openvibers/openvibe.codes Commits
Talk to a running task: messages wait for the next attempt (#412)
Changes
@@@ -110,6 +110,8 @@ const { task } = await res.json();// GET /api/v1/tasks?state=running your tasks: running, needs (left unfinished for you), done// GET /api/v1/tasks/:id/events the event stream (SSE; Last-Event-ID resumes it)// POST /api/v1/tasks/:id/continue { note?, agent? } → the next attempt now++// POST /api/v1/tasks/:id/message { text } → 202 while an attempt runs (it waits for the next attempt),++// or the next attempt now when nothing runs, like a continue with a note// POST /api/v1/tasks/:id/stop the running attempt stops; nothing carries it on```nnew file mode 100644BBinary files /dev/null and b/docs/screenshots/ask-354-message-1280.png differnnew file mode 100644BBinary files /dev/null and b/docs/screenshots/ask-354-message-390.png differnnew file mode 100644@@@ -0,0 +1,17 @@++-- phase: expand++-- Ask 354: a message to a running task. A Run job on the person's own machine cannot be interrupted mid-run, so a++-- message sent while an attempt is live waits here (the conductor's turn queue, codes_conductor_messages, works the++-- same way): when the attempt ends, the next attempt carries the waiting notes in its prompt. n is the message's++-- place in the queue; state is queued until an attempt takes it, then delivered. Every row hangs under a task and is++-- erased with it (server/account-data.js).++++CREATE TABLE IF NOT EXISTS codes_task_messages (++ id text COLLATE "C" PRIMARY KEY, -- msg_<ULID>++ task_id text COLLATE "C" NOT NULL REFERENCES run_tasks(id) ON DELETE CASCADE,++ n integer NOT NULL, -- 1, 2, … the message's place in the queue++ text text NOT NULL, -- what the person said (at most 4000 characters)++ state text COLLATE "C" NOT NULL, -- queued | delivered++ created_at timestamptz NOT NULL DEFAULT now()++);++CREATE UNIQUE INDEX IF NOT EXISTS codes_task_messages_task_n ON codes_task_messages (task_id, n);++CREATE INDEX IF NOT EXISTS codes_task_messages_queued ON codes_task_messages (task_id) WHERE state = 'queued';@@@ -45,13 +45,20 @@if (!run || !host.isConnected) return;if (oldView) oldView.destroy();host.textContent = '';++ // A Run job on the person's machine cannot be interrupted mid-run, so a message to a live task waits++ // for the next attempt (POST /message, 202); the reload shows it waiting under Live.++ var live = section.hasAttribute('data-events');var view = window.OpenVibeAgentSession.mount(host, {title: run.title || section.getAttribute('data-title') || 'Task', status: run.status || 'running',-- follow: section.hasAttribute('data-events'), files: false,-- onSend: section.getAttribute('data-owner') === '1' && !section.hasAttribute('data-events') ? function (text) {-- return fetch('/api/v1/tasks/' + encodeURIComponent(id) + '/continue', {++ follow: live, files: false,++ sayPlaceholder: live ? 'Message the run: it starts the next attempt' : undefined,++ onSend: section.getAttribute('data-owner') === '1' ? function (text) {++ return fetch('/api/v1/tasks/' + encodeURIComponent(id) + (live ? '/message' : '/continue'), {method: 'POST', credentials: 'same-origin', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ note: text })-- }).then(function (r) { if (!r.ok) throw new Error('The task could not take the message yet.'); });++ }).then(function (r) {++ if (!r.ok) throw new Error('The task could not take the message yet.');++ if (r.status === 202) window.location.reload();++ });} : undefined});section.dataset.agentUi = '1';@@@ -80,6 +80,12 @@ const TASK_KEYED = [},// The approvals the person decided, and the ones still waiting when their account is erased.{ table: 'codes_approvals', file: 'approvals.json', columns: 'l.id, l.task_id, l.tool, l.summary, l.decision, l.message, l.created_at, l.decided_at', order: 'l.created_at ASC', kind: 'approvals', max: TASK_ROWS_MAX },++ // A message sent while an attempt ran (ask 354): what the person said, taken by the next attempt.++ {++ table: 'codes_task_messages', file: 'task-messages.json',++ columns: 'l.id, l.task_id, l.n, l.text, l.state, l.created_at',++ order: 'l.n ASC', kind: 'task_messages', max: TASK_ROWS_MAX,++ },{ table: 'codes_events_inbox', file: null, kind: 'events_received' },// The one notice sent per task and state (Codes C9c): no content of its own, erased with the task, not exported.{ table: 'codes_notices', file: null, kind: 'notices' },@@@ -962,15 +962,33 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;try { await writeSessionReport(saved, collected, ending); } catch (e) {log.warn('[Codes] the session report of task ' + row.id + ' failed: ' + String((e && e.message) || e).slice(0, 200));}++ // A message sent while the attempt ran carries the task on after all: the engine is done, the person is++ // not. This is the manual continue the owner would press next, with the waiting notes taken into it.++ const waiting = await pendingMessages(row.id);++ if (waiting.length) {++ const fresh = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);++ const carried = fresh ? await continueTask(fresh, {}) : { error: 'no such task' };++ if (!carried.error && carried.job) {++ log.info(`[Codes] task ${row.id}: ${waiting.length} waiting message(s) started the next attempt`);++ return carried.row;++ }++ if (carried.error) log.warn(`[Codes] task ${row.id}: waiting messages could not carry it on: ${carried.error}`);++ }return saved;}if (!plan.kind) return row;++ // A message sent while the attempt ran rides the follow-up the engine planned: the person is answered.++ // Marked delivered only once the job is submitted; a follow-up that never starts leaves the queue for the++ // owner's next press, and a message arriving after this read stays queued for the attempt after that.++ const riding = await pendingMessages(row.id);++ if (riding.length) plan.prompt = `${plan.prompt || ''}${notesSection(riding)}`;// The goal's budgets are checked before the next turn starts: a spent budget ends the task here, with its reason.if (goal) {const hit = goalBudgetHit(attempts, asObject(row.goal) || goal, row);if (hit) return await finishGoal(row, { outcome: 'unfinished', note: `stopped by budget: ${hit}`, end_state: 'stopped_by_budget' });}const done = await startFollowUp(row, plan, ending);++ if (riding.length && done.created && done.job) await db.exec(`UPDATE codes_task_messages SET state = 'delivered' WHERE task_id = $1 AND state = 'queued' AND id = ANY($2)`, [row.id, riding.map((m) => m.id)]);return done.row || row;}@@@ -1068,6 +1086,61 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;return { attempt, created: true, job, row: updated };}++ /**++ * A message to the task while it works. A Run job on the person's own machine cannot be interrupted mid-run++ * (Run has no job-input channel), so a message sent while an attempt is live waits in codes_task_messages — the++ * conductor's turn queue works the same way. The next attempt carries the waiting notes in its prompt: the++ * engine's own follow-up takes them, and when the engine ends the task instead, afterEnd carries it on with++ * them. Nothing the person sent is lost; on a stopped task the queue waits until the owner carries it on.++ */++ const MAX_MESSAGE = 4000;++ function cleanMessage(text) {++ const t = typeof text === 'string' ? text.trim() : '';++ if (!t) return { error: 'a message needs words' };++ if (t.length > MAX_MESSAGE) return { error: `a message is at most ${MAX_MESSAGE} characters` };++ return { text: t };++ }++ /** The messages still waiting for the next attempt, oldest first (also what the task page shows). */++ async function pendingMessages(taskId) {++ if (!taskId) return [];++ return db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 AND state = $2 ORDER BY n ASC', [taskId, 'queued']);++ }++ /** Take the waiting messages inside the attempt's own transaction: taken exactly once, by the attempt that runs. */++ async function takeMessages(taskId) {++ const rows = await db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 AND state = $2 ORDER BY n ASC FOR UPDATE', [taskId, 'queued']);++ if (rows.length) await db.exec('UPDATE codes_task_messages SET state = $2 WHERE task_id = $1 AND state = $3', [taskId, 'delivered', 'queued']);++ return rows;++ }++ function notesSection(rows) {++ return `\n\n---\n## Notes from the owner (sent while the last attempt ran)\n\n${rows.map((m) => String(m.text)).join('\n\n---\n\n')}\n`;++ }++ async function queueMessage(row, text) {++ const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_task_messages WHERE task_id = $1', [row.id]);++ return db.one(`INSERT INTO codes_task_messages (id, task_id, n, text, state) VALUES ($1, $2, $3, $4, 'queued') RETURNING *`,++ [store.newId('msg'), row.id, Number(n.n), text]);++ }++ /** A message to the task: it waits while an attempt is live, and carries the task on when nothing runs. */++ async function messageTask(row, { note = null } = {}) {++ if (!row || !row.id) return { error: 'no such task' };++ const clean = cleanMessage(note);++ if (clean.error) return { error: clean.error };++ const attempts = await attemptStore.list(row.id);++ if (!attempts.length) return { error: 'the task has no attempt to carry on' };++ const live = attempts.slice().reverse().find((a) => !ENDED.has(a.state));++ if (live) {++ const message = await queueMessage(row, clean.text);++ log.info(`[Codes] task ${row.id}: a message waits for the next attempt`);++ return { queued: true, message, attempt: live, row };++ }++ const carried = await continueTask(row, { note: clean.text });++ if (carried.live) {++ // An attempt started between the check and the carry-on: the message waits after all.++ const message = await queueMessage(carried.row || row, clean.text);++ return { queued: true, message, attempt: carried.attempt, row: carried.row || row };++ }++ return carried;++ }++/*** 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.@@@ -1125,6 +1198,13 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;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' };++ // A message sent while the last attempt ran rides this one: taken here, in the attempt's own++ // transaction, so exactly one attempt carries each message.++ const waiting = await takeMessages(row.id);++ if (waiting.length) {++ plan.prompt += notesSection(waiting);++ if (!text) plan.why = `carried on by the owner: ${String(waiting[0].text).slice(0, 200)}`;++ }const made = await attemptStore.create({task: current, agent: plan.agent, model: plan.model, effort: plan.effort || null, session: null, why: plan.why, kind: plan.kind, state: 'queued', budget: left.left,});@@@ -1260,6 +1340,7 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;return {start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,latestProject, present, who, attempts: attemptStore, setConductor, attachBindings, fundTurn,++ messageTask, pendingMessages,setForge(f) { forge = f || null; },};}@@@ -339,6 +339,9 @@ ${i.summary ? html`<p class="run-proof-summary" data-proof-summary>${i.summary}<const lines = live ? null : await sessionStore.linesForRow(row);const title = String(row.prompt || '').split('\n')[0].slice(0, 90);const session = lines && lines.length ? html`<section class="run-session" data-session data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" data-state="${row.state}"><h2>The session</h2><div data-session-view></div><script type="application/json" data-session-log>${raw(JSON.stringify(lines).replace(/</g, '\\u003c'))}</script></section>` : '';++ // A message sent while the attempt runs waits for the next one (ask 354): say so under Live.++ const waiting = live && mine ? await taskStore.pendingMessages(row.id) : [];++ const waitingLine = waiting.length ? html`<p class="run-pending" data-pending>${waiting.length === 1 ? 'A message waits' : `${waiting.length} messages wait`} for the next attempt</p>` : '';const body = html`${shareHead ? raw(shareHead) : ''}<p class="crumbs"><a href="/run">Run locally</a></p><h1 class="run-repo">${repoName(row.repo)}${row.ref ? html` <small>${row.ref}</small>` : ''}</h1><p><span class="run-pill run-pill-${row.state}" data-state>${STATE_LABEL[row.state] || row.state}</span>${row.agent ? html` <span class="run-pill">${row.agent}</span>` : ''}</p>@@@ -347,7 +350,7 @@ ${mine ? shareButton(row) : html`<p class="muted">This task was shared with you.${await timelineHtml(row)}${goalHtml(row)}${await proofHtml(row)}--${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events" data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" aria-live="polite"><h2>Live</h2><ol class="run-log" data-log></ol></section>++${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events" data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" aria-live="polite"><h2>Live</h2>${waitingLine}<ol class="run-log" data-log></ol></section>${mine ? html`<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'] });});@@@ -571,6 +571,29 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, taskreturn res.json(await timeline(out.row));});++ /**++ * A message to the task (ask 354): while an attempt is live the message waits for the next attempt (202, nothing++ * lost — a Run job on the person's own machine cannot be interrupted mid-run); when nothing runs it carries the++ * task on at once, like a continue with a note.++ */++ r.post('/tasks/:id/message', 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 text = typeof b.text === 'string' ? b.text : (typeof b.note === 'string' ? b.note : null);++ const out = await taskStore.messageTask(row, { note: text });++ if (out.error) {++ const status = out.error === 'no such task' ? 404 : /needs words|at most/.test(out.error) ? 422 : 409;++ return problem(req, res, status,++ status === 404 ? 'codes.task_not_found' : status === 422 ? 'codes.message_invalid' : 'codes.task_conflict', out.error);++ }++ if (out.queued) {++ return res.status(202).json({ queued: true, message: { id: out.message.id, n: out.message.n, text: out.message.text } });++ }++ log.info(`[Codes] task ${row.id} carried on by a message`);++ return res.json(await timeline(out.row));++ });++/*** The running attempt stops: the outcome is "stopped" and nothing carries it on. The owner (the Dock's shape)* gets { ok }; a service keeps the timeline its capability was built on. Another person's task is 404, nevernnew file mode 100644@@@ -0,0 +1,171 @@++'use strict';++/**++ * A message to a task (POST /api/v1/tasks/:id/message, ask 354): while an attempt is live the message waits in++ * codes_task_messages (202, nothing lost — a Run job on the person's machine cannot be interrupted mid-run); when++ * the attempt ends the next attempt carries the waiting notes in its prompt, whether the engine planned a follow-up++ * or ended the task. When nothing runs the message carries the task on at once, like a continue with a note. A++ * stopped task refuses (409) and keeps its queue.++ */++const assert = require('assert');++const { boot, check, done } = require('./helpers/boot');++++(async () => {++ const submits = [];++ let counter = 0;++ const jobs = new Map();++ const run = {++ enabled: true,++ async submit(o) {++ counter++;++ const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${String(counter).padStart(2, '0')}`;++ jobs.set(id, { id, state: 'queued', args: o.args });++ submits.push(o);++ return { id, state: 'queued' };++ },++ async read(jobId) { return jobs.get(jobId) || { id: jobId, state: 'queued' }; },++ async cancel(jobId) { const j = jobs.get(jobId); if (j) j.state = 'cancelled'; return j || { id: jobId, state: 'cancelled' }; },++ async streamUrl() { throw new Error('not used'); },++ };++ const t = await boot({ run });++ const db = t.ctx.store.db;++ const ada = t.network.addUser('ada');++ const bo = t.network.addUser('bo');++ const actor = t.network.mintService('svc:actor', 'openvibe.codes', ['codes.task.start', 'codes.task.read', 'codes.task.cancel']);++ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';++ // Ada ran a task on /run before (so Codes knows her project); Bo never did.++ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)++ VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, $2, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z200', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject, project]);++++ const asAda = async (method, path, body, token = t.network.userToken(ada)) => {++ const headers = { Accept: 'application/json', Authorization: `Bearer ${token}` };++ if (body !== undefined) headers['Content-Type'] = 'application/json';++ const res = await fetch(`${t.base}/api/v1${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body), redirect: 'manual' });++ return { status: res.status, json: await res.json().catch(() => null) };++ };++++ let seq = 0;++ const startTask = async () => {++ seq++;++ const r = await asAda('POST', '/tasks', { repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a line to the README.', agent: 'deepseek', idempotency_key: `act_msg_${seq}` });++ assert.strictEqual(r.status, 201, JSON.stringify(r.json));++ return r.json.task.id;++ };++ const attemptsOf = async (taskId) => db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);++ const messagesOf = async (taskId) => db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 ORDER BY n ASC', [taskId]);++ /** The attempt ends the way Run delivers it, then the engine decides what happens next. */++ const endAttempt = async (taskId, ok, text) => {++ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [taskId]);++ const job = {++ id: row.job_id, state: ok ? 'succeeded' : 'failed',++ result: {++ ok, summary: text, files: [],++ log: [{ t: 'agent', agent: 'deepseek', session: 'sess-1' }, { t: 'done', ok, text }],++ },++ ...(ok ? {} : { error: { code: 'run.job.worker_failed', detail: text } }),++ };++ const updated = await t.ctx.tasks.applyEvent(row, { state: ok ? 'succeeded' : 'failed', job, error: ok ? null : text });++ return t.ctx.tasks.afterEnd(updated.id);++ };++++ try {++ await check('a message to a live task waits: 202, queued, no new attempt, no job', async () => {++ const id = await startTask();++ const before = submits.length;++ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'also check the spelling' });++ assert.strictEqual(r.status, 202, JSON.stringify(r.json));++ assert.strictEqual(r.json.queued, true);++ assert.strictEqual(r.json.message.n, 1);++ assert.strictEqual(r.json.message.text, 'also check the spelling');++ assert.strictEqual((await attemptsOf(id)).length, 1, 'no new attempt while one runs');++ assert.strictEqual(submits.length, before, 'no job while one runs');++ const rows = await messagesOf(id);++ assert.strictEqual(rows.length, 1);++ assert.strictEqual(rows[0].state, 'queued');++ });++++ await check('a second message joins the queue in order', async () => {++ const id = await startTask();++ await asAda('POST', `/tasks/${id}/message`, { text: 'first' });++ const r = await asAda('POST', `/tasks/${id}/message`, { note: 'second' });++ assert.strictEqual(r.status, 202, JSON.stringify(r.json));++ assert.strictEqual(r.json.message.n, 2);++ const rows = await messagesOf(id);++ assert.deepStrictEqual(rows.map((m) => m.text), ['first', 'second']);++ assert.ok(rows.every((m) => m.state === 'queued'));++ });++++ await check('when the attempt crashes, the follow-up carries the waiting notes', async () => {++ const id = await startTask();++ await asAda('POST', `/tasks/${id}/message`, { text: 'also check the spelling' });++ await asAda('POST', `/tasks/${id}/message`, { text: 'and the license header' });++ const before = submits.length;++ await endAttempt(id, false, 'the machine went away');++ const attempts = await attemptsOf(id);++ assert.strictEqual(attempts.length, 2, 'the task was carried on');++ assert.strictEqual(submits.length, before + 1, 'one follow-up job');++ const prompt = String(submits[submits.length - 1].args.prompt);++ assert.ok(prompt.includes('Notes from the owner (sent while the last attempt ran)'), 'the notes ride the follow-up');++ assert.ok(prompt.includes('also check the spelling') && prompt.includes('and the license header'), 'both notes');++ assert.ok((await messagesOf(id)).every((m) => m.state === 'delivered'), 'taken exactly once');++ });++++ await check('when the engine ends the task with notes waiting, it carries on by itself', async () => {++ const id = await startTask();++ await asAda('POST', `/tasks/${id}/message`, { text: 'double-check the edge cases' });++ const before = submits.length;++ await endAttempt(id, true, 'done, all green');++ const attempts = await attemptsOf(id);++ assert.strictEqual(attempts.length, 2, 'the waiting message started the next attempt');++ assert.strictEqual(submits.length, before + 1, 'one follow-up job');++ assert.ok(String(submits[submits.length - 1].args.prompt).includes('double-check the edge cases'), 'the note rides it');++ assert.ok((await messagesOf(id)).every((m) => m.state === 'delivered'), 'taken exactly once');++ });++++ await check('a message to an ended task carries it on at once, like a continue with a note', async () => {++ const id = await startTask();++ await endAttempt(id, true, 'done, all green');++ assert.strictEqual((await attemptsOf(id)).length, 1, 'the engine left the finished task alone');++ const before = submits.length;++ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'and the tests' });++ assert.strictEqual(r.status, 200, JSON.stringify(r.json));++ assert.strictEqual((await attemptsOf(id)).length, 2, 'the message is the next attempt');++ assert.strictEqual(submits.length, before + 1, 'one job');++ assert.ok(String(submits[submits.length - 1].args.prompt).includes('and the tests'), 'the note rides it');++ });++++ await check('a stopped task refuses and keeps its queue: 409, nothing starts', async () => {++ const id = await startTask();++ await asAda('POST', `/tasks/${id}/message`, { text: 'wait for me' });++ const stopped = await asAda('POST', `/tasks/${id}/stop`);++ assert.strictEqual(stopped.status, 200, JSON.stringify(stopped.json));++ const before = submits.length;++ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'one more thing' });++ assert.strictEqual(r.status, 409, JSON.stringify(r.json));++ assert.strictEqual(r.json.code, 'codes.task_conflict');++ assert.strictEqual(submits.length, before, 'no job on a stopped task');++ assert.strictEqual((await attemptsOf(id)).length, 1, 'no attempt on a stopped task');++ assert.deepStrictEqual((await messagesOf(id)).map((m) => m.state), ['queued'], 'the queue waits for a hand carry-on');++ });++++ await check('empty and too long are 422; another person and a missing task are 404', async () => {++ const id = await startTask();++ assert.strictEqual((await asAda('POST', `/tasks/${id}/message`, { text: ' ' })).status, 422);++ assert.strictEqual((await asAda('POST', `/tasks/${id}/message`, { text: 'x'.repeat(4001) })).status, 422);++ const other = await asAda('POST', `/tasks/${id}/message`, { text: 'hello' }, t.network.userToken(bo));++ assert.strictEqual(other.status, 404);++ assert.strictEqual((await asAda('POST', '/tasks/tsk_01J8Z3Q4R5S6T7V8W9X0Y9Z9Q9/message', { text: 'hello' })).status, 404);++ });++++ await check('the live task page says a message waits for the next attempt', async () => {++ const id = await startTask();++ await asAda('POST', `/tasks/${id}/message`, { text: 'wait for me' });++ const page = await t.get(`/run/tasks/${id}`, { as: ada });++ assert.strictEqual(page.status, 200);++ assert.ok(page.text.includes('data-pending'), 'the waiting line is on the page');++ assert.ok(page.text.includes('A message waits for the next attempt'), 'in plain words');++ });++ } finally {++ await t.close();++ }++ done();++})().catch((e) => { console.error(e); process.exit(1); });
@@ -110,6 +110,8 @@ const { task } = await res.json();
@@ -110,6 +110,8 @@ const { task } = await res.json();
// GET /api/v1/tasks?state=running your tasks: running, needs (left unfinished for you), done
// GET /api/v1/tasks?state=running your tasks: running, needs (left unfinished for you), done
// GET /api/v1/tasks/:id/events the event stream (SSE; Last-Event-ID resumes it)
// GET /api/v1/tasks/:id/events the event stream (SSE; Last-Event-ID resumes it)
// POST /api/v1/tasks/:id/continue { note?, agent? } → the next attempt now
// POST /api/v1/tasks/:id/continue { note?, agent? } → the next attempt now
+// POST /api/v1/tasks/:id/message { text } → 202 while an attempt runs (it waits for the next attempt),
+// or the next attempt now when nothing runs, like a continue with a note
// POST /api/v1/tasks/:id/stop the running attempt stops; nothing carries it on
// POST /api/v1/tasks/:id/stop the running attempt stops; nothing carries it on
```
```
new file mode 100644
new file mode 100644
Binary files /dev/null and b/docs/screenshots/ask-354-message-1280.png differ
Binary files /dev/null and b/docs/screenshots/ask-354-message-1280.png differ
new file mode 100644
new file mode 100644
Binary files /dev/null and b/docs/screenshots/ask-354-message-390.png differ
Binary files /dev/null and b/docs/screenshots/ask-354-message-390.png differ
new file mode 100644
new file mode 100644
@@ -0,0 +1,17 @@
@@ -0,0 +1,17 @@
+-- phase: expand
+-- Ask 354: a message to a running task. A Run job on the person's own machine cannot be interrupted mid-run, so a
+-- message sent while an attempt is live waits here (the conductor's turn queue, codes_conductor_messages, works the
+-- same way): when the attempt ends, the next attempt carries the waiting notes in its prompt. n is the message's
+-- place in the queue; state is queued until an attempt takes it, then delivered. Every row hangs under a task and is
+-- erased with it (server/account-data.js).
+
+CREATE TABLE IF NOT EXISTS codes_task_messages (
+ id text COLLATE "C" PRIMARY KEY, -- msg_<ULID>
+ task_id text COLLATE "C" NOT NULL REFERENCES run_tasks(id) ON DELETE CASCADE,
+ n integer NOT NULL, -- 1, 2, … the message's place in the queue
+ text text NOT NULL, -- what the person said (at most 4000 characters)
+ state text COLLATE "C" NOT NULL, -- queued | delivered
+ created_at timestamptz NOT NULL DEFAULT now()
+);
+CREATE UNIQUE INDEX IF NOT EXISTS codes_task_messages_task_n ON codes_task_messages (task_id, n);
+CREATE INDEX IF NOT EXISTS codes_task_messages_queued ON codes_task_messages (task_id) WHERE state = 'queued';
@@ -45,13 +45,20 @@
@@ -45,13 +45,20 @@
if (!run || !host.isConnected) return;
if (!run || !host.isConnected) return;
if (oldView) oldView.destroy();
if (oldView) oldView.destroy();
host.textContent = '';
host.textContent = '';
+ // A Run job on the person's machine cannot be interrupted mid-run, so a message to a live task waits
+ // for the next attempt (POST /message, 202); the reload shows it waiting under Live.
+ var live = section.hasAttribute('data-events');
var view = window.OpenVibeAgentSession.mount(host, {
var view = window.OpenVibeAgentSession.mount(host, {
title: run.title || section.getAttribute('data-title') || 'Task', status: run.status || 'running',
title: run.title || section.getAttribute('data-title') || 'Task', status: run.status || 'running',
- follow: section.hasAttribute('data-events'), files: false,
- onSend: section.getAttribute('data-owner') === '1' && !section.hasAttribute('data-events') ? function (text) {
- return fetch('/api/v1/tasks/' + encodeURIComponent(id) + '/continue', {
+ follow: live, files: false,
+ sayPlaceholder: live ? 'Message the run: it starts the next attempt' : undefined,
+ onSend: section.getAttribute('data-owner') === '1' ? function (text) {
+ return fetch('/api/v1/tasks/' + encodeURIComponent(id) + (live ? '/message' : '/continue'), {
method: 'POST', credentials: 'same-origin', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ note: text })
method: 'POST', credentials: 'same-origin', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ note: text })
- }).then(function (r) { if (!r.ok) throw new Error('The task could not take the message yet.'); });
+ }).then(function (r) {
+ if (!r.ok) throw new Error('The task could not take the message yet.');
+ if (r.status === 202) window.location.reload();
+ });
} : undefined
} : undefined
});
});
section.dataset.agentUi = '1';
section.dataset.agentUi = '1';
@@ -80,6 +80,12 @@ const TASK_KEYED = [
@@ -80,6 +80,12 @@ const TASK_KEYED = [
},
},
// The approvals the person decided, and the ones still waiting when their account is erased.
// The approvals the person decided, and the ones still waiting when their account is erased.
{ table: 'codes_approvals', file: 'approvals.json', columns: 'l.id, l.task_id, l.tool, l.summary, l.decision, l.message, l.created_at, l.decided_at', order: 'l.created_at ASC', kind: 'approvals', max: TASK_ROWS_MAX },
{ table: 'codes_approvals', file: 'approvals.json', columns: 'l.id, l.task_id, l.tool, l.summary, l.decision, l.message, l.created_at, l.decided_at', order: 'l.created_at ASC', kind: 'approvals', max: TASK_ROWS_MAX },
+ // A message sent while an attempt ran (ask 354): what the person said, taken by the next attempt.
+ {
+ table: 'codes_task_messages', file: 'task-messages.json',
+ columns: 'l.id, l.task_id, l.n, l.text, l.state, l.created_at',
+ order: 'l.n ASC', kind: 'task_messages', max: TASK_ROWS_MAX,
+ },
{ table: 'codes_events_inbox', file: null, kind: 'events_received' },
{ table: 'codes_events_inbox', file: null, kind: 'events_received' },
// The one notice sent per task and state (Codes C9c): no content of its own, erased with the task, not exported.
// The one notice sent per task and state (Codes C9c): no content of its own, erased with the task, not exported.
{ table: 'codes_notices', file: null, kind: 'notices' },
{ table: 'codes_notices', file: null, kind: 'notices' },
@@ -962,15 +962,33 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
@@ -962,15 +962,33 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
try { await writeSessionReport(saved, collected, ending); } catch (e) {
try { await writeSessionReport(saved, collected, ending); } catch (e) {
log.warn('[Codes] the session report of task ' + row.id + ' failed: ' + String((e && e.message) || e).slice(0, 200));
log.warn('[Codes] the session report of task ' + row.id + ' failed: ' + String((e && e.message) || e).slice(0, 200));
}
}
+ // A message sent while the attempt ran carries the task on after all: the engine is done, the person is
+ // not. This is the manual continue the owner would press next, with the waiting notes taken into it.
+ const waiting = await pendingMessages(row.id);
+ if (waiting.length) {
+ const fresh = await db.maybe('SELECT * FROM run_tasks WHERE id = $1', [row.id]);
+ const carried = fresh ? await continueTask(fresh, {}) : { error: 'no such task' };
+ if (!carried.error && carried.job) {
+ log.info(`[Codes] task ${row.id}: ${waiting.length} waiting message(s) started the next attempt`);
+ return carried.row;
+ }
+ if (carried.error) log.warn(`[Codes] task ${row.id}: waiting messages could not carry it on: ${carried.error}`);
+ }
return saved;
return saved;
}
}
if (!plan.kind) return row;
if (!plan.kind) return row;
+ // A message sent while the attempt ran rides the follow-up the engine planned: the person is answered.
+ // Marked delivered only once the job is submitted; a follow-up that never starts leaves the queue for the
+ // owner's next press, and a message arriving after this read stays queued for the attempt after that.
+ const riding = await pendingMessages(row.id);
+ if (riding.length) plan.prompt = `${plan.prompt || ''}${notesSection(riding)}`;
// The goal's budgets are checked before the next turn starts: a spent budget ends the task here, with its reason.
// The goal's budgets are checked before the next turn starts: a spent budget ends the task here, with its reason.
if (goal) {
if (goal) {
const hit = goalBudgetHit(attempts, asObject(row.goal) || goal, row);
const hit = goalBudgetHit(attempts, asObject(row.goal) || goal, row);
if (hit) return await finishGoal(row, { outcome: 'unfinished', note: `stopped by budget: ${hit}`, end_state: 'stopped_by_budget' });
if (hit) return await finishGoal(row, { outcome: 'unfinished', note: `stopped by budget: ${hit}`, end_state: 'stopped_by_budget' });
}
}
const done = await startFollowUp(row, plan, ending);
const done = await startFollowUp(row, plan, ending);
+ if (riding.length && done.created && done.job) await db.exec(`UPDATE codes_task_messages SET state = 'delivered' WHERE task_id = $1 AND state = 'queued' AND id = ANY($2)`, [row.id, riding.map((m) => m.id)]);
return done.row || row;
return done.row || row;
}
}
@@ -1068,6 +1086,61 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
@@ -1068,6 +1086,61 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
return { attempt, created: true, job, row: updated };
return { attempt, created: true, job, row: updated };
}
}
+ /**
+ * A message to the task while it works. A Run job on the person's own machine cannot be interrupted mid-run
+ * (Run has no job-input channel), so a message sent while an attempt is live waits in codes_task_messages — the
+ * conductor's turn queue works the same way. The next attempt carries the waiting notes in its prompt: the
+ * engine's own follow-up takes them, and when the engine ends the task instead, afterEnd carries it on with
+ * them. Nothing the person sent is lost; on a stopped task the queue waits until the owner carries it on.
+ */
+ const MAX_MESSAGE = 4000;
+ function cleanMessage(text) {
+ const t = typeof text === 'string' ? text.trim() : '';
+ if (!t) return { error: 'a message needs words' };
+ if (t.length > MAX_MESSAGE) return { error: `a message is at most ${MAX_MESSAGE} characters` };
+ return { text: t };
+ }
+ /** The messages still waiting for the next attempt, oldest first (also what the task page shows). */
+ async function pendingMessages(taskId) {
+ if (!taskId) return [];
+ return db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 AND state = $2 ORDER BY n ASC', [taskId, 'queued']);
+ }
+ /** Take the waiting messages inside the attempt's own transaction: taken exactly once, by the attempt that runs. */
+ async function takeMessages(taskId) {
+ const rows = await db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 AND state = $2 ORDER BY n ASC FOR UPDATE', [taskId, 'queued']);
+ if (rows.length) await db.exec('UPDATE codes_task_messages SET state = $2 WHERE task_id = $1 AND state = $3', [taskId, 'delivered', 'queued']);
+ return rows;
+ }
+ function notesSection(rows) {
+ return `\n\n---\n## Notes from the owner (sent while the last attempt ran)\n\n${rows.map((m) => String(m.text)).join('\n\n---\n\n')}\n`;
+ }
+ async function queueMessage(row, text) {
+ const n = await db.one('SELECT COALESCE(MAX(n), 0) + 1 AS n FROM codes_task_messages WHERE task_id = $1', [row.id]);
+ return db.one(`INSERT INTO codes_task_messages (id, task_id, n, text, state) VALUES ($1, $2, $3, $4, 'queued') RETURNING *`,
+ [store.newId('msg'), row.id, Number(n.n), text]);
+ }
+ /** A message to the task: it waits while an attempt is live, and carries the task on when nothing runs. */
+ async function messageTask(row, { note = null } = {}) {
+ if (!row || !row.id) return { error: 'no such task' };
+ const clean = cleanMessage(note);
+ if (clean.error) return { error: clean.error };
+ const attempts = await attemptStore.list(row.id);
+ if (!attempts.length) return { error: 'the task has no attempt to carry on' };
+ const live = attempts.slice().reverse().find((a) => !ENDED.has(a.state));
+ if (live) {
+ const message = await queueMessage(row, clean.text);
+ log.info(`[Codes] task ${row.id}: a message waits for the next attempt`);
+ return { queued: true, message, attempt: live, row };
+ }
+ const carried = await continueTask(row, { note: clean.text });
+ if (carried.live) {
+ // An attempt started between the check and the carry-on: the message waits after all.
+ const message = await queueMessage(carried.row || row, clean.text);
+ return { queued: true, message, attempt: carried.attempt, row: carried.row || row };
+ }
+ return carried;
+ }
+
/**
/**
* The next attempt now: the owner carries the task on (with a note), or switches to another agent. The task's row
* 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.
* is locked while the attempt is written, so two presses start one attempt and not two.
@@ -1125,6 +1198,13 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
@@ -1125,6 +1198,13 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
if (!current) return { error: 'no such task' };
if (!current) return { error: 'no such task' };
if (!TERMINAL.has(current.state)) return { error: 'the task is still running' };
if (!TERMINAL.has(current.state)) return { error: 'the task is still running' };
if (current.outcome === 'stopped') return { error: 'the task was stopped on purpose' };
if (current.outcome === 'stopped') return { error: 'the task was stopped on purpose' };
+ // A message sent while the last attempt ran rides this one: taken here, in the attempt's own
+ // transaction, so exactly one attempt carries each message.
+ const waiting = await takeMessages(row.id);
+ if (waiting.length) {
+ plan.prompt += notesSection(waiting);
+ if (!text) plan.why = `carried on by the owner: ${String(waiting[0].text).slice(0, 200)}`;
+ }
const made = await attemptStore.create({
const made = await attemptStore.create({
task: current, agent: plan.agent, model: plan.model, effort: plan.effort || null, session: null, why: plan.why, kind: plan.kind, state: 'queued', budget: left.left,
task: current, agent: plan.agent, model: plan.model, effort: plan.effort || null, session: null, why: plan.why, kind: plan.kind, state: 'queued', budget: left.left,
});
});
@@ -1260,6 +1340,7 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
@@ -1260,6 +1340,7 @@ ${String(verdict.check_output || '').slice(0, 50000)}`;
return {
return {
start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,
start, refresh, applyJob, applyEvent, recordState, afterEnd, continueTask, stopTask, proof: taskProof,
latestProject, present, who, attempts: attemptStore, setConductor, attachBindings, fundTurn,
latestProject, present, who, attempts: attemptStore, setConductor, attachBindings, fundTurn,
+ messageTask, pendingMessages,
setForge(f) { forge = f || null; },
setForge(f) { forge = f || null; },
};
};
}
}
@@ -339,6 +339,9 @@ ${i.summary ? html`<p class="run-proof-summary" data-proof-summary>${i.summary}<
@@ -339,6 +339,9 @@ ${i.summary ? html`<p class="run-proof-summary" data-proof-summary>${i.summary}<
const lines = live ? null : await sessionStore.linesForRow(row);
const lines = live ? null : await sessionStore.linesForRow(row);
const title = String(row.prompt || '').split('\n')[0].slice(0, 90);
const title = String(row.prompt || '').split('\n')[0].slice(0, 90);
const session = lines && lines.length ? html`<section class="run-session" data-session data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" data-state="${row.state}"><h2>The session</h2><div data-session-view></div><script type="application/json" data-session-log>${raw(JSON.stringify(lines).replace(/</g, '\\u003c'))}</script></section>` : '';
const session = lines && lines.length ? html`<section class="run-session" data-session data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" data-state="${row.state}"><h2>The session</h2><div data-session-view></div><script type="application/json" data-session-log>${raw(JSON.stringify(lines).replace(/</g, '\\u003c'))}</script></section>` : '';
+ // A message sent while the attempt runs waits for the next one (ask 354): say so under Live.
+ const waiting = live && mine ? await taskStore.pendingMessages(row.id) : [];
+ const waitingLine = waiting.length ? html`<p class="run-pending" data-pending>${waiting.length === 1 ? 'A message waits' : `${waiting.length} messages wait`} for the next attempt</p>` : '';
const body = html`${shareHead ? raw(shareHead) : ''}<p class="crumbs"><a href="/run">Run locally</a></p>
const body = html`${shareHead ? raw(shareHead) : ''}<p class="crumbs"><a href="/run">Run locally</a></p>
<h1 class="run-repo">${repoName(row.repo)}${row.ref ? html` <small>${row.ref}</small>` : ''}</h1>
<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>
<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>
@@ -347,7 +350,7 @@ ${mine ? shareButton(row) : html`<p class="muted">This task was shared with you.
@@ -347,7 +350,7 @@ ${mine ? shareButton(row) : html`<p class="muted">This task was shared with you.
${await timelineHtml(row)}
${await timelineHtml(row)}
${goalHtml(row)}
${goalHtml(row)}
${await proofHtml(row)}
${await proofHtml(row)}
-${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events" data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" aria-live="polite"><h2>Live</h2><ol class="run-log" data-log></ol></section>
+${live ? html`<section class="run-live" data-events="/run/tasks/${row.id}/events" data-run-id="${row.id}" data-owner="${mine ? '1' : '0'}" data-title="${title}" aria-live="polite"><h2>Live</h2>${waitingLine}<ol class="run-log" data-log></ol></section>
${mine ? html`<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) })}`}`;
${mine ? html`<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'] });
return page(req, res, { title: 'Task', body, sharedScripts: ['agent-session.js'] });
});
});
@@ -571,6 +571,29 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
@@ -571,6 +571,29 @@ function createTasksApi({ config, store, keys, run, harnesses, actorLimits, task
return res.json(await timeline(out.row));
return res.json(await timeline(out.row));
});
});
+ /**
+ * A message to the task (ask 354): while an attempt is live the message waits for the next attempt (202, nothing
+ * lost — a Run job on the person's own machine cannot be interrupted mid-run); when nothing runs it carries the
+ * task on at once, like a continue with a note.
+ */
+ r.post('/tasks/:id/message', 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 text = typeof b.text === 'string' ? b.text : (typeof b.note === 'string' ? b.note : null);
+ const out = await taskStore.messageTask(row, { note: text });
+ if (out.error) {
+ const status = out.error === 'no such task' ? 404 : /needs words|at most/.test(out.error) ? 422 : 409;
+ return problem(req, res, status,
+ status === 404 ? 'codes.task_not_found' : status === 422 ? 'codes.message_invalid' : 'codes.task_conflict', out.error);
+ }
+ if (out.queued) {
+ return res.status(202).json({ queued: true, message: { id: out.message.id, n: out.message.n, text: out.message.text } });
+ }
+ log.info(`[Codes] task ${row.id} carried on by a message`);
+ return res.json(await timeline(out.row));
+ });
+
/**
/**
* The running attempt stops: the outcome is "stopped" and nothing carries it on. The owner (the Dock's shape)
* The running attempt stops: the outcome is "stopped" and nothing carries it on. The owner (the Dock's shape)
* gets { ok }; a service keeps the timeline its capability was built on. Another person's task is 404, never
* gets { ok }; a service keeps the timeline its capability was built on. Another person's task is 404, never
new file mode 100644
new file mode 100644
@@ -0,0 +1,171 @@
@@ -0,0 +1,171 @@
+'use strict';
+/**
+ * A message to a task (POST /api/v1/tasks/:id/message, ask 354): while an attempt is live the message waits in
+ * codes_task_messages (202, nothing lost — a Run job on the person's machine cannot be interrupted mid-run); when
+ * the attempt ends the next attempt carries the waiting notes in its prompt, whether the engine planned a follow-up
+ * or ended the task. When nothing runs the message carries the task on at once, like a continue with a note. A
+ * stopped task refuses (409) and keeps its queue.
+ */
+const assert = require('assert');
+const { boot, check, done } = require('./helpers/boot');
+
+(async () => {
+ const submits = [];
+ let counter = 0;
+ const jobs = new Map();
+ const run = {
+ enabled: true,
+ async submit(o) {
+ counter++;
+ const id = `job_01J8Z3Q4R5S6T7V8W9X0Y1Z2J${String(counter).padStart(2, '0')}`;
+ jobs.set(id, { id, state: 'queued', args: o.args });
+ submits.push(o);
+ return { id, state: 'queued' };
+ },
+ async read(jobId) { return jobs.get(jobId) || { id: jobId, state: 'queued' }; },
+ async cancel(jobId) { const j = jobs.get(jobId); if (j) j.state = 'cancelled'; return j || { id: jobId, state: 'cancelled' }; },
+ async streamUrl() { throw new Error('not used'); },
+ };
+ const t = await boot({ run });
+ const db = t.ctx.store.db;
+ const ada = t.network.addUser('ada');
+ const bo = t.network.addUser('bo');
+ const actor = t.network.mintService('svc:actor', 'openvibe.codes', ['codes.task.start', 'codes.task.read', 'codes.task.cancel']);
+ const project = 'prj_01J8Z3Q4R5S6T7V8W9X0Y1Z2P1';
+ // Ada ran a task on /run before (so Codes knows her project); Bo never did.
+ await db.exec(`INSERT INTO run_tasks (id, subject, project_id, job_id, repo, prompt, state, created_at, updated_at)
+ VALUES ('tsk_01J8Z3Q4R5S6T7V8W9X0Y1Z2T1', $1, $2, 'job_01J8Z3Q4R5S6T7V8W9X0Y1Z200', 'https://github.com/a/b', 'by hand', 'succeeded', 1, 2)`, [ada.subject, project]);
+
+ const asAda = async (method, path, body, token = t.network.userToken(ada)) => {
+ const headers = { Accept: 'application/json', Authorization: `Bearer ${token}` };
+ if (body !== undefined) headers['Content-Type'] = 'application/json';
+ const res = await fetch(`${t.base}/api/v1${path}`, { method, headers, body: body === undefined ? undefined : JSON.stringify(body), redirect: 'manual' });
+ return { status: res.status, json: await res.json().catch(() => null) };
+ };
+
+ let seq = 0;
+ const startTask = async () => {
+ seq++;
+ const r = await asAda('POST', '/tasks', { repo: 'https://github.com/octocat/Hello-World', prompt: 'Add a line to the README.', agent: 'deepseek', idempotency_key: `act_msg_${seq}` });
+ assert.strictEqual(r.status, 201, JSON.stringify(r.json));
+ return r.json.task.id;
+ };
+ const attemptsOf = async (taskId) => db.many('SELECT * FROM codes_attempts WHERE task_id = $1 ORDER BY n ASC', [taskId]);
+ const messagesOf = async (taskId) => db.many('SELECT * FROM codes_task_messages WHERE task_id = $1 ORDER BY n ASC', [taskId]);
+ /** The attempt ends the way Run delivers it, then the engine decides what happens next. */
+ const endAttempt = async (taskId, ok, text) => {
+ const row = await db.one('SELECT * FROM run_tasks WHERE id = $1', [taskId]);
+ const job = {
+ id: row.job_id, state: ok ? 'succeeded' : 'failed',
+ result: {
+ ok, summary: text, files: [],
+ log: [{ t: 'agent', agent: 'deepseek', session: 'sess-1' }, { t: 'done', ok, text }],
+ },
+ ...(ok ? {} : { error: { code: 'run.job.worker_failed', detail: text } }),
+ };
+ const updated = await t.ctx.tasks.applyEvent(row, { state: ok ? 'succeeded' : 'failed', job, error: ok ? null : text });
+ return t.ctx.tasks.afterEnd(updated.id);
+ };
+
+ try {
+ await check('a message to a live task waits: 202, queued, no new attempt, no job', async () => {
+ const id = await startTask();
+ const before = submits.length;
+ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'also check the spelling' });
+ assert.strictEqual(r.status, 202, JSON.stringify(r.json));
+ assert.strictEqual(r.json.queued, true);
+ assert.strictEqual(r.json.message.n, 1);
+ assert.strictEqual(r.json.message.text, 'also check the spelling');
+ assert.strictEqual((await attemptsOf(id)).length, 1, 'no new attempt while one runs');
+ assert.strictEqual(submits.length, before, 'no job while one runs');
+ const rows = await messagesOf(id);
+ assert.strictEqual(rows.length, 1);
+ assert.strictEqual(rows[0].state, 'queued');
+ });
+
+ await check('a second message joins the queue in order', async () => {
+ const id = await startTask();
+ await asAda('POST', `/tasks/${id}/message`, { text: 'first' });
+ const r = await asAda('POST', `/tasks/${id}/message`, { note: 'second' });
+ assert.strictEqual(r.status, 202, JSON.stringify(r.json));
+ assert.strictEqual(r.json.message.n, 2);
+ const rows = await messagesOf(id);
+ assert.deepStrictEqual(rows.map((m) => m.text), ['first', 'second']);
+ assert.ok(rows.every((m) => m.state === 'queued'));
+ });
+
+ await check('when the attempt crashes, the follow-up carries the waiting notes', async () => {
+ const id = await startTask();
+ await asAda('POST', `/tasks/${id}/message`, { text: 'also check the spelling' });
+ await asAda('POST', `/tasks/${id}/message`, { text: 'and the license header' });
+ const before = submits.length;
+ await endAttempt(id, false, 'the machine went away');
+ const attempts = await attemptsOf(id);
+ assert.strictEqual(attempts.length, 2, 'the task was carried on');
+ assert.strictEqual(submits.length, before + 1, 'one follow-up job');
+ const prompt = String(submits[submits.length - 1].args.prompt);
+ assert.ok(prompt.includes('Notes from the owner (sent while the last attempt ran)'), 'the notes ride the follow-up');
+ assert.ok(prompt.includes('also check the spelling') && prompt.includes('and the license header'), 'both notes');
+ assert.ok((await messagesOf(id)).every((m) => m.state === 'delivered'), 'taken exactly once');
+ });
+
+ await check('when the engine ends the task with notes waiting, it carries on by itself', async () => {
+ const id = await startTask();
+ await asAda('POST', `/tasks/${id}/message`, { text: 'double-check the edge cases' });
+ const before = submits.length;
+ await endAttempt(id, true, 'done, all green');
+ const attempts = await attemptsOf(id);
+ assert.strictEqual(attempts.length, 2, 'the waiting message started the next attempt');
+ assert.strictEqual(submits.length, before + 1, 'one follow-up job');
+ assert.ok(String(submits[submits.length - 1].args.prompt).includes('double-check the edge cases'), 'the note rides it');
+ assert.ok((await messagesOf(id)).every((m) => m.state === 'delivered'), 'taken exactly once');
+ });
+
+ await check('a message to an ended task carries it on at once, like a continue with a note', async () => {
+ const id = await startTask();
+ await endAttempt(id, true, 'done, all green');
+ assert.strictEqual((await attemptsOf(id)).length, 1, 'the engine left the finished task alone');
+ const before = submits.length;
+ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'and the tests' });
+ assert.strictEqual(r.status, 200, JSON.stringify(r.json));
+ assert.strictEqual((await attemptsOf(id)).length, 2, 'the message is the next attempt');
+ assert.strictEqual(submits.length, before + 1, 'one job');
+ assert.ok(String(submits[submits.length - 1].args.prompt).includes('and the tests'), 'the note rides it');
+ });
+
+ await check('a stopped task refuses and keeps its queue: 409, nothing starts', async () => {
+ const id = await startTask();
+ await asAda('POST', `/tasks/${id}/message`, { text: 'wait for me' });
+ const stopped = await asAda('POST', `/tasks/${id}/stop`);
+ assert.strictEqual(stopped.status, 200, JSON.stringify(stopped.json));
+ const before = submits.length;
+ const r = await asAda('POST', `/tasks/${id}/message`, { text: 'one more thing' });
+ assert.strictEqual(r.status, 409, JSON.stringify(r.json));
+ assert.strictEqual(r.json.code, 'codes.task_conflict');
+ assert.strictEqual(submits.length, before, 'no job on a stopped task');
+ assert.strictEqual((await attemptsOf(id)).length, 1, 'no attempt on a stopped task');
+ assert.deepStrictEqual((await messagesOf(id)).map((m) => m.state), ['queued'], 'the queue waits for a hand carry-on');
+ });
+
+ await check('empty and too long are 422; another person and a missing task are 404', async () => {
+ const id = await startTask();
+ assert.strictEqual((await asAda('POST', `/tasks/${id}/message`, { text: ' ' })).status, 422);
+ assert.strictEqual((await asAda('POST', `/tasks/${id}/message`, { text: 'x'.repeat(4001) })).status, 422);
+ const other = await asAda('POST', `/tasks/${id}/message`, { text: 'hello' }, t.network.userToken(bo));
+ assert.strictEqual(other.status, 404);
+ assert.strictEqual((await asAda('POST', '/tasks/tsk_01J8Z3Q4R5S6T7V8W9X0Y9Z9Q9/message', { text: 'hello' })).status, 404);
+ });
+
+ await check('the live task page says a message waits for the next attempt', async () => {
+ const id = await startTask();
+ await asAda('POST', `/tasks/${id}/message`, { text: 'wait for me' });
+ const page = await t.get(`/run/tasks/${id}`, { as: ada });
+ assert.strictEqual(page.status, 200);
+ assert.ok(page.text.includes('data-pending'), 'the waiting line is on the page');
+ assert.ok(page.text.includes('A message waits for the next attempt'), 'in plain words');
+ });
+ } finally {
+ await t.close();
+ }
+ done();
+})().catch((e) => { console.error(e); process.exit(1); });