feat(agents): resolve managed identities from authoritative storage

litellm/proxy/_lazy_openapi_snapshot.json

@@ -2521,6 +2521,91 @@
           "title": "AgentExtension",
           "type": "object"
         },
+        "AgentIdentityBinding": {
+          "properties": {
+            "active": {
+              "default": true,
+              "title": "Active",
+              "type": "boolean"
+            },
+            "agent_id": {
+              "title": "Agent Id",
+              "type": "string"
+            },
+            "client_id": {
+              "title": "Client Id",
+              "type": "string"
+            },
+            "issuer": {
+              "title": "Issuer",
+              "type": "string"
+            },
+            "last_authenticated_at": {
+              "anyOf": [
+                {
+                  "format": "date-time",
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "title": "Last Authenticated At"
+            },
+            "provider": {
+              "const": "microsoft_entra",
+              "title": "Provider",
+              "type": "string"
+            },
+            "required_roles": {
+              "default": [],
+              "items": {
+                "type": "string"
+              },
+              "title": "Required Roles",
+              "type": "array"
+            },
+            "required_scopes": {
+              "default": [
+                "user_impersonation"
+              ],
+              "items": {
+                "type": "string"
+              },
+              "title": "Required Scopes",
+              "type": "array"
+            },
+            "revision": {
+              "title": "Revision",
+              "type": "string"
+            },
+            "service_principal_id": {
+              "anyOf": [
+                {
+                  "type": "string"
+                },
+                {
+                  "type": "null"
+                }
+              ],
+              "title": "Service Principal Id"
+            },
+            "tenant_id": {
+              "title": "Tenant Id",
+              "type": "string"
+            }
+          },
+          "required": [
+            "agent_id",
+            "provider",
+            "tenant_id",
+            "client_id",
+            "issuer",
+            "revision"
+          ],
+          "title": "AgentIdentityBinding",
+          "type": "object"
+        },
         "AgentInterface": {
           "description": "Declares a combination of a target URL and a transport protocol.",
           "properties": {
@@ -2972,6 +3057,21 @@
               ],
               "title": "Created By"
             },
+            "enabled": {
+              "default": true,
+              "title": "Enabled",
+              "type": "boolean"
+            },
+            "execution_mode": {
+              "default": "autonomous",
+              "enum": [
+                "autonomous",
+                "delegated",
+                "both"
+              ],
+              "title": "Execution Mode",
+              "type": "string"
+            },
             "extra_headers": {
               "anyOf": [
                 {
@@ -2986,6 +3086,26 @@
               ],
               "title": "Extra Headers"
             },
+            "identity": {
+              "anyOf": [
+                {
+                  "$ref": "#/components/schemas/AgentIdentityBinding"
+                },
+                {
+                  "type": "null"
+                }
+              ]
+            },
+            "identity_managed": {
+              "default": false,
+              "title": "Identity Managed",
+              "type": "boolean"
+            },
+            "jwt_auth_configured": {
+              "default": false,
+              "title": "Jwt Auth Configured",
+              "type": "boolean"
+            },
             "keys": {
               "anyOf": [
                 {

litellm/proxy/agent_endpoints/identity_store.py

@@ -0,0 +1,222 @@
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING, Final
+
+from litellm.proxy.agent_endpoints.managed_identity import classify_agent_subject
+from litellm.repositories.table_repositories import (
+    AgentIdentityRepository,
+    AgentsRepository,
+    RetiredAgentIdentityRepository,
+    RetiredAgentRepository,
+    VerifiedSubjectRepository,
+)
+from litellm.types.agents import AgentResponse
+from litellm.types.proxy.agent_identity import (
+    AgentIdentityFailure,
+    ManagedAgentContext,
+    MicrosoftInteractiveSubject,
+    VerifiedHumanSubject,
+)
+
+if TYPE_CHECKING:
+    from prisma.models import LiteLLM_VerifiedSubject
+    from prisma.types import (
+        LiteLLM_AgentIdentityUpdateManyMutationInput,
+        LiteLLM_AgentIdentityWhereInput,
+        LiteLLM_AgentIdentityWhereUniqueInput,
+        LiteLLM_AgentsTableInclude,
+        LiteLLM_AgentsTableWhereUniqueInput,
+        LiteLLM_VerifiedSubjectCreateInput,
+        LiteLLM_VerifiedSubjectUpsertInput,
+        LiteLLM_VerifiedSubjectWhereUniqueInput,
+    )
+
+
+class AgentIdentityStore:
+    @classmethod
+    def from_client(cls, client: object) -> "AgentIdentityStore":
+        return cls(
+            AgentsRepository(client, use_writer=True),
+            AgentIdentityRepository(client, use_writer=True),
+            VerifiedSubjectRepository(client, use_writer=True),
+            RetiredAgentIdentityRepository(client, use_writer=True),
+            RetiredAgentRepository(client, use_writer=True),
+        )
+
+    def __init__(
+        self,
+        agents: AgentsRepository,
+        identities: AgentIdentityRepository,
+        humans: VerifiedSubjectRepository,
+        retired: RetiredAgentIdentityRepository | None = None,
+        retired_agents: RetiredAgentRepository | None = None,
+    ) -> None:
+        self.agents = agents
+        self.identities = identities
+        self.humans = humans
+        self.retired = retired
+        self.retired_agents = retired_agents
+
+    async def agent(self, agent_id: str) -> AgentResponse | AgentIdentityFailure | None:
+        try:
+            where: Final[LiteLLM_AgentsTableWhereUniqueInput] = {"agent_id": agent_id}
+            include: Final[LiteLLM_AgentsTableInclude] = {
+                "identity": True,
+                "object_permission": True,
+            }
+            row: Final = await self.agents.table.find_unique(where=where, include=include)
+            if row is None:
+                return None
+            return AgentResponse.model_validate(row.model_dump())
+        except Exception:
+            return AgentIdentityFailure(code="policy_unavailable", message="Agent policy could not be loaded")
+
+    async def unbound_client(self, where: "LiteLLM_AgentIdentityWhereUniqueInput") -> AgentIdentityFailure | None:
+        if self.retired is not None:
+            try:
+                retired: Final = await self.retired.table.find_unique(where=where)
+            except Exception:
+                return AgentIdentityFailure(
+                    code="policy_unavailable", message="Retired agent identity could not be checked"
+                )
+            if retired is not None:
+                return AgentIdentityFailure(message="This agent identity binding has been retired")
+        return None
+
+    async def resolve_verified_claims(
+        self, claims: Mapping[str, object]
+    ) -> ManagedAgentContext | AgentIdentityFailure | None:
+        issuer: Final = claims.get("iss")
+        tenant: Final = claims.get("tid")
+        client: Final = claims.get("azp")
+        if not isinstance(issuer, str) or not isinstance(tenant, str) or not isinstance(client, str):
+            return None
+        where: Final[LiteLLM_AgentIdentityWhereUniqueInput] = {
+            "provider_tenant_id_client_id": {"provider": "microsoft_entra", "tenant_id": tenant, "client_id": client}
+        }
+        try:
+            row: Final = await self.identities.table.find_unique(where=where)
+        except Exception:
+            return AgentIdentityFailure(code="policy_unavailable", message="Agent identity could not be loaded")
+        proven: Final = await self.subject(issuer, tenant, claims.get("oid"))
+        if isinstance(proven, AgentIdentityFailure):
+            return proven
+        if row is None:
+            return await self.unbound_client(where)
+        agent: Final = await self.agent(row.agent_id)
+        if isinstance(agent, AgentIdentityFailure):
+            return agent
+        if (
+            agent is None
+            or not agent.identity_managed
+            or not agent.enabled
+            or agent.identity is None
+            or not agent.identity.active
+        ):
+            return AgentIdentityFailure(message="Agent is disabled or no longer bound to an identity")
+        subject: Final = classify_agent_subject(agent.identity, claims, agent.execution_mode)
+        if isinstance(subject, AgentIdentityFailure):
+            return subject
+        if subject.kind == "application":
+            return ManagedAgentContext(
+                agent_id=agent.agent_id,
+                binding_revision=agent.identity.revision,
+                mode=subject.mode,
+                subject_oid=subject.oid,
+            )
+        human: Final = (
+            VerifiedHumanSubject.model_validate(proven.model_dump())
+            if proven is not None
+            and proven.kind == "human"
+            and proven.verified_via == "sso_interactive"
+            and proven.user_id is not None
+            else None
+        )
+        if human is None:
+            return AgentIdentityFailure(message="The delegated user must first sign in through trusted Microsoft SSO")
+        return ManagedAgentContext(
+            agent_id=agent.agent_id,
+            binding_revision=agent.identity.revision,
+            mode=subject.mode,
+            user_id=human.user_id,
+            subject_oid=subject.oid,
+        )
+
+    async def subject(
+        self, issuer: str, tenant_id: str, oid: object
+    ) -> "LiteLLM_VerifiedSubject | AgentIdentityFailure | None":
+        if not isinstance(oid, str):
+            return None
+        try:
+            where: Final[LiteLLM_VerifiedSubjectWhereUniqueInput] = {
+                "issuer_tenant_id_oid": {"issuer": issuer, "tenant_id": tenant_id, "oid": oid}
+            }
+            return await self.humans.table.find_unique(where=where)
+        except Exception:
+            return AgentIdentityFailure(code="policy_unavailable", message="Subject classification is unavailable")
+
+    async def retired_agent(self, agent_id: str) -> bool | AgentIdentityFailure:
+        if self.retired_agents is None:
+            return AgentIdentityFailure(code="policy_unavailable", message="Agent history is unavailable")
+        try:
+            return await self.retired_agents.table.find_unique(where={"original_agent_id": agent_id}) is not None
+        except Exception:
+            return AgentIdentityFailure(code="policy_unavailable", message="Agent history is unavailable")
+
+    async def record_authentication(self, context: ManagedAgentContext) -> AgentIdentityFailure | None:
+        try:
+            if context.binding_revision is None:
+                return AgentIdentityFailure(message="Agent authentication requires a binding revision")
+            where: Final[LiteLLM_AgentIdentityWhereInput] = {
+                "agent_id": context.agent_id,
+                "revision": context.binding_revision,
+            }
+            data: Final[LiteLLM_AgentIdentityUpdateManyMutationInput] = {
+                "last_authenticated_at": datetime.now(timezone.utc)
+            }
+            count: Final = await self.identities.table.update_many(where=where, data=data)
+            if count != 1:
+                return AgentIdentityFailure(message="Agent identity changed during authentication; retry")
+            return None
+        except Exception:
+            return AgentIdentityFailure(code="policy_unavailable", message="Agent authentication could not be recorded")
+
+    async def enroll_interactive_human(
+        self,
+        subject: MicrosoftInteractiveSubject,
+        user_id: str,
+    ) -> AgentIdentityFailure | None:
+        try:
+            where: Final[LiteLLM_VerifiedSubjectWhereUniqueInput] = {
+                "issuer_tenant_id_oid": {"issuer": subject.issuer, "tenant_id": subject.tenant_id, "oid": subject.oid}
+            }
+            create_data: Final[LiteLLM_VerifiedSubjectCreateInput] = {
+                "issuer": subject.issuer,
+                "tenant_id": subject.tenant_id,
+                "oid": subject.oid,
+                "user_id": user_id,
+                "verified_via": "sso_interactive",
+            }
+            data: Final[LiteLLM_VerifiedSubjectUpsertInput] = {"create": create_data, "update": {}}
+            row: Final = await self.humans.table.upsert(where=where, data=data)
+            if row.kind != "human" or row.user_id != user_id or row.verified_via != "sso_interactive":
+                return AgentIdentityFailure(message="Microsoft subject is already bound to another local identity")
+            return None
+        except Exception:
+            return AgentIdentityFailure(
+                code="policy_unavailable", message="Microsoft subject enrollment is unavailable"
+            )
+
+
+async def resolve_managed_agent(
+    claims: Mapping[str, object],
+    client: object,
+) -> ManagedAgentContext | None:
+    from litellm.proxy.agent_endpoints.managed_identity import raise_identity_failure
+
+    if client is None:
+        return None
+    result: Final = await AgentIdentityStore.from_client(client).resolve_verified_claims(claims)
+    if isinstance(result, AgentIdentityFailure):
+        raise_identity_failure(result)
+    return result

litellm/proxy/agent_endpoints/managed_identity.py

@@ -1,6 +1,8 @@
 from collections.abc import Mapping
 from typing import Final
 
+from fastapi import HTTPException
+
 from litellm.types.proxy.agent_identity import (
     AgentExecutionMode,
     AgentIdentityBinding,
@@ -44,3 +46,7 @@ def classify_agent_subject(
     if not frozenset(binding.required_roles).issubset(roles):
         return AgentIdentityFailure(message="Token lacks the required application roles")
     return AgentSubject(kind="application", oid=oid, mode="autonomous")
+
+
+def raise_identity_failure(failure: AgentIdentityFailure, status_code: int = 403) -> None:
+    raise HTTPException(503 if failure.code == "policy_unavailable" else status_code, failure.message)

litellm/repositories/table_repositories.py

@@ -21,8 +21,9 @@ class PrismaTableRepository(Generic[RowT_co]):
 
     table_name: str
 
-    def __init__(self, prisma_client: object):
+    def __init__(self, prisma_client: object, *, use_writer: bool = False) -> None:
         self._prisma_client = prisma_client
+        self._use_writer = use_writer
 
     @property
     def prisma_client(self) -> Any:
@@ -32,7 +33,9 @@ class PrismaTableRepository(Generic[RowT_co]):
 
     @property
     def table(self) -> TableActions[RowT_co]:
-        actions: Final[TableActions[RowT_co]] = getattr(self.prisma_client.db, self.table_name)
+        actions: Final[TableActions[RowT_co]] = getattr(
+            self.prisma_client.writer_db if self._use_writer else self.prisma_client.db, self.table_name
+        )
         return wrap_table_actions_for_config_sync(actions=actions, table_name=self.table_name)
 
 
@@ -44,6 +47,18 @@ class AgentsRepository(PrismaTableRepository["prisma_models.LiteLLM_AgentsTable"
     table_name = "litellm_agentstable"
 
 
+class AgentIdentityRepository(PrismaTableRepository["prisma_models.LiteLLM_AgentIdentity"]):
+    table_name = "litellm_agentidentity"
+
+
+class RetiredAgentIdentityRepository(PrismaTableRepository["prisma_models.LiteLLM_RetiredAgentIdentity"]):
+    table_name = "litellm_retiredagentidentity"
+
+
+class VerifiedSubjectRepository(PrismaTableRepository["prisma_models.LiteLLM_VerifiedSubject"]):
+    table_name = "litellm_verifiedsubject"
+
+
 class ObjectPermissionRepository(PrismaTableRepository["prisma_models.LiteLLM_ObjectPermissionTable"]):
     table_name = "litellm_objectpermissiontable"
 
@@ -246,3 +261,7 @@ class AuditLogRepository(PrismaTableRepository["prisma_models.LiteLLM_AuditLog"]
 
 class AdaptiveRouterSessionRepository(PrismaTableRepository["prisma_models.LiteLLM_AdaptiveRouterSession"]):
     table_name = "litellm_adaptiveroutersession"
+
+
+class RetiredAgentRepository(PrismaTableRepository["prisma_models.LiteLLM_RetiredAgent"]):
+    table_name = "litellm_retiredagent"

litellm/types/agents.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, StrictInt, field
 from typing_extensions import ReadOnly, Required, TypedDict
 
 from litellm.types.llms.base import LiteLLMPydanticObjectBase
+from litellm.types.proxy.agent_identity import AgentExecutionMode, AgentIdentityBinding
 
 if TYPE_CHECKING:
     from a2a.types import SendMessageResponse
@@ -301,6 +302,11 @@ class AgentKeySummary(BaseModel):
 
 
 class AgentResponse(BaseModel):
+    identity: AgentIdentityBinding | None = None
+    identity_managed: bool = False
+    enabled: bool = True
+    execution_mode: AgentExecutionMode = "autonomous"
+    jwt_auth_configured: bool = False
     agent_id: str
     agent_name: str
     litellm_params: dict[str, object] | None = None

tests/test_litellm/proxy/agent_endpoints/test_identity_store.py

@@ -0,0 +1,381 @@
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from typing import Final
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from fastapi import HTTPException
+from prisma.models import LiteLLM_VerifiedSubject
+
+from litellm.proxy.agent_endpoints.identity_store import AgentIdentityStore, resolve_managed_agent
+from litellm.repositories.table_repositories import (
+    AgentIdentityRepository,
+    AgentsRepository,
+    VerifiedSubjectRepository,
+)
+from litellm.types.agents import AgentResponse
+from litellm.types.proxy.agent_identity import (
+    AgentIdentityBinding,
+    AgentIdentityFailure,
+    ManagedAgentContext,
+    MicrosoftInteractiveSubject,
+)
+
+TENANT: Final = "11111111-1111-4111-8111-111111111111"
+CLIENT: Final = "22222222-2222-4222-8222-222222222222"
+PRINCIPAL: Final = "33333333-3333-4333-8333-333333333333"
+HUMAN: Final = "44444444-4444-4444-8444-444444444444"
+ISSUER: Final = f"https://login.microsoftonline.com/{TENANT}/v2.0"
+BINDING: Final = AgentIdentityBinding(
+    agent_id="agent-one",
+    provider="microsoft_entra",
+    tenant_id=TENANT,
+    client_id=CLIENT,
+    service_principal_id=PRINCIPAL,
+    issuer=ISSUER,
+    required_roles=("Agent.Invoke",),
+    revision="revision-one",
+)
+CLAIMS: Final = {"iss": ISSUER, "tid": TENANT, "azp": CLIENT, "oid": PRINCIPAL, "roles": ["Agent.Invoke"]}
+
+
+def stored_agent(**overrides: object) -> AgentResponse:
+    return AgentResponse.model_validate(
+        {
+            "agent_id": "agent-one",
+            "agent_name": "Research",
+            "agent_card_params": {},
+            "identity": BINDING,
+            "identity_managed": True,
+            "execution_mode": "both",
+            **overrides,
+        }
+    )
+
+
+def setup_store(
+    agent: AgentResponse | None = stored_agent(),
+    human: LiteLLM_VerifiedSubject | None = None,
+) -> tuple[AgentIdentityStore, AsyncMock, AsyncMock, AsyncMock]:
+    agents: Final = AsyncMock()
+    identities: Final = AsyncMock()
+    humans: Final = AsyncMock()
+    agents.find_unique.return_value = agent
+    identities.find_unique.return_value = BINDING
+    identities.update_many.return_value = 1
+    humans.find_unique.return_value = human
+    db: Final = SimpleNamespace(
+        db=SimpleNamespace(
+            litellm_agentstable=agents,
+            litellm_agentidentity=identities,
+            litellm_verifiedsubject=humans,
+        )
+    )
+    return (
+        AgentIdentityStore(AgentsRepository(db), AgentIdentityRepository(db), VerifiedSubjectRepository(db)),
+        agents,
+        identities,
+        humans,
+    )
+
+
+@pytest.mark.asyncio
+async def test_application_authentication_has_no_fabricated_human() -> None:
+    store, _, _, humans = setup_store()
+    result: Final = await store.resolve_verified_claims(CLAIMS)
+    assert isinstance(result, ManagedAgentContext)
+    assert result.agent_id == "agent-one"
+    assert result.mode == "autonomous"
+    assert result.user_id is None
+    humans.upsert.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_lifecycle_is_read_on_every_request_without_cached_allow() -> None:
+    store, agents, _, _ = setup_store()
+    agents.find_unique.side_effect = [stored_agent(), stored_agent(enabled=False)]
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), ManagedAgentContext)
+    denial: Final = await store.resolve_verified_claims(CLAIMS)
+    assert isinstance(denial, AgentIdentityFailure)
+    assert denial.code == "identity_denied"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("unavailable_table", ["agents", "identities", "humans"])
+async def test_identity_store_failure_never_becomes_a_legacy_allow(unavailable_table: str) -> None:
+    store, agents, identities, humans = setup_store()
+    table: Final = {"agents": agents, "identities": identities, "humans": humans}[unavailable_table]
+    table.find_unique.side_effect = RuntimeError("database unavailable")
+    result: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "policy_unavailable"
+
+
+@pytest.mark.asyncio
+async def test_unclassified_delegated_subject_cannot_authenticate_as_a_user() -> None:
+    store, _, _, _ = setup_store()
+    result: Final = await store.resolve_verified_claims(
+        {**CLAIMS, "oid": HUMAN, "scp": "user_impersonation", "idtyp": "user"}
+    )
+    assert isinstance(result, AgentIdentityFailure)
+    assert "first sign in" in result.message
+
+
+@pytest.mark.asyncio
+async def test_delegated_subject_uses_canonical_sso_user_not_email_claim() -> None:
+    human: Final = LiteLLM_VerifiedSubject(
+        kind="human",
+        subject_id="subject-one",
+        issuer=ISSUER,
+        tenant_id=TENANT,
+        oid=HUMAN,
+        user_id="canonical-user",
+        verified_via="sso_interactive",
+        verified_at=datetime.now(timezone.utc),
+    )
+    store, _, _, humans = setup_store(human=human)
+    result: Final = await store.resolve_verified_claims(
+        {
+            **CLAIMS,
+            "oid": HUMAN,
+            "scp": "user_impersonation",
+            "email": "untrusted-alias@example.com",
+        }
+    )
+    assert isinstance(result, ManagedAgentContext)
+    assert result.mode == "delegated"
+    assert result.user_id == "canonical-user"
+    humans.find_unique.assert_awaited_once_with(
+        where={"issuer_tenant_id_oid": {"issuer": ISSUER, "tenant_id": TENANT, "oid": HUMAN}}
+    )
+
+
+@pytest.mark.asyncio
+async def test_rebinding_during_authentication_does_not_mark_new_identity_verified() -> None:
+    store, _, identities, _ = setup_store()
+    identities.update_many.return_value = 0
+    context: Final = ManagedAgentContext(agent_id="agent-one", binding_revision="old-revision", mode="autonomous")
+    result: Final = await store.record_authentication(context)
+    assert isinstance(result, AgentIdentityFailure)
+    assert "changed" in result.message
+    assert identities.update_many.call_args.kwargs["where"] == {"agent_id": "agent-one", "revision": "old-revision"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("agent", [None, stored_agent(identity=None), stored_agent(identity_managed=False)])
+async def test_stale_binding_cannot_bypass_lifecycle(agent: AgentResponse | None) -> None:
+    store, _, _, _ = setup_store(agent=agent)
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), AgentIdentityFailure)
+
+
+@pytest.mark.asyncio
+async def test_unrelated_non_entra_claims_do_not_query_identity_store() -> None:
+    store, agents, identities, _ = setup_store()
+    assert await store.resolve_verified_claims({"sub": "ordinary-user"}) is None
+    identities.find_unique.assert_not_awaited()
+    agents.find_unique.assert_not_awaited()
+
+
+HUMAN_CLAIMS: Final = {"iss": ISSUER, "tid": TENANT, "azp": CLIENT, "oid": HUMAN, "scp": "user_impersonation"}
+
+
+@pytest.mark.asyncio
+async def test_bound_agents_and_policy_failures_are_never_served_from_the_miss_cache() -> None:
+    store, _, identities, _ = setup_store()
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), ManagedAgentContext)
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), ManagedAgentContext)
+    assert identities.find_unique.await_count == 2
+    identities.find_unique.side_effect = ConnectionError("database down")
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), AgentIdentityFailure)
+    assert isinstance(await store.resolve_verified_claims(CLAIMS), AgentIdentityFailure)
+    assert identities.find_unique.await_count == 4
+
+
+@pytest.mark.asyncio
+async def test_retired_client_cannot_fall_back_to_ordinary_user_authentication() -> None:
+    from prisma.models import LiteLLM_RetiredAgentIdentity
+
+    from litellm.repositories.table_repositories import RetiredAgentIdentityRepository
+
+    identities: Final = AsyncMock()
+    identities.find_unique.return_value = None
+    retired: Final = AsyncMock()
+    retired.find_unique.return_value = LiteLLM_RetiredAgentIdentity(
+        binding_id="retired",
+        agent_id="agent-one",
+        provider="microsoft_entra",
+        issuer=ISSUER,
+        tenant_id=TENANT,
+        client_id=CLIENT,
+    )
+    db: Final = SimpleNamespace(
+        db=SimpleNamespace(
+            litellm_agentidentity=identities,
+            litellm_retiredagentidentity=retired,
+            litellm_agentstable=AsyncMock(),
+            litellm_verifiedsubject=AsyncMock(),
+        )
+    )
+    store: Final = AgentIdentityStore(
+        AgentsRepository(db),
+        AgentIdentityRepository(db),
+        VerifiedSubjectRepository(db),
+        RetiredAgentIdentityRepository(db),
+    )
+    result: Final = await store.resolve_verified_claims({**CLAIMS, "oid": HUMAN, "scp": "user_impersonation"})
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "identity_denied"
+    assert "retired" in result.message
+
+
+@pytest.mark.asyncio
+async def test_missing_revision_cannot_create_entra_authentication_evidence() -> None:
+    store, _, identities, _ = setup_store()
+    result: Final = await store.record_authentication(ManagedAgentContext(agent_id="agent-one", mode="autonomous"))
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "identity_denied"
+    identities.update_many.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_authentication_evidence_write_failure_is_not_success() -> None:
+    store, _, identities, _ = setup_store()
+    identities.update_many.side_effect = RuntimeError("writer unavailable")
+    result: Final = await store.record_authentication(
+        ManagedAgentContext(agent_id="agent-one", binding_revision="revision-one", mode="autonomous")
+    )
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "policy_unavailable"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("unavailable", [True, False])
+async def test_retired_binding_denies_and_history_outage_cannot_become_legacy_fallback(unavailable: bool) -> None:
+
+    database: Final = MagicMock()
+    database.writer_db.litellm_agentidentity.find_unique = AsyncMock(return_value=None)
+    database.writer_db.litellm_verifiedsubject.find_unique = AsyncMock(return_value=None)
+    database.writer_db.litellm_retiredagentidentity.find_unique = AsyncMock(
+        return_value={"client_id": CLIENT}, side_effect=RuntimeError("unavailable") if unavailable else None
+    )
+    result: Final = await AgentIdentityStore.from_client(database).resolve_verified_claims(CLAIMS)
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == ("policy_unavailable" if unavailable else "identity_denied")
+    assert result.message == (
+        "Retired agent identity could not be checked" if unavailable else "This agent identity binding has been retired"
+    )
+
+
+@pytest.mark.asyncio
+async def test_new_binding_is_enforced_after_another_worker_commits_it() -> None:
+    _, agents, identities, humans = setup_store()
+    identities.find_unique.return_value = None
+    retired: Final = AsyncMock()
+    retired.find_unique.return_value = None
+    db: Final = SimpleNamespace(
+        writer_db=SimpleNamespace(
+            litellm_agentstable=agents,
+            litellm_agentidentity=identities,
+            litellm_verifiedsubject=humans,
+            litellm_retiredagentidentity=retired,
+            litellm_retiredagent=retired,
+        )
+    )
+    worker: Final = AgentIdentityStore.from_client(db)
+    claims: Final = {**CLAIMS, "oid": "55555555-5555-4555-8555-555555555555"}
+    assert await worker.resolve_verified_claims(claims) is None
+    identities.find_unique.return_value = BINDING
+    denied: Final = await worker.resolve_verified_claims(claims)
+    assert isinstance(denied, AgentIdentityFailure)
+    assert denied.code == "identity_denied"
+    assert "Application token contradicts" in denied.message
+
+
+@pytest.mark.asyncio
+async def test_non_string_subject_does_not_query_directory_ownership() -> None:
+    store, _, _, humans = setup_store()
+    assert await store.subject(ISSUER, TENANT, None) is None
+    humans.find_unique.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("configured", [False, True])
+async def test_missing_or_unavailable_retirement_history_fails_closed(configured: bool) -> None:
+    database: Final = MagicMock()
+    database.writer_db.litellm_retiredagent.find_unique = AsyncMock(side_effect=RuntimeError("history unavailable"))
+    store: Final = AgentIdentityStore.from_client(database) if configured else setup_store()[0]
+    result: Final = await store.retired_agent("deleted-agent")
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "policy_unavailable"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("owner", ["canonical-user", "another-user"])
+async def test_interactive_enrollment_preserves_existing_subject_ownership(owner: str) -> None:
+    store, _, _, humans = setup_store()
+    humans.upsert.return_value = LiteLLM_VerifiedSubject(
+        subject_id="subject-one",
+        issuer=ISSUER,
+        tenant_id=TENANT,
+        oid=HUMAN,
+        user_id=owner,
+        kind="human",
+        verified_via="sso_interactive",
+        verified_at=datetime.now(timezone.utc),
+    )
+    result: Final = await store.enroll_interactive_human(
+        MicrosoftInteractiveSubject(issuer=ISSUER, tenant_id=TENANT, oid=HUMAN), "canonical-user"
+    )
+    if owner == "canonical-user":
+        assert result is None
+    else:
+        assert isinstance(result, AgentIdentityFailure)
+        assert result.code == "identity_denied"
+    assert humans.upsert.call_args.kwargs["data"]["update"] == {}
+    assert humans.upsert.call_args.kwargs["data"]["create"]["user_id"] == "canonical-user"
+
+
+@pytest.mark.asyncio
+async def test_interactive_enrollment_outage_fails_closed() -> None:
+    store, _, _, humans = setup_store()
+    humans.upsert.side_effect = ConnectionError("writer unavailable")
+    result: Final = await store.enroll_interactive_human(
+        MicrosoftInteractiveSubject(issuer=ISSUER, tenant_id=TENANT, oid=HUMAN), "canonical-user"
+    )
+    assert isinstance(result, AgentIdentityFailure)
+    assert result.code == "policy_unavailable"
+
+
+@pytest.mark.asyncio
+async def test_matching_revision_records_successful_authentication() -> None:
+    store, _, identities, _ = setup_store()
+    assert (
+        await store.record_authentication(
+            ManagedAgentContext(agent_id="agent-one", binding_revision="revision-one", mode="autonomous")
+        )
+        is None
+    )
+    identities.update_many.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("outage", [False, True])
+async def test_resolver_maps_denials_and_outages_to_public_errors(outage: bool) -> None:
+    database: Final = MagicMock()
+    database.writer_db.litellm_agentidentity.find_unique = AsyncMock(
+        return_value=BINDING, side_effect=ConnectionError("unavailable") if outage else None
+    )
+    database.writer_db.litellm_verifiedsubject.find_unique = AsyncMock(return_value=None)
+    database.writer_db.litellm_agentstable.find_unique = AsyncMock(return_value=stored_agent(enabled=False))
+    with pytest.raises(HTTPException) as exc:
+        await resolve_managed_agent(CLAIMS, database)
+    assert exc.value.status_code == (503 if outage else 403)
+
+
+@pytest.mark.asyncio
+async def test_resolver_preserves_unconfigured_and_unrelated_authentication() -> None:
+    assert await resolve_managed_agent(CLAIMS, None) is None
+    assert await resolve_managed_agent({"sub": "ordinary-user"}, MagicMock()) is None
+    store, _, identities, _ = setup_store()
+    identities.find_unique.return_value = None
+    assert await store.resolve_verified_claims(CLAIMS) is None

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -24297,6 +24297,45 @@ export interface components {
             /** Uri */
             uri?: string;
         };
+        /** AgentIdentityBinding */
+        AgentIdentityBinding: {
+            /**
+             * Active
+             * @default true
+             */
+            active: boolean;
+            /** Agent Id */
+            agent_id: string;
+            /** Client Id */
+            client_id: string;
+            /** Issuer */
+            issuer: string;
+            /** Last Authenticated At */
+            last_authenticated_at?: string | null;
+            /**
+             * Provider
+             * @constant
+             */
+            provider: "microsoft_entra";
+            /**
+             * Required Roles
+             * @default []
+             */
+            required_roles: string[];
+            /**
+             * Required Scopes
+             * @default [
+             *       "user_impersonation"
+             *     ]
+             */
+            required_scopes: string[];
+            /** Revision */
+            revision: string;
+            /** Service Principal Id */
+            service_principal_id?: string | null;
+            /** Tenant Id */
+            tenant_id: string;
+        };
         /**
          * AgentInterface
          * @description Declares a combination of a target URL and a transport protocol.
@@ -24452,8 +24491,30 @@ export interface components {
             created_at?: string | null;
             /** Created By */
             created_by?: string | null;
+            /**
+             * Enabled
+             * @default true
+             */
+            enabled: boolean;
+            /**
+             * Execution Mode
+             * @default autonomous
+             * @enum {string}
+             */
+            execution_mode: "autonomous" | "delegated" | "both";
             /** Extra Headers */
             extra_headers?: string[] | null;
+            identity?: components["schemas"]["AgentIdentityBinding"] | null;
+            /**
+             * Identity Managed
+             * @default false
+             */
+            identity_managed: boolean;
+            /**
+             * Jwt Auth Configured
+             * @default false
+             */
+            jwt_auth_configured: boolean;
             /** Keys */
             keys?: components["schemas"]["AgentKeySummary"][] | null;
             kill_switch?: components["schemas"]["AgentKillSwitchConfig"] | null;