diff --git a/app/core/api_token.py b/app/core/api_token.py
new file mode 100644
index 0000000..e162966
--- /dev/null
+++ b/app/core/api_token.py
@@ -0,0 +1,180 @@
+"""API Token Service — create, verify, revoke, list Bearer tokens.
+
+Uses ApiToken model with token_hash (SHA-256). Tokens are shown once at creation
+and never stored in plaintext. Verification hashes the incoming token and
+matches against the database.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import secrets
+import uuid
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import select, update, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.auth import ApiToken
+from app.models.user import User, UserTenant
+
+
+def _hash_token(token: str) -> str:
+    """Hash a plaintext token with SHA-256."""
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+def _generate_token() -> str:
+    """Generate a secure random token (URL-safe, 32 bytes)."""
+    return secrets.token_urlsafe(32)
+
+
+async def create_api_token(
+    db: AsyncSession,
+    tenant_id: uuid.UUID,
+    user_id: uuid.UUID,
+    name: str,
+    scopes: list[str] | None = None,
+    expires_at: datetime | None = None,
+) -> dict[str, Any]:
+    """Create a new API token. Returns the plaintext token ONCE."""
+    plaintext = _generate_token()
+    token_hash = _hash_token(plaintext)
+
+    token = ApiToken(
+        tenant_id=tenant_id,
+        user_id=user_id,
+        token_hash=token_hash,
+        name=name,
+        scopes=scopes or [],
+        expires_at=expires_at,
+    )
+    db.add(token)
+    await db.flush()
+    await db.refresh(token)
+
+    return {
+        "id": str(token.id),
+        "token": plaintext,  # Only returned once at creation
+        "name": token.name,
+        "scopes": token.scopes,
+        "expires_at": token.expires_at.isoformat() if token.expires_at else None,
+        "created_at": token.created_at.isoformat() if token.created_at else None,
+    }
+
+
+async def verify_api_token(
+    db: AsyncSession, token: str
+) -> dict[str, Any] | None:
+    """Verify a Bearer token. Returns user context dict or None.
+
+    Checks:
+    - Token hash matches a database record
+    - Token is not revoked (revoked_at is NULL)
+    - Token is not expired (expires_at is NULL or in the future)
+    - User is active
+    - User has an active membership in the token's tenant
+    """
+    token_hash = _hash_token(token)
+
+    q = select(ApiToken).where(
+        ApiToken.token_hash == token_hash,
+        ApiToken.revoked_at.is_(None),
+    )
+    result = await db.execute(q)
+    api_token = result.scalar_one_or_none()
+
+    if api_token is None:
+        return None
+
+    # Check expiry
+    now = datetime.now(UTC)
+    if api_token.expires_at is not None:
+        expires_at = api_token.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=UTC)
+        if now > expires_at:
+            return None
+
+    # Load user
+    user_q = select(User).where(User.id == api_token.user_id, User.is_active == True)  # noqa: E712
+    user_result = await db.execute(user_q)
+    user = user_result.scalar_one_or_none()
+    if user is None:
+        return None
+
+    # Check active membership
+    ut_q = select(UserTenant).where(
+        UserTenant.user_id == user.id,
+        UserTenant.tenant_id == api_token.tenant_id,
+        UserTenant.status == "active",
+    )
+    ut_result = await db.execute(ut_q)
+    ut = ut_result.scalar_one_or_none()
+    if ut is None:
+        return None
+
+    # Update last_used_at (non-blocking)
+    await db.execute(
+        update(ApiToken)
+        .where(ApiToken.id == api_token.id)
+        .values(last_used_at=now)
+    )
+    await db.flush()
+
+    # Build user context dict (same shape as get_current_user)
+    return {
+        "user_id": str(user.id),
+        "tenant_id": str(api_token.tenant_id),
+        "email": user.email,
+        "name": user.name,
+        "role": ut.role,
+        "is_system_admin": user.is_system_admin,
+        "permissions": [],  # Loaded by require_permission if needed
+        "_auth_method": "api_token",
+        "_token_id": str(api_token.id),
+        "_token_scopes": api_token.scopes or [],
+    }
+
+
+async def revoke_api_token(
+    db: AsyncSession, tenant_id: uuid.UUID, token_id: uuid.UUID
+) -> bool:
+    """Revoke an API token."""
+    now = datetime.now(UTC)
+    result = await db.execute(
+        update(ApiToken)
+        .where(
+            ApiToken.id == token_id,
+            ApiToken.tenant_id == tenant_id,
+            ApiToken.revoked_at.is_(None),
+        )
+        .values(revoked_at=now)
+    )
+    await db.flush()
+    return result.rowcount > 0
+
+
+async def list_api_tokens(
+    db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID
+) -> list[dict[str, Any]]:
+    """List all API tokens for a user (without token hashes)."""
+    q = select(ApiToken).where(
+        ApiToken.tenant_id == tenant_id,
+        ApiToken.user_id == user_id,
+        ApiToken.revoked_at.is_(None),
+    ).order_by(ApiToken.created_at.desc())
+    result = await db.execute(q)
+    tokens = result.scalars().all()
+    return [
+        {
+            "id": str(t.id),
+            "name": t.name,
+            "scopes": t.scopes or [],
+            "expires_at": t.expires_at.isoformat() if t.expires_at else None,
+            "last_used_at": t.last_used_at.isoformat() if t.last_used_at else None,
+            "created_at": t.created_at.isoformat() if t.created_at else None,
+        }
+        for t in tokens
+    ]
diff --git a/app/core/delegation_token.py b/app/core/delegation_token.py
new file mode 100644
index 0000000..f2de3e6
--- /dev/null
+++ b/app/core/delegation_token.py
@@ -0,0 +1,93 @@
+"""Delegation Token Service — HMAC-signed short-lived tokens for internal AI calls.
+
+Tokens are signed with the app SECRET_KEY using HMAC-SHA256.
+Max lifetime: 60 seconds. No persistent storage — stateless verification.
+
+Token format: base64(payload).base64(signature)
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import uuid
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from app.config import get_settings
+
+
+DELEGATION_AUDIENCE = "internal-ai-delegation"
+MAX_TOKEN_LIFETIME = 60  # seconds
+
+
+def _get_secret() -> bytes:
+    """Get the signing secret from app settings."""
+    return get_settings().secret_key.encode()
+
+
+def _sign(payload: dict) -> str:
+    """Sign payload with HMAC-SHA256 and return base64(payload).base64(sig)."""
+    payload_bytes = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+    sig = hmac.new(_get_secret(), payload_bytes, hashlib.sha256).digest()
+    return f"{base64.b64encode(payload_bytes).decode()}.{base64.b64encode(sig).decode()}"
+
+
+def _verify(token: str) -> dict | None:
+    """Verify a delegation token. Returns payload dict or None."""
+    try:
+        payload_b64, sig_b64 = token.rsplit(".", 1)
+        payload_bytes = base64.b64decode(payload_b64)
+        expected_sig = hmac.new(_get_secret(), payload_bytes, hashlib.sha256).digest()
+        actual_sig = base64.b64decode(sig_b64)
+        if not hmac.compare_digest(expected_sig, actual_sig):
+            return None
+        payload = json.loads(payload_bytes)
+        # Check expiry
+        now = datetime.now(UTC)
+        expires_at = datetime.fromisoformat(payload["expires_at"])
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=UTC)
+        if now > expires_at:
+            return None
+        # Check audience
+        if payload.get("audience") != DELEGATION_AUDIENCE:
+            return None
+        return payload
+    except Exception:
+        return None
+
+
+def create_delegation_token(
+    user_id: str,
+    tenant_id: str,
+    agent_id: str = "ai-copilot",
+    lifetime_seconds: int = MAX_TOKEN_LIFETIME,
+) -> str:
+    """Create a short-lived delegation token for an internal AI call.
+
+    The token contains:
+    - user_id, tenant_id: who the AI acts on behalf of
+    - agent_id: which agent/service is calling
+    - audience: fixed to internal-ai-delegation
+    - expires_at: max 60 seconds from now
+    - token_id: unique ID for audit tracing
+    """
+    now = datetime.now(UTC)
+    expires_at = now + timedelta(seconds=min(lifetime_seconds, MAX_TOKEN_LIFETIME))
+    payload = {
+        "user_id": user_id,
+        "tenant_id": tenant_id,
+        "agent_id": agent_id,
+        "audience": DELEGATION_AUDIENCE,
+        "expires_at": expires_at.isoformat(),
+        "token_id": str(uuid.uuid4()),
+    }
+    return _sign(payload)
+
+
+def verify_delegation_token(token: str) -> dict[str, Any] | None:
+    """Verify a delegation token. Returns payload or None if invalid/expired."""
+    return _verify(token)
diff --git a/app/deps.py b/app/deps.py
index 4c46bea..5490890 100644
--- a/app/deps.py
+++ b/app/deps.py
@@ -154,6 +154,76 @@ async def get_current_user(
     return session_data
 
 
+async def get_current_user_bearer(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+) -> dict[str, Any]:
+    """Get the current user from a Bearer API token.
+
+    Alternative to session-based auth for programmatic access (MCP, API clients).
+    Returns the same dict shape as get_current_user.
+    """
+    auth_header = request.headers.get("Authorization", "")
+    if not auth_header.startswith("Bearer "):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail={"detail": "Bearer token required", "code": "not_authenticated"},
+        )
+
+    token = auth_header[7:]  # Strip "Bearer "
+    from app.core.api_token import verify_api_token
+
+    user_data = await verify_api_token(db, token)
+    if user_data is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail={"detail": "Invalid or expired token", "code": "token_invalid"},
+        )
+
+    # Set RLS tenant context
+    tenant_id = uuid.UUID(user_data["tenant_id"])
+    await set_tenant_context(db, tenant_id)
+
+    # Set RLS user context
+    user_id = uuid.UUID(user_data["user_id"])
+    from app.models.group import UserGroup
+    groups_q = await db.execute(
+        select(UserGroup.group_id)
+        .where(UserGroup.user_id == user_id)
+        .where(UserGroup.tenant_id == tenant_id)
+    )
+    group_ids = [row[0] for row in groups_q]
+    is_admin = user_data.get("is_system_admin", False)
+    await set_user_context(db, user_id, group_ids, is_admin)
+
+    # Load resolved permissions
+    from app.core.permissions import get_cached_permissions
+    redis = get_redis()
+    resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
+    user_data["permissions"] = resolved.get("permissions", [])
+    user_data["denied_permissions"] = resolved.get("denied", [])
+    user_data["field_permissions"] = resolved.get("field_permissions", {})
+    user_data["is_system_admin"] = resolved.get("is_system_admin", False)
+
+    return user_data
+
+
+async def get_current_user_or_bearer(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    redis: aioredis.Redis = Depends(get_redis_dep),
+) -> dict[str, Any]:
+    """Get current user from session cookie OR Bearer token.
+
+    Tries session auth first, falls back to Bearer token.
+    Used by MCP routes that accept both auth methods.
+    """
+    auth_header = request.headers.get("Authorization", "")
+    if auth_header.startswith("Bearer "):
+        return await get_current_user_bearer(request, db)
+    return await get_current_user(request, db, redis)
+
+
 async def require_admin(
     current_user: dict[str, Any] = Depends(get_current_user),
 ) -> dict[str, Any]:
diff --git a/app/main.py b/app/main.py
index 43d1485..bdfe0cd 100644
--- a/app/main.py
+++ b/app/main.py
@@ -70,6 +70,7 @@ from app.routes import (
     guest_auth,
     guests,
     outbox,
+    api_tokens,
 )
 
 
@@ -445,6 +446,7 @@ def create_app() -> FastAPI:
     app.include_router(guests.router)
     app.include_router(workspaces.router)
     app.include_router(outbox.router)
+    app.include_router(api_tokens.router)
 
     # ── Register plugin routes for all built-in plugins ──
     # Routes are registered at app creation time so OpenAPI docs are complete.
diff --git a/app/plugins/builtins/mcp_server/routes.py b/app/plugins/builtins/mcp_server/routes.py
index bb9084f..20cc5be 100644
--- a/app/plugins/builtins/mcp_server/routes.py
+++ b/app/plugins/builtins/mcp_server/routes.py
@@ -10,7 +10,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.db import get_db
-from app.deps import get_current_user, require_permission
+from app.deps import get_current_user, get_current_user_or_bearer, require_permission
 from app.plugins.builtins.mcp_server.schemas import (
     McpServerConfig,
     McpToolDefinition,
@@ -58,10 +58,11 @@ async def execute_mcp_tool(
     tool_name: str,
     request: McpToolExecuteRequest,
     db: AsyncSession = Depends(get_db),
-    current_user: dict[str, Any] = Depends(get_current_user),
+    current_user: dict[str, Any] = Depends(get_current_user_or_bearer),
 ) -> McpToolExecuteResponse:
     """Execute an MCP tool by name with provided arguments.
 
+    Accepts session cookie OR Bearer token (for programmatic access).
     Requires mcp:read for read tools, mcp:write for write tools.
     """
     tool_def = get_tool_definition(tool_name)
@@ -93,8 +94,23 @@ async def execute_mcp_tool(
         "user_id": current_user.get("user_id"),
         "role": current_user.get("role"),
         "permissions": current_user.get("permissions", []),
+        "auth_method": current_user.get("_auth_method", "session"),
     }
 
+    # Audit log
+    from app.core.audit import log_audit
+    import uuid as uuid_mod
+    correlation_id = str(uuid_mod.uuid4())
+    await log_audit(
+        db,
+        tenant_id=uuid.UUID(current_user["tenant_id"]),
+        user_id=uuid.UUID(current_user["user_id"]),
+        action="mcp.tool.execute",
+        entity_type="mcp_tool",
+        entity_id=tool_name,
+        details={"tool": tool_name, "arguments": request.arguments, "correlation_id": correlation_id, "auth_method": context["auth_method"]},
+    )
+
     try:
         result = await handler(db, request.arguments, context)
         await db.commit()
diff --git a/app/routes/api_tokens.py b/app/routes/api_tokens.py
new file mode 100644
index 0000000..35cda3e
--- /dev/null
+++ b/app/routes/api_tokens.py
@@ -0,0 +1,78 @@
+"""API Token routes — create, list, revoke Bearer tokens for programmatic access."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.api_token import create_api_token, list_api_tokens, revoke_api_token
+from app.core.db import get_db
+from app.deps import get_current_user, require_permission
+
+router = APIRouter(prefix="/api/v1/tokens", tags=["api-tokens"])
+
+
+class TokenCreateRequest(BaseModel):
+    name: str
+    scopes: list[str] = []
+    expires_in_days: int | None = None
+
+
+class TokenRevokeRequest(BaseModel):
+    token_id: str
+
+
+@router.post("", status_code=status.HTTP_201_CREATED)
+async def create_token(
+    body: TokenCreateRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("mcp:write")),
+):
+    """Create a new API token. The plaintext token is returned ONCE."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+
+    expires_at = None
+    if body.expires_in_days is not None:
+        expires_at = datetime.now(timezone.utc) + timedelta(days=body.expires_in_days)
+
+    result = await create_api_token(
+        db, tenant_id, user_id, body.name, body.scopes, expires_at,
+    )
+    await db.commit()
+    return result
+
+
+@router.get("")
+async def list_tokens(
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("mcp:read")),
+):
+    """List all API tokens for the current user (without token hashes)."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    user_id = uuid.UUID(current_user["user_id"])
+    tokens = await list_api_tokens(db, tenant_id, user_id)
+    return {"items": tokens, "total": len(tokens)}
+
+
+@router.delete("/{token_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def revoke_token(
+    token_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_permission("mcp:write")),
+):
+    """Revoke an API token."""
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+    try:
+        tid = uuid.UUID(token_id)
+    except ValueError:
+        raise HTTPException(400, detail={"detail": "Invalid token_id", "code": "invalid_id"})
+    revoked = await revoke_api_token(db, tenant_id, tid)
+    if not revoked:
+        raise HTTPException(404, detail={"detail": "Token not found or already revoked", "code": "not_found"})
+    await db.commit()
diff --git a/tests/test_api_tokens.py b/tests/test_api_tokens.py
new file mode 100644
index 0000000..770b0ff
--- /dev/null
+++ b/tests/test_api_tokens.py
@@ -0,0 +1,218 @@
+"""Tests for Phase 5 — API Token Service and Delegation Token.
+
+Covers:
+- API Token: create, verify, revoke, list
+- API Token: expired token rejected
+- API Token: revoked token rejected
+- API Token: inactive user rejected
+- Delegation Token: create, verify, expiry, audience check
+- Delegation Token: tampered token rejected
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.api_token import create_api_token, verify_api_token, revoke_api_token, list_api_tokens, _hash_token
+from app.core.delegation_token import create_delegation_token, verify_delegation_token, DELEGATION_AUDIENCE
+from app.models.auth import ApiToken
+from app.models.tenant import Tenant
+from app.models.user import User, UserTenant
+
+
+async def _seed_tenant_and_user(db: AsyncSession) -> dict:
+    tenant = Tenant(name="Test Tenant", slug="test-tenant-phase5")
+    db.add(tenant)
+    await db.flush()
+    user = User(
+        email="phase5@example.com",
+        name="Phase5 User",
+        password_hash="dummy",
+        is_active=True,
+        preferences={},
+    )
+    db.add(user)
+    await db.flush()
+    ut = UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin", status="active")
+    db.add(ut)
+    await db.flush()
+    return {"tenant": tenant, "user": user}
+
+
+# ─── API Token Tests ──────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_create_api_token_returns_plaintext(db_session: AsyncSession):
+    """create_api_token returns the plaintext token once."""
+    seed = await _seed_tenant_and_user(db_session)
+    result = await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
+    )
+    assert "token" in result
+    assert len(result["token"]) > 20  # URL-safe token
+    assert result["name"] == "Test Token"
+
+
+@pytest.mark.asyncio
+async def test_verify_api_token_valid(db_session: AsyncSession):
+    """verify_api_token returns user context for a valid token."""
+    seed = await _seed_tenant_and_user(db_session)
+    result = await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Test Token",
+    )
+    user_data = await verify_api_token(db_session, result["token"])
+    assert user_data is not None
+    assert user_data["user_id"] == str(seed["user"].id)
+    assert user_data["tenant_id"] == str(seed["tenant"].id)
+    assert user_data["_auth_method"] == "api_token"
+
+
+@pytest.mark.asyncio
+async def test_verify_api_token_invalid(db_session: AsyncSession):
+    """verify_api_token returns None for an invalid token."""
+    user_data = await verify_api_token(db_session, "invalid-token-string")
+    assert user_data is None
+
+
+@pytest.mark.asyncio
+async def test_revoke_api_token(db_session: AsyncSession):
+    """revoked tokens are rejected by verify_api_token."""
+    seed = await _seed_tenant_and_user(db_session)
+    result = await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="To Revoke",
+    )
+    token_id = uuid.UUID(result["id"])
+    revoked = await revoke_api_token(db_session, seed["tenant"].id, token_id)
+    assert revoked is True
+    # Token should no longer verify
+    user_data = await verify_api_token(db_session, result["token"])
+    assert user_data is None
+
+
+@pytest.mark.asyncio
+async def test_verify_api_token_expired(db_session: AsyncSession):
+    """expired tokens are rejected."""
+    seed = await _seed_tenant_and_user(db_session)
+    expires_at = datetime.now(UTC) - timedelta(seconds=1)  # Already expired
+    result = await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Expired",
+        expires_at=expires_at,
+    )
+    user_data = await verify_api_token(db_session, result["token"])
+    assert user_data is None
+
+
+@pytest.mark.asyncio
+async def test_list_api_tokens(db_session: AsyncSession):
+    """list_api_tokens returns tokens without hashes."""
+    seed = await _seed_tenant_and_user(db_session)
+    await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Token 1",
+    )
+    await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Token 2",
+    )
+    tokens = await list_api_tokens(db_session, seed["tenant"].id, seed["user"].id)
+    assert len(tokens) == 2
+    assert "token" not in tokens[0]  # No plaintext in list
+    assert "token_hash" not in tokens[0]  # No hash in list
+
+
+@pytest.mark.asyncio
+async def test_verify_api_token_inactive_user(db_session: AsyncSession):
+    """inactive users are rejected."""
+    seed = await _seed_tenant_and_user(db_session)
+    # Deactivate user
+    seed["user"].is_active = False
+    await db_session.flush()
+    result = await create_api_token(
+        db_session, seed["tenant"].id, seed["user"].id, name="Inactive User",
+    )
+    user_data = await verify_api_token(db_session, result["token"])
+    assert user_data is None
+
+
+# ─── Delegation Token Tests ──────────────────────────────────
+
+
+def test_create_delegation_token_returns_string():
+    """create_delegation_token returns a signed string."""
+    token = create_delegation_token(
+        user_id="user-123", tenant_id="tenant-456",
+    )
+    assert isinstance(token, str)
+    assert "." in token  # payload.signature format
+
+
+def test_verify_delegation_token_valid():
+    """verify_delegation_token returns payload for a valid token."""
+    token = create_delegation_token(
+        user_id="user-123", tenant_id="tenant-456",
+    )
+    payload = verify_delegation_token(token)
+    assert payload is not None
+    assert payload["user_id"] == "user-123"
+    assert payload["tenant_id"] == "tenant-456"
+    assert payload["audience"] == DELEGATION_AUDIENCE
+    assert "expires_at" in payload
+    assert "token_id" in payload
+
+
+def test_verify_delegation_token_invalid():
+    """verify_delegation_token returns None for invalid token."""
+    payload = verify_delegation_token("invalid.token")
+    assert payload is None
+
+
+def test_verify_delegation_token_tampered():
+    """tampered tokens are rejected."""
+    token = create_delegation_token(
+        user_id="user-123", tenant_id="tenant-456",
+    )
+    # Tamper with the payload part
+    parts = token.split(".")
+    tampered = parts[0] + "x." + parts[1]
+    payload = verify_delegation_token(tampered)
+    assert payload is None
+
+
+def test_verify_delegation_token_wrong_audience():
+    """tokens with wrong audience are rejected."""
+    from app.core.delegation_token import _sign
+    import json
+    from datetime import UTC, datetime, timedelta
+    import uuid as uuid_mod
+
+    now = datetime.now(UTC)
+    payload = {
+        "user_id": "user-123",
+        "tenant_id": "tenant-456",
+        "agent_id": "test",
+        "audience": "wrong-audience",
+        "expires_at": (now + timedelta(seconds=30)).isoformat(),
+        "token_id": str(uuid_mod.uuid4()),
+    }
+    token = _sign(payload)
+    result = verify_delegation_token(token)
+    assert result is None
+
+
+def test_delegation_token_max_lifetime():
+    """token lifetime is capped at MAX_TOKEN_LIFETIME."""
+    from app.core.delegation_token import MAX_TOKEN_LIFETIME
+    token = create_delegation_token(
+        user_id="user-123", tenant_id="tenant-456",
+        lifetime_seconds=3600,  # Request 1 hour
+    )
+    payload = verify_delegation_token(token)
+    assert payload is not None
+    # Should be capped at 60 seconds
+    expires_at = datetime.fromisoformat(payload["expires_at"])
+    now = datetime.now(UTC)
+    lifetime = (expires_at - now).total_seconds()
+    assert lifetime <= MAX_TOKEN_LIFETIME + 5  # Allow small timing variance