Stéphane Tailland
ebf864f7ba
Entry point for the estate: understands a request, consults the specialists that can settle it, composes one answer. LangGraph 1.2.11, a2a-sdk 1.1.2, both current as of August 2026. It holds no data, and the realm is what enforces that rather than the prompt. classic-models-agent-assistant is granted two optional scopes — one per specialist, plus the inference endpoint — and every domain audience it could request comes back Invalid scopes. An injection that succeeds here finds no path to a domain because there is none. "No scopes at all" was too absolute: the agent has to reason, and inference is not a domain. The two-hop token chain was the heaviest unproven thing in the design and it holds: sub stays the end user through all four hops, so their own roles apply at the far end and the MCP server filters tool discovery for the right human. azp names the last agent, not the first — Keycloak accepts actor_token and emits no act claim, so an orchestrator's identity survives only in the traces. Verified end to end: a routed question answered from the specialist's data; a question outside every remit refused rather than answered from memory; and a write proposed through the orchestrator, held by the specialist in input-required, relayed here unchanged, approved by the user through the same conversation, and written. The orchestrator never owns that decision. The graph is plan → delegate → plan, tools described by what a specialist decides rather than what it holds, delegations in one step run concurrently and capped at eight counted in calls. Rebuilt per turn, because its tools carry the user's token. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
57 lines
2.6 KiB
Python
57 lines
2.6 KiB
Python
"""What the orchestrator must do without a model in the loop.
|
|
|
|
The interesting properties here are structural: which specialists exist, what
|
|
they advertise, and what happens to a delegation that is refused or that comes
|
|
back waiting on a decision. None of that needs inference to test, and testing it
|
|
without inference is what makes it a check rather than an observation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from assistant_agent.delegation import Answer
|
|
from assistant_agent.specialists import SPECIALISTS, by_name, describe
|
|
|
|
|
|
class TestRegistry:
|
|
def test_every_specialist_is_reachable_by_name(self) -> None:
|
|
for specialist in SPECIALISTS:
|
|
assert by_name(specialist.name) is specialist or by_name(specialist.name) == specialist
|
|
|
|
def test_an_unknown_name_yields_none_rather_than_raising(self) -> None:
|
|
# The model can emit a tool call for something that no longer exists.
|
|
# That must reach it as an answer, not as a stack trace.
|
|
assert by_name("ventes") is None
|
|
|
|
def test_a_specialist_url_is_overridable_for_local_runs(self, monkeypatch) -> None:
|
|
monkeypatch.setenv("ASSISTANT_URL_CATALOGUE", "http://localhost:8099/")
|
|
assert by_name("catalogue").url == "http://localhost:8099/"
|
|
|
|
def test_every_specialist_names_a_distinct_audience_scope(self) -> None:
|
|
# One token per callee is the whole point: a token minted for one
|
|
# specialist must be refused by every other.
|
|
scopes = [s.audience_scope for s in SPECIALISTS]
|
|
assert len(scopes) == len(set(scopes))
|
|
|
|
@pytest.mark.parametrize("specialist", SPECIALISTS, ids=lambda s: s.name)
|
|
def test_a_specialist_advertises_decisions_not_a_query(self, specialist) -> None:
|
|
# An agent advertised as "ask it anything" has become a data proxy and
|
|
# defeats the per-domain scoping the estate is built on.
|
|
text = describe(specialist).casefold()
|
|
assert "it answers; it does not hand back raw records" in text
|
|
assert specialist.decides, "a specialist with no stated decisions is a query endpoint"
|
|
|
|
|
|
class TestAnswer:
|
|
def test_input_required_is_recognised_as_needing_the_user(self) -> None:
|
|
answer = Answer("catalogue", "…", "TASK_STATE_INPUT_REQUIRED")
|
|
assert answer.needs_the_user is True
|
|
|
|
@pytest.mark.parametrize(
|
|
"state",
|
|
["TASK_STATE_COMPLETED", "TASK_STATE_FAILED", "TASK_STATE_CANCELLED"],
|
|
)
|
|
def test_every_other_state_is_the_orchestrator_s_to_compose(self, state: str) -> None:
|
|
assert Answer("catalogue", "…", state).needs_the_user is False
|