openvibers/openvibe.codes Commits

Tasks with attempts, a live event stream, and the engine deciding what happens next (C2) (#162)

OP OpenVibers committed 14 days ago · 491a45ae91 · 18 files, +1554 −56

* Tasks with attempts, a live event stream, and the engine deciding what happens next (C2)

A task is a timeline of attempts, never a new task per try. Each Run job is one
attempt; when an attempt ends the engine (engine/chain.js, engine/budget.js,
engine/router.js) looks at the task's attempts and decides: done, continue in the
same session, switch to the next agent, wrap up, or leave it unfinished.

- migrations/0010_task_attempts.sql: codes_attempts (id, task_id, n, agent, model,
  machine, job_id, session, state, reason, why, cost_usd, tokens, steps, started_at,
  finished_at, created_at, kind, follow_up_of), run_tasks.outcome / outcome_note /
  attempts_count, codes_session_lines.attempt_n. Every existing task becomes its own
  attempt 1.
- server/domain/attempts.js: the attempts and the decision. The engine's vocabulary
  is mapped onto Codes' (the reason is passed on, since the engine only reads the
  runners it knows).
- server/domain/run-tasks.js: the events inbox and the run.job.* handler update the
  attempt; the decision runs outside the transaction, because a follow-up starts a
  Run job. A follow-up is fenced by the unique follow_up_of, so two endings of the
  same attempt start one job, not two.
- server/http/task-events.js: GET /api/v1/tasks/:id/events, the session's lines in
  seq order, resumable with Last-Event-ID, a heartbeat every 15 s, the attempt each
  line belongs to and the attempts themselves.
- server/http/tasks-api.js: GET /api/v1/tasks/:id answers { task, attempts[],
  outcome }; POST /api/v1/tasks/:id/continue and /stop. The same auth as the tasks
  API: the owner, or a service with the codes.task capability.
- /run/tasks/:id shows the attempts above the session, each with its why.

* The read answers codes.task@2: the task at the top level, its attempts added

GET /api/v1/tasks/:id answers codes.task@2 now: the task's own fields stay at
the top level, exactly as codes.task@1 had them, and its attempts and its
outcome are added to them. A caller that reads the task — OpenVibe.Actor's
codes agent, which takes the answer as it comes — keeps working; one that
expected the task under a "task" key would not have.

Codes pins openvibe-contracts v0.242.0, which carries codes.task@2 (the v1
contract is unchanged, and codes.task.read declares the v2).

* Pin openvibe-contracts v0.242.1 (the release whose generated files carry
their own version)

* Pin openvibe-contracts v0.243.0 (main's release of codes.task@2)

Brings the contract package up from v0.242.1; contracts/codes is unchanged
(task.v2.json is the one the API answers), and the release adds AI credential
v2 schemas. README's "Depends on" line names the same versions as the
versions block again.

Changes

ddiff --git a/README.md b/README.md
iindex 53343d7..79dc5a3 100644
---- a/README.md
++++ b/README.md
@@@ -20,9 +20,9 @@ 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 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). |
++| 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 **timeline of attempts** above the agents' **session** (each attempt with its why: "Codex hit its limit, so Command Code took over with what it had done") and the 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. |
++| Task API (services) | `POST /api/v1/tasks`, `GET /api/v1/tasks/:id`, `GET /api/v1/tasks/:id/events`, `POST /api/v1/tasks/:id/continue`, `POST /api/v1/tasks/:id/stop`, `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 owner needs no token for their own tasks. 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. **A task's attempts** (stage C2, migration 0010 `codes_attempts`): every Run job is one attempt, so a continue, a switch to another agent and a wrap-up are attempts on one timeline, never new tasks. `GET /api/v1/tasks/:id` answers `codes.task@2`: the task's own fields at the top level, exactly as `codes.task@1` had them, and, added to them, `attempts[]` (the timeline) and `outcome` — `done`, `unfinished` (with what is left) or `stopped`. `GET /api/v1/tasks/:id/events` is the task's event stream (server-sent events): the session's lines in order with the attempt each belongs to, the attempts themselves (started, ended with why), a heartbeat every 15 s, and a resume with `Last-Event-ID`. `POST /api/v1/tasks/:id/continue` starts the next attempt now (with `agent` it switches), `POST /api/v1/tasks/:id/stop` stops the running attempt and nothing carries it on. |
| 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`). |
| Improve OpenVibe | `/improve` | Every OpenVibe repository, from the pinned contracts' service manifests (the services that run, then the shared libraries), and the path from a change to a reviewed pull request. |
| Community | `/policy`, `/policy/<document>` | The code of conduct ([CODE_OF_CONDUCT.md](CODE_OF_CONDUCT.md)), the contribution guide ([CONTRIBUTING.md](CONTRIBUTING.md)), the [contributor ladder](docs/governance/contributor-ladder.md) and the [moderation policy](docs/governance/moderation.md), rendered from those files. All four are **drafts pending owner review**: marked at the top, shown with a notice, not indexed; deleting the draft line publishes one. They govern the repositories and the apps and releases on OpenVibe.Services alike. |
@@@ -84,7 +84,7 @@ openvibe-codes resume <run id> "now add a test"
- **OpenVibe.Network** — SSO (OAuth client `codes`, PKCE S256) and its JWKS.
- **PostgreSQL 18 and Valkey 9** (OpenVibe.Host `roles/data/`, ADR-035): `openvibe-sdk/db`; Valkey holds the per-actor limit counters (optional).
--- **openvibe-contracts v0.143.0**, **openvibe-sdk v0.67.0**, **openvibe-shared v4.23.0** (pinned tag tarballs).
++- **openvibe-contracts v0.243.0**, **openvibe-sdk v0.73.0**, **openvibe-shared v4.48.0** (pinned tag tarballs).
No path in Codes sends or accepts a shared loopback key (tested by grep and at runtime).
@@@ -158,7 +158,7 @@ Reporting a vulnerability: [SECURITY.md](SECURITY.md).
Part of the [OpenVibe network](https://openvibe.network). Built in the open by [OpenVibers](https://github.com/OpenVibers).
<!-- versions:start -->
--- openvibe-contracts: v0.232.0
++- openvibe-contracts: v0.243.0
- openvibe-sdk: v0.73.0
- openvibe-shared: v4.48.0
<!-- versions:end -->
ddiff --git a/STATUS.json b/STATUS.json
iindex 07e3221..9364e24 100644
---- a/STATUS.json
++++ b/STATUS.json
@@@ -7,11 +7,11 @@
"runtime": "node 22, express 4, PostgreSQL 18 (openvibe-sdk/db, async) + Valkey 9",
"port": 4900,
"domain": "openvibe.codes",
-- "contracts": "openvibe-contracts v0.232.0",
++ "contracts": "openvibe-contracts v0.243.0",
"sdk": "openvibe-sdk v0.73.0",
"shared": "openvibe-shared v4.48.0",
"packages": [
-- "openvibe-contracts v0.232.0",
++ "openvibe-contracts v0.243.0",
"openvibe-sdk v0.73.0",
"openvibe-shared v4.48.0"
],
ddiff --git a/harness/job.js b/harness/job.js
iindex 2385d51..db3beee 100644
---- a/harness/job.js
++++ b/harness/job.js
@@@ -84,8 +84,13 @@ function parseArgs(raw, env = process.env) {
} : null;
if (push && push.branch && (!BRANCH.test(push.branch) || push.branch.startsWith('-'))) throw new Error('push.branch must be a branch name');
if (push && push.branch && PROTECTED.test(push.branch)) throw new Error(`push.branch must be a new branch of the task's own, never ${push.branch}`);
++ // The session a resume continues: the job's own input, or RESUME in the environment (a task carried on in the
++ // same conversation). It is a harness's session id, never a path and never a credential.
++ const resumeIn = a.resume == null || a.resume === '' ? env.RESUME : a.resume;
++ const resume = typeof resumeIn === 'string' && /^[A-Za-z0-9._:-]{1,200}$/.test(resumeIn) ? resumeIn : null;
return {
repo: repo.toString(), ref, prompt, permission, push,
++ ...(resume ? { resume } : {}),
agent: typeof a.agent === 'string' && /^[a-z0-9-]{1,40}$/.test(a.agent) ? a.agent : null,
model: typeof a.model === 'string' && a.model.length <= 100 ? a.model : null,
};
@@@ -129,7 +134,7 @@ const resultText = (c) => (typeof c === 'string' ? c : Array.isArray(c) ? c.map(
function compact(e) {
const cut = (s) => cutTo(s, MAX_TEXT);
const parent = e.parent_tool_use_id ? { parent: String(e.parent_tool_use_id) } : {};
-- if (e.type === 'system' && e.subtype === 'init') return { t: 'agent', agent: e.agent, model: e.model || null };
++ if (e.type === 'system' && e.subtype === 'init') return { t: 'agent', agent: e.agent, model: e.model || null, session: e.session_id || null };
if (e.type === 'system' && e.subtype === 'handoff') return { t: 'handoff', from: e.from, ...(e.to ? { to: e.to } : {}), reason: cut(e.reason) };
if (e.type === 'system' && (e.subtype === 'error' || e.subtype === 'warning')) return { t: 'note', level: e.subtype, text: cut(e.text) };
if (e.type === 'system' && e.subtype === 'tool_progress') return { t: 'tool_progress', id: String(e.tool_use_id || ''), chunk: clean(cutTo(e.text, MAX_OUTPUT)) };
@@@ -266,6 +271,9 @@ async function runJob({ input, harness, emit, writeResult, env = process.env, tm
const run = harness.run({
prompt: job.prompt, task: 'edit', agent: job.agent, model: job.model, cwd: work,
permission: job.permission, handoff: !job.agent, maxAttempts: 3,
++ // A task carried on in the same conversation resumes its session (RESUME in the environment, or the job's
++ // own resume field).
++ resume: job.resume || null,
});
let last = null;
for await (const e of run) {
ddiff --git a/migrations/0010_task_attempts.sql b/migrations/0010_task_attempts.sql
nnew file mode 100644
iindex 0000000..44d300a
---- /dev/null
++++ b/migrations/0010_task_attempts.sql
@@@ -0,0 +1,109 @@
++-- phase: expand
++-- Stage C2 (openvibe.codes v1): a task's attempts. A continue, a switch to another agent and a wrap-up are attempts on
++-- one task's timeline, never new tasks with cryptic names (the portal's "-c1"/"-r2" confusion of 10-30). Every Run job
++-- is one attempt: the attempt carries the job, the agent, the machine and the harness session a resume needs.
++--
++-- n the attempt's place on the timeline (1, 2, …)
++-- state queued | running | done | failed | stopped | budget-stopped | interrupted
++-- reason limit | auth | capacity | refused | error | budget | stopped | hung | context (why it ended)
++-- why one sentence in plain words: why this attempt happened (for the timeline)
++-- kind first | continue | switch | wrap-up | manual: how this attempt came to be
++-- follow_up_of the attempt whose end started this one. It is unique, so one ending can start only one follow-up:
++-- the fence against the portal's 10-30 bug, where two racing ticks carried one job on twice.
++--
++-- run_tasks grows the task's own verdict: outcome ('done' | 'unfinished' | 'stopped', NULL while it runs), outcome_note
++-- (what is left, in plain words) and attempts_count. codes_session_lines grows attempt_n, so every line of the session
++-- says which attempt it belongs to.
++--
++-- Every task that exists now becomes attempt 1 of itself: the timeline of an old task is not empty.
++
++CREATE TABLE IF NOT EXISTS codes_attempts (
++ id text COLLATE "C" PRIMARY KEY, -- att_<ULID>
++ task_id text COLLATE "C" NOT NULL, -- tsk_<ULID>
++ n integer NOT NULL, -- 1, 2, … the attempt's place on the timeline
++ agent text COLLATE "C", -- the harness (claude-code, codex, …) or 'auto'
++ model text COLLATE "C",
++ machine text COLLATE "C", -- the node's id, or 'sandbox'
++ job_id text COLLATE "C", -- Run's job_<ULID> for this attempt
++ session text COLLATE "C", -- the harness session id, for a resume
++ state text COLLATE "C" NOT NULL, -- queued, running, done, failed, stopped, budget-stopped, interrupted
++ reason text COLLATE "C", -- limit, auth, capacity, refused, error, budget, stopped, hung, context
++ why text COLLATE "C", -- one sentence in plain words
++ kind text COLLATE "C" NOT NULL DEFAULT 'first',-- first | continue | switch | wrap-up | manual
++ cost_usd numeric(14,6),
++ tokens bigint,
++ steps integer,
++ started_at timestamptz,
++ finished_at timestamptz,
++ created_at timestamptz NOT NULL DEFAULT now(),
++ follow_up_of text COLLATE "C" -- the attempt whose end started this one
++);
++CREATE UNIQUE INDEX IF NOT EXISTS codes_attempts_task_n ON codes_attempts (task_id, n);
++-- One ending, one follow-up: the unique index is the fence (the second insert is a no-op).
++CREATE UNIQUE INDEX IF NOT EXISTS codes_attempts_follow_up ON codes_attempts (follow_up_of) WHERE follow_up_of IS NOT NULL;
++-- A Run job is one attempt: the job of an attempt is found by it.
++CREATE INDEX IF NOT EXISTS codes_attempts_job ON codes_attempts (job_id) WHERE job_id IS NOT NULL;
++CREATE INDEX IF NOT EXISTS codes_attempts_task ON codes_attempts (task_id, n);
++
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS outcome text COLLATE "C"; -- done | unfinished | stopped
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS outcome_note text COLLATE "C"; -- what is left, in plain words
++ALTER TABLE run_tasks ADD COLUMN IF NOT EXISTS attempts_count integer NOT NULL DEFAULT 0;
++ALTER TABLE codes_session_lines ADD COLUMN IF NOT EXISTS attempt_n integer;
++
++-- Every task that exists becomes attempt 1 of itself. The id is the task's own id hashed into the ULID alphabet
++-- (26 characters of Crockford base32), so the same task always gets the same attempt and a rerun of the migration
++-- changes nothing.
++INSERT INTO codes_attempts (id, task_id, n, agent, job_id, state, reason, why, kind, started_at, finished_at, created_at)
++SELECT 'att_' || upper(substr(md5(t.id), 1, 26)),
++ t.id,
++ 1,
++ t.agent,
++ t.job_id,
++ CASE t.state
++ WHEN 'queued' THEN 'queued'
++ WHEN 'placed' THEN 'queued'
++ WHEN 'running' THEN 'running'
++ WHEN 'succeeded' THEN 'done'
++ WHEN 'cancelled' THEN 'stopped'
++ ELSE 'failed'
++ END,
++ CASE t.state
++ WHEN 'cancelled' THEN 'stopped'
++ WHEN 'expired' THEN 'capacity'
++ WHEN 'failed' THEN 'error'
++ ELSE NULL
++ END,
++ CASE t.state
++ WHEN 'succeeded' THEN 'the first attempt of this task finished the work'
++ WHEN 'cancelled' THEN 'the first attempt of this task was stopped on purpose'
++ WHEN 'expired' THEN 'no machine took the first attempt in time'
++ WHEN 'failed' THEN 'the first attempt of this task failed'
++ WHEN 'running' THEN 'the first attempt of this task is running'
++ ELSE 'the first attempt of this task is waiting for a machine'
++ END,
++ 'first',
++ t.started_at,
++ t.finished_at,
++ to_timestamp(GREATEST(t.created_at, 0) / 1000.0)
++FROM run_tasks t
++WHERE NOT EXISTS (SELECT 1 FROM codes_attempts a WHERE a.task_id = t.id);
++
++UPDATE run_tasks t
++SET outcome = CASE t.state
++ WHEN 'succeeded' THEN 'done'
++ WHEN 'cancelled' THEN 'stopped'
++ WHEN 'failed' THEN 'unfinished'
++ WHEN 'expired' THEN 'unfinished'
++ ELSE NULL
++ END,
++ outcome_note = CASE t.state
++ WHEN 'succeeded' THEN 'the task finished before attempts were counted'
++ WHEN 'cancelled' THEN 'the task was stopped on purpose'
++ WHEN 'failed' THEN 'the attempt failed; what it left is in the session'
++ WHEN 'expired' THEN 'no machine took the task in time'
++ ELSE NULL
++ END,
++ attempts_count = 1
++WHERE t.attempts_count = 0;
++
++UPDATE codes_session_lines SET attempt_n = 1 WHERE attempt_n IS NULL;
ddiff --git a/package-lock.json b/package-lock.json
iindex 0725a20..b18c1cd 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.232.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.243.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.73.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
@@@ -1149,9 +1149,9 @@
}
},
"node_modules/openvibe-contracts": {
-- "version": "0.232.0",
-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.232.0",
-- "integrity": "sha512-6HI5nn//nAUEDdG4xRTAlcb40oc/6yO1nx7R5exGZ6Bo3mL5fJ5szw9IONddoLN6qJsO1CXzB5TzTqkBAh3OzQ==",
++ "version": "0.243.0",
++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.243.0",
++ "integrity": "sha512-ZFrwTpgALUEOIvdYi2r654murdEwludenAZ5u3CuvPMVcYuu/3kC+J20jVeqPaCFKYkzRzLyO8fH/iOdh44aYg==",
"license": "AGPL-3.0-only",
"dependencies": {
"ajv": "^8.17.1",
ddiff --git a/package.json b/package.json
iindex 0e6213c..1e2b5c7 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.232.0",
++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.243.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.73.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v4.48.0",
"pg": "^8.23.0"
ddiff --git a/server/app.js b/server/app.js
iindex bcea35e..99bc489 100644
---- a/server/app.js
++++ b/server/app.js
@@@ -32,6 +32,7 @@ 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 { createAttempts } = require('./domain/attempts');
const { createSessionLines } = require('./domain/session-lines');
const { createOutbox, createPublisher } = require('./events/outbox');
const { createEventsConsumer } = require('./events-consumer');
@@@ -66,8 +67,10 @@ async function createApp(opts = {}) {
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 };
++ // The task's attempts (stage C2): every Run job is one attempt, and the engine decides what happens next.
++ const attempts = createAttempts({ store, config, log, harnesses });
++ const tasks = createRunTasks({ store, run, config, log, sessions, outbox, attempts });
++ const ctx = { config, store, keys, sso, harnesses, log, sessions, outbox, run, tasks, attempts };
ctx.forgejo = createForgejoClient({ config, log });
const app = express();
ddiff --git a/server/config.js b/server/config.js
iindex 4a71991..2b893c6 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -81,6 +81,18 @@ function load(env = process.env) {
// 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') },
++ // A task's own budget (engine/budget.js): steps (tool calls), minutes (wall clock) and dollars. A limit that is
++ // not set is no limit — 0 means "no limit", never a zero budget. Steps and minutes buy one wrap-up turn.
++ budget: {
++ steps: int(env.CODES_TASK_STEPS, 0),
++ minutes: int(env.CODES_TASK_MINUTES, 0),
++ usd: Number(env.CODES_TASK_USD || 0) || 0,
++ },
++ // The task's timeline (stage C2): how many automatic continues one task may have before it needs a person.
++ task: { maxContinues: Math.max(0, int(env.CODES_TASK_MAX_CONTINUES, 2)) },
++ // The event stream of a task (GET /api/v1/tasks/:id/events): the heartbeat that keeps a connection alive.
++ sse: { heartbeatMs: Math.max(1000, int(env.CODES_SSE_HEARTBEAT_MS, 15_000)) },
++
};
}
ddiff --git a/server/domain/attempts.js b/server/domain/attempts.js
nnew file mode 100644
iindex 0000000..1412a5a
---- /dev/null
++++ b/server/domain/attempts.js
@@@ -0,0 +1,418 @@
++'use strict';
++
++/**
++ * A coding task's attempts (codes_attempts): the timeline of one task.
++ *
++ * A continue, a switch to another agent and a wrap-up are attempts on one task's timeline, never new tasks with
++ * cryptic names (the portal's "-c1"/"-r2" confusion of 10-30). Every Run job is one attempt: the attempt carries the
++ * job, the agent, the machine and the harness session a resume needs.
++ *
++ * The engine decides what happens next (engine/chain.js outcome/nextStep, engine/budget.js wrapUp): this module turns
++ * its decision into a plan — a follow-up attempt, or the task's outcome in plain words. It never starts a Run job
++ * itself (server/domain/run-tasks.js does that, outside a transaction) and it never decides twice for one ending:
++ * the follow-up's follow_up_of is unique, so one ending can start only one follow-up.
++ *
++ * createAttempts({ store, config, log, harnesses })
++ * list(taskId) the attempts of a task, in order
++ * get(id) one attempt
++ * create({ task, …, followUpOf }) one attempt; the same followUpOf answers the same attempt (the fence)
++ * bind(id, { job_id, state }) the Run job of an attempt
++ * end(id, { state, reason, … }) an attempt ended: its state, its reason, its cost
++ * decide({ task, attempts, ending, result }) the engine's decision, as a plan
++ * present(attempt) the attempt as the API shows it
++ */
++
++const { chain, budget, router } = require('../../engine');
++
++/** Codes' harness ids and the engine's short runner ids for the same harness. */
++const RUNNER_OF = { 'claude-code': 'claude', codex: 'cx', 'command-code': 'cc', opencode: 'oc', deepseek: 'dsh' };
++const HARNESS_OF = { claude: 'claude-code', cx: 'codex', cc: 'command-code', oc: 'opencode', dsh: 'deepseek' };
++
++/** The order a runner switch tries the harnesses in (the ladder's order, flattened). */
++const FALLBACK_ORDER = ['cx', 'claude', 'cc', 'oc', 'dsh'];
++
++/** Which harnesses may take which kind of task, cheapest first (the engine's ladder, Codes' harnesses). */
++const TIER_ORDER = {
++ basic: ['opencode', 'command-code', 'deepseek'],
++ research: ['codex', 'opencode', 'deepseek'],
++ standard: ['codex', 'claude-code', 'command-code', 'opencode', 'deepseek'],
++ hard: ['codex', 'claude-code', 'deepseek'],
++ design: ['codex', 'claude-code', 'opencode'],
++};
++const TIER_EFFORT = { basic: 'low', research: 'medium', standard: 'high', hard: 'high', design: 'high' };
++
++/** The attempt's own states, and the reasons it may have ended. */
++const STATES = ['queued', 'running', 'done', 'failed', 'stopped', 'budget-stopped', 'interrupted'];
++const REASONS = ['limit', 'auth', 'capacity', 'refused', 'error', 'budget', 'stopped', 'hung', 'context'];
++const KINDS = ['first', 'continue', 'switch', 'wrap-up', 'restart', 'manual'];
++
++/** The engine's own words for a reason: the timeline reads like a person wrote it. */
++const REASON_TEXT = chain.REASON_TEXT;
++
++/**
++ * The text the engine classifies a reason from. The engine reads the reason of a failed attempt out of what it said
++ * (engine/chain.js reason), so the text has to say the same thing our reason column says.
++ */
++const ERROR_TEXT = {
++ limit: 'usage limit reached',
++ auth: 'authentication failed: log in again',
++ capacity: 'the model was at capacity',
++ refused: 'turned off in the agent portal',
++ error: 'the work failed',
++ budget: 'stopped at its limit',
++ stopped: 'stopped on purpose',
++ context: 'the context window is too big',
++ hung: 'it went quiet for too long',
++};
++
++/** The engine's status for our state (the two vocabularies are the same, except 'queued' and 'running'). */
++const ENGINE_STATE = {
++ queued: 'queued', running: 'running', done: 'done', failed: 'failed',
++ stopped: 'stopped', 'budget-stopped': 'budget-stopped', interrupted: 'interrupted',
++};
++
++/** The attempt's state for a task's state (a task from before the attempts still has one). */
++const TASK_STATE = {
++ queued: 'queued', placed: 'queued', running: 'running', succeeded: 'done',
++ failed: 'failed', cancelled: 'stopped', expired: 'failed',
++};
++
++const iso = (v) => {
++ if (!v) return null;
++ const t = v instanceof Date ? v.getTime() : Date.parse(String(v));
++ return Number.isFinite(t) ? new Date(t).toISOString() : null;
++};
++const clip = (s, n) => String(s == null ? '' : s).slice(0, n);
++const runnerOf = (agent) => RUNNER_OF[agent] || (agent || 'auto');
++const wordsOf = (reason) => REASON_TEXT[reason] || REASON_TEXT.error;
++
++function createAttempts({ store, config = {}, log = console, harnesses = null }) {
++ const db = store.db;
++ const catalog = harnesses || require('./harnesses').createHarnesses();
++ // How many automatic continues one task may have before it needs a person (CODES_TASK_MAX_CONTINUES, two).
++ const wanted = config.task ? config.task.maxContinues : undefined;
++ const maxContinues = Number.isFinite(Number(wanted)) ? Math.max(0, Number(wanted)) : 2;
++
++ const nameOf = (agent) => {
++ const offer = agent ? catalog.get(agent) : null;
++ return offer ? offer.name : (agent || 'the agent');
++ };
++
++ /** The attempts of a task, in order. A task written before the attempts (or by another path) still has a first one. */
++ async function list(taskId) {
++ if (!taskId) return [];
++ const rows = await db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);
++ if (rows.length) return rows;
++ const task = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
++ if (!task) return [];
++ await db.exec(
++ `INSERT INTO codes_attempts (id, task_id, n, agent, job_id, state, why, kind, started_at, finished_at, created_at)
++ VALUES ($1, $2, 1, $3, $4, $5, $6, 'first', $7::timestamptz, $8::timestamptz, $9::timestamptz)
++ ON CONFLICT DO NOTHING`,
++ [store.newId('att'), task.id, task.agent || null, task.job_id || null, TASK_STATE[task.state] || 'failed',
++ 'the first attempt of this task', task.started_at || null, task.finished_at || null, store.iso()],
++ );
++ return await db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);
++ }
++
++ /** One attempt. */
++ async function get(id) {
++ if (!id) return null;
++ return await db.maybe('SELECT * FROM codes_attempts WHERE id = $1', [id]);
++ }
++
++ /** The attempt a job belongs to, or null. */
++ async function byJob(jobId) {
++ if (!jobId) return null;
++ return await db.maybe('SELECT * FROM codes_attempts WHERE job_id = $1', [jobId]);
++ }
++
++ /**
++ * One attempt. The fence: follow_up_of is unique, so an ending that starts a follow-up twice answers the same
++ * attempt the second time and starts no second job. created tells the caller which of the two it was.
++ */
++ async function create({ task, agent = null, model = null, machine = null, session = null, why = null, kind = 'first', followUpOf = null, state = 'queued' }) {
++ if (!task || !task.id) throw new Error('an attempt belongs to a task');
++ for (let tries = 0; tries < 5; tries++) {
++ const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_attempts WHERE task_id = $1', [task.id]);
++ const id = store.newId('att');
++ const inserted = await db.maybe(
++ `INSERT INTO codes_attempts (id, task_id, n, agent, model, machine, session, state, why, kind, follow_up_of, created_at)
++ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::timestamptz)
++ ON CONFLICT DO NOTHING RETURNING *`,
++ [id, task.id, n.n, agent, model, machine, session, state, why, kind, followUpOf, store.iso()],
++ );
++ if (inserted) {
++ await db.exec('UPDATE run_tasks SET attempts_count = $2 WHERE id = $1', [task.id, n.n]);
++ return { attempt: inserted, created: true };
++ }
++ // Someone else wrote it: the same follow-up answers the same attempt.
++ const seen = followUpOf ? await db.maybe('SELECT * FROM codes_attempts WHERE follow_up_of = $1', [followUpOf]) : null;
++ if (seen) return { attempt: seen, created: false };
++ }
++ throw new Error('the attempt could not be written');
++ }
++
++ /** The Run job of an attempt. */
++ async function bind(id, { job_id = null, state = null, started_at = null, machine = null, model = null } = {}) {
++ return await db.one(
++ `UPDATE codes_attempts SET job_id = COALESCE($2, job_id),
++ state = COALESCE($3, state),
++ machine = COALESCE($4, machine),
++ model = COALESCE($5, model),
++ started_at = COALESCE($6::timestamptz, started_at)
++ WHERE id = $1 RETURNING *`,
++ [id, job_id, state, machine, model, started_at],
++ );
++ }
++
++ /** An attempt ended: its state, why, what it cost. */
++ async function end(id, { state = 'failed', reason = null, why = null, cost_usd = null, tokens = null, steps = null, session = null, model = null, machine = null, finished_at = null } = {}) {
++ return await db.one(
++ `UPDATE codes_attempts SET state = $2,
++ reason = COALESCE($3, reason),
++ why = COALESCE($4, why),
++ cost_usd = COALESCE($5::numeric, cost_usd),
++ tokens = COALESCE($6, tokens),
++ steps = COALESCE($7, steps),
++ session = COALESCE($8, session),
++ model = COALESCE($9, model),
++ machine = COALESCE($10, machine),
++ finished_at = COALESCE($11::timestamptz, finished_at)
++ WHERE id = $1 RETURNING *`,
++ [id, state, reason, why, cost_usd == null ? null : String(cost_usd), tokens, steps, session, model, machine, finished_at || store.iso()],
++ );
++ }
++
++ /** The engine's own vocabulary for Codes' harnesses: the ladder, the health and the catalogs it routes with. */
++ function routing() {
++ const subagents = {}; const models = {}; const runners = [];
++ for (const [harnessId, runner] of Object.entries(RUNNER_OF)) {
++ const offer = catalog.get(harnessId);
++ if (!offer) continue;
++ subagents[runner] = { enabled: true, max: 4, model: null };
++ models[runner] = (offer.agents || []).map((a) => a.model).filter(Boolean);
++ runners.push(runner);
++ }
++ const health = {};
++ for (const r of runners) health[r] = { state: 'ok' };
++ const ladder = {};
++ for (const tier of router.TIERS) {
++ ladder[tier] = (TIER_ORDER[tier] || []).map((id) => {
++ const runner = RUNNER_OF[id];
++ const offer = catalog.get(id);
++ const agent = offer && offer.agents && offer.agents[0];
++ return [runner, agent ? agent.model : null, TIER_EFFORT[tier]];
++ }).filter(([runner, model]) => subagents[runner] && (!model || models[runner].includes(model)));
++ }
++ return {
++ cfg: { subagents, budget: config.budget || {}, fallback: { order: FALLBACK_ORDER.filter((r) => subagents[r]) } },
++ health, counts: {}, models, runners, ladder,
++ };
++ }
++
++ /**
++ * The next agent for a task, from the engine's router (engine/router.js): the tier of the brief decides the
++ * ladder, and the first harness of it that may take the task takes it. The agent that just stopped is skipped.
++ */
++ function nextAgent(prompt, ctx, exclude) {
++ const usable = (runner) => {
++ const sub = ctx.cfg.subagents[runner];
++ if (!sub || !sub.enabled) return false;
++ if ((ctx.counts[runner] || 0) >= sub.max) return false;
++ const h = ctx.health[runner];
++ return !!h && h.state === 'ok';
++ };
++ const picked = router.pick(prompt, ctx);
++ const tier = picked ? picked.tier : router.classify(prompt).tier;
++ const ladder = ctx.ladder[tier] || [];
++ const shape = (runner, model, effort) => ({
++ runner,
++ agent: HARNESS_OF[runner] || null,
++ model: model || null,
++ effort: router.fitEffort(runner, effort),
++ tier,
++ why: picked ? picked.why : `${tier} → ${runner}`,
++ });
++ if (picked && picked.runner !== exclude && usable(picked.runner)) return shape(picked.runner, picked.model, picked.effort);
++ for (const [runner, model, effort] of ladder) {
++ if (runner === exclude || !usable(runner)) continue;
++ const m = model && ctx.models[runner] && !ctx.models[runner].includes(model) ? null : model;
++ return shape(runner, m, effort);
++ }
++ return null;
++ }
++
++ /** One attempt as the engine sees it. */
++ function engineAttempt(a, all, facts = null) {
++ const n = Number(a.n) || 1;
++ const reason = a.reason && REASONS.includes(a.reason) ? a.reason : null;
++ const next = all.find((x) => x.follow_up_of && x.follow_up_of === a.id);
++ const prev = all.find((x) => Number(x.n) === n - 1);
++ const created = a.created_at ? Date.parse(String(a.created_at)) : 0;
++ return {
++ key: a.id,
++ runner: runnerOf(a.agent),
++ name: n === 1 ? 'a' : `a-r${n}`,
++ runnerName: nameOf(a.agent),
++ status: ENGINE_STATE[a.state] || 'failed',
++ // The order the engine reads the chain in: the attempt's place, never a tie.
++ updated: created + n,
++ attempt: n,
++ error: reason ? ERROR_TEXT[reason] : null,
++ // The engine reads the reason of a failed attempt out of what it said; Codes' attempts carry their own
++ // reason, and the engine only reads the runners it knows, so it is passed on.
++ reason: reason || null,
++ next: next ? next.id : null,
++ prev: prev ? prev.id : null,
++ parent: a.follow_up_of || null,
++ stopped: a.state === 'stopped',
++ budget: reason === 'budget' ? { hit: 'steps' } : undefined,
++ wrapUp: a.kind === 'wrap-up',
++ hung: reason === 'hung',
++ facts: {
++ session: a.session || null,
++ apiError: (facts && facts.apiError) || (reason ? ERROR_TEXT[reason] : null),
++ hasFinal: !!(facts && facts.hasFinal),
++ finalText: (facts && facts.finalText) || '',
++ },
++ };
++ }
++
++ /**
++ * What the job's log says about the attempt: its last words (the engine reads them to know whether the agent
++ * said it was done) and the error it ended with.
++ */
++ function factsOf(result) {
++ const d = result && typeof result === 'object' && !Array.isArray(result) ? result : {};
++ const log = Array.isArray(d.log) ? d.log : [];
++ let hasFinal = false; let finalText = ''; let apiError = '';
++ for (const line of log) {
++ if (!line || typeof line !== 'object') continue;
++ if (line.t === 'done') {
++ hasFinal = true;
++ if (line.text) finalText = String(line.text);
++ if (line.ok === false && !apiError) apiError = finalText || 'the agent ended without an answer';
++ } else if (line.t === 'note' && line.level === 'error' && !apiError) {
++ apiError = String(line.text || '');
++ }
++ }
++ return { hasFinal, finalText, apiError };
++ }
++
++ /** The note a continued attempt gets: it keeps the work, it does not redo it. */
++ function continuationNote(ending, words) {
++ return `\n\n---\n## Continuation (written by openvibe.codes)\n\nAn earlier attempt${ending.agent ? ` on ${nameOf(ending.agent)}` : ''} ${words}. Keep the work it finished, do not redo it, and continue from where it stopped. If the task is already complete, check it and report that.\n`;
++ }
++
++ /** The note a new agent gets: what changed, what it said, where it stopped. */
++ function handoverNote(ending, result, words) {
++ const d = result && typeof result === 'object' ? result : {};
++ const files = Array.isArray(d.files) ? d.files.map((f) => f && f.path).filter(Boolean).slice(0, 20) : [];
++ const summary = clip(d.summary || '', 1200);
++ return `\n\n---\n## Hand-over (written by openvibe.codes)\n\n${nameOf(ending.agent)} ${words}. Before it stopped it:\n`
++ + `- changed: ${files.length ? files.join(', ') : 'nothing yet'}\n`
++ + `- said: ${summary || 'nothing'}\n`
++ + `- stopped at: ${words}\n\n`
++ + 'Keep the work it did, do not redo it, and finish the task.\n';
++ }
++
++ /**
++ * What happens after an attempt ends: the engine decides, this turns the decision into a plan.
++ *
++ * @param {object} what
++ * @param {object} what.task the task row
++ * @param {object[]} what.attempts every attempt of the task, in order
++ * @param {object} what.ending the attempt that ended
++ * @param {object} [what.result] the job's result (what it changed and what it said)
++ * @returns {object} the plan: an outcome ('done', 'unfinished', 'stopped') or a follow-up attempt
++ */
++ function decide({ task, attempts, ending, result = null }) {
++ const ctx = routing();
++ const upto = (attempts || []).filter((a) => Number(a.n) <= Number(ending.n));
++ const facts = factsOf(result);
++ const engineAttempts = upto.map((a) => engineAttempt(a, upto, a.id === ending.id ? facts : null));
++ const out = chain.outcome(engineAttempts, ctx);
++ const step = out.step || { action: 'none' };
++ const reason = ending.reason || out.reason || 'error';
++ const words = wordsOf(reason);
++ const continues = upto.filter((a) => a.kind === 'continue').length;
++
++ if (step.action === 'done') return { outcome: 'done', note: 'the task finished' };
++ if (step.action === 'stop') return { outcome: 'stopped', note: 'the task was stopped on purpose' };
++ if (step.action === 'wait') return { outcome: null, note: null };
++
++ if (step.action === 'wrap-up') {
++ const hit = step.hit || 'steps';
++ return {
++ kind: 'wrap-up', agent: ending.agent, model: ending.model, session: ending.session || null,
++ prompt: String(task.prompt || '') + budget.wrapUpPrompt(hit),
++ why: `one last turn to land the work after the ${hit} limit`,
++ };
++ }
++
++ if (step.action === 'continue') {
++ if (continues >= maxContinues) {
++ return { outcome: 'unfinished', note: `the task was carried on ${continues} times already; it needs a person` };
++ }
++ return {
++ kind: 'continue', agent: ending.agent, model: ending.model, session: step.session || ending.session || null,
++ prompt: String(task.prompt || '') + continuationNote(ending, words),
++ why: `carried on in the same conversation: ${words}`,
++ };
++ }
++
++ if (step.action === 'restart') {
++ return {
++ kind: 'restart', agent: ending.agent, model: ending.model, session: null,
++ prompt: String(task.prompt || '') + continuationNote(ending, words),
++ why: `started again with a fresh session: ${words}`,
++ };
++ }
++
++ if (step.action === 'switch') {
++ const choice = nextAgent(task.prompt, ctx, runnerOf(ending.agent));
++ const agent = choice ? choice.agent : (step.runner ? HARNESS_OF[step.runner] : null);
++ if (!agent) return { outcome: 'unfinished', note: `no other agent could take it: ${step.why || words}` };
++ return {
++ kind: 'switch', agent, model: choice ? choice.model : (step.model || null), session: null,
++ prompt: String(task.prompt || '') + handoverNote(ending, result, words),
++ why: `${nameOf(agent)} took over: ${words}`,
++ };
++ }
++
++ return { outcome: 'unfinished', note: `${step.why || 'the task stopped'}. What is left is in the session.` };
++ }
++
++ /** An attempt as the API and the page show it. */
++ function present(a) {
++ if (!a) return null;
++ return {
++ id: a.id,
++ n: Number(a.n) || 0,
++ agent: a.agent || null,
++ model: a.model || null,
++ machine: a.machine || null,
++ job_id: a.job_id || null,
++ session: a.session || null,
++ state: a.state,
++ reason: a.reason || null,
++ why: a.why || null,
++ kind: a.kind || 'first',
++ cost_usd: a.cost_usd == null ? null : Number(a.cost_usd),
++ tokens: a.tokens == null ? null : Number(a.tokens),
++ steps: a.steps == null ? null : Number(a.steps),
++ started_at: iso(a.started_at),
++ finished_at: iso(a.finished_at),
++ created_at: iso(a.created_at),
++ end_why: a.reason ? wordsOf(a.reason) : null,
++ };
++ }
++
++ return {
++ STATES, REASONS, KINDS, REASON_TEXT, RUNNER_OF, HARNESS_OF,
++ list, get, byJob, create, bind, end, decide, present, routing, nextAgent, nameOf, wordsOf,
++ };
++}
++
++module.exports = { createAttempts, RUNNER_OF, HARNESS_OF, STATES, REASONS, KINDS, REASON_TEXT, ERROR_TEXT };
ddiff --git a/server/domain/run-tasks.js b/server/domain/run-tasks.js
iindex b8bc16b..d6f9450 100644
---- a/server/domain/run-tasks.js
++++ b/server/domain/run-tasks.js
@@@ -4,10 +4,13 @@
* 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, sessions, outbox })
++ * createRunTasks({ store, run, config, log, sessions, outbox, attempts })
* start({ subject, project, args, via, key }) → the row: queued (Run took the job) or failed (Run's reason)
* 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
++ * afterEnd(taskId) → the engine's decision after an attempt ended
++ * continueTask(row, { note, agent }) → the next attempt now (a switch when an agent is named)
++ * stopTask(row) → stop the running attempt; nothing carries it on
* latestProject(subject) → the project of the person's latest task, or null
* present(row) → codes.task@1
* TERMINAL, STATE_LABEL, STALE_MS
@@@ -15,8 +18,15 @@
* 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.
++ *
++ * Stage C2: every Run job is one attempt on the task's timeline (server/domain/attempts.js). When an attempt ends,
++ * the engine decides what happens next — done, a continue in the same session, a switch to another agent, a wrap-up or
++ * an unfinished task — and this module carries the decision out. One ending starts at most one follow-up: the
++ * follow-up's follow_up_of is unique, so two racing ticks never carry one job on twice.
*/
const { RunError } = require('./run-client');
++const { chain, budget } = require('../../engine');
++const { parseArgs } = require('../../harness/job');
const TERMINAL = new Set(['succeeded', 'failed', 'cancelled', 'expired']);
const STATE_LABEL = {
@@@ -85,12 +95,60 @@ function errorOf(job) {
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 }) {
++/** The machine a job ran on: the worker's device id, or the hosted sandbox. */
++function machineOf(job) {
++ if (!job || typeof job !== 'object') return null;
++ const node = job.placement && typeof job.placement === 'object' ? job.placement.node : null;
++ if (typeof node === 'string' && node) return node.slice(0, 128);
++ return typeof job.machine === 'string' && job.machine ? job.machine.slice(0, 128) : null;
++}
++
++/** The attempt's own state for a Run job's state. */
++const ATTEMPT_STATE = { queued: 'queued', placed: 'queued', running: 'running', succeeded: 'done', failed: 'failed', cancelled: 'stopped', expired: 'failed' };
++
++/** The reason an attempt ended: the engine's own words (engine/chain.js reason), plus the two it reads from the text. */
++function reasonOf(state, text) {
++ if (state === 'cancelled') return 'stopped';
++ if (state === 'expired') return 'capacity';
++ const s = String(text || '');
++ if (/went quiet|hung|stale/i.test(s)) return 'hung';
++ if (/prompt is too long|context length|maximum context|context window|too many tokens|input too long/i.test(s)) return 'context';
++ const why = chain.reason(s);
++ return why === 'unknown' ? 'error' : why;
++}
++
++/** What an attempt spent: its cost, its tokens, its steps and the session a resume needs (from the job's result log). */
++function usageOf(result) {
++ const d = asObject(result) || {};
++ const log = Array.isArray(d.log) ? d.log : [];
++ let tokens = 0; let steps = 0; let session = null; let model = null;
++ for (const line of log) {
++ if (!line || typeof line !== 'object') continue;
++ if (line.t === 'usage') tokens += (Number(line.input_tokens) || 0) + (Number(line.output_tokens) || 0);
++ else if (line.t === 'tool') steps++;
++ else if (line.t === 'agent') {
++ if (!session && line.session) session = String(line.session);
++ if (!model && line.model) model = String(line.model);
++ }
++ }
++ return {
++ cost_usd: typeof d.cost_usd === 'number' && Number.isFinite(d.cost_usd) ? d.cost_usd : null,
++ tokens: tokens || null,
++ steps: steps || null,
++ session: session || null,
++ model: model || null,
++ };
++}
++
++function createRunTasks({ store, run, config, log = console, sessions = null, outbox = null, attempts = 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 });
++ const attemptStore = attempts || require('./attempts').createAttempts({ store, config, log });
++ /** An attempt's state when it ended: a limit reached is a budget-stop, never a plain failure. */
++ const ENDED = new Set(['done', 'failed', 'stopped', 'budget-stopped', 'interrupted']);
/**
* One state change: update the row, write its codes.task.* outbox row in the same transaction, and on a terminal
@@@ -121,36 +179,79 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
[current.id, state, result == null ? null : JSON.stringify(result), error, startedAt, finishedAt, store.now()],
);
if (changed) await taskOutbox.emitTask(updated, current.state);
++ // Every Run job is one attempt: the attempt of this job follows the job.
++ const attempt = await attemptStore.byJob(current.job_id);
if (TERMINAL.has(updated.state) && !updated.session_complete_at) {
-- await sessionLines.backfillAndComplete(updated, updated.result);
++ await sessionLines.backfillAndComplete(updated, updated.result, attempt ? Number(attempt.n) : null);
++ }
++ if (attempt) {
++ if (TERMINAL.has(state)) {
++ const usage = usageOf(updated.result);
++ const why = reasonOf(state, updated.error);
++ const hit = budgetHit(attempt, usage, why);
++ await attemptStore.end(attempt.id, {
++ state: hit ? 'budget-stopped' : (ATTEMPT_STATE[state] || 'failed'),
++ reason: hit ? 'budget' : why,
++ cost_usd: usage.cost_usd, tokens: usage.tokens, steps: usage.steps, session: usage.session,
++ model: usage.model, machine: machineOf(next.job),
++ finished_at: updated.finished_at || store.iso(),
++ });
++ } else {
++ await attemptStore.bind(attempt.id, {
++ state: ATTEMPT_STATE[state] || 'queued', started_at: updated.started_at || null, machine: machineOf(next.job),
++ });
++ }
}
return updated;
});
}
++ /**
++ * The limit an attempt reached, if any (engine/budget.js): steps, minutes or dollars. A provider's own limit
++ * (usage limit, signed out) is not the owner's budget and never buys a wrap-up turn.
++ */
++ function budgetHit(attempt, usage, reason) {
++ if (!['error', 'hung', 'context'].includes(reason)) return null;
++ const limits = budget.limitsFor(config, attempt.agent, null);
++ if (!limits.steps && !limits.minutes && !limits.usd) return null;
++ const started = attempt.started_at ? Date.parse(String(attempt.started_at)) : 0;
++ const finished = attempt.finished_at ? Date.parse(String(attempt.finished_at)) : 0;
++ const minutes = finished > started ? (finished - started) / 60000 : null;
++ return budget.evaluate({ steps: usage.steps || 0, minutes, usd: usage.cost_usd }, limits).hit;
++ }
++
async function start({ subject, project, args, via = null, key = null, actorRef = null }) {
const id = store.newId('tsk');
const at = store.now();
const actorRefValue = actorRef && typeof actorRef === 'object' && !Array.isArray(actorRef) ? JSON.stringify(actorRef) : null;
++ let inserted;
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, actor_ref, created_at, updated_at)
++ inserted = await db.one(`INSERT INTO run_tasks (id, subject, project_id, repo, ref, prompt, agent, push, pull_request, state, via, idempotency_key, actor_ref, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'queued', $10, $11, $12::jsonb, $13, $13) RETURNING *`,
[id, subject, project, args.repo, args.ref, args.prompt, args.agent, !!args.push, !!(args.push && args.push.pull_request), via, key, actorRefValue, at]);
await taskOutbox.emitTask(inserted, null);
return inserted;
});
++ // The task's first attempt: every Run job is one attempt, and this one is the task as it was asked.
++ const first = await attemptStore.create({
++ task: inserted, agent: args.agent || null, model: args.model || null,
++ why: args.agent ? `the first try, with ${attemptStore.nameOf(args.agent)}` : 'the first try, with the agent that fits the task',
++ kind: 'first', state: 'queued',
++ });
let job;
try {
job = await run.submit({ subject, project, args, key: id });
} catch (e) {
const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
++ await attemptStore.end(first.attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
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()]);
++ const failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, outcome = 'unfinished', outcome_note = $3, finished_at = $4::timestamptz, updated_at = $5
++ WHERE id = $1 RETURNING *`, [id, why, `the task could not start: ${why}`, store.iso(), store.now()]);
await taskOutbox.emitTask(failed, 'queued');
return failed;
});
}
++ await attemptStore.bind(first.attempt.id, { job_id: job.id, state: 'queued' });
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;
@@@ -165,7 +266,7 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
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, {
++ const updated = await recordState(row, {
state: job.state,
result,
error: TERMINAL.has(job.state) ? errorOf(job) : undefined,
@@@ -173,6 +274,8 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
finishedAt: job.timings && job.timings.finished_at,
job,
});
++ // The attempt ended: the engine decides, and the row carries the outcome (or the next attempt).
++ return (await afterEnd(updated.id)) || updated;
}
/**
@@@ -180,7 +283,15 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
* 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 (!row.job_id) return row;
++ if (TERMINAL.has(row.state)) {
++ // The task has ended. Its decision (the outcome, or the attempt that follows) is made once, when the
++ // attempt ended; if that failed, it is made here, as the fallback reads Run after 10 quiet minutes.
++ if (!row.outcome && (force || store.now() - Number(row.updated_at) >= STALE_MS)) {
++ return (await afterEnd(row.id)) || row;
++ }
++ 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; }
@@@ -201,6 +312,160 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
});
}
++ /**
++ * An attempt ended: the engine decides what happens next (server/domain/attempts.js) and this carries the
++ * decision out — a follow-up attempt, or the task's outcome in plain words.
++ *
++ * Calling it twice for one ending changes nothing: the follow-up's follow_up_of is unique, so the second call
++ * finds the attempt the first one wrote and starts no second job (the portal's 10-30 bug, where two racing ticks
++ * carried one job on twice).
++ */
++ async function afterEnd(taskId) {
++ const row = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
++ if (!row) return null;
++ if (!TERMINAL.has(row.state)) return row;
++ if (row.outcome === 'stopped') return row; // a person stopped it: nothing carries it on
++ const attempts = await attemptStore.list(row.id);
++ const ending = attempts[attempts.length - 1];
++ if (!ending || !ENDED.has(ending.state)) return row;
++ const plan = await attemptStore.decide({ task: row, attempts, ending, result: row.result });
++ if (plan.outcome) {
++ return await db.one('UPDATE run_tasks SET outcome = $2, outcome_note = $3, updated_at = $4 WHERE id = $1 RETURNING *',
++ [row.id, plan.outcome, plan.note, store.now()]);
++ }
++ if (!plan.kind) return row;
++ const done = await startFollowUp(row, plan, ending);
++ return done.row || row;
++ }
++
++ /**
++ * The next attempt: a Run job for the task with the engine's plan in it. The attempt is written first and the job
++ * second, so the fence (follow_up_of) is in place before anything is started.
++ */
++ async function startFollowUp(row, plan, ending) {
++ const { attempt, created } = await attemptStore.create({
++ task: row, agent: plan.agent || null, model: plan.model || null, session: plan.session || null,
++ why: plan.why || null, kind: plan.kind || 'manual', followUpOf: ending.id, state: 'queued',
++ });
++ // Someone else already started the follow-up of this ending: no second job.
++ if (!created) return { attempt, created: false, row };
++ let args;
++ try {
++ args = followUpArgs(row, plan);
++ } catch (e) {
++ await attemptStore.end(attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
++ const failed = await db.one(`UPDATE run_tasks SET outcome = 'unfinished', outcome_note = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
++ [row.id, `the next attempt could not be prepared: ${e.message}`, store.now()]);
++ return { attempt, created: true, row: failed, error: e.message };
++ }
++ let job;
++ try {
++ job = await run.submit({ subject: row.subject, project: row.project_id, args, key: attempt.id });
++ } catch (e) {
++ const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
++ await attemptStore.end(attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
++ const failed = await db.one(`UPDATE run_tasks SET outcome = 'unfinished', outcome_note = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
++ [row.id, `the next attempt could not start: ${why}`, store.now()]);
++ return { attempt, created: true, row: failed, error: why };
++ }
++ await attemptStore.bind(attempt.id, { job_id: job.id, state: 'queued' });
++ const updated = await db.one(
++ `UPDATE run_tasks SET job_id = $2, state = 'queued', outcome = NULL, outcome_note = NULL, error = NULL,
++ finished_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *`,
++ [row.id, job.id, store.now()]);
++ await taskOutbox.emitTask(updated, row.state);
++ return { attempt, created: true, job, row: updated };
++ }
++
++ /**
++ * The next attempt now: the owner carries the task on (with a note), or switches to another agent. The task's row
++ * is locked while the attempt is written, so two presses start one attempt and not two.
++ */
++ async function continueTask(row, { note = null, agent = null } = {}) {
++ if (!row || !row.id) return { error: 'no such task' };
++ const attempts = await attemptStore.list(row.id);
++ const ending = attempts[attempts.length - 1];
++ if (!ending) return { error: 'the task has no attempt to carry on' };
++ const text = note ? String(note).trim().slice(0, 4000) : '';
++ const plan = agent
++ ? {
++ kind: 'switch', agent, model: null, session: null,
++ prompt: `${row.prompt}\n\n---\n## Hand-over (written by openvibe.codes)\n\n${attemptStore.nameOf(agent)} takes over from ${attemptStore.nameOf(ending.agent)}, which ${attemptStore.wordsOf(ending.reason || 'error')}. Keep the work that was done, do not redo it, and finish the task.\n`,
++ why: `${attemptStore.nameOf(agent)} took over at the owner's request`,
++ }
++ : {
++ kind: 'manual', agent: ending.agent, model: ending.model, session: null,
++ prompt: `${row.prompt}${text ? `\n\n---\n## A note from the owner\n\n${text}\n` : ''}`,
++ why: text ? `carried on by the owner: ${text.slice(0, 200)}` : 'carried on by the owner',
++ };
++ let attempt = null;
++ const locked = await store.tx(async () => {
++ const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
++ if (!current) return { error: 'no such task' };
++ if (!TERMINAL.has(current.state)) return { error: 'the task is still running' };
++ if (current.outcome === 'stopped') return { error: 'the task was stopped on purpose' };
++ const made = await attemptStore.create({
++ task: current, agent: plan.agent, model: plan.model, session: null, why: plan.why, kind: plan.kind, state: 'queued',
++ });
++ attempt = made.attempt;
++ await db.one(`UPDATE run_tasks SET state = 'queued', outcome = NULL, outcome_note = NULL, error = NULL, job_id = NULL, finished_at = NULL, updated_at = $2 WHERE id = $1 RETURNING *`,
++ [row.id, store.now()]);
++ return { row: current };
++ });
++ if (locked.error) return { error: locked.error };
++ let job;
++ try {
++ job = await run.submit({ subject: row.subject, project: row.project_id, args: followUpArgs(row, plan), key: attempt.id });
++ } catch (e) {
++ const why = e instanceof RunError ? `${e.code}: ${e.message}` : e.message;
++ await attemptStore.end(attempt.id, { state: 'failed', reason: 'error', finished_at: store.iso() });
++ const failed = await db.one(`UPDATE run_tasks SET state = 'failed', error = $2, outcome = 'unfinished', outcome_note = $3, finished_at = $4::timestamptz, updated_at = $5 WHERE id = $1 RETURNING *`,
++ [row.id, why, `the next attempt could not start: ${why}`, store.iso(), store.now()]);
++ return { error: why, row: failed, attempt };
++ }
++ await attemptStore.bind(attempt.id, { job_id: job.id, state: 'queued' });
++ const updated = await db.one('UPDATE run_tasks SET job_id = $2, updated_at = $3 WHERE id = $1 RETURNING *', [row.id, job.id, store.now()]);
++ await taskOutbox.emitTask(updated, row.state);
++ return { attempt, job, row: updated };
++ }
++
++ /** The job's arguments for a follow-up: the task's own repository, the plan's prompt and the session to resume. */
++ function followUpArgs(row, plan) {
++ return parseArgs(JSON.stringify({
++ repo: row.repo, ref: row.ref || null, prompt: plan.prompt || row.prompt,
++ agent: plan.agent || null, permission: 'edit',
++ ...(row.push ? { push: { pull_request: !!row.pull_request } } : {}),
++ ...(plan.session ? { resume: plan.session } : {}),
++ }));
++ }
++
++ /**
++ * Stop the running attempt: the outcome is stopped and nothing carries it on. The task's row is locked, so a stop
++ * and a follow-up never race.
++ */
++ async function stopTask(row) {
++ if (!row || !row.id) return { error: 'no such task' };
++ const attempts = await attemptStore.list(row.id);
++ const running = attempts.slice().reverse().find((a) => !ENDED.has(a.state));
++ if (row.job_id && !TERMINAL.has(row.state)) {
++ try { await run.cancel(row.job_id, who(row)); } catch (e) { log.warn(`[Codes] stop ${row.id}: ${e.message}`); }
++ }
++ const stopped = await store.tx(async () => {
++ const current = await db.maybe('SELECT * FROM run_tasks WHERE id = $1 FOR UPDATE', [row.id]);
++ if (!current) return { error: 'no such task' };
++ if (TERMINAL.has(current.state) && current.outcome) return { error: 'the task has already ended', row: current };
++ const updated = await db.one(
++ `UPDATE run_tasks SET state = 'cancelled', outcome = 'stopped', outcome_note = $2,
++ finished_at = COALESCE(finished_at, $3::timestamptz), updated_at = $4 WHERE id = $1 RETURNING *`,
++ [row.id, 'the task was stopped on purpose', store.iso(), store.now()]);
++ await taskOutbox.emitTask(updated, current.state);
++ return { row: updated };
++ });
++ if (stopped.error) return stopped;
++ if (running) await attemptStore.end(running.id, { state: 'stopped', reason: 'stopped', finished_at: store.iso() });
++ return { row: stopped.row, attempt: running || null };
++ }
++
async function latestProject(subject) {
const row = await db.maybe('SELECT project_id FROM run_tasks WHERE subject = $1 ORDER BY created_at DESC LIMIT 1', [subject]);
return row ? row.project_id : null;
@@@ -215,7 +480,10 @@ function createRunTasks({ store, run, config, log = console, sessions = null, ou
});
}
-- return { start, refresh, applyJob, applyEvent, recordState, latestProject, present, who };
++ return {
++ start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask,
++ latestProject, present, who, attempts: attemptStore,
++ };
}
--module.exports = { createRunTasks, presentResult, TERMINAL, STATE_LABEL, STATES, RANK, STALE_MS, sessionLogOf, errorOf, asObject };
++module.exports = { createRunTasks, presentResult, TERMINAL, STATE_LABEL, STATES, RANK, STALE_MS, sessionLogOf, errorOf, asObject, ATTEMPT_STATE, reasonOf, usageOf };
ddiff --git a/server/domain/session-lines.js b/server/domain/session-lines.js
iindex 7bc210b..6471ac6 100644
---- a/server/domain/session-lines.js
++++ b/server/domain/session-lines.js
@@@ -98,7 +98,7 @@ function createSessionLines({ store, config = {}, log = console }) {
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) {
++ async function append(taskId, lines, attemptN = null) {
if (!taskId || !Array.isArray(lines) || !lines.length) return 0;
const clean = [];
for (const line of lines) {
@@@ -114,12 +114,12 @@ function createSessionLines({ store, config = {}, log = console }) {
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));
++ const b = j * 6;
++ values.push(`($${b + 1}, $${b + 2}, $${b + 3}::timestamptz, $${b + 4}, $${b + 5}::jsonb, $${b + 6})`);
++ params.push(taskId, line.seq, line.at, line.kind, JSON.stringify(line.body), attemptN);
});
inserted += await db.exec(
-- `INSERT INTO codes_session_lines (task_id, seq, at, kind, body) VALUES ${values.join(', ')}
++ `INSERT INTO codes_session_lines (task_id, seq, at, kind, body, attempt_n) VALUES ${values.join(', ')}
ON CONFLICT (task_id, seq) DO NOTHING`, params);
}
return inserted;
@@@ -127,7 +127,7 @@ function createSessionLines({ store, config = {}, log = console }) {
/** 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]);
++ return await db.many('SELECT seq, at, kind, body, attempt_n FROM codes_session_lines WHERE task_id = $1 ORDER BY seq ASC', [taskId]);
}
/**
@@@ -135,7 +135,7 @@ function createSessionLines({ store, config = {}, log = console }) {
* 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) {
++ function recorder(taskId, { attemptN = null } = {}) {
let frameBuffer = '';
let outputBuffer = '';
let fallbackSeq = 0;
@@@ -196,7 +196,7 @@ function createSessionLines({ store, config = {}, log = console }) {
frameBuffer = `${frameBuffer}${String(text == null ? '' : text)}`.replace(/\r\n/g, '\n');
const out = [];
drainFrames(out);
-- if (out.length) await append(taskId, out);
++ if (out.length) await append(taskId, out, attemptN);
},
async flush() {
const out = [];
@@@ -208,7 +208,7 @@ function createSessionLines({ store, config = {}, log = console }) {
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);
++ if (out.length) await append(taskId, out, attemptN);
},
ended: () => ended,
};
@@@ -219,7 +219,7 @@ function createSessionLines({ store, config = {}, log = console }) {
* 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) {
++ async function backfill(row, result, attemptN = null) {
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)
@@@ -242,7 +242,7 @@ function createSessionLines({ store, config = {}, log = console }) {
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,
-- })));
++ })), attemptN);
}
async function complete(taskId) {
@@@ -252,8 +252,8 @@ function createSessionLines({ store, config = {}, log = console }) {
);
}
-- async function backfillAndComplete(row, result) {
-- const inserted = await backfill(row, result);
++ async function backfillAndComplete(row, result, attemptN = null) {
++ const inserted = await backfill(row, result, attemptN);
await complete(row.id);
return inserted;
}
@@@ -276,11 +276,11 @@ function createSessionLines({ store, config = {}, log = console }) {
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 }));
++ lines = rows.filter((r) => Number(r.seq) > from).map((r) => ({ seq: Number(r.seq), at: r.at, kind: r.kind, body: r.body, attempt_n: r.attempt_n == null ? null : Number(r.attempt_n) }));
} 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 }))
++ lines = (sessionLogOf(row) || []).map((body, i) => ({ seq: i + 1, at, kind: typeof body.t === 'string' ? body.t : 'text', body, attempt_n: 1 }))
.filter((line) => line.seq > from);
}
const page = lines.slice(0, max);
ddiff --git a/server/events-consumer.js b/server/events-consumer.js
iindex 72928e1..24b1f15 100644
---- a/server/events-consumer.js
++++ b/server/events-consumer.js
@@@ -77,7 +77,7 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
let job = null;
if (TERMINAL.has(state)) job = await run.read(jobId, tasks.who(row));
-- return await store.tx(async () => {
++ const applied = 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`,
@@@ -90,6 +90,13 @@ function createEventsConsumer({ config, store, tasks, run, accountData, accountS
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 };
});
++ // The attempt ended: the engine decides what happens next. It runs outside the transaction, because a
++ // follow-up starts a Run job. A failure here is logged, not fatal: the delivery is applied, and the decision
++ // is made again by the fallback (the read of Run after 10 quiet minutes).
++ if (!applied.duplicate) {
++ try { await tasks.afterEnd(applied.task_id); } catch (err) { log.warn(`[Codes] task ${applied.task_id} decision failed: ${err.message}`); }
++ }
++ return applied;
}
const router = express.Router();
ddiff --git a/server/http/run.js b/server/http/run.js
iindex 109765d..99d78de 100644
---- a/server/http/run.js
++++ b/server/http/run.js
@@@ -5,7 +5,8 @@
* GET /run the form, the person's projects and their recent tasks (a sign-in prompt for a guest)
* POST /run start one: a run_tasks row and an OpenVibe Run job for the person (server/domain/run-client.js)
* POST /run/project create a project for the runs when the person has none they may run jobs in
-- * GET /run/tasks/:id one task: live while it runs, then the agents' summary, the changed files and the patch
++ * GET /run/tasks/:id one task: its attempts (the timeline), live while it runs, then the agents' summary,
++ * the changed files and the patch
* GET /run/tasks/:id/events its live lines, proxied from Run's job stream (server-sent events)
* POST /run/tasks/:id/cancel stop it
*
@@@ -26,6 +27,9 @@ const DELEGABLE = new Set(['owner', 'admin']);
const MACHINES_URL = 'https://openvibe.bot/machines';
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++/** The task's outcome in the words a person reads. */
++const OUTCOME_TEXT = { done: 'Done', unfinished: "Didn't finish", stopped: 'Stopped by you' };
++
/** 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) {
const [host, ...path] = String(url).replace(/^https:\/\//, '').split('/');
@@@ -182,6 +186,24 @@ ${typeof d.duration_ms === 'number' ? html`<p class="muted">${Math.round(d.durat
</section>`;
}
++ /** The task's timeline: every attempt and why it happened. The data-* hooks are for the design of stage C4. */
++ async function timelineHtml(row) {
++ const attempts = await taskStore.attempts.list(row.id);
++ if (!attempts.length) return '';
++ const outcome = row.outcome
++ ? html`<p class="run-outcome" data-outcome="${row.outcome}"><b>${OUTCOME_TEXT[row.outcome] || row.outcome}</b>${row.outcome_note ? html` ${row.outcome_note}` : ''}</p>`
++ : '';
++ return html`<section class="run-timeline" data-timeline data-attempts="${attempts.length}">
++<h2>The attempts</h2>
++<ol class="run-attempts">${attempts.map((a) => html`<li class="run-attempt" data-attempt="${a.n}" data-attempt-id="${a.id}" data-state="${a.state}" data-reason="${a.reason || ''}" data-kind="${a.kind || 'first'}" data-agent="${a.agent || ''}">
++<p class="run-attempt-head"><b>Attempt ${a.n}</b>${a.agent ? html` · ${a.agent}` : ''} <span class="run-pill run-pill-${a.state}">${a.state}</span></p>
++<p class="run-attempt-why" data-why>${a.why || ''}</p>
++${a.reason ? html`<p class="run-attempt-end" data-end-why>${taskStore.attempts.wordsOf(a.reason)}</p>` : ''}
++</li>`)}</ol>
++${outcome}
++</section>`;
++ }
++
r.get('/run/tasks/:id', async (req, res) => {
if (!signedIn(req)) return res.redirect(302, `/auth/login?next=${encodeURIComponent(req.originalUrl)}`);
let row = await own(req, req.params.id);
@@@ -197,6 +219,7 @@ ${typeof d.duration_ms === 'number' ? html`<p class="muted">${Math.round(d.durat
<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>
++${await timelineHtml(row)}
${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 && lines.length) })}`}`;
return page(req, res, { title: 'Task', body, sharedScripts: ['agent-session.js'] });
@@@ -242,7 +265,9 @@ ${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events
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 attempts = await taskStore.attempts.list(row.id);
++ const attemptN = attempts.length ? Number(attempts[attempts.length - 1].n) : null;
++ const recorder = sessionStore.recorder(row.id, { attemptN });
const decoder = new TextDecoder();
try {
for await (const chunk of resp.body) {
@@@ -286,13 +311,8 @@ ${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events
if (!sameOrigin(req)) return res.status(403).type('text/plain').send('Cross-site form posts are refused.\n');
const row = await own(req, req.params.id);
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}`); }
-- }
++ // The running attempt stops: the outcome is "stopped" and nothing carries it on.
++ try { await taskStore.stopTask(row); } catch (e) { log.warn(`[Codes] cancel ${row.id}: ${e.message}`); }
return res.redirect(303, `/run/tasks/${row.id}`);
});
ddiff --git a/server/http/task-events.js b/server/http/task-events.js
nnew file mode 100644
iindex 0000000..8fd1f36
---- /dev/null
++++ b/server/http/task-events.js
@@@ -0,0 +1,151 @@
++'use strict';
++
++/**
++ * A task's event stream (GET /api/v1/tasks/:id/events): the session's lines in the order they happened, the attempt
++ * each line belongs to, and the attempts themselves (started, ended with why). The stream is resumable: a client
++ * that reconnects sends the last line's seq as Last-Event-ID and gets only what came after it. A heartbeat every 15
++ * seconds keeps the connection alive through a proxy that would otherwise drop it.
++ *
++ * createTaskEvents({ store, config, log, attempts })
++ * stream(req, res, row) the stream of one task
++ *
++ * The lines come from codes_session_lines (the session the machine streamed, back-filled from its final result) and
++ * the attempts from codes_attempts. Nothing here writes anything: the stream only reads.
++ */
++
++const { TERMINAL } = require('../domain/run-tasks');
++
++const POLL_MS = 1000;
++const PAGE = 500;
++const ENDED = ['done', 'failed', 'stopped', 'budget-stopped', 'interrupted'];
++
++function createTaskEvents({ store, config = {}, log = console, attempts = null }) {
++ const db = store.db;
++ const heartbeatMs = Math.max(1000, Number(config.sse && config.sse.heartbeatMs) || 15_000);
++ const attemptStore = attempts || require('../domain/attempts').createAttempts({ store, config, log });
++
++ /** One SSE frame. */
++ const frame = (event, data, id = null) => (id == null ? '' : `id: ${id}\n`) + `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
++
++ const started = (a) => ({
++ type: 'attempt_started', attempt_n: Number(a.n), id: a.id, agent: a.agent || null, model: a.model || null,
++ why: a.why || null, state: 'queued', at: a.created_at ? new Date(a.created_at).toISOString() : null,
++ });
++ const ended = (a) => ({
++ type: 'attempt_ended', attempt_n: Number(a.n), id: a.id, agent: a.agent || null, model: a.model || null,
++ why: a.why || null, state: a.state, reason: a.reason || null,
++ end_why: a.reason ? attemptStore.wordsOf(a.reason) : null,
++ cost_usd: a.cost_usd == null ? null : Number(a.cost_usd),
++ tokens: a.tokens == null ? null : Number(a.tokens),
++ steps: a.steps == null ? null : Number(a.steps),
++ at: a.finished_at ? new Date(a.finished_at).toISOString() : null,
++ });
++
++ /** The stream of one task: its lines, its attempts, a heartbeat, and an end when the task has ended. */
++ async function stream(req, res, row) {
++ res.writeHead(200, {
++ 'Content-Type': 'text/event-stream',
++ 'Cache-Control': 'no-store',
++ 'X-Accel-Buffering': 'no',
++ Connection: 'keep-alive',
++ });
++ if (res.flushHeaders) res.flushHeaders();
++
++ const asked = String(req.get('last-event-id') || (req.query && req.query.last_event_id) || '0').trim();
++ let lastSeq = /^\d+$/.test(asked) ? Number(asked) : 0;
++ let closed = false;
++ let busy = false;
++ let terminalAt = 0;
++
++ const write = (text) => {
++ if (closed || !text) return;
++ res.write(text);
++ };
++
++ // The attempts a client has not seen: from the attempt its last line belongs to.
++ const anchor = lastSeq
++ ? await db.maybe('SELECT attempt_n FROM codes_session_lines WHERE task_id = $1 AND seq = $2', [row.id, lastSeq])
++ : null;
++ const fromN = anchor && anchor.attempt_n ? Number(anchor.attempt_n) : 1;
++ const known = await attemptStore.list(row.id);
++ const sent = new Map(known.map((a) => [a.id, a.state]));
++ let out = '';
++ for (const a of known) {
++ if (Number(a.n) < fromN) continue;
++ out += frame('attempt', started(a));
++ if (ENDED.includes(a.state)) out += frame('attempt', ended(a));
++ }
++ write(out);
++
++ const sendLines = async () => {
++ const lines = await db.many(
++ 'SELECT seq, at, kind, body, attempt_n FROM codes_session_lines WHERE task_id = $1 AND seq > $2 ORDER BY seq ASC LIMIT $3',
++ [row.id, lastSeq, PAGE],
++ );
++ for (const line of lines) {
++ write(frame('line', {
++ seq: Number(line.seq),
++ at: line.at ? new Date(line.at).toISOString() : null,
++ kind: line.kind,
++ attempt_n: line.attempt_n == null ? null : Number(line.attempt_n),
++ body: line.body,
++ }, Number(line.seq)));
++ lastSeq = Math.max(lastSeq, Number(line.seq));
++ }
++ return lines.length;
++ };
++ await sendLines();
++
++ const tick = async () => {
++ if (closed || busy) return;
++ busy = true;
++ try {
++ const sentLines = await sendLines();
++ const fresh = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);
++ const rows = await attemptStore.list(row.id);
++ let updates = '';
++ for (const a of rows) {
++ const was = sent.get(a.id);
++ if (was === a.state) continue;
++ sent.set(a.id, a.state);
++ if (was === undefined) updates += frame('attempt', started(a));
++ if (ENDED.includes(a.state)) updates += frame('attempt', ended(a));
++ }
++ write(updates);
++ if (fresh && TERMINAL.has(fresh.state) && !rows.some((a) => ['queued', 'running'].includes(a.state))) {
++ if (!terminalAt) terminalAt = Date.now();
++ // The task has ended and its outcome is known: the stream is done. A task whose decision is still
++ // being made (no outcome yet) keeps the stream a little longer, so a follow-up is not missed.
++ if (fresh.outcome || Date.now() - terminalAt > 10_000) {
++ closed = true;
++ clearInterval(poll);
++ clearInterval(beat);
++ res.end();
++ }
++ } else {
++ terminalAt = 0;
++ }
++ } catch (e) {
++ log.warn(`[Codes] task ${row.id} events: ${e.message}`);
++ } finally {
++ busy = false;
++ }
++ };
++
++ const poll = setInterval(tick, POLL_MS);
++ const beat = setInterval(() => write(': heartbeat\n\n'), heartbeatMs);
++ const stop = () => {
++ if (closed) return;
++ closed = true;
++ clearInterval(poll);
++ clearInterval(beat);
++ };
++ req.on('close', stop);
++ res.on('close', stop);
++ return undefined;
++ }
++
++ return { stream, heartbeatMs };
++}
++
++module.exports = { createTaskEvents };
ddiff --git a/server/http/tasks-api.js b/server/http/tasks-api.js
iindex 10a3a5c..f7d7fbe 100644
---- a/server/http/tasks-api.js
++++ b/server/http/tasks-api.js
@@@ -7,19 +7,29 @@
* machine, in the project of their latest /run task.
*
* POST /tasks codes.task.start codes.task-start-request@1 → codes.task-start-result@1 (201, or 200 again)
-- * GET /tasks/:id codes.task.read → codes.task@1
++ * GET /tasks/:id codes.task.read → codes.task@2 (the task as codes.task@1, with its attempts and
++ * its outcome added to it)
* POST /tasks/:id/cancel codes.task.cancel → codes.task@1
++ * GET /tasks/:id/events codes.task.read → the task's event stream (server-sent events)
++ * POST /tasks/:id/continue codes.task.start → the next attempt now (a switch when an agent is named)
++ * POST /tasks/:id/stop codes.task.cancel → the running attempt stops; nothing carries it on
*
* A Network service token for audience openvibe.codes (CODES_AUDIENCE), from a svc: principal holding the route's
-- * capability. A service sees only the tasks it started, for that person; anything else is 404. The person's task
-- * page (the codes.task@1 url) shows every task they own, whoever started it. Calls count against the person's own
-- * per-actor budget (actor-limits.js), not the service's.
++ * capability. A service sees only the tasks it started, for that person; anything else is 404. The owner may read
++ * and act on their own tasks without a token. The person's task page (the codes.task@1 url) shows every task they
++ * own, whoever started it. Calls count against the person's own per-actor budget (actor-limits.js), not the service's.
++ *
++ * Stage C2: a task's timeline is its attempts. GET /tasks/:id answers codes.task@2: the task itself with the same
++ * fields it always had, and, added to them, its attempts and its outcome in plain words ("done", "unfinished" with
++ * what is left, or "stopped"). GET /tasks/:id/events replays the session's lines in order — resumable with
++ * Last-Event-ID, a heartbeat every 15 s — with the attempt each line belongs to.
*/
const express = require('express');
const { serviceAuth, capabilities, http, validate } = require('openvibe-contracts');
const { asyncRouter } = require('./router');
const { parseArgs } = require('../../harness/job');
const { createRunTasks, TERMINAL } = require('../domain/run-tasks');
++const { createTaskEvents } = require('./task-events');
const USER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
@@@ -28,6 +38,7 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
const r = asyncRouter();
const db = store.db;
const taskStore = tasks || createRunTasks({ store, run, config, log });
++ const taskEvents = createTaskEvents({ store, config, log, attempts: taskStore.attempts });
const problem = (req, res, status, code, detail) => http.sendProblem(res, status, code, { detail, ctx: req.ov });
const json = express.json({ limit: '32kb' });
@@@ -53,10 +64,32 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
/** The task this service started for this person, or null. */
const mine = async (req) => {
if (!TASK.test(String(req.params.id))) return null;
++ if (req.principal.owner) return db.maybe('SELECT * FROM run_tasks WHERE id = $1 AND subject = $2', [req.params.id, req.principal.subject]);
return db.maybe('SELECT * FROM run_tasks WHERE id = $1 AND subject = $2 AND via = $3', [req.params.id, req.principal.subject, req.principal.via]);
};
const byKey = (p, key) => db.maybe('SELECT * FROM run_tasks WHERE subject = $1 AND via = $2 AND idempotency_key = $3', [p.subject, p.via, key]);
++ /** The owner (signed in), or a service token with the capability acting for the person in X-OV-Subject. */
++ const signedIn = (req) => req.viewer && req.viewer.kind === 'user' && req.viewer.subject;
++ const guardAny = (cap) => async (req, res, next) => {
++ if (String(req.headers.authorization || '').startsWith('Bearer ')) return guard(cap)(req, res, next);
++ if (signedIn(req)) {
++ req.principal = { sub: `user:${req.viewer.subject}`, subject: req.viewer.subject, via: null, owner: true };
++ return next();
++ }
++ return problem(req, res, 401, 'token.required', `${cap} needs a Network service token for ${config.serviceAudience}, or a signed-in owner`);
++ };
++
++ /** The task, its attempts and its outcome in plain words. The task's own fields stay at the top level. */
++ const timeline = async (row) => {
++ const attempts = await taskStore.attempts.list(row.id);
++ return {
++ ...taskStore.present(row),
++ attempts: attempts.map((a) => taskStore.attempts.present(a)),
++ outcome: { state: row.outcome || null, note: row.outcome_note || null },
++ };
++ };
++
r.post('/tasks', guard('codes.task.start'), actorLimits.budget('codes.task.start'), json, async (req, res) => {
const b = req.body;
// Actor back-reference: the contract does not carry these yet, so they are stripped before validation and stored
@@@ -102,10 +135,41 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
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) => {
++ r.get('/tasks/:id', guardAny('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(await timeline(await taskStore.refresh(row)));
++ });
++
++ /** The task's event stream: the session's lines in order, resumable with Last-Event-ID, plus the attempts. */
++ r.get('/tasks/:id/events', guardAny('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 await taskEvents.stream(req, res, row);
++ });
++
++ /** The next attempt now: the owner carries the task on, or hands it to another agent. */
++ r.post('/tasks/:id/continue', guardAny('codes.task.start'), actorLimits.budget('codes.task.start'), json, 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');
++ const b = req.body || {};
++ const note = typeof b.note === 'string' ? b.note.slice(0, 4000) : null;
++ const agent = typeof b.agent === 'string' && b.agent ? b.agent : null;
++ if (agent && !harnesses.list().some((a) => a.id === agent)) return problem(req, res, 422, 'codes.agent_unknown', 'agent is one of GET /api/v1/harnesses');
++ const out = await taskStore.continueTask(row, { note, agent });
++ if (out.error) return problem(req, res, 409, 'codes.task_conflict', out.error);
++ log.info(`[Codes] task ${row.id} carried on (${agent ? `with ${agent}` : 'by the owner'})`);
++ return res.json(await timeline(out.row));
++ });
++
++ /** The running attempt stops: the outcome is "stopped" and nothing carries it on. */
++ r.post('/tasks/:id/stop', guardAny('codes.task.cancel'), actorLimits.budget('codes.task.start'), 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(taskStore.present(await taskStore.refresh(row)));
++ const out = await taskStore.stopTask(row);
++ if (out.error) return problem(req, res, 409, 'codes.task_conflict', out.error);
++ log.info(`[Codes] task ${row.id} stopped by ${req.principal.via || req.principal.subject}`);
++ return res.json(await timeline(out.row));
});
r.post('/tasks/:id/cancel', guard('codes.task.cancel'), actorLimits.budget('codes.task.start'), async (req, res) => {
ddiff --git a/test/attempts.test.js b/test/attempts.test.js
nnew file mode 100644
iindex 0000000..89e1dcc
---- /dev/null
++++ b/test/attempts.test.js
@@@ -0,0 +1,288 @@
++'use strict';
++
++/**
++ * Stage C2: a task's attempts. Every Run job is one attempt; when an attempt ends the engine decides what happens
++ * next — done, a continue in the same conversation, a switch to another agent, a wrap-up after a limit, or an
++ * unfinished task — and one ending starts at most one follow-up (the portal's 10-30 bug, where two racing ticks
++ * carried one job on twice).
++ *
++ * The migration (an existing task becomes attempt 1), every engine decision, the fence under two concurrent endings,
++ * the stop fence, and the auth of every new route.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++
++const TASK = /^tsk_[0-9A-HJKMNP-TV-Z]{26}$/;
++
++(async () => {
++ const calls = [];
++ const jobs = new Map();
++ let counter = 0;
++ const run = {
++ enabled: true,
++ async submit(o) {
++ counter++;
++ const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${String(counter).padStart(2, '0')}`;
++ jobs.set(id, { id, state: 'queued', args: o.args });
++ calls.push(['submit', o]);
++ return { id, state: 'queued' };
++ },
++ async read(jobId) { return jobs.get(jobId) || { id: jobId, state: 'queued' }; },
++ async cancel(jobId, who) { calls.push(['cancel', jobId, who]); const j = jobs.get(jobId); if (j) j.state = 'cancelled'; return j; },
++ async streamUrl() { throw new Error('not used'); },
++ };
++ const t = await boot({ run, env: { CODES_TASK_STEPS: '2' } });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const bo = t.network.addUser('bo');
++ const CAPS = ['codes.task.start', 'codes.task.read', 'codes.task.cancel'];
++ const actor = t.network.mintService('svc:actor', 'openvibe.codes', CAPS);
++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
++
++ const call = async (method, path, { token = actor, subject = ada.subject, body, headers = {} } = {}) => {
++ const h = { Accept: 'application/json', ...headers, ...(token ? { Authorization: `Bearer ${token}` } : {}), ...(subject ? { 'X-OV-Subject': subject } : {}) };
++ if (body !== undefined) h['Content-Type'] = 'application/json';
++ const res = await fetch(`${t.base}/api/v1${path}`, { method, headers: h, body: body === undefined ? undefined : JSON.stringify(body) });
++ return { status: res.status, json: await res.json().catch(() => null) };
++ };
++
++ /** A task for Ada, started through the API (so it has its first attempt). */
++ const startTask = async (prompt = 'Fix the parser and add a test for it.') => {
++ const r = await call('POST', '/tasks', { body: { repo: 'https://github.com/octocat/Hello-World', prompt, idempotency_key: `act_${Math.random().toString(36).slice(2, 10)}` } });
++ assert.strictEqual(r.status, 201, JSON.stringify(r.json));
++ return r.json.task.id;
++ };
++
++ /** The attempt ends: the state change and then the engine's decision (what the events consumer does). */
++ const endAttempt = async (taskId, { state = 'failed', result = null, error = null } = {}) => {
++ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
++ const job = { id: row.job_id, state, result, error: error ? { code: 'run.job.worker_failed', detail: error } : null };
++ const updated = await t.ctx.tasks.applyEvent(row, { state, job, error });
++ await t.ctx.tasks.afterEnd(updated.id);
++ return await rowOf(updated.id);
++ };
++
++ const attemptsOf = async (taskId) => await db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);
++ const rowOf = async (taskId) => await db.one('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
++
++ try {
++ await check('the migration: every task that exists becomes attempt 1 of itself', async () => {
++ // A task from before the attempts (written straight into the table, as an old row would be).
++ const old = 'tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T9';
++ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, agent, state, created_at, updated_at)
++ VALUES ($1, $2, $3, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J99', 'https://github.com/a/b', 'by hand', 'claude-code', 'succeeded', 1, 2)`, [old, ada.subject, project]);
++ const read = await t.ctx.attempts.list(old);
++ assert.strictEqual(read.length, 1, 'one attempt');
++ assert.strictEqual(Number(read[0].n), 1);
++ assert.strictEqual(read[0].state, 'done', 'the task succeeded, so its first attempt did');
++ assert.ok(read[0].why, 'the attempt says why it happened');
++ const shown = await t.get(`/api/v1/tasks/${old}`, { as: ada });
++ assert.strictEqual(shown.status, 200, shown.text.slice(0, 200));
++ assert.strictEqual(shown.json().attempts.length, 1);
++ assert.strictEqual(shown.json().attempts[0].n, 1);
++ });
++
++ await check('a task starts as one attempt, and the timeline says why', async () => {
++ const id = await startTask();
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 1);
++ assert.strictEqual(Number(attempts[0].n), 1);
++ assert.strictEqual(attempts[0].state, 'queued');
++ assert.strictEqual(attempts[0].kind, 'first');
++ assert.ok(attempts[0].job_id, 'the attempt carries the Run job');
++ const read = await call('GET', `/tasks/${id}`);
++ assert.strictEqual(read.json.id, id, 'the task at the top level, as codes.task@1 had it');
++ assert.strictEqual(read.json.state, 'queued');
++ assert.ok(read.json.url, 'the task carries its own fields');
++ assert.strictEqual(read.json.attempts.length, 1);
++ assert.strictEqual(read.json.outcome.state, null, 'a running task has no outcome yet');
++ });
++
++ await check('a finished attempt: the task is done, and nothing follows it', async () => {
++ const id = await startTask();
++ const updated = await endAttempt(id, { state: 'succeeded', result: { ok: true, summary: 'all done', files: [], log: [{ t: 'done', ok: true, text: 'all done' }] } });
++ assert.strictEqual(updated.outcome, 'done');
++ assert.strictEqual((await attemptsOf(id)).length, 1, 'no follow-up');
++ const read = await call('GET', `/tasks/${id}`);
++ assert.strictEqual(read.json.outcome.state, 'done');
++ assert.strictEqual(read.json.attempts[0].state, 'done');
++ });
++
++ await check('a crash: the same conversation continues, with the session to resume', async () => {
++ const id = await startTask();
++ const result = {
++ ok: false, summary: 'I will continue with the rest of it', files: [],
++ log: [{ t: 'agent', agent: 'claude-code', session: 'sess-42' }, { t: 'done', ok: false, text: 'I will continue with the rest of it' }],
++ };
++ await endAttempt(id, { state: 'failed', result, error: 'the machine went away' });
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 2, 'the task was carried on');
++ assert.strictEqual(attempts[1].kind, 'continue');
++ assert.strictEqual(attempts[1].session, 'sess-42', 'the session to resume');
++ assert.ok(/^carried on in the same conversation: /.test(attempts[1].why), attempts[1].why);
++ // The new attempt is a Run job of its own, with RESUME set to the session.
++ const submits = calls.filter((c) => c[0] === 'submit');
++ const last = submits[submits.length - 1][1];
++ assert.strictEqual(last.args.resume, 'sess-42');
++ assert.ok(last.args.prompt.includes('Continuation'), 'the prompt tells the agent to carry on');
++ assert.notStrictEqual(last.key, id, 'the follow-up has its own idempotency key');
++ });
++
++ await check('a limit: another agent takes over, with a hand-over note of what the last one did', async () => {
++ const id = await startTask();
++ const result = { ok: false, summary: 'I fixed the parser', files: [{ path: 'src/parse.js', added: 3, removed: 1 }], log: [{ t: 'done', ok: false, text: 'usage limit reached' }] };
++ await endAttempt(id, { state: 'failed', result, error: 'usage limit reached' });
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 2);
++ assert.strictEqual(attempts[1].kind, 'switch');
++ assert.ok(attempts[1].agent && attempts[1].agent !== attempts[0].agent, `${attempts[1].agent} took over from ${attempts[0].agent}`);
++ assert.ok(/took over: /.test(attempts[1].why), attempts[1].why);
++ const submits = calls.filter((c) => c[0] === 'submit');
++ const last = submits[submits.length - 1][1];
++ assert.ok(last.args.prompt.includes('Hand-over'), 'the new agent gets a hand-over note');
++ assert.ok(last.args.prompt.includes('src/parse.js'), 'the note says what changed');
++ assert.ok(last.args.prompt.includes('I fixed the parser'), 'the note says what it said');
++ });
++
++ await check('a limit of the owner: one last turn to land the work', async () => {
++ const id = await startTask();
++ const result = {
++ ok: false, summary: 'half done', files: [],
++ log: [{ t: 'tool', id: 't1', name: 'Edit' }, { t: 'tool', id: 't2', name: 'Edit' }, { t: 'done', ok: false, text: 'maximum turns' }],
++ };
++ await endAttempt(id, { state: 'failed', result, error: 'the agent stopped' });
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 2);
++ assert.strictEqual(attempts[0].state, 'budget-stopped', 'a limit is not a failure');
++ assert.strictEqual(attempts[0].reason, 'budget');
++ assert.strictEqual(attempts[1].kind, 'wrap-up');
++ assert.ok(/one last turn to land the work/.test(attempts[1].why), attempts[1].why);
++ const submits = calls.filter((c) => c[0] === 'submit');
++ assert.ok(submits[submits.length - 1][1].args.prompt.includes('one short last turn'), 'the wrap-up prompt');
++ });
++
++ await check('the same failure twice: the task is unfinished, with what is left', async () => {
++ const id = await startTask();
++ const fail = { ok: false, summary: 'it broke', files: [], log: [{ t: 'done', ok: false, text: 'the tool call failed' }] };
++ await endAttempt(id, { state: 'failed', result: fail, error: 'the tool call failed' });
++ await endAttempt(id, { state: 'failed', result: fail, error: 'the tool call failed' });
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 2, 'a second opinion, and no more');
++ const row = await rowOf(id);
++ assert.strictEqual(row.outcome, 'unfinished');
++ assert.ok(row.outcome_note && row.outcome_note.length > 10, 'the outcome says what is left');
++ });
++
++ await check('one ending starts at most one follow-up, even when two endings race', async () => {
++ const id = await startTask();
++ const before = calls.filter((c) => c[0] === 'submit').length;
++ const row = await rowOf(id);
++ const job = { id: row.job_id, state: 'failed', result: { ok: false, summary: 'x', files: [], log: [{ t: 'done', ok: false, text: 'usage limit reached' }] }, error: { code: 'run.job.worker_failed', detail: 'usage limit reached' } };
++ // Two ticks see the same ending at the same time (the portal's 10-30 bug).
++ await Promise.all([
++ t.ctx.tasks.applyEvent(row, { state: 'failed', job, error: 'usage limit reached' }),
++ t.ctx.tasks.applyEvent(row, { state: 'failed', job, error: 'usage limit reached' }),
++ ]);
++ await Promise.all([t.ctx.tasks.afterEnd(id), t.ctx.tasks.afterEnd(id)]);
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.length, 2, 'one follow-up, not two');
++ const submits = calls.filter((c) => c[0] === 'submit').length - before;
++ assert.strictEqual(submits, 1, 'one job started, not two');
++ });
++
++ await check('a stop fences the task: the outcome is stopped and nothing carries it on', async () => {
++ const id = await startTask();
++ const before = calls.filter((c) => c[0] === 'submit').length;
++ const stopped = await call('POST', `/tasks/${id}/stop`);
++ assert.strictEqual(stopped.status, 200, JSON.stringify(stopped.json));
++ assert.strictEqual(stopped.json.outcome.state, 'stopped');
++ assert.ok(calls.some((c) => c[0] === 'cancel'), 'the job was cancelled');
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts[0].state, 'stopped');
++ assert.strictEqual(attempts[0].reason, 'stopped');
++ // The machine's late report changes nothing.
++ await endAttempt(id, { state: 'failed', error: 'the machine went away' });
++ assert.strictEqual((await attemptsOf(id)).length, 1, 'nothing carried it on');
++ assert.strictEqual(calls.filter((c) => c[0] === 'submit').length - before, 0, 'no job was started');
++ assert.strictEqual((await rowOf(id)).outcome, 'stopped');
++ });
++
++ await check('the owner carries the task on, or hands it to another agent', async () => {
++ const id = await startTask();
++ await endAttempt(id, { state: 'succeeded', result: { ok: true, summary: 'done', files: [], log: [{ t: 'done', ok: true, text: 'done' }] } });
++ const continued = await call('POST', `/tasks/${id}/continue`, { body: { note: 'Only the tests are left.' } });
++ assert.strictEqual(continued.status, 200, JSON.stringify(continued.json));
++ assert.strictEqual(continued.json.attempts.length, 2);
++ assert.strictEqual(continued.json.attempts[1].kind, 'manual');
++ assert.ok(/carried on by the owner/.test(continued.json.attempts[1].why), continued.json.attempts[1].why);
++ // While the task runs again, nobody carries it on a second time.
++ const again = await call('POST', `/tasks/${id}/continue`, { body: { agent: 'codex' } });
++ assert.strictEqual(again.status, 409, 'the task is running again');
++ });
++
++ await check('the auth of every new route: no token, no capability, another person', async () => {
++ const id = await startTask();
++ const noCap = t.network.mintService('svc:nocap', 'openvibe.codes', ['codes.other']);
++ const other = t.network.mintService('svc:other', 'openvibe.codes', CAPS);
++ // No token at all.
++ for (const [method, path] of [['GET', `/tasks/${id}`], ['GET', `/tasks/${id}/events`], ['POST', `/tasks/${id}/continue`], ['POST', `/tasks/${id}/stop`]]) {
++ const res = await call(method, path, { token: null });
++ assert.strictEqual(res.status, 401, `${method} ${path} without a token`);
++ }
++ // A service without the capability.
++ assert.strictEqual((await call('GET', `/tasks/${id}`, { token: noCap })).status, 403);
++ assert.strictEqual((await call('POST', `/tasks/${id}/continue`, { token: noCap, body: {} })).status, 403);
++ // Another person, and another service (which sees only the tasks it started).
++ assert.strictEqual((await call('GET', `/tasks/${id}`, { subject: bo.subject })).status, 404);
++ assert.strictEqual((await call('GET', `/tasks/${id}`, { token: other })).status, 404);
++ assert.strictEqual((await call('POST', `/tasks/${id}/stop`, { subject: bo.subject })).status, 404);
++ // A task that does not exist.
++ assert.strictEqual((await call('GET', '/tasks/tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1')).status, 404);
++ assert.strictEqual((await call('GET', '/tasks/nope')).status, 404);
++ });
++
++ await check('the task page shows the timeline of attempts, each with its why', async () => {
++ const id = await startTask();
++ const result = { ok: false, summary: 'x', files: [], log: [{ t: 'done', ok: false, text: 'usage limit reached' }] };
++ await endAttempt(id, { state: 'failed', result, error: 'usage limit reached' });
++ const page = await t.get(`/run/tasks/${id}`, { as: ada });
++ assert.strictEqual(page.status, 200, page.text.slice(0, 300));
++ assert.ok(page.text.includes('data-timeline'), 'the timeline is there');
++ assert.ok(page.text.includes('data-attempt="1"') && page.text.includes('data-attempt="2"'), 'every attempt is a row');
++ assert.ok(page.text.includes('data-state="failed"'), 'the attempt says how it ended');
++ assert.ok(page.text.includes('data-why'), 'the attempt says why it happened');
++ assert.ok(/took over: /.test(page.text), 'the second attempt says who took over');
++ assert.ok(page.text.indexOf('data-timeline') < page.text.indexOf('data-session') || !page.text.includes('data-session'),
++ 'the timeline is above the session');
++ });
++
++ await check('a task is carried on twice, and then it needs a person', async () => {
++ const id = await startTask();
++ const crash = {
++ ok: false, summary: 'I will continue with the rest of it', files: [],
++ log: [{ t: 'agent', agent: 'claude-code', session: 'sess-1' }, { t: 'done', ok: false, text: 'I will continue with the rest of it' }],
++ };
++ await endAttempt(id, { state: 'failed', result: crash, error: 'the machine went away' });
++ await endAttempt(id, { state: 'failed', result: crash, error: 'the machine went away' });
++ const attempts = await attemptsOf(id);
++ assert.strictEqual(attempts.filter((a) => a.kind === 'continue').length, 2, 'two automatic continues');
++ await endAttempt(id, { state: 'failed', result: crash, error: 'the machine went away' });
++ const row = await rowOf(id);
++ assert.strictEqual(row.outcome, 'unfinished', 'the third one needs a person');
++ assert.ok(/carried on 2 times/.test(row.outcome_note), row.outcome_note);
++ assert.strictEqual((await attemptsOf(id)).length, 3, 'no third continue');
++ });
++
++ await check('the owner needs no token: their own task reads, continues and stops', async () => {
++ const id = await startTask();
++ const mine = await t.get(`/api/v1/tasks/${id}`, { as: ada });
++ assert.strictEqual(mine.status, 200, mine.text.slice(0, 200));
++ assert.strictEqual(mine.json().attempts.length, 1);
++ const theirs = await t.get(`/api/v1/tasks/${id}`, { as: bo });
++ assert.strictEqual(theirs.status, 404);
++ });
++ } finally {
++ await t.close();
++ }
++ done();
++})().catch((e) => { console.error(e); process.exit(1); });
ddiff --git a/test/task-events.test.js b/test/task-events.test.js
nnew file mode 100644
iindex 0000000..3933632
---- /dev/null
++++ b/test/task-events.test.js
@@@ -0,0 +1,138 @@
++'use strict';
++
++/**
++ * A task's event stream (GET /api/v1/tasks/:id/events): the session's lines in the order they happened, the attempt
++ * each line belongs to, and the attempts themselves (started, ended with why). A client that reconnects sends the last
++ * line's seq as Last-Event-ID and gets only what came after it; a heartbeat keeps the connection alive.
++ */
++const assert = require('assert');
++const http = require('http');
++const { boot, check, done } = require('./helpers/boot');
++
++(async () => {
++ const calls = [];
++ let counter = 0;
++ const run = {
++ enabled: true,
++ async submit(o) {
++ counter++;
++ const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${String(counter).padStart(2, '0')}`;
++ calls.push(['submit', o]);
++ return { id, state: 'queued' };
++ },
++ async read(jobId) { return { id: jobId, state: 'queued' }; },
++ async cancel(jobId) { return { id: jobId, state: 'cancelled' }; },
++ async streamUrl() { throw new Error('not used'); },
++ };
++ const t = await boot({ run, env: { CODES_SSE_HEARTBEAT_MS: '300' } });
++ const db = t.ctx.store.db;
++ const ada = t.network.addUser('ada');
++ const bo = t.network.addUser('bo');
++ const CAPS = ['codes.task.start', 'codes.task.read', 'codes.task.cancel'];
++ const actor = t.network.mintService('svc:actor', 'openvibe.codes', CAPS);
++
++ const call = async (method, path, { token = actor, subject = ada.subject, body, headers = {} } = {}) => {
++ const h = { Accept: 'text/event-stream', ...headers, ...(token ? { Authorization: `Bearer ${token}` } : {}), ...(subject ? { 'X-OV-Subject': subject } : {}) };
++ if (body !== undefined) h['Content-Type'] = 'application/json';
++ const res = await fetch(`${t.base}/api/v1${path}`, { method, headers: h, body: body === undefined ? undefined : JSON.stringify(body) });
++ return { status: res.status, text: await res.text() };
++ };
++
++ /** The stream, read until it ends (a finished task) or until the marker arrives. */
++ const stream = (path, { headers = {}, until = null, ms = 8000 } = {}) => new Promise((resolve, reject) => {
++ const req = http.get(`${t.base}${path}`, { headers }, (res) => {
++ let text = '';
++ res.setEncoding('utf8');
++ const timer = setTimeout(() => { req.destroy(); resolve({ status: res.statusCode, text }); }, ms);
++ res.on('data', (chunk) => {
++ text += chunk;
++ if (until && text.includes(until)) { clearTimeout(timer); req.destroy(); resolve({ status: res.statusCode, text }); }
++ });
++ res.on('end', () => { clearTimeout(timer); resolve({ status: res.statusCode, text }); });
++ });
++ req.on('error', (e) => (e.code === 'ECONNRESET' ? resolve({ status: 0, text: '' }) : reject(e)));
++ });
++
++ /** A task of Ada's with two attempts and three lines of session. */
++ const makeTask = async () => {
++ const r = await fetch(`${t.base}/api/v1/tasks`, {
++ method: 'POST',
++ headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${actor}`, 'X-OV-Subject': ada.subject },
++ body: JSON.stringify({ repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a test for the date parser.', idempotency_key: `act_${Math.random().toString(36).slice(2, 10)}` }),
++ });
++ const json = await r.json();
++ const id = json.task.id;
++ await db.exec(`INSERT INTO codes_session_lines (task_id, seq, at, kind, body, attempt_n) VALUES
++ ($1, 1, now(), 'text', '{"t":"text","text":"first"}'::jsonb, 1),
++ ($1, 2, now(), 'text', '{"t":"text","text":"second"}'::jsonb, 1),
++ ($1, 3, now(), 'text', '{"t":"text","text":"third"}'::jsonb, 2)`, [id]);
++ return id;
++ };
++
++ try {
++ // Ada ran a task on /run before (so Codes knows her project).
++ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)
++ VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1', 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J00', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject]);
++
++ await check('the stream: the lines in order, the attempt each belongs to, and the attempts themselves', async () => {
++ const id = await makeTask();
++ // The task ends, so the stream ends with it.
++ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [id]);
++ const job = { id: row.job_id, state: 'succeeded', result: { ok: true, summary: 'done', files: [], log: [{ t: 'done', ok: true, text: 'done' }] }, error: null };
++ await t.ctx.tasks.applyEvent(row, { state: 'succeeded', job });
++ await t.ctx.tasks.afterEnd(id);
++ const s = await call('GET', `/tasks/${id}/events`);
++ assert.strictEqual(s.status, 200);
++ assert.ok(s.text.includes('event: line'), s.text.slice(0, 300));
++ const seqs = [...s.text.matchAll(/^id: (\d+)$/gm)].map((m) => Number(m[1]));
++ assert.deepStrictEqual(seqs.slice(0, 3), [1, 2, 3], 'the lines in order, each with its seq');
++ const bodies = [...s.text.matchAll(/^data: (.*)$/gm)].map((m) => JSON.parse(m[1]));
++ const lines = bodies.filter((b) => b.seq);
++ assert.deepStrictEqual(lines.slice(0, 3).map((l) => l.attempt_n), [1, 1, 2], 'each line says which attempt it belongs to');
++ assert.deepStrictEqual(lines.slice(0, 3).map((l) => l.body.text), ['first', 'second', 'third']);
++ const attempts = bodies.filter((b) => b.type);
++ assert.strictEqual(attempts[0].type, 'attempt_started');
++ assert.strictEqual(attempts[attempts.length - 1].type, 'attempt_ended');
++ assert.strictEqual(attempts[attempts.length - 1].state, 'done');
++ });
++
++ await check('a client that reconnects gets only what came after the line it last saw', async () => {
++ const id = await makeTask();
++ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [id]);
++ const job = { id: row.job_id, state: 'succeeded', result: { ok: true, summary: 'done', files: [], log: [{ t: 'done', ok: true, text: 'done' }] }, error: null };
++ await t.ctx.tasks.applyEvent(row, { state: 'succeeded', job });
++ await t.ctx.tasks.afterEnd(id);
++ const s = await call('GET', `/tasks/${id}/events`, { headers: { 'Last-Event-ID': '2' } });
++ assert.strictEqual(s.status, 200);
++ const seqs = [...s.text.matchAll(/^id: (\d+)$/gm)].map((m) => Number(m[1]));
++ assert.deepStrictEqual(seqs[0], 3, 'only the line after the last one it saw');
++ assert.ok(!seqs.includes(1) && !seqs.includes(2), 'and not the ones before');
++ assert.ok(!s.text.includes('"first"') && !s.text.includes('"second"'), 'and not the ones before');
++ });
++
++ await check('the heartbeat keeps the connection alive', async () => {
++ const id = await makeTask();
++ const s = await stream(`/api/v1/tasks/${id}/events`, {
++ headers: { Authorization: `Bearer ${actor}`, 'X-OV-Subject': ada.subject },
++ until: ': heartbeat',
++ ms: 5000,
++ });
++ assert.ok(s.text.includes(': heartbeat'), s.text.slice(0, 200));
++ });
++
++ await check('the auth of the stream: no token, another person, another service', async () => {
++ const id = await makeTask();
++ assert.strictEqual((await call('GET', `/tasks/${id}/events`, { token: null })).status, 401);
++ assert.strictEqual((await call('GET', `/tasks/${id}/events`, { subject: bo.subject })).status, 404);
++ const other = t.network.mintService('svc:other', 'openvibe.codes', CAPS);
++ assert.strictEqual((await call('GET', `/tasks/${id}/events`, { token: other })).status, 404);
++ // The owner needs no token (the task runs, so the stream is read until the heartbeat).
++ const mine = await stream(`/api/v1/tasks/${id}/events`, { headers: { cookie: t.signIn(ada) }, until: ': heartbeat', ms: 5000 });
++ assert.strictEqual(mine.status, 200);
++ assert.ok(mine.text.includes('event: line'), mine.text.slice(0, 200));
++ });
++ } finally {
++ await t.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 365881f..97cb755 100644
---- a/test/tasks-api.test.js
++++ b/test/tasks-api.test.js
@@@ -92,12 +92,24 @@ const { boot, check, done } = require('./helpers/boot');
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);
++ // The read is the task (its own fields at the top level, as codes.task@1 had them), with its
++ // attempts and its outcome added to it: codes.task@2 (stage C2).
++ const v = validate('codes.task@2', r.json);
assert.ok(v.valid, JSON.stringify(v.errors));
++ const v1doc = { ...r.json };
++ delete v1doc.attempts; delete v1doc.outcome;
++ const v1 = validate('codes.task@1', v1doc);
++ assert.ok(v1.valid, 'the task itself still answers codes.task@1');
assert.strictEqual(r.json.state, 'succeeded');
++ assert.strictEqual(r.json.id, task.id, 'the task at the top level, not wrapped');
assert.strictEqual(r.json.result.summary.length, 4000);
assert.deepStrictEqual(r.json.result.files, [{ path: 'README', added: 1, removed: 0 }]);
assert.deepStrictEqual(r.json.result.push, { pushed: false, reason: 'no token' });
++ assert.strictEqual(r.json.attempts.length, 1, 'one attempt: the task as it was asked');
++ assert.strictEqual(r.json.attempts[0].n, 1);
++ assert.strictEqual(r.json.attempts[0].state, 'done');
++ assert.ok(r.json.attempts[0].why, 'the attempt says why it happened');
++ assert.strictEqual(r.json.outcome.state, 'done');
});
await check('another person, another service, or a task the person started by hand: 404', async () => {