ai.run.* events (W13 gate): queued, cached, succeeded and failed runs each queue their openvibe-contracts ai.run.<status>@1 event in the same transaction as the state change (succeeded with its citations; interrupted runs at boot too), published by an openvibe-sdk outbox with AI's service token; internal visibility, ids, workflow, requester, usage and counts only, never input or output. EVENTS_URL enables it. test/events.test.js

ddiff --git a/package-lock.json b/package-lock.json
iindex bd5a3ae..02a2439 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -14,6 +14,7 @@
          "dotenv": "^16.4.7",
          "express": "^4.21.2",
          "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",
++        "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
          "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
        },
        "engines": {
@@@ -1409,6 +1410,15 @@
          "node": ">=20"
        }
      },
++    "node_modules/openvibe-sdk": {
++      "version": "0.7.0",
++      "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
++      "integrity": "sha512-G+nevM1cujNypp0oU0Ss1iVTQjbnDRmeUhS5uup6+OBBYQEoUwCBP/d2O3TBioQ2mjroAEbsCTACTxulok/BAw==",
++      "license": "MIT",
++      "engines": {
++        "node": ">=20"
++      }
++    },
      "node_modules/openvibe-shared": {
        "version": "1.7.0",
        "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0",
ddiff --git a/package.json b/package.json
iindex 4f8c61c..a500d79 100644
---- a/package.json
++++ b/package.json
@@@ -20,6 +20,7 @@
      "dotenv": "^16.4.7",
      "express": "^4.21.2",
      "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.33.0",
++    "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.7.0",
      "openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.7.0"
    },
    "optionalDependencies": {
ddiff --git a/server/events.js b/server/events.js
nnew file mode 100644
iindex 0000000..1e732b3
---- /dev/null
++++ b/server/events.js
@@@ -0,0 +1,97 @@
++'use strict';
++/**
++ * AI → OpenVibe.Events: ai.run.queued | cached | succeeded | failed (openvibe-contracts payloads
++ * ai.run.*@1), so products and the operator console learn a run's outcome without polling.
++ *
++ * Each event is written to AI's own outbox in the SAME transaction as the run's state change (the SDK
++ * outbox refuses anything else) and a relay publishes it with AI's service token (audience
++ * openvibe.events, capability events.event.publish). Events down: rows wait and are retried; a run
++ * never waits on Events. Payloads carry ids, the workflow, the requester, usage and counts, never
++ * the input, the output or a prompt. Every ai.run.* event is internal (runs are private).
++ *
++ * Off unless EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set (EVENTS_PUBLISH=off disables it).
++ */
++const { createClient } = require('openvibe-sdk/core');
++const { createServiceTokenClient } = require('openvibe-sdk/auth');
++const { createEventsClient, createOutbox } = require('openvibe-sdk/events');
++
++let outbox = null;
++const stats = { queued: 0, lastError: null };
++const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
++
++function init(db, { eventsUrl = process.env.EVENTS_URL, clientSecret = process.env.OV_OAUTH_CLIENT_SECRET, clientId = process.env.OV_OAUTH_CLIENT_ID || 'ai',
++    networkUrl = process.env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000', fetchImpl, intervalMs, log = console } = {}) {
++    if (outbox) return outbox;
++    if (process.env.EVENTS_PUBLISH === 'off' || !eventsUrl || !clientSecret) return null;
++    const tokens = createServiceTokenClient({ tokenUrl: `${String(networkUrl).replace(/\/+$/, '')}/oauth/token`, clientId, clientSecret, fetch: fetchImpl });
++    const client = createClient({ baseUrls: { events: String(eventsUrl).replace(/\/+$/, '') }, tokenProvider: tokens, fetch: fetchImpl, retries: 0 });
++    outbox = createOutbox(db, {
++        events: createEventsClient(client, { source: 'ai' }),
++        intervalMs: intervalMs || 2000,
++        onError: (err) => { const m = err && err.message; if (m !== stats.lastError) log.warn('[Events] publish failed (will retry):', m); stats.lastError = m; },
++    });
++    outbox.ensureSchema();
++    outbox.start();
++    const prune = setInterval(() => { try { outbox.prune(); } catch { /* next time */ } }, PRUNE_EVERY_MS);
++    if (prune.unref) prune.unref();
++    log.log(`[Events] ai → ${eventsUrl} (${outbox.pending()} pending)`);
++    return outbox;
++}
++
++const iso = (v) => (v ? new Date(v).toISOString() : null);
++const parse = (s) => { try { return s ? JSON.parse(s) : null; } catch { return null; } };
++
++/** A runs row → the ai.run.<status>@1 payload. */
++function payloadOf(row) {
++    const status = row.status;
++    const failed = status === 'failed';
++    const done = status === 'succeeded' || status === 'failed' || status === 'cached';
++    return {
++        run_id: row.id,
++        status,
++        workflow: { key: row.workflow_key, version: Number(row.workflow_version) },
++        requester: { type: row.requester_type, id: row.requester_id },
++        on_behalf_of: parse(row.on_behalf_of),
++        target: parse(row.target),
++        attribution: parse(row.attribution),
++        provider: status === 'succeeded' ? row.provider_key || null : null,
++        model: status === 'succeeded' || status === 'cached' ? row.model_key || null : null,
++        fallback_used: Boolean(row.fallback_used),
++        synthetic: Boolean(row.synthetic),
++        cached_from: status === 'cached' ? row.cached_from || null : null,
++        usage: { tokens_in: Number(row.tokens_in) || 0, tokens_out: Number(row.tokens_out) || 0, cost_usd: Number(row.cost_usd) || 0, attempts: Number(row.attempts) || 0 },
++        error: failed ? { code: row.error_code || 'run.internal', detail: String(row.error_detail || '').slice(0, 500) } : null,
++        citations_count: Number(row.citations_count) || 0,
++        retry_of: row.retry_of || null,
++        created_at: iso(row.created_at),
++        finished_at: done ? iso(row.finished_at || row.created_at) : null,
++    };
++}
++
++/**
++ * Queue the event for a run's new status. MUST run inside the transaction that changed it; a no-op
++ * while publishing is off, and for states without an event (running, cancelled).
++ */
++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({
++        event_type: `ai.run.${row.status}`,
++        actor: requester.type === 'user' || requester.type === 'service' ? requester : { type: 'service', id: 'ai' },
++        subject: { type: 'run', id: row.id },
++        visibility: 'internal',
++        priority: row.status === 'failed' ? 'important' : 'low',
++        payload: payloadOf(row),
++    });
++    stats.queued++;
++    setImmediate(() => outbox && outbox.kick());
++    return env;
++}
++
++function status() {
++    if (!outbox) return { enabled: false };
++    return { enabled: true, pending: outbox.pending(), rejected: outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
++}
++function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
++
++module.exports = { init, runChanged, payloadOf, status, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex e0aba82..d164df1 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -31,6 +31,8 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
      const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log });
      const fetcher = createFetcher(config);
      const engine = createEngine({ registry, pool, fetcher });
++    // ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
++    require('./events').init(db, { log });
      const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log });
      seed({ registry, quotas, config, env, db });
      const interrupted = runs.recoverInterrupted();
ddiff --git a/server/runs.js b/server/runs.js
iindex 7adc65d..f02beda 100644
---- a/server/runs.js
++++ b/server/runs.js
@@@ -19,6 +19,7 @@
  const { ids } = require('openvibe-contracts');
  const { AiError, sha256, stableStringify, parseJson, iso } = require('./util');
  const schemas = require('./schemas');
++const events = require('./events');
  
  const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'cached']);
  const QUEUE_RETRY_AFTER_S = 5;
@@@ -192,13 +193,16 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });
              const hit = cache.get(cacheKey, scope);
              if (hit) {
--                db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
--                    source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, created_at, started_at, finished_at, citations_count)
--                    VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'cached', @requester_type, @requester_id, @on_behalf_of, @attribution,
--                    @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @created_at, @created_at, @created_at, 0)`)
--                    .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });
--                const src = citations(hit.run_id);
--                if (src.length) addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
++                db.transaction(() => {
++                    db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
++                        source_service, target, target_key, input, input_hash, output, model_key, cache_key, cached_from, retry_of, idempotency_key, trace_id, request_id, options, created_at, started_at, finished_at, citations_count)
++                        VALUES (@id, @workflow_key, @workflow_version, @template_key, @template_version, @route_key, @route_version, 'cached', @requester_type, @requester_id, @on_behalf_of, @attribution,
++                        @source_service, @target, @target_key, @input, @input_hash, @output, @model_key, @cache_key, @cached_from, @retry_of, @idempotency_key, @trace_id, @request_id, @options, @created_at, @created_at, @created_at, 0)`)
++                        .run({ ...base, output: JSON.stringify(hit.output), model_key: hit.model_key, cache_key: cacheKey, cached_from: hit.run_id });
++                    const src = citations(hit.run_id);
++                    if (src.length) addCitations(id, src.map(c => ({ ...c, provenance: { ...c.provenance, via_cache: hit.run_id } })), 'cache');
++                    events.runChanged(getRow(id));
++                })();
                  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: get(id), created: true, promise: null };
              }
@@@ -209,11 +213,14 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          admit(caller);
          const reserved = quotas.reserve(ctx);
  
--        db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, 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)
--            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 });
++        db.transaction(() => {
++            db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, 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)
++                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(getRow(id));
++        })();
          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 } });
  
          const controller = new AbortController();
@@@ -269,12 +276,18 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
          try {
              const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash });
              if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
--            const done = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,
--                route_key = COALESCE(?, route_key), route_version = COALESCE(?, route_version), finished_at = ? WHERE id = ? AND status = 'running'`)
--                .run(JSON.stringify(r.output), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,
--                    r.route ? r.route.key : null, r.route ? r.route.version : null, iso(clock.now()), id);
++            // The status, its citations and the ai.run.succeeded event commit together.
++            const done = db.transaction(() => {
++                const d = db.prepare(`UPDATE runs SET status = 'succeeded', output = ?, synthetic = ?, provider_key = ?, model_key = ?, fallback_used = ?, attempts = ?, tokens_in = ?, tokens_out = ?, cost_usd = ?,
++                    route_key = COALESCE(?, route_key), route_version = COALESCE(?, route_version), finished_at = ? WHERE id = ? AND status = 'running'`)
++                    .run(JSON.stringify(r.output), r.synthetic ? 1 : 0, r.provider, r.model, r.fallbackUsed ? 1 : 0, attempts, r.usage.input, r.usage.output, r.cost,
++                        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) addCitations(id, r.citations, 'workflow');
++                events.runChanged(getRow(id));
++                return d;
++            })();
              if (!done.changes) return;       // cancelled while finishing
--            if (r.citations.length) addCitations(id, r.citations, 'workflow');
              if (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
              if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
              if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none') {
@@@ -287,8 +300,11 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
              const detail = interrupted ? 'the service shut down while this run was in progress' : cancelled ? 'cancelled' : (err instanceof AiError ? err.detail : 'internal error');
              if (!(err instanceof AiError) && !cancelled) log.error(`[runs] ${id} crashed: ${err && err.stack || err}`);
              const extra = err instanceof AiError && err.extra ? JSON.stringify(err.extra).slice(0, 2000) : null;
--            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);
++            db.transaction(() => {
++                const f = 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(getRow(id));
++            })();
              const spent = db.prepare("SELECT COALESCE(SUM(tokens_in),0) ti, COALESCE(SUM(tokens_out),0) tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
              if (spent && (spent.ti || spent.tout)) quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });
          } finally {
@@@ -358,8 +374,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
  
      /** Boot: runs interrupted by a restart are failed explicitly (retry is one call away). */
      function recoverInterrupted() {
--        const r = 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()));
--        return r.changes;
++        return db.transaction(() => {
++            const open = db.prepare("SELECT id FROM runs WHERE status IN ('queued', 'running')").all();
++            const r = 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(getRow(id));
++            return r.changes;
++        })();
      }
  
      function prune(days = config.runs.retentionDays) {
ddiff --git a/test/events.test.js b/test/events.test.js
nnew file mode 100644
iindex 0000000..e7d52af
---- /dev/null
++++ b/test/events.test.js
@@@ -0,0 +1,64 @@
++'use strict';
++// ai.run.* events: queued, cached, succeeded and failed runs each queue one event in the run's own
++// transaction, the payloads validate against openvibe-contracts, and nothing of the input or output travels.
++const assert = require('assert');
++const contracts = require('openvibe-contracts');
++
++process.env.EVENTS_URL = 'http://127.0.0.1:9';           // unreachable: rows stay in the outbox
++process.env.OV_OAUTH_CLIENT_SECRET = 'ai-secret-for-tests';
++process.env.OV_NETWORK_INTERNAL_URL = 'http://127.0.0.1:9';
++const { boot, request, token, suite, ALL } = require('./helpers');
++
++const t = suite('events');
++const live = token('live', ALL);
++let h;
++const queued = () => h.db.prepare('SELECT envelope FROM event_outbox ORDER BY id').all().map((r) => JSON.parse(r.envelope));
++const valid = (env) => { const r = contracts.validate(`${env.event_type}@1`, env.payload); assert.ok(r.valid, `${env.event_type}: ${JSON.stringify(r.errors)}`); };
++
++t.test('boot with events on', async () => {
++    h = await boot({ env: { AI_STUB_FALLBACK: 'true' } });
++    assert.strictEqual(require('../server/events').status().enabled, true);
++});
++
++t.test('a run queues ai.run.queued then ai.run.succeeded, valid and without the prompt or the text', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'ai.generate', input: { prompt: 'secret prompt words' } } });
++    assert.strictEqual(r.status, 201, r.text);
++    const evs = queued().filter((e) => e.payload.run_id === r.body.run.id);
++    assert.deepStrictEqual(evs.map((e) => e.event_type), ['ai.run.queued', 'ai.run.succeeded']);
++    evs.forEach(valid);
++    assert.strictEqual(evs[0].visibility, 'internal');
++    assert.deepStrictEqual(evs[1].payload.requester, { type: 'service', id: 'live' });
++    assert.strictEqual(evs[1].payload.finished_at !== null, true);
++    assert.ok(!JSON.stringify(evs).includes('secret prompt'), 'no input');
++    assert.ok(!JSON.stringify(evs).includes(r.body.run.output.text.slice(0, 20)), 'no output');
++});
++
++t.test('a cache hit queues ai.run.cached with cached_from', async () => {
++    const body = { workflow: 'ai.generate', input: { prompt: 'cache me' } };
++    const a = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body });
++    const b = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body });
++    const ev = queued().find((e) => e.payload.run_id === b.body.run.id);
++    if (b.body.run.status !== 'cached') { assert.ok(ev, 'an event for the second run'); return; }   // this workflow may not cache
++    assert.strictEqual(ev.event_type, 'ai.run.cached');
++    assert.strictEqual(ev.payload.cached_from, a.body.run.id);
++    valid(ev);
++});
++
++t.test('a failed run queues ai.run.failed with its error code; a restart announces interrupted runs', async () => {
++    const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: live, body: { workflow: 'no.such.workflow', input: {} } });
++    assert.ok(r.status >= 400, 'refused before a run exists: no event');
++    const before = queued().length;
++    // A queued run left behind by a restart: recoverInterrupted fails it and announces it.
++    const row = h.db.prepare("SELECT * FROM runs WHERE status = 'succeeded' LIMIT 1").get();
++    h.db.prepare("INSERT INTO runs (id, workflow_key, workflow_version, status, requester_type, requester_id, input, input_hash, created_at) VALUES ('run_01JAB2C3D4E5F6G7H8J9K0MNPQ', ?, ?, 'running', 'service', 'live', '{}', 'x', ?)")
++        .run(row.workflow_key, row.workflow_version, new Date().toISOString());
++    assert.strictEqual(h.runs.recoverInterrupted(), 1);
++    const ev = queued().slice(before).find((e) => e.payload.run_id === 'run_01JAB2C3D4E5F6G7H8J9K0MNPQ');
++    assert.strictEqual(ev.event_type, 'ai.run.failed');
++    assert.strictEqual(ev.payload.error.code, 'run.interrupted');
++    assert.strictEqual(ev.priority, 'important');
++    valid(ev);
++});
++
++t.test('stop', async () => { require('../server/events')._reset(); await h.stop(); });
++t.run();