openvibers/openvibe.actor Commits

Expose stored Actor work details (#403)

OP OpenVibers committed 3 days ago · c97e8481b1 · 8 files, +125 −9

* Expose stored Actor work details

* Return stored Actor evaluation verdict

* Allow deleting completed Actor work

* Use declared Actor fabric routes

* Retain completed Actor work for 30 days

* Merge remote-tracking branch 'origin/main' into feat/s6-stored-work

* Record successful Actor fabric reads

Changes

ddiff --git a/migrations/0064_actor_fabric_task_triggers.sql b/migrations/0064_actor_fabric_task_triggers.sql
nnew file mode 100644
iindex 0000000..4b35c7d
---- /dev/null
++++ b/migrations/0064_actor_fabric_task_triggers.sql
@@@ -0,0 +1,3 @@
++-- phase: expand
++-- Keep the exact source inputs assigned to each stored task so a successful report can mark only those inputs read.
++ALTER TABLE actor_fabric_tasks ADD COLUMN trigger_ids text[] NOT NULL DEFAULT '{}';
ddiff --git a/server/fabric/adapters.js b/server/fabric/adapters.js
iindex e14c505..0773bb1 100644
---- a/server/fabric/adapters.js
++++ b/server/fabric/adapters.js
@@@ -107,8 +107,8 @@ function createFabricAdapters({ s, config, fabric, models, ctx: taskCtx = null,
parentId,
// The binding commits with the task's own insert, so a run can never end without a report target.
afterInsert: async (task) => {
-- await s.db.query(`INSERT INTO actor_fabric_tasks (task_id, agent_id, owner, ticket_id, created_at)
-- VALUES ($1, $2, $3, $4, $5) ON CONFLICT (task_id) DO NOTHING`, [task.id, agentId, owner, ticket.id, s.iso()]);
++ await s.db.query(`INSERT INTO actor_fabric_tasks (task_id, agent_id, owner, ticket_id, created_at, trigger_ids)
++ VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (task_id) DO NOTHING`, [task.id, agentId, owner, ticket.id, s.iso(), [trigger.id]]);
},
});
if (!out || out.status !== 201 || !out.task) {
@@@ -259,7 +259,7 @@ function createFabricAdapters({ s, config, fabric, models, ctx: taskCtx = null,
const via = String(row.via || '');
if (!/^agent:agt_/.test(via)) return;
let agentId = via.slice(6), owner = row.requester, ticketId = null;
-- const link = await s.db.maybe('SELECT agent_id, owner, ticket_id FROM actor_fabric_tasks WHERE task_id = $1', [row.id]);
++ const link = await s.db.maybe('SELECT agent_id, owner, ticket_id, trigger_ids FROM actor_fabric_tasks WHERE task_id = $1', [row.id]);
if (link) { agentId = link.agent_id; owner = link.owner; ticketId = link.ticket_id; }
const found = await s.db.maybe('SELECT state FROM actor_work_fabric WHERE agent_id = $1 AND owner = $2', [agentId, owner]);
if (!found) return;
@@@ -273,7 +273,11 @@ function createFabricAdapters({ s, config, fabric, models, ctx: taskCtx = null,
const ticket = state.tickets[ticketId];
if (ticket.state !== 'open') return;
const report = reportFor(row, ticket);
-- await fabric.command(owner, agentId, 'report', { ticket_id: ticketId, report, end: true, spent_usd: Number(row.cost_usd) || 0, task_id: row.id });
++ const owned = new Set((state.meta[ticketId] && state.meta[ticketId].owned) || []);
++ const readTriggerIds = row.state === 'succeeded'
++ ? (Array.isArray(link && link.trigger_ids) ? link.trigger_ids : []).filter(id => owned.has(id))
++ : [];
++ await fabric.command(owner, agentId, 'report', { ticket_id: ticketId, report, end: true, spent_usd: Number(row.cost_usd) || 0, task_id: row.id, read_trigger_ids: readTriggerIds });
// One confirmation to the person on the Source thread, not a stream of progress.
const source = state.sources[ticket.mandate.actor.source];
if (source) {
ddiff --git a/server/fabric/api.js b/server/fabric/api.js
iindex a8ce03b..a18608e 100644
---- a/server/fabric/api.js
++++ b/server/fabric/api.js
@@@ -75,6 +75,35 @@ function createRoutes(ctx) {
r.get('/:agent/reports', run(async (req) => {
const snap = await ctx.fabric.snapshot(req.principal.requester, req.params.agent);
const state = snap.state;
++ if (req.query.detail === 'work') {
++ const owner = req.principal.requester, agent = req.params.agent;
++ const links = await ctx.s.db.many(`SELECT ft.ticket_id, t.id AS task_id, t.state AS task_state,
++ t.result, t.error_code, t.cost_usd, t.created_at, t.finished_at
++ FROM actor_fabric_tasks ft JOIN tasks t ON t.id = ft.task_id
++ WHERE ft.owner = $1 AND ft.agent_id = $2 ORDER BY t.created_at DESC LIMIT 200`, [owner, agent]);
++ const taskIds = links.map(x => x.task_id);
++ const files = taskIds.length ? await ctx.s.db.many(`SELECT task_id, id, name, mime, size, created_at
++ FROM actor_artifacts WHERE owner = $1 AND task_id = ANY($2::text[]) ORDER BY created_at`, [owner, taskIds]) : [];
++ const byTicket = new Map();
++ for (const row of links) {
++ const list = byTicket.get(row.ticket_id) || [];
++ list.push({ task_id: row.task_id, state: row.task_state, result: row.result,
++ error_code: row.error_code, cost_usd: Number(row.cost_usd) || 0,
++ created_at: row.created_at, finished_at: row.finished_at });
++ byTicket.set(row.ticket_id, list);
++ }
++ const tickets = Object.entries(state.tickets || {}).map(([id, ticket]) => {
++ const meta = (state.meta || {})[id] || {}, runs = byTicket.get(id) || [];
++ return { ticket_id: id, state: ticket.state, stage: ticket.stage, mandate: ticket.mandate,
++ evaluation: (state.decisions || []).filter(x => x.trigger_ids && x.trigger_ids.includes(ticket.trigger_id)),
++ reports: meta.reports || [], runs,
++ artifacts: files.filter(x => runs.some(run => run.task_id === x.task_id)).map(x => ({
++ task_id: x.task_id, id: x.id, name: x.name, mime: x.mime, size: Number(x.size),
++ url: `/api/v1/tasks/${encodeURIComponent(x.task_id)}/artifacts/${encodeURIComponent(x.id)}`, created_at: x.created_at,
++ })) };
++ });
++ return { agent: req.params.agent, tickets };
++ }
const tickets = Object.entries(state.tickets || {}).map(([id, ticket]) => ({
ticket_id: id,
state: ticket.state,
@@@ -86,7 +115,10 @@ function createRoutes(ctx) {
return { agent: req.params.agent, tickets, report_states: reportStates() };
}));
-- r.post('/:agent/commands/:command', run(req => ctx.fabric.command(req.principal.requester, req.params.agent, req.params.command, req.body)));
++ /** Durable work detail: ticket decision, final reports, spend and artifact references. Bytes stay behind the task artifact route. */
++ r.post('/:agent/commands/:command', run(req => req.params.command === 'deleteTicket'
++ ? ctx.fabric.deleteTicket(req.principal.requester, req.params.agent, req.body && req.body.ticket_id)
++ : ctx.fabric.command(req.principal.requester, req.params.agent, req.params.command, req.body)));
return r;
}
module.exports = { createRoutes };
ddiff --git a/server/fabric/engine.js b/server/fabric/engine.js
iindex 1b38a3d..8ed513a 100644
---- a/server/fabric/engine.js
++++ b/server/fabric/engine.js
@@@ -271,7 +271,7 @@ class Engine {
this.validate('agent.ticket', ticket);
this.s.tickets[ticket.id] = ticket;
for (const id of e.ids) this.s.accepted[id] = ticket.id;
-- this.s.meta[ticket.id] = { basis: e.basis, owned: [], handbacks: [], read: [], reports: [], tasks: [] };
++ this.s.meta[ticket.id] = { basis: e.basis, created_at: this.iso(), owned: [], handbacks: [], read: [], reports: [], tasks: [] };
for (const key of t.threads) this.thread(key).subscriptions[ticket.id] = 'origin';
this.enqueue('launch', `launch:${ticket.id}`, { ticket, triggers: e.ids.map(id => this.s.triggers[id]) }, `worker:${ticket.id}`);
}
@@@ -293,7 +293,7 @@ class Engine {
return k;
}
/** Record one agent.report@1 on the Ticket; a terminal report may end it. */
-- report({ ticket_id, report, end = false, spent_usd = 0, task_id = null }) {
++ report({ ticket_id, report, end = false, spent_usd = 0, task_id = null, read_trigger_ids = [] }) {
const k = own(this.s.tickets, ticket_id), m = own(this.s.meta, ticket_id);
assert(k && m && report && typeof report === 'object' && !Array.isArray(report), 'Report needs an open Ticket');
assert(k.state === 'open', 'Ticket has ended');
@@@ -306,6 +306,8 @@ class Engine {
this.validate('agent.report', r);
m.reports ||= [];
m.reports.push(r);
++ // A completed executor turn claims only inputs bound to this task, atomically with its report.
++ if (read_trigger_ids.length) this.read({ ticket_id, trigger_ids: read_trigger_ids });
m.spent_usd = round(Number(m.spent_usd || 0) + Number(spent_usd || 0));
if (REPORT_STAGES[r.state]) k.stage = REPORT_STAGES[r.state];
// The mandate's budget is a cap, not advice: a spent budget puts the lane that produced the Ticket back in
ddiff --git a/server/fabric/service.js b/server/fabric/service.js
iindex c0214fa..a631f79 100644
---- a/server/fabric/service.js
++++ b/server/fabric/service.js
@@@ -7,6 +7,7 @@ const COMMANDS = new Set(['configure', 'start', 'decide', 'subscribe', 'attach',
const round = (n) => Math.round((Number(n) || 0) * 1e6) / 1e6;
// The first promotion review needs a little history; a single run is not a basis for an estimate.
const ESTIMATE_RUNS = 3;
++const WORK_RETENTION_MS = 30 * 24 * 60 * 60 * 1000;
function createFabric({ s, enabled = false, readMail, validate = (name, value) => {
try { return contracts.assertValid(`${name}@1`, value); }
catch (err) { if (err.code === 'contract.invalid') err.status = 422; throw err; }
@@@ -61,10 +62,50 @@ function createFabric({ s, enabled = false, readMail, validate = (name, value) =
}
return {
enabled,
++ async deleteTicket(owner, agent, ticketId) {
++ enabledCheck();
++ return s.db.tx(async db => {
++ const exists = await db.maybe('SELECT id FROM actor_agents WHERE id = $1 AND owner = $2 FOR UPDATE', [agent, owner]);
++ if (!exists) throw Object.assign(new Error('Agent not found'), { status: 404 });
++ const row = await db.maybe('SELECT state FROM actor_work_fabric WHERE agent_id = $1 AND owner = $2 FOR UPDATE', [agent, owner]);
++ if (!row) throw Object.assign(new Error('Fabric not found'), { status: 404 });
++ const state = typeof row.state === 'string' ? JSON.parse(row.state) : row.state;
++ const ticket = state.tickets && state.tickets[ticketId];
++ if (!ticket) throw Object.assign(new Error('Ticket not found'), { status: 404 });
++ if (ticket.state !== 'ended') throw Object.assign(new Error('Only ended work can be deleted'), { status: 409 });
++ const triggerIds = new Set([ticket.trigger_id]);
++ const tasks = await db.many('SELECT task_id FROM actor_fabric_tasks WHERE owner = $1 AND agent_id = $2 AND ticket_id = $3', [owner, agent, ticketId]);
++ for (const task of tasks) await db.query('DELETE FROM tasks WHERE id = $1 AND requester = $2', [task.task_id, owner]);
++ delete state.tickets[ticketId]; delete state.meta[ticketId];
++ state.work = (state.work || []).filter(w => w.conversation !== `worker:${ticketId}` && w.payload?.ticket_id !== ticketId);
++ for (const thread of Object.values(state.threads || {})) {
++ delete thread.subscriptions?.[ticketId];
++ thread.pending = (thread.pending || []).filter(id => !triggerIds.has(id));
++ thread.triggers = (thread.triggers || []).filter(id => !triggerIds.has(id));
++ if (thread.evaluation) thread.evaluation.ids = (thread.evaluation.ids || []).filter(id => !triggerIds.has(id));
++ }
++ for (const id of triggerIds) {
++ const stillUsed = Object.values(state.tickets).some(t => t.trigger_id === id);
++ if (stillUsed) continue;
++ delete state.triggers[id]; delete state.accepted[id];
++ state.decisions = (state.decisions || []).filter(d => !(d.trigger_ids || []).includes(id));
++ }
++ const decisionIds = new Set((state.decisions || []).map(d => d.id));
++ for (const thread of Object.values(state.threads || {})) thread.history = (thread.history || []).filter(id => decisionIds.has(id));
++ await db.query('UPDATE actor_work_fabric SET state = $3::jsonb, revision = revision + 1, updated_at = $4 WHERE agent_id = $1 AND owner = $2', [agent, owner, JSON.stringify(state), s.iso()]);
++ return { deleted: true, ticket_id: ticketId, tasks: tasks.length };
++ });
++ },
async snapshot(owner, agent) {
enabledCheck();
-- const row = await s.db.maybe('SELECT f.state, f.revision FROM actor_work_fabric f JOIN actor_agents a ON a.id = f.agent_id WHERE f.agent_id = $1 AND f.owner = $2 AND a.owner = $2', [agent, owner]);
++ let row = await s.db.maybe('SELECT f.state, f.revision FROM actor_work_fabric f JOIN actor_agents a ON a.id = f.agent_id WHERE f.agent_id = $1 AND f.owner = $2 AND a.owner = $2', [agent, owner]);
if (!row) throw Object.assign(new Error('Fabric not found'), { status: 404 });
++ const state = typeof row.state === 'string' ? JSON.parse(row.state) : row.state;
++ const cutoff = Date.now() - WORK_RETENTION_MS;
++ const expired = Object.entries(state.tickets || {}).filter(([id, ticket]) => ticket.state === 'ended'
++ && Date.parse(state.meta?.[id]?.created_at || '') < cutoff).map(([id]) => id);
++ for (const id of expired) await this.deleteTicket(owner, agent, id);
++ if (expired.length) row = await s.db.maybe('SELECT f.state, f.revision FROM actor_work_fabric f JOIN actor_agents a ON a.id = f.agent_id WHERE f.agent_id = $1 AND f.owner = $2 AND a.owner = $2', [agent, owner]);
return row;
},
async command(owner, agent, name, args = {}) {
ddiff --git a/test/fabric-adapters.test.js b/test/fabric-adapters.test.js
iindex 3f366df..a69135a 100644
---- a/test/fabric-adapters.test.js
++++ b/test/fabric-adapters.test.js
@@@ -100,12 +100,16 @@ function test(name, fn) { return Promise.resolve(fn()).then(() => { count++; con
await test('task end: ready becomes an agent.report@1 on the Ticket, ends it and queues one reply', async () => {
const state = fabricState({ ticket: ticket() }); state.meta.ticket_1.task_id = 'tsk_1';
++ state.meta.ticket_1.owned = ['follow_1'];
const fabric = fakeFabric(state);
-- const s = { db: { maybe: async () => ({ state }) } };
++ const s = { db: { maybe: async (sql) => sql.includes('actor_fabric_tasks')
++ ? { agent_id: AGENT, owner: OWNER, ticket_id: 'ticket_1', trigger_ids: ['follow_1'] }
++ : { state } } };
const a = createFabricAdapters({ s, config: {}, fabric, models: {} });
await a.onTaskState({ id: 'tsk_1', via: `agent:${AGENT}`, requester: OWNER, state: 'succeeded', assignment_id: 'asg_1', result: { answer: 'The test is fixed.' } });
const report = fabric.commands.find((c) => c.name === 'report');
assert.equal(report.args.report.state, 'ready'); assert.equal(report.args.report.summary, 'The test is fixed.'); assert.equal(report.args.end, true);
++ assert.deepEqual(report.args.read_trigger_ids, ['follow_1']);
const reply = fabric.commands.find((c) => c.name === 'reply');
assert.match(reply.args.text, /The test is fixed/); assert.equal(reply.args.key, 'report:tsk_1');
});
ddiff --git a/test/fabric-observe-store.test.js b/test/fabric-observe-store.test.js
iindex 85ec689..ff62580 100644
---- a/test/fabric-observe-store.test.js
++++ b/test/fabric-observe-store.test.js
@@@ -82,6 +82,7 @@ const policy = () => ({
const out = await fabric.command(owner, agent, 'decide', { evaluation: state.threads[thread].evaluation.id, decision: 'accept', assessment });
assert.equal(out.decision.decision, 'accept'); assert.ok(out.ticket);
first = out.ticket;
++ assert.ok(Date.parse((await fabric.snapshot(owner, agent)).state.meta[first.id].created_at));
const after = await fabric.observe(owner, agent, { lane: 'fix' });
assert.equal(after[0].mode, 'auto'); assert.equal(after[0].decision, 'accept');
});
@@@ -107,9 +108,31 @@ const policy = () => ({
assert.equal((await t.get(`/api/v1/fabric/${agent}/commands/demote`, { as: alice, json: { lane: 'fix', reason: 'Reading a few more observe records first.' }, headers: { Origin: 'https://openvibe.actor' } })).status, 200);
});
++ await check('stored work API returns the durable mandate, evaluation, reports and run/artifact references', async () => {
++ const res = await t.get(`/api/v1/fabric/${agent}/reports?detail=work`, { as: alice });
++ assert.equal(res.status, 200);
++ const body = res.json();
++ assert.equal(body.tickets.length, 1);
++ assert.equal(body.tickets[0].ticket_id, first.id);
++ assert.equal(body.tickets[0].mandate.scope, first.mandate.scope);
++ assert.equal(body.tickets[0].evaluation[0].decision, 'accept');
++ assert.equal(body.tickets[0].reports[0].state, 'progress');
++ assert.ok(Array.isArray(body.tickets[0].runs));
++ assert.ok(Array.isArray(body.tickets[0].artifacts));
++ });
++
await check('owner-only: another signed-in person gets 404 from every new read', async () => {
for (const path of ['lanes', 'observe', 'reports']) assert.equal((await t.get(`/api/v1/fabric/${agent}/${path}`, { as: bob })).status, 404);
});
++ await check('the owner can delete ended stored work and active tickets are protected', async () => {
++ const active = await t.get(`/api/v1/fabric/${agent}/commands/deleteTicket`, { as: alice, json: { ticket_id: first.id }, headers: { Origin: 'https://openvibe.actor' } });
++ assert.equal(active.status, 409);
++ await t.get(`/api/v1/fabric/${agent}/commands/end`, { as: alice, json: { ticket_id: first.id, reason: 'owner_finished' }, headers: { Origin: 'https://openvibe.actor' } });
++ const deleted = await t.get(`/api/v1/fabric/${agent}/commands/deleteTicket`, { as: alice, json: { ticket_id: first.id }, headers: { Origin: 'https://openvibe.actor' } });
++ assert.equal(deleted.status, 200);
++ assert.equal(deleted.json().deleted, true);
++ assert.equal((await t.get(`/api/v1/fabric/${agent}/reports?detail=work`, { as: alice })).json().tickets.length, 0);
++ });
} finally { await t.close(); }
done();
})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/fabric.test.js b/test/fabric.test.js
iindex 36d95fe..6fc5c31 100644
---- a/test/fabric.test.js
++++ b/test/fabric.test.js
@@@ -127,6 +127,13 @@ test('002 Release unread: read input stays accounted for; unread returns to rout
const e = engine(), k = accept(e); e.receive(trigger('read')); e.receive(trigger('unread'));
e.read({ ticket_id: k.id, trigger_ids: ['read'] }); e.end({ ticket_id: k.id }); assert.deepEqual(e.s.threads.room.pending, ['unread']);
});
++test('002 A successful report atomically claims only inputs assigned to its run', () => {
++ const e = engine(), k = accept(e); e.receive(trigger('read')); e.receive(trigger('unread'));
++ e.report({ ticket_id: k.id, task_id: 'tsk_1', read_trigger_ids: ['read'], end: true,
++ report: { ticket_id: k.id, assignment_id: 'asg_1', state: 'ready', summary: 'Done', pr: null, tests_run: [], risks: [] } });
++ assert(e.s.meta[k.id].read.includes('read'));
++ assert.deepEqual(e.s.threads.room.pending, ['unread']);
++});
test('002 Eval timeout: undelivered never expires; third delivered expiry is unavailable', () => {
const e = engine(); e.receive(trigger('a')); let ev = e.start({ thread: 'room' }); e.now = 9999999; e.expire(); assert(e.s.threads.room.evaluation);
for (let i = 0; i < 3; i++) { e.delivered({ evaluation: ev.id }); e.now += 300001; e.expire(); if (i < 2) ev = e.start({ thread: 'room' }); }