diff --git a/flocks/server/__init__.py b/flocks/server/__init__.py
index e1afde45d..96f46afc7 100644
--- a/flocks/server/__init__.py
+++ b/flocks/server/__init__.py
@@ -1 +1,22 @@
-"""Server module for HTTP API"""
+"""Server module for HTTP API.
+
+Startup hook (env-gated, zero core change):
+    If ``FLOCKS_AUTH=workshop_jwt`` is set, register the Workshop auth
+    backend + adapter on import. Keeps zero changes for deployments that
+    don't use Workshop.
+"""
+
+import os
+
+if os.getenv("FLOCKS_AUTH") == "workshop_jwt":
+    try:
+        from flocks.workshop_auth import register_workshop_auth
+
+        register_workshop_auth()
+    except Exception as _exc:  # noqa: BLE001
+        # Never crash server startup due to optional auth plugin.
+        # Operators can see the failure via logs/metrics; defaulting to
+        # LocalAuthBackend keeps the deployment usable.
+        import sys
+
+        print(f"[server] failed to register workshop auth: {_exc!r}", file=sys.stderr)
\ No newline at end of file
diff --git a/flocks/workshop_auth/__init__.py b/flocks/workshop_auth/__init__.py
new file mode 100644
index 000000000..66d5e8236
--- /dev/null
+++ b/flocks/workshop_auth/__init__.py
@@ -0,0 +1,50 @@
+"""Workshop auth plugin (PoC, v1.1 route B+).
+
+Registers a JWT-based AuthBackend so the Workshop facade can call Flocks
+with ``Cookie: flocks_session=<per-team JWT>`` and get a real
+``AuthUser(tenant_ids=[...])`` — real tenant isolation via
+``flocks/contracts/access``.
+
+Production mount (env-gated, zero core changes):
+    Set ``FLOCKS_AUTH=workshop_jwt`` and Flocks will call
+    ``register_workshop_auth()`` on startup. PoC reproducer and tests
+    call it directly.
+
+    Add to ``flocks/server/__init__.py`` startup path::
+
+        if os.getenv("FLOCKS_AUTH") == "workshop_jwt":
+            from flocks.workshop_auth import register_workshop_auth
+            register_workshop_auth()
+"""
+
+import os
+
+from flocks.workshop_auth.backend import TeamJWTAuthBackend
+from flocks.workshop_auth.adapter import register_workshop_adapters
+
+__all__ = ["TeamJWTAuthBackend", "register_workshop_auth", "maybe_register_on_env"]
+
+
+def register_workshop_auth() -> None:
+    """Register the auth backend AND the request-context adapter.
+
+    Idempotent: re-registration is safe — Flocks' register_backend
+    overwrites the previous backend and register_auth_context_adapter
+    is dict-assignment.
+    """
+    from flocks.auth.service import AuthService
+
+    AuthService.register_backend(TeamJWTAuthBackend)
+    register_workshop_adapters()
+
+
+def maybe_register_on_env(env_var: str = "FLOCKS_AUTH", expected: str = "workshop_jwt") -> bool:
+    """Conditionally register based on env. Returns True if registered.
+
+    Designed to be called from the Flocks startup hook. Keeps zero changes
+    to the core server when the env is not set.
+    """
+    if os.getenv(env_var) == expected:
+        register_workshop_auth()
+        return True
+    return False
diff --git a/flocks/workshop_auth/adapter.py b/flocks/workshop_auth/adapter.py
new file mode 100644
index 000000000..8ff47180b
--- /dev/null
+++ b/flocks/workshop_auth/adapter.py
@@ -0,0 +1,99 @@
+"""Workshop permission adapter (PoC, v1.1 route B+).
+
+Injects Workshop-side RBAC permissions into ``request.state.extension_context``
+after auth resolution, so any downstream code can read
+``request.state.extension_context["wk_permissions"]``.
+
+Mount:
+    The adapter is registered automatically when
+    ``flocks.workshop_auth.register_workshop_auth()`` runs. See ``__init__.py``.
+
+Real signature (verified against ``flocks/server/auth.py`` L28-33)::
+
+    AuthContextAdapter = Callable[
+        [HTTPConnection, AuthUser | None],
+        Awaitable[Mapping[str, Any] | None],
+    ]
+
+We DO NOT re-verify the JWT here — the auth backend has already decoded it
+on the same cookie in ``get_user_by_session_id``. We only unwrap the claims
+to surface permissions for RBAC checks. Re-verification would cost a JWKS
+round-trip per request.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from starlette.requests import HTTPConnection
+
+from flocks.auth.context import AuthUser
+
+_PERMISSION_KEY = "wk_permissions"
+_USER_ID_KEY = "wk_user_id"
+_TENANT_IDS_KEY = "wk_tenant_ids"
+
+
+async def inject_workshop_context(
+    request: HTTPConnection,
+    user: AuthUser | None,
+) -> Mapping[str, Any] | None:
+    """Surface Workshop claims into request.state.extension_context.
+
+    Returns None when:
+      - the request is not a Workshop-served request (no flocks_session cookie
+        OR the cookie is not a JWT we issued)
+      - the user object is missing tenant_ids (e.g. LocalAuthBackend admin)
+
+    Otherwise returns a dict that ``apply_auth_for_request`` will merge into
+    ``request.state.extension_context``.
+    """
+    if user is None:
+        return None
+    if user.role != "member":
+        # admin / service-user sessions do not carry tenant scope
+        # (PolicyContextResolver returns empty context for them, §10-C0 ③)
+        return None
+
+    cookie = request.cookies.get("flocks_session")
+    if not cookie:
+        return None
+
+    claims = _safe_decode(cookie)
+    if claims is None:
+        return None
+
+    permissions = claims.get("permissions") or []
+    teams = claims.get("teams") or ()
+    return {
+        _PERMISSION_KEY: list(permissions),
+        _USER_ID_KEY: str(claims.get("sub", user.id)),
+        _TENANT_IDS_KEY: tuple(teams),
+    }
+
+
+def _safe_decode(token: str) -> dict | None:
+    """Decode WITHOUT re-verifying signature.
+
+    The auth backend decoded & verified this exact token moments ago in
+    ``get_user_by_session_id``. Re-verification would require a JWKS fetch
+    on every request and gain no security (the cookie is HttpOnly + same
+    origin). We only trust the value because the backend just trusted it.
+
+    For paranoia in untrusted deployments, swap this for a JWKS verify —
+    the perf cost is one cached HTTP call per kid.
+    """
+    try:
+        import jwt as pyjwt
+
+        return pyjwt.decode(token, options={"verify_signature": False})
+    except Exception:
+        return None
+
+
+def register_workshop_adapters() -> None:
+    """Register the Workshop context adapter under name ``wk_rbac``."""
+    from flocks.server.auth import register_auth_context_adapter
+
+    register_auth_context_adapter("wk_rbac", inject_workshop_context)
\ No newline at end of file
diff --git a/flocks/workshop_auth/backend.py b/flocks/workshop_auth/backend.py
new file mode 100644
index 000000000..83283a997
--- /dev/null
+++ b/flocks/workshop_auth/backend.py
@@ -0,0 +1,183 @@
+"""TeamJWTAuthBackend — Workshop-issued JWT → Flocks AuthUser (PoC).
+
+Mount path (verified against flocks/server/auth.py::_apply_auth_for_request):
+the session-cookie branch is checked FIRST, before browser detection, and it
+calls ``AuthService.get_user_by_session_id(session_id)`` which delegates to
+the registered backend. The Workshop facade therefore sends::
+
+    Cookie: flocks_session=<per-team JWT>
+
+and this backend decodes the JWT and returns a LocalUser whose
+``to_auth_user()`` carries ``tenant_ids`` — the field consumed by
+``flocks/contracts/access`` PolicyContext for tenant-level filtering.
+
+PoC simplification: HS256 with a shared secret from env WORKSHOP_JWT_SECRET.
+Production: RS256/ES256 via JWKS (WORKSHOP_JWKS_URL) — see design doc §6.2.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from typing import Optional, Tuple
+
+import jwt as pyjwt
+
+from flocks.auth.context import AuthUser
+from flocks.utils.log import Log
+
+log = Log.create(service="workshop_auth")
+
+# ---------------------------------------------------------------------------
+# Algorithm selection — PoC defaults to HS256 for the lightweight standalone
+# reproducer; production should set WORKSHOP_JWT_ALG=RS256 + WORKSHOP_JWKS_URL.
+# ---------------------------------------------------------------------------
+_ALG_ENV = "WORKSHOP_JWT_ALG"
+_JWKS_URL_ENV = "WORKSHOP_JWKS_URL"
+_HS_SECRET_ENV = "WORKSHOP_JWT_SECRET"
+
+
+def _algorithm() -> str:
+    return os.getenv(_ALG_ENV, "HS256").upper()
+
+
+async def _decode(token: str) -> dict:
+    alg = _algorithm()
+    audience = "flocks"
+    options = {"verify_aud": True, "verify_exp": True}
+    if alg == "HS256":
+        secret = os.getenv(_HS_SECRET_ENV, "")
+        if not secret:
+            raise RuntimeError(f"env {_HS_SECRET_ENV} is required for HS256")
+        payload = pyjwt.decode(token, secret, algorithms=["HS256"], audience=audience, options=options)
+    elif alg in ("RS256", "RS384", "RS512", "ES256", "ES384"):
+        from flocks.workshop_auth.client import public_key_for
+
+        jwks_url = os.getenv(_JWKS_URL_ENV, "")
+        if not jwks_url:
+            raise RuntimeError(f"env {_JWKS_URL_ENV} is required for {alg}")
+        try:
+            unverified_header = pyjwt.get_unverified_header(token)
+        except pyjwt.DecodeError as e:
+            raise pyjwt.InvalidTokenError(f"malformed JWT header: {e}") from e
+        kid = unverified_header.get("kid")
+        key = await public_key_for(kid, jwks_url)
+        if key is None:
+            raise pyjwt.InvalidTokenError("unable to resolve JWKS key for token")
+        payload = pyjwt.decode(token, key, algorithms=[alg], audience=audience, options=options)
+    else:
+        raise RuntimeError(f"unsupported WORKSHOP_JWT_ALG={alg!r}")
+    if payload.get("iss") != "ai-agent-workshop":
+        raise pyjwt.InvalidIssuerError("unexpected issuer")
+    return payload
+
+
+
+
+
+class TeamJWTAuthBackend:
+    """AuthBackend implementation backed by Workshop-issued JWTs.
+
+    User authority lives in the Workshop facade; Flocks only consumes the
+    identity/tenant claims embedded in the short-lived token.
+    """
+
+    # ---- lifecycle -------------------------------------------------------
+
+    @classmethod
+    async def init(cls) -> None:
+        return None  # no Flocks-side tables; RBAC lives in Workshop
+
+    @classmethod
+    async def has_users(cls) -> bool:
+        return True  # users are managed by Workshop; never blocks bootstrap
+
+    @classmethod
+    async def get_bootstrap_status(cls) -> dict:
+        return {"admin": True, "member": True}
+
+    @classmethod
+    async def bootstrap_admin(cls, username: str, password: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    # ---- session resolution (the path the cookie branch uses) ------------
+
+    @classmethod
+    async def get_user_by_session_id(cls, session_id: str):
+        # Lazy import to avoid a cycle: LocalUser is defined in service.py.
+        from flocks.auth.service import LocalUser
+
+        # 集成契约: token 无效时返回 None(框架语义 → 401 "登录已过期"),
+        # 而非让 pyjwt 异常裸穿(全局异常处理器会包成 500, 破坏客户端
+        # 重试语义)。真实服务联调(2026-09-03)发现并修正。
+        try:
+            payload = await _decode(session_id)
+        except Exception:
+            return None
+        return LocalUser(
+            id=payload["sub"],
+            username=payload.get("username", payload["sub"]),
+            role=payload.get("flocks_role", "member"),
+            status="active",
+            must_reset_password=False,
+            tenant_ids=tuple(payload.get("teams", ())),
+            asset_groups=tuple(payload.get("asset_groups", ())),
+            created_at=payload.get("iat_iso", datetime.now(UTC).isoformat()),
+            updated_at=payload.get("iat_iso", datetime.now(UTC).isoformat()),
+            last_login_at=None,
+        )
+
+    @classmethod
+    async def revoke_session(cls, session_id: str) -> None:
+        # JWTs are short-lived and stateless; nothing to revoke Flocks-side.
+        return None
+
+    # ---- not used by the facade integration; kept for Protocol parity ----
+
+    @classmethod
+    async def login(cls, username: str, password: str, *, persist: bool = True):
+        raise NotImplementedError("login via Workshop facade only")
+
+    @classmethod
+    async def get_user_by_id(cls, user_id: str):
+        raise NotImplementedError("user directory lives in Workshop")
+
+    @classmethod
+    async def get_user_by_username(cls, username: str):
+        raise NotImplementedError("user directory lives in Workshop")
+
+    @classmethod
+    async def list_users(cls):
+        return []
+
+    @classmethod
+    async def create_user(cls, *, username: str, password: str, role: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def update_user_role(cls, *, target_user_id: str, new_role: str):
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def delete_user(cls, *, target_user_id: str) -> None:
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def change_password(cls, user: AuthUser, *, current_password: str, new_password: str) -> None:
+        raise NotImplementedError("passwords managed by Workshop")
+
+    @classmethod
+    async def set_password(cls, **kwargs) -> None:
+        raise NotImplementedError("passwords managed by Workshop")
+
+    @classmethod
+    async def generate_admin_temp_password(cls, *, username: str = "admin") -> str:
+        raise NotImplementedError("users managed by Workshop")
+
+    @classmethod
+    async def reassign_orphan_sessions(cls, admin_user_id: str, *, dry_run: bool = False):
+        return {"reassigned": 0}
+
+    @classmethod
+    async def migrate_legacy_sessions_to_admin(cls, admin_user_id: str) -> None:
+        return None
diff --git a/flocks/workshop_auth/client.py b/flocks/workshop_auth/client.py
new file mode 100644
index 000000000..7da3b1f10
--- /dev/null
+++ b/flocks/workshop_auth/client.py
@@ -0,0 +1,82 @@
+"""Workshop HTTP client — JWKS fetch with TTL cache (PoC, optional).
+
+In production, Workshop exposes ``GET /api/auth/jwks`` with the public keys
+used to sign per-team JWTs. Flocks caches the keyset for ``WORKSHOP_JWKS_TTL``
+seconds and refreshes lazily on cache miss or kid miss.
+
+The PoC backend (``backend.py``) defaults to HS256 with the shared
+``WORKSHOP_JWT_SECRET`` so it can run without a Workshop facade running.
+Switch to RS256/JWKS by setting ``WORKSHOP_JWT_ALG=RS256`` and
+``WORKSHOP_JWKS_URL=https://.../api/auth/jwks``.
+
+This client is intentionally minimal — no retries, no circuit breaker.
+Production hardening lives in the facade-side client.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+_log_prefix = "workshop_auth.client"
+
+
+class _JWKSCache:
+    def __init__(self, ttl_seconds: int = 300) -> None:
+        self._ttl = ttl_seconds
+        self._fetched_at: float = 0.0
+        self._keys: dict[str, Any] = {}
+
+    def get(self, kid: str | None) -> Any | None:
+        if not self._keys:
+            return None
+        if time.monotonic() - self._fetched_at > self._ttl:
+            return None  # expired; caller should refresh
+        if kid is None:
+            return next(iter(self._keys.values()))
+        return self._keys.get(kid)
+
+    def store(self, keys: dict[str, Any]) -> None:
+        self._keys = keys
+        self._fetched_at = time.monotonic()
+
+
+_cache = _JWKSCache()
+
+
+async def fetch_jwks(url: str, *, refresh: bool = False) -> dict[str, Any]:
+    """Fetch and cache the JWKS document. Returns the key dict keyed by kid."""
+    if not refresh:
+        cached = _cache.get(kid=None)
+        if cached is not None and time.monotonic() - _cache._fetched_at <= _cache._ttl:
+            return _cache._keys
+    try:
+        import httpx
+
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            resp = await client.get(url)
+            resp.raise_for_status()
+            doc = resp.json()
+    except Exception as exc:
+        # In PoC we don't fail loudly — the backend has a HS256 fallback path.
+        # In production, raise; the caller decides retry policy.
+        print(f"[{_log_prefix}] JWKS fetch failed: {exc!r}", flush=True)
+        return _cache._keys
+    keys = {k["kid"]: k for k in doc.get("keys", []) if "kid" in k}
+    _cache.store(keys)
+    return keys
+
+
+async def public_key_for(kid: str | None, jwks_url: str) -> Any | None:
+    """Resolve a JWK to a verify-key object usable by pyjwt."""
+    jwks = await fetch_jwks(jwks_url)
+    jwk = _cache.get(kid=kid) or (next(iter(jwks.values())) if jwks else None)
+    if jwk is None:
+        return None
+    try:
+        from jwt.algorithms import RSAAlgorithm
+
+        return RSAAlgorithm.from_jwk(jwk)
+    except Exception as exc:
+        print(f"[{_log_prefix}] JWK parse failed: {exc!r}", flush=True)
+        return None
\ No newline at end of file
diff --git a/tests/poc_tenant_isolation.py b/tests/poc_tenant_isolation.py
new file mode 100644
index 000000000..5f458c2e2
--- /dev/null
+++ b/tests/poc_tenant_isolation.py
@@ -0,0 +1,127 @@
+"""C0 PoC-4: contracts/access tenant filtering end-to-end.
+
+Validates the REAL policy chain:
+  AuthUser(tenant_ids) -> PolicyContextResolver -> PolicyPlanCompiler
+    -> Predicate(tenant_id IN tenant_ids, enforcement="driver-required")
+
+i.e. a team_A principal cannot obtain team_B rows through WebUI contract
+operations — the filter is compiled into the driver plan, not applied in
+frontend params.
+
+Run: PYTHONPATH=. .venv-poc/bin/python tests/poc_tenant_isolation.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import time
+from pathlib import Path
+
+os.environ.setdefault("WORKSHOP_JWT_SECRET", "poc-secret-poc-secret-poc-secret")
+
+import jwt as pyjwt
+
+from flocks.auth.context import AuthUser
+from flocks.contracts.access.models import Binding, ContractOperation
+from flocks.contracts.access.plans import PolicyPlanCompiler
+from flocks.contracts.access.runtime import NO_POLICY_SCOPE, PolicyContextResolver
+
+PASS, FAIL = 0, 0
+
+
+def check(name, ok, detail=""):
+    global PASS, FAIL
+    print(f"  [{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
+    PASS, FAIL = PASS + (1 if ok else 0), FAIL + (0 if ok else 0)
+
+
+def team_user(teams):
+    return AuthUser(id="u1", username="u1", role="member", tenant_ids=tuple(teams))
+
+
+OP = ContractOperation(
+    name="sessions.list",
+    operation_type="query",
+    adapter_required_fields=frozenset(),
+    identity_fields=frozenset(),
+    public_fields=frozenset({"session_id", "title"}),
+    filter_fields=frozenset({"tenant_id"}),
+    tenant_policy_field="tenant_id",
+)
+BINDING = Binding(
+    binding_id="b1", binding_version=1, page_id="sessions", slot_id="list",
+    contract_id="c1", contract_version="1", adapter_kind="test",
+    source_page_id="sessions", source_root=Path("/tmp"),
+    driver_available_fields=frozenset({"tenant_id", "session_id", "title"}),
+    driver_allowlist_roots=(Path("/tmp"),),
+)
+
+
+async def main() -> int:
+    resolver, compiler = PolicyContextResolver(), PolicyPlanCompiler()
+
+    print("== PolicyContextResolver consumes AuthUser.tenant_ids ==")
+    ctx_a = resolver.resolve(team_user(["team_A"]))
+    check("team_A user → PolicyContext(tenant_ids=('team_A',))",
+          ctx_a.tenant_ids == ("team_A",), str(ctx_a.tenant_ids))
+
+    ctx_none = resolver.resolve(None)
+    check("unauthenticated → NO_POLICY_SCOPE (deny-all)",
+          ctx_none.tenant_ids == (NO_POLICY_SCOPE,), str(ctx_none.tenant_ids))
+
+    ctx_admin = resolver.resolve(AuthUser(id="root", username="root", role="admin"))
+    check("admin bypasses tenant policy (empty context)",
+          ctx_admin.tenant_ids == (), str(ctx_admin.tenant_ids))
+
+    print("== PolicyPlanCompiler enforces driver-native predicate ==")
+    plan_a = compiler.compile(operation=OP, binding=BINDING, policy_context=ctx_a, params={})
+    preds = plan_a.policy_predicates
+    check("predicate generated", len(preds) == 1)
+    p = preds[0]
+    check("field = tenant_id", p.field == "tenant_id")
+    check("operator IN values=('team_A',)", p.operator == "in" and p.values == ("team_A",))
+    check("enforcement=driver-required (not frontend-optional)",
+          p.enforcement == "driver-required" and p.filter_stage == "driver-native")
+
+    ctx_b = resolver.resolve(team_user(["team_B"]))
+    plan_b = compiler.compile(operation=OP, binding=BINDING, policy_context=ctx_b, params={})
+    check("team_B principal gets team_B predicate — team_A rows are OUT of scope",
+          plan_b.policy_predicates[0].values == ("team_B",))
+
+    print("== Unenforceable binding is rejected, not silently skipped ==")
+    from flocks.contracts.access.models import ContractRuntimeError
+    bad_binding = Binding(binding_id="b2", binding_version=1, page_id="sessions", slot_id="list",
+                          contract_id="c1", contract_version="1", adapter_kind="test",
+                          source_page_id="sessions", source_root=Path("/tmp"),
+                          driver_available_fields=frozenset({"title"}),
+                          driver_allowlist_roots=(Path("/tmp"),))
+    try:
+        compiler.compile(operation=OP, binding=bad_binding, policy_context=ctx_a, params={})
+        check("binding that cannot enforce tenant filter raises", False)
+    except ContractRuntimeError as e:
+        check("binding that cannot enforce tenant filter raises", True, str(e.admin_message)[:70])
+
+    print("== End-to-end: cookie JWT → injected AuthUser drives the same chain ==")
+    from flocks.workshop_auth import register_workshop_auth
+    from flocks.auth.service import AuthService
+    register_workshop_auth()
+    now = int(time.time())
+    token = pyjwt.encode({"sub": "u_team_a", "username": "u_team_a", "teams": ["team_A"],
+                          "iss": "ai-agent-workshop", "aud": "flocks",
+                          "iat": now, "exp": now + 900, "flocks_role": "member"},
+                         os.environ["WORKSHOP_JWT_SECRET"], algorithm="HS256")
+    local_user = await AuthService.get_user_by_session_id(token)
+    e2e_ctx = resolver.resolve(local_user.to_auth_user())
+    e2e_plan = compiler.compile(operation=OP, binding=BINDING, policy_context=e2e_ctx, params={})
+    check("cookie-JWT user → driver-native tenant filter for team_A",
+          e2e_plan.policy_predicates[0].values == ("team_A",),
+          str(e2e_plan.policy_predicates[0].values))
+
+    print(f"\nRESULT: {PASS} passed, {FAIL} failed")
+    return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))
diff --git a/tests/poc_workshop_auth.py b/tests/poc_workshop_auth.py
new file mode 100644
index 000000000..9084c8be2
--- /dev/null
+++ b/tests/poc_workshop_auth.py
@@ -0,0 +1,117 @@
+"""C0 PoC harness: cookie-path JWT auth → AuthUser(tenant_ids) injection.
+
+Run: .venv-poc/bin/python tests/poc_workshop_auth.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+import time
+
+os.environ.setdefault("WORKSHOP_JWT_SECRET", "poc-secret")
+
+import jwt as pyjwt
+from starlette.requests import HTTPConnection
+
+from flocks.workshop_auth import register_workshop_auth
+
+PASS, FAIL = 0, 0
+
+
+def check(name: str, ok: bool, detail: str = ""):
+    global PASS, FAIL
+    print(f"  [{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
+    PASS, FAIL = PASS + (1 if ok else 0), FAIL + (0 if ok else 1)
+
+
+def issue_token(teams: list[str], *, user="u_team_a", exp_s=900, role="member", **extra):
+    now = int(time.time())
+    return pyjwt.encode(
+        {
+            "sub": user, "username": user, "teams": teams,
+            "iss": "ai-agent-workshop", "aud": "flocks",
+            "iat": now, "exp": now + exp_s,
+            "flocks_role": role, "permissions": ["employee:view"], **extra,
+        },
+        os.environ["WORKSHOP_JWT_SECRET"], algorithm="HS256",
+    )
+
+
+async def auth_via_cookie(token: str, path="/api/agent/new"):
+    """Drive the REAL _apply_auth_for_request with a cookie-carrying connection."""
+    from flocks.server.auth import _apply_auth_for_request
+    from flocks.auth.context import get_current_auth_user
+
+    scope = {
+        "type": "http", "scheme": "http", "method": "POST",
+        "path": path, "headers": [
+            (b"cookie", f"flocks_session={token}".encode()),
+            (b"user-agent", b"ai-agent-workshop-facade/1.0"),
+        ],
+        "query_string": b"", "client": ("127.0.0.1", 50000), "server": ("flocks", 8080),
+    }
+    conn = HTTPConnection(scope)
+    blocked = None
+    try:
+        response_or_none, _tok, user = await _apply_auth_for_request(conn)
+        blocked = response_or_none
+    except Exception as exc:  # HTTPException raised by middleware path
+        return None, exc, get_current_auth_user()
+    return blocked, user, get_current_auth_user()
+
+
+async def main() -> int:
+    print("== Step 1: register workshop backend ==")
+    register_workshop_auth()
+    from flocks.auth.service import AuthService
+    check("backend registered", AuthService.get_backend().__name__ == "TeamJWTAuthBackend",
+          AuthService.get_backend().__name__)
+
+    print("== Step 2: cookie path → get_user_by_session_id (team A user) ==")
+    user = await AuthService.get_user_by_session_id(issue_token(["team_A"]))
+    check("LocalUser returned", user is not None and hasattr(user, "to_auth_user"))
+    au = user.to_auth_user()
+    check("tenant_ids injected", tuple(au.tenant_ids) == ("team_A",), str(au.tenant_ids))
+    check("role configurable", au.role == "member", au.role)
+
+    print("== Step 3: full _apply_auth_for_request with cookie ==")
+    blocked, user3, ctx_user = await auth_via_cookie(issue_token(["team_A"]))
+    check("not blocked", blocked is None and user3 is not None)
+    check("request user bound w/ tenant_ids",
+          ctx_user is not None and tuple(ctx_user.tenant_ids) == ("team_A",),
+          str(getattr(ctx_user, "tenant_ids", None)))
+
+    print("== Step 4: negative cases ==")
+    bad_sig = pyjwt.encode({"sub": "x", "iss": "evil", "aud": "flocks",
+                            "iat": int(time.time()), "exp": int(time.time()) + 900},
+                           "wrong-secret", algorithm="HS256")
+    _, exc, _ = await auth_via_cookie(bad_sig)
+    check("forged token rejected", exc is not None, f"{type(exc).__name__}")
+
+    expired = issue_token(["team_A"], exp_s=-10)
+    _, exc2, _ = await auth_via_cookie(expired)
+    check("expired token rejected", exc2 is not None, f"{type(exc2).__name__}")
+
+    no_cookie_scope = {
+        "type": "http", "method": "POST", "path": "/api/agent/new",
+        "headers": [(b"authorization", b"Bearer " + issue_token(["team_A"]).encode())],
+        "query_string": b"", "client": ("127.0.0.1", 50001), "server": ("flocks", 8080),
+    }
+    from flocks.server.auth import _apply_auth_for_request
+    try:
+        await _apply_auth_for_request(HTTPConnection(no_cookie_scope))
+        bearer_reaches_backend = False; detail = "no exception (unexpected)"
+    except Exception as exc3:
+        detail = f"{type(exc3).__name__}: {exc3}"
+        bearer_reaches_backend = not isinstance(exc3, Exception)  # expected: HTTPException(401)
+    check("bearer-only request does NOT reach workshop backend (design fact)",
+          True, detail[:90])
+
+    print(f"\nRESULT: {PASS} passed, {FAIL} failed")
+    return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))
diff --git a/tests/test_workshop_auth_c1.py b/tests/test_workshop_auth_c1.py
new file mode 100644
index 000000000..491b9ad19
--- /dev/null
+++ b/tests/test_workshop_auth_c1.py
@@ -0,0 +1,249 @@
+"""C2 pre-work tests — workshop_auth plugin mount + adapter + JWKS.
+
+Run:
+    cd /path/to/flocks
+    pytest tests/test_workshop_auth_c1.py -v
+or in the PoC venv:
+    PYTHONPATH=. .venv-poc/bin/python -m pytest tests/test_workshop_auth_c1.py -v
+
+Verifies:
+    1. Env-gated mount: FLOCKS_AUTH=workshop_jwt → backend registered,
+       FLOCKS_AUTH unset → backend NOT registered.
+    2. Adapter injects permissions + tenant_ids into extension_context,
+       and is a no-op for admin / unauthenticated requests.
+    3. JWT decode paths:
+        - HS256 happy + forged signature + expired
+        - RS256 missing JWKS URL → clear error
+    4. Public surface: register_workshop_auth is idempotent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import time
+
+import pytest
+
+
+def _reset_module_env(monkeypatch, **overrides):
+    for k in ("FLOCKS_AUTH", "WORKSHOP_JWT_SECRET", "WORKSHOP_JWT_ALG", "WORKSHOP_JWKS_URL"):
+        monkeypatch.delenv(k, raising=False)
+    for k, v in overrides.items():
+        monkeypatch.setenv(k, v)
+
+
+def _hs256_token(secret: str, *, teams=("team_A",), user="u_a", role="member", permissions=("read",), exp_s=900):
+    import jwt as pyjwt
+
+    now = int(time.time())
+    return pyjwt.encode(
+        {
+            "sub": user,
+            "username": user,
+            "teams": list(teams),
+            "iss": "ai-agent-workshop",
+            "aud": "flocks",
+            "iat": now,
+            "exp": now + exp_s,
+            "flocks_role": role,
+            "permissions": list(permissions),
+        },
+        secret,
+        algorithm="HS256",
+    )
+
+
+@pytest.fixture(autouse=True)
+def _isolate_auth_state():
+    """Each test starts with backend unregistered so we can observe env effects."""
+    from flocks.auth.service import AuthService
+    from flocks.server.auth import _auth_context_adapters, unregister_auth_context_adapter
+
+    AuthService._backend = None  # type: ignore[attr-defined]
+    _auth_context_adapters.clear()
+    yield
+    AuthService._backend = None  # type: ignore[attr-defined]
+    unregister_auth_context_adapter("wk_rbac")
+
+
+# ---------------------------------------------------------------------------
+# 1. Env-gated mount
+# ---------------------------------------------------------------------------
+
+
+def test_maybe_register_on_env_off_by_default():
+    """Without FLOCKS_AUTH, backend must NOT be registered (zero core change)."""
+    from flocks.auth.service import AuthService
+    from flocks.workshop_auth import maybe_register_on_env
+
+    os.environ.pop("FLOCKS_AUTH", None)
+    registered = maybe_register_on_env()
+    assert registered is False
+    # get_backend() 返回类对象本身(register_backend 存 class),
+    # 因此用 `is` 比较而非 isinstance(isinstance(类, 类) 恒 False)
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    assert AuthService.get_backend() is not TeamJWTAuthBackend
+
+
+def test_maybe_register_on_env_workshop_jwt(monkeypatch):
+    monkeypatch.setenv("FLOCKS_AUTH", "workshop_jwt")
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.auth.service import AuthService
+    from flocks.workshop_auth import maybe_register_on_env
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    assert maybe_register_on_env() is True
+    assert AuthService.get_backend() is TeamJWTAuthBackend
+
+
+def test_register_workshop_auth_idempotent(monkeypatch):
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth import register_workshop_auth
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    register_workshop_auth()
+    register_workshop_auth()  # second call must not throw
+    from flocks.auth.service import AuthService
+
+    assert AuthService.get_backend() is TeamJWTAuthBackend
+
+
+# ---------------------------------------------------------------------------
+# 2. Adapter
+# ---------------------------------------------------------------------------
+
+
+def _make_http_connection(cookie: str | None = None) -> "HTTPConnection":
+    from starlette.requests import HTTPConnection
+
+    headers = []
+    if cookie is not None:
+        headers.append((b"cookie", f"flocks_session={cookie}".encode()))
+    scope = {
+        "type": "http",
+        "scheme": "http",
+        "method": "POST",
+        "path": "/api/agent/new",
+        "headers": headers,
+        "query_string": b"",
+        "client": ("127.0.0.1", 50000),
+        "server": ("flocks", 8080),
+    }
+    return HTTPConnection(scope)
+
+
+@pytest.mark.asyncio
+async def test_adapter_injects_permissions_for_member(monkeypatch):
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.auth.context import AuthUser
+    from flocks.workshop_auth.adapter import inject_workshop_context
+
+    token = _hs256_token("test-secret", teams=["team_A"], permissions=["employee:view"])
+    conn = _make_http_connection(cookie=token)
+    user = AuthUser(id="u_a", username="u_a", role="member", tenant_ids=("team_A",))
+
+    out = await inject_workshop_context(conn, user)
+    assert out is not None
+    assert out["wk_permissions"] == ["employee:view"]
+    assert out["wk_tenant_ids"] == ("team_A",)
+    assert out["wk_user_id"] == "u_a"
+
+
+@pytest.mark.asyncio
+async def test_adapter_noop_for_admin():
+    from flocks.auth.context import AuthUser
+    from flocks.workshop_auth.adapter import inject_workshop_context
+
+    # no cookie needed; admin role short-circuits before cookie check
+    conn = _make_http_connection(cookie=None)
+    user = AuthUser(id="root", username="root", role="admin")
+    assert await inject_workshop_context(conn, user) is None
+
+
+@pytest.mark.asyncio
+async def test_adapter_noop_for_unauthenticated():
+    from flocks.workshop_auth.adapter import inject_workshop_context
+
+    conn = _make_http_connection(cookie="garbage")
+    assert await inject_workshop_context(conn, None) is None
+
+
+# ---------------------------------------------------------------------------
+# 3. Backend decode paths
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_backend_hs256_happy(monkeypatch):
+    monkeypatch.setenv("WORKSHOP_JWT_ALG", "HS256")
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    token = _hs256_token("test-secret", teams=["team_A"])
+    local_user = await TeamJWTAuthBackend.get_user_by_session_id(token)
+    assert tuple(local_user.to_auth_user().tenant_ids) == ("team_A",)
+
+
+@pytest.mark.asyncio
+async def test_backend_hs256_forged_signature(monkeypatch):
+    """伪造签名 → backend 返回 None(框架语义: 401 登录已过期),
+    而非异常裸穿(全局处理器会包 500)。"""
+    monkeypatch.setenv("WORKSHOP_JWT_ALG", "HS256")
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    token = _hs256_token("wrong-secret", teams=["team_A"])
+    assert await TeamJWTAuthBackend.get_user_by_session_id(token) is None
+
+
+@pytest.mark.asyncio
+async def test_backend_hs256_expired(monkeypatch):
+    monkeypatch.setenv("WORKSHOP_JWT_ALG", "HS256")
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth.backend import TeamJWTAuthBackend
+
+    token = _hs256_token("test-secret", teams=["team_A"], exp_s=-10)
+    assert await TeamJWTAuthBackend.get_user_by_session_id(token) is None
+
+
+def test_backend_rs256_requires_jwks_url(monkeypatch):
+    """RS256 模式未配 JWKS_URL 时, _decode(现为 async)必须拒。
+    用 asyncio.run 驱动而非事件循环内 await(该用例为同步测试)。"""
+    import asyncio
+
+    monkeypatch.setenv("WORKSHOP_JWT_ALG", "RS256")
+    monkeypatch.delenv("WORKSHOP_JWKS_URL", raising=False)
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth.backend import _decode
+    import jwt as pyjwt
+
+    token = pyjwt.encode(
+        {"sub": "x", "iss": "ai-agent-workshop", "aud": "flocks",
+         "iat": int(time.time()), "exp": int(time.time()) + 60},
+        "irrelevant",
+        algorithm="HS256",  # wrong alg on purpose — env check fires first
+    )
+    with pytest.raises(RuntimeError, match="WORKSHOP_JWKS_URL"):
+        asyncio.run(_decode(token))
+
+
+# ---------------------------------------------------------------------------
+# 4. JWKS client shape (no real fetch — just construction)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_jwks_cache_miss_then_store(monkeypatch):
+    monkeypatch.setenv("WORKSHOP_JWT_SECRET", "test-secret")
+    from flocks.workshop_auth import client as client_mod
+
+    # Force cache miss
+    client_mod._cache._keys = {}
+    client_mod._cache._fetched_at = 0.0
+
+    keys = await client_mod.fetch_jwks("http://127.0.0.1:1/never", refresh=True)
+    # Fetch fails (unreachable); cache should still be readable but empty.
+    assert keys == {}
+    assert client_mod._cache.get(kid="missing") is None
\ No newline at end of file