Codes: every attempt is capped, and the outbox stops retrying rejected events (#186)

ddiff --git a/migrations/0017_outbox_dead_letters.sql b/migrations/0017_outbox_dead_letters.sql
nnew file mode 100644
iindex 0000000..2eca30d
---- /dev/null
++++ b/migrations/0017_outbox_dead_letters.sql
@@@ -0,0 +1,6 @@
++-- phase: expand
++-- The outbox's dead letters: an event OpenVibe.Events refuses (its envelope can never pass) is marked dead after a
++-- fixed number of tries instead of being retried forever. dead_at says when it died, dead_reason why.
++ALTER TABLE codes_outbox ADD COLUMN IF NOT EXISTS dead_at timestamptz;
++ALTER TABLE codes_outbox ADD COLUMN IF NOT EXISTS dead_reason text COLLATE "C";   -- refused | failed
++CREATE INDEX IF NOT EXISTS codes_outbox_dead ON codes_outbox (dead_at) WHERE dead_at IS NOT NULL;
ddiff --git a/migrations/0018_attempt_run_mode.sql b/migrations/0018_attempt_run_mode.sql
nnew file mode 100644
iindex 0000000..50aed66
---- /dev/null
++++ b/migrations/0018_attempt_run_mode.sql
@@@ -0,0 +1,5 @@
++-- phase: expand
++-- How an attempt was paid for: 'ai' (it ran through OpenVibe.AI under a run token, metered by the task's budget),
++-- 'plan' (the person's own CLI subscription on their machine: no AI token, nothing metered), 'none' (no run token
++-- could be minted, so the harness ran as it did before).
++ALTER TABLE codes_attempts ADD COLUMN IF NOT EXISTS run_mode text COLLATE "C";
ddiff --git a/server/app.js b/server/app.js
iindex 1271e42..f0c5a7f 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -81,7 +81,7 @@ async function createApp(opts = {}) {
      const run = opts.run || createRunClient({ config, fetchImpl: fetchImpl || globalThis.fetch });
      // A person's own connections (Codes C3): their provider keys live in OpenVibe.AI, and their harnesses run with a
      // short-lived run token that names them.
--    const connections = createConnections({ config, fetchImpl: fetchImpl || globalThis.fetch, log });
++    const connections = opts.connections || createConnections({ config, fetchImpl: fetchImpl || globalThis.fetch, log });
      const runTokens = createRunTokens({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
      // The task's attempts (stage C2): every Run job is one attempt, and the engine decides what happens next.
      const attempts = createAttempts({ store, config, log, harnesses });
ddiff --git a/server/config.js b/server/config.js
iindex 16c940d..ee716b5 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -101,6 +101,10 @@ function load(env = process.env) {
              secrets: String(env.CODES_EVENTS_SECRET || '').split(',').map((s) => s.trim()).filter(Boolean),
              intervalMs: Math.max(500, int(env.CODES_EVENTS_INTERVAL_MS, 5000)),
              scope: env.CODES_EVENTS_SCOPE || 'events.event.publish',
++            // How many tries an event gets before it is marked dead: an event OpenVibe.Events refused (its envelope
++            // can never pass) fewer than one that keeps failing for another reason.
++            deadAfterRefusals: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_REFUSALS, 3)),
++            deadAfterAttempts: Math.max(1, int(env.CODES_EVENTS_DEAD_AFTER_ATTEMPTS, 20)),
          },
          // Session lines in PostgreSQL: no total cap; finished tasks keep theirs for this many days (default 90).
          sessions: {
ddiff --git a/server/domain/attempts.js b/server/domain/attempts.js
iindex 22a3164..b0bc597 100644
---- a/server/domain/attempts.js
++++ b/server/domain/attempts.js
@@@ -173,16 +173,17 @@ function createAttempts({ store, config = {}, log = console, harnesses = null })
      }
  
      /** The Run job of an attempt. The session is set for a conductor's attempt, which outlives each turn's job. */
--    async function bind(id, { job_id = null, state = null, started_at = null, machine = null, model = null, session = null } = {}) {
++    async function bind(id, { job_id = null, state = null, started_at = null, machine = null, model = null, session = null, run_mode = null } = {}) {
          return await db.one(
              `UPDATE codes_attempts SET job_id = COALESCE($2, job_id),
                  state = COALESCE($3, state),
                  machine = COALESCE($4, machine),
                  model = COALESCE($5, model),
                  started_at = COALESCE($6::timestamptz, started_at),
--                session = COALESCE($7, session)
++                session = COALESCE($7, session),
++                run_mode = COALESCE($8, run_mode)
               WHERE id = $1 RETURNING *`,
--            [id, job_id, state, machine, model, started_at, session],
++            [id, job_id, state, machine, model, started_at, session, run_mode],
          );
      }
  
ddiff --git a/server/domain/run-tasks.js b/server/domain/run-tasks.js
iindex 99a3ca6..b220c6f 100644
---- a/server/domain/run-tasks.js
++++ b/server/domain/run-tasks.js
@@@ -23,6 +23,11 @@
   * the engine decides what happens next — done, a continue in the same session, a switch to another agent, a wrap-up or
   * an unfinished task — and this module carries the decision out. One ending starts at most one follow-up: the
   * follow-up's follow_up_of is unique, so two racing ticks never carry one job on twice.
++ *
++ * Every attempt that calls models through OpenVibe.AI runs on a run token of its own, carrying what is left of the
++ * task's budget: the task's budget minus what the earlier attempts spent, as OpenVibe.AI counted it. When nothing
++ * is left of it, the follow-up does not start and the task ends at its budget. An attempt on the person's own plan
++ * (their CLI subscription on their machine) needs no token at all; the attempt's run_mode says which it was.
   */
  const { RunError } = require('./run-client');
  const { chain, budget } = require('../../engine');
@@@ -164,7 +169,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
      // The answer is cached for a minute, so a task that ends does not ask again for every attempt.
      const connectionCache = new Map();
      const CONNECTION_TTL_MS = 60_000;
--    let providerMap = null;
++    let harnessInfoCache = null;
      async function personConnections(subject) {
          if (!connections || !subject) return null;
          const hit = connectionCache.get(subject);
@@@ -182,22 +187,47 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
      }
  
      /**
--     * The environment the job's harness gets (Codes C3): a run token that names the person and says where to
--     * send its requests. A failure to mint one is not fatal: the job runs as it did before, without the token.
++     * The environment an attempt's harness gets (Codes C3): a run token that names the person, says where to send
++     * its requests and carries the budget that is left for this attempt. A failure to mint one is not fatal: the
++     * job runs as it did before, without the token.
++     *
++     * An attempt on the person's own plan (the CLI's own login on their machine) needs no token at all: the CLI
++     * signs in itself, and nothing of it is metered by the task's budget. The attempt says which of the two it was.
++     *
++     * → { env, mode } with mode 'ai' (through OpenVibe.AI), 'plan' (the person's own subscription) or 'none'
       */
--    async function harnessEnv({ subject, taskId, attemptId = null, agent = null, limits = null }) {
--        if (!runTokens) return null;
++    async function attemptEnv({ subject, taskId, attemptId = null, agent = null, budget = null }) {
++        if (!runTokens) return { env: null, mode: 'none' };
          try {
              const provider = providerFor(agent);
              const own = provider ? await hasOwnKey(subject, provider) : false;
--            const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own, budget: budgetFor(limits) });
--            return minted.env;
++            // The person's own key wins: it is the credential they chose, and the run is metered all the same.
++            if (!own && await planFor(subject, agent)) return { env: null, mode: 'plan' };
++            const minted = await runTokens.mint({ subject, task: taskId, attempt: attemptId, harness: agent, provider, ownKey: own, budget: budget || budgetFor(null) });
++            return { env: minted.env, mode: 'ai' };
          } catch (e) {
              log.warn(`[Codes] no run token for task ${taskId}: ${String(e && e.message).slice(0, 200)}`);
--            return null;
++            return { env: null, mode: 'none' };
          }
      }
  
++    /**
++     * The person's own plan for a harness: the CLI's own login on their machine (their Claude plan, their ChatGPT
++     * plan, …). It is not OpenVibe.AI, so an attempt on it spends nothing the task's budget can run out of. Only a
++     * CLI can sign in itself: an API harness always needs a token.
++     */
++    async function planFor(subject, agent) {
++        const info = harnessInfo(agent);
++        if (!info.provider || info.kind !== 'cli') return null;
++        const conns = await personConnections(subject);
++        if (!conns || !conns.plans || !conns.plans.length) return null;
++        return conns.plans.find((p) => {
++            if (!p || typeof p !== 'object') return false;
++            const names = [p.provider, p.harness, p.id].filter(Boolean).map((v) => String(v).toLowerCase());
++            return names.includes(info.provider.toLowerCase()) || names.includes(String(agent).toLowerCase());
++        }) || null;
++    }
++
      /**
       * The budget of one attempt: the task's limits (a benchmark's per-task limits) first, then the person's default
       * in Codes' settings, and nothing at all when neither says (an uncapped run is an uncapped run).
@@@ -227,6 +257,66 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
          return await runTokens.spend(row.jti);
      }
  
++    /**
++     * What the task's attempts spent through OpenVibe.AI: the sum of what AI counted for each of their run tokens
++     * (GET /api/v1/run-budgets/:jti). An attempt on the person's own plan has no token and spends nothing here.
++     */
++    async function taskSpend(taskId) {
++        const out = { usd: 0, tokens: 0, known: true };
++        if (!runTokens || !taskId) return out;
++        const tokens = await runTokens.forTask(taskId);
++        for (const t of tokens) {
++            const s = await runTokens.spend(t.jti);
++            if (s) {
++                out.usd += Number(s.spent_usd) || 0;
++                out.tokens += Number(s.spent_tokens) || 0;
++                continue;
++            }
++            // AI cannot say (its row goes with the token): what the harness reported is the next best thing.
++            out.known = false;
++            const a = t.attempt_id ? await attemptStore.get(t.attempt_id) : null;
++            out.usd += Number(a && a.cost_usd) || 0;
++            out.tokens += Number(a && a.tokens) || 0;
++        }
++        return out;
++    }
++
++    /**
++     * The budget of the task and what is left of it: the task's budget (the first attempt's) minus what the earlier
++     * attempts spent, as OpenVibe.AI counted it. A task with no budget at all has nothing to run out of.
++     */
++    async function remainingBudget(taskId) {
++        const attempts = await attemptStore.list(taskId);
++        const first = attempts[0] || null;
++        const cap = {
++            usd: first && first.budget_usd != null ? Number(first.budget_usd) : null,
++            tokens: first && first.budget_tokens != null ? Number(first.budget_tokens) : null,
++        };
++        if (cap.usd == null && cap.tokens == null) {
++            const fallback = budgetFor(null);   // a task written before the budgets: the person's default
++            cap.usd = fallback.usd;
++            cap.tokens = fallback.tokens;
++        }
++        if (cap.usd == null && cap.tokens == null) {
++            return { cap, spent: { usd: 0, tokens: 0 }, left: { usd: null, tokens: null }, capped: false, exhausted: false };
++        }
++        const spent = await taskSpend(taskId);
++        const left = {
++            usd: cap.usd == null ? null : Math.max(0, Math.round((cap.usd - spent.usd) * 10000) / 10000),
++            tokens: cap.tokens == null ? null : Math.max(0, cap.tokens - spent.tokens),
++        };
++        const exhausted = (cap.usd != null && left.usd <= 0) || (cap.tokens != null && left.tokens <= 0);
++        return { cap, spent, left, capped: true, exhausted };
++    }
++
++    /** What a task that ran out of budget says, in plain words. */
++    function budgetLeftNote(left) {
++        const parts = [];
++        if (left.cap.usd != null) parts.push(`$${(left.spent.usd || 0).toFixed(4)} of $${left.cap.usd.toFixed(2)}`);
++        if (left.cap.tokens != null) parts.push(`${left.spent.tokens || 0} of ${left.cap.tokens} tokens`);
++        return parts.length ? `stopped at its budget: ${parts.join(', ')} spent` : 'stopped at its budget';
++    }
++
      /** Whether the run was refused for its budget (OpenVibe.AI said 402 ai.run_budget_reached). */
      function budgetRefused(result, error) {
          return /ai\.run_budget_reached/.test(budgetText(result, error));
@@@ -247,12 +337,18 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
  
      /** The provider a harness runs on: the catalog's own word for it (anthropic, openai, deepseek, …). */
      function providerFor(agent) {
--        if (!agent) return null;
--        if (!providerMap) {
--            providerMap = {};
--            for (const offer of require('./harnesses').createHarnesses().list()) providerMap[offer.id] = offer.provider;
++        return harnessInfo(agent).provider;
++    }
++
++    /** What the catalog says about a harness: its provider, and whether it is a CLI (which can sign in itself). */
++    function harnessInfo(agent) {
++        if (!harnessInfoCache) {
++            harnessInfoCache = {};
++            for (const offer of require('./harnesses').createHarnesses().list()) {
++                harnessInfoCache[offer.id] = { provider: offer.provider || null, kind: (offer.address && offer.address.kind) || null };
++            }
          }
--        return providerMap[agent] || null;
++        return harnessInfoCache[agent] || { provider: null, kind: null };
      }
  
      /** Whether the person keeps their own key for this provider. */
@@@ -385,8 +481,9 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
          try {
              // The job's harness runs with a short-lived run token that names the person (Codes C3): the environment
              // travels with the job's input and the machine's worker hands it to the harness.
--            const env = await harnessEnv({ subject, taskId: id, attemptId: first.attempt.id, agent: args.agent || null, limits: args.limits });
--            job = await run.submit({ subject, project, args: env ? { ...args, env } : args, key: id });
++            const funding = await attemptEnv({ subject, taskId: id, attemptId: first.attempt.id, agent: args.agent || null, budget });
++            await attemptStore.bind(first.attempt.id, { run_mode: funding.mode });
++            job = await run.submit({ subject, project, args: funding.env ? { ...args, env: funding.env } : args, key: id });
          } catch (e) {
              const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
              await attemptStore.end(first.attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
@@@ -527,11 +624,21 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
      /**
       * The next attempt: a Run job for the task with the engine's plan in it. The attempt is written first and the job
       * second, so the fence (follow_up_of) is in place before anything is started.
++     *
++     * The attempt gets a run token of its own, carrying what is left of the task's budget: the task's budget minus
++     * what the earlier attempts spent, as OpenVibe.AI counted it. When nothing is left, the follow-up does not start
++     * and the task ends there, at its budget.
       */
      async function startFollowUp(row, plan, ending) {
++        const left = await remainingBudget(row.id);
++        if (left.exhausted) {
++            const stopped = await db.one(`UPDATE run_tasks SET outcome = 'unfinished', outcome_note = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
++                [row.id, budgetLeftNote(left), store.now()]);
++            return { attempt: null, created: false, row: stopped, error: 'the task has no budget left' };
++        }
          const { attempt, created } = await attemptStore.create({
              task: row, agent: plan.agent || null, model: plan.model || null, session: plan.session || null,
--            why: plan.why || null, kind: plan.kind || 'manual', followUpOf: ending.id, state: 'queued',
++            why: plan.why || null, kind: plan.kind || 'manual', followUpOf: ending.id, state: 'queued', budget: left.left,
          });
          // Someone else already started the follow-up of this ending: no second job.
          if (!created) return { attempt, created: false, row };
@@@ -546,7 +653,11 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
          }
          let job;
          try {
--            job = await run.submit({ subject: row.subject, project: row.project_id, args, key: attempt.id });
++            // The follow-up runs on a run token of its own, with the budget that is left: a follow-up that spends
++            // through AI is capped like any other attempt.
++            const funding = await attemptEnv({ subject: row.subject, taskId: row.id, attemptId: attempt.id, agent: plan.agent || null, budget: left.left });
++            await attemptStore.bind(attempt.id, { run_mode: funding.mode });
++            job = await run.submit({ subject: row.subject, project: row.project_id, args: funding.env ? { ...args, env: funding.env } : args, key: attempt.id });
          } catch (e) {
              const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
              await attemptStore.end(attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
@@@ -591,13 +702,25 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
                  why: text ? `carried on by the owner: ${text.slice(0, 200)}` : 'carried on by the owner',
              };
          let attempt = null;
++        const before = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);
++        if (!before) return { error: 'no such task' };
++        if (!TERMINAL.has(before.state)) return { error: 'the task is still running' };
++        if (before.outcome === 'stopped') return { error: 'the task was stopped on purpose' };
++        // The next attempt gets a run token of its own, with what is left of the task's budget. When nothing is left
++        // of it, the attempt does not start and the task ends at its budget.
++        const left = await remainingBudget(row.id);
++        if (left.exhausted) {
++            const stopped = await db.one(`UPDATE run_tasks SET outcome = 'unfinished', outcome_note = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
++                [row.id, budgetLeftNote(left), store.now()]);
++            return { error: 'the task has no budget left', row: stopped };
++        }
          const locked = await store.tx(async () => {
              const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
              if (!current) return { error: 'no such task' };
              if (!TERMINAL.has(current.state)) return { error: 'the task is still running' };
              if (current.outcome === 'stopped') return { error: 'the task was stopped on purpose' };
              const made = await attemptStore.create({
--                task: current, agent: plan.agent, model: plan.model, session: null, why: plan.why, kind: plan.kind, state: 'queued',
++                task: current, agent: plan.agent, model: plan.model, session: null, why: plan.why, kind: plan.kind, state: 'queued', budget: left.left,
              });
              attempt = made.attempt;
              await db.one(`UPDATE run_tasks SET state = 'queued', outcome = NULL, outcome_note = NULL, error = NULL, job_id = NULL, finished_at = NULL, updated_at = $2 WHERE id = $1 RETURNING *`,
@@@ -607,7 +730,10 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
          if (locked.error) return { error: locked.error };
          let job;
          try {
--            job = await run.submit({ subject: row.subject, project: row.project_id, args: followUpArgs(row, plan), key: attempt.id });
++            const funding = await attemptEnv({ subject: row.subject, taskId: row.id, attemptId: attempt.id, agent: plan.agent || null, budget: left.left });
++            await attemptStore.bind(attempt.id, { run_mode: funding.mode });
++            const args = followUpArgs(row, plan);
++            job = await run.submit({ subject: row.subject, project: row.project_id, args: funding.env ? { ...args, env: funding.env } : args, key: attempt.id });
          } catch (e) {
              const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
              await attemptStore.end(attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
iindex cf42b8b..2f6fcc0 100644
---- a/server/events/outbox.js
++++ b/server/events/outbox.js
@@@ -9,17 +9,26 @@
   *
   * The fixed evt_ id is chosen at insert, so a retry or a restart never makes a second event. The publisher claims due
   * rows with a short lease (two processes never send one row at once), sends each envelope once under that id, records
-- * a failure's attempt and backoff, and leaves it for the next pass. Under CODES_DRILL no loop starts.
++ * a failure's attempt and backoff, and leaves it for the next pass. An event OpenVibe.Events refuses (its envelope
++ * can never pass) is marked dead after a fixed number of tries instead of being retried forever. Under CODES_DRILL
++ * no loop starts.
   *
   *   { task_id, state, agent, repo, cost_usd, files_changed, started_at, finished_at, error }
   */
--const { ids, serviceAuth } = require('openvibe-contracts');
++const { ids, serviceAuth, validate } = require('openvibe-contracts');
  
  const TOPICS = Object.freeze([
      'codes.task.queued', 'codes.task.running', 'codes.task.succeeded', 'codes.task.failed', 'codes.task.cancelled',
  ]);
  const TOPIC_SET = new Set(TOPICS);
  const MAX_BACKOFF_MS = 5 * 60_000;
++// The envelope contract (events.event-envelope@1): the envelope Codes sends has to match it, or OpenVibe.Events
++// refuses the event with a 422 and the row is retried forever.
++const ENVELOPE_CONTRACT = 'events.event-envelope@1';
++// An event Events refused (4xx: its envelope can never pass) is dead after this many tries; an event that keeps
++// failing for another reason is dead after this many. A dead event is not sent again.
++const DEAD_AFTER_REFUSALS = 3;
++const DEAD_AFTER_ATTEMPTS = 20;
  
  /** queued | running | succeeded | failed | cancelled: the five states codes.task.* announces. */
  function groupOf(state) {
@@@ -114,7 +123,11 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
          scope: (config.events && config.events.scope) || 'events.event.publish',
          fetch: fetchImpl,
      }) : null;
--    const stats = { published: 0, failed: 0, last_error: null };
++    const stats = { published: 0, failed: 0, dead: 0, last_error: null };
++    // How many tries an event gets before it is marked dead: a refusal (its envelope can never pass) fewer than a
++    // failure of another kind.
++    const deadAfterRefusals = Math.max(1, Number((config.events && config.events.deadAfterRefusals) || DEAD_AFTER_REFUSALS));
++    const deadAfterAttempts = Math.max(1, Number((config.events && config.events.deadAfterAttempts) || DEAD_AFTER_ATTEMPTS));
      let timer = null;
      let running = false;
      let active = false;
@@@ -130,7 +143,7 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
              `UPDATE codes_outbox o SET claimed_at = $1::timestamptz, claimed_by = $2
               WHERE o.event_id IN (
                   SELECT o2.event_id FROM codes_outbox o2
--                 WHERE o2.published_at IS NULL AND o2.next_attempt_at <= $1::timestamptz
++                 WHERE o2.published_at IS NULL AND o2.dead_at IS NULL AND o2.next_attempt_at <= $1::timestamptz
                     AND (o2.claimed_at IS NULL OR o2.claimed_at < $3::timestamptz)
                   ORDER BY o2.created_at ASC, o2.event_id ASC
                   LIMIT $4
@@@ -141,6 +154,10 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
          );
      }
  
++    /**
++     * The envelope OpenVibe.Events is given (events.event-envelope@1): the priority is one of the contract's own
++     * three words (critical | important | low), never a word of Codes' own.
++     */
      function envelope(row) {
          return {
              event_id: row.event_id,
@@@ -150,16 +167,35 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
              actor: { type: 'user', id: row.subject },
              subject: { type: 'user', id: row.subject },
              timestamp: store.iso(),
--            priority: 'normal',
++            priority: 'important',
              visibility: 'internal',
              payload: row.payload,
          };
      }
  
++    /**
++     * What is wrong with an envelope, per the contract itself: null when it matches. It is a warning, not a refusal:
++     * OpenVibe.Events is the authority on what it accepts, and an envelope that does not match here is sent anyway
++     * (and refused there, which is what marks the row dead).
++     */
++    function envelopeIssue(doc) {
++        try {
++            const out = validate(ENVELOPE_CONTRACT, doc);
++            if (!out || out.valid) return null;
++            const errors = Array.isArray(out.errors) ? out.errors : [];
++            return errors.map((e) => `${e.path || ''} ${e.message || ''}`.trim()).join('; ') || 'the envelope does not match the contract';
++        } catch (e) {
++            return null;   // the contract is not in the pinned version: nothing to say about it
++        }
++    }
++
      async function send(row) {
          const headers = { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) };
++        const doc = envelope(row);
++        const issue = envelopeIssue(doc);
++        if (issue) log.warn(`[Codes] outbox ${row.event_id} does not match ${ENVELOPE_CONTRACT}: ${issue}`);
          const res = await fetchImpl(`${eventsUrl}/api/v1/events`, {
--            method: 'POST', headers, body: JSON.stringify(envelope(row)), signal: AbortSignal.timeout(15_000),
++            method: 'POST', headers, body: JSON.stringify(doc), signal: AbortSignal.timeout(15_000),
          });
          if (!res.ok) {
              const text = await res.text().catch(() => '');
@@@ -190,6 +226,21 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
                      const message = String((err && err.message) || err).slice(0, 500);
                      stats.last_error = message;
                      const attempts = Number(row.attempts) + 1;
++                    // A refusal (4xx) is the envelope being turned down: it can never pass, so it dies after a fixed
++                    // number of tries. Another failure is worth more tries, and dies too in the end.
++                    const status = Number(err && err.status) || 0;
++                    const refused = status >= 400 && status < 500;
++                    const deadAfter = refused ? deadAfterRefusals : deadAfterAttempts;
++                    if (attempts >= deadAfter) {
++                        await db.exec(
++                            `UPDATE codes_outbox SET attempts = attempts + 1, last_error = $1, dead_at = $2::timestamptz, dead_reason = $3,
++                               claimed_at = NULL, claimed_by = NULL WHERE event_id = $4`,
++                            [message, store.iso(), refused ? 'refused' : 'failed', row.event_id],
++                        );
++                        stats.dead++;
++                        log.warn(`[Codes] outbox ${row.event_id} is dead after ${attempts} tries: ${message}`);
++                        continue;
++                    }
                      const next = new Date(store.now() + backoffMs(attempts)).toISOString();
                      await db.exec(
                          `UPDATE codes_outbox SET attempts = attempts + 1, last_error = $1, claimed_at = NULL, claimed_by = NULL,
@@@ -217,10 +268,16 @@ function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = co
          configured, runOnce,
          start() { if (!configured || config.drill || timer) return false; active = true; schedule(0); return true; },
          stop() { active = false; if (timer) clearTimeout(timer); timer = null; return Promise.resolve(); },
--        pending: async () => Number(await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE published_at IS NULL')) || 0,
--        status: async () => ({ enabled: configured, pending: await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE published_at IS NULL'), last_error: stats.last_error }),
++        pending: async () => Number(await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE published_at IS NULL AND dead_at IS NULL')) || 0,
++        dead: async () => Number(await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE dead_at IS NOT NULL')) || 0,
++        status: async () => ({
++            enabled: configured,
++            pending: await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE published_at IS NULL AND dead_at IS NULL'),
++            dead: await db.value('SELECT COUNT(*)::int AS n FROM codes_outbox WHERE dead_at IS NOT NULL'),
++            last_error: stats.last_error,
++        }),
          stats: () => ({ ...stats }),
      };
  }
  
--module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS };
++module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS, DEAD_AFTER_REFUSALS, DEAD_AFTER_ATTEMPTS };
ddiff --git a/test/followup-budget.test.js b/test/followup-budget.test.js
nnew file mode 100644
iindex 0000000..feab8b8
---- /dev/null
++++ b/test/followup-budget.test.js
@@@ -0,0 +1,182 @@
++'use strict';
++
++/**
++ * The budget of every attempt (Codes C3): the first try and every follow-up after it (a continue, a switch, a
++ * wrap-up) gets a run token of its own, carrying what is left of the task's budget — the task's budget minus what
++ * the earlier attempts spent, as OpenVibe.AI counted it. When nothing is left of it, the follow-up does not start
++ * and the task ends at its budget. An attempt on the person's own plan (their CLI subscription on their machine)
++ * needs no token at all, and the attempt says so.
++ */
++const assert = require('assert');
++const http = require('http');
++const { boot, check, done } = require('./helpers/boot');
++
++(async () => {
++    // A stand-in for OpenVibe.AI: it keeps the budgets Codes registers and what each run spent.
++    const budgets = new Map();
++    const registered = [];
++    const ai = http.createServer((req, res) => {
++        const chunks = [];
++        req.on('data', (c) => chunks.push(c));
++        req.on('end', () => {
++            const send = (status, doc) => { res.writeHead(status, { 'Content-Type': 'application/json' }); res.end(JSON.stringify(doc)); };
++            if (req.url.startsWith('/api/v1/credentials')) return send(404, { detail: 'no key for this provider' });
++            const jti = decodeURIComponent(req.url.split('/').pop());
++            if (req.method === 'PUT') {
++                const body = JSON.parse(Buffer.concat(chunks).toString('utf8') || '{}');
++                budgets.set(jti, {
++                    budget_usd: body.budget_usd == null ? null : Number(body.budget_usd),
++                    budget_tokens: body.budget_tokens == null ? null : Number(body.budget_tokens),
++                    spent_usd: 0, spent_tokens: 0,
++                });
++                registered.push({ jti, body });
++                return send(200, { jti, ...body });
++            }
++            const row = budgets.get(jti) || { budget_usd: null, budget_tokens: null, spent_usd: 0, spent_tokens: 0 };
++            return send(200, { jti, ...row, reached: row.budget_tokens != null && row.spent_tokens >= row.budget_tokens });
++        });
++    });
++    await new Promise((r) => ai.listen(0, '127.0.0.1', r));
++    const aiUrl = `http://127.0.0.1:${ai.address().port}`;
++
++    const submits = [];
++    const jobs = new Map();
++    let counter = 0;
++    const run = {
++        enabled: true,
++        async submit(o) {
++            counter++;
++            const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${String(counter).padStart(2, '0')}`;
++            jobs.set(id, { id, state: 'queued', args: o.args });
++            submits.push(o);
++            return { id, state: 'queued' };
++        },
++        async read(jobId) { return jobs.get(jobId) || { id: jobId, state: 'queued' }; },
++        async cancel(jobId) { const j = jobs.get(jobId); if (j) j.state = 'cancelled'; return j; },
++        async streamUrl() { throw new Error('not used'); },
++    };
++
++    // The person's connections: no key of their own, and a Claude plan on their machine.
++    const connections = {
++        async list() { return { keys: [], plans: [{ provider: 'anthropic', machine: 'ada-mac', label: 'Claude Max' }] }; },
++    };
++
++    const t = await boot({
++        run, connections,
++        env: { CODES_AI_URL: aiUrl, CODES_AI_INTERNAL_URL: aiUrl, CODES_TASK_TOKENS: '1000', CODES_TASK_USD: '2' },
++    });
++    const db = t.ctx.store.db;
++    const ada = t.network.addUser('ada');
++    const actor = t.network.mintService('svc:actor', 'openvibe.codes', ['codes.task.start', 'codes.task.read', 'codes.task.cancel']);
++    const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++    // A task the person ran in the browser: a service's task runs in the project of their latest one there.
++    await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)
++        VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, $2, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2Z200', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject, project]);
++
++    const call = async (method, path, body) => {
++        const headers = { Accept: 'application/json', Authorization: `Bearer ${actor}`, 'X-OV-Subject': ada.subject };
++        if (body !== undefined) headers['Content-Type'] = 'application/json';
++        const res = await fetch(`${t.base}/api/v1${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body) });
++        return { status: res.status, json: await res.json().catch(() => null) };
++    };
++
++    const startTask = async (agent) => {
++        const r = await call('POST', '/tasks', {
++            repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a line to the README.', agent,
++            idempotency_key: `act_budget_${Math.random().toString(36).slice(2, 10)}`,
++        });
++        assert.strictEqual(r.status, 201, JSON.stringify(r.json));
++        return r.json.task.id;
++    };
++
++    const attemptsOf = async (taskId) => await db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);
++    const rowOf = async (taskId) => await db.one('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
++
++    /** The attempt ends in a crash, so the engine carries the task on in the same conversation. */
++    const endAttempt = async (taskId, error = 'the machine went away') => {
++        const row = await rowOf(taskId);
++        const result = {
++            ok: false, summary: 'I will continue with the rest of it', files: [],
++            log: [{ t: 'agent', agent: 'deepseek', session: 'sess-42' }, { t: 'done', ok: false, text: 'I will continue with the rest of it' }],
++        };
++        const job = { id: row.job_id, state: 'failed', result, error: { code: 'run.job.worker_failed', detail: error } };
++        const updated = await t.ctx.tasks.applyEvent(row, { state: 'failed', job, error });
++        await t.ctx.tasks.afterEnd(updated.id);
++        return updated;
++    };
++
++    try {
++        await check('a follow-up gets a run token of its own, with what is left of the task\'s budget', async () => {
++            const id = await startTask('deepseek');
++            assert.strictEqual(registered.length, 1, 'the first attempt minted a token');
++            assert.strictEqual(registered[0].body.budget_tokens, 1000);
++            assert.strictEqual(registered[0].body.budget_usd, 2);
++            // What the first attempt spent, as OpenVibe.AI counted it.
++            budgets.get(registered[0].jti).spent_usd = 0.42;
++            budgets.get(registered[0].jti).spent_tokens = 400;
++
++            await endAttempt(id);
++            const attempts = await attemptsOf(id);
++            assert.strictEqual(attempts.length, 2, 'the task was carried on');
++            assert.strictEqual(attempts[1].kind, 'continue');
++            assert.strictEqual(Number(attempts[1].budget_tokens), 600, 'the follow-up carries what is left of the tokens');
++            assert.strictEqual(Number(attempts[1].budget_usd), 1.58, 'the follow-up carries what is left of the dollars');
++            assert.strictEqual(attempts[1].run_mode, 'ai', 'the attempt ran through OpenVibe.AI');
++            assert.strictEqual(registered.length, 2, 'the follow-up minted a token of its own');
++            assert.strictEqual(registered[1].body.budget_tokens, 600);
++            assert.strictEqual(registered[1].body.budget_usd, 1.58);
++            const job = submits[submits.length - 1];
++            assert.ok(job.args.env, 'the follow-up runs with a token');
++            assert.ok(job.args.env.DEEPSEEK_API_KEY, 'the token is in the environment');
++        });
++
++        await check('a follow-up after the budget is spent does not start, and the task ends at its budget', async () => {
++            const before = submits.length;
++            const id = await startTask('deepseek');
++            const token = registered[registered.length - 1];
++            budgets.get(token.jti).spent_usd = 2;
++            budgets.get(token.jti).spent_tokens = 1000;
++
++            await endAttempt(id);
++            const attempts = await attemptsOf(id);
++            assert.strictEqual(attempts.length, 1, 'no follow-up: there is no budget left');
++            assert.strictEqual(submits.length, before + 1, 'only the first attempt ran');
++            const task = await rowOf(id);
++            assert.strictEqual(task.outcome, 'unfinished');
++            assert.ok(/stopped at its budget/.test(task.outcome_note), `the outcome says it: ${task.outcome_note}`);
++            assert.ok(/1000 of 1000 tokens/.test(task.outcome_note), `the outcome says what it spent: ${task.outcome_note}`);
++        });
++
++        await check('a switch after the budget is spent does not start', async () => {
++            const id = await startTask('deepseek');
++            const token = registered[registered.length - 1];
++            budgets.get(token.jti).spent_usd = 2;
++            budgets.get(token.jti).spent_tokens = 1000;
++            await endAttempt(id);
++            const before = submits.length;
++
++            const r = await call('POST', `/tasks/${id}/continue`, { agent: 'codex' });
++            assert.strictEqual(r.status, 409, JSON.stringify(r.json));
++            assert.ok(/budget/.test(String(r.json && r.json.detail || r.json && r.json.title || '')), JSON.stringify(r.json));
++            assert.strictEqual((await attemptsOf(id)).length, 1, 'the switch started no attempt');
++            assert.strictEqual(submits.length, before, 'the switch started no job');
++            const task = await rowOf(id);
++            assert.strictEqual(task.outcome, 'unfinished');
++            assert.ok(/stopped at its budget/.test(task.outcome_note), `the outcome says it: ${task.outcome_note}`);
++        });
++
++        await check('an attempt on the person\'s own plan needs no token, and the attempt says so', async () => {
++            const before = registered.length;
++            const id = await startTask('claude-code');
++            const attempts = await attemptsOf(id);
++            assert.strictEqual(attempts[0].run_mode, 'plan', 'the attempt ran on their own subscription');
++            assert.strictEqual(registered.length, before, 'no run token was minted');
++            const job = submits[submits.length - 1];
++            assert.ok(!job.args.env, 'the job carries no token');
++        });
++    } finally {
++        await t.close();
++        await new Promise((r) => ai.close(r));
++    }
++    done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/helpers/boot.js b/test/helpers/boot.js
iindex 1b4bd69..45e1ed9 100644
---- a/test/helpers/boot.js
++++ b/test/helpers/boot.js
@@@ -37,7 +37,7 @@ async function boot(opts = {}) {
      const { createStore } = require('../../server/db');
      // One database per boot (PGlite, or CODES_TEST_STORE=pg: the containers), dropped when the boot closes.
      const testdb = await require('./db').testDb();
--    const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.benchAttemptEnded ? { benchAttemptEnded: opts.benchAttemptEnded } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}), ...(opts.notify ? { notify: opts.notify } : {}) });
++    const built = await createApp({ config, store: createStore(testdb.db), log, limitsNow: opts.limitsNow, actorLimits: opts.actorLimits === true, ...(opts.run ? { run: opts.run } : {}), ...(opts.benchAttemptEnded ? { benchAttemptEnded: opts.benchAttemptEnded } : {}), ...(opts.accountSend ? { accountSend: opts.accountSend } : {}), ...(opts.connections ? { connections: opts.connections } : {}), ...(opts.notify ? { notify: opts.notify } : {}) });
      await built.ctx.keys.ensure();
      const server = await new Promise((resolve) => { const s = http.createServer(built.app); s.listen(0, '127.0.0.1', () => resolve(s)); });
      const base = `http://127.0.0.1:${server.address().port}`;
ddiff --git a/test/outbox.test.js b/test/outbox.test.js
iindex 621484f..c2b4604 100644
---- a/test/outbox.test.js
++++ b/test/outbox.test.js
@@@ -6,6 +6,7 @@
   */
  const assert = require('assert');
  const http = require('http');
++const { validate } = require('openvibe-contracts');
  const { parseArgs } = require('../harness/job');
  const { boot, check, done } = require('./helpers/boot');
  const { createPublisher } = require('../server/events/outbox');
@@@ -13,11 +14,18 @@ const { createPublisher } = require('../server/events/outbox');
  (async () => {
      const received = [];
      let failNext = 0;
++    let refuseNext = 0;
      const events = http.createServer((req, res) => {
          let body = '';
          req.on('data', (c) => { body += c; });
          req.on('end', () => {
              if (failNext > 0) { failNext--; res.writeHead(503, { 'Content-Type': 'application/json' }); return res.end('{"code":"events.down"}'); }
++            // Events refuses an envelope it does not accept (a 422 that no retry can fix).
++            if (refuseNext > 0) {
++                refuseNext--;
++                res.writeHead(422, { 'Content-Type': 'application/json' });
++                return res.end('{"code":"events.invalid_envelope","detail":"/priority must be equal to one of the allowed values"}');
++            }
              const event = JSON.parse(body);
              received.push(event);
              res.writeHead(201, { 'Content-Type': 'application/json' });
@@@ -97,6 +105,14 @@ const { createPublisher } = require('../server/events/outbox');
              assert.strictEqual(await t.ctx.publisher.pending(), 0);
          });
  
++        await check('the envelope is the one the contract describes, so OpenVibe.Events accepts it', async () => {
++            assert.ok(received.length >= 2, 'events were published');
++            for (const event of received) {
++                const v = validate('events.event-envelope@1', event);
++                assert.ok(v.valid, `the envelope matches events.event-envelope@1: ${JSON.stringify(v.errors)}`);
++            }
++        });
++
          await check('a failure records attempts/last_error, backs off, then retries the same id', async () => {
              const second = await t.ctx.tasks.start({ subject: ada.subject, project, args });
              const row = await db.one('SELECT * FROM codes_outbox WHERE payload->>\'task_id\' = $1', [second.id]);
@@@ -118,6 +134,29 @@ const { createPublisher } = require('../server/events/outbox');
              assert.strictEqual(received.filter((x) => x.event_id === row.event_id).length, 1, 'one event under the fixed id');
          });
  
++        await check('an event OpenVibe.Events refuses is dead after a fixed number of tries, not retried forever', async () => {
++            const third = await t.ctx.tasks.start({ subject: ada.subject, project, args });
++            const row = await db.one('SELECT * FROM codes_outbox WHERE payload->>\'task_id\' = $1', [third.id]);
++            refuseNext = 3;
++            for (let tries = 0; tries < 3; tries++) {
++                await db.exec("UPDATE codes_outbox SET next_attempt_at = now() - interval '1 second' WHERE event_id = $1", [row.event_id]);
++                const r = await t.ctx.publisher.runOnce();
++                assert.strictEqual(r.claimed, 1, 'the row is claimed');
++                assert.strictEqual(r.failed, 1, 'the refusal is a failure');
++            }
++            const dead = await db.one('SELECT * FROM codes_outbox WHERE event_id = $1', [row.event_id]);
++            assert.ok(dead.dead_at, 'the event is dead');
++            assert.strictEqual(dead.dead_reason, 'refused');
++            assert.strictEqual(dead.published_at, null);
++            assert.strictEqual(dead.attempts, 3, 'it got the tries it was given');
++            assert.match(dead.last_error, /events 422/);
++            assert.strictEqual(await t.ctx.publisher.pending(), 0, 'a dead event is not pending');
++            assert.strictEqual(await t.ctx.publisher.dead(), 1);
++            await db.exec("UPDATE codes_outbox SET next_attempt_at = now() - interval '1 second' WHERE event_id = $1", [row.event_id]);
++            assert.strictEqual((await t.ctx.publisher.runOnce()).claimed, 0, 'a dead event is never sent again');
++            refuseNext = 0;
++        });
++
          await check('a restart over the same rows does not duplicate', async () => {
              const before = received.length;
              const restarted = createPublisher({ config: t.config, store: t.ctx.store, fetchImpl: globalThis.fetch, log: { warn() {}, log() {}, error() {} } });