openvibers/openvibe.actor Commits
Bridge Codes session results into Actor (#398)
Changes
@@@ -60,6 +60,15 @@ function answerOf(t) {return lines.join('\n');}++/** Validate the Codes v1 session envelope before its untrusted harness output enters Actor's step stream. */++function sessionEventsOf(page, taskId) {++ if (!page || page.task_id !== taskId || !Array.isArray(page.lines)) return [];++ const allowed = new Set(['step', 'tool', 'tool_result', 'done']);++ return page.lines.filter(line => line && Number.isSafeInteger(line.seq) && line.seq > 0 && allowed.has(line.kind)++ && line.body && typeof line.body === 'object' && !Array.isArray(line.body) && line.body.t === line.kind)++ .sort((a,b) => a.seq - b.seq).slice(-500);++}++function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverride = null, sleep = (ms, signal) => new Promise((ok) => { const h = setTimeout(ok, ms); if (signal) signal.addEventListener('abort', () => { clearTimeout(h); ok(); }, { once: true }); }), pollMs = 3000, now = () => Date.now() }) {const base = config.services.codesUrl;const pages = config.services.codesPublicUrl || 'https://openvibe.codes';@@@ -79,6 +88,24 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverrconst data = await res.json().catch(() => null);return { status: res.status, ok: res.ok, data };}++ async function sessionPage(taskId, subject, signal) {++ const lines = [];++ let after = 0, complete = false;++ for (let pageN = 0; pageN < 20; pageN++) {++ const sig = signal ? AbortSignal.any([signal, AbortSignal.timeout(20_000)]) : AbortSignal.timeout(20_000);++ const res = await fetchImpl(`${base}/run/tasks/${encodeURIComponent(taskId)}/session.json?after=${after}`, {++ headers: { Accept: 'application/json', 'X-OV-Subject': subject, ...(await tokens.authHeaders()) }, signal: sig,++ });++ if (!res.ok) return null;++ const page = await res.json().catch(() => null);++ if (!page || page.task_id !== taskId || !Array.isArray(page.lines)) return null;++ lines.push(...page.lines);++ complete = page.complete === true;++ if (!page.has_more || !Number.isSafeInteger(page.next_after) || page.next_after <= after) break;++ after = page.next_after;++ }++ return { task_id: taskId, lines, complete };++ }const why = (r) => (r.data && (r.data.detail || r.data.title)) || `OpenVibe.Codes answered ${r.status}`;async function adapter({ task, requester, signal, emit, taskId = null }) {@@@ -125,10 +152,18 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverrif (t.state === 'expired') throw new ProviderError(`No machine of yours took the task in time. Is it online at https://openvibe.link/machines, with the Codes job runner (install it with --worker)? The task: ${t.url}`, { code: 'actor.agent.failed' });if (t.state !== 'succeeded') throw new ProviderError(`The task ${t.state === 'cancelled' ? 'was cancelled' : 'failed'}${t.error ? `: ${t.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });if (!t.result || !t.result.ok) throw new ProviderError(`The agents on your machine did not finish${t.result && t.result.error ? `: ${t.result.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });-- return { text: answerOf(t), sources, model: (t.result.agents || [])[0] || 'openvibe-codes' };++ let text = answerOf(t);++ if (config.codesSessionBridge === true) {++ const page = await sessionPage(t.id, subject, signal).catch(() => null);++ const lines = sessionEventsOf(page, t.id);++ for (const line of lines) await emit({ step: 'tool_result', tool: `codes.session.${line.kind}`, text: JSON.stringify(line.body).slice(0, 3000) });++ const done = [...lines].reverse().find(line => line.kind === 'done' && line.body.ok === true);++ if (done && typeof done.body.text === 'string' && done.body.text.trim()) text += `\n\nCodes session result: ${done.body.text.trim().slice(0, 4000)}`;++ }++ return { text, sources, model: (t.result.agents || [])[0] || 'openvibe-codes' };}return { enabled, adapter };}--module.exports = { createCodes, repoIn, answerOf, personOf };++module.exports = { createCodes, repoIn, answerOf, personOf, sessionEventsOf };@@@ -28,6 +28,7 @@ function load(env = process.env) {return {service: 'actor',++ codesSessionBridge: env.ACTOR_CODES_SESSION_BRIDGE === '1' || env.ACTOR_CODES_SESSION_BRIDGE === 'true',a4ModelSwitch: env.ACTOR_A4_MODEL_SWITCH === '1' || env.ACTOR_A4_MODEL_SWITCH === 'true',fabric: { enabled: bool(env.ACTOR_WORK_FABRIC) },port,@@@ -32,9 +32,20 @@ function fakeCodes({ start = 201, startBody = null, states = ['running', 'succeeconst fetchImpl = async (url, init = {}) => {const u = new URL(url);const headers = init.headers || {};-- calls.push({ method: init.method || 'GET', path: u.pathname, subject: headers['X-OV-Subject'], auth: headers.Authorization, body: init.body ? JSON.parse(init.body) : null });++ calls.push({ method: init.method || 'GET', path: u.pathname, query:u.searchParams.toString(), subject: headers['X-OV-Subject'], auth: headers.Authorization, body: init.body ? JSON.parse(init.body) : null });if (u.pathname === '/api/v1/tasks' && init.method === 'POST') return reply(start, startBody || { task: task('queued'), created: true });if (u.pathname === `/api/v1/tasks/${TASK}` && (init.method || 'GET') === 'GET') return reply(200, task(states[Math.min(reads++, states.length - 1)]));++ if (u.pathname === `/run/tasks/${TASK}/session.json`) {++ const all = [++ { seq:1, kind:'step', body:{ t:'step', text:'Reading README' } },++ { seq:2, kind:'tool', body:{ t:'tool', id:'tool-1', name:'Read', input:{ file_path:'README' } } },++ { seq:3, kind:'tool_result', body:{ t:'tool_result', id:'tool-1', ok:true, output:'README contents' } },++ { seq:4, kind:'done', body:{ t:'done', ok:true, text:'Finished the README edit.' } },++ ];++ const after = Number(u.searchParams.get('after') || 0), lines = all.filter(line => line.seq > after);++ const page = lines.slice(0, after === 0 ? 3 : 500);++ return reply(200, { task_id:TASK, source:'session', after, lines:page, next_after:page.at(-1)?.seq || after, has_more:page.at(-1)?.seq !== all.at(-1).seq, complete:true });++ }if (u.pathname === `/api/v1/tasks/${TASK}/cancel`) return reply(200, task('cancelled'));return reply(404, { code: 'not_found' });};@@@ -63,9 +74,20 @@ const run = (codes, extra = {}) => {assert.strictEqual(start.body.idempotency_key, 'act_task_01');assert.ok(out.text.includes('Done on your machine') && out.text.includes('Appended the line.') && out.text.includes('`README` (+1 −0)') && out.text.includes('```diff'), out.text);assert.deepStrictEqual(out.sources, [`https://openvibe.codes/run/tasks/${TASK}`]);++ assert.ok(!codes.calls.some(c=>c.path.endsWith('/session.json')), 'the session bridge is off unless the owner enables its flag');assert.ok(events.some((e) => e.text && e.text.startsWith('Your machine is working on it')));});++ await check('the default-off session bridge reads the Codes session and emits its tool calls and result as Actor steps', async () => {++ const codes = fakeCodes();++ const { out, events } = await run(createCodes({ config:{...config,codesSessionBridge:true}, fetchImpl:codes.fetchImpl, tokens, sleep:async()=>{}, pollMs:0 }));++ assert.ok(codes.calls.some(c=>c.path===`/run/tasks/${TASK}/session.json` && c.subject===ME && c.auth==='Bearer svc-token'));++ assert.ok(codes.calls.some(c=>c.path===`/run/tasks/${TASK}/session.json` && c.query==='after=3'), 'the client follows session pagination');++ assert.ok(events.some(e=>e.tool==='codes.session.tool' && e.step==='tool_result'));++ assert.ok(events.some(e=>e.tool==='codes.session.tool_result' && e.text.includes('README contents')));++ assert.match(out.text,/Codes session result: Finished the README edit\./);++ });++await check('signed out, no repository, or no /run task yet: a refusal that says what to do', async () => {const codes = createCodes({ config, fetchImpl: fakeCodes().fetchImpl, tokens, sleep: async () => {}, pollMs: 0 });await assert.rejects(codes.adapter({ task: 'https://github.com/a/b fix', requester: 'ip:1.2.3.4', emit: async () => {} }), /signed in/);
@@ -60,6 +60,15 @@ function answerOf(t) {
@@ -60,6 +60,15 @@ function answerOf(t) {
return lines.join('\n');
return lines.join('\n');
}
}
+/** Validate the Codes v1 session envelope before its untrusted harness output enters Actor's step stream. */
+function sessionEventsOf(page, taskId) {
+ if (!page || page.task_id !== taskId || !Array.isArray(page.lines)) return [];
+ const allowed = new Set(['step', 'tool', 'tool_result', 'done']);
+ return page.lines.filter(line => line && Number.isSafeInteger(line.seq) && line.seq > 0 && allowed.has(line.kind)
+ && line.body && typeof line.body === 'object' && !Array.isArray(line.body) && line.body.t === line.kind)
+ .sort((a,b) => a.seq - b.seq).slice(-500);
+}
+
function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverride = null, sleep = (ms, signal) => new Promise((ok) => { const h = setTimeout(ok, ms); if (signal) signal.addEventListener('abort', () => { clearTimeout(h); ok(); }, { once: true }); }), pollMs = 3000, now = () => Date.now() }) {
function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverride = null, sleep = (ms, signal) => new Promise((ok) => { const h = setTimeout(ok, ms); if (signal) signal.addEventListener('abort', () => { clearTimeout(h); ok(); }, { once: true }); }), pollMs = 3000, now = () => Date.now() }) {
const base = config.services.codesUrl;
const base = config.services.codesUrl;
const pages = config.services.codesPublicUrl || 'https://openvibe.codes';
const pages = config.services.codesPublicUrl || 'https://openvibe.codes';
@@ -79,6 +88,24 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverr
@@ -79,6 +88,24 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverr
const data = await res.json().catch(() => null);
const data = await res.json().catch(() => null);
return { status: res.status, ok: res.ok, data };
return { status: res.status, ok: res.ok, data };
}
}
+ async function sessionPage(taskId, subject, signal) {
+ const lines = [];
+ let after = 0, complete = false;
+ for (let pageN = 0; pageN < 20; pageN++) {
+ const sig = signal ? AbortSignal.any([signal, AbortSignal.timeout(20_000)]) : AbortSignal.timeout(20_000);
+ const res = await fetchImpl(`${base}/run/tasks/${encodeURIComponent(taskId)}/session.json?after=${after}`, {
+ headers: { Accept: 'application/json', 'X-OV-Subject': subject, ...(await tokens.authHeaders()) }, signal: sig,
+ });
+ if (!res.ok) return null;
+ const page = await res.json().catch(() => null);
+ if (!page || page.task_id !== taskId || !Array.isArray(page.lines)) return null;
+ lines.push(...page.lines);
+ complete = page.complete === true;
+ if (!page.has_more || !Number.isSafeInteger(page.next_after) || page.next_after <= after) break;
+ after = page.next_after;
+ }
+ return { task_id: taskId, lines, complete };
+ }
const why = (r) => (r.data && (r.data.detail || r.data.title)) || `OpenVibe.Codes answered ${r.status}`;
const why = (r) => (r.data && (r.data.detail || r.data.title)) || `OpenVibe.Codes answered ${r.status}`;
async function adapter({ task, requester, signal, emit, taskId = null }) {
async function adapter({ task, requester, signal, emit, taskId = null }) {
@@ -125,10 +152,18 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverr
@@ -125,10 +152,18 @@ function createCodes({ config, fetchImpl = globalThis.fetch, tokens: tokensOverr
if (t.state === 'expired') throw new ProviderError(`No machine of yours took the task in time. Is it online at https://openvibe.link/machines, with the Codes job runner (install it with --worker)? The task: ${t.url}`, { code: 'actor.agent.failed' });
if (t.state === 'expired') throw new ProviderError(`No machine of yours took the task in time. Is it online at https://openvibe.link/machines, with the Codes job runner (install it with --worker)? The task: ${t.url}`, { code: 'actor.agent.failed' });
if (t.state !== 'succeeded') throw new ProviderError(`The task ${t.state === 'cancelled' ? 'was cancelled' : 'failed'}${t.error ? `: ${t.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });
if (t.state !== 'succeeded') throw new ProviderError(`The task ${t.state === 'cancelled' ? 'was cancelled' : 'failed'}${t.error ? `: ${t.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });
if (!t.result || !t.result.ok) throw new ProviderError(`The agents on your machine did not finish${t.result && t.result.error ? `: ${t.result.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });
if (!t.result || !t.result.ok) throw new ProviderError(`The agents on your machine did not finish${t.result && t.result.error ? `: ${t.result.error}` : ''}. ${t.url}`, { code: 'actor.agent.failed' });
- return { text: answerOf(t), sources, model: (t.result.agents || [])[0] || 'openvibe-codes' };
+ let text = answerOf(t);
+ if (config.codesSessionBridge === true) {
+ const page = await sessionPage(t.id, subject, signal).catch(() => null);
+ const lines = sessionEventsOf(page, t.id);
+ for (const line of lines) await emit({ step: 'tool_result', tool: `codes.session.${line.kind}`, text: JSON.stringify(line.body).slice(0, 3000) });
+ const done = [...lines].reverse().find(line => line.kind === 'done' && line.body.ok === true);
+ if (done && typeof done.body.text === 'string' && done.body.text.trim()) text += `\n\nCodes session result: ${done.body.text.trim().slice(0, 4000)}`;
+ }
+ return { text, sources, model: (t.result.agents || [])[0] || 'openvibe-codes' };
}
}
return { enabled, adapter };
return { enabled, adapter };
}
}
-module.exports = { createCodes, repoIn, answerOf, personOf };
+module.exports = { createCodes, repoIn, answerOf, personOf, sessionEventsOf };
@@ -28,6 +28,7 @@ function load(env = process.env) {
@@ -28,6 +28,7 @@ function load(env = process.env) {
return {
return {
service: 'actor',
service: 'actor',
+ codesSessionBridge: env.ACTOR_CODES_SESSION_BRIDGE === '1' || env.ACTOR_CODES_SESSION_BRIDGE === 'true',
a4ModelSwitch: env.ACTOR_A4_MODEL_SWITCH === '1' || env.ACTOR_A4_MODEL_SWITCH === 'true',
a4ModelSwitch: env.ACTOR_A4_MODEL_SWITCH === '1' || env.ACTOR_A4_MODEL_SWITCH === 'true',
fabric: { enabled: bool(env.ACTOR_WORK_FABRIC) },
fabric: { enabled: bool(env.ACTOR_WORK_FABRIC) },
port,
port,
@@ -32,9 +32,20 @@ function fakeCodes({ start = 201, startBody = null, states = ['running', 'succee
@@ -32,9 +32,20 @@ function fakeCodes({ start = 201, startBody = null, states = ['running', 'succee
const fetchImpl = async (url, init = {}) => {
const fetchImpl = async (url, init = {}) => {
const u = new URL(url);
const u = new URL(url);
const headers = init.headers || {};
const headers = init.headers || {};
- calls.push({ method: init.method || 'GET', path: u.pathname, subject: headers['X-OV-Subject'], auth: headers.Authorization, body: init.body ? JSON.parse(init.body) : null });
+ calls.push({ method: init.method || 'GET', path: u.pathname, query:u.searchParams.toString(), subject: headers['X-OV-Subject'], auth: headers.Authorization, body: init.body ? JSON.parse(init.body) : null });
if (u.pathname === '/api/v1/tasks' && init.method === 'POST') return reply(start, startBody || { task: task('queued'), created: true });
if (u.pathname === '/api/v1/tasks' && init.method === 'POST') return reply(start, startBody || { task: task('queued'), created: true });
if (u.pathname === `/api/v1/tasks/${TASK}` && (init.method || 'GET') === 'GET') return reply(200, task(states[Math.min(reads++, states.length - 1)]));
if (u.pathname === `/api/v1/tasks/${TASK}` && (init.method || 'GET') === 'GET') return reply(200, task(states[Math.min(reads++, states.length - 1)]));
+ if (u.pathname === `/run/tasks/${TASK}/session.json`) {
+ const all = [
+ { seq:1, kind:'step', body:{ t:'step', text:'Reading README' } },
+ { seq:2, kind:'tool', body:{ t:'tool', id:'tool-1', name:'Read', input:{ file_path:'README' } } },
+ { seq:3, kind:'tool_result', body:{ t:'tool_result', id:'tool-1', ok:true, output:'README contents' } },
+ { seq:4, kind:'done', body:{ t:'done', ok:true, text:'Finished the README edit.' } },
+ ];
+ const after = Number(u.searchParams.get('after') || 0), lines = all.filter(line => line.seq > after);
+ const page = lines.slice(0, after === 0 ? 3 : 500);
+ return reply(200, { task_id:TASK, source:'session', after, lines:page, next_after:page.at(-1)?.seq || after, has_more:page.at(-1)?.seq !== all.at(-1).seq, complete:true });
+ }
if (u.pathname === `/api/v1/tasks/${TASK}/cancel`) return reply(200, task('cancelled'));
if (u.pathname === `/api/v1/tasks/${TASK}/cancel`) return reply(200, task('cancelled'));
return reply(404, { code: 'not_found' });
return reply(404, { code: 'not_found' });
};
};
@@ -63,9 +74,20 @@ const run = (codes, extra = {}) => {
@@ -63,9 +74,20 @@ const run = (codes, extra = {}) => {
assert.strictEqual(start.body.idempotency_key, 'act_task_01');
assert.strictEqual(start.body.idempotency_key, 'act_task_01');
assert.ok(out.text.includes('Done on your machine') && out.text.includes('Appended the line.') && out.text.includes('`README` (+1 −0)') && out.text.includes('```diff'), out.text);
assert.ok(out.text.includes('Done on your machine') && out.text.includes('Appended the line.') && out.text.includes('`README` (+1 −0)') && out.text.includes('```diff'), out.text);
assert.deepStrictEqual(out.sources, [`https://openvibe.codes/run/tasks/${TASK}`]);
assert.deepStrictEqual(out.sources, [`https://openvibe.codes/run/tasks/${TASK}`]);
+ assert.ok(!codes.calls.some(c=>c.path.endsWith('/session.json')), 'the session bridge is off unless the owner enables its flag');
assert.ok(events.some((e) => e.text && e.text.startsWith('Your machine is working on it')));
assert.ok(events.some((e) => e.text && e.text.startsWith('Your machine is working on it')));
});
});
+ await check('the default-off session bridge reads the Codes session and emits its tool calls and result as Actor steps', async () => {
+ const codes = fakeCodes();
+ const { out, events } = await run(createCodes({ config:{...config,codesSessionBridge:true}, fetchImpl:codes.fetchImpl, tokens, sleep:async()=>{}, pollMs:0 }));
+ assert.ok(codes.calls.some(c=>c.path===`/run/tasks/${TASK}/session.json` && c.subject===ME && c.auth==='Bearer svc-token'));
+ assert.ok(codes.calls.some(c=>c.path===`/run/tasks/${TASK}/session.json` && c.query==='after=3'), 'the client follows session pagination');
+ assert.ok(events.some(e=>e.tool==='codes.session.tool' && e.step==='tool_result'));
+ assert.ok(events.some(e=>e.tool==='codes.session.tool_result' && e.text.includes('README contents')));
+ assert.match(out.text,/Codes session result: Finished the README edit\./);
+ });
+
await check('signed out, no repository, or no /run task yet: a refusal that says what to do', async () => {
await check('signed out, no repository, or no /run task yet: a refusal that says what to do', async () => {
const codes = createCodes({ config, fetchImpl: fakeCodes().fetchImpl, tokens, sleep: async () => {}, pollMs: 0 });
const codes = createCodes({ config, fetchImpl: fakeCodes().fetchImpl, tokens, sleep: async () => {}, pollMs: 0 });
await assert.rejects(codes.adapter({ task: 'https://github.com/a/b fix', requester: 'ip:1.2.3.4', emit: async () => {} }), /signed in/);
await assert.rejects(codes.adapter({ task: 'https://github.com/a/b fix', requester: 'ip:1.2.3.4', emit: async () => {} }), /signed in/);