[codex] fix(outbox): preserve local FIFO event order

server/src/__tests__/plugin-event-outbox.test.ts

@@ -1,14 +1,15 @@
 import { randomUUID } from "node:crypto";
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
-import { companies, createDb, pluginEventOutbox } from "@paperclipai/db";
-import { eq } from "drizzle-orm";
+import { activityLog, companies, createDb, pluginEventOutbox, type Db } from "@paperclipai/db";
+import { asc, eq } from "drizzle-orm";
 import type { PluginEvent } from "@paperclipai/plugin-sdk";
 import {
   getEmbeddedPostgresTestSupport,
   startEmbeddedPostgresTestDatabase,
 } from "./helpers/embedded-postgres.js";
 import { createPluginEventBus } from "../services/plugin-event-bus.js";
 import { pollOnce, resetStaleProcessing } from "../services/plugin-event-outbox.js";
+import { logActivity, setPluginEventOutboxDb } from "../services/activity-log.js";
 
 const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
@@ -40,6 +41,8 @@ describeEmbeddedPostgres("plugin event outbox", () => {
   });
 
   beforeEach(async () => {
+    setPluginEventOutboxDb(db);
+    await db.delete(activityLog);
     await db.delete(pluginEventOutbox);
   });
 
@@ -103,6 +106,60 @@ describeEmbeddedPostgres("plugin event outbox", () => {
     expect(order).toEqual(["approval.created", "approval.decided"]);
   });
 
+  it("preserves sequential logActivity order when an outbox insert is delayed", async () => {
+    const pendingInserts: Promise<unknown>[] = [];
+    let resolveSecondInsertScheduled!: () => void;
+    const secondInsertScheduled = new Promise<void>((resolve) => {
+      resolveSecondInsertScheduled = resolve;
+    });
+    let insertCount = 0;
+    let delayFirstInsert = true;
+    const delayedOutboxDb = {
+      insert: (table: typeof pluginEventOutbox) => {
+        const builder = db.insert(table);
+        return {
+          values: (values: typeof pluginEventOutbox.$inferInsert) => {
+            const query = builder.values(values);
+            const completion = delayFirstInsert
+              ? new Promise<void>((resolve) => setImmediate(resolve)).then(() => query)
+              : Promise.resolve(query);
+            insertCount += 1;
+            if (insertCount === 2) resolveSecondInsertScheduled();
+            delayFirstInsert = false;
+            pendingInserts.push(completion);
+            return completion;
+          },
+        };
+      },
+    } as unknown as Db;
+    setPluginEventOutboxDb(delayedOutboxDb);
+
+    await logActivity(db, {
+      companyId,
+      actorType: "system",
+      actorId: "test",
+      action: "issue.created",
+      entityType: "issue",
+      entityId: randomUUID(),
+    });
+    await logActivity(db, {
+      companyId,
+      actorType: "system",
+      actorId: "test",
+      action: "issue.updated",
+      entityType: "issue",
+      entityId: randomUUID(),
+    });
+    await secondInsertScheduled;
+    await Promise.all(pendingInserts);
+
+    const rows = await db
+      .select({ eventType: pluginEventOutbox.eventType })
+      .from(pluginEventOutbox)
+      .orderBy(asc(pluginEventOutbox.seq));
+    expect(rows.map((row) => row.eventType)).toEqual(["issue.created", "issue.updated"]);
+  });
+
   it("marks processed even when a handler throws (no poison loop)", async () => {
     const bus = createPluginEventBus();
     bus.forPlugin("test").subscribe("approval.created", async () => {

server/src/services/activity-log.ts

@@ -36,6 +36,7 @@ const ACTIVITY_ACTION_TO_PLUGIN_EVENT: Readonly<Record<string, PluginEventType>>
 
 let _pluginEventBus: PluginEventBus | null = null;
 let _outboxDb: Db | null = null;
+let _outboxEnqueueTail: Promise<void> | null = null;
 
 /** Wire the plugin event bus so domain events are forwarded to plugins. */
 export function setPluginEventBus(bus: PluginEventBus): void {
@@ -68,27 +69,42 @@ function eventTypeForActivityAction(action: string): PluginEventType | null {
  * in-process: the worker-tier poller (plugin-event-outbox.ts) is the sole
  * emitter, so events raised on any tier (notably the API tier, where plugins
  * are not loaded) reliably reach subscribed plugins. One writer + one emitter
- * ⇒ no double-delivery. Fire-and-forget to keep the signature synchronous.
+ * ⇒ no double-delivery. Writes are serialized within this process so the
+ * database `seq` reflects call order even though this remains fire-and-forget.
+ * Each insert is still best-effort: a failure is logged and does not block the
+ * next event or surface to the activity caller.
  */
 export function publishPluginDomainEvent(event: PluginEvent): void {
-  if (!_outboxDb) {
+  const outboxDb = _outboxDb;
+  if (!outboxDb) {
     logger.warn(
       { eventType: event.eventType, eventId: event.eventId },
       "plugin event outbox db not set; dropping event",
     );
     return;
   }
-  void _outboxDb
-    .insert(pluginEventOutbox)
-    .values({
-      eventId: event.eventId,
-      companyId: event.companyId,
-      eventType: event.eventType,
-      payload: event as unknown as Record<string, unknown>,
-    })
-    .catch((err) =>
-      logger.warn({ err, eventType: event.eventType }, "failed to enqueue plugin event to outbox"),
-    );
+  const values = {
+    eventId: event.eventId,
+    companyId: event.companyId,
+    eventType: event.eventType,
+    payload: event as unknown as Record<string, unknown>,
+  };
+
+  const enqueue = async () => {
+    try {
+      await outboxDb.insert(pluginEventOutbox).values(values);
+    } catch (err) {
+      logger.warn({ err, eventType: event.eventType }, "failed to enqueue plugin event to outbox");
+    }
+  };
+  // Start the first writer immediately, preserving the original detached timing.
+  // Later calls chain behind it, so they cannot claim an earlier `seq`.
+  const pending = _outboxEnqueueTail
+    ? _outboxEnqueueTail.then(enqueue, enqueue)
+    : enqueue();
+  // Keep an unexpected failure from poisoning the chain or becoming unhandled;
+  // normal insert failures are already logged and swallowed above.
+  _outboxEnqueueTail = pending.catch(() => {});
 }
 
 /**