refactor(delivery): centralize durable queue routing metadata (#114928)

* refactor(delivery): centralize durable queue routing metadata

* fix(delivery): keep queue metadata private to its owner

src/infra/delivery-queue-sqlite.test.ts

@@ -8,6 +8,8 @@ import {
   promoteDeliveryQueueEntryPlatformSend,
 } from "./delivery-queue-sqlite-claim.js";
 import {
+  commitStagedDeliveryQueueEntry,
+  commitStagedDeliveryQueueEntryOnce,
   completeDeliveryQueueEntry,
   countFailedDeliveryQueueEntries,
   deleteDeliveryQueueEntry,
@@ -140,6 +142,156 @@ describe("delivery-queue-sqlite corrupt JSON resilience", () => {
       expect(loaded).toMatchObject({ id: "rt-1", enqueuedAt: 1000, retryCount: 0 });
     });
 
+    it.each([
+      {
+        name: "outbound delivery",
+        queueName: "outbound",
+        entry: {
+          id: "metadata-outbound",
+          channel: "discord",
+          to: "channel:123",
+          accountId: "bot-a",
+          session: { key: "agent:main:discord:channel:123" },
+        },
+        expected: {
+          entry_kind: "outbound",
+          session_key: "agent:main:discord:channel:123",
+          channel: "discord",
+          target: "channel:123",
+          account_id: "bot-a",
+        },
+      },
+      {
+        name: "routed session delivery",
+        queueName: "session",
+        entry: {
+          id: "metadata-session-route",
+          kind: "agentTurn",
+          sessionKey: "agent:main:discord:channel:123",
+          route: { channel: "discord", to: "channel:123", accountId: "bot-a" },
+          deliveryContext: { channel: "telegram", to: "999", accountId: "bot-b" },
+        },
+        expected: {
+          entry_kind: "agentTurn",
+          session_key: "agent:main:discord:channel:123",
+          channel: "discord",
+          target: "channel:123",
+          account_id: "bot-a",
+        },
+      },
+      {
+        name: "context-only session delivery",
+        queueName: "session",
+        entry: {
+          id: "metadata-session-context",
+          kind: "systemEvent",
+          sessionKey: "agent:main:telegram:direct:123",
+          deliveryContext: { channel: "telegram", to: "123", accountId: "bot-a" },
+        },
+        expected: {
+          entry_kind: "systemEvent",
+          session_key: "agent:main:telegram:direct:123",
+          channel: "telegram",
+          target: "123",
+          account_id: "bot-a",
+        },
+      },
+    ])("indexes canonical $name metadata", ({ queueName, entry, expected }) => {
+      upsertDeliveryQueueEntry({
+        queueName,
+        entry: { ...entry, enqueuedAt: 1000, retryCount: 0 },
+        stateDir,
+      });
+
+      const { db } = openOpenClawStateDatabase({
+        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
+      });
+      const readMetadata = () =>
+        db
+          .prepare(
+            `SELECT entry_kind, session_key, channel, target, account_id
+             FROM delivery_queue_entries WHERE queue_name = ? AND id = ?`,
+          )
+          .get(queueName, entry.id);
+      expect(readMetadata()).toEqual(expected);
+
+      updateDeliveryQueueEntry(queueName, entry.id, stateDir, (current) => ({
+        ...current,
+        retryCount: current.retryCount + 1,
+      }));
+      expect(readMetadata()).toEqual(expected);
+    });
+
+    it("preserves explicit queue metadata ownership", () => {
+      upsertDeliveryQueueEntry({
+        queueName: "outbound-media-staging",
+        entry: { id: "metadata-media-stage", enqueuedAt: 1000, retryCount: 0 },
+        metadata: { entryKind: "outbound-media-stage" },
+        stateDir,
+      });
+
+      const { db } = openOpenClawStateDatabase({
+        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
+      });
+      expect(
+        db
+          .prepare("SELECT entry_kind FROM delivery_queue_entries WHERE queue_name = ? AND id = ?")
+          .get("outbound-media-staging", "metadata-media-stage"),
+      ).toEqual({ entry_kind: "outbound-media-stage" });
+    });
+
+    it.each([
+      { name: "ordinary", commit: commitStagedDeliveryQueueEntry, expected: true },
+      { name: "insert-only", commit: commitStagedDeliveryQueueEntryOnce, expected: "created" },
+    ])("indexes $name staged outbound commits", ({ commit, expected }) => {
+      const stagingQueueName = "outbound-media-staging";
+      const stagingId = "metadata-staged-media";
+      const outboundEntry = {
+        id: "metadata-staged-outbound",
+        enqueuedAt: 1000,
+        retryCount: 0,
+        channel: "discord",
+        to: "channel:123",
+        accountId: "bot-a",
+        session: { key: "agent:main:discord:channel:123" },
+      };
+      upsertDeliveryQueueEntry({
+        queueName: stagingQueueName,
+        entry: { id: stagingId, enqueuedAt: 1000, retryCount: 0 },
+        metadata: { entryKind: "outbound-media-stage" },
+        stateDir,
+      });
+
+      expect(
+        commit({
+          queueName: "outbound",
+          entry: outboundEntry,
+          stagingId,
+          stagingQueueName,
+          stateDir,
+        }),
+      ).toBe(expected);
+
+      const { db } = openOpenClawStateDatabase({
+        env: { ...process.env, OPENCLAW_STATE_DIR: stateDir },
+      });
+      expect(
+        db
+          .prepare(
+            `SELECT entry_kind, session_key, channel, target, account_id
+             FROM delivery_queue_entries WHERE queue_name = ? AND id = ?`,
+          )
+          .get("outbound", "metadata-staged-outbound"),
+      ).toEqual({
+        entry_kind: "outbound",
+        session_key: "agent:main:discord:channel:123",
+        channel: "discord",
+        target: "channel:123",
+        account_id: "bot-a",
+      });
+      expect(loadDeliveryQueueEntry(stagingQueueName, stagingId, stateDir)).toBeNull();
+    });
+
     it("update increments retry count", () => {
       upsertDeliveryQueueEntry({
         queueName: QUEUE,

src/infra/delivery-queue-sqlite.ts

@@ -25,7 +25,7 @@ export type DeliveryQueueCompletionRetention =
     }>;
 
 /** Indexed metadata extracted from queue payloads for diagnostics and recovery. */
-export type DeliveryQueueRowMetadata = {
+type DeliveryQueueRowMetadata = {
   entryKind?: string;
   sessionKey?: string;
   channel?: string;
@@ -114,7 +114,7 @@ function inflate(row: QueueRow): DeliveryQueueEntryState | null {
   };
 }
 
-function metadata(entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata {
+function metadata(queueName: string, entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata {
   const item = entry as DeliveryQueueEntryState & {
     kind?: string;
     sessionKey?: string;
@@ -126,7 +126,7 @@ function metadata(entry: DeliveryQueueEntryState): DeliveryQueueRowMetadata {
     deliveryContext?: { channel?: string; to?: string; accountId?: string };
   };
   return {
-    entryKind: item.kind,
+    entryKind: item.kind ?? queueName,
     sessionKey: item.sessionKey ?? item.session?.key,
     channel: item.channel ?? item.route?.channel ?? item.deliveryContext?.channel,
     target: item.to ?? item.route?.to ?? item.deliveryContext?.to,
@@ -140,7 +140,7 @@ function upsertDeliveryQueueEntryInDatabase(
 ): boolean {
   const now = Date.now();
   const status = params.status ?? "pending";
-  const meta = params.metadata ?? metadata(params.entry);
+  const meta = params.metadata ?? metadata(params.queueName, params.entry);
   const queueDb = getNodeSqliteKysely<DeliveryQueueDatabase>(database.db);
   const insert = queueDb.insertInto("delivery_queue_entries").values({
     queue_name: params.queueName,

src/infra/outbound/delivery-queue-storage.ts

@@ -22,7 +22,6 @@ import {
   reserveDeliveryQueueEntryAttempt,
   updateDeliveryQueueEntry,
   upsertDeliveryQueueEntry,
-  type DeliveryQueueRowMetadata,
   type DeliveryQueueCompletionRetention,
 } from "../delivery-queue-sqlite.js";
 import { generateSecureUuid } from "../secure-random.js";
@@ -120,16 +119,6 @@ export interface QueuedDelivery extends QueuedDeliveryPayload {
   recoveryState?: "producer_claimed" | "send_attempt_started" | "unknown_after_send";
 }
 
-function queuedDeliveryMetadata(entry: QueuedDelivery): DeliveryQueueRowMetadata {
-  return {
-    entryKind: "outbound",
-    sessionKey: entry.session?.key,
-    channel: entry.channel,
-    target: entry.to,
-    accountId: entry.accountId,
-  };
-}
-
 function createQueuedDelivery(params: QueuedDeliveryPayload, id: string): QueuedDelivery {
   return {
     id,
@@ -172,12 +161,10 @@ export async function enqueueDelivery(
 ): Promise<string> {
   const id = generateSecureUuid();
   const entry = createQueuedDelivery(params, id);
-  const metadata = queuedDeliveryMetadata(entry);
   if (mediaStageId) {
     const committed = commitStagedDeliveryQueueEntry({
       queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
       entry,
-      metadata,
       stagingId: mediaStageId,
       stagingQueueName: DELIVERY_QUEUE_MEDIA_STAGING_QUEUE_NAME,
       stateDir,
@@ -189,7 +176,6 @@ export async function enqueueDelivery(
     upsertDeliveryQueueEntry({
       queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
       entry,
-      metadata,
       stateDir,
     });
   }
@@ -208,13 +194,11 @@ export async function enqueueDeliveryOnce(
     throw new Error("Stable delivery queue id is required");
   }
   const entry = createQueuedDelivery(params, normalizedId);
-  const metadata = queuedDeliveryMetadata(entry);
   const created = mediaStageId
     ? (() => {
         const result = commitStagedDeliveryQueueEntryOnce({
           queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
           entry,
-          metadata,
           stagingId: mediaStageId,
           stagingQueueName: DELIVERY_QUEUE_MEDIA_STAGING_QUEUE_NAME,
           stateDir,
@@ -227,7 +211,6 @@ export async function enqueueDeliveryOnce(
     : upsertDeliveryQueueEntry({
         queueName: OUTBOUND_DELIVERY_QUEUE_NAME,
         entry,
-        metadata,
         stateDir,
         insertOnly: true,
       });

src/infra/session-delivery-queue-storage.ts

@@ -12,7 +12,6 @@ import {
   updateDeliveryQueueEntry,
   upsertDeliveryQueueEntry,
   type DeliveryQueueCompletionRetention,
-  type DeliveryQueueRowMetadata,
 } from "./delivery-queue-sqlite.js";
 import { generateSecureUuid } from "./secure-random.js";
 
@@ -113,17 +112,6 @@ function buildEntryId(idempotencyKey?: string): string {
   return sha256Hex(idempotencyKey);
 }
 
-function queuedSessionDeliveryMetadata(entry: QueuedSessionDelivery): DeliveryQueueRowMetadata {
-  const route = entry.kind === "agentTurn" ? entry.route : undefined;
-  return {
-    entryKind: entry.kind,
-    sessionKey: entry.sessionKey,
-    channel: route?.channel ?? entry.deliveryContext?.channel,
-    target: route?.to ?? entry.deliveryContext?.to,
-    accountId: route?.accountId ?? entry.deliveryContext?.accountId,
-  };
-}
-
 /** Enqueue a session delivery and return its durable id. */
 export async function enqueueSessionDelivery(
   params: QueuedSessionDeliveryPayload,
@@ -140,7 +128,6 @@ export async function enqueueSessionDelivery(
   upsertDeliveryQueueEntry({
     queueName: QUEUE_NAME,
     entry,
-    metadata: queuedSessionDeliveryMetadata(entry),
     stateDir,
     ...(params.completionRetention === "permanent"
       ? { insertOnly: true }
@@ -170,7 +157,6 @@ export async function enqueueClaimedSessionDelivery(
   const claimed = upsertDeliveryQueueEntry({
     queueName: QUEUE_NAME,
     entry,
-    metadata: queuedSessionDeliveryMetadata(entry),
     stateDir,
     insertOnly: true,
   });
@@ -241,7 +227,6 @@ export async function markSessionDeliveryAttemptStarted(
         ...entry,
         deliveryStartedAt: entry.deliveryStartedAt ?? Date.now(),
       } as QueuedSessionDelivery,
-      metadata: queuedSessionDeliveryMetadata(entry),
       stateDir,
       updatePendingOnly: true,
     });
@@ -278,7 +263,6 @@ export async function markSessionDeliverySettlement(
         settlementOutcome: outcome,
         ...(outcome === "recovered" ? { acknowledgedAt: entry.acknowledgedAt ?? Date.now() } : {}),
       } as QueuedSessionDelivery,
-      metadata: queuedSessionDeliveryMetadata(entry),
       stateDir,
       updatePendingOnly: true,
     });