Answer errors, outbox and shutdown through openvibe-sdk/service (#6)

ddiff --git a/server/api/admin.js b/server/api/admin.js
iindex b685dc2..a82203f 100644
---- a/server/api/admin.js
++++ b/server/api/admin.js
@@@ -22,8 +22,7 @@
   */
  const express = require('express');
  const { CAPS } = require('../auth');
--const { AiError } = require('../util');
--const { sendError } = require('./runs');
++const { AiError, sendError } = require('../util');
  const { createOps } = require('../ops');
  
  function adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops = createOps({ db, registry, pool, quotas, cache, runs }), log = console }) {
ddiff --git a/server/api/runs.js b/server/api/runs.js
iindex 4be3013..c8493de 100644
---- a/server/api/runs.js
++++ b/server/api/runs.js
@@@ -17,22 +17,9 @@
   * problem+json; a quota refusal or a full run queue (queue.full) is 429 with Retry-After.
   */
  const express = require('express');
--const { http } = require('openvibe-contracts');
--const { AiError } = require('../util');
++const { AiError, sendError } = require('../util');
  const { CAPS, namespaceAllowed } = require('../auth');
  
--function sendError(res, err, ctx, log = console) {
--    if (err instanceof AiError) {
--        const extra = err.extra && typeof err.extra === 'object' ? { ...err.extra } : undefined;
--        const errors = extra && Array.isArray(extra.errors) ? extra.errors : undefined;
--        if (extra) delete extra.errors;
--        if (err.status === 429 && extra && extra.retry_after_seconds) res.setHeader('Retry-After', String(extra.retry_after_seconds));
--        return http.sendProblem(res, err.status, err.code, { detail: err.detail, ctx, errors, extra });
--    }
--    log.error(`[api] ${err && err.stack || err}`);
--    return http.sendProblem(res, 500, 'ai.internal', { detail: 'internal error', ctx });
--}
--
  function runsRouter({ runs, registry, auth, config, log = console }) {
      const r = express.Router();
      const create = auth.requireCap(CAPS.runCreate);
@@@ -130,4 +117,4 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
      return r;
  }
  
--module.exports = { runsRouter, sendError };
++module.exports = { runsRouter };
ddiff --git a/server/app.js b/server/app.js
iindex 1ad9f24..3c8bbbb 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -7,6 +7,7 @@ const { instrument } = require('openvibe-shared/metrics');
  const { createRelease } = require('openvibe-shared/release');
  const { createAiReadiness, registerAiGauges } = require('./observability');
  const { runsRouter } = require('./api/runs');
++const { sendError } = require('./util');
  const { adminRouter } = require('./api/admin');
  const { createOps } = require('./ops');
  const { consoleRouter } = require('./console');
@@@ -46,8 +47,8 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
      // Operator actions shared by the admin API and the operator console (one implementation, one audit row).
      const ops = createOps({ db, registry, pool, quotas, cache, runs });
      app.use(runsRouter({ runs, registry, auth, config, log }));
--    app.use(require('./api/attribution-quotas').attributionQuotasRouter({ db, quotas, auth, clock, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));
--    if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));
++    app.use(require('./api/attribution-quotas').attributionQuotasRouter({ db, quotas, auth, clock, sendError: (res, err, ctx) => sendError(res, err, ctx, log) }));
++    if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => sendError(res, err, ctx, log) }));
      app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));
      // The operator console (/console, /auth/*): Network staff, server-rendered, no scripts (server/console).
      app.use(consoleRouter({ config, db, registry, pool, quotas, cache, runs, ops, env, clock, fetchImpl, log }));
@@@ -78,13 +79,14 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
  
      app.use((req, res) => http.sendProblem(res, 404, 'ai.not_found', { detail: `no route ${req.method} ${req.path}`, ctx: req.ov }));
  
++    // AI's own codes for a miss and a bad body (openvibe-sdk/service jsonErrors() would answer not_found and
++    // request.invalid_json); everything else goes through sendError (openvibe-sdk/service): 500 ai.internal.
      // eslint-disable-next-line no-unused-vars
      app.use((err, req, res, _next) => {
          if (err.type === 'entity.parse.failed') return http.sendProblem(res, 400, 'input.bad_json', { detail: 'request body is not valid JSON', ctx: req.ov });
          if (err.type === 'entity.too.large') return http.sendProblem(res, 413, 'input.too_large', { detail: 'request body too large', ctx: req.ov });
--        log.error(`[app] ${req.method} ${req.path}: ${err.stack || err}`);
--        if (res.headersSent) return res.end();
--        return http.sendProblem(res, 500, 'ai.internal', { detail: 'internal error', ctx: req.ov });
++        if (res.headersSent) { log.error(`[app] ${req.method} ${req.path}: ${err.stack || err}`); return res.end(); }
++        return sendError(res, { stack: `${req.method} ${req.path}: ${err.stack || err}` }, req.ov, log);   // never the error's own status: a 500
      });
  
      app.locals.metrics = metrics;
ddiff --git a/server/events.js b/server/events.js
iindex a1872dc..d72b99e 100644
---- a/server/events.js
++++ b/server/events.js
@@@ -11,34 +11,30 @@
   *
   * 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, createPgOutbox } = require('openvibe-sdk/events');
++const { createServiceOutbox } = require('openvibe-sdk/events');
  
--let outbox = null;
--let dbRef = null;   // the handle enqueue joins the running transaction through (ADR-035)
++let svc = null;     // openvibe-sdk/events createServiceOutbox (event_outbox) while publishing is on
++let dbRef = null;   // the handle a run change commits through (its afterCommit kicks the relay)
  let pruneTimer = null;
--const stats = { queued: 0, lastError: null };
++let queued = 0;
  const PRUNE_EVERY_MS = 6 * 60 * 60 * 1000;
  
++/** Starts the relay and returns the PostgreSQL outbox under it (null while publishing is off). */
  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 (svc) return svc.outbox;
      dbRef = db;
      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 });
--    // The PostgreSQL outbox (ADR-035): rows are written in the change's own transaction (enqueue(db, …) joins it).
--    outbox = createPgOutbox(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; },
++    // The PostgreSQL outbox (ADR-035): emit joins the change's own transaction.
++    svc = createServiceOutbox({
++        db, source: 'ai', eventsUrl: String(eventsUrl).replace(/\/+$/, ''), networkInternalUrl: networkUrl, clientId, clientSecret,
++        intervalMs: intervalMs || 2000, fetch: fetchImpl, log,
      });
--    outbox.start();
--    pruneTimer = setInterval(() => { if (outbox) outbox.prune().catch((err) => log.warn('[Events] outbox prune failed:', err && err.message)); }, PRUNE_EVERY_MS);
++    svc.start();
++    pruneTimer = setInterval(() => { if (svc) svc.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;
++    log.log(`[Events] ai → ${eventsUrl} (${svc.outbox.pending()} pending)`);
++    return svc.outbox;
  }
  
  const iso = (v) => (v ? new Date(v).toISOString() : null);
@@@ -76,9 +72,9 @@ function payloadOf(row) {
   * while publishing is off, and for states without an event (running, cancelled).
   */
  async function runChanged(row) {
--    if (!outbox || !row || !['queued', 'cached', 'succeeded', 'failed'].includes(row.status)) return null;
++    if (!svc || !row || !['queued', 'cached', 'succeeded', 'failed'].includes(row.status)) return null;
      const requester = { type: row.requester_type, id: row.requester_id };
--    const env = await outbox.enqueue(dbRef, {
++    const env = await svc.emitIn(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 },
@@@ -86,15 +82,16 @@ async function runChanged(row) {
          priority: row.status === 'failed' ? 'important' : 'low',
          payload: payloadOf(row),
      });
--    stats.queued++;
++    queued++;
      // Relay once the change (and its event) committed.
--    dbRef.afterCommit(() => outbox && outbox.kick());
++    dbRef.afterCommit(() => svc && svc.kick());
      return env;
  }
  
  async function status() {
--    if (!outbox) return { enabled: false };
--    return { enabled: true, pending: await outbox.pending(), rejected: await outbox.rejected(), queued_since_boot: stats.queued, last_error: stats.lastError };
++    if (!svc) return { enabled: false };
++    const s = await svc.status();
++    return { enabled: true, pending: s.pending, rejected: s.rejected, queued_since_boot: queued, last_error: s.last_error };
  }
  /**
   * Graceful stop (start().close(), after runs.drain()): the relay stops and nothing more is queued; resolves
@@@ -103,10 +100,10 @@ async function status() {
  async function stop() {
      if (pruneTimer) clearInterval(pruneTimer);
      pruneTimer = null;
--    const o = outbox;
--    outbox = null;
++    const o = svc;
++    svc = null;
      if (o) await o.stop();
  }
--function _reset() { if (outbox) outbox.stop(); outbox = null; stats.queued = 0; stats.lastError = null; }
++function _reset() { if (svc) svc.stop(); svc = null; queued = 0; }
  
  module.exports = { init, runChanged, payloadOf, status, stop, _reset };
ddiff --git a/server/index.js b/server/index.js
iindex 9304455..6f5b126 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -21,6 +21,7 @@ const { createAuth, createNetworkKeys } = require('./auth');
  const { jwksClient } = require('openvibe-sdk/auth');
  const schemas = require('./schemas');
  const { createApp } = require('./app');
++const { gracefulStop } = require('openvibe-sdk/service');
  
  async function start({ config, db: givenDb = null, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true, credentialFetch = null } = {}) {
      config = config || load(env);
@@@ -102,13 +103,10 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
  if (require.main === module) {
      require('dotenv').config();
      start().then((handles) => {
--        const shutdown = (sig) => {
--            console.log(`[ai] ${sig}: shutting down`);
--            handles.close().then(() => process.exit(0), () => process.exit(1));
--            setTimeout(() => process.exit(1), 10000).unref();
--        };
--        process.on('SIGTERM', () => shutdown('SIGTERM'));
--        process.on('SIGINT', () => shutdown('SIGINT'));
++        // SIGTERM/SIGINT (openvibe-sdk/service, docs/service.md's handles family): requests in flight get 8 s,
++        // then handles.close() (runs drained, the relay stopped, the database closed; a rejection exits 1);
++        // past 10 s the process exits 1.
++        gracefulStop({ name: 'ai', server: handles.server, handles, drainMs: 8000, deadlineMs: 10000 });
      }).catch((err) => {
          console.error(`[ai] failed to start: ${err.stack || err}`);
          process.exit(1);
ddiff --git a/server/providers/index.js b/server/providers/index.js
iindex f14928d..5b0bd45 100644
---- a/server/providers/index.js
++++ b/server/providers/index.js
@@@ -222,7 +222,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
              return ea - eb || (a.score ?? Infinity) - (b.score ?? Infinity);
          });
          for (const x of rest) order.push(x.id);
--        return { order: order.map(splitOfferId), explain };
++        return { order: order.map(splitOfferId), explain, placement: result };
      }
  
      /** A pinned route: [primary, ...fallbacks], exactly as before. */
@@@ -239,14 +239,14 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
      }
  
      async function candidates(route, features, operation) {
--        const { order, explain } = route.capability && operation
++        const { order, explain, placement: result = null } = route.capability && operation
              ? await poolCandidates(route, features, operation)
              : await pinnedCandidates(route, features);
          if (config.stubFallback && !order.some(c => c.provider === 'stub') && await registry.getProvider('stub')) {
              order.push({ provider: 'stub', model: null });
              explain.candidates.push({ provider: 'stub', model: null, eligible: true, excluded_reason: null, cost: 0, latency: null });
          }
--        return { order, explain };
++        return { order, explain, placement: result };   // placement: the platform.placement-result@1 explain is built from (pool routes)
      }
  
      async function skipReason(p, a, features) {
ddiff --git a/server/util.js b/server/util.js
iindex c7957ed..49654ce 100644
---- a/server/util.js
++++ b/server/util.js
@@@ -1,6 +1,7 @@
  'use strict';
  /** Small helpers shared by every module: hashing, stable JSON, row decoding, errors. */
  const crypto = require('crypto');
++const svc = require('openvibe-sdk/service');
  
  /** JSON with sorted object keys, so equal inputs always hash equal. */
  function stableStringify(value) {
@@@ -23,15 +24,22 @@ function parseJson(text, fallback = null) {
  
  const iso = (ms = Date.now()) => new Date(ms).toISOString();
  
--/** An error that the API layer turns into a problem+json response. */
--class AiError extends Error {
--    constructor(status, code, detail, extra) {
--        super(detail || code);
--        this.status = status;
--        this.code = code;
--        this.detail = detail;
--        this.extra = extra;
--    }
++/** An error that the API layer turns into a problem+json response: (status, code, detail?, extra?). */
++const AiError = svc.createServiceError('AiError');
++
++const PROBLEMS = { name: 'api', internalCode: 'ai.internal', internalDetail: 'internal error' };
++
++/**
++ * Any error -> problem+json (openvibe-sdk/service sendError). Only an AiError answers with its own status,
++ * code and extra (extra.errors is the problem's errors list; a 429 with retry_after_seconds sets Retry-After).
++ * Anything else is logged and answers 500 ai.internal: it is passed on as its stack alone, so a provider
++ * error's upstream HTTP status never becomes AI's answer.
++ */
++function sendError(res, err, ctx, log = console) {
++    const req = { ov: ctx };
++    if (!(err instanceof AiError)) return svc.sendError(res, req, { stack: (err && err.stack) || String(err) }, log, PROBLEMS);
++    if (err.status === 429 && err.extra && err.extra.retry_after_seconds && !res.headersSent) res.setHeader('Retry-After', String(err.extra.retry_after_seconds));
++    return svc.sendError(res, req, err, log, PROBLEMS);
  }
  
  /** 'env:NAME' -> the value of process.env.NAME (read by name; never logged). */
@@@ -64,4 +72,4 @@ function parseJsonLoose(text) {
      } catch { return null; }
  }
  
--module.exports = { stableStringify, sha256, parseJson, iso, AiError, resolveSecret, secretRefValid, parseJsonLoose };
++module.exports = { stableStringify, sha256, parseJson, iso, AiError, sendError, resolveSecret, secretRefValid, parseJsonLoose };
ddiff --git a/test/provider-router.test.js b/test/provider-router.test.js
iindex 84a49f0..2018a9f 100644
---- a/test/provider-router.test.js
++++ b/test/provider-router.test.js
@@@ -9,6 +9,7 @@
  //   - explain{objective,reasons,selected,candidates} rides every run response and GET /runs/:id, with
  //     the per-attempt request log still there.
  const assert = require('assert');
++const contracts = require('openvibe-contracts');
  const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
  
  const t = suite('provider-router');
@@@ -94,6 +95,16 @@ t.test('two providers on one capability: the cheaper is chosen; a circuit-open o
      const byProvider = Object.fromEntries(exec.explain.candidates.map((c) => [c.provider, c]));
      assert.ok(byProvider.cheap.cost < byProvider.pricey.cost, 'rate cards carry the real prices');
  
++    // explain is AI's projection of the placement result; that result is a platform.placement-result@1.
++    const placed = await h.pool.candidates(r, ['classify'], 'classify');
++    let v = null;
++    try { v = contracts.validate('platform.placement-result@1', placed.placement); } catch (e) { if (!/unknown contract/.test(e.message)) throw e; }
++    if (v) assert.ok(v.valid, `platform.placement-result@1: ${JSON.stringify(v.errors)}`);
++    else console.log(`placement-result: skipped (openvibe-contracts ${require('openvibe-contracts/package.json').version} has no platform.placement-result@1; it lands in 0.77.0)`);
++    assert.strictEqual(placed.placement.selected, 'cheap:cheap-m');
++    assert.strictEqual(placed.explain.selected, placed.placement.selected, 'explain.selected is the placement\'s');
++    assert.deepStrictEqual(placed.explain.candidates.map((c) => (c.model ? `${c.provider}:${c.model}` : c.provider)), placed.placement.candidates.map((c) => c.id), 'one explain candidate per placement candidate');
++
      // A provider that keeps failing opens its circuit; placement then excludes it and the other answers.
      await provider({ key: 'flaky', base_url: flaky.url, capabilities: ['extract'], default_model: 'flaky-m' });
      await model('flaky', 'flaky-m', 1);
ddiff --git a/test/service-kit.test.js b/test/service-kit.test.js
nnew file mode 100644
iindex 0000000..96ddecc
---- /dev/null
++++ b/test/service-kit.test.js
@@@ -0,0 +1,67 @@
++'use strict';
++// openvibe-sdk/service in AI: sendError keeps AI's problem bodies (an AiError's status, code, detail, errors and
++// extra; Retry-After on a 429; anything else 500 ai.internal, never a provider's upstream status), and the
++// process stop drains the server, then closes the handles (exit 0, or 1 when handles.close() fails).
++const assert = require('assert');
++const http = require('http');
++const { AiError, sendError } = require('../server/util');
++const { ProviderError } = require('../server/providers/common');
++const { gracefulStop } = require('openvibe-sdk/service');
++const { suite } = require('./helpers');
++
++const t = suite('service-kit');
++const quiet = { log() {}, warn() {}, error() {} };
++const ctx = { requestId: 'req_1', traceId: 'trace_1' };
++
++function fakeRes() {
++    const headers = {};
++    return {
++        headers, statusCode: 0, body: null, headersSent: false,
++        setHeader(k, v) { headers[k.toLowerCase()] = v; },
++        getHeader(k) { return headers[k.toLowerCase()]; },
++        end(s) { this.body = JSON.parse(s); this.headersSent = true; },
++    };
++}
++
++t.test('an AiError answers its own status, code, detail, errors and extra; a 429 sets Retry-After', () => {
++    const res = fakeRes();
++    sendError(res, new AiError(429, 'quota.exceeded', 'quota exceeded for global *', { retry_after_seconds: 42, quota: { id: 1 }, errors: [{ path: '/x', message: 'bad' }] }), ctx, quiet);
++    assert.strictEqual(res.statusCode, 429);
++    assert.strictEqual(res.headers['retry-after'], '42');
++    assert.strictEqual(res.headers['content-type'], 'application/problem+json');
++    assert.deepStrictEqual(res.body, {
++        type: 'https://openvibe.network/problems/quota.exceeded', title: 'Too Many Requests', status: 429, code: 'quota.exceeded',
++        detail: 'quota exceeded for global *', errors: [{ path: '/x', message: 'bad' }], request_id: 'req_1', trace_id: 'trace_1',
++        error: 'quota exceeded for global *', retry_after_seconds: 42, quota: { id: 1 },
++    });
++});
++
++t.test('anything else is a logged 500 ai.internal, even an error that carries an HTTP status and a code', () => {
++    for (const err of [new Error('boom'), new ProviderError('upstream said no', { status: 401, code: 'provider.error' })]) {
++        const res = fakeRes();
++        const logged = [];
++        sendError(res, err, ctx, { ...quiet, error: (...a) => logged.push(a.join(' ')) });
++        assert.strictEqual(res.statusCode, 500);
++        assert.deepStrictEqual(res.body, {
++            type: 'https://openvibe.network/problems/ai.internal', title: 'Internal Server Error', status: 500, code: 'ai.internal',
++            detail: 'internal error', request_id: 'req_1', trace_id: 'trace_1', error: 'internal error',
++        });
++        assert.ok(logged.length === 1 && logged[0].includes(err.message), 'the stack is logged');
++    }
++});
++
++t.test('the stop drains the server, then closes the handles; a failing close exits 1', async () => {
++    for (const fails of [false, true]) {
++        const server = http.createServer((_req, res) => res.end('ok'));
++        await new Promise((r) => server.listen(0, '127.0.0.1', r));
++        const order = [];
++        const handles = { close: async () => { order.push(server.listening ? 'close while listening' : 'close'); if (fails) throw new Error('db'); } };
++        let exited = null;
++        const code = await gracefulStop({ name: 'ai', server, handles, drainMs: 8000, deadlineMs: 10000, signals: false, exit: (c) => { exited = c; }, log: quiet }).stop('SIGTERM');
++        assert.deepStrictEqual(order, ['close']);
++        assert.strictEqual(code, fails ? 1 : 0);
++        assert.strictEqual(exited, code);
++    }
++});
++
++t.run();