diff --git a/docs/operations/observability.md b/docs/operations/observability.md
index 306895116..1a3c82eaa 100644
--- a/docs/operations/observability.md
+++ b/docs/operations/observability.md
@@ -8,9 +8,10 @@ Every HTTP request emits one JSON line at completion (or earlier on error), tagg
 
 - `req.id` — UUID, present on the response as `X-Request-Id`. Quote it in bug reports.
 - `userId` / `orgId` / `authMethod` — propagated from the authenticated request.
-- `req.method`, `req.url`, `res.statusCode`, `responseTime`.
+- `req.method`, `req.url` (path only, no query string), `res.statusCode`, `responseTime`.
+- `req.mcpMethod` (the `Mcp-Method` header, on MCP requests), `req.userAgent`, `req.ip` (first `X-Forwarded-For` hop, else the socket address).
 
-Sensitive fields are stripped at the pino layer:
+Request and response headers are not logged (`packages/backend/src/common/log-serializers.ts` keeps only the fields above), so no credential header can reach a line. As a second line of defence, pino also strips:
 - headers: `authorization`, `cookie`, `x-api-key`, `set-cookie`.
 - bodies: any `password`, `token`, `refreshToken`, `accessToken`, `apiKey`, `secret` field.
 
diff --git a/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql b/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql
new file mode 100644
index 000000000..5d4f8d0ce
--- /dev/null
+++ b/packages/backend/prisma/migrations/20261021100000_kg_value_seen_connector_fk/migration.sql
@@ -0,0 +1,37 @@
+-- =============================================================================
+-- Migration: kg_value_seen rows belong to their connector
+-- =============================================================================
+-- kg_value_seen.connector_id had no foreign key, so deleting a connector left
+-- its value hashes behind (102 deleted connectors, 57.5k rows on cloud by
+-- 21 Oct 2026). The observational ingest kept pairing fresh values with them
+-- and tried to create kg_nodes for the deleted connector, which Postgres
+-- refused on kg_nodes_connector_id_fkey some 21.8k times a day.
+--
+-- Delete the orphans, then add the foreign key with ON DELETE CASCADE plus an
+-- index on connector_id, so the cascade does not scan the whole table every
+-- time a connector is deleted.
+--
+-- Locking: the migration runs as one transaction. On cloud (2.7 M rows,
+-- 1.5 GB) the orphan scan takes under a second and the index build about
+-- four. CREATE INDEX blocks writes to kg_value_seen only (the KG ingest runs
+-- in the background and retries on the next call). The foreign key is added
+-- NOT VALID, which is instant, so "connectors" is locked against writes only
+-- for the short second delete and VALIDATE that follow. The second delete
+-- removes rows of a connector deleted between the first one and the moment
+-- the constraint started to apply.
+-- =============================================================================
+
+DELETE FROM "kg_value_seen" v
+WHERE NOT EXISTS (SELECT 1 FROM "connectors" c WHERE c."id" = v."connector_id");
+
+CREATE INDEX "kg_value_seen_connector_id_idx" ON "kg_value_seen"("connector_id");
+
+ALTER TABLE "kg_value_seen"
+    ADD CONSTRAINT "kg_value_seen_connector_id_fkey"
+    FOREIGN KEY ("connector_id") REFERENCES "connectors"("id")
+    ON DELETE CASCADE ON UPDATE CASCADE NOT VALID;
+
+DELETE FROM "kg_value_seen" v
+WHERE NOT EXISTS (SELECT 1 FROM "connectors" c WHERE c."id" = v."connector_id");
+
+ALTER TABLE "kg_value_seen" VALIDATE CONSTRAINT "kg_value_seen_connector_id_fkey";
diff --git a/packages/backend/prisma/schema.prisma b/packages/backend/prisma/schema.prisma
index 86024c8c2..4d3e499b5 100644
--- a/packages/backend/prisma/schema.prisma
+++ b/packages/backend/prisma/schema.prisma
@@ -468,6 +468,7 @@ model Connector {
   kgNodes         KgNode[]
   kgState         KgConnectorState?
   kgSkills        KgSkillSuggestion[]
+  kgValuesSeen    KgValueSeen[]
 
   @@index([organizationId])
   @@map("connectors")
@@ -1460,6 +1461,10 @@ model KgValueSeen {
   organizationId String       @map("organization_id")
   organization   Organization @relation(fields: [organizationId], references: [id], onDelete: Cascade)
   connectorId    String       @map("connector_id")
+  // Deleting a connector deletes its values. Without the relation they stayed
+  // behind and every correlate pass tried to create nodes for the deleted
+  // connector (21.8k foreign-key errors a day on cloud by 21 Oct 2026).
+  connector      Connector    @relation(fields: [connectorId], references: [id], onDelete: Cascade)
   valueHash      String       @map("value_hash")
   entity         String
   field          String
@@ -1472,6 +1477,7 @@ model KgValueSeen {
   @@unique([organizationId, connectorId, valueHash, entity, field, direction], map: "kg_value_seen_occurrence_key")
   @@index([organizationId, valueHash])
   @@index([seenAt])
+  @@index([connectorId])
   @@map("kg_value_seen")
 }
 
diff --git a/packages/backend/src/auth/email-verified.guard.spec.ts b/packages/backend/src/auth/email-verified.guard.spec.ts
index 59a93c3ca..c6e9f763c 100644
--- a/packages/backend/src/auth/email-verified.guard.spec.ts
+++ b/packages/backend/src/auth/email-verified.guard.spec.ts
@@ -1,5 +1,6 @@
 import { ForbiddenException } from '@nestjs/common';
 import { EmailVerifiedGuard } from './email-verified.guard';
+import { McpEmailVerificationRequiredException } from './mcp-email-verification';
 
 /**
  * Global guards run before the route's AuthGuard('jwt'), so req.user is unset
@@ -8,7 +9,9 @@ import { EmailVerifiedGuard } from './email-verified.guard';
 describe('EmailVerifiedGuard', () => {
   function makeGuard(opts: { cloud?: boolean; verified?: boolean }) {
     const prisma = {
-      user: { findUnique: jest.fn(async () => ({ emailVerified: opts.verified ?? false })) },
+      user: {
+        findUnique: jest.fn(async () => ({ emailVerified: opts.verified ?? false, email: 'jane.doe@example.com' })),
+      },
     };
     const auth = {
       verifyToken: jest.fn((t: string) => {
@@ -35,8 +38,17 @@ describe('EmailVerifiedGuard', () => {
   it('tells an AI client on /mcp what the person has to do', async () => {
     const { guard } = makeGuard({ verified: false });
     const refusal = guard.canActivate(ctx({ path: '/mcp/srv1', headers: { authorization: 'Bearer good' } }));
+    await expect(refusal).rejects.toBeInstanceOf(McpEmailVerificationRequiredException);
     await expect(refusal).rejects.toBeInstanceOf(ForbiddenException);
-    await expect(refusal).rejects.toThrow(/confirm the email address.*sign in at https:\/\//);
+    await expect(refusal).rejects.toThrow(/not been verified.*ja\*\*\*@example\.com.*Sign in at https:\/\/\S+\/login/);
+    await expect(refusal).rejects.not.toThrow(/jane\.doe/);
+  });
+
+  it('keeps the plain refusal off /mcp', async () => {
+    const { guard } = makeGuard({ verified: false });
+    const refusal = guard.canActivate(ctx({ path: '/api/connectors', headers: { authorization: 'Bearer good' } }));
+    await expect(refusal).rejects.not.toBeInstanceOf(McpEmailVerificationRequiredException);
+    await expect(refusal).rejects.toThrow('Email verification required');
   });
 
   it('lets a verified user through', async () => {
diff --git a/packages/backend/src/auth/email-verified.guard.ts b/packages/backend/src/auth/email-verified.guard.ts
index 316e9d605..774cf7dfc 100644
--- a/packages/backend/src/auth/email-verified.guard.ts
+++ b/packages/backend/src/auth/email-verified.guard.ts
@@ -3,6 +3,10 @@ import { ConfigService } from '@nestjs/config';
 import { PrismaService } from '../common/prisma.service';
 import { AuthService } from './auth.service';
 import { resolveUserIdFromTokenPayload } from './resolve-user-id.util';
+import {
+  emailVerificationNotice,
+  McpEmailVerificationRequiredException,
+} from './mcp-email-verification';
 
 /**
  * In cloud mode, users must verify their email before they can access any
@@ -80,21 +84,20 @@ export class EmailVerifiedGuard implements CanActivate {
 
     const dbUser = await this.prisma.user.findUnique({
       where: { id: userId },
-      select: { emailVerified: true },
+      select: { emailVerified: true, email: true },
     });
     // Unknown user: not ours to judge (see the class comment).
     if (!dbUser) return true;
     if (!dbUser.emailVerified) {
       if (path.startsWith('/mcp')) {
         this.logger.warn(`Refused ${path}: user ${userId} has not verified their email`);
-        // An AI client shows this text to the person, who otherwise only sees
-        // the connection fail: on 9-10 Oct 2026 seven people who had connected
-        // Claude retried up to 122 times without knowing what was missing.
-        const dashboard = (this.configService.get<string>('FRONTEND_URL') || 'https://cloud.anythingmcp.com').replace(/\/+$/, '');
-        throw new ForbiddenException(
-          'Email verification required: confirm the email address of your AnythingMCP account ' +
-            `with the link we sent when you signed up, or sign in at ${dashboard} to get a new ` +
-            'link. Then try again; nothing else needs to change.',
+        // The person otherwise only sees the connection fail: on 9-10 Oct
+        // 2026 seven people who had connected Claude retried up to 122 times
+        // without knowing what was missing. McpEmailVerificationFilter turns
+        // this into an MCP answer the client shows and the model can read.
+        const dashboard = this.configService.get<string>('FRONTEND_URL') || 'https://cloud.anythingmcp.com';
+        throw new McpEmailVerificationRequiredException(
+          emailVerificationNotice(dashboard, dbUser.email),
         );
       }
       throw new ForbiddenException('Email verification required');
diff --git a/packages/backend/src/auth/mcp-email-verification.spec.ts b/packages/backend/src/auth/mcp-email-verification.spec.ts
new file mode 100644
index 000000000..e8be556f0
--- /dev/null
+++ b/packages/backend/src/auth/mcp-email-verification.spec.ts
@@ -0,0 +1,161 @@
+import { ForbiddenException, Logger } from '@nestjs/common';
+import { Client, StreamableHTTPClientTransport } from '@modelcontextprotocol/client';
+import express from 'express';
+import type { AddressInfo } from 'net';
+import type { Server } from 'http';
+import {
+  emailVerificationNotice,
+  maskEmail,
+  McpEmailVerificationFilter,
+  McpEmailVerificationRequiredException,
+  VERIFY_EMAIL_TOOL,
+} from './mcp-email-verification';
+
+describe('maskEmail', () => {
+  it('keeps enough to recognise the address and no more', () => {
+    expect(maskEmail('jane.doe@example.com')).toBe('ja***@example.com');
+    expect(maskEmail('jo@example.com')).toBe('j***@example.com');
+    expect(maskEmail('j@example.com')).toBe('j***@example.com');
+  });
+
+  it('gives nothing for what is not an address', () => {
+    expect(maskEmail(undefined)).toBeUndefined();
+    expect(maskEmail('no-at-sign')).toBeUndefined();
+    expect(maskEmail('@example.com')).toBeUndefined();
+  });
+});
+
+describe('emailVerificationNotice', () => {
+  it('says what is missing, where the email went, and how to get a new code', () => {
+    const text = emailVerificationNotice('https://cloud.anythingmcp.com/', 'jane.doe@example.com');
+    expect(text).toMatch(/email address has not been verified/);
+    expect(text).toContain('ja***@example.com');
+    expect(text).toContain('https://cloud.anythingmcp.com/login');
+    expect(text).not.toContain('jane.doe');
+  });
+});
+
+/**
+ * The guard's refusal, as an MCP client receives it. A plain 403 reads as a
+ * broken connection; this answers the protocol so the notice reaches the
+ * person and the model, with nothing of the workspace in it.
+ */
+describe('McpEmailVerificationFilter', () => {
+  const NOTICE = emailVerificationNotice('https://cloud.example.com', 'jane.doe@example.com');
+  let server: Server;
+  let base: string;
+  const savedJson = process.env.MCP_STREAMABLE_JSON_RESPONSE;
+
+  beforeAll(async () => {
+    jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
+    const filter = new McpEmailVerificationFilter();
+    const app = express();
+    app.use(express.json());
+    // Stands in for Nest: the guard throws, the filter gets the exception.
+    app.all('/mcp', (req, res) => {
+      const host = { switchToHttp: () => ({ getRequest: () => req, getResponse: () => res }) } as any;
+      filter.catch(new McpEmailVerificationRequiredException(NOTICE), host);
+    });
+    await new Promise<void>((resolve) => {
+      server = app.listen(0, '127.0.0.1', () => resolve());
+    });
+    base = `http://127.0.0.1:${(server.address() as AddressInfo).port}/mcp`;
+  });
+
+  afterAll(async () => {
+    process.env.MCP_STREAMABLE_JSON_RESPONSE = savedJson;
+    jest.restoreAllMocks();
+    await new Promise((resolve) => server.close(resolve));
+  });
+
+  const rpc = (method: string, params: Record<string, unknown> = {}, id = 1) =>
+    fetch(base, {
+      method: 'POST',
+      headers: {
+        'content-type': 'application/json',
+        accept: 'application/json, text/event-stream',
+        'mcp-protocol-version': '2025-06-18',
+      },
+      body: JSON.stringify({ jsonrpc: '2.0', id, method, params }),
+    });
+
+  /** The JSON-RPC message of a JSON or SSE-framed answer. */
+  const message = async (res: Response) => {
+    const text = await res.text();
+    if ((res.headers.get('content-type') ?? '').includes('text/event-stream')) {
+      const data = text.split('\n').find((l) => l.startsWith('data: '));
+      return JSON.parse(data!.slice(6));
+    }
+    return JSON.parse(text);
+  };
+
+  const initialize = () =>
+    rpc('initialize', {
+      protocolVersion: '2025-06-18',
+      capabilities: {},
+      clientInfo: { name: 'test', version: '1' },
+    });
+
+  for (const json of [true, false]) {
+    describe(json ? 'with JSON responses' : 'with SSE responses', () => {
+      beforeAll(() => {
+        process.env.MCP_STREAMABLE_JSON_RESPONSE = json ? 'true' : 'false';
+      });
+
+      it('lets the client connect and carries the notice in the instructions', async () => {
+        const res = await initialize();
+        expect(res.status).toBe(200);
+        const msg = await message(res);
+        expect(msg.result.instructions).toBe(NOTICE);
+        expect(msg.result.serverInfo.name).toBe('AnythingMCP');
+      });
+
+      it('lists only the verification tool', async () => {
+        const msg = await message(await rpc('tools/list'));
+        expect(msg.result.tools.map((t: { name: string }) => t.name)).toEqual([VERIFY_EMAIL_TOOL]);
+      });
+
+      it('answers any tool call with the notice as an error the model reads', async () => {
+        const msg = await message(await rpc('tools/call', { name: 'crm_list_customers', arguments: {} }));
+        const text = JSON.stringify(msg);
+        expect(text).toContain('ja***@example.com');
+        expect(text).not.toContain('jane.doe');
+        const call = await message(await rpc('tools/call', { name: VERIFY_EMAIL_TOOL, arguments: {} }));
+        expect(call.result.isError).toBe(true);
+        expect(call.result.content[0].text).toBe(NOTICE);
+      });
+    });
+  }
+
+  it('works for the current SDK client, whatever protocol revision it negotiates', async () => {
+    delete process.env.MCP_STREAMABLE_JSON_RESPONSE;
+    const client = new Client({ name: 'spec', version: '1.0.0' });
+    await client.connect(new StreamableHTTPClientTransport(new URL(base)));
+    try {
+      expect(client.getInstructions()).toBe(NOTICE);
+      expect((await client.listTools()).tools.map((t) => t.name)).toEqual([VERIFY_EMAIL_TOOL]);
+      const result: any = await client.callTool({ name: 'crm_list_customers', arguments: {} });
+      expect(result.isError).toBe(true);
+      expect(result.content[0].text).toBe(NOTICE);
+    } finally {
+      await client.close();
+    }
+  });
+
+  it('refuses a GET and a listen stream with the notice as a JSON-RPC error', async () => {
+    const get = await fetch(base, { headers: { accept: 'text/event-stream' } });
+    expect(get.status).toBe(403);
+    expect((await get.json()).error.message).toBe(NOTICE);
+
+    const listen = await rpc('subscriptions/listen');
+    expect(listen.status).toBe(403);
+    expect((await listen.json()).error).toEqual({ code: -32001, message: NOTICE });
+  });
+
+  it('is still a 403 with the notice wherever the filter is not in play', () => {
+    const err = new McpEmailVerificationRequiredException(NOTICE);
+    expect(err).toBeInstanceOf(ForbiddenException);
+    expect(err.getStatus()).toBe(403);
+    expect(err.message).toBe(NOTICE);
+  });
+});
diff --git a/packages/backend/src/auth/mcp-email-verification.ts b/packages/backend/src/auth/mcp-email-verification.ts
new file mode 100644
index 000000000..f100de5b2
--- /dev/null
+++ b/packages/backend/src/auth/mcp-email-verification.ts
@@ -0,0 +1,177 @@
+import { ArgumentsHost, Catch, ExceptionFilter, ForbiddenException, Logger } from '@nestjs/common';
+import type { Request, Response } from 'express';
+import { McpServer, isLegacyRequest } from '@modelcontextprotocol/server';
+import {
+  NodeStreamableHTTPServerTransport,
+  toNodeHandler,
+  toWebRequest,
+} from '@modelcontextprotocol/node';
+import { APP_VERSION } from '../common/app-version';
+import { createQuietMcpHandler } from '../mcp-server/mcp-handler.util';
+
+/**
+ * What an AI client gets from /mcp while its user has not verified their
+ * email address (cloud only, see EmailVerifiedGuard).
+ *
+ * WHY. The guard answered a plain HTTP 403. MCP clients treat that as a
+ * failed connection and show neither the person nor the model why: on
+ * 21 Oct 2026, 13 people who had connected Claude without verifying were
+ * refused 438 times in under six hours, one of them 80 times, with nothing
+ * telling them what was missing.
+ *
+ * The request is still refused: the guard throws before any controller runs,
+ * and nothing of the workspace is read. This filter only shapes the refusal
+ * as MCP: a server with no tools of the user's, whose instructions and only
+ * tool carry the notice, so the client connects and the model can tell the
+ * person what to do. Nothing here sends an email: a new code comes from
+ * signing in, as before, so the rate of verification emails is unchanged.
+ */
+
+/** Name of the one tool served to an unverified user. */
+export const VERIFY_EMAIL_TOOL = 'anythingmcp_verify_email';
+
+/** `jo***@example.com`: enough for the person to recognise, no more. */
+export function maskEmail(email: string | null | undefined): string | undefined {
+  if (typeof email !== 'string') return undefined;
+  const at = email.lastIndexOf('@');
+  if (at < 1 || at === email.length - 1) return undefined;
+  const local = email.slice(0, at);
+  const keep = local.length > 2 ? 2 : 1;
+  return `${local.slice(0, keep)}***${email.slice(at)}`;
+}
+
+/** The text shown to the person and the model. */
+export function emailVerificationNotice(dashboardUrl: string, email?: string | null): string {
+  const base = dashboardUrl.replace(/\/+$/, '');
+  const masked = maskEmail(email);
+  return (
+    'Your AnythingMCP account cannot be used yet: its email address has not been verified, ' +
+    'so no tools are available. ' +
+    `We sent a verification email${masked ? ` to ${masked}` : ''} when you signed up: open the link ` +
+    'in it (or enter its code), then try again. ' +
+    `No email, or the code expired? Sign in at ${base}/login and a new code is sent; check the spam ` +
+    'folder too. Nothing needs to change in your AI client.'
+  );
+}
+
+/** Thrown by EmailVerifiedGuard on /mcp; a 403 with the notice wherever this filter does not apply. */
+export class McpEmailVerificationRequiredException extends ForbiddenException {
+  constructor(readonly notice: string) {
+    super(notice);
+  }
+}
+
+/** A tool name worth answering under (what MCP clients send; anything else gets "not found"). */
+const TOOL_NAME = /^[A-Za-z0-9_.-]{1,128}$/;
+
+/**
+ * The server an unverified user is answered from: the notice, and nothing of
+ * theirs. `calledTool` is the tool the request calls, if any: it answers with
+ * the notice too, so a client that still knows a tool from earlier gets the
+ * explanation rather than "tool not found". It is never listed.
+ */
+export function emailVerificationServer(notice: string, calledTool?: string): McpServer {
+  const server = new McpServer(
+    { name: 'AnythingMCP', version: APP_VERSION },
+    { capabilities: { tools: {} }, instructions: notice },
+  );
+  const answer = async () => ({ content: [{ type: 'text' as const, text: notice }], isError: true });
+  server.registerTool(
+    VERIFY_EMAIL_TOOL,
+    {
+      description:
+        'The AnythingMCP account behind this connection has not verified its email address, so its ' +
+        'tools are unavailable. Call this to get the steps, and pass them on to the user.',
+      annotations: { readOnlyHint: true, openWorldHint: false },
+    },
+    answer,
+  );
+  if (calledTool && calledTool !== VERIFY_EMAIL_TOOL && TOOL_NAME.test(calledTool)) {
+    server.registerTool(calledTool, { description: notice }, answer);
+  }
+  return server;
+}
+
+/** The tool a `tools/call` request names. */
+function calledToolOf(body: unknown): string | undefined {
+  const b = body as { method?: unknown; params?: { name?: unknown } };
+  return b?.method === 'tools/call' && typeof b.params?.name === 'string' ? b.params.name : undefined;
+}
+
+/** A single JSON-RPC message other than a subscriptions/listen stream. */
+function isServableRequest(body: unknown): boolean {
+  if (!body || typeof body !== 'object' || Array.isArray(body)) return false;
+  const method = (body as { method?: unknown }).method;
+  // A listen stream would stay open for minutes and has nothing to deliver.
+  return typeof method === 'string' && method !== 'subscriptions/listen';
+}
+
+@Catch(McpEmailVerificationRequiredException)
+export class McpEmailVerificationFilter implements ExceptionFilter {
+  private readonly logger = new Logger(McpEmailVerificationFilter.name);
+
+  catch(exception: McpEmailVerificationRequiredException, host: ArgumentsHost): void {
+    const ctx = host.switchToHttp();
+    const req = ctx.getRequest<Request>();
+    const res = ctx.getResponse<Response>();
+    if (res.headersSent) return;
+
+    const body = (req as { body?: unknown }).body;
+    if (req.method === 'POST' && isServableRequest(body)) {
+      void this.serve(req, res, body, exception.notice).catch((err) => {
+        this.logger.warn(`Could not serve the email verification notice: ${err?.message ?? err}`);
+        this.refuse(res, exception.notice);
+      });
+      return;
+    }
+    this.refuse(res, exception.notice);
+  }
+
+  private refuse(res: Response, notice: string): void {
+    if (res.headersSent) return;
+    res.status(403).json({ jsonrpc: '2.0', error: { code: -32001, message: notice }, id: null });
+  }
+
+  private async isLegacy(req: Request, body: unknown): Promise<boolean> {
+    try {
+      return await isLegacyRequest(await toWebRequest(req, body), body);
+    } catch {
+      return false;
+    }
+  }
+
+  /**
+   * Same framing as the MCP endpoint (McpEndpointController.serveStateless):
+   * the handler for 2026-07-28 traffic, and single-shot JSON for 2025-era
+   * requests when JSON responses are enabled, which the handler's own
+   * fallback cannot produce.
+   */
+  private async serve(req: Request, res: Response, body: unknown, notice: string): Promise<void> {
+    const json = process.env.MCP_STREAMABLE_JSON_RESPONSE === 'true';
+    const build = () => emailVerificationServer(notice, calledToolOf(body));
+    if (json && (await this.isLegacy(req, body))) {
+      const server = build();
+      const transport = new NodeStreamableHTTPServerTransport({
+        sessionIdGenerator: undefined,
+        enableJsonResponse: true,
+      });
+      try {
+        await server.connect(transport);
+        await transport.handleRequest(req, res, body);
+      } finally {
+        await transport.close().catch(() => undefined);
+        await server.close().catch(() => undefined);
+      }
+      return;
+    }
+    const handler = createQuietMcpHandler(build, {
+      legacy: 'stateless',
+      responseMode: json ? 'json' : 'sse',
+    });
+    try {
+      await toNodeHandler(handler)(req, res, body);
+    } finally {
+      await handler.close().catch(() => undefined);
+    }
+  }
+}
diff --git a/packages/backend/src/common/log-serializers.spec.ts b/packages/backend/src/common/log-serializers.spec.ts
new file mode 100644
index 000000000..0f9636e26
--- /dev/null
+++ b/packages/backend/src/common/log-serializers.spec.ts
@@ -0,0 +1,79 @@
+import pino from 'pino';
+import { Writable } from 'stream';
+import { serializeRequest, serializeResponse } from './log-serializers';
+
+/** What an `/mcp` request looked like in the cloud log on 21 Oct 2026, plus credentials. */
+const mcpRequest = () => ({
+  id: 'req-1',
+  method: 'POST',
+  url: '/mcp?token=one-time-secret',
+  query: { token: 'one-time-secret' },
+  params: { path: ['mcp'] },
+  headers: {
+    host: 'cloud.anythingmcp.com',
+    'user-agent': 'Claude-User',
+    accept: 'application/json, text/event-stream',
+    'content-type': 'application/json',
+    'mcp-method': 'tools/list',
+    'mcp-protocol-version': '2026-07-28',
+    via: '1.1 Caddy',
+    'x-forwarded-for': '160.79.106.167, 10.0.0.1',
+    authorization: 'Bearer eyJhbGciOi.secret',
+    cookie: 'amcp_session=secret',
+    'x-api-key': 'mcp_secret',
+    'x-custom-token': 'secret',
+  } as Record<string, string>,
+  remoteAddress: '::ffff:172.18.0.6',
+  remotePort: 32852,
+});
+
+describe('log serializers', () => {
+  it('keeps what a request is looked up by, and nothing else', () => {
+    expect(serializeRequest(mcpRequest())).toEqual({
+      id: 'req-1',
+      method: 'POST',
+      url: '/mcp',
+      mcpMethod: 'tools/list',
+      userAgent: 'Claude-User',
+      ip: '160.79.106.167',
+    });
+  });
+
+  it('never writes a credential, whatever header or query parameter carries it', () => {
+    expect(JSON.stringify(serializeRequest(mcpRequest()))).not.toMatch(/secret/);
+  });
+
+  it('falls back to the socket address and leaves out absent headers', () => {
+    expect(
+      serializeRequest({ id: 'r', method: 'GET', url: '/api/health', headers: {}, remoteAddress: '127.0.0.1' }),
+    ).toEqual({ id: 'r', method: 'GET', url: '/api/health', ip: '127.0.0.1' });
+  });
+
+  it('caps a header value a client made huge', () => {
+    const req = mcpRequest();
+    req.headers['user-agent'] = 'x'.repeat(5_000);
+    expect(serializeRequest(req).userAgent).toHaveLength(200);
+  });
+
+  it('keeps only the status of a response', () => {
+    const res = { statusCode: 200, headers: { 'set-cookie': 'secret' } };
+    expect(serializeResponse(res)).toEqual({ statusCode: 200 });
+  });
+
+  it('makes a request line a fraction of its former size through pino', () => {
+    const lines: string[] = [];
+    const sink = new Writable({
+      write(chunk, _enc, cb) {
+        lines.push(String(chunk));
+        cb();
+      },
+    });
+    const withDefaults = pino({}, sink);
+    const withOurs = pino({ serializers: { req: serializeRequest } }, sink);
+    withDefaults.info({ req: mcpRequest() }, 'request completed');
+    withOurs.info({ req: mcpRequest() }, 'request completed');
+    expect(lines[0]).toMatch(/"headers"/);
+    expect(lines[1]).not.toMatch(/secret|"headers"/);
+    expect(lines[1].length).toBeLessThan(lines[0].length / 2);
+  });
+});
diff --git a/packages/backend/src/common/log-serializers.ts b/packages/backend/src/common/log-serializers.ts
new file mode 100644
index 000000000..f838e3fed
--- /dev/null
+++ b/packages/backend/src/common/log-serializers.ts
@@ -0,0 +1,65 @@
+/**
+ * What a request and its response contribute to a log line.
+ *
+ * pino-http's default serializers write every request and response header,
+ * and the request object is bound to every line logged while the request
+ * runs. On cloud that was most of the backend log: about 1.2 KB of headers
+ * per `/mcp` request, so the docker log (5 x 50 MB) only reached back about
+ * 5.7 hours on 21 Oct 2026. The fields kept here are the ones the logs are
+ * read for: which request (id), what (method, path, MCP method), from whom
+ * (client, address; user and workspace come from customProps) and how it
+ * ended (status; the duration is added by pino-http as responseTime).
+ *
+ * Only named headers are read, so a credential can never end up in a line
+ * whatever header it arrives in. The query string is dropped from the path:
+ * OAuth callbacks and one-time links carry codes and tokens there.
+ */
+
+/** The request as pino's wrapped serializer hands it over (`raw` is the IncomingMessage). */
+interface SerializedRequestInput {
+  id?: unknown;
+  method?: string;
+  url?: string;
+  headers?: Record<string, string | string[] | undefined>;
+  remoteAddress?: string;
+}
+
+export interface LoggedRequest {
+  id?: unknown;
+  method?: string;
+  url?: string;
+  mcpMethod?: string;
+  userAgent?: string;
+  ip?: string;
+}
+
+/** Longest header value kept; a client can send anything. */
+const MAX_HEADER_CHARS = 200;
+
+function header(
+  headers: SerializedRequestInput['headers'],
+  name: string,
+): string | undefined {
+  const value = headers?.[name];
+  const first = Array.isArray(value) ? value[0] : value;
+  if (typeof first !== 'string' || !first) return undefined;
+  return first.length > MAX_HEADER_CHARS ? first.slice(0, MAX_HEADER_CHARS) : first;
+}
+
+export function serializeRequest(req: SerializedRequestInput): LoggedRequest {
+  const url = typeof req?.url === 'string' ? req.url.split('?')[0] : undefined;
+  // The first X-Forwarded-For hop is the client as the reverse proxy saw it.
+  const forwarded = header(req?.headers, 'x-forwarded-for')?.split(',')[0]?.trim();
+  const out: LoggedRequest = { id: req?.id, method: req?.method, url };
+  const mcpMethod = header(req?.headers, 'mcp-method');
+  if (mcpMethod) out.mcpMethod = mcpMethod;
+  const userAgent = header(req?.headers, 'user-agent');
+  if (userAgent) out.userAgent = userAgent;
+  const ip = forwarded || req?.remoteAddress;
+  if (ip) out.ip = ip;
+  return out;
+}
+
+export function serializeResponse(res: { statusCode?: number }): { statusCode?: number } {
+  return { statusCode: res?.statusCode };
+}
diff --git a/packages/backend/src/common/logger.module.ts b/packages/backend/src/common/logger.module.ts
index 7b5b7fbee..42e560f15 100644
--- a/packages/backend/src/common/logger.module.ts
+++ b/packages/backend/src/common/logger.module.ts
@@ -2,6 +2,7 @@ import { Module } from '@nestjs/common';
 import { LoggerModule as PinoLoggerModule } from 'nestjs-pino';
 import { randomUUID } from 'crypto';
 import type { IncomingMessage, ServerResponse } from 'http';
+import { serializeRequest, serializeResponse } from './log-serializers';
 
 /**
  * Structured logging — Pino over the default NestJS console logger.
@@ -12,9 +13,10 @@ import type { IncomingMessage, ServerResponse } from 'http';
  *  - Per-request correlation id: read from X-Request-Id, generated when
  *    missing, attached to every log line and echoed back on the response so
  *    a client can quote it when reporting bugs.
- *  - Authorization, cookies and X-API-Key headers are redacted.
- *  - The MCP endpoint is logged at trace level: it is a high-traffic JSON-RPC
- *    surface and we don't need a line per request unless debugging.
+ *  - Requests and responses are logged without their headers (see
+ *    log-serializers.ts): method, path, MCP method, client, address and
+ *    status only, so no credential header can reach a line. The redaction
+ *    below stays as a second line of defence.
  */
 @Module({
   imports: [
@@ -35,6 +37,10 @@ import type { IncomingMessage, ServerResponse } from 'http';
           res.setHeader('X-Request-Id', id);
           return id;
         },
+        serializers: {
+          req: serializeRequest,
+          res: serializeResponse,
+        },
         customProps: (req: IncomingMessage) => ({
           // Surface authenticated identity (set by passport / mcp-auth guards)
           userId: (req as any).user?.sub,
diff --git a/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts b/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts
index 3161471d0..24bd053cf 100644
--- a/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts
+++ b/packages/backend/src/connectors/engines/bearer-token-renewal.service.spec.ts
@@ -189,6 +189,30 @@ describe('BearerTokenRenewalService', () => {
     expect(mockedAxios.get).toHaveBeenCalledTimes(1);
   });
 
+  it('once a short-lived token has expired, marks the connector as needing a new token instead of warning', async () => {
+    const expiresAt = Date.now() + 3_600_000;
+    graph({ '/debug_token': () => ({ data: { expires_at: Math.floor(expiresAt / 1000) } }) });
+    expect(await service.getToken(withoutSecret(), 'conn-lapsed')).toBe(SHORT);
+    expect(authStatus.mark).toHaveBeenLastCalledWith(
+      'conn-lapsed',
+      expect.objectContaining({ status: 'expiring', code: 'meta_expiring' }),
+    );
+
+    // Two hours later, nobody has pasted a new token.
+    const now = Date.now();
+    jest.spyOn(Date, 'now').mockReturnValue(now + 2 * 3_600_000);
+    // The call still goes out: Meta's answer has the last word.
+    expect(await service.getToken(withoutSecret(), 'conn-lapsed')).toBe(SHORT);
+    expect(authStatus.mark).toHaveBeenLastCalledWith('conn-lapsed', {
+      status: 'needs_reauth',
+      code: 'token_expired',
+      message: expect.stringMatching(/expired on .* UTC and could not be renewed.*Paste a new token/),
+      ref: credentialRef(SHORT),
+    });
+    await service.getToken(withoutSecret(), 'conn-lapsed');
+    expect(authStatus.mark).toHaveBeenCalledTimes(2);
+  });
+
   it('records a refused exchange as a warning and keeps using the pasted token', async () => {
     graph({
       '/debug_token': () => graphError(190, undefined, 400),
diff --git a/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts b/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts
index 5d11fb5fd..1abcaa1b1 100644
--- a/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts
+++ b/packages/backend/src/connectors/engines/bearer-token-renewal.service.ts
@@ -6,7 +6,7 @@ import { decrypt, encrypt } from '../../common/crypto/encryption.util';
 import { getRequiredSecret } from '../../common/secrets.util';
 import { connectorPageUrl } from '../../common/url.util';
 import { getAdapter } from '../../adapters/catalog';
-import { ConnectorAuthStatusService, credentialRef } from './connector-auth-status.service';
+import { ConnectorAuthStatusService, credentialRef, expiredTokenHint } from './connector-auth-status.service';
 
 /**
  * Renewal of pasted bearer tokens that expire, for providers that let a
@@ -411,7 +411,12 @@ export class BearerTokenRenewalService {
     await this.authStatus?.clear(connectorId, { onlyCodes: WARNING_CODES });
   }
 
-  /** A token that will stop working soon and cannot be renewed here: say so. */
+  /**
+   * A token that will stop working soon and cannot be renewed here: say so.
+   * One that already has is no longer a warning: the connector needs a new
+   * token, and its owner gets the single email (an expired token cannot be
+   * exchanged). The call still goes out; Meta's answer has the last word.
+   */
   private async warnIfExpiring(
     connectorId: string,
     st: TokenState,
@@ -419,6 +424,18 @@ export class BearerTokenRenewalService {
     failure?: string,
   ): Promise<void> {
     if (typeof st.expiresAt !== 'number') return;
+    if (st.expiresAt <= Date.now()) {
+      const message = expiredTokenHint(new Date(st.expiresAt), connectorId);
+      if (this.warned.get(connectorId) === message) return;
+      this.warned.set(connectorId, message);
+      await this.authStatus?.mark(connectorId, {
+        status: 'needs_reauth',
+        code: 'token_expired',
+        message,
+        ref: st.sourceRef,
+      });
+      return;
+    }
     const within = (spec.renewWithinDays ?? DEFAULT_RENEW_WITHIN_DAYS) * DAY_MS;
     if (st.expiresAt - Date.now() > within) return;
     if (!failure && this.shouldRenew(st, spec)) return; // will be renewed on this call
diff --git a/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts b/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts
index cb7ea63f3..1cbcea640 100644
--- a/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts
+++ b/packages/backend/src/connectors/engines/connector-auth-status.service.spec.ts
@@ -21,6 +21,11 @@ function fakePrisma(row: Record<string, any>) {
     connector: {
       findUnique: jest.fn(async ({ where }: any) => (where.id === row.id ? { ...row } : null)),
       update: jest.fn(async ({ data }: any) => Object.assign(row, data)),
+      findMany: jest.fn(async ({ where }: any) =>
+        row.authStatus === where.authStatus && row.tokenExpiresAt && row.tokenExpiresAt <= where.tokenExpiresAt.lte
+          ? [{ id: row.id, authStatusRef: row.authStatusRef ?? null, tokenExpiresAt: row.tokenExpiresAt }]
+          : [],
+      ),
       updateMany: jest.fn(async ({ where, data }: any) => {
         if (!matches(row, where)) return { count: 0 };
         Object.assign(row, data);
@@ -122,6 +127,63 @@ describe('ConnectorAuthStatusService', () => {
     expect(issue.ref).toMatch(/^[0-9a-f]{16}$/);
   });
 
+  describe('expireLapsedTokens', () => {
+    const ref = credentialRef('EAAshortlived');
+    const expiresAt = new Date('2026-10-20T17:00:00Z');
+    const warn = { status: 'expiring' as const, code: 'meta_expiring', message: 'This is a short-lived token.', ref };
+
+    it('turns a warning whose token has expired into needs_reauth and emails the owner once', async () => {
+      await service.mark('conn-1', warn);
+      row.tokenExpiresAt = expiresAt;
+      await flush();
+      expect(email.sendConnectorAuthEmail).not.toHaveBeenCalled();
+
+      const later = new Date('2026-10-21T10:00:00Z');
+      expect(await service.expireLapsedTokens(later)).toBe(1);
+      await flush();
+      expect(row).toMatchObject({ authStatus: 'needs_reauth', authStatusCode: 'token_expired', authStatusRef: ref });
+      expect(row.authStatusMessage).toMatch(/expired on 2026-10-20 17:00 UTC and could not be renewed/);
+      expect(row.authNotifiedAt).toBeInstanceOf(Date);
+      expect(email.sendConnectorAuthEmail).toHaveBeenCalledTimes(1);
+
+      // The next sweep finds nothing left to do.
+      expect(await service.expireLapsedTokens(later)).toBe(0);
+      await flush();
+      expect(email.sendConnectorAuthEmail).toHaveBeenCalledTimes(1);
+    });
+
+    it('leaves a warning alone until its token has expired', async () => {
+      await service.mark('conn-1', warn);
+      row.tokenExpiresAt = expiresAt;
+      expect(await service.expireLapsedTokens(new Date('2026-10-20T16:59:00Z'))).toBe(0);
+      expect(row.authStatus).toBe('expiring');
+    });
+
+    it('only looks at expiring warnings, never at other connectors with an expiry', async () => {
+      await service.expireLapsedTokens(new Date());
+      expect(prisma.connector.findMany).toHaveBeenCalledWith(
+        expect.objectContaining({ where: { authStatus: 'expiring', tokenExpiresAt: { lte: expect.any(Date) } } }),
+      );
+    });
+  });
+
+  it('runs the expiry sweep on a timer that does not keep the process alive', () => {
+    jest.useFakeTimers();
+    try {
+      const spy = jest.spyOn(service, 'expireLapsedTokens').mockResolvedValue(0);
+      service.onModuleInit();
+      jest.advanceTimersByTime(2 * 60 * 1000);
+      expect(spy).toHaveBeenCalledTimes(1);
+      jest.advanceTimersByTime(60 * 60 * 1000);
+      expect(spy).toHaveBeenCalledTimes(2);
+      service.onModuleDestroy();
+      jest.advanceTimersByTime(3 * 60 * 60 * 1000);
+      expect(spy).toHaveBeenCalledTimes(2);
+    } finally {
+      jest.useRealTimers();
+    }
+  });
+
   it('works without an email service (self-hosted without SMTP)', async () => {
     const quiet = new ConnectorAuthStatusService(prisma as any);
     await quiet.mark('conn-1', issue);
diff --git a/packages/backend/src/connectors/engines/connector-auth-status.service.ts b/packages/backend/src/connectors/engines/connector-auth-status.service.ts
index ba3ff35e1..27c044782 100644
--- a/packages/backend/src/connectors/engines/connector-auth-status.service.ts
+++ b/packages/backend/src/connectors/engines/connector-auth-status.service.ts
@@ -1,6 +1,7 @@
-import { Injectable, Logger, Optional } from '@nestjs/common';
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit, Optional } from '@nestjs/common';
 import { pbkdf2Sync } from 'crypto';
 import { PrismaService } from '../../common/prisma.service';
+import { connectorPageUrl } from '../../common/url.util';
 import { EmailService } from '../../settings/email.service';
 
 /**
@@ -18,7 +19,9 @@ import { EmailService } from '../../settings/email.service';
  * Statuses:
  *   needs_reauth  blocking: calls fail at once with `message` until the
  *                 credential changes (`ref` is a hash of the one refused)
- *   expiring      a warning: the token works, but stops at tokenExpiresAt
+ *   expiring      a warning: the token works, but stops at tokenExpiresAt;
+ *                 once that time has passed it becomes needs_reauth
+ *                 (token_expired), on the next call or the hourly sweep
  *
  * Nothing here is ever a secret: `ref` is a truncated SHA-256.
  */
@@ -52,20 +55,88 @@ export function credentialRef(value: unknown): string | undefined {
   return ref;
 }
 
+/** How often connectors whose token expired while still marked `expiring` are looked for. */
+const EXPIRY_SWEEP_MS = 60 * 60 * 1000;
+/** First sweep after boot: late enough to stay out of the startup work. */
+const EXPIRY_SWEEP_FIRST_MS = 2 * 60 * 1000;
+/** Connectors handled per sweep; the rest wait for the next one. */
+const EXPIRY_SWEEP_BATCH = 200;
+
+/** The status message of a token past its expiry that nothing could renew. */
+export function expiredTokenHint(expiresAt: Date, connectorId: string): string {
+  const when = expiresAt.toISOString().replace('T', ' ').slice(0, 16) + ' UTC';
+  const page = connectorPageUrl(connectorId);
+  return (
+    `The access token expired on ${when} and could not be renewed; retrying will not help. ` +
+    `Paste a new token into the connector's variables in AnythingMCP${page ? ` (${page})` : ''}. ` +
+    'For Meta, a System User token with expiry Never does not expire, and a user token is renewed ' +
+    'automatically when the app id and secret are set.'
+  );
+}
+
 /** The credential a status is recorded against: the refresh token, else the pasted token. */
 export function authCredentialRef(authConfig: Record<string, unknown> | undefined): string | undefined {
   return credentialRef(authConfig?.refreshToken) ?? credentialRef(authConfig?.token);
 }
 
 @Injectable()
-export class ConnectorAuthStatusService {
+export class ConnectorAuthStatusService implements OnModuleInit, OnModuleDestroy {
   private readonly logger = new Logger(ConnectorAuthStatusService.name);
+  private sweepTimer?: ReturnType<typeof setInterval>;
+  private firstSweep?: ReturnType<typeof setTimeout>;
 
   constructor(
     private readonly prisma: PrismaService,
     @Optional() private readonly email?: EmailService,
   ) {}
 
+  onModuleInit(): void {
+    const sweep = () => void this.expireLapsedTokens();
+    this.firstSweep = setTimeout(sweep, EXPIRY_SWEEP_FIRST_MS);
+    this.firstSweep.unref?.();
+    this.sweepTimer = setInterval(sweep, EXPIRY_SWEEP_MS);
+    this.sweepTimer.unref?.();
+  }
+
+  onModuleDestroy(): void {
+    if (this.firstSweep) clearTimeout(this.firstSweep);
+    if (this.sweepTimer) clearInterval(this.sweepTimer);
+  }
+
+  /**
+   * An `expiring` warning whose token has expired is no longer a warning.
+   * The next call notices it too (Meta refuses the token, or the renewal
+   * service sees the expiry), but a connector nobody calls stayed `expiring`
+   * for good and its owner was never told: a meta-ads token that expired on
+   * 20 Oct 2026 at 17:00 was still a warning, with no email, a day later.
+   * Each one becomes needs_reauth against the same credential, which sends
+   * the single owner email like any other refusal. Returns how many moved.
+   */
+  async expireLapsedTokens(now: Date = new Date()): Promise<number> {
+    let rows: Array<{ id: string; authStatusRef: string | null; tokenExpiresAt: Date | null }>;
+    try {
+      rows = await this.prisma.connector.findMany({
+        where: { authStatus: 'expiring', tokenExpiresAt: { lte: now } },
+        select: { id: true, authStatusRef: true, tokenExpiresAt: true },
+        orderBy: { tokenExpiresAt: 'asc' },
+        take: EXPIRY_SWEEP_BATCH,
+      });
+    } catch (err: any) {
+      this.logger.warn(`Could not look for expired connector tokens: ${err?.message ?? err}`);
+      return 0;
+    }
+    for (const row of rows) {
+      if (!row.tokenExpiresAt) continue;
+      await this.mark(row.id, {
+        status: 'needs_reauth',
+        code: 'token_expired',
+        message: expiredTokenHint(row.tokenExpiresAt, row.id),
+        ref: row.authStatusRef ?? undefined,
+      });
+    }
+    return rows.length;
+  }
+
   /**
    * Record an issue. A repeat of the stored one (same status, code and
    * credential) writes nothing, so a burst of failing calls costs one read
diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts
index 357d19cba..b00860f42 100644
--- a/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts
+++ b/packages/backend/src/knowledge-graph/kg-observational.service.spec.ts
@@ -226,7 +226,7 @@ describe('KgObservationalService.ingestOrganization', () => {
     expect(kinds).toEqual(['produces_consumes', 'same_identity']);
   });
 
-  it('skips values left by a deleted connector instead of aborting the run', async () => {
+  it('never pairs values left by a connector that no longer exists', async () => {
     const { svc, prisma } = make({
       pages: [[row('r1', 'c2', 1, 'erp_get_order')]],
       payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } },
@@ -235,18 +235,52 @@ describe('KgObservationalService.ingestOrganization', () => {
         { valueHash: 'h1', connectorId: 'gone', entity: 'customer', field: 'id', direction: 'input' },
       ],
     });
+
+    await expect(svc.ingestOrganization(ORG)).resolves.toBeDefined();
+    // No node is even attempted for it: each attempt was a failed insert,
+    // 21.8k a day in the Postgres log on cloud.
+    expect(prisma.kgNode.upsert).not.toHaveBeenCalledWith(
+      expect.objectContaining({ create: expect.objectContaining({ connectorId: 'gone' }) }),
+    );
+    expect(prisma.kgEdge.create).not.toHaveBeenCalled();
+    expect(prisma.kgConnectorState.upsert).toHaveBeenCalled();
+  });
+
+  it('skips a connector deleted during the run, asking once instead of once per value', async () => {
+    const occ = (hash: string) => [
+      { valueHash: hash, connectorId: 'c2', entity: 'order', field: 'customer_id', direction: 'output' },
+      { valueHash: hash, connectorId: 'c1', entity: 'customer', field: 'id', direction: 'input' },
+    ];
+    const { svc, prisma } = make({
+      pages: [[row('r1', 'c2', 1, 'erp_get_order')]],
+      payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } },
+      valueSeen: ['h1', 'h2', 'h3'].flatMap(occ),
+    });
+    // c1 was listed at the start of the run and deleted since.
     prisma.kgNode.upsert.mockImplementation(async (args: any) => {
-      if (args.create.connectorId === 'gone') {
+      if (args.create.connectorId === 'c1') {
         throw Object.assign(new Error('Foreign key constraint violated: kg_nodes_connector_id_fkey'), { code: 'P2003' });
       }
       return { id: `node-${args.create.connectorId}-${args.create.entity}` };
     });
 
     await expect(svc.ingestOrganization(ORG)).resolves.toBeDefined();
+    const forC1 = prisma.kgNode.upsert.mock.calls.filter((c: any[]) => c[0].create.connectorId === 'c1');
+    expect(forC1).toHaveLength(1);
     expect(prisma.kgEdge.create).not.toHaveBeenCalled();
     expect(prisma.kgConnectorState.upsert).toHaveBeenCalled();
   });
 
+  it('drops value rows of a connector that is gone when it writes them', async () => {
+    const { svc, prisma } = make({
+      pages: [[row('r1', 'c2', 1, 'erp_get_order')]],
+      payloads: { r1: { input: {}, output: { order: { customer_id: 'CUST-12345' } } } },
+    });
+    await svc.ingestOrganization(ORG);
+    const [sql] = prisma.$executeRaw.mock.calls[0];
+    expect(sql.join('?')).toMatch(/WHERE EXISTS \(SELECT 1 FROM connectors c WHERE c\.id = r\.connector_id\)/);
+  });
+
   it('writes each edge once per pass, however many values link the same nodes', async () => {
     const occ = (hash: string) => [
       { valueHash: hash, connectorId: 'c2', entity: 'order', field: 'customer_id', direction: 'output' },
diff --git a/packages/backend/src/knowledge-graph/kg-observational.service.ts b/packages/backend/src/knowledge-graph/kg-observational.service.ts
index f1a6887b9..b3132cbb1 100644
--- a/packages/backend/src/knowledge-graph/kg-observational.service.ts
+++ b/packages/backend/src/knowledge-graph/kg-observational.service.ts
@@ -128,6 +128,10 @@ export class KgObservationalService {
     const slugByConnector = new Map(
       connectors.map((c) => [c.id, deriveSlug(c.tools.map((t) => t.name))]),
     );
+    // The organization's connectors as of this run. correlate() only pairs
+    // values of these: a value row of a connector deleted since can never get
+    // a node (kg_nodes has a foreign key), so trying cost a failed insert.
+    const liveConnectors = new Set(connectors.map((c) => c.id));
     const states = await this.prisma.kgConnectorState.findMany({
       where: { organizationId },
       select: { connectorId: true, lastObservedAt: true },
@@ -279,7 +283,11 @@ export class KgObservationalService {
       }
       const hashes = [...newHashes];
       for (let i = 0; i < hashes.length; i += WRITE_CHUNK) {
-        edges += await this.correlate(organizationId, hashes.slice(i, i + WRITE_CHUNK));
+        edges += await this.correlate(
+          organizationId,
+          hashes.slice(i, i + WRITE_CHUNK),
+          liveConnectors,
+        );
         await yieldToEventLoop();
       }
 
@@ -291,7 +299,7 @@ export class KgObservationalService {
     } // while pages
 
     // Apply references edges mined from response shapes.
-    const refCache = new Map<string, string>();
+    const refCache = new Map<string, string | null>();
     for (const b of refBumps.values()) {
       const src = await this.nodeId(refCache, organizationId, b.connectorId, b.from);
       const tgt = await this.nodeId(refCache, organizationId, b.connectorId, b.to);
@@ -313,7 +321,7 @@ export class KgObservationalService {
     // Intent co-occurrence: entities whose tools satisfied the SAME captured
     // user request are related. A weak, suggested signal (awaits confirmation)
     // — this is how the chat history that led to calls extends the graph.
-    const coCache = new Map<string, string>();
+    const coCache = new Map<string, string | null>();
     for (const members of intentGroups.values()) {
       const list = [...members];
       if (list.length < 2) continue;
@@ -422,7 +430,9 @@ export class KgObservationalService {
   /**
    * One statement with seven array parameters, however many rows. Prisma's
    * createMany compiles a parameter per column per row, synchronously: 1,000
-   * rows held the production event loop for 330-630 ms per call.
+   * rows held the production event loop for 330-630 ms per call. Rows of a
+   * connector deleted while the run was going are dropped rather than
+   * failing the whole batch on the foreign key.
    */
   private async insertValueRows(rows: ValueRow[]): Promise<void> {
     if (rows.length === 0) return;
@@ -430,7 +440,7 @@ export class KgObservationalService {
     await this.prisma.$executeRaw`
       INSERT INTO kg_value_seen
         (id, organization_id, connector_id, value_hash, entity, field, direction)
-      SELECT * FROM unnest(
+      SELECT r.* FROM unnest(
         ${rows.map(() => randomUUID())}::text[],
         ${col('organizationId')}::text[],
         ${col('connectorId')}::text[],
@@ -438,7 +448,8 @@ export class KgObservationalService {
         ${col('entity')}::text[],
         ${col('field')}::text[],
         ${col('direction')}::text[]
-      )
+      ) AS r(id, organization_id, connector_id, value_hash, entity, field, direction)
+      WHERE EXISTS (SELECT 1 FROM connectors c WHERE c.id = r.connector_id)
       ON CONFLICT (organization_id, connector_id, value_hash, entity, field, direction) DO NOTHING`;
   }
 
@@ -459,6 +470,7 @@ export class KgObservationalService {
   private async correlate(
     organizationId: string,
     hashes: string[],
+    liveConnectors: ReadonlySet<string>,
   ): Promise<number> {
     const rows = await this.prisma.kgValueSeen.findMany({
       where: { organizationId, valueHash: { in: hashes } },
@@ -479,6 +491,9 @@ export class KgObservationalService {
     const byHash = new Map<string, typeof rows>();
     const seen = new Set<string>();
     for (const r of rows) {
+      // Deleting a connector deletes its rows (ON DELETE CASCADE); this
+      // covers a connector deleted during the run.
+      if (!liveConnectors.has(r.connectorId)) continue;
       const key = `${r.valueHash}|${r.connectorId}|${r.entity}|${r.direction}`;
       if (seen.has(key)) continue;
       seen.add(key);
@@ -487,7 +502,7 @@ export class KgObservationalService {
       byHash.set(r.valueHash, list);
     }
 
-    const nodeCache = new Map<string, string>(); // `${connectorId}::${entity}` -> nodeId
+    const nodeCache = new Map<string, string | null>(); // `${connectorId}::${entity}` -> nodeId
     // Many values link the same two nodes, so one pass used to write the same
     // edge again for every one of them: a read and an update per value, most
     // of the database work behind an MCP call. The bumps are counted here and
@@ -569,14 +584,15 @@ export class KgObservationalService {
   }
 
   private async nodeId(
-    cache: Map<string, string>,
+    cache: Map<string, string | null>,
     organizationId: string,
     connectorId: string,
     entity: string,
   ): Promise<string | null> {
     const key = `${connectorId}::${entity}`;
-    const cached = cache.get(key);
-    if (cached) return cached;
+    // A cached null is a connector that is gone: ask once per pass, not once
+    // per value that names it.
+    if (cache.has(key)) return cache.get(key) ?? null;
     let node: { id: string };
     try {
       node = await this.prisma.kgNode.upsert({
@@ -593,11 +609,13 @@ export class KgObservationalService {
         select: { id: true },
       });
     } catch (e: any) {
-      // kg_value_seen keeps the values of a connector after it is deleted (no
-      // foreign key there), so a match can name a connector that is gone. A
-      // node for it would violate kg_nodes_connector_id_fkey and abort the
-      // whole run for the organization, on every call: skip it instead.
-      if (e?.code === 'P2003') return null;
+      // A connector deleted while the run was going: a node for it would
+      // violate kg_nodes_connector_id_fkey and abort the whole run for the
+      // organization. Skip it instead.
+      if (e?.code === 'P2003') {
+        cache.set(key, null);
+        return null;
+      }
       throw e;
     }
     cache.set(key, node.id);
diff --git a/packages/backend/src/main.ts b/packages/backend/src/main.ts
index 7e7051c6c..f6be80277 100644
--- a/packages/backend/src/main.ts
+++ b/packages/backend/src/main.ts
@@ -29,6 +29,7 @@ import helmet from 'helmet';
 import { AppModule } from './app.module';
 import { mcpStrategy } from './mcp-server/mcp-strategy';
 import { McpAuthExceptionFilter } from './auth/mcp-auth-exception.filter';
+import { McpEmailVerificationFilter } from './auth/mcp-email-verification';
 import { validateRequiredSecretsAtStartup } from './common/secrets.util';
 import { startMetricsServer, stopMetricsServer } from './common/metrics';
 
@@ -104,8 +105,9 @@ async function bootstrap() {
     }),
   );
 
-  // Add WWW-Authenticate header to MCP 401 responses for OAuth discovery
-  app.useGlobalFilters(new McpAuthExceptionFilter());
+  // Add WWW-Authenticate header to MCP 401 responses for OAuth discovery,
+  // and answer an unverified cloud user's MCP client with what to do.
+  app.useGlobalFilters(new McpAuthExceptionFilter(), new McpEmailVerificationFilter());
 
   // Global validation
   app.useGlobalPipes(
diff --git a/packages/backend/src/mcp-server/mcp-endpoint.controller.ts b/packages/backend/src/mcp-server/mcp-endpoint.controller.ts
index dd74ce34f..0ee244e5b 100644
--- a/packages/backend/src/mcp-server/mcp-endpoint.controller.ts
+++ b/packages/backend/src/mcp-server/mcp-endpoint.controller.ts
@@ -16,7 +16,7 @@ import { SkipThrottle, Throttle } from '@nestjs/throttler';
 import { Request, Response } from 'express';
 import { randomUUID } from 'crypto';
 import { z } from 'zod';
-import { McpServer, createMcpHandler, isLegacyRequest } from '@modelcontextprotocol/server';
+import { McpServer, isLegacyRequest } from '@modelcontextprotocol/server';
 import {
   NodeStreamableHTTPServerTransport as StreamableHTTPServerTransport,
   toNodeHandler,
@@ -25,6 +25,7 @@ import {
 import { McpCombinedAuthGuard } from '../auth/mcp-combined-auth.guard';
 import { McpPrincipalRateLimitGuard } from '../auth/mcp-principal-rate-limit.guard';
 import { mcpHttpTransport } from './mcp-strategy';
+import { createQuietMcpHandler } from './mcp-handler.util';
 import { toolVisibilityRole } from './mcp-server.service';
 import { McpServersService } from '../mcp-servers/mcp-servers.service';
 import { McpSessionManager } from '../mcp-servers/mcp-session.manager';
@@ -1296,7 +1297,8 @@ export class McpEndpointController {
       return;
     }
 
-    const handler = createMcpHandler(() => buildServer(), {
+    // Built per request; its JSON-mode warning is logged once per process.
+    const handler = createQuietMcpHandler(() => buildServer(), {
       legacy: 'stateless',
       // Mapped from the old `enableJsonResponse` boolean, deliberately NOT to
       // 'auto'. 'auto' answers with JSON whenever it can and upgrades to SSE
diff --git a/packages/backend/src/mcp-server/mcp-handler.util.spec.ts b/packages/backend/src/mcp-server/mcp-handler.util.spec.ts
new file mode 100644
index 000000000..5075c490a
--- /dev/null
+++ b/packages/backend/src/mcp-server/mcp-handler.util.spec.ts
@@ -0,0 +1,49 @@
+import { Logger } from '@nestjs/common';
+import { McpServer } from '@modelcontextprotocol/server';
+import {
+  createQuietMcpHandler,
+  JSON_MODE_WARNING_PREFIX,
+  resetJsonModeWarningForTests,
+} from './mcp-handler.util';
+
+/**
+ * The SDK printed its JSON-mode warning once per handler, and the endpoint
+ * builds a handler per request: 61k lines a day in the cloud log.
+ */
+describe('createQuietMcpHandler', () => {
+  const factory = () => new McpServer({ name: 't', version: '1' });
+  let warn: jest.SpyInstance;
+  let nestWarn: jest.SpyInstance;
+
+  beforeEach(() => {
+    resetJsonModeWarningForTests();
+    warn = jest.spyOn(console, 'warn').mockImplementation(() => undefined);
+    nestWarn = jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
+  });
+
+  afterEach(() => jest.restoreAllMocks());
+
+  it('logs the JSON-mode warning once per process, however many handlers are built', async () => {
+    for (let i = 0; i < 5; i++) {
+      await createQuietMcpHandler(factory, { legacy: 'stateless', responseMode: 'json' }).close();
+    }
+    expect(warn).not.toHaveBeenCalledWith(expect.stringContaining(JSON_MODE_WARNING_PREFIX));
+    expect(nestWarn).toHaveBeenCalledTimes(1);
+    expect(nestWarn).toHaveBeenCalledWith(expect.stringContaining(JSON_MODE_WARNING_PREFIX));
+  });
+
+  it('puts console.warn back, and lets every other warning through', async () => {
+    const before = console.warn;
+    await createQuietMcpHandler(factory, { responseMode: 'json' }).close();
+    expect(console.warn).toBe(before);
+    console.warn('something else');
+    expect(warn).toHaveBeenCalledWith('something else');
+  });
+
+  it('leaves SSE handlers untouched', async () => {
+    const before = console.warn;
+    await createQuietMcpHandler(factory, { responseMode: 'sse' }).close();
+    expect(console.warn).toBe(before);
+    expect(nestWarn).not.toHaveBeenCalled();
+  });
+});
diff --git a/packages/backend/src/mcp-server/mcp-handler.util.ts b/packages/backend/src/mcp-server/mcp-handler.util.ts
new file mode 100644
index 000000000..c88460d04
--- /dev/null
+++ b/packages/backend/src/mcp-server/mcp-handler.util.ts
@@ -0,0 +1,51 @@
+import { Logger } from '@nestjs/common';
+import { createMcpHandler } from '@modelcontextprotocol/server';
+
+/** Start of the warning `createMcpHandler` prints for `responseMode: 'json'`. */
+export const JSON_MODE_WARNING_PREFIX = "responseMode: 'json' drops mid-call notifications";
+
+const logger = new Logger('McpHandler');
+let jsonModeWarningLogged = false;
+
+/**
+ * `createMcpHandler`, with its `responseMode: 'json'` warning logged once per
+ * process instead of once per handler.
+ *
+ * The SDK prints the warning with console.warn every time a handler is built
+ * in JSON mode, and the endpoints build one per request: 61k plain-text lines
+ * a day on cloud, a third of what pushed the docker log past its cap within
+ * six hours. JSON mode is chosen on purpose there (MCP_STREAMABLE_JSON_RESPONSE,
+ * for clients that cannot read SSE), so the warning says nothing new after the
+ * first time.
+ *
+ * Only that one message is held back, and only while the handler is built:
+ * `createMcpHandler` is synchronous, so console.warn is restored before any
+ * other code runs, and any other warning printed meanwhile goes through.
+ */
+export function createQuietMcpHandler(
+  ...args: Parameters<typeof createMcpHandler>
+): ReturnType<typeof createMcpHandler> {
+  if (args[1]?.responseMode !== 'json') return createMcpHandler(...args);
+  const original = console.warn;
+  console.warn = (...warnArgs: unknown[]) => {
+    const first = warnArgs[0];
+    if (typeof first === 'string' && first.startsWith(JSON_MODE_WARNING_PREFIX)) {
+      if (!jsonModeWarningLogged) {
+        jsonModeWarningLogged = true;
+        logger.warn(`${first} (logged once per process)`);
+      }
+      return;
+    }
+    original.apply(console, warnArgs);
+  };
+  try {
+    return createMcpHandler(...args);
+  } finally {
+    console.warn = original;
+  }
+}
+
+/** Test hook: forget that the warning was logged. */
+export function resetJsonModeWarningForTests(): void {
+  jsonModeWarningLogged = false;
+}