openvibers/openvibe.watch Commits
A routine that is over stops its own subscription (#234)
Changes
@@@ -37,7 +37,10 @@* else, including an allowance spent for the day, is retried on the backoff. It needs a person as* owner: an agent- or service-owned watch is `skipped`.* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.-- * A missing agent or allowance refusal fails this firing; a service error retries.++ * A missing agent or allowance refusal fails this firing; a service error retries. When Actor++ * answers 409 actor.trigger.paused or actor.trigger.done, the routine is over (a goal reached its++ * check, or its runs, money or deadline ran out): the watch pauses itself and the action is++ * `skipped` — only the person's own token may pause a watch for them, and Watch does not have it.* run-job one OpenVibe.Run job for the owner in the watch's project (server/run-jobs.js): the action's* run.job-create-request@1 fields, with the firing under args[firing_arg] when it names one. Watch* submits as itself, the owner in X-OV-Subject and the project in X-OV-Project (project delegation,@@@ -229,6 +232,9 @@ function createActions({ db, config, notifications = null, post = null, actor =COUNT(*) FILTER (WHERE state = 'sent') AS sent FROM action_runs`),recent: db.prepare(`SELECT check_run_id, action_index, kind, state, attempts, last_error, request, created_at, finished_at FROM action_runsWHERE watch_id = ? ORDER BY rid DESC LIMIT ?`),++ // A routine that is over stops its own subscription: Actor answered the delivery with the code that says++ // no run will happen again (a person's token, not Watch's, is what Actor may pause a watch with).++ endRoutine: db.prepare("UPDATE watches SET status = 'paused', next_due_at = 0, updated_at = ? WHERE id = ? AND status = 'active' AND deleted_at IS NULL"),};/** One row per action of the watch, for the run that fired. Call inside the run's transaction. */@@@ -357,6 +363,13 @@ function createActions({ db, config, notifications = null, post = null, actor =});return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };} catch (err) {++ // The routine is over (a goal reached its check, or Actor paused it): nothing will run again, so this++ // subscription stops checking. Actor cannot pause a watch for the person (only the person's own token++ // reaches Watch), so Watch stops itself on the first delivery it refuses for that reason.++ if (err && err.status === 409 && /^actor\.trigger\.(paused|done)$/.test(String(err.code || ''))) {++ await st.endRoutine.run(now(), row.watch_id);++ return { state: 'skipped', error: `watch.action_routine_over: ${err.code}` };++ }if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };if (err && (err.status === 429 || err.status === 400 || err.status === 422 || (err.status === 403 && /(?:limit|free.tier|over_tier)/i.test(String(err.code || ''))))) {return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };@@@ -454,6 +454,33 @@ t('actor-agent sends note and bounded firing context, records its task, and clasactorMode = 'ok';});++t('a routine that is over stops its own subscription, without a person\'s token', async () => {++ // Actor answers a delivery for a trigger that ended with actor.trigger.paused / actor.trigger.done. Only the++ // person's own token may pause a watch for them (Actor#325's rule), so Watch stops itself on that answer.++ const w = await create(person, { name: 'a goal that finished', action: [{ kind: 'actor-agent', agent_id: agentId }] });++ const over = async (code) => { throw Object.assign(new Error('refused'), { status: 409, code }); };++ const original = actor.runAgent;++ try {++ actor.runAgent = over.bind(null, 'actor.trigger.done');++ await fire(w);++ const r = await svc.actions.tick();++ const row = (await acts(w.id))[0];++ assert.deepStrictEqual([row.state, r.skipped], ['skipped', 1], 'not a failure, and not retried');++ assert.match(row.last_error, /watch\.action_routine_over: actor\.trigger\.done/);++ const after = await svc.registry.get(w.id, person);++ assert.strictEqual(after.status, 'paused', 'the subscription stopped checking');++++ const paused = await create(person, { name: 'a routine paused by its person', action: [{ kind: 'actor-agent', agent_id: agentId }] });++ actor.runAgent = over.bind(null, 'actor.trigger.paused');++ await fire(paused);++ await svc.actions.tick();++ assert.strictEqual((await svc.registry.get(paused.id, person)).status, 'paused');++ assert.strictEqual((await acts(paused.id))[0].state, 'skipped');++ } finally {++ actor.runAgent = original;++ }++});++t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {const st = await svc.actions.status();assert.strictEqual(st.enabled, true);
@@ -37,7 +37,10 @@
@@ -37,7 +37,10 @@
* else, including an allowance spent for the day, is retried on the backoff. It needs a person as
* else, including an allowance spent for the day, is retried on the backoff. It needs a person as
* owner: an agent- or service-owned watch is `skipped`.
* owner: an agent- or service-owned watch is `skipped`.
* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.
* actor-agent runs one of the owner's personal agents on Actor with a note and bounded firing context.
- * A missing agent or allowance refusal fails this firing; a service error retries.
+ * A missing agent or allowance refusal fails this firing; a service error retries. When Actor
+ * answers 409 actor.trigger.paused or actor.trigger.done, the routine is over (a goal reached its
+ * check, or its runs, money or deadline ran out): the watch pauses itself and the action is
+ * `skipped` — only the person's own token may pause a watch for them, and Watch does not have it.
* run-job one OpenVibe.Run job for the owner in the watch's project (server/run-jobs.js): the action's
* run-job one OpenVibe.Run job for the owner in the watch's project (server/run-jobs.js): the action's
* run.job-create-request@1 fields, with the firing under args[firing_arg] when it names one. Watch
* run.job-create-request@1 fields, with the firing under args[firing_arg] when it names one. Watch
* submits as itself, the owner in X-OV-Subject and the project in X-OV-Project (project delegation,
* submits as itself, the owner in X-OV-Subject and the project in X-OV-Project (project delegation,
@@ -229,6 +232,9 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -229,6 +232,9 @@ function createActions({ db, config, notifications = null, post = null, actor =
COUNT(*) FILTER (WHERE state = 'sent') AS sent FROM action_runs`),
COUNT(*) FILTER (WHERE state = 'sent') AS sent FROM action_runs`),
recent: db.prepare(`SELECT check_run_id, action_index, kind, state, attempts, last_error, request, created_at, finished_at FROM action_runs
recent: db.prepare(`SELECT check_run_id, action_index, kind, state, attempts, last_error, request, created_at, finished_at FROM action_runs
WHERE watch_id = ? ORDER BY rid DESC LIMIT ?`),
WHERE watch_id = ? ORDER BY rid DESC LIMIT ?`),
+ // A routine that is over stops its own subscription: Actor answered the delivery with the code that says
+ // no run will happen again (a person's token, not Watch's, is what Actor may pause a watch with).
+ endRoutine: db.prepare("UPDATE watches SET status = 'paused', next_due_at = 0, updated_at = ? WHERE id = ? AND status = 'active' AND deleted_at IS NULL"),
};
};
/** One row per action of the watch, for the run that fired. Call inside the run's transaction. */
/** One row per action of the watch, for the run that fired. Call inside the run's transaction. */
@@ -357,6 +363,13 @@ function createActions({ db, config, notifications = null, post = null, actor =
@@ -357,6 +363,13 @@ function createActions({ db, config, notifications = null, post = null, actor =
});
});
return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };
return { state: 'sent', error: `Actor task ${task.id || '(no id)'}` };
} catch (err) {
} catch (err) {
+ // The routine is over (a goal reached its check, or Actor paused it): nothing will run again, so this
+ // subscription stops checking. Actor cannot pause a watch for the person (only the person's own token
+ // reaches Watch), so Watch stops itself on the first delivery it refuses for that reason.
+ if (err && err.status === 409 && /^actor\.trigger\.(paused|done)$/.test(String(err.code || ''))) {
+ await st.endRoutine.run(now(), row.watch_id);
+ return { state: 'skipped', error: `watch.action_routine_over: ${err.code}` };
+ }
if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };
if (err && err.status === 404 && err.code === 'actor.agent.not_found') return { state: 'failed', error: `watch.action_refused: ${err.code}: this agent was not found for the watch owner` };
if (err && (err.status === 429 || err.status === 400 || err.status === 422 || (err.status === 403 && /(?:limit|free.tier|over_tier)/i.test(String(err.code || ''))))) {
if (err && (err.status === 429 || err.status === 400 || err.status === 422 || (err.status === 403 && /(?:limit|free.tier|over_tier)/i.test(String(err.code || ''))))) {
return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };
return { state: 'failed', error: `watch.action_refused: ${err.code || err.status}: ${String(err.detail || err.message || '').slice(0, 200)}` };
@@ -454,6 +454,33 @@ t('actor-agent sends note and bounded firing context, records its task, and clas
@@ -454,6 +454,33 @@ t('actor-agent sends note and bounded firing context, records its task, and clas
actorMode = 'ok';
actorMode = 'ok';
});
});
+t('a routine that is over stops its own subscription, without a person\'s token', async () => {
+ // Actor answers a delivery for a trigger that ended with actor.trigger.paused / actor.trigger.done. Only the
+ // person's own token may pause a watch for them (Actor#325's rule), so Watch stops itself on that answer.
+ const w = await create(person, { name: 'a goal that finished', action: [{ kind: 'actor-agent', agent_id: agentId }] });
+ const over = async (code) => { throw Object.assign(new Error('refused'), { status: 409, code }); };
+ const original = actor.runAgent;
+ try {
+ actor.runAgent = over.bind(null, 'actor.trigger.done');
+ await fire(w);
+ const r = await svc.actions.tick();
+ const row = (await acts(w.id))[0];
+ assert.deepStrictEqual([row.state, r.skipped], ['skipped', 1], 'not a failure, and not retried');
+ assert.match(row.last_error, /watch\.action_routine_over: actor\.trigger\.done/);
+ const after = await svc.registry.get(w.id, person);
+ assert.strictEqual(after.status, 'paused', 'the subscription stopped checking');
+
+ const paused = await create(person, { name: 'a routine paused by its person', action: [{ kind: 'actor-agent', agent_id: agentId }] });
+ actor.runAgent = over.bind(null, 'actor.trigger.paused');
+ await fire(paused);
+ await svc.actions.tick();
+ assert.strictEqual((await svc.registry.get(paused.id, person)).status, 'paused');
+ assert.strictEqual((await acts(paused.id))[0].state, 'skipped');
+ } finally {
+ actor.runAgent = original;
+ }
+});
+
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
t('readiness counts the queue, and the watch page says what the watch did when it fired', async () => {
const st = await svc.actions.status();
const st = await svc.actions.status();
assert.strictEqual(st.enabled, true);
assert.strictEqual(st.enabled, true);