openvibers/openvibe.surf Commits

A6: Surf publishes the agent's step (agent.step.started|ended) (#91)

OP OpenVibers committed 6 days ago · 06cd50f591 · 5 files, +197 −4

A browsing task is one Browse step of the agent's timeline: a generated title, its
state and its page. Surf keeps no database and no outbox — the task is Actor's — so
the step goes straight to Events with Surf's service token, best effort, and a new
state per task. Never the person's query, a page, a source or the report.

Changes

ddiff --git a/server/config.js b/server/config.js
iindex 4382ba9..72be710 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -55,6 +55,11 @@ function load(env = process.env) {
// Ride: maps are cached 60 s; Games has a little longer than one Live source to answer.
rideCacheMs: Math.max(0, int(env.SURF_RIDE_CACHE_MS, 60_000)),
gamesTimeoutMs: atLeast(env.SURF_GAMES_TIMEOUT_MS, 200, 4000),
++ // The agent's step (design 2026-11-07): Surf publishes agent.step.started/ended as the person's task changes
++ // state. Off until EVENTS_URL and the OAuth client secret are set, like every service's outbox.
++ events: {
++ url: String(env.EVENTS_URL || '').replace(/\/+$/, ''),
++ },
// Web: Actor creates a task quickly, but a cold allowance check may take a moment.
actorTimeoutMs: atLeast(env.SURF_ACTOR_TIMEOUT_MS, 200, 15_000),
// Watch together: in-memory parties, hard caps 200 open parties and 50 members each. The environment may
ddiff --git a/server/http/api.js b/server/http/api.js
iindex 0227dd3..fdeb3e3 100644
---- a/server/http/api.js
++++ b/server/http/api.js
@@@ -97,6 +97,7 @@ function createApi(ctx) {
token: req.webToken,
query,
idempotencyKey: req.get('idempotency-key') || null,
++ subject: req.viewer && req.viewer.subject ? req.viewer.subject : null,
});
res.set('Location', out.page);
return res.status(201).json(out);
@@@ -108,7 +109,7 @@ function createApi(ctx) {
});
r.get('/v1/web/tasks/:id', requireToken, async (req, res) => {
-- try { return res.json(await web.task({ token: req.webToken, id: req.params.id })); }
++ try { return res.json(await web.task({ token: req.webToken, id: req.params.id, subject: req.viewer && req.viewer.subject ? req.viewer.subject : null })); }
catch (err) { return problem(req, res, err.status || 502, 'web.task.unavailable', err.message); }
});
r.get('/v1/web/tasks/:id/events', requireToken, (req, res) => web.events({ token: req.webToken, id: req.params.id, after: req.get('last-event-id'), res }));
ddiff --git a/server/web/agent-step.js b/server/web/agent-step.js
nnew file mode 100644
iindex 0000000..504bc64
---- /dev/null
++++ b/server/web/agent-step.js
@@@ -0,0 +1,109 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): OpenVibe.Surf tells the one timeline behind the Dock what the browsing task
++ * it started for a person is doing. Surf keeps no database and no outbox — the task is OpenVibe.Actor's — so the
++ * step is published straight to OpenVibe.Events with Surf's service token, best effort: Events being down never fails
++ * the task. The payload is a generated title, an outcome and a link: never the person's query, a page, a source or
++ * the report.
++ */
++const AGENT_TOPICS = Object.freeze({
++ started: 'agent.step.started',
++ updated: 'agent.step.updated',
++ ended: 'agent.step.ended',
++});
++const USER_SUBJECT = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
++const STATUS_OF = Object.freeze({
++ queued: 'waiting', running: 'running', verifying: 'running', needs_you: 'waiting',
++ succeeded: 'done', failed: 'failed', cancelled: 'stopped',
++});
++const ENDED = new Set(['done', 'failed', 'stopped']);
++
++/** One browsing task → the agent's step: a generated title, its state and its page. Never the query or a source. */
++function agentStepOf(task, { status = null, baseUrl = 'https://openvibe.surf', startedAt = null, endedAt = null, costUsd = null, reason = null } = {}) {
++ const state = STATUS_OF[status] || status || STATUS_OF[task.state] || 'running';
++ const base = String(baseUrl || 'https://openvibe.surf').replace(/\/+$/, '');
++ const cost = costUsd == null && task.cost ? Number(task.cost.usd) : costUsd;
++ return {
++ step_id: String(task.id),
++ mode: 'browse',
++ product: 'surf',
++ ref: String(task.id),
++ title: 'Surfing the web for you',
++ status: state,
++ started_at: new Date(startedAt == null ? Date.now() : startedAt).toISOString(),
++ ended_at: ENDED.has(state) ? new Date(endedAt == null ? Date.now() : endedAt).toISOString() : null,
++ placement: { where: 'On a private desk', why: 'it browses for you, never with your accounts, and stops at logins and anything that would spend money' },
++ cost_usd: Number.isFinite(cost) && cost >= 0 ? cost : null,
++ link: `${base}/web/t/${encodeURIComponent(String(task.id))}`,
++ reason: state === 'failed' || state === 'stopped' ? (reason == null ? null : String(reason).slice(0, 240)) : null,
++ };
++}
++
++/** The topic a state is: the step starts, moves on, or ends. */
++function topicOf(state) {
++ return state === 'waiting' || state === 'running' ? 'started' : 'ended';
++}
++
++/**
++ * The publisher. `publish` may be passed in (tests, or a deployment with its own client); otherwise an
++ * openvibe-sdk Events client with Surf's service token is made once. Without EVENTS_URL and a client secret, every
++ * call is a quiet no-op.
++ */
++function createAgentSteps({ config = {}, publish = null, log = console, now = Date.now } = {}) {
++ const enabled = Boolean(publish || (config.events && config.events.url && config.oauth && config.oauth.clientSecret));
++ const sent = new Map(); // subject|task → the last state published, so a polled page is not a row every second
++ let client = null;
++
++ function eventsClient() {
++ if (client) return client;
++ const { createClient } = require('openvibe-sdk/core');
++ const { createServiceTokenClient } = require('openvibe-sdk/auth');
++ const { createEventsClient } = require('openvibe-sdk/events');
++ const tokenProvider = createServiceTokenClient({
++ tokenUrl: `${String(config.networkInternalUrl || 'http://127.0.0.1:4000').replace(/\/+$/, '')}/oauth/token`,
++ clientId: config.oauth.clientId || 'surf', clientSecret: config.oauth.clientSecret,
++ scope: { 'openvibe.events': 'events.event.publish' },
++ });
++ client = createEventsClient(createClient({ baseUrls: { events: config.events.url }, tokenProvider, retries: 0 }), { source: 'surf' });
++ return client;
++ }
++
++ /**
++ * Publish the task's step if its state is new for this person. `subject` is the person whose token ran the task.
++ * Returns the envelope (or null); never throws.
++ */
++ async function note(task, { subject = null } = {}) {
++ if (!enabled || !task || !String(task.id || '') || !USER_SUBJECT.test(String(subject || ''))) return null;
++ const state = STATUS_OF[task.state] || null;
++ if (!state) return null;
++ const key = `${subject}|${task.id}`;
++ if (sent.get(key) === state) return null;
++ const envelope = {
++ event_type: AGENT_TOPICS[topicOf(state)],
++ actor: { type: 'user', id: subject },
++ subject: { type: 'user', id: subject },
++ visibility: 'internal',
++ priority: ENDED.has(state) ? 'important' : 'low',
++ timestamp: new Date(now()).toISOString(),
++ payload: agentStepOf(task, {
++ status: state, baseUrl: config.baseUrl, startedAt: task.created_at ? Date.parse(task.created_at) : now(), now,
++ reason: state === 'failed' ? (task.error && (task.error.detail || task.error.code)) || 'it did not finish' : null,
++ }),
++ };
++ sent.set(key, state);
++ if (sent.size > 2000) sent.delete(sent.keys().next().value);
++ try {
++ if (publish) await publish(envelope);
++ else await eventsClient().publish(envelope);
++ return envelope;
++ } catch (err) {
++ sent.delete(key); // the next read tries the same state again
++ if (log && log.warn) log.warn('[OpenVibe.Surf] agent step:', err && err.message);
++ return null;
++ }
++ }
++
++ return { enabled, note, _sent: sent };
++}
++
++module.exports = { AGENT_TOPICS, agentStepOf, topicOf, createAgentSteps, STATUS_OF, USER_SUBJECT };
ddiff --git a/server/web/tasks.js b/server/web/tasks.js
iindex 347d2ae..82592fb 100644
---- a/server/web/tasks.js
++++ b/server/web/tasks.js
@@@ -17,6 +17,7 @@
* pass through. A timeout or a 5xx is this service's 502.
*/
const crypto = require('crypto');
++const { createAgentSteps } = require('./agent-step');
const MIN_QUERY = 3;
const MAX_QUERY = 500;
@@@ -39,11 +40,13 @@ function newIdempotencyKey() {
return `surf:web:${crypto.randomBytes(16).toString('hex')}`;
}
--function createWeb({ config, fetchImpl = globalThis.fetch, log = console }) {
++function createWeb({ config, fetchImpl = globalThis.fetch, log = console, steps = null }) {
++ // The agent's step (design 2026-11-07): the browsing task is the Browse step of the one timeline behind the Dock.
++ const agentSteps = steps || createAgentSteps({ config, log });
const taskPath = (id) => `${config.actorInternalUrl}/api/v1/tasks/${encodeURIComponent(id)}`;
const headers = (token) => ({ Accept: 'application/json', Authorization: `Bearer ${token}` });
-- async function task({ token, id }) {
++ async function task({ token, id, subject = null }) {
if (!token) throw Object.assign(new Error('Sign in to see your task.'), { status: 401 });
if (!TASK_ID.test(String(id || ''))) throw Object.assign(new Error('Task not found.'), { status: 404 });
let res;
@@@ -56,6 +59,8 @@ function createWeb({ config, fetchImpl = globalThis.fetch, log = console }) {
let data;
try { data = await res.json(); } catch { /* invalid Actor response */ }
if (!data || data.id !== id) throw Object.assign(new Error('OpenVibe.Actor sent an invalid task.'), { status: 502 });
++ // A read of the task's own page is how Surf sees its outcome: publish the new state, once per state.
++ agentSteps.note(data, { subject }).catch(() => {});
return data;
}
@@@ -89,7 +94,7 @@ function createWeb({ config, fetchImpl = globalThis.fetch, log = console }) {
* Create one task for the person whose token this is. Resolves { task_id, page } (Surf's task page) or throws
* { status, code, detail, actor } — a 4xx from Actor keeps its status and detail.
*/
-- async function createTask({ token, query, idempotencyKey = null } = {}) {
++ async function createTask({ token, query, idempotencyKey = null, subject = null } = {}) {
if (!token) throw Object.assign(new Error('Sign in to surf the web.'), { status: 401, code: 'token.required' });
if (!validQuery(query)) throw Object.assign(new Error('query must be 3 to 500 characters.'), { status: 400, code: 'web.invalid_query' });
const key = IDEMPOTENCY.test(String(idempotencyKey || '')) ? String(idempotencyKey) : newIdempotencyKey();
@@@ -130,6 +135,7 @@ function createWeb({ config, fetchImpl = globalThis.fetch, log = console }) {
log.warn('[OpenVibe.Surf] Actor answered without a task id');
throw Object.assign(new Error('OpenVibe.Actor answered without a task id.'), { status: 502, code: 'actor.invalid_response' });
}
++ agentSteps.note(data, { subject }).catch(() => {});
return { task_id: id, page: `/web/t/${encodeURIComponent(id)}` };
}
ddiff --git a/test/agent-step.test.js b/test/agent-step.test.js
nnew file mode 100644
iindex 0000000..4a61ff4
---- /dev/null
++++ b/test/agent-step.test.js
@@@ -0,0 +1,72 @@
++'use strict';
++/**
++ * The agent's step (design 2026-11-07): the browsing task Surf starts for a person is the Browse step of the one
++ * timeline behind the Dock. Surf has no database and no outbox, so the step is published to OpenVibe.Events with
++ * Surf's service token, best effort, once per state. The payload is a generated title, an outcome and a link: never
++ * the person's query, a page it read, a source or the report.
++ */
++const assert = require('assert');
++const { AGENT_TOPICS, agentStepOf, topicOf, createAgentSteps, STATUS_OF } = require('../server/web/agent-step');
++
++const SUBJECT = 'usr_01JAB2C3D4E5F6G7H8J9K0MNPS';
++const TASK = {
++ id: 'tsk_01JAB2C3D4E5F6G7H8J9K0MNPS', state: 'running',
++ task: 'find me a cheap flight to Lisbon, the person\'s own words',
++ result: { answer: 'the report the agent wrote', sources: ['https://example.test/private-page'] },
++ cost: { usd: 0.07 }, created_at: '2026-11-07T05:00:00.000Z',
++};
++
++assert.strictEqual(AGENT_TOPICS.started, 'agent.step.started');
++assert.strictEqual(AGENT_TOPICS.ended, 'agent.step.ended');
++assert.strictEqual(STATUS_OF.queued, 'waiting');
++assert.strictEqual(STATUS_OF.verifying, 'running');
++assert.strictEqual(topicOf('waiting'), 'started');
++assert.strictEqual(topicOf('done'), 'ended');
++
++const running = agentStepOf(TASK, { status: 'running', startedAt: Date.parse(TASK.created_at), baseUrl: 'https://openvibe.surf' });
++assert.strictEqual(running.step_id, TASK.id);
++assert.strictEqual(running.mode, 'browse');
++assert.strictEqual(running.product, 'surf');
++assert.strictEqual(running.status, 'running');
++assert.strictEqual(running.title, 'Surfing the web for you');
++assert.strictEqual(running.link, `https://openvibe.surf/web/t/${TASK.id}`);
++assert.strictEqual(running.ended_at, null);
++assert.strictEqual(running.cost_usd, 0.07);
++assert.ok(running.placement && running.placement.where && running.placement.why);
++const raw = JSON.stringify(running);
++assert.ok(!raw.includes('cheap flight'), 'the query never travels');
++assert.ok(!raw.includes('the report the agent wrote'), 'the report never travels');
++assert.ok(!raw.includes('example.test/private-page'), 'a source never travels');
++assert.ok(!('task' in running) && !('result' in running) && !('sources' in running), 'no field for the content');
++
++const failed = agentStepOf({ ...TASK, state: 'failed', error: { detail: 'the site refused the desk' } }, { reason: 'the site refused the desk' });
++assert.strictEqual(failed.status, 'failed');
++assert.strictEqual(failed.reason, 'the site refused the desk');
++assert.ok(failed.ended_at, 'an ended step says when it ended');
++
++(async () => {
++ const sent = [];
++ const steps = createAgentSteps({
++ config: { events: { url: 'http://events.test' }, oauth: { clientId: 'surf', clientSecret: 'x' }, baseUrl: 'https://openvibe.surf' },
++ publish: async (env) => { sent.push(env); },
++ });
++ assert.strictEqual(steps.enabled, true);
++ await steps.note(TASK, { subject: SUBJECT });
++ await steps.note(TASK, { subject: SUBJECT });
++ assert.strictEqual(sent.length, 1, 'one row per state, however often the page is read');
++ assert.strictEqual(sent[0].event_type, 'agent.step.started');
++ assert.deepStrictEqual(sent[0].subject, { type: 'user', id: SUBJECT });
++ assert.strictEqual(sent[0].visibility, 'internal');
++ await steps.note({ ...TASK, state: 'succeeded' }, { subject: SUBJECT });
++ assert.strictEqual(sent.length, 2);
++ assert.strictEqual(sent[1].event_type, 'agent.step.ended');
++ assert.strictEqual(sent[1].payload.status, 'done');
++ await steps.note(TASK, { subject: 'usr_nope' });
++ assert.strictEqual(sent.length, 2, 'no person subject, no step');
++ // Off (no EVENTS_URL or no client secret) is a quiet no-op, like every service whose relay is off.
++ const off = createAgentSteps({ config: { events: { url: '' } } });
++ assert.strictEqual(off.enabled, false);
++ assert.strictEqual(await off.note(TASK, { subject: SUBJECT }), null);
++ console.log('surf agent step: a generated title, an outcome and a link — never the query, a page or a source');
++ console.log('all passed');
++})().catch((err) => { console.error(err); process.exit(1); });