From 6ebe9387653363799fce03fc4592d18696524ee2 Mon Sep 17 00:00:00 2001 From: Jacob Malm <28159678+Jacobobber@users.noreply.github.com> Date: Wed, 8 Jul 2026 14:27:50 -0600 Subject: [PATCH] Add auth-native routing (Router): route by capability = scope set The differentiator the harness research called out. Route a task to a worker whose harness (its granted scopes) covers the scopes the task needs: - Router.register(worker, scopes); route(required) -> a capable worker (round-robin across eligible), or None if none is capable (visible, not a silent misroute); candidates(required) lists them. - The same scopes that gate tool calls decide who is capable of a task, so work can't be routed to an agent that isn't authorized for it. Pair with dispatch or the task queue: route first, hand off to the chosen worker. 292 tests (+6), mypy --strict, ruff (whole repo), catalog freshness all green. v1.6.0. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 4 +++ README.md | 19 ++++++++++++- docs/IDEAS.md | 1 + pyproject.toml | 2 +- src/hallpass/__init__.py | 5 ++-- src/hallpass/orchestrator.py | 42 +++++++++++++++++++++++++++- tests/test_router.py | 54 ++++++++++++++++++++++++++++++++++++ 7 files changed, 122 insertions(+), 5 deletions(-) create mode 100644 tests/test_router.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 622833e..1ad479e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ All notable changes to hallpass. This project follows [semantic versioning](https://semver.org): from 1.0.0 on, the public API (everything exported from the top-level `hallpass` package) is stable, and breaking changes bump the major version. Per-release detail is in the [GitHub releases](https://github.com/Jacobobber/hallpass/releases). +## [1.6.0] + +- **Auth-native routing** (`Router`) — route a task to a worker by capability, where capability is the auth scope set. A worker registers its harness (granted scopes); a task declares the scopes it needs; `route` returns a worker whose harness covers them, round-robin across the eligible ones, or `None` if none is capable (visible, not a silent misroute). The differentiator the harness research called out: the same scopes that gate tool calls decide who is capable of a task, so work can't be routed to an agent that isn't authorized for it. Pair with `dispatch` or the task queue. Additive. + ## [1.5.0] - **Durable task queue** (`hallpass.taskqueue`) — `TaskQueue` on SQLite, closing the two gaps the harness research flagged (event-log resume, dedup + lease). `enqueue` writes a pending task; `claim` hands one worker exactly one task under a write-locked transaction (concurrent workers never claim the same one) and leases it; an expired lease makes an abandoned task claimable again (a dead worker doesn't strand it); `complete` is idempotent by id, so a re-run cannot overwrite a recorded result; `result` / `outstanding` survive a restart, so a resuming orchestrator sees what is still in flight. A coordination/durability primitive; the auth boundary stays on the tools a worker calls. Additive. diff --git a/README.md b/README.md index e85aab0..596fab8 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Multi-user auth core for MCP servers: per-user OAuth 2.1 verification against any OIDC provider, an encrypted per-user credential vault, and scope-derived tool gating that is enforced at call time, not just in the catalog. The same identity and scope model also governs agent-to-agent channels, agent orchestration, and relevance-ranked tool search, so one auth layer covers agent-to-tools, agent-to-agent, and finding the right tool among many. -**Status: v1.5 — stable.** Core, MCP adapter, operational layer (audit, rate limiting, availability, idempotency), agent-to-agent channels (with FLEX, a token-efficient message language) and an orchestrator that spawns scoped worker agents and drives them over those channels, tool search, batteries-included setup, a catalog of prewired connectors, a per-provider OAuth connect flow with self-healing token refresh and consent/revoke, transient-error retry with backoff, untrusted-message sanitization, a response-size guard, a `doctor()` config self-check, and a runnable HTTP reference server + `hallpass` CLI are in place and green. The public API (everything exported from `hallpass`) is committed to under semver from 1.0; see [CHANGELOG.md](CHANGELOG.md). +**Status: v1.6 — stable.** Core, MCP adapter, operational layer (audit, rate limiting, availability, idempotency), agent-to-agent channels (with FLEX, a token-efficient message language) and an orchestrator that spawns scoped worker agents and drives them over those channels, tool search, batteries-included setup, a catalog of prewired connectors, a per-provider OAuth connect flow with self-healing token refresh and consent/revoke, transient-error retry with backoff, untrusted-message sanitization, a response-size guard, a `doctor()` config self-check, and a runnable HTTP reference server + `hallpass` CLI are in place and green. The public API (everything exported from `hallpass`) is committed to under semver from 1.0; see [CHANGELOG.md](CHANGELOG.md). The design essay behind this: [Multi-user is the hard part of an MCP server](docs/multi-user-is-the-hard-part.md). @@ -214,6 +214,23 @@ print(orch.gather([task_id])[task_id]) # Result(ok=True, fields={"status": "don Because it rides `A2ABus`, dispatch and results are gated by the channel's scopes (the harness does not bypass the auth core), durable (a worker that dies mid-task sees it on reconnect), and audited through the same sink. Delivery is at-least-once, so `gather` de-duplicates by task id and handlers should be idempotent; a handler that raises produces a failed result carrying only the exception type, never its message. Addressing is by convention on a shared channel; for hard isolation, give each worker its own channel with its own read scope. A runnable end-to-end demo is [`examples/orchestrator.py`](examples/orchestrator.py). +### Routing by capability + +`Router` picks a worker for a task by capability, where capability is the auth scope set. Each worker registers its harness; a task declares the scopes it needs; `route` returns a worker whose harness covers them (round-robin across the eligible ones): + +```python +from hallpass import Router + +router = Router() +router.register("reviewer", {"github:read", "github:write"}) +router.register("messenger", {"slack:write"}) + +router.route({"github:write"}) # -> "reviewer" +router.route({"pagerduty:read"}) # -> None: nobody is capable, and that's visible +``` + +The routing decision uses the same scopes that gate tool calls, so work never lands on an agent that isn't authorized to do it, and an unroutable task surfaces as `None` rather than a silent misroute. Pair it with `dispatch` or the task queue: route first, then hand the task to the chosen worker. + ### Spawning agents The step past dispatching to workers that already exist is *creating* them, each with a different harness and a different task. `Team.spawn` mints each agent a token carrying only its harness scopes and launches it through a pluggable `Spawner`, passing the identity, task, and channel by environment: diff --git a/docs/IDEAS.md b/docs/IDEAS.md index f2bb3a3..d074b5a 100644 --- a/docs/IDEAS.md +++ b/docs/IDEAS.md @@ -9,6 +9,7 @@ only when it is built and tested. - Identity (OAuth 2.1 resource-server verification), credential vault, scope gating, MCP adapter. - Operational layer: audit trail (denials included), per-principal rate limiting, connector availability. `SqliteAuditLog` is a durable, queryable sink (`query` by subject/tool/decision/time); `AuditEvent.duration_ms` records call latency, so the audit trail doubles as an observability source without an OpenTelemetry dependency. - Agent-to-agent channels: authenticated, authorized, durable. +- Auth-native routing (`hallpass.Router`): route a task to a worker by capability, where capability is the auth scope set (the worker's harness). A task declares required scopes; `route` returns a worker whose harness covers them (round-robin), or None if none is capable. The differentiator: the same scopes that gate tool calls decide who is capable, so work can't land on an unauthorized agent. - Durable task queue (`hallpass.taskqueue`): `TaskQueue` on SQLite gives crash-recovery and exactly-once. `claim` hands one worker exactly one task under a write lock (no double-runs); an expired lease makes an abandoned task claimable again (dead-worker recovery); `complete` is idempotent by id; results and backlog survive a restart, so a resuming orchestrator sees what's outstanding. The two gaps the harness research flagged (event-log resume, dedup+lease) in one primitive. - Agent spawning (`hallpass.agents`): the orchestrator creates scoped agents, not just dispatches to existing ones. `Team.spawn(AgentSpec)` mints each agent a token carrying only its harness scopes and launches it through a pluggable `Spawner` (default `SubprocessSpawner`), passing name/token/task/channel by environment; `AgentContext.from_env()` picks it up inside the process. Each agent is a scoped identity, not a trusted one; hallpass stays model-agnostic (the process's loop is yours). Demo: `examples/spawn_agents.py`. - Orchestrator (`hallpass.orchestrator`): an agent that drives worker agents over an A2A channel. `Orchestrator.dispatch`/`gather` + `Worker` with registered handlers; tasks/results are FLEX messages tagged by id. Rides `A2ABus` so it is scope-gated, durable, and audited; at-least-once with gather de-dup; failed handlers report only the exception type. The coordination layer the auth core composes into. diff --git a/pyproject.toml b/pyproject.toml index ffac01f..b964293 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "hallpass" -version = "1.5.0" +version = "1.6.0" description = "Multi-user auth core for MCP servers: per-user OAuth, encrypted credential vault, scope-gated tools." readme = "README.md" requires-python = ">=3.10" diff --git a/src/hallpass/__init__.py b/src/hallpass/__init__.py index 4065f83..2935e2d 100644 --- a/src/hallpass/__init__.py +++ b/src/hallpass/__init__.py @@ -43,7 +43,7 @@ SqlitePendingStore, TokenHttp, ) -from .orchestrator import Handler, Orchestrator, Result, Task, Worker +from .orchestrator import Handler, Orchestrator, Result, Router, Task, Worker from .taskqueue import LeasedTask, TaskQueue from .ratelimit import FixedWindowRateLimiter, RateLimited, RateLimiter from .sanitize import frame_untrusted, sanitize @@ -66,7 +66,7 @@ from .toolkit import ToolKit from .vault import CredentialVault, VaultError -__version__ = "1.5.0" +__version__ = "1.6.0" __all__ = [ "A2ABus", @@ -116,6 +116,7 @@ "RestService", "RetryPolicy", "RetryingHttpClient", + "Router", "SqliteAuditLog", "SqlitePendingStore", "Spawner", diff --git a/src/hallpass/orchestrator.py b/src/hallpass/orchestrator.py index 206dfd3..f30e591 100644 --- a/src/hallpass/orchestrator.py +++ b/src/hallpass/orchestrator.py @@ -26,6 +26,7 @@ from __future__ import annotations import secrets +import threading from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field @@ -33,7 +34,7 @@ from .a2a import A2ABus from .identity import Principal -__all__ = ["Task", "Result", "Orchestrator", "Worker", "Handler"] +__all__ = ["Task", "Result", "Orchestrator", "Worker", "Handler", "Router"] @dataclass(frozen=True) @@ -195,3 +196,42 @@ def _reply( note=note, ) self._bus.post(self._principal, self._channel, flex.encode(message)) + + +class Router: + """Route a task to a worker by capability, where capability is the auth + scope set. Each worker registers its harness (the scopes its token carries); + a task declares the scopes it needs; ``route`` returns a worker whose harness + covers them, round-robin across the eligible ones. Auth-native: the same + scopes that gate tool calls decide who is *capable* of a task, so work never + lands on an agent that could not perform it anyway (and if none is capable, + that is visible, not a silent misroute). Pair it with ``dispatch`` or a + ``TaskQueue``: route first, then hand the task to the chosen worker.""" + + def __init__(self) -> None: + self._workers: dict[str, frozenset[str]] = {} + self._rr = 0 + self._lock = threading.Lock() + + def register(self, worker: str, scopes: Iterable[str]) -> None: + """Declare a worker and the scopes its harness grants.""" + with self._lock: + self._workers[worker] = frozenset(scopes) + + def candidates(self, required: Iterable[str]) -> list[str]: + """Every registered worker whose harness covers ``required``, sorted.""" + need = frozenset(required) + with self._lock: + return sorted(w for w, s in self._workers.items() if need <= s) + + def route(self, required: Iterable[str]) -> str | None: + """A capable worker for a task needing ``required`` scopes, or None if + none is capable. Round-robins across the eligible workers so repeated + routes spread the load.""" + eligible = self.candidates(required) + if not eligible: + return None + with self._lock: + pick = eligible[self._rr % len(eligible)] + self._rr += 1 + return pick diff --git a/tests/test_router.py b/tests/test_router.py new file mode 100644 index 0000000..affa646 --- /dev/null +++ b/tests/test_router.py @@ -0,0 +1,54 @@ +"""Auth-native routing: a task is routed to a worker whose harness (its granted +scopes) covers the scopes the task needs, so work never lands on an agent that +isn't authorized to do it. What matters: only capable workers are candidates, +routing round-robins across them, and an unroutable task is visibly None (not a +silent misroute).""" + +from hallpass import Router + + +def _router(): + r = Router() + r.register("reviewer", {"github:read", "github:write"}) + r.register("reader", {"github:read"}) + r.register("everything", {"github:read", "github:write", "slack:write"}) + return r + + +def test_candidates_are_only_capable_workers(): + r = _router() + assert r.candidates({"github:read"}) == ["everything", "reader", "reviewer"] + assert r.candidates({"github:write"}) == ["everything", "reviewer"] + assert r.candidates({"slack:write"}) == ["everything"] + assert r.candidates({"pagerduty:read"}) == [] # nobody has it + + +def test_route_picks_a_capable_worker(): + r = _router() + picked = r.route({"github:write"}) + assert picked in {"everything", "reviewer"} # both can, neither "reader" + + +def test_route_round_robins_across_eligible(): + r = _router() + # three workers can do github:read; consecutive routes should rotate + picks = [r.route({"github:read"}) for _ in range(3)] + assert set(picks) == {"everything", "reader", "reviewer"} # spread, not stuck + + +def test_unroutable_task_returns_none(): + r = _router() + assert r.route({"pagerduty:read"}) is None # no capable worker -> visible None + + +def test_route_needs_all_required_scopes(): + r = Router() + r.register("partial", {"a:read"}) # has one of the two + assert r.route({"a:read", "b:read"}) is None + r.register("full", {"a:read", "b:read"}) + assert r.route({"a:read", "b:read"}) == "full" + + +def test_empty_requirement_matches_any_worker(): + r = _router() + assert r.route(set()) is not None # a task needing nothing can go anywhere