fix: harden API key handling and bound agent task claims

apps/agent/agent/lib/tasks.ts

@@ -42,11 +42,7 @@ export async function claimDue(
 	const onlyMode = "only" in kinds;
 
 	const claimed = await db.$queryRaw<LeasedTask[]>`
-		UPDATE "agentTask" AS t
-		SET "leasedUntil" = ${until},
-			"startedAt" = COALESCE(t."startedAt", ${now}),
-			"attempts" = t."attempts" + 1
-		FROM (
+		WITH due AS MATERIALIZED (
 			SELECT t2.id FROM "agentTask" AS t2
 			WHERE t2."finishedAt" IS NULL
 				AND t2."dueAt" <= ${now}
@@ -59,7 +55,12 @@ export async function claimDue(
 			ORDER BY t2."priority" DESC, t2."dueAt" ASC
 			LIMIT ${limit}
 			FOR UPDATE SKIP LOCKED
-		) AS due
+		)
+		UPDATE "agentTask" AS t
+		SET "leasedUntil" = ${until},
+			"startedAt" = COALESCE(t."startedAt", ${now}),
+			"attempts" = t."attempts" + 1
+		FROM due
 		WHERE t.id = due.id
 		RETURNING t.id, t."contactId", t."companyId", t."dealId", t.kind, t.reason, t.payload,
 			t.budget, t.attempts, t.priority, t."dueAt";
@@ -72,23 +73,30 @@ export async function claimDue(
 
 export async function retireExhausted(
 	limit: number = DISPATCH.reconcile.retire,
+	kinds?: { only: readonly string[] },
 ): Promise<TaskSubject[]> {
+	const only = kinds?.only ?? [];
+	if (kinds && only.length === 0) return [];
+
 	const now = new Date();
 
 	return db.$queryRaw<TaskSubject[]>`
-		UPDATE "agentTask" AS t
-		SET "finishedAt" = ${now},
-			"outcome" = ${RETIRED_OUTCOME}
-		WHERE t.id IN (
+		WITH exhausted AS MATERIALIZED (
 			SELECT c.id
 			FROM "agentTask" AS c
 			WHERE c."finishedAt" IS NULL
 				AND c."attempts" >= ${MAX_ATTEMPTS}
 				AND (c."leasedUntil" IS NULL OR c."leasedUntil" < ${now})
+				AND (${only}::text[] = '{}' OR c.kind = ANY(${only}::text[]))
 			ORDER BY c."dueAt" ASC
 			LIMIT ${limit}
 			FOR UPDATE SKIP LOCKED
 		)
+		UPDATE "agentTask" AS t
+		SET "finishedAt" = ${now},
+			"outcome" = ${RETIRED_OUTCOME}
+		FROM exhausted
+		WHERE t.id = exhausted.id
 		RETURNING t.id, t."contactId", t."companyId", t."dealId", t.kind;
 	`;
 }

apps/agent/test/tasks.integration.spec.ts

@@ -1,6 +1,5 @@
 import { afterEach, beforeEach, describe, expect, it } from "bun:test";
 import { db } from "@crm/db";
-import { DIRECT_KINDS } from "@crm/db/agent-tasks";
 import {
 	claimDue,
 	completeTask,
@@ -9,9 +8,14 @@ import {
 	scheduleTask,
 } from "../agent/lib/tasks";
 
-const kind = "test-lease";
+const suffix = process.env.TEST_RUN_ID ?? "tasks-integration-spec";
+const kind = `test-lease-${suffix}`;
 
-const RESEARCH = { except: DIRECT_KINDS } as const;
+const RESEARCH = { only: [kind] } as const;
+
+function retire(limit?: number) {
+	return retireExhausted(limit, RESEARCH);
+}
 
 async function clear() {
 	await db.agentTask.deleteMany({ where: { kind } });
@@ -44,6 +48,13 @@ async function expire(taskId: string) {
 	});
 }
 
+async function exhaust() {
+	for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
+		const claimed = await claimDue(10, RESEARCH);
+		for (const task of claimed) await expire(task.id);
+	}
+}
+
 async function someone() {
 	return db.contact.create({
 		data: {
@@ -141,6 +152,27 @@ describe("claimDue", () => {
 
 		expect(await claimDue(10, RESEARCH)).toHaveLength(0);
 	});
+
+	it("gives each due row to one caller and no caller more than its limit", async () => {
+		const mine: string[] = [];
+		for (let row = 0; row < 8; row++) mine.push((await queue()).id);
+
+		const results = await Promise.all(
+			Array.from({ length: 4 }, () => claimDue(2, RESEARCH)),
+		);
+		for (const claimed of results) {
+			expect(claimed.length).toBeLessThanOrEqual(2);
+		}
+
+		const ids = results.flatMap((claimed) => claimed.map((task) => task.id));
+		expect(new Set(ids).size).toBe(ids.length);
+		for (const id of ids) expect(mine).toContain(id);
+
+		const leased = await db.agentTask.count({
+			where: { id: { in: mine }, attempts: 1, leasedUntil: { not: null } },
+		});
+		expect(leased).toBe(ids.length);
+	});
 });
 
 describe("retireExhausted", () => {
@@ -153,7 +185,7 @@ describe("retireExhausted", () => {
 			await expire(task.id);
 		}
 
-		const retired = await retireExhausted();
+		const retired = await retire();
 		expect(retired.map((t) => t.id)).toContain(task.id);
 		expect(retired.find((t) => t.id === task.id)?.contactId).toBe(contact.id);
 
@@ -170,14 +202,14 @@ describe("retireExhausted", () => {
 			if (attempt < MAX_ATTEMPTS - 1) await expire(task.id);
 		}
 
-		expect(await retireExhausted()).toHaveLength(0);
+		expect(await retire()).toHaveLength(0);
 	});
 
 	it("leaves work that still has attempts left", async () => {
 		await queue();
 		await claimDue(10, RESEARCH);
 
-		expect(await retireExhausted()).toHaveLength(0);
+		expect(await retire()).toHaveLength(0);
 	});
 
 	it("retires no more rows than the limit allows", async () => {
@@ -190,14 +222,37 @@ describe("retireExhausted", () => {
 		}
 
 		for (let pass = 0; pass < 3; pass++) {
-			expect((await retireExhausted(2)).length).toBeLessThanOrEqual(2);
+			expect((await retire(2)).length).toBeLessThanOrEqual(2);
 		}
 
 		const open = await db.agentTask.count({
 			where: { id: { in: mine }, finishedAt: null },
 		});
 		expect(open).toBe(0);
 	});
+
+	it("gives each exhausted row to one caller and no caller more than its limit", async () => {
+		const mine: string[] = [];
+		for (let row = 0; row < 8; row++) mine.push((await queue()).id);
+
+		await exhaust();
+
+		const results = await Promise.all(
+			Array.from({ length: 4 }, () => retire(2)),
+		);
+		for (const retired of results) {
+			expect(retired.length).toBeLessThanOrEqual(2);
+		}
+
+		const ids = results.flatMap((retired) => retired.map((task) => task.id));
+		expect(new Set(ids).size).toBe(ids.length);
+		for (const id of ids) expect(mine).toContain(id);
+
+		const open = await db.agentTask.count({
+			where: { id: { in: mine }, finishedAt: null },
+		});
+		expect(open).toBe(mine.length - ids.length);
+	});
 });
 
 describe("completeTask", () => {

apps/api/package.json

@@ -51,6 +51,7 @@
 		"zod": "^4.4.3"
 	},
 	"devDependencies": {
+		"@better-auth/api-key": "1.6.25",
 		"@crm/typescript-config": "workspace:*",
 		"@nestjs/testing": "^11.0.1",
 		"@types/bun": "^1.3.14",

apps/api/src/archive/archive-retention.controller.ts

@@ -20,6 +20,7 @@ import {
 	ApiTags,
 } from "@nestjs/swagger";
 import { AllowAnonymous } from "@thallesp/nestjs-better-auth";
+import { timingSafeEquals } from "../common/timing-safe-equal";
 import { CompaniesService } from "../companies/companies.service";
 import type { EnvironmentVariables } from "../config/env.validation";
 import { ContactsService } from "../contacts/contacts.service";
@@ -103,14 +104,3 @@ export class ArchiveRetentionController {
 		return { retentionDays, companies, contacts, deals };
 	}
 }
-
-function timingSafeEquals(a: string, b: string): boolean {
-	if (a.length !== b.length) return false;
-
-	let mismatch = 0;
-	for (let index = 0; index < a.length; index += 1) {
-		mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index);
-	}
-
-	return mismatch === 0;
-}

apps/api/src/common/timing-safe-equal.ts

@@ -0,0 +1,10 @@
+import { timingSafeEqual } from "node:crypto";
+
+export function timingSafeEquals(a: string, b: string): boolean {
+	const bufferA = Buffer.from(a);
+	const bufferB = Buffer.from(b);
+
+	if (bufferA.length !== bufferB.length) return false;
+
+	return timingSafeEqual(bufferA, bufferB);
+}

apps/api/src/config/env.validation.ts

@@ -8,6 +8,7 @@ import {
 	Max,
 	Min,
 	MinLength,
+	ValidateIf,
 	validateSync,
 } from "class-validator";
 
@@ -120,7 +121,12 @@ export class EnvironmentVariables {
 	AGENT_URL?: string;
 
 	@IsOptional()
+	@ValidateIf((_environment, value) => value !== "")
 	@IsString()
+	@MinLength(32, {
+		message:
+			"AGENT_BRIDGE_SECRET must be at least 32 characters. Generate one with: openssl rand -base64 32",
+	})
 	AGENT_BRIDGE_SECRET?: string;
 
 	@IsOptional()

apps/api/src/create-app.ts

@@ -109,7 +109,7 @@ export async function createApp(): Promise<NestExpressApplication> {
 
 	restBridge = createOpenApiExpressMiddleware({
 		router: appRouter,
-		createContext: ({ req }) => createBaseTrpcContext(req),
+		createContext: ({ req, res }) => createBaseTrpcContext(req, res),
 	});
 
 	return app;

apps/api/src/currency/rates.controller.ts

@@ -18,6 +18,7 @@ import {
 	ApiTags,
 } from "@nestjs/swagger";
 import { AllowAnonymous } from "@thallesp/nestjs-better-auth";
+import { timingSafeEquals } from "../common/timing-safe-equal";
 import type { EnvironmentVariables } from "../config/env.validation";
 import { ConversionService } from "./conversion.service";
 import { RatesService } from "./rates.service";
@@ -81,14 +82,3 @@ export class RatesController {
 		return { ...refresh, converted: filled.converted, missing: filled.missing };
 	}
 }
-
-function timingSafeEquals(a: string, b: string): boolean {
-	if (a.length !== b.length) return false;
-
-	let mismatch = 0;
-	for (let index = 0; index < a.length; index += 1) {
-		mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index);
-	}
-
-	return mismatch === 0;
-}

apps/api/src/logging/all-exceptions.filter.ts

@@ -42,7 +42,7 @@ export class AllExceptionsFilter implements ExceptionFilter {
 		const payload = {
 			message: describe(exception),
 			method: request.method,
-			path: request.originalUrl,
+			path: request.path,
 			statusCode: status,
 			exception: exception?.constructor?.name,
 		};

apps/api/src/logging/request-logger.middleware.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import { API_KEY_HEADER } from "@crm/auth";
 import { Injectable, Logger, type NestMiddleware } from "@nestjs/common";
 import type { NextFunction, Request, Response } from "express";
 import { type RequestContext, runInRequestContext } from "./request-context";
@@ -18,7 +19,7 @@ export class RequestLoggerMiddleware implements NestMiddleware {
 		const context: RequestContext = {
 			requestId,
 			method: request.method,
-			path: request.originalUrl,
+			path: request.path,
 		};
 		const startedAt = process.hrtime.bigint();
 
@@ -56,6 +57,7 @@ export class RequestLoggerMiddleware implements NestMiddleware {
 			durationMs: Number(durationMs.toFixed(1)),
 			ip: request.ip,
 			userAgent: request.get("user-agent"),
+			apiKeyPresented: Boolean(request.get(API_KEY_HEADER)),
 		};
 
 		if (statusCode >= 500) {

apps/api/src/sync/sync.controller.ts

@@ -18,6 +18,7 @@ import {
 	ApiTags,
 } from "@nestjs/swagger";
 import { AllowAnonymous } from "@thallesp/nestjs-better-auth";
+import { timingSafeEquals } from "../common/timing-safe-equal";
 import type { EnvironmentVariables } from "../config/env.validation";
 import { MailboxSyncService } from "./mailbox-sync.service";
 
@@ -88,14 +89,3 @@ export class SyncController {
 		return this.sync.runDue();
 	}
 }
-
-function timingSafeEquals(a: string, b: string): boolean {
-	if (a.length !== b.length) return false;
-
-	let mismatch = 0;
-	for (let index = 0; index < a.length; index += 1) {
-		mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index);
-	}
-
-	return mismatch === 0;
-}

apps/api/src/telemetry/rollup.service.ts

@@ -128,7 +128,13 @@ export class RollupService {
 		]);
 
 		return {
-			properties: { ...shape, ...agent, ...ledger, ...crm },
+			properties: {
+				...shape,
+				...agent,
+				...ledger,
+				...crm,
+				api_key_rate_limited: counters.api_key_rate_limited ?? 0,
+			},
 			counters,
 		};
 	}

apps/api/src/telemetry/telemetry.controller.ts

@@ -18,6 +18,7 @@ import {
 	ApiTags,
 } from "@nestjs/swagger";
 import { AllowAnonymous } from "@thallesp/nestjs-better-auth";
+import { timingSafeEquals } from "../common/timing-safe-equal";
 import type { EnvironmentVariables } from "../config/env.validation";
 import { RollupService } from "./rollup.service";
 
@@ -71,14 +72,3 @@ export class TelemetryController {
 		return this.rollup.run();
 	}
 }
-
-function timingSafeEquals(a: string, b: string): boolean {
-	if (a.length !== b.length) return false;
-
-	let mismatch = 0;
-	for (let index = 0; index < a.length; index += 1) {
-		mismatch |= a.charCodeAt(index) ^ b.charCodeAt(index);
-	}
-
-	return mismatch === 0;
-}