A run and its ai.run.* event commit together again: events.runChanged enqueued into the PostgreSQL outbox without awaiting it (the async enqueue arrived with the PostgreSQL move) and all five callers in runs.js dropped the promise, so a failed event write was an unhandled rejection while the run change committed. runChanged is async and awaited inside each run transaction; a failed write rolls the change back. The outbox prune timer catches its async rejection (its try/catch could never see it). test/events.test.js: a rejecting enqueue fails recoverInterrupted and the run stays 'running'; a rejecting prune is not unhandled. Console handlers were checked and are covered by openvibe-shared's routeAsyncErrors. Found by the estate-wide floating-promise sweep; reviewed independently (VERDICT: SHIP). npm test 23/23 test files passed.

ddiff --git a/server/events.js b/server/events.js
iindex 448a028..a1872dc 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -35,7 +35,7 @@ function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = process.e
          onError: (err) => { const m = err && err.message; if (m !== stats.lastError) log.warn('[Events] publish failed (will retry):', m); stats.lastError = m; },
      });
      outbox.start();
--    pruneTimer = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);
++    pruneTimer = setInterval(() => { if (outbox) outbox.prune().catch((err) => log.warn('[Events] outbox prune failed:', err && err.message)); }, PRUNE_EVERY_MS);
      if (pruneTimer.unref) pruneTimer.unref();
      log.log(`[Events] ai → ${eventsUrl} (${outbox.pending()} pending)`);
      return outbox;
@@@ -72,13 +72,13 @@ function payloadOf(row) {
  }
  
  /**
-- * Queue the event for a run's new status. MUST run inside the transaction that changed it; a no-op
++ * Queue the event for a run's new status. MUST be awaited inside the transaction that changed it; a no-op
   * while publishing is off, and for states without an event (running, cancelled).
   */
--function runChanged(row) {
++async function runChanged(row) {
      if (!outbox || !row || !['queued', 'cached', 'succeeded', 'failed'].includes(row.status)) return null;
      const requester = { type: row.requester_type, id: row.requester_id };
--    const env = outbox.enqueue(dbRef, {
++    const env = await outbox.enqueue(dbRef, {
          event_type: `ai.run.${row.status}`,
          actor: requester.type === 'user' || requester.type === 'service' ? requester : { type: 'service', id: 'ai' },
          subject: { type: 'run', id: row.id },
ddiff --git a/server/runs.js b/server/runs.js
iindex b9d7983..22df76f 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -212,7 +212,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                      if (src.length) await addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
                      // The reused output carries the grounding it was produced with.
                      await db.prepare('UPDATE runs SET grounding = (SELECT grounding FROM runs WHERE id = ?) WHERE id = ?').run(hit.run_id, id);
--                    events.runChanged(await getRow(id));
++                    await events.runChanged(await getRow(id));
                  });
                  await registry.audit(principal.sub, 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, status: 'cached', cached_from: hit.run_id } });
                  return { run: await get(id), created: true, promise: null };
@@@ -231,7 +231,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                  VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'queued', @requester_type, @requester_id, @on_behalf_of, @attribution,
                  @source_service, @target, @target_key, @input, @input_hash, @cache_key, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @created_at)`)
                  .run({ ...base, cache_key: cacheKey });
--            events.runChanged(await getRow(id));
++            await events.runChanged(await getRow(id));
          });
          await registry.audit(principal.sub, retryOf ? 'run.retry' : 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, target: ctx.targetKey, retry_of: retryOf } });
  
@@@ -304,7 +304,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
                          r.route ? r.route.key : null, r.route ? r.route.version : null, iso(clock.now()), id);
                  if (!d.changes) return d;
                  if (r.citations.length) await addCitations(id, r.citations, 'workflow');
--                events.runChanged(await getRow(id));
++                await events.runChanged(await getRow(id));
                  return d;
              });
              if (!done.changes) return;       // cancelled while finishing
@@@ -324,7 +324,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              await db.tx(async () => {
                  const f = await db.prepare(`UPDATE runs SET status = ?, error_code = ?, error_detail = ?, attempts = ?, finished_at = ? WHERE id = ? AND status IN ('queued', 'running')`)
                      .run(cancelled ? 'cancelled' : 'failed', code, extra ? `${detail} ${extra}` : detail, attempts, iso(clock.now()), id);
--                if (f.changes) events.runChanged(await getRow(id));
++                if (f.changes) await events.runChanged(await getRow(id));
              });
              const spent = await db.prepare("SELECT COALESCE(SUM(tokens_in),0)::bigint ti, COALESCE(SUM(tokens_out),0)::bigint tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
              if (reserved && spent && (spent.ti || spent.tout)) await quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });
@@@ -398,7 +398,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          return await db.tx(async () => {
              const open = await db.prepare("SELECT id FROM runs WHERE status IN ('queued', 'running')").all();
              const r = await db.prepare("UPDATE runs SET status = 'failed', error_code = 'run.interrupted', error_detail = 'the service restarted while this run was in progress', finished_at = ? WHERE status IN ('queued', 'running')").run(iso(clock.now()));
--            for (const { id } of open) events.runChanged(await getRow(id));
++            for (const { id } of open) await events.runChanged(await getRow(id));
              return r.changes;
          });
      }
ddiff --git a/test/events.test.js b/test/events.test.js
iindex d5a30fa..4621106 100644
---- a/test/events.test.js
++++ b/test/events.test.js
@@@ -60,6 +60,48 @@ t.test('a failed run queues ai.run.failed with its error code; a restart announc
      valid(ev);
  });
  
++t.test('an outbox write that fails rejects the caller, so a run and its event cannot diverge silently', async () => {
++    const events = require('../server/events');
++    const outbox = events.init(h.db);                       // the live outbox: init returns it once events is on
++    assert.ok(outbox, 'events is on, so there is an outbox to fail');
++    const row = await h.db.prepare("SELECT * FROM runs WHERE status = 'succeeded' LIMIT 1").get();
++    await h.db.prepare("INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, created_at) VALUES ('run_01JAB2C3D4E5F6G7H8J9K0MNPS', ?, ?, 'running', 'service', 'live', '{}', 'x', ?)")
++        .run(row.workflow_key, row.workflow_version, new Date().toISOString());
++    const real = outbox.enqueue;
++    outbox.enqueue = async () => { throw new Error('outbox write failed'); };
++    try {
++        await assert.rejects(h.runs.recoverInterrupted(), /outbox write failed/, 'the transaction must fail rather than commit the run change with its event silently absent');
++        const after = await h.db.prepare("SELECT status FROM runs WHERE id = 'run_01JAB2C3D4E5F6G7H8J9K0MNPS'").get();
++        assert.strictEqual(after.status, 'running', 'the run change rolled back with its event');
++    } finally {
++        outbox.enqueue = real;
++        await h.db.prepare("DELETE FROM runs WHERE id = 'run_01JAB2C3D4E5F6G7H8J9K0MNPS'").run();
++    }
++});
++
++t.test('a failing outbox prune is caught, not left as an unhandled rejection', async () => {
++    const events = require('../server/events');
++    events._reset();                                        // stop the live outbox so init builds a fresh one
++    const realSetInterval = global.setInterval;
++    let tick = null;
++    global.setInterval = (fn) => { tick = fn; return { unref() {} }; };
++    let rejection;
++    const onUnhandled = (e) => { rejection = e; };
++    process.on('unhandledRejection', onUnhandled);
++    try {
++        const outbox = events.init(h.db, { log: { log() {}, warn() {} } });
++        assert.ok(outbox && tick, 'the prune timer was scheduled');
++        outbox.prune = async () => { throw new Error('prune failed'); };
++        tick();                                             // the timer fires; the rejection must be caught
++        await new Promise((r) => setImmediate(r));
++        await new Promise((r) => setImmediate(r));
++        assert.strictEqual(rejection, undefined, 'the prune rejection was handled, not left unhandled');
++    } finally {
++        process.removeListener('unhandledRejection', onUnhandled);
++        global.setInterval = realSetInterval;
++    }
++});
++
  t.test('close() stops the ai.run.* relay (unsent rows stay in the outbox)', async () => {
      const events = require('../server/events');
      const pending = (await events.status()).pending;