openvibers/openvibe.codes Commits

Coding sessions kept in PostgreSQL as they stream, run.job.* events finalize tasks without polling, codes.task.* published from an outbox; contracts v0.232.0 (#143)

OP OpenVibers committed 16 days ago · 1c7748d323 · 26 files, +1741 −91

- codes_session_lines (bigint seq: the stream's order), back-filled from the final log, replayed at
  /run/tasks/:id/session.json; the export carries them and a deletion erases them.
- POST /internal/events: signed v2 deliveries, the inbox claimed with the task update.
- codes_outbox: fixed evt_ ids, leased claims, backoff; off under CODES_DRILL.

Changes

ddiff --git a/README.md b/README.md
iindex 0cb96c7..4691e3b 100644
---- a/README.md
++++ b/README.md
@@@ -20,7 +20,7 @@ One place to send a coding task to the agent that fits it — by what the task n
| The harness | `openvibe-codes` ([bin/openvibe-codes.js](bin/openvibe-codes.js), [harness/](harness/)) | Runs on a person's own machine with their CLIs and keys; nothing goes through OpenVibe. See **Use the harness** below. |
| Home | `/` | What the harness is, the agents in the catalog, how a task finds its agent, Improve OpenVibe, and an honest table of what works today and what comes next. |
| Get started | `/start` | Install, the agents and what each needs, the commands, permission levels, hand-offs, the event format and using it from code. |
--| Run on your machine | `/run`, `/run/tasks/:id` ([server/http/run.js](server/http/run.js)) | A signed-in person starts a coding task in any browser and their own machine runs it: an OpenVibe Run job (class `code`, artifact `[email protected]`, `trust: user-owned`) that Codes starts for them in their project (project delegation: `X-OV-Subject`, `X-OV-Project`; [server/domain/run-client.js](server/domain/run-client.js)). Run places it only on a machine of theirs (added on openvibe.bot/machines), whose OpenVibe Node runs `openvibe-codes job` ([harness/job.js](harness/job.js)) in its job sandbox: clone, run the agents with hand-offs, hand back the summary, the changed files and the patch, and, when asked and the machine holds `GITHUB_TOKEN`, push a new branch (never the one it started from) and open a pull request. The page streams the job's lines (Run's job stream through Codes) and keeps the result (`run_tasks`, migration 0003). Each task is its owner's; account export and deletion cover them (ADR-033, [server/account-data.js](server/account-data.js)). The task page shows the agents' **session** with openvibe-shared/agent-session.js: steps grouped, each step's command, change and output (live while it runs), helpers nested, hand-offs and the plan; after the end, the session the result kept (job lines 1.1, `log`). |
++| Run on your machine | `/run`, `/run/tasks/:id` ([server/http/run.js](server/http/run.js)) | A signed-in person starts a coding task in any browser and their own machine runs it: an OpenVibe Run job (class `code`, artifact `[email protected]`, `trust: user-owned`) that Codes starts for them in their project (project delegation: `X-OV-Subject`, `X-OV-Project`; [server/domain/run-client.js](server/domain/run-client.js)). Run places it only on a machine of theirs (added on openvibe.bot/machines), whose OpenVibe Node runs `openvibe-codes job` ([harness/job.js](harness/job.js)) in its job sandbox: clone, run the agents with hand-offs, hand back the summary, the changed files and the patch, and, when asked and the machine holds `GITHUB_TOKEN`, push a new branch (never the one it started from) and open a pull request. The page streams the job's lines (Run's job stream through Codes) and keeps the result (`run_tasks`, migration 0003). Each task is its owner's; account export and deletion cover them (ADR-033, [server/account-data.js](server/account-data.js)). The task page shows the agents' **session** with openvibe-shared/agent-session.js: steps grouped, each step's command, change and output (live while it runs), helpers nested, hand-offs and the plan; after the end, the session from `codes_session_lines` (migration 0007): the lines are kept as the stream arrives, back-filled from the machine's final log, and replayed at `GET /run/tasks/:id/session.json?after=` (the owner, or a service with `codes.task.read` and `X-OV-Subject`); tasks from before keep the old `log`. Run's `run.job.*` events finalize a task at `POST /internal/events` ([server/events-consumer.js](server/events-consumer.js), no polling but a 10-minute fallback), and every state change is published as `codes.task.queued|running|succeeded|failed|cancelled` from an outbox ([server/events/outbox.js](server/events/outbox.js), contracts 0.232.0). |
| The agents | `/harnesses` | Every harness in [`server/data/harness-offers.json`](server/data/harness-offers.json) (validated at boot as `platform.harness-offer@1`, each model as `platform.agent-offer@1`) with its provider, address kind, what it can do, price and limits, and its models with context limits and token prices. No local path is ever shown. |
| Task API (services) | `POST /api/v1/tasks`, `GET /api/v1/tasks/:id`, `POST /api/v1/tasks/:id/cancel` ([server/http/tasks-api.js](server/http/tasks-api.js)) | Contracts 0.162 `codes.task.start` / `read` / `cancel`. A first-party service (OpenVibe.Actor) holding the capability starts a coding task for a person named in `X-OV-Subject`, with a Network service token for audience `openvibe.codes` (`CODES_AUDIENCE`). The task is the same one `/run` starts ([server/domain/run-tasks.js](server/domain/run-tasks.js)): it runs on that person's own machine, in the project of their latest `/run` task (none yet → 409 `codes.no_project`). It shows on their `/run` page. `idempotency_key` answers a retry with the first task (200, `created: false`). A service reads and cancels only the tasks it started for that person (otherwise 404). The result the machine returned is passed on only as `codes.task@1` allows, each field checked and cut to its limit (migration 0005: `via`, `idempotency_key`). Calls count against the person's budget: starts and cancels 6 a minute, 60 an hour. |
| Agent API | `GET /api/v1/harnesses`, `POST /api/v1/harnesses/route` | Public, no key. The router takes `{ task, requirements? }` (`task` is `edit`, `review`, `long` or `resume`; `requirements.capabilities` adds hard constraints) and answers `{ selected, reasons, candidates }` from `openvibe-sdk/placement` — hard constraints first (a missing capability excludes an agent, and says so), then cost. RFC 9457 problems (`harness.invalid`, `harness.task_unknown`). |
ddiff --git a/migrations/0007_session_lines.sql b/migrations/0007_session_lines.sql
nnew file mode 100644
iindex 0000000..3f2966e
---- /dev/null
++++ b/migrations/0007_session_lines.sql
@@@ -0,0 +1,17 @@
++-- phase: expand
++-- OpenVibe.Codes: a coding task's session, line by line, so a finished task replays from any device instead of only
++-- from the 240 KB log the machine returned in run_tasks.result. A line is the job line the session view understands
++-- (openvibe-shared/agent-session.js: agent, text, tool, tool_progress, tool_result, handoff, note, usage, done, step)
++-- or the stream's own state/end line. seq is the stream's order, so a reconnect that replays an event is a no-op on
++-- (task_id, seq). Bodies are clipped to 64 KB each; there is no total cap (session retention removes old lines).
++CREATE TABLE IF NOT EXISTS codes_session_lines (
++ task_id text COLLATE "C" NOT NULL, -- tsk_<ULID>
++ seq bigint NOT NULL, -- the stream's order (event seq × 250 000 + line), never renumbered
++ at timestamptz NOT NULL,
++ kind text COLLATE "C" NOT NULL, -- the line type the session view reads
++ body jsonb NOT NULL,
++ PRIMARY KEY (task_id, seq)
++);
++
++-- Set once the machine's final result has been reconciled with the stream and any missing lines were back-filled.
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS session_complete_at timestamptz;
ddiff --git a/migrations/0008_task_events.sql b/migrations/0008_task_events.sql
nnew file mode 100644
iindex 0000000..583b611
---- /dev/null
++++ b/migrations/0008_task_events.sql
@@@ -0,0 +1,34 @@
++-- phase: expand
++-- OpenVibe.Codes: task events in and out (plan 2026-10-28 §6, decisions 2 and 3). run.job.* deliveries are claimed in
++-- codes_events_inbox in the same transaction that finalizes the task, so a replay changes nothing. Every state change
++-- writes its codes.task.* event into codes_outbox in the same transaction; the publisher sends each fixed evt_ id
++-- once, backs off on failure and survives a restart. started_at/finished_at are the Run timings carried in the event
++-- payload (run_tasks.updated_at stays the local write clock).
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS started_at timestamptz;
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS finished_at timestamptz;
++CREATE INDEX IF NOT EXISTS run_tasks_job_id ON run_tasks (job_id) WHERE job_id IS NOT NULL;
++
++-- Inbound OpenVibe.Events deliveries already applied: a redelivery is a no-op.
++CREATE TABLE IF NOT EXISTS codes_events_inbox (
++ event_id text COLLATE "C" PRIMARY KEY, -- evt_<ULID>
++ event_type text COLLATE "C" NOT NULL,
++ task_id text COLLATE "C", -- tsk_… when the event named a Codes task
++ outcome text COLLATE "C",
++ processed_at timestamptz NOT NULL DEFAULT now()
++);
++
++-- The fixed Events evt_ id is chosen when the row is written, so a retry (or a restart) never makes a second event.
++CREATE TABLE IF NOT EXISTS codes_outbox (
++ event_id text COLLATE "C" PRIMARY KEY, -- evt_<ULID>
++ topic text COLLATE "C" NOT NULL, -- codes.task.queued|running|succeeded|failed|cancelled
++ subject text COLLATE "C" NOT NULL, -- usr_… the task's owner
++ payload jsonb NOT NULL,
++ created_at timestamptz NOT NULL DEFAULT now(),
++ published_at timestamptz,
++ attempts integer NOT NULL DEFAULT 0,
++ last_error text,
++ next_attempt_at timestamptz NOT NULL DEFAULT now(),
++ claimed_at timestamptz,
++ claimed_by text
++);
++CREATE INDEX IF NOT EXISTS codes_outbox_pending ON codes_outbox (next_attempt_at) WHERE published_at IS NULL;
ddiff --git a/package-lock.json b/package-lock.json
iindex 45adb45..da9d4ee 100644
---- a/package-lock.json
++++ b/package-lock.json
@@@ -15,7 +15,7 @@
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.232.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.31.0",
"pg": "^8.23.0"
@@@ -874,9 +874,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.229.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
-- "integrity": "sha512-BYazi/dna8bgdM6+Go8t+D7lTyI9UaWYnJIx0hYEdD0v2ps87iTPONkjnCBadk5RUrBDtvcNoSONS3R+bwFG0w==",
++ "version": "0.232.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.232.0",
++ "integrity": "sha512-6HI5nn//nAUEDdG4xRTAlcb40oc/6yO1nx7R5exGZ6Bo3mL5fJ5szw9IONddoLN6qJsO1CXzB5TzTqkBAh3OzQ==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex c7049b3..60159a5 100644
---- a/package.json
++++ b/package.json
@@@ -30,7 +30,7 @@
"express-rate-limit": "^7.5.0",
"helmet": "^8.1.0",
"iovalkey": "^0.4.0",
-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.229.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.232.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.67.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.31.0",
"pg": "^8.23.0"
ddiff --git a/server/account-data.js b/server/account-data.js
iindex d88fbb5..12297f9 100644
---- a/server/account-data.js
++++ b/server/account-data.js
@@@ -1,21 +1,47 @@
'use strict';
/**
* Account export and deletion → Codes (ADR-033; openvibe-sdk/account-data). What Codes holds about a person is the
-- * coding tasks they started on their own machine (run_tasks, plan T16): what they asked, the Run job and its result.
-- * The export carries them whole; a deletion erases them. The Run jobs themselves are OpenVibe.Run's records (usage
-- * billed to the project) and are Run's to answer for.
++ * coding tasks they started on their own machine (run_tasks, plan T16) and the session lines of those tasks
++ * (codes_session_lines): what they asked, the Run job, its result and its session. The export carries the tasks whole
++ * and their sessions as session-lines.json; a deletion erases the session lines and then the tasks, in one
++ * transaction. The Run jobs themselves are OpenVibe.Run's records (usage billed to the project) and are Run's to
++ * answer for.
*/
const { createAccountData, TOPICS } = require('openvibe-sdk/account-data');
++const SESSION_LINES_MAX = 20000;
++
const TABLES = [
{
table: 'run_tasks', subject: 'subject', file: 'run-tasks.json', order: 'created_at',
-- columns: ['id', 'project_id', 'job_id', 'repo', 'ref', 'prompt', 'agent', 'push', 'pull_request', 'state', 'result', 'error', 'created_at', 'updated_at'],
++ columns: ['id', 'project_id', 'job_id', 'repo', 'ref', 'prompt', 'agent', 'push', 'pull_request', 'state', 'result', 'error', 'started_at', 'finished_at', 'created_at', 'updated_at'],
++ // Erased by extraErase below, after the session lines that point at these task ids (the SDK deletes the
++ // table's rows before calling it, so a default delete here would lose the ids first).
++ erase: { keep: 'erased with its codes_session_lines by the service extraErase' },
},
];
function create({ db, log = console }) {
-- return createAccountData({ db, service: 'codes', tables: TABLES, log });
++ return createAccountData({
++ db, service: 'codes', tables: TABLES, log,
++ // The sessions of their tasks, line by line, in task and stream order (at most SESSION_LINES_MAX, oldest
++ // tasks first out).
++ extraExport: async (d, subject) => {
++ const lines = await d.many(
++ `SELECT l.task_id, l.seq, l.at, l.kind, l.body FROM codes_session_lines l JOIN run_tasks r ON r.id = l.task_id
++ WHERE r.subject = $1 ORDER BY r.created_at DESC, l.seq ASC LIMIT ${SESSION_LINES_MAX}`, [subject]);
++ return lines.length ? [{ name: 'session-lines.json', content: lines }] : [];
++ },
++ extraErase: async (t, subjects, { erased, retained, add }) => {
++ const rows = await t.many('SELECT id FROM run_tasks WHERE subject = ANY($1::text[])', [subjects]);
++ if (rows.length) {
++ const ids = rows.map((r) => r.id);
++ add(erased, 'session_lines', await t.exec('DELETE FROM codes_session_lines WHERE task_id = ANY($1::text[])', [ids]));
++ }
++ delete retained.run_tasks;
++ add(erased, 'run_tasks', await t.exec('DELETE FROM run_tasks WHERE subject = ANY($1::text[])', [subjects]));
++ },
++ });
}
module.exports = { create, TABLES, TOPICS };
ddiff --git a/server/app.js b/server/app.js
iindex bacb6cd..9f9aec6 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -31,6 +31,10 @@ const { createRunRoutes } = require('./http/run');
const { createTasksApi } = require('./http/tasks-api');
const { createActionsApi } = require('./http/actions-api');
const { createRunClient } = require('./domain/run-client');
++const { createRunTasks } = require('./domain/run-tasks');
++const { createSessionLines } = require('./domain/session-lines');
++const { createOutbox, createPublisher } = require('./events/outbox');
++const { createEventsConsumer } = require('./events-consumer');
const { createApi } = require('./http/api');
const { createMovedRoutes } = require('./http/moved');
const { createCodesReadiness } = require('./observability');
@@@ -56,7 +60,12 @@ async function createApp(opts = {}) {
const sso = createSso({ config, keys, fetchImpl: fetchImpl || globalThis.fetch, now: store.now, log });
// The harness catalog (server/data/harness-offers.json), read once: a bad seed row fails boot.
const harnesses = createHarnesses();
-- const ctx = { config, store, keys, sso, harnesses, log };
++ // Sessions, the run.job.* consumer and the codes.task.* outbox share one store and one clock.
++ const sessions = createSessionLines({ store, config, log });
++ const outbox = createOutbox({ store, log });
++ const run = opts.run || createRunClient({ config, fetchImpl: fetchImpl || globalThis.fetch });
++ const tasks = createRunTasks({ store, run, config, log, sessions, outbox });
++ const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks };
const app = express();
app.disable('x-powered-by');
@@@ -103,15 +112,21 @@ async function createApp(opts = {}) {
app.use(cookieParser());
// ── OpenVibe.Events → Codes (loopback only: nginx answers 404 for /internal/) ──
-- // network.account.export_requested and network.account.deleted (ADR-033) for the coding tasks people started here.
-- // The consumer reads the raw body itself (the v2 signature covers it) and refuses a request that came through a proxy.
++ // run.job.* finalizes the coding tasks people started here; network.account.export_requested and
++ // network.account.deleted (ADR-033) go to openvibe-sdk/account-data. The consumer reads the raw body itself (the
++ // v2 signature covers it) and refuses a request that came through a proxy.
{
const { createNetworkSender } = require('openvibe-sdk/account-data');
const accountData = require('./account-data').create({ db: store.db, log });
const accountSend = opts.accountSend || (config.oauth.clientSecret
? createNetworkSender({ networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret, fetch: fetchImpl || globalThis.fetch })
: null);
-- app.post('/internal/events', accountData.consumer({ secrets: config.events.secrets, send: accountSend, log }));
++ const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });
++ ctx.accountData = accountData;
++ ctx.eventsConsumer = consumer;
++ ctx.publisher = createPublisher({ config, store, fetchImpl: fetchImpl || globalThis.fetch, log });
++ ctx.sessionRetention = sessions.retention;
++ app.use('/internal/events', consumer.router);
}
// ── Machine endpoints ───────────────────────────────────
@@@ -149,16 +164,15 @@ async function createApp(opts = {}) {
// ── API ─────────────────────────────────────────────────
// Run on your machine (plan T16): coding tasks on the person's own machine, through OpenVibe Run, from the /run
// pages below or, for a first-party service acting for the person, from /api/v1/tasks (codes.task.*).
-- ctx.run = opts.run || createRunClient({ config, fetchImpl: fetchImpl || globalThis.fetch });
app.use('/api/v1', createApi(ctx));
-- app.use('/api/v1', createTasksApi({ config, store, keys, run: ctx.run, harnesses, actorLimits: ctx.actorLimits, log }));
-- app.use('/api/v1', createActionsApi({ config, store, keys, run: ctx.run, actorLimits: ctx.actorLimits, log }));
++ app.use('/api/v1', createTasksApi({ config, store, keys, run, harnesses, tasks, actorLimits: ctx.actorLimits, log }));
++ app.use('/api/v1', createActionsApi({ config, store, keys, run, tasks, actorLimits: ctx.actorLimits, log }));
app.use('/api', (req, res) => contracts.http.sendProblem(res, 404, 'route.not_found', { detail: 'No such API route', ctx: req.ov }));
// ── Pages ───────────────────────────────────────────────
app.use(rateLimit({ windowMs: 60_000, limit: Number(process.env.CODES_RATE_LIMIT_PER_MIN) || 300, standardHeaders: true, legacyHeaders: false })); // per address; tests raise it
app.use(createPageRoutes(ctx));
-- app.use(createRunRoutes({ config, store, sso, harnesses, run: ctx.run, fetchImpl: fetchImpl || globalThis.fetch, log }));
++ app.use(createRunRoutes({ config, store, sso, harnesses, run, tasks, sessions, keys, fetchImpl: fetchImpl || globalThis.fetch, log }));
app.use((req, res) => send(res, 404, { viewer: req.viewer, config, path: req.originalUrl, title: 'Not found', body: html`<h1>Not found</h1><p>No page here. Try <a href="/harnesses">the agents</a>, <a href="/improve">Improve OpenVibe</a>, or the developer platform at <a href="https://openvibe.services">openvibe.services</a>.</p>` }));
// eslint-disable-next-line no-unused-vars
ddiff --git a/server/config.js b/server/config.js
iindex d4ac49a..4a71991 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -25,6 +25,8 @@ function load(env = process.env) {
nodeEnv,
isProduction,
baseUrl,
++ // Keep every background loop off in a drill (CODES_DRILL=1), the network's convention.
++ drill: env.CODES_DRILL === '1' || env.CODES_DRILL === 'true',
trustProxy: env.TRUST_PROXY != null ? Number(env.TRUST_PROXY) : 2,
// Per-actor limits (server/http/actor-limits.js, roadmap WS-R task 4): the reads one caller (an
// app, a person, else an address) may make per minute and per hour. Writes set tighter numbers there.
@@@ -61,10 +63,20 @@ function load(env = process.env) {
sessionAudience: env.OV_SESSION_AUDIENCE || 'openvibe.network',
},
cookies: { secure: env.COOKIE_SECURE ? env.COOKIE_SECURE === 'true' : isProduction },
-- // OpenVibe.Events (ADR-033): account export and deletion arrive at POST /internal/events, signed with one of these.
++ // OpenVibe.Events (ADR-033): account export and deletion arrive at POST /internal/events, signed with one of
++ // these; run.job.* subscriptions are created at boot. The outbox relay publishes codes.task.* with the
++ // service's events.event.publish token.
events: {
-- url: trim(env.EVENTS_URL || ''),
++ url: trim(env.CODES_EVENTS_URL || env.EVENTS_URL || ''),
++ endpoint: env.CODES_EVENTS_ENDPOINT || '',
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',
++ },
++ // Session lines in PostgreSQL: no total cap; finished tasks keep theirs for this many days (default 90).
++ sessions: {
++ retentionDays: Math.max(1, int(env.CODES_SESSION_RETENTION_DAYS, 90)),
++ retentionIntervalMs: Math.max(60_000, int(env.CODES_SESSION_RETENTION_INTERVAL_MS, 86_400_000)),
},
// OpenVibe.Run (plan T16): a coding task on the person's own machine is a Run job Codes starts for them.
run: { url: trim(env.CODES_RUN_URL || 'http://127.0.0.1:4920') },
ddiff --git a/server/db.js b/server/db.js
iindex dfee53e..4c0ab36 100644
---- a/server/db.js
++++ b/server/db.js
@@@ -3,9 +3,10 @@
/**
* OpenVibe.Codes' own PostgreSQL database (ADR-035, roadmap WS-X2): the schema is migrations/NNNN_*.sql, applied at boot.
*
-- * Codes holds no table today: the developer console's records (manifests, releases, trust tiers, playground logs)
-- * moved to OpenVibe.Services on 2026-10-08 and migration 0002 dropped them here. The harness's runs, sessions and
-- * hand-offs will live here when they arrive; nothing written here will ever contain a secret, token or credential.
++ * Codes holds its own tables (run_tasks, account_data_events, telemetry_samples, codes_session_lines,
++ * codes_events_inbox and codes_outbox): the developer console's records (manifests, releases, trust tiers, playground
++ * logs) moved to OpenVibe.Services on 2026-10-08 and migration 0002 dropped them here. Nothing written here will ever
++ * contain a secret, token or credential.
*/
const fs = require('fs');
const path = require('path');
ddiff --git a/server/domain/run-tasks.js b/server/domain/run-tasks.js
iindex 3c78f28..17f3a54 100644
---- a/server/domain/run-tasks.js
++++ b/server/domain/run-tasks.js
@@@ -4,15 +4,17 @@
* Coding tasks on a person's own machine (run_tasks, plan T16): what the /run pages (server/http/run.js) and the
* service API (server/http/tasks-api.js, codes.task.*) share.
*
-- * createRunTasks({ store, run, config, log })
++ * createRunTasks({ store, run, config, log, sessions, outbox })
* start({ subject, project, args, via, key }) → the row: queued (Run took the job) or failed (Run's reason)
-- * refresh(row) → the row, updated from Run while its job has not ended
++ * refresh(row, { force }) → the row, read from Run only as a fallback after 10 quiet minutes
++ * applyEvent(row, event) → a run.job.* delivery: state, timings, result, event, in one tx
* latestProject(subject) → the project of the person's latest task, or null
* present(row) → codes.task@1
-- * TERMINAL, STATE_LABEL
++ * TERMINAL, STATE_LABEL, STALE_MS
*
-- * A task's result comes from the person's machine: present() passes on only the contract's fields, each checked and
-- * cut to its limit.
++ * Every state change writes its codes.task.* outbox row in the same transaction; a terminal state also back-fills
++ * the session's missing lines and marks it complete. A task's result comes from the person's machine: present() passes
++ * on only the contract's fields, each checked and cut to its limit.
*/
const { RunError } = require('./run-client');
@@@ -22,6 +24,9 @@ const STATE_LABEL = {
failed: 'Failed', cancelled: 'Cancelled', expired: 'Expired (no machine took it in time)',
};
const STATES = new Set(Object.keys(STATE_LABEL));
++const RANK = { queued: 0, placed: 1, running: 2, succeeded: 3, failed: 3, cancelled: 3, expired: 3 };
++/** A non-terminal task is read from Run at most once after this long without an event (the fallback, not polling). */
++const STALE_MS = 10 * 60_000;
const iso = (ms) => new Date(Number(ms)).toISOString();
const str = (v, max) => (typeof v === 'string' ? v.slice(0, max) : undefined);
@@@ -31,6 +36,11 @@ const httpsUrl = (v) => {
try { return new URL(v).protocol === 'https:' ? v : undefined; } catch { return undefined; }
};
const drop = (o) => Object.fromEntries(Object.entries(o).filter(([, v]) => v !== undefined));
++const asObject = (v) => {
++ if (v && typeof v === 'object' && !Array.isArray(v)) return v;
++ if (typeof v === 'string') { try { const p = JSON.parse(v); return p && typeof p === 'object' && !Array.isArray(p) ? p : null; } catch { return null; } }
++ return null;
++};
/** The job's result document (harness/job.js) as codes.task@1's `result`, or undefined. */
function presentResult(d) {
@@@ -69,35 +79,125 @@ function sessionLogOf(row) {
return lines.length ? lines : null;
}
--function createRunTasks({ store, run, config, log = console }) {
++function errorOf(job) {
++ if (!job || !job.error) return null;
++ if (typeof job.error === 'string') return job.error.slice(0, 1000);
++ return `${job.error.code || 'run.error'}${job.error.detail ? `: ${job.error.detail}` : ''}`.slice(0, 1000);
++}
++
++function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null }) {
const db = store.db;
const who = (row) => ({ subject: row.subject, project: row.project_id });
++ // Lazily, so a caller that only reads tasks does not need the session/outbox modules (and there is no cycle).
++ const sessionLines = sessions || require('./session-lines').createSessionLines({ store, config, log });
++ const taskOutbox = outbox || require('../events/outbox').createOutbox({ store, log });
++
++ /**
++ * One state change: update the row, write its codes.task.* outbox row in the same transaction, and on a terminal
++ * state reconcile the session with the result and mark it complete. A late or out-of-order state never moves a
++ * task backwards, and a terminal state is final.
++ */
++ async function recordState(row, next) {
++ const state = STATES.has(next && next.state) ? next.state : null;
++ if (!state || !row || !row.id) return row;
++ return await store.tx(async () => {
++ const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
++ if (!current) return row;
++ if (RANK[state] < RANK[current.state]) return current;
++ if (TERMINAL.has(current.state) && current.state !== state) return current;
++ const changed = current.state !== state;
++ const startedAt = next.startedAt || (next.job && next.job.timings && next.job.timings.started_at) || null;
++ const finishedAt = next.finishedAt || (next.job && next.job.timings && next.job.timings.finished_at) || (TERMINAL.has(state) ? store.iso() : null);
++ const result = next.result === undefined ? null : next.result;
++ const error = next.error === undefined ? null : next.error;
++ const updated = await db.one(
++ `UPDATE run_tasks SET state = $2,
++ result = COALESCE($3::jsonb, result),
++ error = COALESCE($4, error),
++ started_at = COALESCE($5::timestamptz, started_at),
++ finished_at = COALESCE($6::timestamptz, finished_at),
++ updated_at = $7
++ WHERE id = $1 RETURNING *`,
++ [current.id, state, result == null ? null : JSON.stringify(result), error, startedAt, finishedAt, store.now()],
++ );
++ if (changed) await taskOutbox.emitTask(updated, current.state);
++ if (TERMINAL.has(updated.state) && !updated.session_complete_at) {
++ await sessionLines.backfillAndComplete(updated, updated.result);
++ }
++ return updated;
++ });
++ }
async function start({ subject, project, args, via = null, key = null }) {
const id = store.newId('tsk');
const at = store.now();
-- await db.exec(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, created_at, updated_at)
-- VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12, $12)`,
-- [id, subject, project, args.repo, args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, at]);
++ await store.tx(async () => {
++ const inserted = await db.one(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, created_at, updated_at)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12, $12) RETURNING *`,
++ [id, subject, project, args.repo, args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, at]);
++ await taskOutbox.emitTask(inserted, null);
++ return inserted;
++ });
++ let job;
try {
-- const job = await run.submit({ subject, project, args, key: id });
-- await db.exec('UPDATE run_tasks SET job_id = $2, state = $3, updated_at = $4 WHERE id = $1', [id, job.id, job.state, store.now()]);
++ job = await run.submit({ subject, project, args, key: id });
} catch (e) {
const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
-- await db.exec("UPDATE run_tasks SET state = 'failed', error = $2, updated_at = $3 WHERE id = $1", [id, why, store.now()]);
++ return await store.tx(async () => {
++ const failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, finished_at = $3::timestamptz, updated_at = $4
++ WHERE id = $1 RETURNING *`, [id, why, store.iso(), store.now()]);
++ await taskOutbox.emitTask(failed, 'queued');
++ return failed;
++ });
}
-- return db.one('SELECT * FROM run_tasks WHERE id = $1', [id]);
++ return await store.tx(async () => {
++ const state = STATES.has(job.state) ? job.state : 'queued';
++ const startedAt = job.timings && job.timings.started_at ? job.timings.started_at : null;
++ const updated = await db.one(`UPDATE run_tasks SET job_id = $2, state = $3, started_at = COALESCE($4::timestamptz, started_at), updated_at = $5
++ WHERE id = $1 RETURNING *`, [id, job.id, state, startedAt, store.now()]);
++ await taskOutbox.emitTask(updated, 'queued');
++ return updated;
++ });
++ }
++
++ /** A Run job document (run.job-read-result@1) as the task row. */
++ async function applyJob(row, job) {
++ if (!job || !STATES.has(job.state)) return row;
++ const result = TERMINAL.has(job.state) ? (asObject(job.result) || null) : undefined;
++ return await recordState(row, {
++ state: job.state,
++ result,
++ error: TERMINAL.has(job.state) ? errorOf(job) : undefined,
++ startedAt: job.timings && job.timings.started_at,
++ finishedAt: job.timings && job.timings.finished_at,
++ job,
++ });
}
-- async function refresh(row) {
++ /**
++ * Read Run once when a task has been non-terminal for more than 10 minutes without an event (the single fallback
++ * for a missing delivery, not page-view polling). force reads now, for a cancel or an explicit check.
++ */
++ async function refresh(row, { force = false } = {}) {
if (!row.job_id || TERMINAL.has(row.state)) return row;
++ if (!force && store.now() - Number(row.updated_at) < STALE_MS) return row;
let job;
try { job = await run.read(row.job_id, who(row)); } catch (e) { log.warn(`[Codes] task ${row.id}: ${e.message}`); return row; }
-- if (job.state === row.state && !TERMINAL.has(job.state)) return row;
-- const error = job.error ? `${job.error.code}${job.error.detail ? `: ${job.error.detail}` : ''}` : null;
-- const result = job.result && typeof job.result === 'object' ? job.result : null;
-- return db.one('UPDATE run_tasks SET state = $2, result = $3, error = $4, updated_at = $5 WHERE id = $1 RETURNING *',
-- [row.id, job.state, result ? JSON.stringify(result) : null, error, store.now()]);
++ return await applyJob(row, job);
++ }
++
++ /** A run.job.* delivery: the event's state/timings and the result read once from Run, in the caller's tx. */
++ async function applyEvent(row, { state, job = null, error = null, timings = null } = {}) {
++ if (!STATES.has(state)) return row;
++ const result = TERMINAL.has(state) ? (job && asObject(job.result)) || null : undefined;
++ const why = error || errorOf(job);
++ return await recordState(row, {
++ state,
++ result,
++ error: TERMINAL.has(state) ? why : undefined,
++ startedAt: timings && timings.started_at,
++ finishedAt: timings && timings.finished_at,
++ });
}
async function latestProject(subject) {
@@@ -106,7 +206,7 @@ function createRunTasks({ store, run, config, log = console }) {
}
function present(row) {
-- const result = TERMINAL.has(row.state) ? presentResult(typeof row.result === 'string' ? JSON.parse(row.result) : row.result) : undefined;
++ const result = TERMINAL.has(row.state) ? presentResult(asObject(row.result)) : undefined;
return drop({
id: row.id, state: STATES.has(row.state) ? row.state : 'failed', repo: String(row.repo).slice(0, 300), ref: row.ref || null,
agent: row.agent || null, url: `${config.baseUrl}/run/tasks/${row.id}`, created_at: iso(row.created_at), updated_at: iso(row.updated_at),
@@@ -114,7 +214,7 @@ function createRunTasks({ store, run, config, log = console }) {
});
}
-- return { start, refresh, latestProject, present, who };
++ return { start, refresh, applyJob, applyEvent, recordState, latestProject, present, who };
}
--module.exports = { createRunTasks, presentResult, TERMINAL, STATE_LABEL, sessionLogOf };
++module.exports = { createRunTasks, presentResult, TERMINAL, STATE_LABEL, STATES, RANK, STALE_MS, sessionLogOf, errorOf, asObject };
ddiff --git a/server/domain/session-lines.js b/server/domain/session-lines.js
nnew file mode 100644
iindex 0000000..7bc210b
---- /dev/null
++++ b/server/domain/session-lines.js
@@@ -0,0 +1,334 @@
++'use strict';
++
++/**
++ * A coding task's session in PostgreSQL (codes_session_lines): the job lines the session view reads, written as the
++ * Run stream carries them, back-filled from the machine's final result log, and replayed by GET
++ * /run/tasks/:id/session.json. The stream's own order is the seq, so a reconnect that replays the same events is a
++ * no-op on (task_id, seq); a line is clipped to 64 KB with a marker and there is no total cap.
++ *
++ * createSessionLines({ store, config, log })
++ * append(taskId, lines) one INSERT … ON CONFLICT DO NOTHING per line
++ * recorder(taskId) push(text) / flush(): parses Run's output|state|end SSE frames
++ * backfill(row, result) adds the final log's lines the stream missed, in content order
++ * backfillAndComplete(row, result) backfill, then set run_tasks.session_complete_at
++ * linesForRow(row) the table's bodies, else the old result.log (tasks before this change)
++ * replay(row, { after, limit }) a page for the session.json route
++ * retention { runOnce, start, stop } (CODES_SESSION_RETENTION_DAYS, default 90)
++ */
++const { TERMINAL, sessionLogOf } = require('./run-tasks');
++
++const MAX_BODY_BYTES = 64 * 1024;
++const CLIP_MARKER = '… [clipped]';
++const PAGE_LIMIT = 500;
++const SEQ_STRIDE = 250_000; // one stream event's lines fit inside this (a 64 KB chunk has < 66k newlines); the
++ // column is bigint, so any event seq Run sends fits (an integer would stop at ~8 600)
++const STATE_END_INDEX = 0;
++const PARTIAL_LINE_INDEX = SEQ_STRIDE - 1;
++
++const iso = (ms) => new Date(Number(ms)).toISOString();
++const atIso = (value) => {
++ const t = value instanceof Date ? value.getTime() : Date.parse(String(value || ''));
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++};
++
++/** Every string in a JSON value, with the object/key that owns it, longest first. */
++function stringSlots(value, out = [], seen = new Set()) {
++ if (!value || typeof value !== 'object' || seen.has(value)) return out;
++ seen.add(value);
++ if (Array.isArray(value)) {
++ value.forEach((v, i) => (typeof v === 'string' ? out.push({ parent: value, key: i, value: v }) : stringSlots(v, out, seen)));
++ return out;
++ }
++ for (const key of Object.keys(value)) {
++ if (typeof value[key] === 'string') out.push({ parent: value, key, value: value[key] });
++ else stringSlots(value[key], out, seen);
++ }
++ return out;
++}
++
++/**
++ * A body no larger than 64 KB: the longest strings are cut and end in the marker. A body that cannot be made small
++ * (thousands of short keys) becomes the small marker line instead, so a line never fails to store.
++ */
++function clipBody(body) {
++ let value = body;
++ if (value == null) value = { t: 'text', text: '' };
++ else if (typeof value !== 'object') value = { t: 'text', text: String(value) };
++ let json;
++ try { json = JSON.stringify(value); } catch { return { t: 'note', level: 'warning', text: CLIP_MARKER, clipped: true }; }
++ if (!json) json = '{}';
++ if (Buffer.byteLength(json) <= MAX_BODY_BYTES) return value;
++ let clone;
++ try { clone = JSON.parse(json); } catch { return { t: 'note', level: 'warning', text: CLIP_MARKER, clipped: true }; }
++ const slots = stringSlots(clone).sort((a, b) => b.value.length - a.value.length);
++ let current = Buffer.byteLength(JSON.stringify(clone));
++ for (let pass = 0; pass < 12 && current > MAX_BODY_BYTES; pass++) {
++ let changed = false;
++ for (const slot of slots) {
++ if (current <= MAX_BODY_BYTES) break;
++ const text = slot.parent[slot.key];
++ if (typeof text !== 'string' || !text.length) continue;
++ const over = current - MAX_BODY_BYTES;
++ const perChar = Math.max(1, Buffer.byteLength(text) / text.length);
++ const cut = Math.max(1, Math.ceil(over / perChar) + CLIP_MARKER.length);
++ slot.parent[slot.key] = `${text.slice(0, Math.max(0, text.length - cut))}${CLIP_MARKER}`;
++ current = Buffer.byteLength(JSON.stringify(clone));
++ changed = true;
++ }
++ if (!changed) break;
++ }
++ if (current > MAX_BODY_BYTES) {
++ const t = typeof clone.t === 'string' ? clone.t.slice(0, 40) : 'note';
++ return { t, level: 'warning', text: CLIP_MARKER, clipped: true };
++ }
++ return clone;
++}
++
++/** The session line a raw stdout line becomes: the job's own JSON object when it parses, else the text itself. */
++function lineOf(raw) {
++ let body;
++ try { body = JSON.parse(raw); } catch { body = null; }
++ if (!body || typeof body !== 'object' || Array.isArray(body)) return { kind: 'text', body: { t: 'text', text: raw } };
++ const kind = typeof body.t === 'string' ? body.t : (typeof body.kind === 'string' ? body.kind : 'output');
++ return { kind, body };
++}
++
++function createSessionLines({ store, config = {}, log = console }) {
++ const db = store.db;
++ const retentionDays = Math.max(1, Number(config.sessions && config.sessions.retentionDays) || 90);
++
++ /** Insert lines in order; an existing (task_id, seq) wins, so a replay never duplicates. → rows inserted. */
++ async function append(taskId, lines) {
++ if (!taskId || !Array.isArray(lines) || !lines.length) return 0;
++ const clean = [];
++ for (const line of lines) {
++ const seq = Number(line && line.seq);
++ if (!Number.isSafeInteger(seq) || seq < 1) continue;
++ const at = atIso(line.at) || store.iso();
++ const kind = String(line.kind || 'text').slice(0, 64) || 'text';
++ clean.push({ seq, at, kind, body: clipBody(line.body) });
++ }
++ let inserted = 0;
++ for (let i = 0; i < clean.length; i += 200) {
++ const chunk = clean.slice(i, i + 200);
++ const values = [];
++ const params = [];
++ chunk.forEach((line, j) => {
++ const b = j * 5;
++ values.push(`($${b + 1}, $${b + 2}, $${b + 3}::timestamptz, $${b + 4}, $${b + 5}::jsonb)`);
++ params.push(taskId, line.seq, line.at, line.kind, JSON.stringify(line.body));
++ });
++ inserted += await db.exec(
++ `INSERT INTO codes_session_lines (task_id, seq, at, kind, body) VALUES ${values.join(', ')}
++ ON CONFLICT (task_id, seq) DO NOTHING`, params);
++ }
++ return inserted;
++ }
++
++ /** All rows of a task in stream order. */
++ async function allRows(taskId) {
++ return await db.many('SELECT seq, at, kind, body FROM codes_session_lines WHERE task_id = $1 ORDER BY seq ASC', [taskId]);
++ }
++
++ /**
++ * A recorder for the SSE proxy: push the decoded upstream text as it arrives, then flush. Output frames are split
++ * into the job's JSON lines; state and end frames become their own lines. The stream's seq is the base, so the
++ * same event replayed after a reconnect maps to the same rows.
++ */
++ function recorder(taskId) {
++ let frameBuffer = '';
++ let outputBuffer = '';
++ let fallbackSeq = 0;
++ let lastBase = 0;
++ let ended = false;
++
++ function baseOf(event) {
++ const n = Number(event && event.seq);
++ if (Number.isSafeInteger(n) && n >= 1 && n * SEQ_STRIDE <= Number.MAX_SAFE_INTEGER) lastBase = n * SEQ_STRIDE;
++ else lastBase = (++fallbackSeq) * SEQ_STRIDE;
++ return lastBase;
++ }
++ function recordsOf(frame) {
++ const lines = String(frame).split('\n');
++ let type = '';
++ const data = [];
++ for (const line of lines) {
++ if (line.startsWith('event:')) type = line.slice(6).trim();
++ else if (line.startsWith('data:')) data.push(line.slice(5).replace(/^ /, ''));
++ }
++ if (!data.length) return [];
++ let event;
++ try { event = JSON.parse(data.join('\n')); } catch { return []; }
++ if (!event || typeof event !== 'object') return [];
++ const at = atIso(event.at) || store.iso();
++ const kind = type || event.type;
++ if (kind === 'output') {
++ outputBuffer += String(event.chunk == null ? '' : event.chunk);
++ const parts = outputBuffer.split('\n');
++ outputBuffer = parts.pop();
++ const base = baseOf(event);
++ const out = [];
++ parts.forEach((raw, i) => {
++ if (!raw.trim()) return;
++ const line = lineOf(raw);
++ out.push({ seq: base + i, at, kind: line.kind, body: line.body });
++ });
++ return out;
++ }
++ if (kind === 'state' || kind === 'end') {
++ if (kind === 'end') ended = true;
++ const state = event.state == null ? null : String(event.state);
++ return [{ seq: baseOf(event) + STATE_END_INDEX, at, kind, body: { t: kind, state } }];
++ }
++ return [];
++ }
++ function drainFrames(out) {
++ for (;;) {
++ const i = frameBuffer.indexOf('\n\n');
++ if (i < 0) break;
++ const frame = frameBuffer.slice(0, i);
++ frameBuffer = frameBuffer.slice(i + 2);
++ out.push(...recordsOf(frame));
++ }
++ }
++ return {
++ async push(text) {
++ frameBuffer = `${frameBuffer}${String(text == null ? '' : text)}`.replace(/\r\n/g, '\n');
++ const out = [];
++ drainFrames(out);
++ if (out.length) await append(taskId, out);
++ },
++ async flush() {
++ const out = [];
++ drainFrames(out);
++ if (frameBuffer.trim()) out.push(...recordsOf(frameBuffer));
++ frameBuffer = '';
++ if (outputBuffer !== '') {
++ const line = lineOf(outputBuffer);
++ out.push({ seq: (lastBase || 0) + PARTIAL_LINE_INDEX, at: store.iso(), kind: line.kind, body: line.body });
++ outputBuffer = '';
++ }
++ if (out.length) await append(taskId, out);
++ },
++ ended: () => ended,
++ };
++ }
++
++ /**
++ * Add the final log's lines the stream missed, in content order. Existing harness lines are matched by their
++ * JSON, so a reconnect or a second result never adds a line twice; the new seqs sit after the streamed output and
++ * before the terminal state/end line.
++ */
++ async function backfill(row, result) {
++ if (!row || !row.id) return 0;
++ const rows = await allRows(row.id);
++ const harness = rows.filter((r) => r && r.body && typeof r.body === 'object' && !Array.isArray(r.body)
++ && typeof r.body.t === 'string' && r.kind !== 'state' && r.kind !== 'end');
++ const existing = harness.map((r) => JSON.stringify(r.body));
++ const log = sessionLogOf({ result: result === undefined ? row.result : result }) || [];
++ let cursor = 0;
++ const missing = [];
++ for (const body of log) {
++ const text = JSON.stringify(body);
++ let found = -1;
++ for (let i = cursor; i < existing.length; i++) { if (existing[i] === text) { found = i; break; } }
++ if (found >= 0) cursor = found + 1;
++ else missing.push(body);
++ }
++ if (!missing.length) return 0;
++ const used = new Set(rows.map((r) => Number(r.seq)));
++ let start = harness.reduce((max, r) => Math.max(max, Number(r.seq) || 0), 0) + 1;
++ while (used.has(start)) start++;
++ const at = atIso(row.finished_at) || (row.updated_at ? iso(row.updated_at) : store.iso());
++ return await append(row.id, missing.map((body, i) => ({
++ seq: start + i, at, kind: typeof body.t === 'string' ? body.t : 'text', body,
++ })));
++ }
++
++ async function complete(taskId) {
++ return await db.exec(
++ 'UPDATE run_tasks SET session_complete_at = COALESCE(session_complete_at, $2::timestamptz) WHERE id = $1',
++ [taskId, store.iso()],
++ );
++ }
++
++ async function backfillAndComplete(row, result) {
++ const inserted = await backfill(row, result);
++ await complete(row.id);
++ return inserted;
++ }
++
++ /** The bodies the session view replays: the table first, the old result.log for a task from before this change. */
++ async function linesForRow(row) {
++ if (!row || !row.id) return [];
++ const rows = await allRows(row.id);
++ const session = rows.filter((r) => r.kind !== 'state' && r.kind !== 'end').map((r) => r.body);
++ if (session.length) return session;
++ return sessionLogOf(row) || [];
++ }
++
++ /** One replay page (at most 500 lines after `after`), from the table or the old result.log. */
++ async function replay(row, { after = 0, limit = PAGE_LIMIT } = {}) {
++ const from = Math.max(0, Number.isSafeInteger(Number(after)) ? Number(after) : 0);
++ const max = Math.min(PAGE_LIMIT, Math.max(1, Number.isSafeInteger(Number(limit)) ? Number(limit) : PAGE_LIMIT));
++ const rows = await allRows(row.id);
++ let lines;
++ let source;
++ if (rows.length) {
++ source = 'session';
++ lines = rows.filter((r) => Number(r.seq) > from).map((r) => ({ seq: Number(r.seq), at: r.at, kind: r.kind, body: r.body }));
++ } else {
++ source = 'result.log';
++ const at = row.created_at ? iso(row.created_at) : store.iso();
++ lines = (sessionLogOf(row) || []).map((body, i) => ({ seq: i + 1, at, kind: typeof body.t === 'string' ? body.t : 'text', body }))
++ .filter((line) => line.seq > from);
++ }
++ const page = lines.slice(0, max);
++ return {
++ task_id: row.id,
++ source,
++ after: from,
++ lines: page,
++ next_after: page.length ? page[page.length - 1].seq : from,
++ has_more: lines.length > max,
++ complete: row.session_complete_at != null || TERMINAL.has(row.state),
++ };
++ }
++
++ /** Delete the lines of tasks that finished more than CODES_SESSION_RETENTION_DAYS ago (default 90). */
++ async function retentionOnce() {
++ const cutoff = new Date(store.now() - retentionDays * 86_400_000).toISOString();
++ return await db.exec(
++ `DELETE FROM codes_session_lines WHERE task_id IN (
++ SELECT id FROM run_tasks
++ WHERE state = ANY($1)
++ AND COALESCE(finished_at, to_timestamp(updated_at / 1000.0)) < $2::timestamptz
++ )`, [[...TERMINAL], cutoff]);
++ }
++
++ let running = false;
++ let timer = null;
++ const retention = {
++ days: retentionDays,
++ runOnce: async () => {
++ if (running) return 0;
++ running = true;
++ try { return await retentionOnce(); } catch (err) { log.warn(`[Codes] session retention failed: ${err.message}`); return 0; } finally { running = false; }
++ },
++ start() {
++ if (timer || config.drill) return false;
++ const jobs = require('../utils/jobs');
++ const every = jobs.every('codes-session-retention', (Number(config.sessions && config.sessions.retentionIntervalMs) || 86_400_000), () => retention.runOnce(), { initialDelayMs: 60_000, jitterMs: 60_000, log });
++ timer = every;
++ return true;
++ },
++ stop() { if (timer) timer(); timer = null; },
++ };
++
++ return {
++ MAX_BODY_BYTES, retentionDays, append, allRows, recorder, backfill, backfillAndComplete, complete,
++ linesForRow, replay, retention, clipBody,
++ };
++}
++
++module.exports = { createSessionLines, clipBody, lineOf, MAX_BODY_BYTES, CLIP_MARKER, PAGE_LIMIT, SEQ_STRIDE };
ddiff --git a/server/events-consumer.js b/server/events-consumer.js
nnew file mode 100644
iindex 0000000..72928e1
---- /dev/null
++++ b/server/events-consumer.js
@@@ -0,0 +1,202 @@
++'use strict';
++
++/**
++ * OpenVibe.Events → Codes, and the subscriptions that bring the deliveries (plan 2026-10-28 §6, decision 2).
++ *
++ * run.job.queued|started|succeeded|failed|cancelled|expired
++ * a job that is a Codes task is finalized from its event (state, timings, and on an end state
++ * the result read once from Run). The event id is claimed in codes_events_inbox in the same
++ * transaction as the row update, so a redelivery changes nothing.
++ * network.account.export_requested / network.account.deleted
++ * handed to openvibe-sdk/account-data (server/account-data.js), which keeps its own idempotency.
++ *
++ * POST /internal/events is a signed v2 delivery endpoint (loopback only; nginx answers 404 for /internal), the shape
++ * Help and Today use. startSubscriptions() creates the missing subscriptions at Events at boot, idempotently, with
++ * backoff; it is off without EVENTS_URL/CODES_EVENTS_SECRET/the OAuth client secret, under CODES_DRILL, or when
++ * CODES_EVENTS_SUBSCRIBE=0.
++ *
++ * const consumer = createEventsConsumer({ config, store, tasks, run, accountData, accountSend, log });
++ * app.use('/internal/events', consumer.router);
++ */
++const express = require('express');
++const { http, serviceAuth } = require('openvibe-contracts');
++const { parseDelivery } = require('openvibe-sdk/events');
++const accountDataLib = require('openvibe-sdk/account-data');
++const { TERMINAL, STATES } = require('./domain/run-tasks');
++
++const RUN_TOPICS = Object.freeze([
++ 'run.job.queued', 'run.job.started', 'run.job.succeeded', 'run.job.failed', 'run.job.cancelled', 'run.job.expired',
++]);
++const TOPICS = Object.freeze([...RUN_TOPICS, ...accountDataLib.TOPICS]);
++const EVENT_ID_RE = /^evt_[0-9A-HJKMNP-TV-Z]{26}$/;
++const JOB_RE = /^job_[0-9A-HJKMNP-TV-Z]{26}$/;
++const STATE_OF_TOPIC = {
++ 'run.job.queued': 'queued', 'run.job.started': 'running', 'run.job.succeeded': 'succeeded',
++ 'run.job.failed': 'failed', 'run.job.cancelled': 'cancelled', 'run.job.expired': 'expired',
++};
++
++function errorText(error) {
++ if (!error) return null;
++ if (typeof error === 'string') return error.slice(0, 1000);
++ if (typeof error === 'object') return `${error.code || 'run.error'}${error.detail ? `: ${error.detail}` : ''}`.slice(0, 1000);
++ return null;
++}
++
++function createEventsConsumer({ config, store, tasks, run, accountData, accountSend = null, log = console, now = store.now }) {
++ const db = store.db;
++ const secrets = (config.events && config.events.secrets ? config.events.secrets : []).filter((x) => typeof x === 'string' && x.length >= 32);
++ const stats = { received: 0, applied: 0, duplicates: 0, ignored: 0, refused: 0, failed: 0, last_at: null };
++
++ /** Record an event that had no effect; a replay answers duplicate. */
++ async function ignore(event, outcome, taskId = null) {
++ const inserted = await db.maybe(
++ `INSERT INTO codes_events_inbox (event_id, event_type, task_id, outcome, processed_at)
++ VALUES ($1, $2, $3, $4, $5::timestamptz) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,
++ [event.event_id, event.event_type, taskId, outcome, store.iso()],
++ );
++ return { duplicate: !inserted, outcome: inserted ? outcome : null, task_id: taskId || undefined };
++ }
++
++ async function handleRun(event) {
++ const payload = event.payload && typeof event.payload === 'object' && !Array.isArray(event.payload) ? event.payload : null;
++ if (!payload) return await ignore(event, 'ignored:payload');
++ const jobId = String(payload.job_id || '');
++ if (!JOB_RE.test(jobId)) return await ignore(event, 'ignored:payload');
++ const state = STATES.has(payload.state) ? payload.state : STATE_OF_TOPIC[event.event_type];
++ if (!state) return await ignore(event, 'ignored:payload');
++ const seen = await db.maybe('SELECT task_id FROM codes_events_inbox WHERE event_id = $1', [event.event_id]);
++ if (seen) return { duplicate: true, outcome: null, task_id: seen.task_id || undefined };
++ const row = await db.maybe('SELECT * FROM run_tasks WHERE job_id = $1', [jobId]);
++ if (!row) return await ignore(event, 'ignored:not_a_codes_task');
++ const requester = payload.requester && typeof payload.requester === 'object' ? payload.requester.id : null;
++ if (requester && requester !== row.subject) return await ignore(event, 'ignored:another_subject', row.id);
++ if (payload.project_id && payload.project_id !== row.project_id) return await ignore(event, 'ignored:another_project', row.id);
++
++ // A terminal event reads the job once for the result (run.job.* carry no result). A failure here is retried by
++ // Events; the inbox row is only written in the transaction that applies the event.
++ let job = null;
++ if (TERMINAL.has(state)) job = await run.read(jobId, tasks.who(row));
++
++ return await store.tx(async () => {
++ const inserted = await db.maybe(
++ `INSERT INTO codes_events_inbox (event_id, event_type, task_id, outcome, processed_at)
++ VALUES ($1, $2, $3, $4, $5::timestamptz) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,
++ [event.event_id, event.event_type, row.id, state, store.iso()],
++ );
++ if (!inserted) return { duplicate: true, outcome: null, task_id: row.id };
++ await tasks.applyEvent(row, {
++ state, job, error: errorText(payload.error), timings: payload.timings && typeof payload.timings === 'object' ? payload.timings : null,
++ });
++ await db.exec('UPDATE codes_events_inbox SET outcome = $2 WHERE event_id = $1', [event.event_id, state]);
++ return { duplicate: false, outcome: state, task_id: row.id };
++ });
++ }
++
++ const router = express.Router();
++ router.post('/', express.raw({ type: () => true, limit: '256kb' }), async (req, res) => {
++ const ctx = http.requestContext(req.headers);
++ const problem = (status, code, detail) => http.sendProblem(res, status, code, { detail, ctx });
++ if (req.headers['x-forwarded-for'] || req.headers['x-real-ip'] || req.headers['cf-connecting-ip']) return problem(403, 'codes.internal_only', 'internal route');
++ if (!secrets.length) return problem(503, 'codes.events_disabled', 'CODES_EVENTS_SECRET is not set');
++ const raw = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
++ let delivery = null;
++ for (const secret of secrets) { delivery = parseDelivery(raw, req.headers, secret, { requireV2: true, now: now() }); if (delivery) break; }
++ if (!delivery) { stats.refused++; return problem(401, 'codes.bad_signature', 'X-OpenVibe-Signature-V2 does not verify or is outside the replay window'); }
++ const event = delivery.event;
++ if (!event || !EVENT_ID_RE.test(String(event.event_id || '')) || typeof event.event_type !== 'string') {
++ stats.refused++;
++ return problem(400, 'codes.bad_delivery', 'body must be { event: <envelope>, seq }');
++ }
++ stats.received++;
++ stats.last_at = new Date(now()).toISOString();
++
++ if (accountDataLib.TOPICS.includes(event.event_type)) {
++ let outcome;
++ try { outcome = await accountData.apply(event, { send: accountSend }); } catch (err) {
++ stats.failed++;
++ log.error(`[Codes] account event ${event.event_id} failed:`, err.message);
++ return problem(503, 'codes.event_failed', 'processing failed; it will be retried');
++ }
++ if (String(outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;
++ return res.json({ event_id: event.event_id, outcome });
++ }
++
++ if (!RUN_TOPICS.includes(event.event_type)) {
++ const result = await ignore(event, 'ignored:type');
++ if (result.duplicate) stats.duplicates++; else stats.ignored++;
++ return res.json({ event_id: event.event_id, duplicate: result.duplicate, outcome: result.outcome });
++ }
++
++ let result;
++ try { result = await handleRun(event); } catch (err) {
++ stats.failed++;
++ log.error(`[Codes] event ${event.event_id} (${event.event_type}) failed:`, err.message);
++ return problem(503, 'codes.event_failed', 'processing failed; it will be retried');
++ }
++ if (result.duplicate) { stats.duplicates++; return res.json({ event_id: event.event_id, duplicate: true, outcome: null }); }
++ if (String(result.outcome).startsWith('ignored:')) stats.ignored++; else stats.applied++;
++ return res.json({ event_id: event.event_id, duplicate: false, outcome: result.outcome, ...(result.task_id ? { task_id: result.task_id } : {}) });
++ });
++
++ return { router, topics: TOPICS, runTopics: RUN_TOPICS, stats: () => ({ ...stats, enabled: secrets.length > 0 }) };
++}
++
++/** Create any missing subscription for `topics` at Events (idempotent; retried in the background at boot). */
++function startSubscriptions(opts = {}) {
++ const config = opts.config || {};
++ const events = config.events || {};
++ const oauth = config.oauth || {};
++ const port = opts.port != null ? opts.port : config.port;
++ const secret = opts.secret != null ? opts.secret : (events.secrets && events.secrets[0]) || '';
++ const eventsUrl = opts.eventsUrl != null ? opts.eventsUrl : events.url;
++ const networkInternalUrl = opts.networkInternalUrl || config.networkInternalUrl;
++ const clientId = opts.clientId || oauth.clientId;
++ const clientSecret = opts.clientSecret || oauth.clientSecret;
++ const drill = opts.drill != null ? opts.drill : config.drill;
++ const topics = opts.topics || TOPICS;
++ const fetchImpl = opts.fetchImpl || globalThis.fetch;
++ const log = opts.log || console;
++ if (!eventsUrl || !secret || !clientSecret || drill || process.env.CODES_EVENTS_SUBSCRIBE === '0') return null;
++ const base = String(eventsUrl).replace(/\/+$/, '');
++ const endpoint = opts.endpoint || events.endpoint || `http://127.0.0.1:${port}/internal/events`;
++ const tokens = serviceAuth.createTokenClient({
++ tokenUrl: `${networkInternalUrl}/oauth/token`, clientId, clientSecret,
++ audience: 'openvibe.events', scope: 'events.subscription.manage', fetchImpl,
++ });
++ const call = async (method, path, body) => {
++ const res = await fetchImpl(`${base}${path}`, {
++ method, headers: { Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}), ...(await tokens.authHeaders()) },
++ body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(15_000),
++ });
++ const json = await res.json().catch(() => ({}));
++ return { status: res.status, ok: res.ok, body: json };
++ };
++ const attempt = async () => {
++ const listed = await call('GET', '/api/v1/subscriptions');
++ if (!listed.ok) throw new Error(`listing subscriptions: ${listed.status}`);
++ const mine = (listed.body.subscriptions || []).filter((s) => s.endpoint === endpoint);
++ for (const topic of topics) {
++ if (mine.some((s) => s.topic_pattern === topic)) continue;
++ const r = await call('POST', '/api/v1/subscriptions', { topic_pattern: topic, endpoint, secret });
++ if (!r.ok && r.status !== 409) throw new Error(`subscribing to ${topic}: ${r.status} ${r.body.code || ''}`);
++ if (r.ok) log.log(`[Events consumer] subscription created: ${r.body.id} (${topic} → ${endpoint})`);
++ }
++ };
++ const delays = [0, 10_000, 60_000, 5 * 60_000, 15 * 60_000];
++ let i = 0;
++ let timer = null;
++ let stopped = false;
++ const run = () => {
++ timer = null;
++ if (stopped) return;
++ attempt().catch((err) => {
++ if (stopped) return;
++ if (++i < delays.length) { timer = setTimeout(run, delays[i]); if (timer.unref) timer.unref(); }
++ else log.warn('[Events consumer] subscriptions not created:', err.message);
++ });
++ };
++ timer = setTimeout(run, delays[0]); if (timer.unref) timer.unref();
++ return { topics, endpoint, stop() { stopped = true; if (timer) clearTimeout(timer); timer = null; } };
++}
++
++module.exports = { createEventsConsumer, startSubscriptions, RUN_TOPICS, TOPICS, EVENT_ID_RE };
ddiff --git a/server/events/outbox.js b/server/events/outbox.js
nnew file mode 100644
iindex 0000000..cf42b8b
---- /dev/null
++++ b/server/events/outbox.js
@@@ -0,0 +1,226 @@
++'use strict';
++
++/**
++ * codes.task.* events out (plan 2026-10-28 §6, decision 3): the transactional outbox the task's state changes write
++ * into, and the relay that publishes it to OpenVibe.Events.
++ *
++ * const outbox = createOutbox({ store });
++ * await store.tx(async () => { …change run_tasks…; await outbox.emitTask(row, before.state); });
++ *
++ * 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.
++ *
++ * { task_id, state, agent, repo, cost_usd, files_changed, started_at, finished_at, error }
++ */
++const { ids, serviceAuth } = 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;
++
++/** queued | running | succeeded | failed | cancelled: the five states codes.task.* announces. */
++function groupOf(state) {
++ if (state === 'queued') return 'queued';
++ if (state === 'placed' || state === 'running') return 'running';
++ if (state === 'succeeded') return 'succeeded';
++ if (state === 'failed' || state === 'expired') return 'failed';
++ if (state === 'cancelled') return 'cancelled';
++ return null;
++}
++
++function topicFor(state, previousState) {
++ const group = groupOf(state);
++ if (!group) return null;
++ if (groupOf(previousState) === group) return null; // placed → running is the same announced state
++ return `codes.task.${group}`;
++}
++
++/** owner/name from a repository URL, or null (never a host or a query string). */
++function repoSlug(url) {
++ if (typeof url !== 'string') return null;
++ try {
++ const u = new URL(url);
++ if (u.protocol !== 'https:' && u.protocol !== 'http:') return null;
++ const parts = u.pathname.replace(/^\/+|\/+$/g, '').replace(/\.git$/i, '').split('/').filter(Boolean);
++ return parts.length >= 2 ? `${parts[0]}/${parts[1]}`.slice(0, 300) : null;
++ } catch { return null; }
++}
++
++function objectOf(value) {
++ if (value && typeof value === 'object' && !Array.isArray(value)) return value;
++ if (typeof value === 'string') { try { const v = JSON.parse(value); return v && typeof v === 'object' && !Array.isArray(v) ? v : null; } catch { return null; } }
++ return null;
++}
++
++const isoOf = (value) => {
++ if (value == null) return null;
++ const t = value instanceof Date ? value.getTime() : Date.parse(String(value));
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++};
++
++/** The payload the proposal schemas describe; `state` is the announced state (placed is running). */
++function payloadOf(row) {
++ const result = objectOf(row.result) || {};
++ const state = groupOf(row.state);
++ const cost = typeof result.cost_usd === 'number' && Number.isFinite(result.cost_usd) && result.cost_usd >= 0 ? result.cost_usd : null;
++ const files = Array.isArray(result.files) ? result.files.length : null;
++ return {
++ task_id: row.id,
++ state: state || row.state,
++ agent: row.agent == null ? null : String(row.agent).slice(0, 80),
++ repo: repoSlug(row.repo),
++ cost_usd: cost,
++ files_changed: files,
++ started_at: isoOf(row.started_at),
++ finished_at: isoOf(row.finished_at),
++ error: row.error == null ? null : String(row.error).slice(0, 500),
++ };
++}
++
++function createOutbox({ store, log = console }) {
++ const db = store.db;
++
++ /** Write the state change's event in the caller's (ambient) transaction. → the evt_ id, or null. */
++ async function emitTask(row, previousState) {
++ if (!row || !row.id || !row.subject) return null;
++ const topic = topicFor(row.state, previousState);
++ if (!topic || !TOPIC_SET.has(topic)) return null;
++ const eventId = ids.newId('event', store.now());
++ const at = store.iso();
++ await db.exec(
++ `INSERT INTO codes_outbox (event_id, topic, subject, payload, created_at, next_attempt_at)
++ VALUES ($1, $2, $3, $4::jsonb, $5::timestamptz, $5::timestamptz)`,
++ [eventId, topic, row.subject, JSON.stringify(payloadOf(row)), at],
++ );
++ return eventId;
++ }
++
++ return { emitTask, payloadOf, topicFor, groupOf };
++}
++
++function createPublisher({ config, store, fetchImpl = globalThis.fetch, log = console, intervalMs = (config.events && config.events.intervalMs) || 5000 }) {
++ const db = store.db;
++ const worker = `codes-${process.pid}-${Math.random().toString(16).slice(2, 10)}`;
++ const eventsUrl = String((config.events && config.events.url) || '').replace(/\/+$/, '');
++ const configured = Boolean(eventsUrl && config.oauth && config.oauth.clientSecret);
++ const tokens = configured ? serviceAuth.createTokenClient({
++ tokenUrl: `${config.networkInternalUrl}/oauth/token`,
++ clientId: config.oauth.clientId,
++ clientSecret: config.oauth.clientSecret,
++ audience: 'openvibe.events',
++ scope: (config.events && config.events.scope) || 'events.event.publish',
++ fetch: fetchImpl,
++ }) : null;
++ const stats = { published: 0, failed: 0, last_error: null };
++ let timer = null;
++ let running = false;
++ let active = false;
++
++ function backoffMs(attempts) {
++ return Math.min(MAX_BACKOFF_MS, 5000 * Math.pow(2, Math.max(0, Math.min(6, attempts))));
++ }
++
++ async function claim(limit = 20) {
++ const now = store.iso();
++ const stale = new Date(store.now() - 2 * 60_000).toISOString();
++ return await db.many(
++ `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
++ AND (o2.claimed_at IS NULL OR o2.claimed_at < $3::timestamptz)
++ ORDER BY o2.created_at ASC, o2.event_id ASC
++ LIMIT $4
++ FOR UPDATE SKIP LOCKED
++ )
++ RETURNING o.*`,
++ [now, worker, stale, Math.min(100, Math.max(1, limit))],
++ );
++ }
++
++ function envelope(row) {
++ return {
++ event_id: row.event_id,
++ event_type: row.topic,
++ version: 1,
++ source: 'codes',
++ actor: { type: 'user', id: row.subject },
++ subject: { type: 'user', id: row.subject },
++ timestamp: store.iso(),
++ priority: 'normal',
++ visibility: 'internal',
++ payload: row.payload,
++ };
++ }
++
++ async function send(row) {
++ const headers = { 'Content-Type': 'application/json', Accept: 'application/json', ...(await tokens.authHeaders()) };
++ const res = await fetchImpl(`${eventsUrl}/api/v1/events`, {
++ method: 'POST', headers, body: JSON.stringify(envelope(row)), signal: AbortSignal.timeout(15_000),
++ });
++ if (!res.ok) {
++ const text = await res.text().catch(() => '');
++ const err = new Error(`events ${res.status}${text ? `: ${text.slice(0, 300)}` : ''}`);
++ err.status = res.status;
++ throw err;
++ }
++ await db.exec('UPDATE codes_outbox SET published_at = $1::timestamptz, claimed_at = NULL, claimed_by = NULL, last_error = NULL WHERE event_id = $2',
++ [store.iso(), row.event_id]);
++ }
++
++ async function runOnce({ limit = 20 } = {}) {
++ if (!configured) return { configured: false, claimed: 0, published: 0, failed: 0 };
++ if (running) return { configured: true, busy: true, claimed: 0, published: 0, failed: 0 };
++ running = true;
++ let claimed = 0; let published = 0; let failed = 0;
++ try {
++ const rows = await claim(limit);
++ claimed = rows.length;
++ for (const row of rows) {
++ try {
++ await send(row);
++ published++;
++ stats.published++;
++ } catch (err) {
++ failed++;
++ stats.failed++;
++ const message = String((err && err.message) || err).slice(0, 500);
++ stats.last_error = message;
++ const attempts = Number(row.attempts) + 1;
++ 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,
++ next_attempt_at = $2::timestamptz WHERE event_id = $3`,
++ [message, next, row.event_id],
++ );
++ log.warn(`[Codes] outbox ${row.event_id} publish failed (attempt ${attempts}): ${message}`);
++ }
++ }
++ return { configured: true, claimed, published, failed };
++ } finally { running = false; }
++ }
++
++ function schedule(ms) {
++ if (!configured || config.drill || !active) return;
++ timer = setTimeout(async () => {
++ timer = null;
++ try { await runOnce(); } catch (err) { log.warn('[Codes] outbox tick failed:', err.message); }
++ schedule(intervalMs);
++ }, ms);
++ if (timer.unref) timer.unref();
++ }
++
++ return {
++ 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 }),
++ stats: () => ({ ...stats }),
++ };
++}
++
++module.exports = { createOutbox, createPublisher, TOPICS, TOPIC_SET, topicFor, groupOf, payloadOf, repoSlug, MAX_BACKOFF_MS };
ddiff --git a/server/http/actions-api.js b/server/http/actions-api.js
iindex 6272ec2..139864e 100644
---- a/server/http/actions-api.js
++++ b/server/http/actions-api.js
@@@ -16,9 +16,9 @@ const items = require('../actions/items');
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const RUNNING = ['queued', 'placed', 'running'];
--function createActionsApi({ config, store, keys, run, actorLimits, log = console }) {
++function createActionsApi({ config, store, keys, run, actorLimits, tasks = null, log = console }) {
const r = asyncRouter();
-- const tasks = createRunTasks({ store, run, config, log });
++ const taskStore = tasks || createRunTasks({ store, run, config, log });
const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });
const guard = async (req, res, next) => {
@@@ -46,7 +46,7 @@ function createActionsApi({ config, store, keys, run, actorLimits, log = console
WHERE subject = $1 AND (state = ANY($2) OR updated_at >= $3)
ORDER BY updated_at DESC
LIMIT 50`, [req.principal.subject, RUNNING, cutoff]);
-- res.json({ items: rows.map((row) => items.toItem(row, tasks.present(row))) });
++ res.json({ items: rows.map((row) => items.toItem(row, taskStore.present(row))) });
});
return r;
ddiff --git a/server/http/run.js b/server/http/run.js
iindex cb4e3d5..109765d 100644
---- a/server/http/run.js
++++ b/server/http/run.js
@@@ -13,15 +13,18 @@
* places it: on a machine the person added on openvibe.bot/machines, never anyone else's (contracts 0.161.0).
*/
const express = require('express');
++const { serviceAuth, capabilities, http } = require('openvibe-contracts');
const { asyncRouter } = require('./router');
const { send } = require('../render/layout');
const { html, raw, esc } = require('../render/html');
const { parseArgs } = require('../../harness/job');
--const { createRunTasks, TERMINAL, STATE_LABEL, sessionLogOf } = require('../domain/run-tasks');
++const { createRunTasks, TERMINAL, STATE_LABEL } = require('../domain/run-tasks');
++const { createSessionLines } = require('../domain/session-lines');
const DELEGABLE = new Set(['owner', 'admin']);
const MACHINES_URL = 'https://openvibe.bot/machines';
++const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
/** A repository URL as a heading: the host small, then owner/name, which may break after a slash but never mid-word. */
function repoName(url) {
@@@ -30,7 +33,7 @@ function repoName(url) {
return html`<small class="run-host">${host}/</small>${path.map((part, i) => html`${i ? raw('/<wbr>') : ''}${part}`)}`;
}
--function createRunRoutes({ config, store, sso, harnesses, run, fetchImpl = globalThis.fetch, log = console }) {
++function createRunRoutes({ config, store, sso, harnesses, run, fetchImpl = globalThis.fetch, log = console, tasks = null, sessions = null, keys = null }) {
const r = asyncRouter();
const db = store.db;
const form = express.urlencoded({ extended: false, limit: '32kb' });
@@@ -58,9 +61,10 @@ function createRunRoutes({ config, store, sso, harnesses, run, fetchImpl = globa
const row = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [id]);
return row && row.subject === req.viewer.subject ? row : null;
};
-- // A task not yet ended asks Run where its job is; an ended job's result is kept on the row (domain/run-tasks).
-- const tasks = createRunTasks({ store, run, config, log });
-- const { who, refresh } = tasks;
++ // A task reads Run only as the 10-minute fallback; its state now arrives from run.job.* events.
++ const taskStore = tasks || createRunTasks({ store, run, config, log });
++ const sessionStore = sessions || createSessionLines({ store, config, log });
++ const { who, refresh } = taskStore;
function formHtml({ projects, values = {}, error = null }) {
const agents = harnesses.list();
@@@ -154,7 +158,7 @@ ${tasksHtml(recent)}`;
}
const projects = await projectsOf(req, res);
if (!projects.some((p) => p.id === values.project)) return runPage(req, res, { values, error: 'Choose one of your projects.', status: 422 });
-- const row = await tasks.start({ subject: req.viewer.subject, project: values.project, args });
++ const row = await taskStore.start({ subject: req.viewer.subject, project: values.project, args });
return res.redirect(303, `/run/tasks/${row.id}`);
});
@@@ -185,19 +189,44 @@ ${typeof d.duration_ms === 'number' ? html`<p class="muted">${Math.round(d.durat
row = await refresh(row);
const live = !TERMINAL.has(row.state);
// The agents' session (openvibe-shared/agent-session.js, js/run.js mounts it): live from the job's stream, and
-- // after the end from the log the result keeps (a machine on an older Codes keeps none: the summary stands).
-- const lines = live ? null : sessionLogOf(row);
++ // after the end from codes_session_lines; a task from before this change falls back to the result's old log.
++ const lines = live ? null : await sessionStore.linesForRow(row);
const title = String(row.prompt || '').split('\n')[0].slice(0, 90);
-- const session = lines ? html`<section class="run-session" data-session data-title="${title}" data-state="${row.state}"><h2>The session</h2><div data-session-view></div><script type="application/json" data-session-log>${raw(JSON.stringify(lines).replace(/</g, '\\u003c'))}</script></section>` : '';
++ const session = lines && lines.length ? html`<section class="run-session" data-session data-title="${title}" data-state="${row.state}"><h2>The session</h2><div data-session-view></div><script type="application/json" data-session-log>${raw(JSON.stringify(lines).replace(/</g, '\\u003c'))}</script></section>` : '';
const body = html`<p class="crumbs"><a href="/run">Run on your machine</a></p>
<h1 class="run-repo">${repoName(row.repo)}${row.ref ? html` <small>${row.ref}</small>` : ''}</h1>
<p><span class="run-pill run-pill-${row.state}" data-state>${STATE_LABEL[row.state] || row.state}</span>${row.agent ? html` <span class="run-pill">${row.agent}</span>` : ''}</p>
<pre tabindex="0" class="run-prompt">${row.prompt}</pre>
${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events" data-title="${title}" aria-live="polite"><h2>Live</h2><ol class="run-log" data-log></ol></section>
--<form method="post" action="/run/tasks/${row.id}/cancel"><button class="run-btn" type="submit">Cancel</button></form>` : html`${session}${resultHtml(row, { session: !!lines })}`}`;
++<form method="post" action="/run/tasks/${row.id}/cancel"><button class="run-btn" type="submit">Cancel</button></form>` : html`${session}${resultHtml(row, { session: !!(lines && lines.length) })}`}`;
return page(req, res, { title: 'Task', body, sharedScripts: ['agent-session.js'] });
});
++ /**
++ * The service half of session.json: a Network service token holding codes.task.read, acting for the person named
++ * in X-OV-Subject. Unlike the task API's writes, a read here is not limited to the service that started the task:
++ * the capability and the person are the grant.
++ */
++ async function serviceOwn(req, id) {
++ if (!keys) return { status: 503, code: 'identity.unavailable', detail: 'the Network signing key is not available' };
++ const header = String(req.headers.authorization || '');
++ if (!header.startsWith('Bearer ')) return { status: 401, code: 'token.required', detail: 'codes.task.read needs a Network service token or a signed-in owner' };
++ if (!keys.loaded()) await keys.ensure();
++ const publicKey = keys.get();
++ if (!publicKey) return { status: 503, code: 'identity.unavailable', detail: 'the Network signing key is not loaded yet' };
++ const v = serviceAuth.verifyServiceToken(header.slice(7).trim(), { publicKey, issuer: config.networkIssuer, audience: config.serviceAudience });
++ if (!v.ok) return { status: 401, code: v.code || 'token.invalid', detail: v.reason || 'the service token is invalid' };
++ const sub = String(v.claims.sub || '');
++ if (!sub.startsWith('svc:')) return { status: 403, code: 'capability.denied', detail: 'codes.task.read is for first-party services' };
++ const decision = capabilities.check(v.claims, 'codes.task.read');
++ if (!decision.allowed) return { status: 403, code: decision.code || 'capability.denied', detail: decision.reason || 'codes.task.read not granted' };
++ const subject = String(req.get('X-OV-Subject') || '').trim();
++ if (!USER.test(subject)) return { status: 422, code: 'subject.required', detail: 'name the person in X-OV-Subject (usr_…)' };
++ if (!/^tsk_[0-9A-HJKMNP-TV-Z]{26}$/.test(String(id))) return { status: 404, code: 'codes.task_not_found', detail: 'no such task for this person' };
++ const row = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 AND subject = $2', [id, subject]);
++ return row ? { row } : { status: 404, code: 'codes.task_not_found', detail: 'no such task for this person' };
++ }
++
r.get('/run/tasks/:id/events', async (req, res) => {
if (!signedIn(req)) return res.status(401).end();
const row = await own(req, req.params.id);
@@@ -212,13 +241,46 @@ ${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events
} catch { return res.status(503).end(); }
if (!resp.ok || !resp.body) return res.status(resp.status === 404 ? 404 : 503).end();
res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-store', 'X-Accel-Buffering': 'no', Connection: 'keep-alive' });
++ // The bytes go to the browser unchanged, and the same frames become codes_session_lines as they stream.
++ const recorder = sessionStore.recorder(row.id);
++ const decoder = new TextDecoder();
try {
-- for await (const chunk of resp.body) { if (!res.write(chunk)) await new Promise((ok) => res.once('drain', ok)); }
++ for await (const chunk of resp.body) {
++ if (!res.write(chunk)) await new Promise((ok) => res.once('drain', ok));
++ try { await recorder.push(decoder.decode(chunk, { stream: true })); } catch (e) { log.warn(`[Codes] session ${row.id}: ${e.message}`); }
++ }
} catch { /* the browser left, or Run closed the stream */ }
++ try { await recorder.push(decoder.decode()); await recorder.flush(); } catch (e) { log.warn(`[Codes] session ${row.id}: ${e.message}`); }
++ // The stream's end is the machine's result arriving: fetch it once, finalize the row and back-fill the log.
++ if (recorder.ended()) {
++ try {
++ const fresh = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);
++ if (fresh) await taskStore.refresh(fresh, { force: true });
++ } catch (e) { log.warn(`[Codes] task ${row.id}: ${e.message}`); }
++ }
res.end();
return undefined;
});
++ /** Session replay: the same page's lines, after a seq, for the owner or a service with codes.task.read. */
++ r.get('/run/tasks/:id/session.json', async (req, res) => {
++ let row = null;
++ if (signedIn(req)) {
++ row = await own(req, req.params.id);
++ } else {
++ const out = await serviceOwn(req, req.params.id);
++ if (!out.row) return http.sendProblem(res, out.status, out.code, { detail: out.detail, ctx: req.ov });
++ row = out.row;
++ }
++ if (!row) return http.sendProblem(res, 404, 'codes.task_not_found', { detail: 'no such task of yours', ctx: req.ov });
++ const raw = String(req.query.after == null ? '0' : req.query.after);
++ const after = /^\d+$/.test(raw) ? Number(raw) : NaN;
++ if (!Number.isSafeInteger(after)) return http.sendProblem(res, 422, 'codes.after_invalid', { detail: 'after is a line sequence number (0 or more)', ctx: req.ov });
++ const replay = await sessionStore.replay(row, { after, limit: 500 });
++ res.set('Cache-Control', 'no-store');
++ return res.json(replay);
++ });
++
r.post('/run/tasks/:id/cancel', form, async (req, res) => {
if (!signedIn(req)) return res.redirect(303, '/auth/login?next=%2Frun');
if (!sameOrigin(req)) return res.status(403).type('text/plain').send('Cross-site form posts are refused.\n');
@@@ -226,6 +288,10 @@ ${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events
if (!row) return res.status(404).type('text/plain').send('No such task of yours.\n');
if (row.job_id && !TERMINAL.has(row.state)) {
try { await run.cancel(row.job_id, who(row)); } catch (e) { log.warn(`[Codes] cancel ${row.id}: ${e.message}`); }
++ try {
++ const fresh = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);
++ if (fresh) await taskStore.refresh(fresh, { force: true });
++ } catch (e) { log.warn(`[Codes] cancel ${row.id}: ${e.message}`); }
}
return res.redirect(303, `/run/tasks/${row.id}`);
});
ddiff --git a/server/http/tasks-api.js b/server/http/tasks-api.js
iindex 52f37bd..7d84ed8 100644
---- a/server/http/tasks-api.js
++++ b/server/http/tasks-api.js
@@@ -24,10 +24,10 @@ const { createRunTasks, TERMINAL } = require('../domain/run-tasks');
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
--function createTasksApi({ config, store, keys, run, harnesses, actorLimits, log = console }) {
++function createTasksApi({ config, store, keys, run, harnesses, actorLimits, tasks = null, log = console }) {
const r = asyncRouter();
const db = store.db;
-- const tasks = createRunTasks({ store, run, config, log });
++ const taskStore = tasks || createRunTasks({ store, run, config, log });
const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });
const json = express.json({ limit: '32kb' });
@@@ -66,7 +66,7 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, log
const key = b.idempotency_key || null;
if (key) {
const seen = await byKey(p, key);
-- if (seen) return res.status(200).json({ task: tasks.present(await tasks.refresh(seen)), created: false });
++ if (seen) return res.status(200).json({ task: taskStore.present(await taskStore.refresh(seen)), created: false });
}
let args;
try {
@@@ -77,37 +77,37 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, log
} catch (e) {
return problem(req, res, 422, 'codes.task_invalid', e.message);
}
-- const project = await tasks.latestProject(p.subject);
++ const project = await taskStore.latestProject(p.subject);
if (!project) {
return problem(req, res, 409, 'codes.no_project', `this person has not run a task on ${config.baseUrl}/run yet: a service's task runs in the project of their latest one there, on a machine of theirs (openvibe.bot/machines)`);
}
let row;
try {
-- row = await tasks.start({ subject: p.subject, project, args, via: p.via, key });
++ row = await taskStore.start({ subject: p.subject, project, args, via: p.via, key });
} catch (e) {
// Two requests with one key at once: the unique index lets one in; the other answers with it.
const seen = key && e && e.code === '23505' ? await byKey(p, key) : null;
if (!seen) throw e;
-- return res.status(200).json({ task: tasks.present(seen), created: false });
++ return res.status(200).json({ task: taskStore.present(seen), created: false });
}
log.info(`[Codes] task ${row.id} started by ${p.via} for ${p.subject}: ${row.state}`);
-- return res.status(201).json({ task: tasks.present(row), created: true });
++ return res.status(201).json({ task: taskStore.present(row), created: true });
});
r.get('/tasks/:id', guard('codes.task.read'), actorLimits.reads('codes.read'), async (req, res) => {
const row = await mine(req);
if (!row) return problem(req, res, 404, 'codes.task_not_found', 'no such task of this service for this person');
-- return res.json(tasks.present(await tasks.refresh(row)));
++ return res.json(taskStore.present(await taskStore.refresh(row)));
});
r.post('/tasks/:id/cancel', guard('codes.task.cancel'), actorLimits.budget('codes.task.start'), async (req, res) => {
let row = await mine(req);
if (!row) return problem(req, res, 404, 'codes.task_not_found', 'no such task of this service for this person');
if (row.job_id && !TERMINAL.has(row.state)) {
-- try { await run.cancel(row.job_id, tasks.who(row)); } catch (e) { log.warn(`[Codes] cancel ${row.id}: ${e.message}`); }
-- row = await tasks.refresh(row);
++ try { await run.cancel(row.job_id, taskStore.who(row)); } catch (e) { log.warn(`[Codes] cancel ${row.id}: ${e.message}`); }
++ row = await taskStore.refresh(row, { force: true });
}
-- return res.json(tasks.present(row));
++ return res.json(taskStore.present(row));
});
return r;
ddiff --git a/server/index.js b/server/index.js
iindex 03e020f..2d2401f 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -7,17 +7,24 @@
const { createApp } = require('./app');
const { gracefulStop, telemetry } = require('openvibe-sdk/service');
const { createPgTelemetrySink, startTelemetryRetention } = require('openvibe-sdk/telemetry');
--const { startSubscriptions } = require('openvibe-sdk/account-data');
++const { startSubscriptions } = require('./events-consumer');
/**
* The process stop (openvibe-sdk/service, docs/service.md's 5 s family): the HTTP drain runs
-- * (defaults 4000/5000), then the JWKS refresher stops and the store closes; past the deadline the process
-- * exits 0, as the hand-rolled 5 s timer did. Exported so a test can inject `exit` and `signals: false`.
++ * (defaults 4000/5000), then the background loops, the JWKS refresher and the store close; past the deadline the
++ * process exits 0, as the hand-rolled 5 s timer did. Exported so a test can inject `exit` and `signals: false`.
*/
function createLifecycle({ server, ctx, exit, signals }) {
return gracefulStop({
name: 'Codes', server, deadlineExitCode: 0, exit, signals,
-- close: [() => ctx.keys.client.stop(), async () => { if (ctx.telemetryRetention) await ctx.telemetryRetention.stop(); await telemetry.stop(); }, () => ctx.store.close()],
++ close: [
++ () => { if (ctx.subscriptions) ctx.subscriptions.stop(); },
++ () => { if (ctx.publisher) return ctx.publisher.stop(); return undefined; },
++ () => { if (ctx.sessionRetention) ctx.sessionRetention.stop(); },
++ () => ctx.keys.client.stop(),
++ async () => { if (ctx.telemetryRetention) await ctx.telemetryRetention.stop(); await telemetry.stop(); },
++ () => ctx.store.close(),
++ ],
});
}
@@@ -37,12 +44,14 @@ async function start() {
// Keep the JWKS cache fresh in the background (openvibe-sdk/auth): one client per URL, the last
// good keys through outages, exponential backoff, unknown-kid floods throttled, an unref'd timer.
ctx.keys.client.start();
-- // The two account subscriptions at OpenVibe.Events (ADR-033), created when missing; off without EVENTS_URL,
-- // CODES_EVENTS_SECRET or the client secret.
-- startSubscriptions({
-- eventsUrl: config.events.url, endpoint: `http://127.0.0.1:${config.port}/internal/events`, secret: config.events.secrets[0],
-- networkInternalUrl: config.networkInternalUrl, clientId: config.oauth.clientId, clientSecret: config.oauth.clientSecret,
++ // The run.job.* and account subscriptions at OpenVibe.Events, created when missing; off without EVENTS_URL,
++ // CODES_EVENTS_SECRET or the client secret (and under CODES_DRILL).
++ ctx.subscriptions = startSubscriptions({
++ config, port: config.port, secret: config.events.secrets[0] || '',
});
++ // codes.task.* events: the outbox relay and the session-retention loop (both off under CODES_DRILL).
++ ctx.publisher.start();
++ ctx.sessionRetention.start();
createLifecycle({ server, ctx });
return { server, ctx };
ddiff --git a/server/utils/jobs.js b/server/utils/jobs.js
nnew file mode 100644
iindex 0000000..f5f5a15
---- /dev/null
++++ b/server/utils/jobs.js
@@@ -0,0 +1,44 @@
++'use strict';
++
++/**
++ * The small background-job loop Codes uses for retention (the shape the network's services share): every() runs a
++ * function on a timeout chain, never overlapping a slow run, and contains a throw so a database hiccup cannot take
++ * the process down. Nothing starts unless start() is called; a drill flag keeps every loop off in a drill.
++ *
++ * const stop = every('codes-session-retention', 24 * 60 * 60_000, run, { initialDelayMs: 60_000 });
++ */
++const timers = new Set();
++
++function every(name, intervalMs, fn, { initialDelayMs = intervalMs, jitterMs = 0, log = console } = {}) {
++ let active = true;
++ let handle = null;
++ let busy = false;
++ const jitter = () => (jitterMs > 0 ? Math.floor(Math.random() * jitterMs) : 0);
++ const schedule = (delay) => {
++ if (!active) return;
++ handle = setTimeout(async () => {
++ timers.delete(handle);
++ handle = null;
++ if (!active) return;
++ if (!busy) {
++ busy = true;
++ try { await fn(); } catch (err) { log.warn(`[Codes] job ${name} failed: ${(err && err.message) || err}`); } finally { busy = false; }
++ }
++ schedule(intervalMs + jitter());
++ }, delay);
++ if (handle.unref) handle.unref();
++ timers.add(handle);
++ };
++ schedule(initialDelayMs + jitter());
++ return () => {
++ active = false;
++ if (handle) { clearTimeout(handle); timers.delete(handle); handle = null; }
++ };
++}
++
++function stopAll() {
++ for (const t of timers) clearTimeout(t);
++ timers.clear();
++}
++
++module.exports = { every, stopAll };
ddiff --git a/test/contracts.test.js b/test/contracts.test.js
iindex d821b63..15891fd 100644
---- a/test/contracts.test.js
++++ b/test/contracts.test.js
@@@ -1,8 +1,9 @@
'use strict';
/**
* The released contracts (the pinned openvibe-contracts tag) describe what the code does. Since 0.113.0 the codes
-- * service manifest declares the coding-task capabilities (codes.task.*, Contracts 0.162) and no event: the developer console moved to OpenVibe.Services, which owns
-- * services.release.* and services.app.*, and the codes.* forms are retired. The harness catalog's rows are the
++ * service manifest declares the coding-task capabilities (codes.task.*, Contracts 0.162): the developer console moved to
++ * OpenVibe.Services, which owns services.release.* and services.app.*, and the codes.* forms are retired. Since 0.232.0
++ * it produces codes.task.queued|running|succeeded|failed|cancelled and consumes run.job.*. The harness catalog's rows are the
* contracts' platform.harness-offer@1 and platform.agent-offer@1 (checked at boot by server/domain/harnesses.js).
*/
const assert = require('assert');
@@@ -14,7 +15,7 @@ const { check, done } = require('./helpers/boot');
(async () => {
const manifest = contracts.services.get('codes');
-- await check('the codes service manifest is released as alpha, live, with the coding-task capabilities and no event', async () => {
++ await check('the codes service manifest is released as alpha, live, with the coding-task capabilities', async () => {
assert.ok(manifest);
assert.strictEqual(manifest.status, 'alpha');
assert.deepStrictEqual(manifest.domains, ['openvibe.codes']);
@@@ -23,7 +24,6 @@ const { check, done } = require('./helpers/boot');
// and reads their Action Center items (http/actions-api.js).
assert.deepStrictEqual(manifest.capabilities, ['codes.task.start', 'codes.task.read', 'codes.task.cancel', 'codes.actions.read']);
for (const id of manifest.capabilities) assert.strictEqual(contracts.capabilities.get(id).status, 'active', id);
-- assert.deepStrictEqual(manifest.eventsProduced, []);
const range = manifest.contractRanges['openvibe-contracts'];
assert.ok(require('openvibe-sdk/core').satisfiesRange(require('openvibe-contracts/package.json').version, range));
});
@@@ -48,11 +48,31 @@ const { check, done } = require('./helpers/boot');
}
});
-- await check('no proposal left behind: everything proposed is released', async () => {
-- const docs = path.join(__dirname, '..', 'docs');
-- assert.ok(!fs.existsSync(path.join(docs, 'capabilities-proposal')));
-- assert.ok(!fs.existsSync(path.join(docs, 'service-manifest-proposal.json')));
-- assert.ok(!fs.existsSync(path.join(docs, 'contracts-proposal')));
++ await check('every codes.task.* payload validates against the pinned contracts, and the manifest names the topics', async () => {
++ const contracts = require('openvibe-contracts');
++ const { TOPICS, payloadOf } = require('../server/events/outbox');
++ const row = { id: 'tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2A3', agent: 'claude', repo: 'https://github.com/OpenVibers/OpenVibe.Codes.git',
++ started_at: '2026-10-28T16:00:00.000Z', finished_at: null, error: null, result: null };
++ const rows = [
++ { ...row, state: 'queued', started_at: null },
++ { ...row, state: 'placed' },
++ { ...row, state: 'succeeded', finished_at: new Date('2026-10-28T16:04:00Z'), result: { cost_usd: 0.42, files: ['a.js', 'b.js'] } },
++ { ...row, state: 'expired', finished_at: '2026-10-28T17:00:00.000Z', error: 'run.job_expired: no machine took the job' },
++ { ...row, state: 'cancelled', repo: 'not a url', agent: null },
++ ];
++ for (const r of rows) {
++ const payload = payloadOf(r);
++ const topic = `codes.task.${payload.state}`;
++ assert.ok(TOPICS.includes(topic), topic);
++ const v = contracts.validate(topic, payload);
++ assert.ok(v.valid, `${topic}: ${JSON.stringify(v.errors)}`);
++ }
++ assert.strictEqual(payloadOf(rows[2]).repo, 'OpenVibers/OpenVibe.Codes');
++ assert.strictEqual(payloadOf(rows[2]).files_changed, 2);
++ const manifest = require('openvibe-contracts/manifests/services/codes.json');
++ assert.deepStrictEqual(manifest.eventsProduced, TOPICS);
++ assert.ok(manifest.eventsConsumed.includes('run.job.succeeded') && manifest.eventsConsumed.includes('network.account.deleted'));
++ assert.ok(!fs.existsSync(path.join(__dirname, '..', 'docs', 'contracts-proposal')), 'the proposal moved to Contracts 0.232.0');
});
done();
ddiff --git a/test/events-consumer.test.js b/test/events-consumer.test.js
nnew file mode 100644
iindex 0000000..5f1eb08
---- /dev/null
++++ b/test/events-consumer.test.js
@@@ -0,0 +1,130 @@
++'use strict';
++/**
++ * run.job.* → Codes (plan 2026-10-28 §6, decision 2): a signed delivery for a job that is a Codes task finalizes the
++ * run_tasks row once (state, timings, result read once from Run), in the same transaction as the inbox claim; a replay
++ * changes nothing; a bad signature is refused; a job that is not a Codes task is ignored.
++ */
++const assert = require('assert');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { parseArgs } = require('../harness/job');
++const { boot, check, done } = require('./helpers/boot');
++
++const SECRET = `fixture_${'consumer'.repeat(6)}`;
++const JOB_A = 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1';
++const JOB_B = 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J2';
++
++function eventOf(id, type, jobId, subject, project, state, timings = {}) {
++ return {
++ event_id: id, event_type: type, version: 1, source: 'run',
++ subject: { type: 'job', id: jobId }, visibility: 'internal',
++ payload: {
++ job_id: jobId, project_id: project, requester: { type: 'user', id: subject }, state,
++ timings: {
++ created_at: new Date(Date.now() - 5000).toISOString(), placed_at: new Date(Date.now() - 4000).toISOString(),
++ started_at: timings.started_at || new Date(Date.now() - 3000).toISOString(),
++ finished_at: timings.finished_at || null,
++ },
++ error: null,
++ },
++ };
++}
++
++(async () => {
++ const reads = [];
++ let readState = 'succeeded';
++ let readResult = null;
++ const run = {
++ enabled: true,
++ async submit() { return { id: JOB_A, state: 'queued' }; },
++ async read(jobId) {
++ reads.push(jobId);
++ return {
++ id: jobId, state: readState, result: readResult, error: null,
++ timings: { started_at: new Date(Date.now() - 3000).toISOString(), finished_at: new Date().toISOString() },
++ };
++ },
++ async cancel() { return { state: 'cancelled' }; },
++ async streamUrl() { throw new Error('not used'); },
++ };
++ const t = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET } });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++ const args = parseArgs(JSON.stringify({ repo: 'https://github.com/Octocat/Hello-World', prompt: 'Fix the typo', permission: 'edit' }));
++ const post = (body) => t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
++ const deliver = (event) => post(JSON.stringify({ event, seq: 1 }));
++
++ try {
++ const task = await t.ctx.tasks.start({ subject: ada.subject, project, args });
++ readResult = { ok: true, summary: 'Fixed', cost_usd: 0.02, files: [{ path: 'README', added: 1, removed: 1 }], log: [{ t: 'text', text: 'done' }, { t: 'done', ok: true, text: 'Fixed' }] };
++
++ await check('a signed run.job.started moves the task to running without reading Run', async () => {
++ const started = eventOf('evt_01M61F7R3WXM5T5ET9CRGNHEP1', 'run.job.started', JOB_A, ada.subject, project, 'running');
++ const r = await deliver(started);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json().outcome, 'running');
++ assert.strictEqual(r.json().task_id, task.id);
++ const row = await db.one('SELECT state, started_at FROM run_tasks WHERE id = $1', [task.id]);
++ assert.strictEqual(row.state, 'running');
++ assert.ok(row.started_at);
++ assert.strictEqual(reads.length, 0, 'no result is read for a non-terminal event');
++ assert.strictEqual(await db.value("SELECT COUNT(*)::int FROM codes_outbox WHERE topic = 'codes.task.running'"), 1);
++ });
++
++ await check('a signed run.job.succeeded finalizes the task once, with its result and one codes.task.succeeded row', async () => {
++ const succeeded = eventOf('evt_01M61F7R3WXM5T5ET9CRGNHEP2', 'run.job.succeeded', JOB_A, ada.subject, project, 'succeeded',
++ { finished_at: new Date().toISOString() });
++ const r = await deliver(succeeded);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json().outcome, 'succeeded');
++ assert.strictEqual(reads.length, 1, 'the result was read once');
++ const row = await db.one('SELECT state, result, finished_at, session_complete_at FROM run_tasks WHERE id = $1', [task.id]);
++ assert.strictEqual(row.state, 'succeeded');
++ assert.strictEqual(row.result.summary, 'Fixed');
++ assert.ok(row.finished_at);
++ assert.ok(row.session_complete_at);
++ const line = await db.maybe("SELECT * FROM codes_session_lines WHERE task_id = $1 AND kind = 'done'", [task.id]);
++ assert.ok(line, 'the result log was back-filled');
++ const outbox = await db.many('SELECT topic, payload FROM codes_outbox WHERE payload->>\'task_id\' = $1 ORDER BY created_at', [task.id]);
++ assert.deepStrictEqual(outbox.map((x) => x.topic), ['codes.task.queued', 'codes.task.running', 'codes.task.succeeded']);
++ const payload = outbox[2].payload;
++ assert.strictEqual(payload.task_id, task.id);
++ assert.strictEqual(payload.state, 'succeeded');
++ assert.strictEqual(payload.repo, 'Octocat/Hello-World');
++ assert.strictEqual(payload.cost_usd, 0.02);
++ assert.strictEqual(payload.files_changed, 1);
++ assert.ok(payload.started_at && payload.finished_at);
++ assert.strictEqual(payload.error, null);
++ });
++
++ await check('a replayed delivery changes nothing', async () => {
++ const before = await db.one('SELECT state, updated_at FROM run_tasks WHERE id = $1', [task.id]);
++ const count = await db.value('SELECT COUNT(*)::int FROM codes_outbox');
++ const succeeded = eventOf('evt_01M61F7R3WXM5T5ET9CRGNHEP2', 'run.job.succeeded', JOB_A, ada.subject, project, 'succeeded',
++ { finished_at: new Date().toISOString() });
++ const r = await deliver(succeeded);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json().duplicate, true);
++ const after = await db.one('SELECT state, updated_at FROM run_tasks WHERE id = $1', [task.id]);
++ assert.strictEqual(after.state, before.state);
++ assert.strictEqual(Number(after.updated_at), Number(before.updated_at));
++ assert.strictEqual(await db.value('SELECT COUNT(*)::int FROM codes_outbox'), count);
++ assert.strictEqual(await db.value('SELECT COUNT(*)::int FROM codes_events_inbox WHERE event_id = $1', ['evt_01M61F7R3WXM5T5ET9CRGNHEP2']), 1);
++ assert.strictEqual(reads.length, 1, 'a replay does not read Run again');
++ });
++
++ await check('a bad signature is refused; a job that is not a Codes task is ignored', async () => {
++ const body = JSON.stringify({ event: eventOf('evt_01M61F7R3WXM5T5ET9CRGNHEP3', 'run.job.failed', JOB_A, ada.subject, project, 'failed'), seq: 1 });
++ const bad = await t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, 'wrong'.repeat(8)) } });
++ assert.strictEqual(bad.status, 401);
++ const other = eventOf('evt_01M61F7R3WXM5T5ET9CRGNHEP4', 'run.job.succeeded', JOB_B, ada.subject, project, 'succeeded');
++ const r = await deliver(other);
++ assert.strictEqual(r.status, 200, r.text);
++ assert.strictEqual(r.json().outcome, 'ignored:not_a_codes_task');
++ assert.strictEqual(await db.value('SELECT COUNT(*)::int FROM codes_events_inbox WHERE event_id = $1', [other.event_id]), 1);
++ });
++ } finally {
++ await t.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/no-polling.test.js b/test/no-polling.test.js
nnew file mode 100644
iindex 0000000..b1c688e
---- /dev/null
++++ b/test/no-polling.test.js
@@@ -0,0 +1,84 @@
++'use strict';
++/**
++ * No page-view polling (plan 2026-10-28 §6, decision 2): a task page view does not call Run while the task got its
++ * run.job.* events; Run is read once only as the fallback after ten minutes without an event.
++ */
++const assert = require('assert');
++const { signDeliveryHeaders } = require('openvibe-sdk/events');
++const { parseArgs } = require('../harness/job');
++const { boot, check, done } = require('./helpers/boot');
++
++const SECRET = `fixture_${'no-polling'.repeat(4)}`;
++const NOW = Date.now();
++
++function event(id, type, jobId, subject, project, state, timings = {}) {
++ return {
++ event_id: id, event_type: type, version: 1, source: 'run', subject: { type: 'job', id: jobId }, visibility: 'internal',
++ payload: {
++ job_id: jobId, project_id: project, requester: { type: 'user', id: subject }, state,
++ timings: { created_at: new Date(NOW).toISOString(), placed_at: new Date(NOW).toISOString(), started_at: new Date(NOW).toISOString(), finished_at: timings.finished_at || null },
++ error: null,
++ },
++ };
++}
++
++(async () => {
++ const reads = [];
++ let nextJob = 1;
++ const states = new Map();
++ const run = {
++ enabled: true,
++ async submit() { const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${nextJob++}`; states.set(id, { state: 'queued', result: null }); return { id, state: 'queued' }; },
++ async read(jobId) {
++ reads.push(jobId);
++ const s = states.get(jobId) || { state: 'running', result: null };
++ return { id: jobId, state: s.state, result: s.result, error: null, timings: { started_at: new Date(NOW).toISOString(), finished_at: s.state === 'succeeded' ? new Date().toISOString() : null } };
++ },
++ async cancel() { return { state: 'cancelled' }; },
++ async streamUrl() { throw new Error('not used'); },
++ };
++ const t = await boot({ run, env: { CODES_EVENTS_SECRET: SECRET } });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++ const args = parseArgs(JSON.stringify({ repo: 'https://github.com/Octocat/Hello-World', prompt: 'Fix the typo', permission: 'edit' }));
++ const deliver = async (e) => {
++ const body = JSON.stringify({ event: e, seq: 1 });
++ return await t.get('/internal/events', { method: 'POST', body, headers: { 'content-type': 'application/json', ...signDeliveryHeaders(body, SECRET) } });
++ };
++
++ try {
++ await check('events finalize a task; its page view then reads nothing from Run', async () => {
++ const task = await t.ctx.tasks.start({ subject: ada.subject, project, args });
++ const jobId = task.job_id;
++ states.set(jobId, { state: 'succeeded', result: { ok: true, summary: 'Fixed', log: [{ t: 'done', ok: true, text: 'Fixed' }] } });
++ assert.strictEqual((await deliver(event('evt_01M61F7R3WXM5T5ET9CRGNHEP5', 'run.job.started', jobId, ada.subject, project, 'running'))).status, 200);
++ assert.strictEqual((await deliver(event('evt_01M61F7R3WXM5T5ET9CRGNHEP6', 'run.job.succeeded', jobId, ada.subject, project, 'succeeded', { finished_at: new Date().toISOString() }))).status, 200);
++ assert.strictEqual(await db.value('SELECT state FROM run_tasks WHERE id = $1', [task.id]), 'succeeded');
++ reads.length = 0;
++ const page = await t.get(`/run/tasks/${task.id}`, { as: ada });
++ assert.strictEqual(page.status, 200);
++ assert.ok(page.text.includes('Finished') && page.text.includes('Fixed'), page.text.slice(-500));
++ assert.strictEqual(reads.length, 0, 'the page did not call Run');
++ });
++
++ await check('a running task is not polled on page views before ten quiet minutes, then read once', async () => {
++ const task = await t.ctx.tasks.start({ subject: ada.subject, project, args });
++ states.set(task.job_id, { state: 'running', result: null });
++ assert.strictEqual((await deliver(event('evt_01M61F7R3WXM5T5ET9CRGNHEP7', 'run.job.started', task.job_id, ada.subject, project, 'running'))).status, 200);
++ reads.length = 0;
++ const first = await t.get(`/run/tasks/${task.id}`, { as: ada });
++ assert.strictEqual(first.status, 200);
++ assert.ok(first.text.includes('data-events'));
++ assert.strictEqual(reads.length, 0, 'no page-view poll');
++ await db.exec('UPDATE run_tasks SET updated_at = updated_at - $2 WHERE id = $1', [task.id, 11 * 60_000]);
++ const fallback = await t.get(`/run/tasks/${task.id}`, { as: ada });
++ assert.strictEqual(fallback.status, 200);
++ assert.strictEqual(reads.length, 1, 'one fallback read after ten quiet minutes');
++ assert.deepStrictEqual(reads, [task.job_id]);
++ });
++ } finally {
++ await t.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/outbox.test.js b/test/outbox.test.js
nnew file mode 100644
iindex 0000000..621484f
---- /dev/null
++++ b/test/outbox.test.js
@@@ -0,0 +1,133 @@
++'use strict';
++/**
++ * codes.task.* events out (plan 2026-10-28 §6, decision 3): every state change writes one codes_outbox row in the
++ * same transaction, the publisher sends each under the evt_ id fixed at insert, a failure records the attempt and
++ * backs off, a retry and a new publisher over the same rows never make a second event.
++ */
++const assert = require('assert');
++const http = require('http');
++const { parseArgs } = require('../harness/job');
++const { boot, check, done } = require('./helpers/boot');
++const { createPublisher } = require('../server/events/outbox');
++
++(async () => {
++ const received = [];
++ let failNext = 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"}'); }
++ const event = JSON.parse(body);
++ received.push(event);
++ res.writeHead(201, { 'Content-Type': 'application/json' });
++ return res.end(JSON.stringify({ event_id: event.event_id, cursor: `c${received.length}` }));
++ });
++ });
++ await new Promise((ok) => events.listen(0, '127.0.0.1', ok));
++
++ let readState = 'queued';
++ let readResult = null;
++ const run = {
++ enabled: true,
++ async submit() { return { id: 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J1', state: 'queued' }; },
++ async read(jobId) {
++ return { id: jobId, state: readState, result: readResult, error: null, timings: { started_at: new Date(Date.now() - 1000).toISOString(), finished_at: readState === 'succeeded' ? new Date().toISOString() : null } };
++ },
++ async cancel() { return { state: 'cancelled' }; },
++ async streamUrl() { throw new Error('not used'); },
++ };
++ const t = await boot({ run, env: { EVENTS_URL: `http://127.0.0.1:${events.address().port}` } });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++ const args = parseArgs(JSON.stringify({ repo: 'https://github.com/Octocat/Hello-World', prompt: 'Fix the typo', permission: 'edit' }));
++
++ try {
++ const task = await t.ctx.tasks.start({ subject: ada.subject, project, args });
++
++ await check('a task start writes its queued row in the change transaction, with the fixed evt_ id', async () => {
++ const rows = await db.many('SELECT * FROM codes_outbox WHERE payload->>\'task_id\' = $1', [task.id]);
++ assert.strictEqual(rows.length, 1);
++ const row = rows[0];
++ assert.match(row.event_id, /^evt_[0-9A-HJKMNP-TV-Z]{26}$/);
++ assert.strictEqual(row.topic, 'codes.task.queued');
++ assert.strictEqual(row.subject, ada.subject);
++ assert.deepStrictEqual(Object.keys(row.payload).sort(), ['agent', 'cost_usd', 'error', 'files_changed', 'finished_at', 'repo', 'started_at', 'state', 'task_id']);
++ assert.strictEqual(row.payload.repo, 'Octocat/Hello-World');
++ assert.strictEqual(row.payload.state, 'queued');
++ const before = await db.value('SELECT COUNT(*)::int FROM codes_outbox');
++ await assert.rejects(t.ctx.store.tx(async () => {
++ await t.ctx.outbox.emitTask({ ...task, state: 'succeeded', result: { ok: true } }, 'running');
++ throw new Error('roll back the change');
++ }), /roll back/);
++ assert.strictEqual(await db.value('SELECT COUNT(*)::int FROM codes_outbox'), before, 'the event is not written without the change');
++ });
++
++ await check('a Run state change writes one row in the same transaction', async () => {
++ readState = 'succeeded';
++ readResult = { ok: true, cost_usd: 0.02, files: [{ path: 'README', added: 1, removed: 1 }], log: [{ t: 'done', ok: true }] };
++ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [task.id]);
++ const done_ = await t.ctx.tasks.refresh(row, { force: true });
++ assert.strictEqual(done_.state, 'succeeded');
++ const rows = await db.many('SELECT topic, payload FROM codes_outbox WHERE payload->>\'task_id\' = $1 ORDER BY created_at, event_id', [task.id]);
++ assert.deepStrictEqual(rows.map((x) => x.topic), ['codes.task.queued', 'codes.task.succeeded']);
++ assert.strictEqual(rows[1].payload.state, 'succeeded');
++ assert.strictEqual(rows[1].payload.cost_usd, 0.02);
++ assert.strictEqual(rows[1].payload.files_changed, 1);
++ });
++
++ await check('publishing sends each row once, with the fixed id and the proposal envelope', async () => {
++ const rows = await db.many('SELECT * FROM codes_outbox ORDER BY created_at, event_id');
++ assert.strictEqual(await t.ctx.publisher.pending(), rows.length);
++ const r = await t.ctx.publisher.runOnce();
++ assert.deepStrictEqual(r, { configured: true, claimed: rows.length, published: rows.length, failed: 0 });
++ assert.strictEqual(received.length, rows.length);
++ for (const row of rows) {
++ const event = received.find((x) => x.event_id === row.event_id);
++ assert.ok(event, `published ${row.event_id}`);
++ assert.strictEqual(event.event_type, row.topic);
++ assert.strictEqual(event.source, 'codes');
++ assert.deepStrictEqual(event.subject, { type: 'user', id: ada.subject });
++ assert.strictEqual(event.visibility, 'internal');
++ assert.deepStrictEqual(event.payload, row.payload);
++ }
++ assert.strictEqual((await t.ctx.publisher.runOnce()).claimed, 0);
++ assert.strictEqual(received.length, rows.length, 'nothing sent twice');
++ assert.strictEqual(await t.ctx.publisher.pending(), 0);
++ });
++
++ 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]);
++ failNext = 1;
++ const failed = await t.ctx.publisher.runOnce();
++ assert.strictEqual(failed.failed, 1);
++ let after = await db.one('SELECT * FROM codes_outbox WHERE event_id = $1', [row.event_id]);
++ assert.strictEqual(after.attempts, 1);
++ assert.match(after.last_error, /events 503/);
++ assert.strictEqual(after.published_at, null);
++ assert.ok(new Date(after.next_attempt_at).getTime() > Date.now(), 'backed off into the future');
++ assert.strictEqual((await t.ctx.publisher.runOnce()).claimed, 0, 'not due yet');
++ await db.exec("UPDATE codes_outbox SET next_attempt_at = now() - interval '1 second' WHERE event_id = $1", [row.event_id]);
++ const retried = await t.ctx.publisher.runOnce();
++ assert.deepStrictEqual(retried, { configured: true, claimed: 1, published: 1, failed: 0 });
++ after = await db.one('SELECT * FROM codes_outbox WHERE event_id = $1', [row.event_id]);
++ assert.ok(after.published_at);
++ assert.strictEqual(after.attempts, 1, 'the failed attempt stays recorded');
++ assert.strictEqual(received.filter((x) => x.event_id === row.event_id).length, 1, 'one event under the fixed id');
++ });
++
++ 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() {} } });
++ assert.strictEqual((await restarted.runOnce()).claimed, 0);
++ assert.strictEqual(received.length, before);
++ assert.strictEqual(await t.ctx.publisher.pending(), 0);
++ });
++ } finally {
++ await t.close();
++ events.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/run-pages.test.js b/test/run-pages.test.js
iindex 2dc2652..b6519b3 100644
---- a/test/run-pages.test.js
++++ b/test/run-pages.test.js
@@@ -85,6 +85,8 @@ const { boot, check, done } = require('./helpers/boot');
assert.deepStrictEqual(stream[2], { subject: ada.subject, project });
state = 'succeeded';
result = { protocol: '1.0.0', lines: '1.1', log: [{ t: 'agent', agent: 'claude-code' }, { t: 'text', text: 'I fix <b>it</b></script>' }, { t: 'tool', id: 'e1', name: 'Edit', input: { file_path: 'README.md', old_string: 'teh', new_string: 'the' } }, { t: 'tool_result', id: 'e1', ok: true, output: 'ok' }, { t: 'mystery', x: 1 }, { t: 'done', ok: true, text: 'Fixed', duration_ms: 900 }], ok: true, agents: ['claude-code'], summary: 'Fixed <the> typo', files: [{ path: 'README.md', added: 1, removed: 1 }], patch: '@@ -1 +1 @@\n-teh\n+the\n', patch_truncated: false, push: { pushed: true, branch: 'openvibe-codes/x', branch_url: 'https://github.com/OpenVibers/OpenVibe.Codes/tree/openvibe-codes%2Fx', pull_request: 'https://github.com/OpenVibers/OpenVibe.Codes/pull/9' }, duration_ms: 42000 };
++ // The page view is not a poll: age the row past the 10-minute fallback so this read checks Run once.
++ await t.ctx.store.db.exec('UPDATE run_tasks SET updated_at = updated_at - $2 WHERE id = $1', [taskId, 11 * 60_000]);
const done_ = await t.get(`/run/tasks/${taskId}`, { as: ada });
// With the session on the page, its closing card says how it ended and what the agents said: the page does not repeat it.
assert.ok(!done_.text.includes('The agents finished the task') && !done_.text.includes('What the agent said') && done_.text.includes('README.md'));
@@@ -93,9 +95,9 @@ const { boot, check, done } = require('./helpers/boot');
// the page's script early.
assert.ok(done_.text.includes('data-session') && /<script src="[^"]*agent-session\.js[^"]*" defer>/.test(done_.text), 'the session and the component');
const kept = JSON.parse(/<script type="application\/json" data-session-log>([\s\S]*?)<\/script>/.exec(done_.text)[1]);
-- assert.deepStrictEqual(kept.map((l) => l.t), ['agent', 'text', 'tool', 'tool_result', 'done'], 'lines of an unknown kind are left out');
++ assert.deepStrictEqual(kept.map((l) => l.t), ['step', 'agent', 'text', 'tool', 'tool_result', 'done'], 'the streamed line first, then the result log’s known lines');
assert.ok(!/<\/script>I|<b>it/.test(/data-session-log>([\s\S]*?)<\/script>/.exec(done_.text)[1]), 'no markup inside the data');
-- assert.strictEqual(kept[1].text, 'I fix <b>it</b></script>', 'and the text reads back as it was');
++ assert.strictEqual(kept[2].text, 'I fix <b>it</b></script>', 'and the text reads back as it was');
const api = await t.get(`/api/v1/tasks/${taskId}`, { as: ada });
if (api.status === 200) assert.ok(!('log' in ((api.json && api.json.result) || {})), 'the task API passes codes.task@1\'s fields only');
const row = await t.ctx.store.db.one('SELECT state, result FROM run_tasks WHERE id = $1', [taskId]);
ddiff --git a/test/security-ssrf.test.js b/test/security-ssrf.test.js
iindex f5e31bc..c19bc7e 100644
---- a/test/security-ssrf.test.js
++++ b/test/security-ssrf.test.js
@@@ -44,6 +44,8 @@ const INTERNAL = [`http://127.0.0.1:3000${PROBE}`, `http://2130706433${PROBE}`,
'server/domain/run-client.js': 'OpenVibe.Run (CODES_RUN_URL, configured) and the Network token endpoint; job ids only, never a URL from a person',
'server/http/run.js': 'Network projects API (configured) with the viewer\'s token, and the Run job stream URL run-client.js builds on its own base',
'server/http/home.js': 'OpenVibe Git\'s public repository search (CODES_GIT_INTERNAL_URL, configured loopback Forgejo) for the home page; no token',
++ 'server/events-consumer.js': 'OpenVibe.Events subscriptions (EVENTS_URL, configured) at boot; no URL from a person',
++ 'server/events/outbox.js': 'OpenVibe.Events publish (EVENTS_URL, configured) of fixed codes.task.* payloads; no URL from a person',
};
const root = path.join(__dirname, '..');
const found = [];
ddiff --git a/test/session-lines.test.js b/test/session-lines.test.js
nnew file mode 100644
iindex 0000000..0d618c3
---- /dev/null
++++ b/test/session-lines.test.js
@@@ -0,0 +1,192 @@
++'use strict';
++/**
++ * Session lines in PostgreSQL (plan 2026-10-28 §6, decision 1): the Run stream's output/state/end frames become
++ * lines as they arrive, a reconnect does not duplicate them, a 70 KB line is clipped with a marker, the machine's
++ * final log back-fills what the stream missed, replay pages with after, the owner or a codes.task.read service reads
++ * them (another person cannot), a task from before this change falls back to result.log, and retention only deletes
++ * lines of tasks finished longer than the retention window ago.
++ */
++const assert = require('assert');
++const http = require('http');
++const { boot, check, done } = require('./helpers/boot');
++
++const TASK = (n) => `tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T${n}`;
++const JOB = (n) => `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${n}`;
++const NOW = Date.now();
++
++function frame(type, seq, data) {
++ return `id: ${seq}\nevent: ${type}\ndata: ${JSON.stringify({ type, seq, ...data })}\n\n`;
++}
++
++(async () => {
++ const stream = { frames: [], requests: 0 };
++ const sse = http.createServer((req, res) => {
++ stream.requests++;
++ res.writeHead(200, { 'Content-Type': 'text/event-stream' });
++ for (const f of stream.frames) res.write(f);
++ res.end();
++ });
++ await new Promise((ok) => sse.listen(0, '127.0.0.1', ok));
++
++ let readResult = null;
++ const run = {
++ enabled: true,
++ async submit() { return { id: JOB('1'), state: 'queued' }; },
++ async read(jobId) { return { id: jobId, state: 'succeeded', result: readResult, error: null, timings: { started_at: new Date(NOW).toISOString(), finished_at: new Date(NOW + 1000).toISOString() } }; },
++ async cancel() { return { state: 'cancelled' }; },
++ async streamUrl() { return `http://127.0.0.1:${sse.address().port}/stream`; },
++ };
++ const t = await boot({ run });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const bo = t.network.addUser('bo');
++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++ const insert = async (n, { state = 'running', job = JOB('1'), result = null, updatedAt = NOW, finishedAt = null } = {}) => {
++ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, result, finished_at, created_at, updated_at)
++ VALUES ($1, $2, $3, $4, 'https://github.com/OpenVibers/OpenVibe.Codes', 'Fix it', $5, $6::jsonb, $7::timestamptz, $8, $9)`,
++ [TASK(n), ada.subject, project, job, state, result == null ? null : JSON.stringify(result), finishedAt, NOW, updatedAt]);
++ };
++ const lines = (n) => db.many('SELECT seq, kind, body FROM codes_session_lines WHERE task_id = $1 ORDER BY seq', [TASK(n)]);
++ const replay = (n, query = '') => t.get(`/run/tasks/${TASK(n)}/session.json${query}`, { as: ada });
++
++ const finalLog = [
++ { t: 'step', text: 'Cloning' },
++ { t: 'tool', id: 't1', name: 'Read', input: { file_path: 'a.js' } },
++ { t: 'tool_result', id: 't1', ok: true, output: 'ok' },
++ { t: 'done', ok: true, text: 'Finished' },
++ ];
++
++ try {
++ await check('the output, state and end frames are lines in order, and the final log back-fills what the stream missed', async () => {
++ await insert('1');
++ stream.frames = [
++ frame('output', 1, { chunk: `${JSON.stringify(finalLog[0])}\n` }),
++ frame('output', 2, { chunk: `${JSON.stringify(finalLog[1])}\n` }),
++ frame('state', 3, { state: 'running' }),
++ frame('output', 4, { chunk: `${JSON.stringify(finalLog[2])}\n` }),
++ frame('end', 5, { state: 'succeeded' }),
++ ];
++ readResult = { ok: true, summary: 'Finished', log: finalLog };
++ const r = await t.get(`/run/tasks/${TASK('1')}/events`, { as: ada });
++ assert.strictEqual(r.status, 200, r.text);
++ const rows = await lines('1');
++ assert.deepStrictEqual(rows.map((x) => x.kind), ['step', 'tool', 'state', 'tool_result', 'done', 'end']);
++ for (let i = 1; i < rows.length; i++) assert.ok(rows[i].seq > rows[i - 1].seq, 'ordered by seq');
++ assert.deepStrictEqual(rows[0].body, finalLog[0]);
++ assert.deepStrictEqual(rows[4].body, finalLog[3], 'the done line came only from the result log');
++ const row = await db.one('SELECT state, session_complete_at FROM run_tasks WHERE id = $1', [TASK('1')]);
++ assert.strictEqual(row.state, 'succeeded');
++ assert.ok(row.session_complete_at, 'the lines are marked complete');
++ const page = await replay('1');
++ assert.strictEqual(page.status, 200, page.text);
++ assert.strictEqual(page.json().source, 'session');
++ assert.strictEqual(page.json().lines.length, 6);
++ });
++
++ await check('a reconnect that replays the stream does not duplicate a line', async () => {
++ const before = await lines('1');
++ const r = await t.get(`/run/tasks/${TASK('1')}/events`, { as: ada });
++ assert.strictEqual(r.status, 200);
++ const after = await lines('1');
++ assert.strictEqual(after.length, before.length);
++ assert.deepStrictEqual(after.map((x) => x.seq), before.map((x) => x.seq));
++ });
++
++ await check('a 70 KB line is clipped and carries a marker', async () => {
++ await insert('2', { job: JOB('2') });
++ stream.frames = [
++ frame('output', 1, { chunk: `${JSON.stringify({ t: 'text', text: 'x'.repeat(70 * 1024) })}\n` }),
++ frame('state', 2, { state: 'succeeded' }),
++ frame('end', 3, { state: 'succeeded' }),
++ ];
++ readResult = { ok: true, log: [] };
++ const r = await t.get(`/run/tasks/${TASK('2')}/events`, { as: ada });
++ assert.strictEqual(r.status, 200);
++ const rows = await lines('2');
++ const text = rows.find((x) => x.kind === 'text');
++ assert.ok(text);
++ assert.ok(text.body.text.length < 70 * 1024);
++ assert.match(text.body.text, /\[clipped\]/);
++ assert.ok(Buffer.byteLength(JSON.stringify(text.body)) <= 64 * 1024);
++ });
++
++ await check('a long session keeps its lines: a stream event past 8 600 does not overflow the seq', async () => {
++ await insert('8', { job: JOB('8') });
++ stream.frames = [
++ frame('output', 12000, { chunk: `${JSON.stringify({ t: 'text', text: 'late in a long run' })}\n` }),
++ frame('state', 12001, { state: 'succeeded' }),
++ frame('end', 12002, { state: 'succeeded' }),
++ ];
++ readResult = { ok: true, log: [] };
++ const r = await t.get(`/run/tasks/${TASK('8')}/events`, { as: ada });
++ assert.strictEqual(r.status, 200);
++ const text = (await lines('8')).find((x) => x.kind === 'text');
++ assert.ok(text, 'the line is kept');
++ assert.ok(text.seq > 2 ** 31, `seq ${text.seq} is past PostgreSQL's integer`);
++ });
++
++ await check('replay pages after a seq, at most 500 lines', async () => {
++ await insert('3', { state: 'succeeded', job: JOB('3'), updatedAt: NOW - 1000, finishedAt: new Date(NOW - 1000).toISOString() });
++ await t.ctx.sessions.append(TASK('3'), Array.from({ length: 510 }, (_, i) => ({
++ seq: i + 1, at: new Date(NOW).toISOString(), kind: 'step', body: { t: 'step', text: `line ${i + 1}` },
++ })));
++ const first = await replay('3', '?after=0');
++ assert.strictEqual(first.status, 200, first.text);
++ assert.strictEqual(first.json().lines.length, 500);
++ assert.strictEqual(first.json().next_after, 500);
++ assert.strictEqual(first.json().has_more, true);
++ const second = await replay('3', '?after=500');
++ assert.strictEqual(second.json().lines.length, 10);
++ assert.strictEqual(second.json().lines[0].seq, 501);
++ assert.strictEqual(second.json().has_more, false);
++ assert.strictEqual((await replay('3', '?after=nope')).status, 422);
++ });
++
++ await check('a person cannot read another person’s session; a codes.task.read service can read it for its subject', async () => {
++ assert.strictEqual((await t.get(`/run/tasks/${TASK('1')}/session.json`, { as: bo })).status, 404);
++ const caps = ['codes.task.read'];
++ const watch = t.network.mintService('svc:watch', 'openvibe.codes', caps);
++ const ok = await t.get(`/run/tasks/${TASK('1')}/session.json`, { headers: { authorization: `Bearer ${watch}`, 'X-OV-Subject': ada.subject } });
++ assert.strictEqual(ok.status, 200, ok.text);
++ assert.strictEqual(ok.json().task_id, TASK('1'));
++ const wrongPerson = await t.get(`/run/tasks/${TASK('1')}/session.json`, { headers: { authorization: `Bearer ${watch}`, 'X-OV-Subject': bo.subject } });
++ assert.strictEqual(wrongPerson.status, 404);
++ const noCap = t.network.mintService('svc:watch', 'openvibe.codes', ['codes.task.start']);
++ assert.strictEqual((await t.get(`/run/tasks/${TASK('1')}/session.json`, { headers: { authorization: `Bearer ${noCap}`, 'X-OV-Subject': ada.subject } })).status, 403);
++ assert.strictEqual((await t.get(`/run/tasks/${TASK('1')}/session.json`)).status, 401);
++ });
++
++ await check('a task from before this change falls back to the old result.log on the page and in replay', async () => {
++ await insert('4', {
++ state: 'succeeded', job: JOB('4'), updatedAt: NOW - 1000, finishedAt: new Date(NOW - 1000).toISOString(),
++ result: { ok: true, log: [{ t: 'text', text: 'old line' }, { t: 'done', ok: true, text: 'old done' }] },
++ });
++ const page = await t.get(`/run/tasks/${TASK('4')}`, { as: ada });
++ assert.strictEqual(page.status, 200);
++ const kept = JSON.parse(/<script type="application\/json" data-session-log>([\s\S]*?)<\/script>/.exec(page.text)[1]);
++ assert.deepStrictEqual(kept.map((x) => x.t), ['text', 'done']);
++ const r = await replay('4');
++ assert.strictEqual(r.json().source, 'result.log');
++ assert.deepStrictEqual(r.json().lines.map((x) => x.body.t), ['text', 'done']);
++ assert.strictEqual(r.json().lines[0].seq, 1);
++ });
++
++ await check('retention deletes only lines of tasks finished more than 90 days ago', async () => {
++ const old = new Date(NOW - 100 * 86_400_000).toISOString();
++ await insert('5', { state: 'succeeded', job: JOB('5'), updatedAt: NOW - 100 * 86_400_000, finishedAt: old });
++ await insert('6', { state: 'running', job: JOB('6'), updatedAt: NOW - 100 * 86_400_000 });
++ await insert('7', { state: 'succeeded', job: JOB('7'), updatedAt: NOW - 1000, finishedAt: new Date(NOW - 1000).toISOString() });
++ for (const n of ['5', '6', '7']) {
++ await t.ctx.sessions.append(TASK(n), [{ seq: 1, at: new Date(NOW).toISOString(), kind: 'step', body: { t: 'step', text: n } }]);
++ }
++ await t.ctx.sessionRetention.runOnce();
++ assert.strictEqual((await lines('5')).length, 0, 'old finished task');
++ assert.strictEqual((await lines('6')).length, 1, 'old but still running');
++ assert.strictEqual((await lines('7')).length, 1, 'recently finished');
++ });
++ } finally {
++ await t.close();
++ sse.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/tasks-api.test.js b/test/tasks-api.test.js
iindex deddc0a..365881f 100644
---- a/test/tasks-api.test.js
++++ b/test/tasks-api.test.js
@@@ -88,6 +88,8 @@ const { boot, check, done } = require('./helpers/boot');
patch: '+line\n', cost_usd: 0.0011, duration_ms: 4200, push: { pushed: false, reason: 'no token', smuggled: '<script>' },
branch_url: 'javascript:alert(1)',
};
++ // A read is not a poll: age the row past the 10-minute fallback so this one checks Run.
++ await db.exec('UPDATE run_tasks SET updated_at = updated_at - $2 WHERE id = $1', [task.id, 11 * 60_000]);
const r = await call('GET', `/tasks/${task.id}`);
assert.strictEqual(r.status, 200, JSON.stringify(r.json));
const v = validate('codes.task@1', r.json);