diff --git a/DESIGN-inline-policy.md b/DESIGN-inline-policy.md
index 7408d59..41e4033 100644
--- a/DESIGN-inline-policy.md
+++ b/DESIGN-inline-policy.md
@@ -218,7 +218,7 @@ consumers:
 **Behavior:**
 
 - **No `consumers` section on localhost / unix socket (default):** Inline policy is allowed for all callers (local DX). Non-loopback binds without consumers fail closed (see DR-037).
-- **`consumers` section present:** Callers need `chat` plus `inline_policy`, with a valid `x-abbenay-token`. Unauthorized requests receive `PERMISSION_DENIED`.
+- **`consumers` section present:** Callers need `chat` plus `inline_policy`, with a valid `x-abbenay-token`. Missing or unrecognized tokens receive `UNAUTHENTICATED`; a recognized consumer lacking `inline_policy` receives `PERMISSION_DENIED`.
 
 The consumer model provides per-app granularity — the admin can trust APME without trusting all Python clients. Token-based auth was chosen over client-type gating for this reason (see DR-024 / DR-037).
 
diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md
index 3412bba..155cd42 100644
--- a/docs/CONFIGURATION.md
+++ b/docs/CONFIGURATION.md
@@ -346,9 +346,19 @@ consumers:
 | Unix socket or loopback TCP (`127.0.0.1`, `::1`) | Allow-all (local DX) | Token + capability required for sensitive RPCs |
 | Non-loopback TCP (`0.0.0.0`, LAN IP, …) | **Refuse to start** unless `--allow-open-auth` or `--insecure` | Token + capability required |
 
-Token comparison uses `crypto.timingSafeEqual` (equal-length buffers). Wrong
-or missing tokens receive `PERMISSION_DENIED`. Health/status/list discovery
-RPCs stay ungated so probes and local tooling keep working.
+Token comparison uses `crypto.timingSafeEqual` (equal-length buffers). Missing
+or unrecognized tokens receive `UNAUTHENTICATED`. A recognized consumer that
+lacks the required capability receives `PERMISSION_DENIED` (same denial
+message as an unrecognized token so the string does not leak validity).
+`HealthCheck`, `GetStatus`, and `ListModels` stay ungated so probes and local
+tooling keep working (`DiscoverModels` still needs the `providers`
+capability). Session CRUD RPCs (`CreateSession` / `GetSession` /
+`ListSessions` / `DeleteSession`) are not capability-gated: no token still
+maps to the `local` owner, but a presented-but-unrecognized token is
+`UNAUTHENTICATED` (DR-049). `SessionChat` and `SummarizeSession` still
+require the `chat` capability. The Python client only attaches
+`x-abbenay-token` on a Unix socket or TLS channel; plaintext TCP +
+`token=` raises before the RPC is sent.
 
 > **WARNING — open auth:** `--allow-open-auth` (or `--insecure`, which implies
 > it) on a non-loopback bind restores allow-all when `consumers` is empty.
@@ -366,6 +376,7 @@ Every session is stamped with an `owner` principal:
 | HTTP + `X-Abbenay-Session-Owner: <name>` | `http:<fingerprint>:<name>` |
 | gRPC with matching consumer token | `consumer:<name>` |
 | gRPC without consumer token | `local` |
+| gRPC with unrecognized consumer token (`consumers` configured) | RPC rejected (`UNAUTHENTICATED`) |
 
 List/get/delete/chat only return sessions for the caller's owner. Cross-owner
 access returns 404 (not 403) so session IDs are not leaked across principals.
diff --git a/docs/SECURITY.md b/docs/SECURITY.md
index 3aee834..3dcd529 100644
--- a/docs/SECURITY.md
+++ b/docs/SECURITY.md
@@ -160,11 +160,12 @@ unless `--allow-open-auth` or `--insecure` is set. Configure consumers per
 
 ```bash
 # After consumers + --grpc-tls on 0.0.0.0: sensitive RPCs need x-abbenay-token
-# Wrong/missing token → PERMISSION_DENIED (see consumer-auth tests / client docs)
+# Missing/wrong token → UNAUTHENTICATED; valid token without capability → PERMISSION_DENIED
 ```
 
 **Pass:** start fails with empty consumers on `0.0.0.0`; with consumers, wrong
-token is denied on gated RPCs.
+token is denied on gated RPCs. Python clients must send `x-abbenay-token` only
+over Unix socket or TLS (plaintext TCP + `token=` is rejected client-side).
 **Fail:** non-loopback gRPC allows all callers with no consumers and no opt-in.
 
 ### 6. MCP HTTP endpoint
@@ -201,3 +202,4 @@ need air-gap / offline posture:
 | DR-029 | Fail-closed TLS for non-loopback gRPC TCP |
 | DR-030 | Secure-by-default HTTP (auth, CORS, bind) |
 | DR-038 | Air-gap docs must not claim isolation equals security |
+| DR-049 | Unrecognized gRPC consumer tokens fail closed for session ownership |
diff --git a/docs/decisions.md b/docs/decisions.md
index 0ff7fa4..e77ddca 100644
--- a/docs/decisions.md
+++ b/docs/decisions.md
@@ -883,3 +883,26 @@ Abbenay remains source of truth for provider credentials. Placing secrets
 next to config reuses the existing config volume contract without introducing
 a separate Gateway SoT or a cloud KMS dependency. Encryption can layer later
 without changing the `(secret_store, secret_name)` address model.
+
+---
+
+## DR-049: Unrecognized gRPC consumer tokens fail closed for session ownership
+
+**Date:** 2026-08-15
+**Decision:** When `consumers` is configured and a caller presents
+`x-abbenay-token` that matches no consumer, session RPCs
+(`CreateSession`, `GetSession`, `ListSessions`, `DeleteSession`,
+`SessionChat`, `SummarizeSession`) return `UNAUTHENTICATED` instead of
+mapping the caller to `local`. Omitting the header still maps to `local`
+(unix-socket / local CLI DX). Privileged RPCs (`authorizeConsumer`) use the
+same split: missing or unrecognized token → `UNAUTHENTICATED`; recognized
+consumer lacking the capability → `PERMISSION_DENIED`. Denial *messages* for
+unrecognized token vs missing capability stay identical so the string does
+not leak token validity; status codes follow gRPC semantics.
+**Rationale:** PR #62 / DR-031 stamped sessions with an owner, but
+`resolveGrpcSessionOwner` treated a present-but-wrong token like no token.
+That let a caller with a bad consumer token create and list sessions in the
+CLI namespace. Fail-closed on unrecognized tokens closes that hole without
+breaking no-token local DX. Tracked as
+[issue #71](https://github.com/redhat-developer/abbenay/issues/71).
+
diff --git a/packages/daemon/src/daemon/server/abbenay-service.test.ts b/packages/daemon/src/daemon/server/abbenay-service.test.ts
index 0371f1d..1075c6e 100644
--- a/packages/daemon/src/daemon/server/abbenay-service.test.ts
+++ b/packages/daemon/src/daemon/server/abbenay-service.test.ts
@@ -469,7 +469,19 @@ describe('protoToPolicyConfig', () => {
 describe('resolveGrpcSessionOwner', () => {
   it('returns local owner without matching token', () => {
     const call = { metadata: new grpc.Metadata() };
-    expect(resolveGrpcSessionOwner(call, { providers: {} })).toBe('local');
+    expect(resolveGrpcSessionOwner(call, { providers: {} })).toEqual({ ok: true, owner: 'local' });
+  });
+
+  it('returns local owner when consumers are configured but no token is sent', async () => {
+    await withEnv('TEST_OWNER_TOKEN', 'owner-tok', () => {
+      const call = { metadata: new grpc.Metadata() };
+      const config: ConfigFile = {
+        consumers: {
+          apme: { token_env: 'TEST_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      };
+      expect(resolveGrpcSessionOwner(call, config)).toEqual({ ok: true, owner: 'local' });
+    });
   });
 
   it('returns consumer owner when token matches', () => {
@@ -484,12 +496,54 @@ describe('resolveGrpcSessionOwner', () => {
           apme: { token_env: 'TEST_OWNER_TOKEN', capabilities: { chat: true } },
         },
       };
-      expect(resolveGrpcSessionOwner(call, config)).toBe('consumer:apme');
+      expect(resolveGrpcSessionOwner(call, config)).toEqual({
+        ok: true,
+        owner: 'consumer:apme',
+      });
     } finally {
       if (prev === undefined) delete process.env.TEST_OWNER_TOKEN;
       else process.env.TEST_OWNER_TOKEN = prev;
     }
   });
+
+  it('fails closed when a token is presented but matches no consumer', async () => {
+    await withEnv('TEST_OWNER_TOKEN', 'owner-tok', () => {
+      const metadata = new grpc.Metadata();
+      metadata.add('x-abbenay-token', 'wrong-tok');
+      const call = { metadata };
+      const config: ConfigFile = {
+        consumers: {
+          apme: { token_env: 'TEST_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      };
+      expect(resolveGrpcSessionOwner(call, config)).toEqual({
+        ok: false,
+        reason: 'Consumer token not recognized.',
+      });
+    });
+  });
+
+  it('fails closed on empty presented token when consumers are configured', async () => {
+    await withEnv('TEST_OWNER_TOKEN', 'owner-tok', () => {
+      const metadata = new grpc.Metadata();
+      metadata.add('x-abbenay-token', '');
+      const config: ConfigFile = {
+        consumers: {
+          apme: { token_env: 'TEST_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      };
+      expect(resolveGrpcSessionOwner({ metadata }, config).ok).toBe(false);
+    });
+  });
+
+  it('ignores unrecognized tokens when no consumers are configured', () => {
+    const metadata = new grpc.Metadata();
+    metadata.add('x-abbenay-token', 'stray-tok');
+    expect(resolveGrpcSessionOwner({ metadata }, { providers: {} })).toEqual({
+      ok: true,
+      owner: 'local',
+    });
+  });
 });
 
 // ── deprecated auth wrappers ─────────────────────────────────────────────────
@@ -922,6 +976,82 @@ describe('createAbbenayService handlers', () => {
     expect(state.mcpClientPool.disconnectByScope).toHaveBeenCalledWith('sess-1');
   });
 
+  it('session RPCs reject unrecognized consumer tokens as UNAUTHENTICATED', async () => {
+    await withEnv('SESS_OWNER_TOKEN', 'good-owner', async () => {
+      mockLoadConfig.mockReturnValue({
+        providers: {},
+        consumers: {
+          apme: { token_env: 'SESS_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      });
+      const state = createMockState();
+      const service = createServiceHandlers(state);
+      const badMeta = new grpc.Metadata();
+      badMeta.add('x-abbenay-token', 'wrong-owner');
+
+      const created = await invokeUnary(service.CreateSession, { model: 'mock/echo' }, badMeta);
+      expect(created.error?.code).toBe(grpc.status.UNAUTHENTICATED);
+      expect(state.sessionStore.create).not.toHaveBeenCalled();
+
+      const listed = await invokeUnary(service.ListSessions, {}, badMeta);
+      expect(listed.error?.code).toBe(grpc.status.UNAUTHENTICATED);
+
+      const got = await invokeUnary(service.GetSession, { session_id: 'sess-1' }, badMeta);
+      expect(got.error?.code).toBe(grpc.status.UNAUTHENTICATED);
+
+      const deleted = await invokeUnary(service.DeleteSession, { session_id: 'sess-1' }, badMeta);
+      expect(deleted.error?.code).toBe(grpc.status.UNAUTHENTICATED);
+    });
+  });
+
+  it('session RPCs keep no-token callers in the local namespace when consumers are configured', async () => {
+    await withEnv('SESS_OWNER_TOKEN', 'good-owner', async () => {
+      mockLoadConfig.mockReturnValue({
+        providers: {},
+        consumers: {
+          apme: { token_env: 'SESS_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      });
+      const state = createMockState();
+      const service = createServiceHandlers(state);
+
+      const created = await invokeUnary(service.CreateSession, { model: 'mock/echo', topic: 't' });
+      expect(created.error).toBeNull();
+      expect(state.sessionStore.create).toHaveBeenCalledWith(
+        'mock/echo',
+        't',
+        undefined,
+        undefined,
+        'local',
+      );
+    });
+  });
+
+  it('CreateSession stamps matching consumer token as consumer owner', async () => {
+    await withEnv('SESS_OWNER_TOKEN', 'good-owner', async () => {
+      mockLoadConfig.mockReturnValue({
+        providers: {},
+        consumers: {
+          apme: { token_env: 'SESS_OWNER_TOKEN', capabilities: { chat: true } },
+        },
+      });
+      const state = createMockState();
+      const service = createServiceHandlers(state);
+      const meta = new grpc.Metadata();
+      meta.add('x-abbenay-token', 'good-owner');
+
+      const created = await invokeUnary(service.CreateSession, { model: 'mock/echo' }, meta);
+      expect(created.error).toBeNull();
+      expect(state.sessionStore.create).toHaveBeenCalledWith(
+        'mock/echo',
+        undefined,
+        undefined,
+        undefined,
+        'consumer:apme',
+      );
+    });
+  });
+
   it('SummarizeSession returns cached summary when counts match', async () => {
     const state = createMockState({
       sessionStore: {
@@ -1396,7 +1526,7 @@ describe('createAbbenayService handlers', () => {
     const state = createMockState();
     const service = createServiceHandlers(state, DEFAULT_CONSUMER_AUTH_CONTEXT);
     const { error } = await invokeUnary(service.GetSecret, { key: 'K' });
-    expect(error?.code).toBe(grpc.status.PERMISSION_DENIED);
+    expect(error?.code).toBe(grpc.status.UNAUTHENTICATED);
   });
 
   it('Chat returns INVALID_ARGUMENT when model missing', async () => {
@@ -1416,7 +1546,7 @@ describe('createAbbenayService handlers', () => {
     expect(written[0]).toEqual({ error: { code: 'INVALID_ARGUMENT', message: 'Model is required' } });
   });
 
-  it('Chat stream emits PERMISSION_DENIED via gRPC error when auth fails', async () => {
+  it('Chat stream emits UNAUTHENTICATED via gRPC error when token is missing', async () => {
     mockLoadConfig.mockReturnValue({
       providers: {},
       consumers: {
@@ -1444,7 +1574,7 @@ describe('createAbbenayService handlers', () => {
     service.Chat(call as never);
     await vi.waitFor(() => expect(call.emit).toHaveBeenCalledWith('error', expect.any(Error)));
     const err = (call.emit as ReturnType<typeof vi.fn>).mock.calls[0][1] as Error & { code?: number };
-    expect(err.code).toBe(grpc.status.PERMISSION_DENIED);
+    expect(err.code).toBe(grpc.status.UNAUTHENTICATED);
   });
 
   it('SessionChat validates session_id and message content', async () => {
@@ -1524,7 +1654,7 @@ describe('createAbbenayService handlers', () => {
       server_id: 'dyn',
       transport: { type: 'stdio', command: 'npx', args: ['x'] },
     });
-    expect(error?.code).toBe(grpc.status.PERMISSION_DENIED);
+    expect(error?.code).toBe(grpc.status.UNAUTHENTICATED);
     expect(error?.message).toMatch(/consumer authentication/i);
   });
 
diff --git a/packages/daemon/src/daemon/server/abbenay-service.ts b/packages/daemon/src/daemon/server/abbenay-service.ts
index 3e95c68..3dd00f7 100644
--- a/packages/daemon/src/daemon/server/abbenay-service.ts
+++ b/packages/daemon/src/daemon/server/abbenay-service.ts
@@ -47,6 +47,7 @@ import {
   authorizeConsumer,
   matchConsumerByToken,
   hasConfiguredConsumers,
+  extractPresentedConsumerToken,
   DEFAULT_CONSUMER_AUTH_CONTEXT,
   type ConsumerAuthContext,
   type ConsumerCapability,
@@ -58,11 +59,13 @@ import {
 import { StdioCommandDeniedError } from '../stdio-command-policy.js';
 
 export type { ConsumerAuthContext, ConsumerCapability, AuthResult };
+export type { AuthDenyCode } from './consumer-auth.js';
 export {
   authorizeConsumer,
   matchConsumerByToken,
   DEFAULT_CONSUMER_AUTH_CONTEXT,
   hasConfiguredConsumers,
+  extractPresentedConsumerToken,
   buildConsumerAuthContext,
   assertConsumersConfiguredForBind,
   resolveAllowOpenAuth,
@@ -435,6 +438,21 @@ function toRole(protoRole: string | number): string {
   }
 }
 
+function grpcStatusForAuth(auth: AuthResult): grpc.status {
+  return auth.code === 'UNAUTHENTICATED'
+    ? grpc.status.UNAUTHENTICATED
+    : grpc.status.PERMISSION_DENIED;
+}
+
+function emitGrpcStatus(
+  call: grpc.ServerWritableStream<unknown, object>,
+  code: grpc.status,
+  message: string,
+): void {
+  const err = Object.assign(new Error(message), { code, details: message });
+  call.emit('error', err);
+}
+
 /**
  * Deny a unary RPC when consumer auth fails. Returns true when the call may proceed.
  */
@@ -447,7 +465,7 @@ function requireCapability(
   const auth = authorizeConsumer(call, loadConfig() || { providers: {} }, capability, authContext);
   if (!auth.allowed) {
     callback({
-      code: grpc.status.PERMISSION_DENIED,
+      code: grpcStatusForAuth(auth),
       message: auth.reason || 'Permission denied',
     });
     return false;
@@ -460,7 +478,7 @@ function requireCapability(
 
 /**
  * Deny a server-streaming RPC when consumer auth fails. Returns auth on success, null when denied.
- * Emits a gRPC PERMISSION_DENIED status (same as unary gates) so clients see a real RPC error,
+ * Emits UNAUTHENTICATED or PERMISSION_DENIED (same as unary gates) so clients see a real RPC error,
  * not only an in-band ChatChunk error.
  */
 function requireCapabilityStream(
@@ -470,16 +488,31 @@ function requireCapabilityStream(
 ): AuthResult | null {
   const auth = authorizeConsumer(call, loadConfig() || { providers: {} }, capability, authContext);
   if (!auth.allowed) {
-    const err = Object.assign(new Error(auth.reason || 'Permission denied'), {
-      code: grpc.status.PERMISSION_DENIED,
-      details: auth.reason || 'Permission denied',
-    });
-    call.emit('error', err);
+    emitGrpcStatus(call, grpcStatusForAuth(auth), auth.reason || 'Permission denied');
     return null;
   }
   return auth;
 }
 
+/**
+ * Resolve session owner or fail the unary RPC with UNAUTHENTICATED.
+ * Returns null when the call was rejected.
+ */
+function requireSessionOwner(
+  call: { metadata: grpc.Metadata },
+  callback: grpc.sendUnaryData<object>,
+): string | null {
+  const resolved = resolveGrpcSessionOwner(call, loadConfig());
+  if (!resolved.ok) {
+    callback({
+      code: grpc.status.UNAUTHENTICATED,
+      message: resolved.reason,
+    });
+    return null;
+  }
+  return resolved.owner;
+}
+
 /**
  * Create the Abbenay service handlers.
  *
@@ -734,7 +767,7 @@ export function createAbbenayService(
           authContext,
         );
         if (!auth.allowed) {
-          call.write({ error: { code: 'PERMISSION_DENIED', message: auth.reason } });
+          call.write({ error: { code: auth.code || 'PERMISSION_DENIED', message: auth.reason } });
           call.end();
           return;
         }
@@ -1321,7 +1354,8 @@ export function createAbbenayService(
         callback({ code: grpc.status.INVALID_ARGUMENT, message: 'model is required' });
         return;
       }
-      const owner = resolveGrpcSessionOwner(call, loadConfig());
+      const owner = requireSessionOwner(call, callback);
+      if (!owner) return;
       state.sessionStore.create(model, topic || undefined, undefined, metadata, owner).then((session) => {
         callback(null, sessionToProto(session));
       }).catch((error: unknown) => {
@@ -1339,7 +1373,8 @@ export function createAbbenayService(
         callback({ code: grpc.status.INVALID_ARGUMENT, message: 'session_id is required' });
         return;
       }
-      const owner = resolveGrpcSessionOwner(call, loadConfig());
+      const owner = requireSessionOwner(call, callback);
+      if (!owner) return;
       state.sessionStore.getOwned(id, owner, includeMessages).then((session) => {
         callback(null, sessionToProto(session));
       }).catch((error: unknown) => {
@@ -1356,7 +1391,8 @@ export function createAbbenayService(
       const rawOffset = call.request.offset;
       const limit = rawLimit == null || rawLimit < 0 ? undefined : rawLimit;
       const offset = rawOffset == null || rawOffset < 0 ? undefined : rawOffset;
-      const owner = resolveGrpcSessionOwner(call, loadConfig());
+      const owner = requireSessionOwner(call, callback);
+      if (!owner) return;
       state.sessionStore.list({ model, limit, offset, owner }).then((result) => {
         callback(null, {
           sessions: result.sessions.map(summaryToProto),
@@ -1376,7 +1412,8 @@ export function createAbbenayService(
         callback({ code: grpc.status.INVALID_ARGUMENT, message: 'session_id is required' });
         return;
       }
-      const owner = resolveGrpcSessionOwner(call, loadConfig());
+      const owner = requireSessionOwner(call, callback);
+      if (!owner) return;
       state.sessionStore.deleteOwned(id, owner).then(async () => {
         // Clean up session-scoped dynamic MCP servers
         await state.mcpClientPool.disconnectByScope(id);
@@ -1446,7 +1483,7 @@ export function createAbbenayService(
           authContext,
         );
         if (!auth.allowed) {
-          call.write({ error: { code: 'PERMISSION_DENIED', message: auth.reason } });
+          call.write({ error: { code: auth.code || 'PERMISSION_DENIED', message: auth.reason } });
           call.end();
           return;
         }
@@ -1455,9 +1492,15 @@ export function createAbbenayService(
         }
       }
 
+      const ownerResult = resolveGrpcSessionOwner(call, loadConfig());
+      if (!ownerResult.ok) {
+        emitGrpcStatus(call, grpc.status.UNAUTHENTICATED, ownerResult.reason);
+        return;
+      }
+      const owner = ownerResult.owner;
+
       (async () => {
         try {
-          const owner = resolveGrpcSessionOwner(call, loadConfig());
           const session = await state.sessionStore.getOwned(sessionId, owner, true);
           await state.sessionStore.appendMessage(sessionId, chatMessage);
 
@@ -1567,9 +1610,11 @@ export function createAbbenayService(
         return;
       }
 
+      const owner = requireSessionOwner(call, callback);
+      if (!owner) return;
+
       (async () => {
         try {
-          const owner = resolveGrpcSessionOwner(call, loadConfig());
           const session = await state.sessionStore.getOwned(sessionId, owner, true);
           const userCount = session.messages.filter((m) => m.role === 'user').length;
 
@@ -1668,7 +1713,7 @@ export function createAbbenayService(
             `(server_id=${serverId}, transport=${transport.type})`,
         );
         callback({
-          code: grpc.status.PERMISSION_DENIED,
+          code: grpcStatusForAuth(auth),
           message: auth.reason || 'Permission denied',
         });
         return;
@@ -2622,24 +2667,34 @@ function transportProtoToConfig(transport: McpTransportProto): McpServerConfig {
   throw new Error(`Unknown transport type: "${type}". Must be "stdio", "http", or "sse".`);
 }
 
-// ── Consumer authorization (DR-024 / DR-025 / DR-037) ─────────────────
+// ── Consumer authorization (DR-024 / DR-025 / DR-037 / DR-049) ─────────
+
+export type GrpcSessionOwnerResult =
+  | { ok: true; owner: string }
+  | { ok: false; reason: string };
 
 /**
  * Resolve the session owner principal for a gRPC call.
  * - Matching consumer token → `consumer:<name>`
- * - Otherwise (local CLI / VS Code / no consumers) → `local`
+ * - No token → `local` (unix-socket / local CLI DX)
+ * - Token presented but unrecognized while `consumers` is configured → fail closed
  */
 export function resolveGrpcSessionOwner(
   call: { metadata: grpc.Metadata },
   config: ConfigFile | null,
-): string {
-  const metadata = call.metadata.get('x-abbenay-token');
-  const token = metadata.length > 0 ? String(metadata[0]) : undefined;
+): GrpcSessionOwnerResult {
+  const token = extractPresentedConsumerToken(call);
   const name = matchConsumerByToken(config, token);
   if (name) {
-    return `consumer:${name}`;
+    return { ok: true, owner: `consumer:${name}` };
+  }
+  if (token !== undefined && hasConfiguredConsumers(config)) {
+    return {
+      ok: false,
+      reason: 'Consumer token not recognized.',
+    };
   }
-  return LOCAL_SESSION_OWNER;
+  return { ok: true, owner: LOCAL_SESSION_OWNER };
 }
 
 /** @deprecated Use authorizeConsumer(call, config, 'mcp_register') instead. */
diff --git a/packages/daemon/src/daemon/server/consumer-auth.test.ts b/packages/daemon/src/daemon/server/consumer-auth.test.ts
index 05f2597..edb341d 100644
--- a/packages/daemon/src/daemon/server/consumer-auth.test.ts
+++ b/packages/daemon/src/daemon/server/consumer-auth.test.ts
@@ -222,6 +222,7 @@ describe('authorizeConsumer', () => {
       { loopbackOnly: false, allowOpenAuth: false },
     );
     expect(result.allowed).toBe(false);
+    expect(result.code).toBe('PERMISSION_DENIED');
     expect(result.reason).toMatch(/beyond localhost|consumers/i);
   });
 
@@ -243,11 +244,12 @@ describe('authorizeConsumer', () => {
         },
       }, 'inline_policy');
       expect(result.allowed).toBe(false);
+      expect(result.code).toBe('UNAUTHENTICATED');
       expect(result.reason).toContain('x-abbenay-token');
     });
   });
 
-  it('rejects wrong token', async () => {
+  it('rejects wrong token as unauthenticated', async () => {
     await withEnv('TEST_TOKEN', 'secret123', () => {
       const result = authorizeConsumer(mockGrpcCall('wrong-token'), {
         consumers: {
@@ -255,6 +257,7 @@ describe('authorizeConsumer', () => {
         },
       }, 'secrets');
       expect(result.allowed).toBe(false);
+      expect(result.code).toBe('UNAUTHENTICATED');
       expect(result.reason).toContain('not recognized');
     });
   });
@@ -283,6 +286,7 @@ describe('authorizeConsumer', () => {
           },
         }, capability);
         expect(result.allowed, capability).toBe(false);
+        expect(result.code, capability).toBe('PERMISSION_DENIED');
       }
     });
   });
@@ -328,6 +332,7 @@ describe('authorizeConsumer', () => {
         },
       }, 'secrets');
       expect(result.allowed).toBe(false);
+      expect(result.code).toBe('PERMISSION_DENIED');
     });
   });
 
@@ -339,6 +344,7 @@ describe('authorizeConsumer', () => {
         },
       }, 'shutdown');
       expect(result.allowed).toBe(false);
+      expect(result.code).toBe('PERMISSION_DENIED');
     });
   });
 });
diff --git a/packages/daemon/src/daemon/server/consumer-auth.ts b/packages/daemon/src/daemon/server/consumer-auth.ts
index 7c78833..3788444 100644
--- a/packages/daemon/src/daemon/server/consumer-auth.ts
+++ b/packages/daemon/src/daemon/server/consumer-auth.ts
@@ -5,7 +5,9 @@
  * - Non-loopback bind: empty `consumers` fails closed unless explicit open mode
  *   (`--allow-open-auth` or `--insecure`).
  * - When `consumers` is configured: sensitive RPCs require a matching token
- *   (timing-safe) and the requested capability.
+ *   (timing-safe) and the requested capability. Missing/unrecognized tokens
+ *   are UNAUTHENTICATED; a recognized consumer lacking the capability is
+ *   PERMISSION_DENIED.
  */
 
 import * as crypto from 'node:crypto';
@@ -16,10 +18,19 @@ import { isLoopbackHost } from '../grpc-tls.js';
 /** Capabilities a consumer may be granted. */
 export type ConsumerCapability = keyof ConsumerCapabilities;
 
+export type AuthDenyCode = 'UNAUTHENTICATED' | 'PERMISSION_DENIED';
+
 export interface AuthResult {
   allowed: boolean;
   consumer?: string;
   reason?: string;
+  /**
+   * Set when `allowed` is false.
+   * UNAUTHENTICATED = missing/unrecognized token when consumers are configured.
+   * PERMISSION_DENIED = recognized consumer lacking the capability, or empty
+   * consumers on a non-loopback bind without open auth.
+   */
+  code?: AuthDenyCode;
 }
 
 /**
@@ -175,9 +186,18 @@ export function matchConsumerByToken(
   return matched;
 }
 
-function extractToken(call: { metadata: grpc.Metadata }): string | undefined {
+/**
+ * First `x-abbenay-token` value when the header is present.
+ * Distinguishes "header omitted" (`undefined`) from "header sent" (including `""`).
+ */
+export function extractPresentedConsumerToken(
+  call: { metadata: grpc.Metadata },
+): string | undefined {
   const metadata = call.metadata.get('x-abbenay-token');
-  return metadata.length > 0 ? String(metadata[0]) : undefined;
+  if (metadata.length === 0) {
+    return undefined;
+  }
+  return String(metadata[0]);
 }
 
 /**
@@ -197,16 +217,18 @@ export function authorizeConsumer(
     }
     return {
       allowed: false,
+      code: 'PERMISSION_DENIED',
       reason:
         'Consumer authentication is required when gRPC is bound beyond localhost. ' +
         'Configure a consumers section in config.yaml, or restart with --allow-open-auth / --insecure.',
     };
   }
 
-  const token = extractToken(call);
-  if (!token) {
+  const token = extractPresentedConsumerToken(call);
+  if (token === undefined || token === '') {
     return {
       allowed: false,
+      code: 'UNAUTHENTICATED',
       reason: `${CAPABILITY_LABELS[capability]} requires consumer authentication. Set the x-abbenay-token gRPC metadata header.`,
     };
   }
@@ -215,6 +237,8 @@ export function authorizeConsumer(
   if (!name) {
     return {
       allowed: false,
+      code: 'UNAUTHENTICATED',
+      // Same message as missing-capability to avoid leaking token validity in the string.
       reason: `Consumer token not recognized or lacks ${CAPABILITY_LABELS[capability]} capability.`,
     };
   }
@@ -226,6 +250,7 @@ export function authorizeConsumer(
 
   return {
     allowed: false,
+    code: 'PERMISSION_DENIED',
     reason: `Consumer token not recognized or lacks ${CAPABILITY_LABELS[capability]} capability.`,
   };
 }
diff --git a/packages/daemon/tests/integration/consumer-auth.test.ts b/packages/daemon/tests/integration/consumer-auth.test.ts
index f5fbbbd..03768ef 100644
--- a/packages/daemon/tests/integration/consumer-auth.test.ts
+++ b/packages/daemon/tests/integration/consumer-auth.test.ts
@@ -127,6 +127,16 @@ async function expectPermissionDenied(promise: Promise<unknown>): Promise<void>
   }
 }
 
+async function expectUnauthenticated(promise: Promise<unknown>): Promise<void> {
+  try {
+    await promise;
+    expect.fail('expected UNAUTHENTICATED');
+  } catch (err) {
+    const e = err as grpc.ServiceError;
+    expect(e.code).toBe(grpc.status.UNAUTHENTICATED);
+  }
+}
+
 describe('Consumer auth RPC gating', () => {
   let server: grpc.Server;
   // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -223,12 +233,12 @@ describe('Consumer auth RPC gating', () => {
     mockSecretStoreData.clear();
   });
 
-  it('denies wrong token on SetSecret', async () => {
-    await expectPermissionDenied(callUnary(client, 'SetSecret', { key: 'K', value: 'v' }, 'wrong'));
+  it('denies wrong token on SetSecret as UNAUTHENTICATED', async () => {
+    await expectUnauthenticated(callUnary(client, 'SetSecret', { key: 'K', value: 'v' }, 'wrong'));
   });
 
-  it('denies missing token on GetSecret', async () => {
-    await expectPermissionDenied(callUnary(client, 'GetSecret', { key: 'K' }));
+  it('denies missing token on GetSecret as UNAUTHENTICATED', async () => {
+    await expectUnauthenticated(callUnary(client, 'GetSecret', { key: 'K' }));
   });
 
   it('denies SetSecret without secrets capability', async () => {
@@ -384,6 +394,51 @@ describe('Consumer auth RPC gating', () => {
     expect(res.healthy).toBe(true);
   });
 
+  it('rejects CreateSession with unrecognized token as UNAUTHENTICATED', async () => {
+    await expectUnauthenticated(callUnary(client, 'CreateSession', {
+      model: 'mock/echo',
+    }, 'wrong'));
+  });
+
+  it('keeps CreateSession without token in the local namespace', async () => {
+    const created = await callUnary(client, 'CreateSession', { model: 'mock/echo', topic: 'local-dx' });
+    expect(created.id).toBeTruthy();
+
+    const fetchedLocal = await callUnary(client, 'GetSession', { session_id: created.id });
+    expect(fetchedLocal.id).toBe(created.id);
+
+    try {
+      await callUnary(client, 'GetSession', { session_id: created.id }, GOOD_TOKEN);
+      expect.fail('consumer should not see local session');
+    } catch (err) {
+      expect((err as grpc.ServiceError).code).toBe(grpc.status.NOT_FOUND);
+    }
+  });
+
+  it('stamps matching consumer token as session owner and isolates from local', async () => {
+    const created = await callUnary(client, 'CreateSession', {
+      model: 'mock/echo',
+      topic: 'apme-owned',
+    }, GOOD_TOKEN);
+    expect(created.id).toBeTruthy();
+
+    const fetchedConsumer = await callUnary(client, 'GetSession', {
+      session_id: created.id,
+    }, GOOD_TOKEN);
+    expect(fetchedConsumer.id).toBe(created.id);
+
+    try {
+      await callUnary(client, 'GetSession', { session_id: created.id });
+      expect.fail('local caller should not see consumer session');
+    } catch (err) {
+      expect((err as grpc.ServiceError).code).toBe(grpc.status.NOT_FOUND);
+    }
+
+    await expectUnauthenticated(callUnary(client, 'GetSession', {
+      session_id: created.id,
+    }, 'wrong'));
+  });
+
   it('full token can call secrets and config', async () => {
     await callUnary(client, 'SetSecret', { key: 'FULL', value: 'ok' }, GOOD_TOKEN);
     await callUnary(client, 'GetConfig', {}, GOOD_TOKEN);
diff --git a/packages/daemon/tests/integration/mcp-stdio-spawn-policy.test.ts b/packages/daemon/tests/integration/mcp-stdio-spawn-policy.test.ts
index f4b8f3e..7f70e54 100644
--- a/packages/daemon/tests/integration/mcp-stdio-spawn-policy.test.ts
+++ b/packages/daemon/tests/integration/mcp-stdio-spawn-policy.test.ts
@@ -332,11 +332,11 @@ describe('MCP stdio spawn policy E2E (H6)', () => {
         server_id: 'no-auth',
         transport: { type: 'stdio', command: 'npx', args: [] },
       });
-      expect.fail('expected PERMISSION_DENIED');
+      expect.fail('expected UNAUTHENTICATED');
     } catch (err) {
       const e = err as grpc.ServiceError;
-      expect(e.code).toBe(grpc.status.PERMISSION_DENIED);
-      expect(e.message).toMatch(/consumer authentication|Permission denied|mcp_register/i);
+      expect(e.code).toBe(grpc.status.UNAUTHENTICATED);
+      expect(e.message).toMatch(/consumer authentication|mcp_register/i);
     }
     expect(stdioTransportConstructs).toHaveLength(0);
   });
diff --git a/packages/python/src/abbenay_grpc/__init__.py b/packages/python/src/abbenay_grpc/__init__.py
index 7d29e4e..fa26e3f 100644
--- a/packages/python/src/abbenay_grpc/__init__.py
+++ b/packages/python/src/abbenay_grpc/__init__.py
@@ -33,6 +33,7 @@
     AbbenayClient,
     AbbenayError,
     ConnectionError,
+    InsecureTokenError,
     NotFoundError,
     ChatChunk,
     Session,
@@ -43,6 +44,7 @@
     "AbbenayClient",
     "AbbenayError",
     "ConnectionError",
+    "InsecureTokenError",
     "NotFoundError",
     "ChatChunk",
     "Session",
diff --git a/packages/python/src/abbenay_grpc/client.py b/packages/python/src/abbenay_grpc/client.py
index ad65d1a..24226bd 100644
--- a/packages/python/src/abbenay_grpc/client.py
+++ b/packages/python/src/abbenay_grpc/client.py
@@ -51,6 +51,11 @@ class NotFoundError(AbbenayError):
     pass
 
 
+class InsecureTokenError(AbbenayError):
+    """Consumer token requested on an unprotected (plaintext TCP) channel."""
+    pass
+
+
 @dataclass
 class ChatChunk:
     """A chunk of chat response."""
@@ -129,6 +134,29 @@ def __init__(
         self._channel: Optional[grpc.aio.Channel] = None
         self._stub = None
         self._client_id: Optional[str] = None
+
+    def _is_unix_target(self) -> bool:
+        return self._target.startswith("unix:")
+
+    def _is_protected_channel(self) -> bool:
+        """True when the transport keeps consumer tokens off the open network.
+
+        Unix sockets are local IPC. TCP requires TLS so x-abbenay-token is not
+        sent in plaintext to a network observer.
+        """
+        return self._is_unix_target() or self._tls
+
+    def _token_metadata(self, token: Optional[str]) -> Optional[List[tuple]]:
+        """Return x-abbenay-token metadata, or reject plaintext TCP token use."""
+        if token is None:
+            return None
+        if not self._is_protected_channel():
+            raise InsecureTokenError(
+                "Consumer tokens require a protected channel. "
+                "Connect via Unix socket (default) or enable TLS "
+                "(tls=True and/or ca_cert=...) before passing token=."
+            )
+        return [("x-abbenay-token", token)]
     
     @staticmethod
     def _get_abbenay_dir() -> Path:
@@ -233,7 +261,7 @@ async def connect(self) -> None:
             self._client_id = None
 
         try:
-            is_unix = self._target.startswith("unix:")
+            is_unix = self._is_unix_target()
             if self._tls and not is_unix:
                 root_certs = None
                 if self._ca_cert:
@@ -329,17 +357,17 @@ async def chat(
                 When set, fully replaces any named policy on the model.
             token: Optional consumer auth token for inline policy
                 authorization (sent as x-abbenay-token gRPC metadata).
-                Required when the server has a ``consumers`` section in
-                config and the consumer needs the ``inline_policy``
-                capability.
+                Requires a Unix socket or TLS channel; plaintext TCP
+                raises InsecureTokenError.
             
         Yields:
             ChatChunk objects containing response data
             
         Raises:
             AbbenayError: If the server streams an error chunk (e.g.,
-                INVALID_ARGUMENT for a malformed inline policy, or
-                PERMISSION_DENIED when consumer auth fails).
+                INVALID_ARGUMENT for a malformed inline policy,
+                UNAUTHENTICATED for a missing/unrecognized consumer token,
+                or PERMISSION_DENIED when the consumer lacks a capability).
         """
         self._ensure_connected()
         
@@ -371,11 +399,7 @@ async def chat(
         if policy is not None:
             request.policy.CopyFrom(_to_policy_proto(policy))
         
-        metadata = []
-        if token is not None:
-            metadata.append(("x-abbenay-token", token))
-        
-        async for chunk in self._stub.Chat(request, metadata=metadata or None):
+        async for chunk in self._stub.Chat(request, metadata=self._token_metadata(token)):
             yield self._parse_chunk(chunk)
     
     async def create_session(
@@ -383,6 +407,8 @@ async def create_session(
         model: str,
         topic: Optional[str] = None,
         metadata: Optional[Dict[str, str]] = None,
+        *,
+        token: Optional[str] = None,
     ) -> Session:
         """Create a new chat session.
         
@@ -390,6 +416,10 @@ async def create_session(
             model: Model ID
             topic: Optional topic/title
             metadata: Optional metadata
+            token: Optional consumer auth token (x-abbenay-token). When
+                consumers are configured, pass the same token used for
+                session_chat so the session is owned by consumer:<name>.
+                Requires a Unix socket or TLS channel.
             
         Returns:
             The created Session
@@ -402,14 +432,24 @@ async def create_session(
             metadata=metadata or {},
         )
         
-        response = await self._stub.CreateSession(request)
+        response = await self._stub.CreateSession(
+            request,
+            metadata=self._token_metadata(token),
+        )
         return self._parse_session(response)
     
-    async def get_session(self, session_id: str) -> Session:
+    async def get_session(
+        self,
+        session_id: str,
+        *,
+        token: Optional[str] = None,
+    ) -> Session:
         """Get a session by ID.
         
         Args:
             session_id: Session ID
+            token: Optional consumer auth token (x-abbenay-token).
+                Requires a Unix socket or TLS channel.
             
         Returns:
             The Session
@@ -424,7 +464,8 @@ async def get_session(self, session_id: str) -> Session:
                 proto.GetSessionRequest(
                     session_id=session_id,
                     include_messages=True,
-                )
+                ),
+                metadata=self._token_metadata(token),
             )
             return self._parse_session(response)
         except grpc.aio.AioRpcError as e:
@@ -436,12 +477,16 @@ async def list_sessions(
         self,
         limit: int = 10,
         offset: int = 0,
+        *,
+        token: Optional[str] = None,
     ) -> List[Session]:
         """List all sessions.
         
         Args:
             limit: Max sessions to return
             offset: Pagination offset
+            token: Optional consumer auth token (x-abbenay-token).
+                Requires a Unix socket or TLS channel.
             
         Returns:
             List of Sessions
@@ -452,7 +497,8 @@ async def list_sessions(
             proto.ListSessionsRequest(
                 limit=limit,
                 offset=offset,
-            )
+            ),
+            metadata=self._token_metadata(token),
         )
         
         return [
@@ -468,17 +514,25 @@ async def list_sessions(
             for s in response.sessions
         ]
     
-    async def delete_session(self, session_id: str) -> None:
+    async def delete_session(
+        self,
+        session_id: str,
+        *,
+        token: Optional[str] = None,
+    ) -> None:
         """Delete a session.
         
         Args:
             session_id: Session ID
+            token: Optional consumer auth token (x-abbenay-token).
+                Requires a Unix socket or TLS channel.
         """
         self._ensure_connected()
         
         try:
             await self._stub.DeleteSession(
-                proto.DeleteSessionRequest(session_id=session_id)
+                proto.DeleteSessionRequest(session_id=session_id),
+                metadata=self._token_metadata(token),
             )
         except grpc.aio.AioRpcError as e:
             if e.code() == grpc.StatusCode.NOT_FOUND:
@@ -551,13 +605,15 @@ async def session_chat(
             enable_tools: Enable tool calling
             tool_filter: Only expose these tools to the LLM (empty = all)
             policy: Optional inline policy override
-            token: Optional consumer auth token
+            token: Optional consumer auth token. Requires a Unix socket
+                or TLS channel.
 
         Yields:
             ChatChunk objects containing response data
 
         Raises:
             AbbenayError: On server error chunks
+            InsecureTokenError: If token is set on plaintext TCP
         """
         self._ensure_connected()
 
@@ -584,11 +640,7 @@ async def session_chat(
         if policy is not None:
             request.policy.CopyFrom(_to_policy_proto(policy))
 
-        metadata = []
-        if token is not None:
-            metadata.append(("x-abbenay-token", token))
-
-        async for chunk in self._stub.SessionChat(request, metadata=metadata or None):
+        async for chunk in self._stub.SessionChat(request, metadata=self._token_metadata(token)):
             yield self._parse_chunk(chunk)
 
     async def register_mcp_server(
@@ -615,7 +667,8 @@ async def register_mcp_server(
                 args: command arguments (stdio)
             session_id: Scope to a session (auto-cleanup on delete)
             tool_filter: Only register these tools from the server
-            token: Consumer auth token for mcp_register capability
+            token: Consumer auth token for mcp_register capability.
+                Requires a Unix socket or TLS channel.
 
         Returns:
             List of discovered tool names (namespaced)
@@ -647,9 +700,10 @@ async def register_mcp_server(
         if tool_filter:
             request.tool_filter.extend(tool_filter)
 
-        metadata = []
-        if token is not None:
-            metadata.append(("x-abbenay-token", token))
+        metadata: List[tuple] = []
+        token_md = self._token_metadata(token)
+        if token_md:
+            metadata.extend(token_md)
         if self._client_id is not None:
             metadata.append(("x-abbenay-client-id", self._client_id))
 
@@ -673,21 +727,17 @@ async def unregister_mcp_server(
 
         Args:
             server_id: Server ID to unregister
-            token: Consumer auth token
+            token: Consumer auth token. Requires a Unix socket or TLS channel.
 
         Returns:
             True if successfully unregistered
         """
         self._ensure_connected()
 
-        metadata = []
-        if token is not None:
-            metadata.append(("x-abbenay-token", token))
-
         try:
             response = await self._stub.UnregisterMcpServer(
                 proto.UnregisterMcpServerRequest(server_id=server_id),
-                metadata=metadata or None,
+                metadata=self._token_metadata(token),
             )
             return response.success
         except grpc.aio.AioRpcError as e:
diff --git a/packages/python/tests/test_session_consumer_token.py b/packages/python/tests/test_session_consumer_token.py
new file mode 100644
index 0000000..4be4e2b
--- /dev/null
+++ b/packages/python/tests/test_session_consumer_token.py
@@ -0,0 +1,138 @@
+"""Unit tests for AbbenayClient consumer-token forwarding on session RPCs."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from abbenay_grpc.client import AbbenayClient, InsecureTokenError
+
+
+def _session_proto(session_id: str = "sess-1", model: str = "mock/echo"):
+    return SimpleNamespace(
+        id=session_id,
+        model=model,
+        topic="t",
+        messages=[],
+        created_by=None,
+        created_at="2024-01-01T00:00:00Z",
+        updated_at="2024-01-01T00:00:00Z",
+    )
+
+
+async def _empty_stream(*_args, **_kwargs):
+    if False:  # pragma: no cover
+        yield None
+
+
+def _connected_client(**kwargs) -> AbbenayClient:
+    c = AbbenayClient(**kwargs)
+    stub = MagicMock()
+    stub.CreateSession = AsyncMock(return_value=_session_proto())
+    stub.GetSession = AsyncMock(return_value=_session_proto())
+    stub.ListSessions = AsyncMock(return_value=SimpleNamespace(sessions=[]))
+    stub.DeleteSession = AsyncMock(return_value=SimpleNamespace())
+    stub.SessionChat = MagicMock(side_effect=_empty_stream)
+    stub.RegisterMcpServer = AsyncMock(
+        return_value=SimpleNamespace(discovered_tools=["tool.a"]),
+    )
+    stub.UnregisterMcpServer = AsyncMock(
+        return_value=SimpleNamespace(success=True),
+    )
+    c._stub = stub
+    c._client_id = "test-client"
+    return c
+
+
+@pytest.fixture
+def unix_client() -> AbbenayClient:
+    return _connected_client(socket_path="/tmp/abbenay-test.sock")
+
+
+@pytest.fixture
+def tls_client() -> AbbenayClient:
+    return _connected_client(host="127.0.0.1", port=50051, tls=True)
+
+
+@pytest.fixture
+def insecure_tcp_client() -> AbbenayClient:
+    return _connected_client(host="127.0.0.1", port=50051, tls=False)
+
+
+@pytest.mark.asyncio
+async def test_session_crud_forwards_consumer_token(unix_client: AbbenayClient):
+    token = "consumer-tok"
+    expected = [("x-abbenay-token", token)]
+
+    await unix_client.create_session("mock/echo", topic="owned", token=token)
+    unix_client._stub.CreateSession.assert_awaited()
+    assert unix_client._stub.CreateSession.await_args.kwargs["metadata"] == expected
+
+    await unix_client.get_session("sess-1", token=token)
+    assert unix_client._stub.GetSession.await_args.kwargs["metadata"] == expected
+
+    await unix_client.list_sessions(token=token)
+    assert unix_client._stub.ListSessions.await_args.kwargs["metadata"] == expected
+
+    await unix_client.delete_session("sess-1", token=token)
+    assert unix_client._stub.DeleteSession.await_args.kwargs["metadata"] == expected
+
+
+@pytest.mark.asyncio
+async def test_create_session_then_session_chat_share_token(unix_client: AbbenayClient):
+    token = "shared-consumer-tok"
+    expected = [("x-abbenay-token", token)]
+
+    session = await unix_client.create_session("mock/echo", token=token)
+    assert session.id == "sess-1"
+    assert unix_client._stub.CreateSession.await_args.kwargs["metadata"] == expected
+
+    chunks = [
+        chunk
+        async for chunk in unix_client.session_chat(session.id, "hi", token=token)
+    ]
+    assert chunks == []
+    assert unix_client._stub.SessionChat.call_args.kwargs["metadata"] == expected
+
+
+@pytest.mark.asyncio
+async def test_tls_tcp_allows_consumer_token(tls_client: AbbenayClient):
+    token = "tls-tok"
+    await tls_client.create_session("mock/echo", token=token)
+    assert tls_client._stub.CreateSession.await_args.kwargs["metadata"] == [
+        ("x-abbenay-token", token),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_plaintext_tcp_rejects_consumer_token(insecure_tcp_client: AbbenayClient):
+    with pytest.raises(InsecureTokenError, match="protected channel"):
+        await insecure_tcp_client.create_session("mock/echo", token="leak-me")
+    insecure_tcp_client._stub.CreateSession.assert_not_awaited()
+
+    with pytest.raises(InsecureTokenError):
+        await insecure_tcp_client.get_session("sess-1", token="leak-me")
+    with pytest.raises(InsecureTokenError):
+        await insecure_tcp_client.list_sessions(token="leak-me")
+    with pytest.raises(InsecureTokenError):
+        await insecure_tcp_client.delete_session("sess-1", token="leak-me")
+    with pytest.raises(InsecureTokenError):
+        async for _ in insecure_tcp_client.session_chat("sess-1", "hi", token="leak-me"):
+            pass
+    with pytest.raises(InsecureTokenError):
+        await insecure_tcp_client.register_mcp_server(
+            "s",
+            {"type": "http", "url": "http://127.0.0.1:9"},
+            token="leak-me",
+        )
+    with pytest.raises(InsecureTokenError):
+        await insecure_tcp_client.unregister_mcp_server("s", token="leak-me")
+
+
+@pytest.mark.asyncio
+async def test_session_crud_omits_metadata_without_token(insecure_tcp_client: AbbenayClient):
+    # No token on plaintext TCP remains allowed (local DX / probes).
+    await insecure_tcp_client.create_session("mock/echo")
+    assert insecure_tcp_client._stub.CreateSession.await_args.kwargs["metadata"] is None