openvibers/openvibe.watch Commits

A6: Watch publishes the agent's step (agent.step.started|updated|ended) (#224)

OP OpenVibers committed 6 days ago · 2452efc8e3 · 6 files, +183 −3

One watch is one step of the agent's timeline: a title, its state and its page. Each
check brings the same entry forward instead of adding a row per check, a pause or a
delete ends it, and a check's failure carries its one-line reason. Never the page,
the condition's value or the watch's source payload.

Changes

ddiff --git a/server/check.js b/server/check.js
iindex 18b500f..a839caa 100644
---- a/server/check.js
++++ b/server/check.js
@@@ -28,6 +28,7 @@ const { ids: contractIds, validate } = require('openvibe-contracts');
const { nextDue } = require('./registry');
const { ExtractError } = require('./extract');
const { envelope } = require('./events/envelope');
++const { emitAgentStep } = require('./events/agent-step');
/** States that count as a failure, back off, and emit watch.check.failed. */
const FAILURE_STATES = new Set(['http_error', 'timeout', 'parse_error', 'rate_limited', 'budget_exceeded', 'skipped', 'disabled']);
@@@ -177,6 +178,19 @@ function createCheck({ db, registry, config, carriers, extract, condition, obser
failures, next_due: armed, not_before: notBefore,
});
for (const e of events || []) await outbox.emit(envelope(e, config.serviceId, now));
++ // The agent's step (design 2026-11-07): the watch is one step, so a check brings it forward rather than
++ // adding an entry per check. A failed check shows failed (with the one-line reason); the next good one
++ // is waiting again.
++ if (watch.status === 'active') {
++ const failed = FAILURE_STATES.has(run.state);
++ await emitAgentStep(outbox, watch, {
++ topic: 'updated', status: failed ? 'failed' : 'waiting',
++ startedAt: run.started_at, endedAt: failed ? run.finished_at : null,
++ reason: failed ? (run.detail || run.error_code || 'the check did not complete') : null,
++ costUsd: typeof run.cost_usd === 'number' ? run.cost_usd : null,
++ baseUrl: config.baseUrl, now,
++ });
++ }
// A run that fired queues the watch's actions in this same commit (server/actions.js): once per trigger.
if (fired && actions) await actions.queue({ watch, run, ...fired });
});
ddiff --git a/server/events/agent-step.js b/server/events/agent-step.js
nnew file mode 100644
iindex 0000000..a322e24
---- /dev/null
++++ b/server/events/agent-step.js
@@@ -0,0 +1,66 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): OpenVibe.Watch publishes a watch as one step of the agent's timeline —
++ * agent.step.started when it is waiting for its next check, agent.step.updated when a check settles, and
++ * agent.step.ended when the watch stops. The payload is a title, an outcome and a link: never the page's body,
++ * the condition's value or the watch's own source payload.
++ *
++ * One watch is one step (step_id is the watch's wch_ id), so a watch that checks every few minutes does not fill
++ * the Dock with one entry per check; each check brings the same entry forward.
++ */
++const { envelope } = require('./envelope');
++
++const AGENT_TOPICS = Object.freeze({
++ started: 'agent.step.started',
++ updated: 'agent.step.updated',
++ ended: 'agent.step.ended',
++});
++
++const PERSON_OWNER = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const STATUS_OPEN = 'waiting';
++const WHERE = 'On the network';
++const WHY = 'it checks the page from our side, on your schedule, and only tells you when it changes';
++
++/** One watch → the agent's step: its name, its state and its page. Never the page, the value or the condition. */
++function agentStepOf(watch, { status = STATUS_OPEN, startedAt = null, endedAt = null, reason = null, costUsd = null, baseUrl = 'https://openvibe.watch', transition = null } = {}) {
++ const id = watch.id;
++ const base = String(baseUrl || 'https://openvibe.watch').replace(/\/+$/, '');
++ return {
++ step_id: id,
++ mode: 'watch',
++ product: 'watch',
++ ref: id,
++ title: `Watching ${watch.name || 'a page'}`,
++ status,
++ started_at: new Date(startedAt == null ? Date.now() : startedAt).toISOString(),
++ ended_at: status === 'waiting' ? null : (endedAt == null ? null : new Date(endedAt).toISOString()),
++ placement: { where: WHERE, why: WHY },
++ cost_usd: typeof costUsd === 'number' && Number.isFinite(costUsd) && costUsd >= 0 ? costUsd : null,
++ link: `${base}/watches/${encodeURIComponent(id)}`,
++ reason: status === 'failed' || status === 'stopped' ? (reason == null ? null : String(reason).slice(0, 240)) : null,
++ // The orchestrating mode's sentence when it hands over to the watch ("Waiting for CI on #167 (Watch, 4 min)…"),
++ // when it gives us one; a check's own update keeps it out of the sentence.
++ transition: transition == null ? null : String(transition).slice(0, 240),
++ };
++}
++
++/**
++ * Write the watch's agent step, in the caller's transaction (the same one as the watch change). A watch owned by
++ * an agent (agt_…) or a service has no person to show it to and publishes nothing.
++ */
++async function emitAgentStep(outbox, watch, { topic, now = () => Date.now(), ...rest } = {}) {
++ const owner = watch && (watch.owner_sub || watch.owner);
++ if (!outbox || !watch || !PERSON_OWNER.test(String(owner || ''))) return null;
++ const event_type = AGENT_TOPICS[topic];
++ if (!event_type) return null;
++ return await outbox.emit(envelope({
++ event_type,
++ actor: { type: 'user', id: owner },
++ subject: { type: 'user', id: owner },
++ visibility: 'internal',
++ priority: topic === 'ended' ? 'important' : 'low',
++ payload: agentStepOf(watch, rest),
++ }, 'watch', now));
++}
++
++module.exports = { AGENT_TOPICS, agentStepOf, emitAgentStep, PERSON_OWNER, STATUS_OPEN };
ddiff --git a/server/index.js b/server/index.js
iindex 012a464..f142ffd 100644
---- a/server/index.js
++++ b/server/index.js
@@@ -56,7 +56,8 @@ async function start({ config, env = process.env, db: givenDb = null, now = () =
intervalMs: config.events.relayIntervalMs, fetch: fetchImpl, log, now, table: 'service_outbox',
eventTypes: ['watch.watch.created', 'watch.watch.updated', 'watch.watch.paused', 'watch.watch.removed',
'watch.observation.recorded', 'watch.watch.triggered', 'watch.check.failed',
-- 'watch.rate.changed', 'watch.rate.approved', 'watch.rate.rejected'],
++ 'watch.rate.changed', 'watch.rate.approved', 'watch.rate.rejected',
++ 'agent.step.started', 'agent.step.updated', 'agent.step.ended'],
validate: (env) => validate('events.event-envelope@1', env),
});
// Actions of fired watches (plan T18 step 4): notifications go to the owner through Network with Watch's own
@@@ -71,7 +72,7 @@ async function start({ config, env = process.env, db: givenDb = null, now = () =
// A watch that names a project: Network says whether its owner is an admin or owner of it, when it is saved (Run checks
// again before every job). Without the client secret no project can be checked, so none can be named.
const projects = givenProjects || (client ? createProjects(client, { networkUrl: config.networkInternalUrl }) : null);
-- const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants, projects });
++ const registry = createRegistry({ db, now, outbox, source: config.serviceId, maxWatchesPerOwner: config.watches.maxPerOwner, secretGrants: config.secretGrants, projects, baseUrl: config.baseUrl });
const observations = createObservations({ db, now });
// The only way out to the internet: SSRF guard (ports, private hosts, every redirect hop), byte
// cap, one deadline, conditional GET; per-host spacing in front of every request.
ddiff --git a/server/registry.js b/server/registry.js
iindex 2bfecd3..ac73f09 100644
---- a/server/registry.js
++++ b/server/registry.js
@@@ -27,6 +27,7 @@
const { validate, ids: contractIds } = require('openvibe-contracts');
const safe = require('./safe-regex');
const { envelope } = require('./events/envelope');
++const { emitAgentStep } = require('./events/agent-step');
/** Source kinds that reach out on a cadence; event and webhook sources are pushed to instead. */
const POLLING = new Set(['http', 'feed', 'api', 'node', 'run']);
@@@ -163,7 +164,7 @@ function validateView(view) {
return view;
}
--function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null, projects = null }) {
++function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox = null, source = 'watch', maxWatchesPerOwner = 50, secretGrants = null, projects = null, baseUrl = 'https://openvibe.watch' }) {
/** A named secret the owner may not use is refused when the watch is saved (server/secret-grants.js). */
function checkSecrets(owner, rec) {
if (!secretGrants) return;
@@@ -377,6 +378,11 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
await db.tx(async () => {
await st.insert.run(row);
if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.created', row, owner, { name: rec.name, status: rec.status }), source, now));
++ // The agent's step (design 2026-11-07): an active watch starts as one waiting step; a watch made paused
++ // has not started, and starts when it is first resumed.
++ if (rec.status === 'active') {
++ await emitAgentStep(outbox, { id: row.id, name: rec.name, owner_sub: owner }, { topic: 'started', status: 'waiting', startedAt: t, baseUrl, now });
++ }
});
if (touchesEvents(rec.source)) eventSourceChanged();
return await st.get.get(row.id);
@@@ -429,6 +435,13 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
if (outbox && touched.length) {
await outbox.emit(envelope(watchEvent(type, row, row.owner_sub, { name: rec.name, status: rec.status, fields: touched.sort() }), source, now));
}
++ // A watch that starts waiting is a step that starts; one that is paused or disabled stops. Another edit
++ // of a waiting watch is the same step, not a second one.
++ if (outbox && changes.status && changes.status !== row.status) {
++ const step = { id: row.id, name: rec.name, owner_sub: row.owner_sub };
++ if (rec.status === 'active') await emitAgentStep(outbox, step, { topic: 'started', status: 'waiting', startedAt: args.updated_at, baseUrl, now });
++ else if (row.status === 'active') await emitAgentStep(outbox, step, { topic: 'ended', status: 'stopped', startedAt: row.created_at, endedAt: args.updated_at, reason: `the watch is ${rec.status}`, baseUrl, now });
++ }
});
if (touchesEvents(current.source, rec.source)) eventSourceChanged();
return await st.get.get(row.id);
@@@ -442,6 +455,7 @@ function createRegistry({ db, now = () => Date.now(), ids = contractIds, outbox
await db.tx(async () => {
await db.prepare('UPDATE watches SET deleted_at = ?, updated_at = ?, next_due_at = 0 WHERE id = ?').run(t, t, row.id);
if (outbox) await outbox.emit(envelope(watchEvent('watch.watch.removed', row, row.owner_sub, { name: row.name, status: row.status }), source, now));
++ if (row.status === 'active') await emitAgentStep(outbox, row, { topic: 'ended', status: 'stopped', startedAt: row.created_at, endedAt: t, reason: 'the watch was deleted', baseUrl, now });
});
if (touchesEvents(fromRow(row).source)) eventSourceChanged();
return { id: row.id, deleted: true };
ddiff --git a/test/agent-step.test.js b/test/agent-step.test.js
nnew file mode 100644
iindex 0000000..2ef1105
---- /dev/null
++++ b/test/agent-step.test.js
@@@ -0,0 +1,67 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): a watch is one step of the agent's timeline — waiting for its next check,
++ * failed with the one-line reason, or stopped when the watch is paused or deleted. The payload is a title, an
++ * outcome and a link: never the page's body, the value it extracted or the condition it compares.
++ *
++ * node test/agent-step.test.js
++ */
++const assert = require('assert');
++const { AGENT_TOPICS, agentStepOf, emitAgentStep } = require('../server/events/agent-step');
++const FIXTURE = require('./fixtures/watch-agent-step.json');
++
++const WATCH = {
++ id: 'wch_01JAB2C3D4E5F6G7H8J9K0MNPS', name: 'Shoe price', owner_sub: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS', status: 'active',
++ source: { kind: 'page', url: 'https://shop.example/private-page' },
++ condition: { kind: 'changed' }, extraction: { kind: 'text', selector: '#price' },
++};
++
++assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');
++assert.strictEqual(AGENT_TOPICS.updated, 'agent.step.updated');
++assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');
++
++const waiting = agentStepOf(WATCH, { status: 'waiting', startedAt: 1762000000000, baseUrl: 'https://openvibe.watch' });
++assert.strictEqual(waiting.step_id, WATCH.id);
++assert.strictEqual(waiting.mode, 'watch');
++assert.strictEqual(waiting.product, 'watch');
++assert.strictEqual(waiting.status, 'waiting');
++assert.strictEqual(waiting.title, 'Watching Shoe price');
++assert.strictEqual(waiting.link, `https://openvibe.watch/watches/${WATCH.id}`);
++assert.strictEqual(waiting.ended_at, null, 'a waiting watch has not ended');
++assert.ok(waiting.placement && waiting.placement.where && waiting.placement.why);
++const raw = JSON.stringify(waiting);
++assert.ok(!raw.includes('shop.example/private-page'), 'the page never travels');
++assert.ok(!raw.includes('private-page'), 'not even a piece of it');
++assert.ok(!raw.includes('"changed"'), 'the condition never travels');
++assert.ok(!('source' in waiting) && !('condition' in waiting) && !('extraction' in waiting), 'no field for the page or the condition');
++
++const failed = agentStepOf(WATCH, { status: 'failed', startedAt: 1762000000000, endedAt: 1762000005000, reason: 'the page answered 500', costUsd: 0.001 });
++assert.strictEqual(failed.status, 'failed');
++assert.strictEqual(failed.reason, 'the page answered 500');
++assert.ok(failed.ended_at, 'a failed check says when it ended');
++assert.strictEqual(failed.cost_usd, 0.001);
++
++const stopped = agentStepOf(WATCH, { status: 'stopped', startedAt: 1762000000000, endedAt: 1762000010000, reason: 'the watch is paused' });
++assert.strictEqual(stopped.status, 'stopped');
++assert.strictEqual(stopped.reason, 'the watch is paused');
++
++// The Act step that waits on CI hands over to Watch: this fixture is the payload Agent's timeline test replays.
++assert.deepStrictEqual(agentStepOf({ id: FIXTURE.step_id, name: '#167 CI', owner_sub: WATCH.owner_sub }, {
++ status: 'waiting', startedAt: Date.parse(FIXTURE.started_at), transition: FIXTURE.transition,
++}), FIXTURE, 'the fixture the Actor transition test uses is exactly what this emitter writes');
++
++(async () => {
++ const rows = [];
++ const outbox = { emit: async (env) => { rows.push(env); return env; } };
++ await emitAgentStep(outbox, WATCH, { topic: 'started', status: 'waiting', startedAt: 1762000000000, now: () => 1762000000000 });
++ assert.strictEqual(rows.length, 1);
++ assert.strictEqual(rows[0].event_type, 'agent.step.started');
++ assert.strictEqual(rows[0].source, 'watch');
++ assert.deepStrictEqual(rows[0].subject, { type: 'user', id: WATCH.owner_sub });
++ assert.strictEqual(rows[0].visibility, 'internal');
++ // An agent's own watch (agt_) has no person to show it to.
++ await emitAgentStep(outbox, { ...WATCH, owner_sub: 'agt_01JAB2C3D4E5F6G7H8J9K0MNPS' }, { topic: 'started', status: 'waiting' });
++ assert.strictEqual(rows.length, 1, 'an agent-owned watch publishes no person\'s step');
++ console.log('watch agent step: one watch, one waiting step — never the page, the value or the condition');
++ console.log('all passed');
++})().catch((err) => { console.error(err); process.exit(1); });
ddiff --git a/test/fixtures/watch-agent-step.json b/test/fixtures/watch-agent-step.json
nnew file mode 100644
iindex 0000000..fbc3e38
---- /dev/null
++++ b/test/fixtures/watch-agent-step.json
@@@ -0,0 +1,18 @@
++{
++ "step_id": "wch_01JAB2C3D4E5F6G7H8J9K0MNPS",
++ "mode": "watch",
++ "product": "watch",
++ "ref": "wch_01JAB2C3D4E5F6G7H8J9K0MNPS",
++ "title": "Watching #167 CI",
++ "status": "waiting",
++ "started_at": "2026-11-07T09:00:00.000Z",
++ "ended_at": null,
++ "placement": {
++ "where": "On the network",
++ "why": "it checks the page from our side, on your schedule, and only tells you when it changes"
++ },
++ "cost_usd": null,
++ "link": "https://openvibe.watch/watches/wch_01JAB2C3D4E5F6G7H8J9K0MNPS",
++ "reason": null,
++ "transition": "Waiting for CI on #167 (Watch, 4 min), then I'll merge it (Act) and deploy on your machine (Run)."
++}