feat(webhooks): publish AUTH_INVALIDATED on federated auth failures

[orhanrauf]

Stacked on #1794 (partial_failures), which is stacked on #1793 (pre-commit). Merge those first.

Summary

When federated search detects a credential failure (e.g. Slack returns `invalid_auth`, an OAuth refresh fails), we now atomically:

1. Flip `is_authenticated=False` on the source connection so the UI's computed status switches to `NEEDS_REAUTH` immediately, instead of waiting for the next sync job to record the error.
2. Publish a `SourceConnectionLifecycleEvent.auth_invalidated` event. Customers subscribed via webhooks can wire this into their own re-auth flow.

Both steps are gated by a per-source-connection Redis lock with a 1 h TTL, so a single broken connection that fails on every search doesn't spam customer webhooks. The notifier fails open (still flips DB + publishes) if Redis is unreachable, and swallows all errors so a search request never 500s because notification had trouble.

What changes

• `SourceConnectionEventType.AUTH_INVALIDATED` added to the enum; the webhook `EventType` auto-derives `SOURCE_CONNECTION_AUTH_INVALIDATED` (no further wiring needed for delivery).
• `SourceConnectionLifecycleEvent` gets:
  • an `auth_invalidated()` factory
  • an `error_reason: Optional[str]` field (machine-readable, never contains tokens)
• `AuthInvalidationNotifier` (domains/source_connections/auth_invalidation.py) is the single entry-point for runtime auth failures. It owns the Redis dedupe, the DB flip, and the event publish.
• `SearchPlanExecutor` (v2) calls the notifier for each `AUTH_INVALIDATED` partial failure (PR feat(search): surface federated source failures as partial_failures #1794), dedup'd per request.
• Legacy `SearchService.search()` replaces the direct `is_authenticated` flip in `_handle_failed_federated_auth` with notifier dispatch when one is wired in (falls back to the old behaviour otherwise so older test paths keep working).
• Container/factory wires the notifier; the `test_container` fixture builds a real notifier against fake Redis/sc_repo/event_bus so tests exercise the actual code path.

Dedupe semantics (per the design discussion)

• TTL 1 hour per source_connection_id. If a customer ignores the alert for a day, they get re-notified.
• Acquires `auth_invalidated:{sc_id}` via `SET NX EX`.
• Within a single request, the executor also dedups before calling the notifier, so the same connection failing on N keywords still produces one notify call (and at most one publish).

Test plan

• `airweave/domains/source_connections/tests/test_auth_invalidation.py` (6 tests): first-call publishes, dedupe within TTL, DB flip skipped when already unauthenticated, Redis failure fails open, missing source connection returns False, publish failure returns False (no propagation).
• `airweave/domains/search/tests/test_executor.py`: 2 new tests — notifier called on auth failure (with correct id + reason); notifier not called on non-auth (transient 502) failure.
• `pytest tests/unit` — 1208 pass.
• `pytest airweave/domains` — 2590 pass.
• Pre-commit hooks (ruff, mypy on changed lines, import-linter, pytest unit) all pass.
• Manual: invalidate a Slack token, hit `/search/classic`, observe `AUTH_INVALIDATED` delivered to a configured webhook endpoint; observe `is_authenticated=False` on the connection.

Follow-ups (deliberately out of scope)

• Fire from OAuth refresh `invalid_grant` and Slack's 401-after-refresh path. Both are good candidates but require touching token-provider plumbing — better as separate PRs.
• Agentic v2 still drops partial_failures (see PR feat(search): surface federated source failures as partial_failures #1794 follow-ups). Once that's threaded through, AUTH_INVALIDATED firing for agentic-tier searches comes for free.

---

Summary by cubic

Immediately publish AUTH_INVALIDATED webhooks and set is_authenticated=false when federated auth fails, so customers can trigger re-auth and the UI updates right away. Dedupe notifications per connection via Redis to avoid spam and never block search requests.

• New Features
  • Added SourceConnectionEventType.AUTH_INVALIDATED and SourceConnectionLifecycleEvent.auth_invalidated() with optional error_reason.
  • Introduced AuthInvalidationNotifier that dedupes per connection (Redis, 1h TTL), flips is_authenticated, and publishes the event; fails open and swallows errors.
  • Integrated notifier in SearchPlanExecutor for AUTH_INVALIDATED partial failures (deduped within a request).
  • Updated legacy SearchService.search() to use the notifier when provided; otherwise falls back to the direct DB flip.
  • Wired through container/factory and API endpoints to inject the notifier.

^{Written for commit 35b62f2. Summary will update on new commits. Review in cubic}

[cubic-dev-ai]

2 issues found across 11 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="backend/airweave/domains/source_connections/auth_invalidation.py">

<violation number="1" location="backend/airweave/domains/source_connections/auth_invalidation.py:81">
P1: Dedupe key set too early and never released on failure. Transient DB/event errors can block retries for 1 hour. Clear dedupe key when notifier exits with failure after winning the lock.</violation>

<violation number="2" location="backend/airweave/domains/source_connections/auth_invalidation.py:82">
P1: DB flip fail still treated as success. Event still publishes and function returns True. This can leave `is_authenticated` unchanged while emitting AUTH_INVALIDATED. Stop and return failure when update fails.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

P1: DB flip fail still treated as success. Event still publishes and function returns True. This can leave is_authenticated unchanged while emitting AUTH_INVALIDATED. Stop and return failure when update fails.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At backend/airweave/domains/source_connections/auth_invalidation.py, line 82:

<comment>DB flip fail still treated as success. Event still publishes and function returns True. This can leave `is_authenticated` unchanged while emitting AUTH_INVALIDATED. Stop and return failure when update fails.</comment>

<file context>
@@ -0,0 +1,146 @@
+            # SET NX EX — only succeeds if the key isn't set.
+            # Returns truthy on first detection within the TTL window.
+            acquired = await self._redis.set(dedupe_key, "1", ex=DEDUPE_TTL_SECONDS, nx=True)
+        except Exception as exc:
+            # Redis is down or misbehaving. Fail open: still flip DB +
+            # publish so we never silently swallow the signal.
</file context>

[cubic-dev-ai]

P1: Dedupe key set too early and never released on failure. Transient DB/event errors can block retries for 1 hour. Clear dedupe key when notifier exits with failure after winning the lock.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At backend/airweave/domains/source_connections/auth_invalidation.py, line 81:

<comment>Dedupe key set too early and never released on failure. Transient DB/event errors can block retries for 1 hour. Clear dedupe key when notifier exits with failure after winning the lock.</comment>

<file context>
@@ -0,0 +1,146 @@
+        try:
+            # SET NX EX — only succeeds if the key isn't set.
+            # Returns truthy on first detection within the TTL window.
+            acquired = await self._redis.set(dedupe_key, "1", ex=DEDUPE_TTL_SECONDS, nx=True)
+        except Exception as exc:
+            # Redis is down or misbehaving. Fail open: still flip DB +
</file context>