openvibers/openvibe.actor Commits
Record why Actor trigger runs started (#434)
Changes
nnew file mode 100644@@@ -0,0 +1,3 @@++-- phase: expand++-- Explain why a schedule or event subscription started each task.++ALTER TABLE actor_trigger_runs ADD COLUMN IF NOT EXISTS why text;@@@ -960,7 +960,11 @@ function createApi(ctx) {r.get('/triggers/:id', principal.requireCapability('actor.task.read'), limits.reads('actor.task.read'), personOnly, async (req, res) => {const card = await triggers.get(triggerOwner(req), req.params.id, { token: watchToken(req), live: true });if (!card) return problem(req, res, 404, 'actor.trigger.not_found', 'No such trigger.');-- res.json({ trigger: card });++ if (req.query.runs !== '1') return res.json({ trigger: card });++ const rows = await require('../triggers/store').runs(s, card.id, 20);++ res.json({ trigger: card, runs: rows.map((row) => ({ task_id: row.task_id, state: row.state, why: row.why || null,++ cost_usd: row.cost_usd == null ? null : Number(row.cost_usd), created_at: new Date(Number(row.created_at)).toISOString(),++ updated_at: new Date(Number(row.updated_at)).toISOString() })) });});r.post('/triggers', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {const b = req.body || {};@@@ -1000,7 +1000,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.await emitPlan(state, ctx);return { ok: true, assumptions: state.plan.assumptions };};-- control.pending = [];++ control.pending = Array.isArray(control.pending) ? control.pending : [];control.assume = applyAssumptions;control.steer = steerOf(state, ctx);control.plan = () => state.plan;@@@ -1094,7 +1094,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.// An assumption edit can arrive while the understand and plan calls are still running (the page already shows// the plan of a previous turn). Hold it here and apply it as soon as the run's own handler exists.if (control && typeof control === 'object') {-- control.pending = [];++ control.pending = Array.isArray(control.pending) ? control.pending : [];control.assume = async (text) => { control.pending.push(String(text == null ? '' : text)); return { ok: true, queued: true }; };control.steer = steerOf(state, ctx);}@@@ -191,7 +191,12 @@ async function createTask({ s, config, engine, principal, body, webhooks: webhoo// its first step; an older task without one gets it lazily (assignments.ensureForTask).const node = await assignments.ensureForTask(s, config, row, { origin: assignmentOrigin, level: assignmentLevel });const task = node ? { ...row, assignment_id: node.id } : row;-- if (trigger) await triggerStore.addRun(s, trigger.id, task.id, 'queued', s.now());++ if (trigger) {++ const why = trigger.kind === 'event' ? `Watch event: ${trigger.event && trigger.event.pattern || 'matched'}`++ : trigger.kind === 'schedule' || trigger.kind === 'goal' ? `Schedule: ${trigger.schedule && trigger.schedule.cron || 'due'}`++ : trigger.kind === 'mention' ? 'Conversation mention' : 'Trigger';++ await triggerStore.addRun(s, trigger.id, task.id, 'queued', s.now(), null, why.slice(0, 300));++ }if (afterInsert) await afterInsert(task);if (chatContext) await s.db.query('INSERT INTO actor_chat_tasks (task_id, agent_id, policy, conversation) VALUES ($1, $2, $3, $4)', [task.id, chatContext.agentId, chatContext.policy, chatContext.conversation]);return { status: 201, task };@@@ -531,6 +531,15 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =: `the one agent loop (${row.mode === 'best' ? 'best' : 'smart'} tier)`;explanation.push({ selected: agentId, objective: row.mode === 'best' ? 'correctness' : 'balanced', reasons: [why], candidates: [], decided_at: iso() });await store.updateTask(s, id, { explanation });++ // The plan event makes the assumptions form visible. Install its control first so an edit sent++ // immediately after the page appears is queued even while the loop is still being prepared.++ const control = { pending: [] };++ control.assume = async (text) => {++ control.pending.push(String(text == null ? '' : text).slice(0, 2000));++ if (control.pending.length > 5) control.pending.shift();++ return { ok: true, queued: true };++ };++ controls.set(id, control);if (!resume && !fork) await emit(id, { kind: 'output', step: 'plan', agent: agentId, text: 'The agent starts with a plan, then works through it.' });// A task that would start coding work on the person's machine asks through the same gate (levels.js):// the level first, then the always-ask floor. Full access or auto-edits may let an edit through; a@@@ -546,8 +555,6 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);if (wakePayload) wakes.delete(id);let out = null;-- const control = {};-- controls.set(id, control);// The files attached to this message in the chat (plan L55 C2): text as data; images described first.const attachments = await inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', err && err.message); return []; });// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.@@@ -96,10 +96,10 @@ async function bumpCounters(s, id, { state = null, runs = 0, costUsd = 0, at = nreturn 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`,-- [triggerId, taskId, state, costUsd, at]);++async function addRun(s, triggerId, taskId, state, at, costUsd = null, why = null) {++ await s.db.query(`INSERT INTO actor_trigger_runs (trigger_id, task_id, state, cost_usd, why, created_at, updated_at)++ VALUES ($1,$2,$3,$4,$5,$6,$6) 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`,++ [triggerId, taskId, state, costUsd, why, at]);}const runForTask = (s, taskId) => s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [taskId]);const runs = (s, triggerId, limit = 20) => s.db.many('SELECT * FROM actor_trigger_runs WHERE trigger_id = $1 ORDER BY created_at DESC LIMIT $2', [triggerId, limit]);@@@ -97,10 +97,19 @@ const SAME = { origin: 'https://openvibe.actor' };const created = await t.get('/tasks', { as: bea, form: { task: 'Find three offers under ten euros', mode: 'balanced' }, headers: SAME });const id = taskIdOf(created);await t.waitFor(id, ['running']);++ // A running task may still be preparing its first plan. Wait for the visible plan event, rather than++ // posting to the form while its section is hidden and the loop has no edit handler yet.++ let planned = false;++ for (const until = Date.now() + 60000; !planned && Date.now() < until;) {++ planned = (await store.eventsAfter(s, id, 0)).some((e) => e.step === 'plan' && taskEvents.plan(e.text));++ if (!planned) await new Promise((r) => setTimeout(r, 50));++ }++ assert.ok(planned, 'the live plan is visible before it is edited');const page = await t.get(`/tasks/${id}`, { as: bea });assert.match(page.text, new RegExp(`action="/tasks/${id}/assumptions"`), 'the live plan has an editable assumptions form');const edited = await t.get(`/tasks/${id}/assumptions`, { as: bea, form: { assumptions: 'Use EUR\nAsk before booking' }, headers: SAME });assert.strictEqual(edited.status, 303);++ assert.ok(!String(edited.headers.get('location')).includes('assume=gone'), 'the visible plan accepts an immediate edit');t.providers.state.runtime = 'tool';// Up to 30 s: the follow-up turn can wait behind the runtime queue on a loaded CI runner.let seen = false;@@@ -277,7 +277,11 @@ async function watchStub() {const root = await s.db.maybe('SELECT * FROM actor_assignments WHERE task_id = $1 AND depth = 0', [task.id]);assert.deepStrictEqual([root.origin, root.effective_level], ['routine', 'ask']);const run = await s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [task.id]);++ assert.match(run.why, /^Schedule:/);assert.strictEqual(run.trigger_id, trigger.id, 'the run is linked back to its card');++ const history = await call('GET', `/api/v1/triggers/${trigger.id}?runs=1`);++ assert.strictEqual(history.status, 200, history.text);++ assert.match(history.json().runs[0].why, /^Schedule:/);});} finally {
new file mode 100644
new file mode 100644
@@ -0,0 +1,3 @@
@@ -0,0 +1,3 @@
+-- phase: expand
+-- Explain why a schedule or event subscription started each task.
+ALTER TABLE actor_trigger_runs ADD COLUMN IF NOT EXISTS why text;
@@ -960,7 +960,11 @@ function createApi(ctx) {
@@ -960,7 +960,11 @@ function createApi(ctx) {
r.get('/triggers/:id', principal.requireCapability('actor.task.read'), limits.reads('actor.task.read'), personOnly, async (req, res) => {
r.get('/triggers/:id', principal.requireCapability('actor.task.read'), limits.reads('actor.task.read'), personOnly, async (req, res) => {
const card = await triggers.get(triggerOwner(req), req.params.id, { token: watchToken(req), live: true });
const card = await triggers.get(triggerOwner(req), req.params.id, { token: watchToken(req), live: true });
if (!card) return problem(req, res, 404, 'actor.trigger.not_found', 'No such trigger.');
if (!card) return problem(req, res, 404, 'actor.trigger.not_found', 'No such trigger.');
- res.json({ trigger: card });
+ if (req.query.runs !== '1') return res.json({ trigger: card });
+ const rows = await require('../triggers/store').runs(s, card.id, 20);
+ res.json({ trigger: card, runs: rows.map((row) => ({ task_id: row.task_id, state: row.state, why: row.why || null,
+ cost_usd: row.cost_usd == null ? null : Number(row.cost_usd), created_at: new Date(Number(row.created_at)).toISOString(),
+ updated_at: new Date(Number(row.updated_at)).toISOString() })) });
});
});
r.post('/triggers', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
r.post('/triggers', principal.requireCapability('actor.task.create'), limits.budget('actor.task.create'), personOnly, async (req, res) => {
const b = req.body || {};
const b = req.body || {};
@@ -1000,7 +1000,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -1000,7 +1000,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
await emitPlan(state, ctx);
await emitPlan(state, ctx);
return { ok: true, assumptions: state.plan.assumptions };
return { ok: true, assumptions: state.plan.assumptions };
};
};
- control.pending = [];
+ control.pending = Array.isArray(control.pending) ? control.pending : [];
control.assume = applyAssumptions;
control.assume = applyAssumptions;
control.steer = steerOf(state, ctx);
control.steer = steerOf(state, ctx);
control.plan = () => state.plan;
control.plan = () => state.plan;
@@ -1094,7 +1094,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
@@ -1094,7 +1094,7 @@ function createLoop({ config, models, registry, log = console, now = () => Date.
// An assumption edit can arrive while the understand and plan calls are still running (the page already shows
// An assumption edit can arrive while the understand and plan calls are still running (the page already shows
// the plan of a previous turn). Hold it here and apply it as soon as the run's own handler exists.
// the plan of a previous turn). Hold it here and apply it as soon as the run's own handler exists.
if (control && typeof control === 'object') {
if (control && typeof control === 'object') {
- control.pending = [];
+ control.pending = Array.isArray(control.pending) ? control.pending : [];
control.assume = async (text) => { control.pending.push(String(text == null ? '' : text)); return { ok: true, queued: true }; };
control.assume = async (text) => { control.pending.push(String(text == null ? '' : text)); return { ok: true, queued: true }; };
control.steer = steerOf(state, ctx);
control.steer = steerOf(state, ctx);
}
}
@@ -191,7 +191,12 @@ async function createTask({ s, config, engine, principal, body, webhooks: webhoo
@@ -191,7 +191,12 @@ async function createTask({ s, config, engine, principal, body, webhooks: webhoo
// its first step; an older task without one gets it lazily (assignments.ensureForTask).
// its first step; an older task without one gets it lazily (assignments.ensureForTask).
const node = await assignments.ensureForTask(s, config, row, { origin: assignmentOrigin, level: assignmentLevel });
const node = await assignments.ensureForTask(s, config, row, { origin: assignmentOrigin, level: assignmentLevel });
const task = node ? { ...row, assignment_id: node.id } : row;
const task = node ? { ...row, assignment_id: node.id } : row;
- if (trigger) await triggerStore.addRun(s, trigger.id, task.id, 'queued', s.now());
+ if (trigger) {
+ const why = trigger.kind === 'event' ? `Watch event: ${trigger.event && trigger.event.pattern || 'matched'}`
+ : trigger.kind === 'schedule' || trigger.kind === 'goal' ? `Schedule: ${trigger.schedule && trigger.schedule.cron || 'due'}`
+ : trigger.kind === 'mention' ? 'Conversation mention' : 'Trigger';
+ await triggerStore.addRun(s, trigger.id, task.id, 'queued', s.now(), null, why.slice(0, 300));
+ }
if (afterInsert) await afterInsert(task);
if (afterInsert) await afterInsert(task);
if (chatContext) await s.db.query('INSERT INTO actor_chat_tasks (task_id, agent_id, policy, conversation) VALUES ($1, $2, $3, $4)', [task.id, chatContext.agentId, chatContext.policy, chatContext.conversation]);
if (chatContext) await s.db.query('INSERT INTO actor_chat_tasks (task_id, agent_id, policy, conversation) VALUES ($1, $2, $3, $4)', [task.id, chatContext.agentId, chatContext.policy, chatContext.conversation]);
return { status: 201, task };
return { status: 201, task };
@@ -531,6 +531,15 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -531,6 +531,15 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
: `the one agent loop (${row.mode === 'best' ? 'best' : 'smart'} tier)`;
: `the one agent loop (${row.mode === 'best' ? 'best' : 'smart'} tier)`;
explanation.push({ selected: agentId, objective: row.mode === 'best' ? 'correctness' : 'balanced', reasons: [why], candidates: [], decided_at: iso() });
explanation.push({ selected: agentId, objective: row.mode === 'best' ? 'correctness' : 'balanced', reasons: [why], candidates: [], decided_at: iso() });
await store.updateTask(s, id, { explanation });
await store.updateTask(s, id, { explanation });
+ // The plan event makes the assumptions form visible. Install its control first so an edit sent
+ // immediately after the page appears is queued even while the loop is still being prepared.
+ const control = { pending: [] };
+ control.assume = async (text) => {
+ control.pending.push(String(text == null ? '' : text).slice(0, 2000));
+ if (control.pending.length > 5) control.pending.shift();
+ return { ok: true, queued: true };
+ };
+ controls.set(id, control);
if (!resume && !fork) await emit(id, { kind: 'output', step: 'plan', agent: agentId, text: 'The agent starts with a plan, then works through it.' });
if (!resume && !fork) await emit(id, { kind: 'output', step: 'plan', agent: agentId, text: 'The agent starts with a plan, then works through it.' });
// A task that would start coding work on the person's machine asks through the same gate (levels.js):
// A task that would start coding work on the person's machine asks through the same gate (levels.js):
// the level first, then the always-ask floor. Full access or auto-edits may let an edit through; a
// the level first, then the always-ask floor. Full access or auto-edits may let an edit through; a
@@ -546,8 +555,6 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
@@ -546,8 +555,6 @@ function createEngine({ config, s, adapters, stream, webhooks = null, now = () =
const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);
const wakePayload = wakes.get(id) || (dayRun ? await day.wakePayload(id) : null);
if (wakePayload) wakes.delete(id);
if (wakePayload) wakes.delete(id);
let out = null;
let out = null;
- const control = {};
- controls.set(id, control);
// The files attached to this message in the chat (plan L55 C2): text as data; images described first.
// The files attached to this message in the chat (plan L55 C2): text as data; images described first.
const attachments = await inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', err && err.message); return []; });
const attachments = await inputs.forRun(s, id, { models, meter, signal, log }).catch((err) => { log.warn('[Actor] attachments:', err && err.message); return []; });
// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.
// The person's own model choice for this run (plan L55 C3): their own key, through OpenVibe.AI.
@@ -96,10 +96,10 @@ async function bumpCounters(s, id, { state = null, runs = 0, costUsd = 0, at = n
@@ -96,10 +96,10 @@ async function bumpCounters(s, id, { state = null, runs = 0, costUsd = 0, at = n
return wire(row);
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`,
- [triggerId, taskId, state, costUsd, at]);
+async function addRun(s, triggerId, taskId, state, at, costUsd = null, why = null) {
+ await s.db.query(`INSERT INTO actor_trigger_runs (trigger_id, task_id, state, cost_usd, why, created_at, updated_at)
+ VALUES ($1,$2,$3,$4,$5,$6,$6) 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`,
+ [triggerId, taskId, state, costUsd, why, at]);
}
}
const runForTask = (s, taskId) => s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [taskId]);
const runForTask = (s, taskId) => s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [taskId]);
const runs = (s, triggerId, limit = 20) => s.db.many('SELECT * FROM actor_trigger_runs WHERE trigger_id = $1 ORDER BY created_at DESC LIMIT $2', [triggerId, limit]);
const runs = (s, triggerId, limit = 20) => s.db.many('SELECT * FROM actor_trigger_runs WHERE trigger_id = $1 ORDER BY created_at DESC LIMIT $2', [triggerId, limit]);
@@ -97,10 +97,19 @@ const SAME = { origin: 'https://openvibe.actor' };
@@ -97,10 +97,19 @@ const SAME = { origin: 'https://openvibe.actor' };
const created = await t.get('/tasks', { as: bea, form: { task: 'Find three offers under ten euros', mode: 'balanced' }, headers: SAME });
const created = await t.get('/tasks', { as: bea, form: { task: 'Find three offers under ten euros', mode: 'balanced' }, headers: SAME });
const id = taskIdOf(created);
const id = taskIdOf(created);
await t.waitFor(id, ['running']);
await t.waitFor(id, ['running']);
+ // A running task may still be preparing its first plan. Wait for the visible plan event, rather than
+ // posting to the form while its section is hidden and the loop has no edit handler yet.
+ let planned = false;
+ for (const until = Date.now() + 60000; !planned && Date.now() < until;) {
+ planned = (await store.eventsAfter(s, id, 0)).some((e) => e.step === 'plan' && taskEvents.plan(e.text));
+ if (!planned) await new Promise((r) => setTimeout(r, 50));
+ }
+ assert.ok(planned, 'the live plan is visible before it is edited');
const page = await t.get(`/tasks/${id}`, { as: bea });
const page = await t.get(`/tasks/${id}`, { as: bea });
assert.match(page.text, new RegExp(`action="/tasks/${id}/assumptions"`), 'the live plan has an editable assumptions form');
assert.match(page.text, new RegExp(`action="/tasks/${id}/assumptions"`), 'the live plan has an editable assumptions form');
const edited = await t.get(`/tasks/${id}/assumptions`, { as: bea, form: { assumptions: 'Use EUR\nAsk before booking' }, headers: SAME });
const edited = await t.get(`/tasks/${id}/assumptions`, { as: bea, form: { assumptions: 'Use EUR\nAsk before booking' }, headers: SAME });
assert.strictEqual(edited.status, 303);
assert.strictEqual(edited.status, 303);
+ assert.ok(!String(edited.headers.get('location')).includes('assume=gone'), 'the visible plan accepts an immediate edit');
t.providers.state.runtime = 'tool';
t.providers.state.runtime = 'tool';
// Up to 30 s: the follow-up turn can wait behind the runtime queue on a loaded CI runner.
// Up to 30 s: the follow-up turn can wait behind the runtime queue on a loaded CI runner.
let seen = false;
let seen = false;
@@ -277,7 +277,11 @@ async function watchStub() {
@@ -277,7 +277,11 @@ async function watchStub() {
const root = await s.db.maybe('SELECT * FROM actor_assignments WHERE task_id = $1 AND depth = 0', [task.id]);
const root = await s.db.maybe('SELECT * FROM actor_assignments WHERE task_id = $1 AND depth = 0', [task.id]);
assert.deepStrictEqual([root.origin, root.effective_level], ['routine', 'ask']);
assert.deepStrictEqual([root.origin, root.effective_level], ['routine', 'ask']);
const run = await s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [task.id]);
const run = await s.db.maybe('SELECT * FROM actor_trigger_runs WHERE task_id = $1', [task.id]);
+ assert.match(run.why, /^Schedule:/);
assert.strictEqual(run.trigger_id, trigger.id, 'the run is linked back to its card');
assert.strictEqual(run.trigger_id, trigger.id, 'the run is linked back to its card');
+ const history = await call('GET', `/api/v1/triggers/${trigger.id}?runs=1`);
+ assert.strictEqual(history.status, 200, history.text);
+ assert.match(history.json().runs[0].why, /^Schedule:/);
});
});
} finally {
} finally {