fix(daemon): preserve tenant scope for queued drains

apps/agor-daemon/src/register-hooks.test.ts

@@ -21,6 +21,7 @@
 import { describe, expect, it } from 'vitest';
 import {
   enrichSessionFindResultWithRemoteRelationships,
+  getTrustedSessionTenantId,
   isPromptFlowPatchOnly,
   PROMPT_FLOW_PATCH_FIELDS,
   shouldDrainQueueAfterSessionPostTurnPatch,
@@ -100,6 +101,25 @@ describe('shouldRunSessionPostTurnHooks', () => {
   });
 });
 
+describe('getTrustedSessionTenantId', () => {
+  it('reads non-enumerable tenant metadata from session DTOs without requiring JSON exposure', () => {
+    const session = makeSession('session-1');
+    Object.defineProperty(session, 'tenant_id', {
+      value: 'tenant-from-row',
+      enumerable: false,
+    });
+
+    expect(getTrustedSessionTenantId(session)).toBe('tenant-from-row');
+    expect(Object.keys(session)).not.toContain('tenant_id');
+    expect(JSON.stringify(session)).not.toContain('tenant_id');
+  });
+
+  it('ignores absent or empty tenant metadata', () => {
+    expect(getTrustedSessionTenantId(makeSession('session-1'))).toBeUndefined();
+    expect(getTrustedSessionTenantId({ tenant_id: '' })).toBeUndefined();
+  });
+});
+
 describe('shouldDrainQueueAfterSessionPostTurnPatch', () => {
   it('drains for promptable ready sessions by default', () => {
     expect(

apps/agor-daemon/src/register-hooks.ts

@@ -139,6 +139,7 @@ import {
   recomputeNextRunAt,
   validateScheduleConfig,
 } from './utils/schedule-hooks.js';
+import { deferWithSessionQueueTenantScope } from './utils/session-queue-tenant-scope.js';
 import {
   isTerminalQueueProcessingSuppressed,
   sessionCanStartTask,
@@ -329,6 +330,11 @@ export function shouldDrainQueueAfterSessionPostTurnPatch(
   );
 }
 
+export function getTrustedSessionTenantId(session: unknown): string | undefined {
+  const tenantId = (session as { tenant_id?: unknown } | undefined)?.tenant_id;
+  return typeof tenantId === 'string' && tenantId.length > 0 ? tenantId : undefined;
+}
+
 export async function enrichSessionFindResultWithRemoteRelationships(
   result: Paginated<Session> | Session[],
   sessionsService: Pick<SessionsServiceImpl, 'enrichRemoteRelationships'>
@@ -2726,23 +2732,37 @@ export function registerHooks(ctx: RegisterHooksContext): void {
             });
 
             if (shouldDrainQueueAfterSessionPostTurnPatch(session, context.params)) {
+              const sessionTenantId = getTrustedSessionTenantId(session);
               // Same fresh-scope pattern: queue processing must run outside the
               // outer transaction but still inside the session tenant for RLS.
-              deferWithTenantDatabaseScope(db, context.params, async () => {
-                try {
+              // Some completion/background paths have minimal params, so this
+              // relies on params.tenant, current tenant ALS, the already-returned
+              // session row tenant_id, or static tenant config and otherwise
+              // fails closed.
+              deferWithSessionQueueTenantScope(
+                {
+                  db,
+                  config,
+                  sessionId: session.session_id,
+                  params: context.params,
+                  tenantIdHint: sessionTenantId,
+                  label: 'SessionsService.after.patch queue drain',
+                },
+                async (queueParams) => {
                   console.log(
                     `🔄 [SessionsService.after.patch] Session ${shortId(session.session_id)} became promptable (${session.status}), checking for queued tasks...`
                   );
 
-                  await sessionsService.triggerQueueProcessing(session.session_id, context.params);
-                } catch (error) {
+                  await sessionsService.triggerQueueProcessing(session.session_id, queueParams);
+                },
+                (error) => {
                   console.error(
                     `❌ [SessionsService.after.patch] Failed to process queue for session ${shortId(session.session_id)}:`,
                     error
                   );
                   // Don't throw - queue processing failure shouldn't break session patches
                 }
-              });
+              );
             } else {
               console.log(
                 `⏭️  [SessionsService.after.patch] Queue drain suppressed for session ${shortId(session.session_id)} (suppressTerminalQueueProcessing or not ready)`

apps/agor-daemon/src/register-routes.ts

@@ -125,6 +125,10 @@ import {
 } from './utils/mcp-header-secrets.js';
 import { canControlCliSession } from './utils/mcp-token-authorization.js';
 import { ensureScheduleRunsAsCaller } from './utils/schedule-hooks.js';
+import {
+  deferWithSessionQueueTenantScope,
+  runWithSessionQueueTenantScope,
+} from './utils/session-queue-tenant-scope.js';
 import { stopSessionPreserveQueue } from './utils/session-stop.js';
 import {
   sessionCanStartTask,
@@ -2181,6 +2185,22 @@ export async function registerRoutes(ctx: RegisterRoutesContext): Promise<void>
   const queueRetryScheduled = new Set<SessionID>();
 
   async function processNextQueuedTask(sessionId: SessionID, params: RouteParams): Promise<void> {
+    await runWithSessionQueueTenantScope(
+      {
+        db,
+        config,
+        sessionId,
+        params,
+        label: 'processNextQueuedTask',
+      },
+      async (scopedParams) => processNextQueuedTaskInTenantScope(sessionId, scopedParams)
+    );
+  }
+
+  async function processNextQueuedTaskInTenantScope(
+    sessionId: SessionID,
+    params: RouteParams
+  ): Promise<void> {
     const existingLock = sessionTurnLocks.get(sessionId);
     if (existingLock) {
       console.log(`⏳ [Queue] Session turn in progress for ${shortId(sessionId)}, waiting...`);
@@ -2208,14 +2228,27 @@ export async function registerRoutes(ctx: RegisterRoutesContext): Promise<void>
 
       if (!queueRetryScheduled.has(sessionId)) {
         queueRetryScheduled.add(sessionId);
-        deferInFreshTenantScope(params, async () => {
-          queueRetryScheduled.delete(sessionId);
-          try {
-            await processNextQueuedTask(sessionId, params);
-          } catch (error) {
+        deferWithSessionQueueTenantScope(
+          {
+            db,
+            config,
+            sessionId,
+            params,
+            label: 'processNextQueuedTask retry',
+          },
+          async (retryParams) => {
+            queueRetryScheduled.delete(sessionId);
+            try {
+              await processNextQueuedTask(sessionId, retryParams);
+            } catch (error) {
+              console.error(`❌ [Queue] Retry failed for session ${shortId(sessionId)}:`, error);
+            }
+          },
+          (error) => {
+            queueRetryScheduled.delete(sessionId);
             console.error(`❌ [Queue] Retry failed for session ${shortId(sessionId)}:`, error);
           }
-        });
+        );
       } else {
         console.log(
           `⏭️  [Queue] Retry already scheduled for session ${shortId(sessionId)}, not queueing another`