🐛 keep the admission advisory-lock key valid PostgreSQL text

libs/backend/agents/execution/runs/main/src/__tests__/admission-lock-key.test.ts

@@ -0,0 +1,30 @@
+import { describe, expect, it } from "vitest";
+
+import { __AdmissionLockKey } from "../admission-lock-key";
+
+/** The byte PostgreSQL text cannot hold, written this way so no source file carries a literal NUL. */
+const _NUL = String.fromCharCode(0);
+
+describe("__AdmissionLockKey", function _AdmissionLockKeySuite()
+{
+	// PostgreSQL rejects a NUL byte in text with SQLSTATE 22021, which fails the whole statement.
+	// Separating the two parts with NUL is what made every message submit return 503.
+	it("introduces no byte PostgreSQL text cannot hold", function _AddsNoControlByte()
+	{
+		expect(__AdmissionLockKey("testv4", "idempotency-key")).not.toContain(_NUL);
+		expect(__AdmissionLockKey("silo-1", "conversation:7753e9bb/message")).not.toContain(_NUL);
+	});
+
+	it("keeps distinct pairs distinct where a plain join would collide", function _StaysInjective()
+	{
+		// A separator either part may itself contain cannot separate them: "a:b" + "c" and "a" + "b:c"
+		// both read as "a:b:c". The length prefix removes the ambiguity without reserving a character.
+		expect(__AdmissionLockKey("a:b", "c")).not.toEqual(__AdmissionLockKey("a", "b:c"));
+		expect(__AdmissionLockKey("ab", "c")).not.toEqual(__AdmissionLockKey("a", "bc"));
+	});
+
+	it("gives one silo and key the same lock on every attempt", function _StaysStable()
+	{
+		expect(__AdmissionLockKey("testv4", "key-1")).toEqual(__AdmissionLockKey("testv4", "key-1"));
+	});
+});

libs/backend/agents/execution/runs/main/src/__tests__/prisma-child-run-reservation-repository.test.ts

@@ -212,6 +212,12 @@ describe("PrismaChildRunReservationRepository", function _describeReservationRep
 
 		expect(result).toEqual({ outcome: "idempotent", snapshot: childSnapshot });
 		expect(transaction.$queryRaw).toHaveBeenCalledTimes(1);
+		// The advisory-lock key is a text parameter, and PostgreSQL rejects a NUL byte in text with
+		// SQLSTATE 22021, failing the whole reservation.
+		for (const [statement] of transaction.$queryRaw.mock.calls)
+		{
+			expect((statement as { values: unknown[] }).values.filter(function _IsText(value): value is string { return typeof value === "string"; }).join("")).not.toContain(String.fromCharCode(0));
+		}
 		expect(transaction.agentRun.create).not.toHaveBeenCalled();
 		expect(transaction.runInputSnapshot.create).not.toHaveBeenCalled();
 		expect(transaction.childRunReservation.create).not.toHaveBeenCalled();

libs/backend/agents/execution/runs/main/src/__tests__/prisma-run-admission-repository.test.ts

@@ -45,7 +45,7 @@
 		const prisma = { $transaction: vi.fn(async function _transaction(callback: (client: typeof transaction) => Promise<unknown>) { return callback(transaction); }) } as unknown as PrismaClient;
 		const repository = new PrismaRunAdmissionRepository(prisma, { now: function _now() { return new Date("2026-07-20T00:00:00.000Z"); } });
 
 		await expect(repository.admit(command, async function _build() { return { outcome: "ready", value: { authority, snapshot } } as const; })).resolves.toMatchObject({ outcome: "accepted" });
 		expect(transaction.agentRun.create).toHaveBeenCalledWith({ data: expect.objectContaining({ trigger: "Schedule", delegatedUserId: null }) });
 	});
 
@@ -55,8 +55,14 @@
 		const prisma = { $transaction: vi.fn(async function _transaction(callback: (client: typeof transaction) => Promise<unknown>) { return callback(transaction); }) } as unknown as PrismaClient;
 		const repository = new PrismaRunAdmissionRepository(prisma, { now: function _now() { return new Date("2026-07-20T00:00:00.000Z"); } });
 
 		await expect(repository.admit(_command(), async function _build() { return { outcome: "ready", value: { authority: _authority(), snapshot: _snapshot() } } as const; })).resolves.toEqual({ outcome: "accepted", snapshot: _snapshot() });
 		expect(transaction.$queryRaw).toHaveBeenCalledTimes(2);
+		// PostgreSQL rejects a NUL byte in text with SQLSTATE 22021 and fails the whole statement, so
+		// an advisory-lock key carrying one made every admission return 503 against a real database.
+		for (const [statement] of transaction.$queryRaw.mock.calls)
+		{
+			expect((statement as { values: unknown[] }).values.filter(function _IsText(value): value is string { return typeof value === "string"; }).join("")).not.toContain(String.fromCharCode(0));
+		}
 		expect(transaction.agentRun.create).toHaveBeenCalledWith({ data: expect.objectContaining({ inputSnapshotDigest: `sha256:${"c".repeat(64)}`, acceptedAt: new Date("2026-07-20T00:00:00.000Z") }) });
 		expect(transaction.runInputSnapshot.create).toHaveBeenCalledWith({ data: expect.objectContaining({ runId: "run-1", digest: `sha256:${"c".repeat(64)}`, messageIds: ["message-1"] }) });
 		expect(transaction.outboxEvent.createMany).toHaveBeenCalledWith({ data: [expect.objectContaining({ sequence: 1, kind: "RunAccepted", idempotencyKey: "run-1:accepted" }), expect.objectContaining({ sequence: 2, kind: "RunAttemptRequested", idempotencyKey: "run-1:attempt:1" })] });
@@ -79,7 +85,7 @@
 			expect(transactionContext.prisma).toBe(transaction);
 			expect(value.snapshot.runId).toBe("run-1");
 			order.push("message");
 		})).resolves.toMatchObject({ outcome: "accepted" });
 		expect(order).toEqual(["run", "snapshot", "outbox", "message"]);
 	});
 
@@ -94,7 +100,7 @@
 		{
 			order.push("build");
 			return { outcome: "ready", value: { authority: _authority(), snapshot: _snapshot() } } as const;
 		}, async function _Commit() { order.push("commit"); }, async function _Prepare() { order.push("prepare"); })).resolves.toMatchObject({ outcome: "accepted" });
 		expect(order).toEqual(["prepare", "build", "run", "snapshot", "outbox", "commit"]);
 	});
 
@@ -118,7 +124,7 @@
 		await expect(repository.admit(_command(), async function _Deny() { return { outcome: "denied", reason: "persona_unavailable" } as const; }, undefined, async function _Prepare(context)
 		{
 			(context.prisma as unknown as { pendingChildren: string[] }).pendingChildren.push("child-1");
 		})).resolves.toEqual({ outcome: "denied", reason: "persona_unavailable" });
 		expect(committedChildren).toEqual([]);
 	});
 
@@ -138,7 +144,7 @@
 		await expect(repository.admit(_command(), async function _BuildConflict() { return { outcome: "ready", value: { authority: { ..._authority(), delegatedUserId: "user-2" }, snapshot: _snapshot() } } as const; }, undefined, async function _Prepare(context)
 		{
 			(context.prisma as unknown as { pendingChildren: string[] }).pendingChildren.push("child-1");
 		})).resolves.toEqual({ outcome: "denied", reason: RunAdmissionDenialReasons.AuthorityConflict });
 		expect(committedChildren).toEqual([]);
 	});
 
@@ -150,7 +156,7 @@
 		const repository = new PrismaRunAdmissionRepository(prisma);
 		const prepare = vi.fn();
 
 		await expect(repository.admit(_command(), async function _UnexpectedBuild() { throw new Error("unexpected build"); }, undefined, prepare)).resolves.toEqual({ outcome: "idempotent", snapshot });
 		expect(prepare).not.toHaveBeenCalled();
 	});
 
@@ -162,7 +168,7 @@
 		const repository = new PrismaRunAdmissionRepository(prisma, { now: function _now() { return new Date("2026-07-20T00:05:00.000Z"); } });
 		let compiled = false;
 
 		await expect(repository.admit({ ..._command(), conversationId: null }, async function _build() { compiled = true; return { outcome: "ready", value: { authority: _authority(), snapshot } } as const; })).resolves.toEqual({ outcome: "idempotent", snapshot });
 		expect(compiled).toBe(false);
 		expect(transaction.agentRun.create).not.toHaveBeenCalled();
 		expect(transaction.runInputSnapshot.create).not.toHaveBeenCalled();
@@ -180,7 +186,7 @@
 		const repository = new PrismaRunAdmissionRepository(prisma);
 		const build = vi.fn();
 
 		await expect(repository.admit(_command(), build)).resolves.toEqual({ outcome: "denied", reason: "authority_conflict" });
 		expect(build).not.toHaveBeenCalled();
 		expect(transaction.runInputSnapshot.findUnique).not.toHaveBeenCalled();
 		expect(transaction.agentRun.create).not.toHaveBeenCalled();

libs/backend/agents/execution/runs/main/src/admission-lock-key.ts

@@ -0,0 +1,14 @@
+/**
+ * Builds the advisory-lock key that serialises one silo's admission of one idempotency key.
+ *
+ * The key is a text parameter to `hashtextextended`, and PostgreSQL text cannot hold a NUL byte —
+ * a value carrying one fails the whole statement with SQLSTATE 22021, so joining the two parts with
+ * NUL made every admission fail. The silo's length prefixes the pair instead: distinct pairs stay
+ * distinct keys without reserving a character either part might legitimately contain.
+ *
+ * Called by: PrismaRunAdmissionRepository.admit, PrismaChildRunReservationRepository (reservation).
+ */
+export function __AdmissionLockKey(siloId: string, requestIdempotencyKey: string): string
+{
+	return `${siloId.length}:${siloId}${requestIdempotencyKey}`;
+}

libs/backend/agents/execution/runs/main/src/prisma-child-run-reservation-repository.ts

@@ -3,6 +3,7 @@ import { AgentRunState, AgentRunTrigger, Prisma, type AgentRun as PrismaAgentRun
 import type { RunInputSnapshot } from "@opencrane/contracts";
 import { ___CreateLogger, type Logger } from "@opencrane/backend/observability";
 
+import { __AdmissionLockKey } from "./admission-lock-key";
 import { __PrepareChildRunAdmission } from "./child-run-admission";
 import type { ChildRunParentAuthority, PrepareChildRunAdmissionCommand, PreparedChildRunAdmission } from "./child-run-admission.types";
 import { _InitialRunOutboxData, _RunInputSnapshot, _RunInputSnapshotData } from "./prisma-run-admission-repository";
@@ -33,7 +34,7 @@ export class PrismaChildRunReservationRepository implements ChildRunReservationR
 			return await this.prisma.$transaction(async function _reserve(transaction): Promise<ChildRunReservationResult>
 			{
 				// 1. Serialize one inherited-silo key before observing or creating a child.
-				await transaction.$queryRaw(Prisma.sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${command.prepared.siloId}\u0000${command.requestIdempotencyKey}`}, 0))`);
+				await transaction.$queryRaw(Prisma.sql`SELECT pg_advisory_xact_lock(hashtextextended(${__AdmissionLockKey(command.prepared.siloId, command.requestIdempotencyKey)}, 0))`);
 				const existing = await transaction.agentRun.findUnique({ where: { siloId_requestIdempotencyKey: { siloId: command.prepared.siloId, requestIdempotencyKey: command.requestIdempotencyKey } } });
 				if (existing !== null)
 				{

libs/backend/agents/execution/runs/main/src/prisma-run-admission-repository.ts

@@ -5,6 +5,7 @@ import { ___CreateLogger, type Logger } from "@opencrane/backend/observability";
 import { ___CloneCanonicalJson } from "@opencrane/util";
 import type { JsonValue } from "@opencrane/util";
 
+import { __AdmissionLockKey } from "./admission-lock-key";
 import { RunAdmissionDenialReasons } from "./run-admission.types";
 import type { InitialRunAuthority, RunAdmissionBuild, RunAdmissionBuildResult, RunAdmissionClock, RunAdmissionCommand, RunAdmissionCommit, RunAdmissionPrepare, RunAdmissionRepository, RunAdmissionResult, RunAdmissionTransaction } from "./run-admission.types";
 
@@ -116,7 +117,7 @@ export class PrismaRunAdmissionRepository implements RunAdmissionRepository
 			return await this.prisma.$transaction(async function _admit(transaction: Prisma.TransactionClient): Promise<RunAdmissionResult<TDenial>>
 			{
 				// 1. Serialize the user-visible key before loading inputs so a duplicate never recompiles at a later instant.
-				await transaction.$queryRaw(Prisma.sql`SELECT pg_advisory_xact_lock(hashtextextended(${`${command.siloId}\u0000${command.requestIdempotencyKey}`}, 0))`);
+				await transaction.$queryRaw(Prisma.sql`SELECT pg_advisory_xact_lock(hashtextextended(${__AdmissionLockKey(command.siloId, command.requestIdempotencyKey)}, 0))`);
 				const existing = await transaction.agentRun.findUnique({ where: { siloId_requestIdempotencyKey: { siloId: command.siloId, requestIdempotencyKey: command.requestIdempotencyKey } } });
 				if (existing !== null)
 				{