openvibers/openvibe.actor Commits

A task's end is published last, and a goal's counters move in one statement (#341)

OP OpenVibers committed 5 days ago · 141e64d6b3 · 3 files, +70 −28

The task's own row is the signal everyone waits for: the Dock, the chat, the person's
notification. Everything a reader of that signal needs is now written, and awaited, before
the row flips: the end event, the assignment tree, the trigger's card and the routine's
answer in the thread, and the notification. The needs-you notification goes out before the
task says it is waiting, for the same reason.

The trigger's counters (the failure streak, and a goal's runs and money) move in one
UPDATE ... RETURNING, so the card is read from the count that was really reached and two
runs that settle at once cannot count one run twice.

PostgreSQL's pool hands out its connections as they come, so the old order committed
"succeeded" while the trigger's card still said "runs_used 0" and the notification was
still on its way: a reader that saw the task was done read a card and a bell that had not
caught up, a state that never existed. PGlite keeps every statement in order, which is
why only the pg store showed it.

Changes

ddiff --git a/server/tasks/engine.js b/server/tasks/engine.js
iindex f3eac89..14e236d 100644
---- a/server/tasks/engine.js
++++ b/server/tasks/engine.js
@@@ -159,31 +159,46 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
stream.publish(id, saved);
return saved;
}
++ /**
++ * A task's state, and everything that hangs off it.
++ *
++ * The task's own row is the signal everyone waits for: the Dock, the chat, the person's notification.
++ * So everything a reader of that signal needs is written, and awaited, first — the event, the
++ * assignment tree, the trigger's card and its answer in the thread, the notification — and the row
++ * flips last. The PostgreSQL pool hands out its connections as they come, so a task that read
++ * "succeeded" while its trigger's card still said "0 runs" would be a state that never existed;
++ * PGlite, which keeps every statement in order, would never have shown it.
++ */
async function setState(id, state, { agentForEvent, ...patch } = {}) {
-- const row = await store.updateTask(s, id, { state, ...patch });
const ended = ['succeeded', 'failed', 'cancelled'].includes(state);
-- await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });
-- // The assignment tree follows the task it runs (plan §12 A-2), and its root's spend is written with it.
-- if (row) await assignments.syncTaskState(s, row).catch((err) => log.warn && log.warn('[Actor] assignment state:', id, (err && err.message) || err));
-- // A trigger's card follows its run: a failure streak grows, five in a row pause it, a success clears it.
-- if (row && triggers) await triggers.taskState(row).catch((err) => log.warn && log.warn('[Actor] trigger state:', id, (err && err.message) || err));
++ // The row as it will be once this state is written: the trigger's card and the notification read it.
++ const before = await store.getTask(s, id);
++ const row = before ? { ...before, ...patch, state } : null;
++ if (row) {
++ await emit(id, { kind: ended ? 'end' : 'state', state, ...(agentForEvent ? { agent: agentForEvent } : {}) });
++ // The assignment tree follows the task it runs (plan §12 A-2), and its root's spend is written with it.
++ await assignments.syncTaskState(s, row).catch((err) => log.warn && log.warn('[Actor] assignment state:', id, (err && err.message) || err));
++ // A trigger's card follows its run: a failure streak grows, five in a row pause it, a success clears it.
++ if (triggers) await triggers.taskState(row).catch((err) => log.warn && log.warn('[Actor] trigger state:', id, (err && err.message) || err));
++ // "Tell me when it's done" (stage V2-6): done or failed, once per task. Cancelled is the person's own action.
++ // A routine's run is the exception (plan HO4): its answer already goes into the conversation that made the
++ // trigger, so the person is told through notifications only when the routine asked for it.
++ if (ended && (state === 'succeeded' || state === 'failed')) {
++ const routine = triggers && typeof triggers.tellsPerson === 'function' ? await triggers.tellsPerson(row).catch(() => null) : null;
++ if (routine !== false) await tellPerson(row, state);
++ }
++ }
++ const saved = await store.updateTask(s, id, { state, ...patch });
if (ended) runTokens.drop(id);
if (ended) stream.end(id);
-- // "Tell me when it's done" (stage V2-6): done or failed, once per task. Cancelled is the person's own action.
-- // A routine's run is the exception (plan HO4): its answer already goes into the conversation that made the
-- // trigger, so the person is told through notifications only when the routine asked for it.
-- if (ended && row && (state === 'succeeded' || state === 'failed')) {
-- const routine = triggers && typeof triggers.tellsPerson === 'function' ? await triggers.tellsPerson(row).catch(() => null) : null;
-- if (routine !== false) await tellPerson(row, state);
-- }
// Webhooks that asked for this state hear it; a delivery never holds the task up or fails it.
-- if (webhooks && row && row.webhooks) {
++ if (webhooks && saved && saved.webhooks) {
try {
await webhooks.enqueue(id, state);
if (ended) await webhooks.forgetSecretIfDone(id);
} catch (err) { log.warn('[Actor] webhook enqueue failed:', id, err && err.message); }
}
-- return row;
++ return saved;
}
/**
@@@ -321,8 +336,11 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
kind: 'output', step: 'question', agent: LOOP_AGENT, text: clean, options: list,
...(questionKind === 'acting' ? { question_kind: 'acting', tool: tool || null, input: input || null } : {}),
});
-- await setState(id, 'needs_you');
++ // The person is told before the task says it needs them: a reader that sees the task waiting
++ // for an answer must find the notification already sent, the same way the run's own end is
++ // published last (see setState).
await tellPerson(row, 'needs_you', { what: `${questionKind === 'acting' ? 'choose for an interrupted action' : 'answer'}: ${clean}`, url: `${config.baseUrl}/tasks/${id}` });
++ await setState(id, 'needs_you');
waitForPerson(15 * 60_000);
const answer = await waiting;
asking.delete(id);
ddiff --git a/server/triggers/service.js b/server/triggers/service.js
iindex 9d9dcdb..ee373a4 100644
---- a/server/triggers/service.js
++++ b/server/triggers/service.js
@@@ -493,26 +493,33 @@ function createTriggers({ s, config, engine = null, webhooks = null, stream = nu
const cost = Number(task.cost_usd) || 0;
await store.addRun(s, run.trigger_id, task.id, mapped, nowMs(), cost);
if (mapped === 'running') return row;
-- let failures = Number(found.failures) || 0;
-- if (mapped === 'failed' && task.state !== 'cancelled') failures += 1;
-- if (mapped === 'ok') failures = 0;
-- const patch = {
-- last_task_id: task.id, last_state: mapped, last_run_at: nowMs(), failures, updated_at: nowMs(),
-- ...(failures >= 5 ? { state: 'paused', paused_reason: 'failed' } : {}),
-- };
// The answer in the thread comes first, so a goal's closing line reads after the run it ends.
await postRun(row, task, mapped);
++ // The counters move in one statement, and the row it returns is the truth: the card is read
++ // from it, and the ending is decided on the count that was really reached. A count read before
++ // this write (and then written back as an absolute number) loses a run when two settle at once.
++ const goal = row.kind === 'goal' && (mapped === 'ok' || mapped === 'failed');
++ const after = await store.bumpCounters(s, row.id, {
++ state: task.state, runs: goal ? 1 : 0, costUsd: goal ? cost : 0, at: nowMs(),
++ }) || row;
++ const failures = Number(after.failures) || 0;
++ const patch = {
++ last_task_id: task.id, last_state: mapped, last_run_at: nowMs(), updated_at: nowMs(),
++ ...(failures >= 5 ? { state: 'paused', paused_reason: 'failed' } : {}),
++ };
++
let ended = null;
-- if (row.kind === 'goal' && (mapped === 'ok' || mapped === 'failed')) {
++ if (goal) {
const result = tasks.parseJson(task.result) || {};
const verdict = mapped === 'ok' ? goalLib.verdictOf(result.answer) : { done: null, reason: '' };
-- patch.runs_used = (Number(row.runs_used) || 0) + 1;
-- patch.spent_usd = usd((Number(row.spent_usd) || 0) + cost);
if (verdict.done != null) patch.verdict = { done: verdict.done, reason: verdict.reason, at: iso(nowMs()) };
if (row.state === 'active') {
-- ended = goalLib.endingOf({ goal: row.goal, verdict, runsUsed: patch.runs_used, spentUsd: patch.spent_usd, at: nowMs() });
++ ended = goalLib.endingOf({
++ goal: row.goal, verdict,
++ runsUsed: Number(after.runs_used) || 0, spentUsd: usd(after.spent_usd), at: nowMs(),
++ });
if (ended) Object.assign(patch, { state: ended.reason === 'done' ? 'done' : 'paused', paused_reason: ended.reason, ended_at: nowMs() });
}
}
ddiff --git a/server/triggers/store.js b/server/triggers/store.js
iindex e4be139..1c0410f 100644
---- a/server/triggers/store.js
++++ b/server/triggers/store.js
@@@ -72,6 +72,23 @@ async function update(s, owner, id, patch) {
return await get(s, owner, id);
}
++/**
++ * The card's counters, moved in one statement: the failure streak, and a goal's runs and money.
++ * The row that comes back (RETURNING) is the truth the card is read from: two runs that settle at
++ * once are counted once, and the ending is decided on the count that was really reached, not on a
++ * count read before the other run wrote its own.
++ */
++async function bumpCounters(s, id, { state = null, runs = 0, costUsd = 0, at = null } = {}) {
++ const row = await s.db.maybe(`UPDATE actor_triggers SET
++ failures = CASE WHEN $2 = 'succeeded' THEN 0 WHEN $2 = 'failed' THEN failures + 1 ELSE failures END,
++ runs_used = runs_used + $3,
++ spent_usd = spent_usd + $4,
++ updated_at = $5
++ WHERE id = $1 AND deleted_at IS NULL
++ RETURNING *`, [id, state, runs, costUsd, at == null ? Date.now() : at]);
++ return wire(row);
++}
++
async function addRun(s, triggerId, taskId, state, at, costUsd = null) {
await s.db.query(`INSERT INTO actor_trigger_runs (trigger_id, task_id, state, cost_usd, created_at, updated_at)
VALUES ($1,$2,$3,$4,$5,$5) ON CONFLICT (task_id) DO UPDATE SET state = EXCLUDED.state, cost_usd = COALESCE(EXCLUDED.cost_usd, actor_trigger_runs.cost_usd), updated_at = EXCLUDED.updated_at`,
@@@ -85,4 +102,4 @@ const remove = async (s, owner, id) => {
await s.db.query('DELETE FROM actor_triggers WHERE id = $1 AND owner = $2', [id, owner]);
};
--module.exports = { wire, get, byWatch, list, countActive, countForAgent, insert, update, addRun, runForTask, runs, removeRuns, remove };
++module.exports = { wire, get, byWatch, list, countActive, countForAgent, insert, update, bumpCounters, addRun, runForTask, runs, removeRuns, remove };