Team rationale: on-demand AI explanation per team

backend/alembic/versions/0009_team_rationale.py

@@ -0,0 +1,27 @@
+"""team rationale
+
+Adds teams.rationale (cached AI explanation per team).
+
+Revision ID: 0009_team_rationale
+Revises: 0008_payout_team_unique
+Create Date: 2026-06-20
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "0009_team_rationale"
+down_revision: Union[str, None] = "0008_payout_team_unique"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("teams") as b:
+        b.add_column(sa.Column("rationale", sa.JSON(), nullable=True))
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("teams") as b:
+        b.drop_column("rationale")

backend/app/api/v1/public.py

@@ -50,6 +50,7 @@ def public_allocation(allocation_id: UUID, db: Session = Depends(get_db)):
             name=team.name,
             fairness_score=team.fairness_score,
             members=[PublicTeamMember.model_validate(m) for m in members],
+            rationale=team.rationale,
         ))
     payouts = []
     for p in db.query(Payout).filter(Payout.allocation_id == allocation.id).all():

backend/app/api/v1/rationale.py

@@ -0,0 +1,25 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from app.api.deps import get_current_user, get_db
+from app.models.user import User
+from app.services.event_service import assert_allocation_organizer
+from app.services import rationale_service
+from app.services.rationale_service import RationaleUnavailable
+
+router = APIRouter()
+
+
+@router.post("/{allocation_id}/rationale")
+def generate_rationale(
+    allocation_id: UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    allocation = assert_allocation_organizer(db, allocation_id, current_user.id)
+    try:
+        return rationale_service.generate(db, allocation)
+    except RationaleUnavailable as exc:
+        raise HTTPException(status_code=400, detail=str(exc))

backend/app/api/v1/teams.py

@@ -36,6 +36,7 @@ def list_teams(allocation_id: UUID, db: Session = Depends(get_db), current_user=
             skill_score=team.skill_score,
             role_balance_score=team.role_balance_score,
             members=[TeamMemberOut.model_validate(m) for m in members],
+            rationale=team.rationale,
         ))
     return result
 
@@ -60,6 +61,7 @@ def get_team(allocation_id: UUID, team_id: UUID, db: Session = Depends(get_db),
         skill_score=team.skill_score,
         role_balance_score=team.role_balance_score,
         members=[TeamMemberOut.model_validate(m) for m in members],
+        rationale=team.rationale,
     )
 
 

backend/app/core/config.py

@@ -27,6 +27,8 @@ def cors_origins(self) -> list[str]:
     # When unset, allocation falls back to a deterministic slug per Other entry.
     ANTHROPIC_API_KEY: str | None = None
     CATEGORIZATION_MODEL: str = "claude-haiku-4-5-20251001"
+    # Model for AI team rationales (separate from categorization so it can be tuned).
+    RATIONALE_MODEL: str = "claude-haiku-4-5-20251001"
 
     # --- Nostr DM sender (all optional; unset → DM sending is a no-op) ---
     # Dedicated *bot* secret key (bech32 `nsec1…`) used ONLY to sign/encrypt

backend/app/main.py

@@ -4,7 +4,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.core.config import settings
-from app.api.v1 import auth, events, participants, allocation, teams, export, public, feedback, payouts
+from app.api.v1 import auth, events, participants, allocation, teams, export, public, feedback, payouts, rationale
 import app.models  # noqa: F401
 
 
@@ -32,6 +32,7 @@ async def lifespan(_: FastAPI):
 app.include_router(public.router, prefix="/api/v1/public", tags=["public"])
 app.include_router(feedback.router, prefix="/api/v1/feedback", tags=["feedback"])
 app.include_router(payouts.router, prefix="/api/v1/allocations", tags=["payouts"])
+app.include_router(rationale.router, prefix="/api/v1/allocations", tags=["rationale"])
 
 
 @app.get("/health")

backend/app/models/team.py

@@ -1,5 +1,5 @@
 import uuid
-from sqlalchemy import Column, String, Float, ForeignKey, Uuid
+from sqlalchemy import Column, String, Float, ForeignKey, Uuid, JSON
 
 from app.core.database import Base
 
@@ -13,6 +13,8 @@ class Team(Base):
     fairness_score = Column(Float, nullable=True)
     skill_score = Column(Float, nullable=True)
     role_balance_score = Column(Float, nullable=True)
+    # Cached AI explanation: {"title","summary","strengths":[...],"gaps":[...]} or None.
+    rationale = Column(JSON, nullable=True)
 
 
 class TeamMember(Base):

backend/app/schemas/allocation.py

@@ -38,6 +38,7 @@ class TeamOut(BaseModel):
     skill_score: Optional[float]
     role_balance_score: Optional[float]
     members: list[TeamMemberOut] = []
+    rationale: Optional[dict] = None
 
     model_config = {"from_attributes": True}
 
@@ -71,6 +72,7 @@ class PublicTeam(BaseModel):
     name: str
     fairness_score: Optional[float]
     members: list[PublicTeamMember] = []
+    rationale: Optional[dict] = None
 
 
 class PublicPayoutSummary(BaseModel):

backend/app/services/allocation_engine.py

@@ -91,6 +91,7 @@ def move_participant(db: Session, allocation: Allocation, participant_id: UUID,
         team.skill_score = round(skill_score, 1)
         team.role_balance_score = round(role_balance_score, 1)
         team.fairness_score = round(fairness_score, 1)
+        team.rationale = None
     db.commit()
 
 

backend/app/services/rationale_service.py

@@ -0,0 +1,151 @@
+"""Generate a short, PII-free "why this team works" rationale per team.
+
+Descriptive only — never reads into or changes the deterministic allocation
+engine. Mirrors categorization_service: pure _build_request/_parse helpers plus a
+monkeypatchable _classify. The AI input carries no participant names or emails.
+"""
+import logging
+
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.models.allocation import Allocation
+from app.models.event import Event
+from app.models.participant import Participant
+from app.models.team import Team, TeamMember
+
+logger = logging.getLogger(__name__)
+
+_MAX_TOKENS = 4096
+
+_SYSTEM_INSTRUCTIONS = (
+    "You explain why each already-formed team is balanced. You do NOT form or change "
+    "teams — you describe what exists. For each team return a short title (<=5 words), a "
+    "one-sentence summary, a list of strengths, and a list of coverage gaps, grounded in "
+    "the members' roles and experience. Never name or invent individuals; describe the "
+    "composition only. If a team is too sparse to assess, omit it from the response."
+)
+
+
+class RationaleUnavailable(Exception):
+    """Raised when AI rationale cannot run (no ANTHROPIC_API_KEY configured)."""
+
+
+def _team_payloads(db: Session, allocation: Allocation) -> list[dict]:
+    """PII-free composition per team: roles, experience, and any tech_stack/interests."""
+    payloads: list[dict] = []
+    teams = db.query(Team).filter(Team.allocation_id == allocation.id).all()
+    for team in teams:
+        members = (
+            db.query(Participant)
+            .join(TeamMember, Participant.id == TeamMember.participant_id)
+            .filter(TeamMember.team_id == team.id)
+            .all()
+        )
+        payloads.append({
+            "id": str(team.id),
+            "name": team.name,
+            "members": [{
+                "role": m.normalized_strength or m.primary_strength,
+                "experience": m.experience_level,
+                "tech_stack": m.tech_stack or [],
+                "interests": m.interests or [],
+            } for m in members],
+        })
+    return payloads
+
+
+def _build_request(event: Event, payloads: list[dict]) -> dict:
+    import json
+
+    tool = {
+        "name": "explain_teams",
+        "description": "Return a structured rationale for each team.",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "rationales": {
+                    "type": "array",
+                    "items": {
+                        "type": "object",
+                        "properties": {
+                            "team_id": {"type": "string"},
+                            "title": {"type": "string"},
+                            "summary": {"type": "string"},
+                            "strengths": {"type": "array", "items": {"type": "string"}},
+                            "gaps": {"type": "array", "items": {"type": "string"}},
+                        },
+                        "required": ["team_id", "title", "summary", "strengths", "gaps"],
+                    },
+                }
+            },
+            "required": ["rationales"],
+        },
+    }
+    return {
+        "model": settings.RATIONALE_MODEL,
+        "max_tokens": _MAX_TOKENS,
+        "system": [{
+            "type": "text",
+            "text": _SYSTEM_INSTRUCTIONS,
+            "cache_control": {"type": "ephemeral"},
+        }],
+        "tools": [tool],
+        "tool_choice": {"type": "tool", "name": "explain_teams"},
+        "messages": [{
+            "role": "user",
+            "content": (
+                f"Event: {event.title}\nDescription: {event.description or '(none)'}\n\n"
+                f"Explain each team from this composition (JSON):\n{json.dumps(payloads)}"
+            ),
+        }],
+    }
+
+
+_REQUIRED_KEYS = {"team_id", "title", "summary", "strengths", "gaps"}
+
+
+def _parse_rationales(content_blocks) -> dict[str, dict]:
+    out: dict[str, dict] = {}
+    for block in content_blocks:
+        if getattr(block, "type", None) == "tool_use":
+            for r in block.input.get("rationales", []):
+                if _REQUIRED_KEYS.issubset(r) and r["team_id"]:
+                    out[r["team_id"]] = {
+                        "title": r["title"], "summary": r["summary"],
+                        "strengths": list(r["strengths"]), "gaps": list(r["gaps"]),
+                    }
+    return out
+
+
+def _classify(event: Event, payloads: list[dict]) -> dict[str, dict]:
+    """Call Claude for all teams; return {team_id: rationale}. Raises on failure."""
+    import anthropic
+
+    client = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
+    msg = client.messages.create(**_build_request(event, payloads))
+    return _parse_rationales(msg.content)
+
+
+def generate(db: Session, allocation: Allocation) -> dict[str, dict]:
+    """Generate + persist a rationale per team. Raises RationaleUnavailable with no key."""
+    if not settings.ANTHROPIC_API_KEY:
+        raise RationaleUnavailable("AI rationale requires ANTHROPIC_API_KEY")
+    event = db.query(Event).filter(Event.id == allocation.event_id).first()
+    if event is None:
+        raise RationaleUnavailable("event not found for allocation")
+    payloads = _team_payloads(db, allocation)
+
+    mapping: dict[str, dict] = {}
+    try:
+        mapping = _classify(event, payloads)
+    except Exception as exc:  # noqa: BLE001 — best-effort; teams without a rationale just stay null
+        logger.warning("Rationale generation failed: %s", exc)
+
+    teams = db.query(Team).filter(Team.allocation_id == allocation.id).all()
+    for team in teams:
+        r = mapping.get(str(team.id))
+        if r:
+            team.rationale = r
+    db.commit()
+    return mapping

backend/tests/test_rationale_endpoint.py

@@ -0,0 +1,41 @@
+from tests.test_payout_endpoint import _setup_team
+
+
+def test_rationale_requires_api_key_returns_400(client, auth_headers, monkeypatch):
+    from app.core.config import settings
+    monkeypatch.setattr(settings, "ANTHROPIC_API_KEY", None, raising=False)
+    _, allocation_id, _team_id, _members = _setup_team(client, auth_headers, all_have_addresses=False)
+    res = client.post(f"/api/v1/allocations/{allocation_id}/rationale", headers=auth_headers)
+    assert res.status_code == 400
+    assert "anthropic" in res.text.lower()
+
+
+def test_rationale_generates_and_persists(client, auth_headers, monkeypatch):
+    from app.core.config import settings
+    from app.services import rationale_service
+    monkeypatch.setattr(settings, "ANTHROPIC_API_KEY", "k", raising=False)
+    monkeypatch.setattr(rationale_service, "_classify", lambda event, payloads: {
+        p["id"]: {"title": "Squad", "summary": "Balanced.", "strengths": ["a"], "gaps": ["b"]}
+        for p in payloads
+    })
+    _, allocation_id, _team_id, _members = _setup_team(client, auth_headers, all_have_addresses=False)
+    res = client.post(f"/api/v1/allocations/{allocation_id}/rationale", headers=auth_headers)
+    assert res.status_code == 200, res.text
+    body = res.json()
+    assert all(r["title"] == "Squad" for r in body.values())
+    teams = client.get(f"/api/v1/allocations/{allocation_id}/teams", headers=auth_headers).json()
+    assert all(t["rationale"]["summary"] == "Balanced." for t in teams)
+
+
+def test_rationale_requires_organizer(client, auth_headers, monkeypatch):
+    from app.core.config import settings
+    monkeypatch.setattr(settings, "ANTHROPIC_API_KEY", "k", raising=False)
+    _, allocation_id, _team_id, _members = _setup_team(client, auth_headers, all_have_addresses=False)
+    from tests.conftest import make_nostr_event
+    from coincurve import PrivateKey
+    other = PrivateKey()
+    pubkey = other.public_key.format(compressed=True)[1:].hex()
+    token = client.post("/auth/nostr", json={"pubkey": pubkey, "event": make_nostr_event(other)}).json()["access_token"]
+    res = client.post(f"/api/v1/allocations/{allocation_id}/rationale",
+                      headers={"Authorization": f"Bearer {token}"})
+    assert res.status_code in (401, 403, 404)

backend/tests/test_rationale_model.py

@@ -0,0 +1,32 @@
+import uuid
+from app.models.allocation import Allocation
+from app.models.team import Team
+
+
+def test_team_rationale_persists_json(db):
+    alloc = Allocation(event_id=uuid.uuid4(), snapshot_hash="h", status="draft",
+                       constraint_warnings={})
+    db.add(alloc)
+    db.flush()
+    team = Team(allocation_id=alloc.id, name="Team 01",
+                rationale={"title": "Build squad", "summary": "Strong delivery.",
+                           "strengths": ["2 advanced engineers"], "gaps": ["limited outreach"]})
+    db.add(team)
+    db.commit()
+    db.refresh(team)
+    assert team.rationale["title"] == "Build squad"
+    assert team.rationale["gaps"] == ["limited outreach"]
+    assert team.rationale["summary"] == "Strong delivery."
+    assert team.rationale["strengths"] == ["2 advanced engineers"]
+
+
+def test_team_rationale_defaults_none(db):
+    alloc = Allocation(event_id=uuid.uuid4(), snapshot_hash="h2", status="draft",
+                       constraint_warnings={})
+    db.add(alloc)
+    db.flush()
+    team = Team(allocation_id=alloc.id, name="Team 02")
+    db.add(team)
+    db.commit()
+    db.refresh(team)
+    assert team.rationale is None

backend/tests/test_rationale_move.py

@@ -0,0 +1,22 @@
+from tests.test_payout_endpoint import _setup_team
+
+
+def test_move_clears_rationale(client, auth_headers, monkeypatch):
+    from app.core.config import settings
+    from app.services import rationale_service
+    monkeypatch.setattr(settings, "ANTHROPIC_API_KEY", "k", raising=False)
+    monkeypatch.setattr(rationale_service, "_classify", lambda event, payloads: {
+        p["id"]: {"title": "Squad", "summary": "Balanced.", "strengths": ["a"], "gaps": []}
+        for p in payloads
+    })
+    _, allocation_id, team_id, members = _setup_team(client, auth_headers, all_have_addresses=False)
+    client.post(f"/api/v1/allocations/{allocation_id}/rationale", headers=auth_headers)
+
+    teams = client.get(f"/api/v1/allocations/{allocation_id}/teams", headers=auth_headers).json()
+    target = next(t["id"] for t in teams if t["id"] != team_id)
+    mover = members[0]["id"]
+    client.patch(f"/api/v1/allocations/{allocation_id}/members/{mover}",
+                 headers=auth_headers, json={"team_id": target})
+
+    teams_after = client.get(f"/api/v1/allocations/{allocation_id}/teams", headers=auth_headers).json()
+    assert all(t["rationale"] is None for t in teams_after)