Make plugin hook interfaces async; drop sync hooks and deprecated params

.claude/CLAUDE.md

@@ -180,11 +180,17 @@ class CreateGroup:
         ...
 ```
 
-Plugin hooks fire from inside `execute()`. Side-effecting Okta/notification work is spawned as
+Plugin hooks fire from inside `execute()`. The plugin interface is **async** (see the Plugin
+system section): each hook call returns coroutines that Access awaits via `run_hooks_to_completion`
+in `api/plugins/_async_dispatch.py`, which uses `asyncio.wait` (not `gather`, so one hook failing
+or a cancelled request can't tear down its siblings' in-flight I/O) and logs a failing hook at
+**ERROR** — a plugin's failure is surfaced, not swallowed, yet never breaks the committed
+operation; plugins are expected to catch their own expected/noisy errors. Separately,
+side-effecting Okta/notification work is spawned as
 `asyncio` tasks after the local DB state commits, then drained at the end of `execute()` via
 `drain_fan_out_tasks` in `api/operations/_fan_out.py` — which surfaces task failures (logged at
-WARNING) instead of dropping them, and uses `asyncio.wait` so a cancelled request doesn't tear
-down in-flight Okta calls. These fan-out tasks do network I/O only, never `db.session` access. Operations fall into two broad groups — request
+WARNING) instead of dropping them, and likewise uses `asyncio.wait` so a cancelled request doesn't
+tear down in-flight Okta calls. These fan-out tasks do network I/O only, never `db.session` access. Operations fall into two broad groups — request
 operations (create/approve/reject for access, group, and role requests) and resource operations
 (create/delete/modify for apps, groups, tags, users, and roles). The exact set changes as
 features land, so read `api/operations/` for the current, authoritative list — one class per
@@ -497,6 +503,19 @@ implementations live in that operator's own private repo. Each operator writes t
 implementations (for notifications, conditional access, etc.); operator-specific behavior
 belongs in one of those plugins, not upstream.
 
+**The plugin interface is async (Access 2.0).** `pluggy` never awaits, so a hook call returns a
+list of coroutines that Access drives to completion via `run_hooks_to_completion`
+(`api/plugins/_async_dispatch.py`). Consequently hook implementations across all four surfaces —
+notifications, conditional access, `metrics_reporter` (`record_*` etc.; `batch_metrics` is an
+async context manager), and the app-group-lifecycle *lifecycle* hooks — must be `async def`, and
+`verify_async_impls` fails fast at hook load if one is registered as a plain `def`. Do blocking
+I/O off the event loop: prefer a native async client (`httpx`, an SDK's async client) and fall
+back to `await asyncio.to_thread(...)` only for sync-only dependencies. App-group-lifecycle hooks
+that mutate state receive an `AsyncSession` (await ORM calls on it; `session.add(...)` is sync);
+the pure metadata/config/status/validation hooks stay **sync**. Plugin-contributed
+`access.commands` CLI commands run as ordinary Click commands and must drive their own
+`asyncio.run(...)`. The README's plugin section documents this in full.
+
 For hook signatures, adding a new plugin type, and implementing an operator override, read the
 hookspecs in `api/plugins/`.
 

POST_MIGRATION_TODO.md

@@ -124,71 +124,6 @@ clean `--snapshot-update` ergonomics.
 
 ---
 
-## Plugin Interface Modernization
-
-### 18. Make plugin interfaces async; deprecate the sync hooks
-
-The four plugin types (`notifications`, `conditional_access`,
-`app_group_lifecycle`, `metrics_reporter`) currently expose synchronous
-hooks via `pluggy`. The application is now async, so every call site
-bridges the sync hook — and the bridge differs by call site, which is
-exactly the smell that motivates going natively async:
-
-- `app_group_lifecycle` hooks (which receive `session=`) run through
-  `AsyncSession.run_sync`, which hands them a working sync `Session` on
-  the greenlet bridge (`create_group`, `delete_group`, `modify_group_*`).
-- The syncer's `access_expiring_*` `notifications` hooks also run through
-  `db.session.run_sync` — the syncer is a batch CLI job holding
-  AsyncSession-bound ORM rows, so run_sync keeps those rows on the
-  session's own thread instead of handing them to an `anyio` worker
-  thread (which was the original cross-thread hazard). Every ORM
-  relationship is `lazy="raise_on_sql"`, so a hook can only read what the
-  syncer eager-loaded regardless — extend the `joinedload`s there if a
-  notification plugin needs a wider graph.
-- `notifications` / `conditional_access` hooks fired from request-path
-  operations still run **inline on the event loop**, so a plugin doing
-  slow network I/O blocks the request. Moving that slow tail off the hot
-  path is item 10 (`BackgroundTasks`); `conditional_access` is harder
-  because its return value gates the request.
-
-Making the hooks natively async collapses all three bridges into one
-`await`. Plugins should eventually follow the app.
-
-Strategy mirrors the existing `DeprecationWarning` pattern in
-`api/plugins/notifications.py:48-74`:
-
-1. Duplicate every hook spec to add an `_async` variant —
-   `access_request_created_async`, `notify_user_async`, etc.
-2. Keep both for one major version. Sync hooks emit
-   `DeprecationWarning` when called.
-3. Remove the sync set at the next major bump.
-
-Plugins authored in this window can pick either flavor; the application
-prefers the async hook when both are registered.
-
-Rules to document with the async hook specs:
-- Hooks must not import `api.extensions.db` — they receive everything
-  they need (the lifecycle hooks' `session` argument is a sync
-  `Session` shim today and becomes an `AsyncSession` for the `_async`
-  variants).
-- **Plugin-contributed CLI commands** (the `access.commands` entry
-  point) now run inside the CLI's `asyncio.run` boundary; commands
-  written against the old sync `db.session` will break. They should
-  declare `async def` bodies and await session calls — document this
-  in the plugin guide alongside the async hook rollout.
-
-### 19. Pre-2.0 release checklist: drop deprecated plugin parameters
-
-Several `access_expiring_*` hooks already carry `DeprecationWarning`
-for legacy parameters: `groups`, `roles`, `users` on
-`access_expiring_owner`; `groups` on `access_expiring_user`. Their
-deprecation window ends at 2.0 — the parameters and their
-backward-compat shims must be removed before tagging the release.
-
-`api/plugins/notifications.py:48-74` is the canonical list.
-
----
-
 ## Security follow-ups (out of scope for the migration PR)
 
 ### 20. Nonce-based CSP

README.md

@@ -471,6 +471,17 @@ Plugins in Access follow the conventions defined by the [Python pluggy framework
 
 Plugins can also extend the `access` CLI with new commands via setuptools entry points — see [health_check_plugin](https://github.com/discord/access/tree/main/examples/plugins/health_check_plugin), which adds an `access health` command.
 
+### Async hook interface (Access 2.0)
+
+As of Access 2.0 the plugin hooks are **async** — Access is an async application and awaits each hook directly on the event loop, with no thread/greenlet bridge. When writing a plugin:
+
+- **Declare hook implementations with `async def`.** This applies to the notification hooks, the conditional-access hooks, the metrics `record_*` / `set_global_tags` / `flush` hooks, and the app-group-lifecycle *lifecycle* hooks (`group_created`, `group_updated`, `group_deleted`, `group_members_added`, `group_members_removed`, `sync_all_group_membership`). Access fails fast at startup with a clear error if one of these is registered as a plain `def`.
+- **Do blocking I/O off the event loop.** Prefer a native async client — `httpx`, `aiohttp`, or an SDK's async client (e.g. Slack's `AsyncWebClient`, as in [notifications_slack](https://github.com/discord/access/tree/main/examples/plugins/notifications_slack)). Only when a dependency is synchronous-only (no async client available) should you wrap its blocking calls in `await asyncio.to_thread(...)`. Never run blocking network calls directly in an `async def` hook.
+- **Lifecycle hooks receive an `AsyncSession`.** Await ORM calls on it (`await session.scalars(...)`); `session.add(...)` is synchronous. Never import `api.extensions.db` — everything a hook needs is passed in.
+- **Metrics `batch_metrics` is an async context manager** — use it as `async with metrics.batch_metrics(): await metrics.record_counter(...)`.
+- **These hooks stay synchronous:** the app-group-lifecycle metadata/config-schema/status-schema/validation hooks (`get_plugin_metadata`, `get_plugin_*_config_properties`, `validate_plugin_*_config`, `get_plugin_*_status_properties`). They are pure schema accessors with no I/O.
+- **Plugin-contributed CLI commands** (the `access.commands` entry point) run as ordinary Click commands, so they must drive their own async boundary — write an `async def` body and run it with `asyncio.run(...)`, as [health_check_plugin](https://github.com/discord/access/tree/main/examples/plugins/health_check_plugin) does.
+
 ### Installing a Plugin in the Docker Container
 
 Below is an example Dockerfile that would install the example notification plugin into the Access Docker container, which was built above using the top-level application [Dockerfile](https://github.com/discord/access/blob/main/Dockerfile). The plugin is copied into the `/app/plugins` directory and installed with `uv` into the application virtualenv at `/app/.venv` (the environment gunicorn and the `access` CLI run from).

api/manage.py

@@ -338,6 +338,7 @@ async def sync_app_group_memberships() -> None:
     """Invoke the periodic membership sync hook for all apps with app group lifecycle plugins configured."""
     from api.extensions import db
     from api.models import App
+    from api.plugins._async_dispatch import run_hooks_to_completion
     from api.plugins.app_group_lifecycle import get_app_group_lifecycle_hook
 
     click.echo("Starting app group lifecycle plugin sync")
@@ -360,14 +361,18 @@ async def sync_app_group_memberships() -> None:
 
     for app in apps:
         click.echo(f"Syncing app '{app.name}' (plugin: {app.app_group_lifecycle_plugin})")
+        # App-group-lifecycle hooks are native async: awaited directly
+        # with the AsyncSession, no run_sync bridge. run_hooks_to_completion uses
+        # asyncio.wait (not gather) and logs any plugin failure itself.
+        _, exceptions = await run_hooks_to_completion(
+            hook.sync_all_group_membership(session=db.session, app=app, plugin_id=app.app_group_lifecycle_plugin),
+            context=f"sync_all_group_membership for app '{app.name}'",
+        )
+        if exceptions:
+            await db.session.rollback()
+            click.echo(f"  ✗ Failed to sync app '{app.name}': {exceptions[0]}", err=True)
+            continue
         try:
-            # App-group-lifecycle hooks are sync plugin code that expects a
-            # sync Session; run them through the session's greenlet bridge.
-            await db.session.run_sync(
-                lambda session, app=app: hook.sync_all_group_membership(  # type: ignore[misc]
-                    session=session, app=app, plugin_id=app.app_group_lifecycle_plugin
-                )
-            )
             await db.session.commit()
             click.echo(f"  ✓ Synced app '{app.name}'")
         except Exception as e:

api/middleware.py

@@ -32,6 +32,7 @@
 from api.config import settings
 from api.context import RequestContext, reset_request_context, set_request_context
 from api.extensions import _session_scope, db
+from api.plugins._async_dispatch import run_hooks_to_completion
 from api.plugins.metrics_reporter import get_metrics_reporter_hook
 
 CSP = (
@@ -187,17 +188,25 @@ async def _send(message: Message) -> None:
             await self.app(scope, receive, _send)
         finally:
             duration_ms = (time.perf_counter() - start) * 1000.0
-            self._record(scope.get("method", ""), status_code, duration_ms)
+            await self._record(scope.get("method", ""), status_code, duration_ms)
 
-    def _record(self, method: str, status_code: int, duration_ms: float) -> None:
+    async def _record(self, method: str, status_code: int, duration_ms: float) -> None:
         try:
             hook = get_metrics_reporter_hook()
         except Exception:
             self._logger.exception("metrics_reporter hook unavailable; skipping emit")
             return
         tags = {"method": method, "status": str(status_code)}
         try:
-            hook.record_counter(metric_name="requests", value=1, tags=tags)
-            hook.record_histogram(metric_name="request.duration", value=duration_ms, tags={**tags, "unit": "ms"})
+            # run_hooks_to_completion uses asyncio.wait (not gather) and logs any
+            # per-plugin failure itself; emit is best-effort and never fails the request.
+            await run_hooks_to_completion(
+                hook.record_counter(metric_name="requests", value=1, tags=tags),
+                context="metrics requests counter",
+            )
+            await run_hooks_to_completion(
+                hook.record_histogram(metric_name="request.duration", value=duration_ms, tags={**tags, "unit": "ms"}),
+                context="metrics request.duration histogram",
+            )
         except Exception:
             self._logger.exception("metrics_reporter emit failed; continuing")

api/operations/approve_access_request.py

@@ -13,7 +13,6 @@
 from api.models import AccessRequest, AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleGroup
 from api.operations.constraints import CheckForReason
 from api.operations.modify_group_users import ModifyGroupUsers
-from api.plugins import get_notification_hook
 from api.schemas import AuditLogSchema, EventType
 
 
@@ -38,8 +37,6 @@ def __init__(
 
         self.notify = notify
 
-        self.notification_hook = get_notification_hook()
-
     async def execute(self) -> AccessRequest:
         # Lock the request row for the duration of the transaction so two
         # concurrent approvers can't both pass the pending-state guard below
@@ -147,34 +144,4 @@ async def execute(self) -> AccessRequest:
                 notify=self.notify,
             ).execute()
 
-        # Now handled inside ModifyGroupUsers
-        # self.access_request.approved_membership_id = (
-        #     db.session.query(OktaUserGroupMember).filter(
-        #         OktaUserGroupMember.user_id == self.access_request.requester_user_id
-        #     )
-        #     .filter(
-        #         OktaUserGroupMember.group_id == self.access_request.requested_group_id
-        #     )
-        #     .filter(OktaUserGroupMember.role_group_map_id.is_(None))
-        #     .filter(
-        #         OktaUserGroupMember.is_owner == self.access_request.request_ownership
-        #     )
-        #     .order_by(OktaUserGroupMember.created_at.desc())
-        #     .first()
-        #     .id
-        # )
-        # db.session.commit()
-
-        # requester = db.session.get(OktaUser, self.access_request.requester_user_id)
-
-        # approvers = get_all_possible_request_approvers(self.access_request)
-
-        # self.notification_hook.access_request_completed(
-        #     access_request=self.access_request,
-        #     group=group,
-        #     requester=requester,
-        #     approvers=approvers,
-        #     notify_requester=True,
-        # )
-
         return access_request

api/operations/approve_group_request.py

@@ -25,7 +25,7 @@
 from api.operations.constraints.check_for_self_add import CheckForSelfAdd
 from api.operations.create_group import CreateGroup
 from api.operations.modify_group_users import ModifyGroupUsers
-from api.plugins import get_notification_hook
+from api.plugins import NotificationHook, send_notification
 from api.schemas import AuditLogSchema, EventType
 
 
@@ -47,7 +47,6 @@ def __init__(
         self.approval_reason = approval_reason
         self.notify = notify
         self.bypass_self_approval = bypass_self_approval
-        self.notification_hook = get_notification_hook()
 
     async def execute(self) -> Optional[GroupRequest]:
         # Lock the request row for the transaction so concurrent approvers
@@ -278,7 +277,8 @@ async def execute(self) -> Optional[GroupRequest]:
         if self.notify:
             requester = await db.session.get(OktaUser, group_request.requester_user_id)
             approvers = await get_all_possible_request_approvers(group_request)
-            self.notification_hook.access_group_request_completed(
+            await send_notification(
+                NotificationHook.ACCESS_GROUP_REQUEST_COMPLETED,
                 group_request=group_request,
                 group=created_group,
                 requester=requester,

api/operations/approve_role_request.py

@@ -13,7 +13,6 @@
 from api.models import AccessRequestStatus, AppGroup, OktaGroup, OktaUser, RoleGroup, RoleRequest
 from api.operations.constraints import CheckForReason
 from api.operations.modify_role_groups import ModifyRoleGroups
-from api.plugins import get_notification_hook
 from api.schemas import AuditLogSchema, EventType
 
 
@@ -38,8 +37,6 @@ def __init__(
 
         self.notify = notify
 
-        self.notification_hook = get_notification_hook()
-
     async def execute(self) -> RoleRequest:
         # Lock the request row for the transaction so concurrent approvers
         # can't both pass the pending-state guard and double-grant. `of=` keeps

api/operations/create_access_request.py

@@ -23,7 +23,7 @@
 from api.models.okta_group import get_group_managers
 from api.operations.approve_access_request import ApproveAccessRequest
 from api.operations.reject_access_request import RejectAccessRequest
-from api.plugins import get_conditional_access_hook, get_notification_hook
+from api.plugins import ConditionalAccessHook, NotificationHook, evaluate_conditional_access, send_notification
 from api.schemas import AuditLogSchema, EventType
 
 
@@ -48,9 +48,6 @@ def __init__(
 
         self.request_approvers = select()
 
-        self.conditional_access_hook = get_conditional_access_hook()
-        self.notification_hook = get_notification_hook()
-
     async def execute(self) -> Optional[AccessRequest]:
         requester = await db.session.get(OktaUser, self.requester_user_id)
 
@@ -131,7 +128,8 @@ async def execute(self) -> Optional[AccessRequest]:
             )
         )
 
-        conditional_access_responses = self.conditional_access_hook.access_request_created(
+        conditional_access_responses = await evaluate_conditional_access(
+            ConditionalAccessHook.ACCESS_REQUEST_CREATED,
             access_request=access_request,
             group=group,
             group_tags=[active_tag_map.enabled_active_tag for active_tag_map in group.active_group_tags],
@@ -156,7 +154,8 @@ async def execute(self) -> Optional[AccessRequest]:
 
                 return access_request
 
-        self.notification_hook.access_request_created(
+        await send_notification(
+            NotificationHook.ACCESS_REQUEST_CREATED,
             access_request=access_request,
             group=group,
             requester=requester,

api/operations/create_group.py

@@ -8,7 +8,7 @@
 
 from api.extensions import db
 from api.models import App, AppGroup, AppTagMap, OktaGroup, OktaGroupTagMap, OktaUser, RoleGroup, Tag
-from api.plugins.app_group_lifecycle import invoke_app_group_lifecycle_hook
+from api.plugins.app_group_lifecycle import AppGroupLifecycleHook, invoke_app_group_lifecycle_hook
 from api.services import okta
 from api.schemas import AuditLogSchema, EventType
 
@@ -120,7 +120,7 @@ async def execute(self, *, _group: Optional[T] = None) -> T:
             await db.session.commit()
 
         # Invoke app group lifecycle plugin hook, if configured
-        await invoke_app_group_lifecycle_hook("group_created", group=self.group)
+        await invoke_app_group_lifecycle_hook(AppGroupLifecycleHook.GROUP_CREATED, group=self.group)
 
         # Audit logging
         email = None