diff --git a/README.md b/README.md index f534a79..6c6dbea 100644 --- a/README.md +++ b/README.md @@ -491,6 +491,15 @@ windows without a new event, when the run is no longer active, or when the collector records that observation finished. Nothing in the timeline restarts a run or calls a model. +When the host advertises `agents.text_stream`, each step a run is working on +shows a **Live answer** pane while it runs: the answer as the model writes it, +labelled provisional, never the verified result. Text written before a tool +call is withdrawn and the pane says so; a reader that fell behind the host's +window is told what is missing rather than shown a spliced passage; when the +run has a receipt the pane yields to the verified result, which decides the +outcome. The pane reconnects from its last sequence on request, and never +restarts a run. + Run `node --test scripts/test-agents.cjs` with the same Playwright environment variables used above. It verifies packaged desktop/mobile editing, persistence, forged save receipts, conflict/draft handling, delayed paging and capability gates. diff --git a/scripts/test-agents.cjs b/scripts/test-agents.cjs index 0cad5e0..8c397d4 100644 --- a/scripts/test-agents.cjs +++ b/scripts/test-agents.cjs @@ -6,7 +6,7 @@ const engines=require(process.env.SCONE_PLAYWRIGHT_MODULE||'playwright'); const contract=require('../tests/fixtures/http-capabilities.json');let browser; before(async()=>{browser=await engines[process.env.SCONE_BROWSER_ENGINE||'chromium'].launch({headless:true,executablePath:process.env.SCONE_BROWSER_PATH});}); after(async()=>{await browser?.close();}); -async function fixture(t,{mobile=false,supported=true,paged=false,runs=false,maxParallel=1,handoffs=false,history=false}={}){ +async function fixture(t,{mobile=false,supported=true,paged=false,runs=false,maxParallel=1,handoffs=false,history=false,answers=false}={}){ const html=fs.readFileSync(path.resolve(__dirname,'../dist/console.html'),'utf8').replaceAll('__SCONE_TOKEN__','agent-fixture'); const state={handoffCapability:handoffs,saved:null,forged:false,hold:null,pageStarted:false,conflict:false,runs:new Map(),starts:0,cancels:0,complete:true,loseStart:false,forgeResult:false,substitute:false,forgeTarget:false,forgeFinal:false}; const server=http.createServer(async(req,res)=>{ @@ -16,7 +16,7 @@ async function fixture(t,{mobile=false,supported=true,paged=false,runs=false,max assert.equal(req.headers.authorization,'Bearer agent-fixture');res.setHeader('content-type','application/json'); const send=value=>res.end(JSON.stringify(value)); if(url.pathname==='/v1/status')return send({space:'alpha',episodes:0}); - if(url.pathname==='/v1/capabilities')return send({...contract.python,features:{...contract.python.features,'agents.handoffs':state.handoffCapability,'agents.catalog':supported,'agents.plans':supported,'agents.runs':runs,'agents.parallel':runs&&maxParallel>1,'agents.history':history}}); + if(url.pathname==='/v1/capabilities')return send({...contract.python,features:{...contract.python.features,'agents.handoffs':state.handoffCapability,'agents.catalog':supported,'agents.plans':supported,'agents.runs':runs,'agents.parallel':runs&&maxParallel>1,'agents.history':history,'agents.text_stream':answers}}); if(url.pathname==='/v1/agents/run-policy')return send({space:'alpha',max_parallel_tasks:maxParallel,max_active_runs:4}); if(url.pathname==='/v1/agents/catalog')return send({agents:(handoffs?['research','write']:['research']).map(agent_id=>({agent_id,default_model:'fast',models:[{model_id:'fast',label:'Fast local',revision:'1'},{model_id:'careful',label:'Careful local',revision:'1'}]}))}); if(url.pathname==='/v1/agent-plans'){ @@ -35,6 +35,19 @@ async function fixture(t,{mobile=false,supported=true,paged=false,runs=false,max if(runs&&url.pathname.startsWith('/v1/agent-runs/')){ const parts=url.pathname.split('/'),run=state.runs.get(decodeURIComponent(parts[3]));if(!run){res.writeHead(404);return send({error:'missing'});} if(parts[4]==='request')return send(run.original); + if(parts[4]==='steps'&&parts[6]==='text'&&parts[7]==='stream'){ + // The answer a running step is writing: two pieces, then a terminal + // once the run has a receipt. A step that does not exist is 404. + state.answerStreams=(state.answerStreams??0)+1;state.answerAfter=url.searchParams.get('after'); + if(parts[5]!=='task-1'){res.writeHead(404);return send({error:'step_not_found',code:'step_not_found'});} + res.writeHead(200,{'content-type':'text/event-stream','cache-control':'no-store'}); + const after=parseInt(state.answerAfter??'0',10); + const pieces=[[1,'Answer from '],[2,'the model.']]; + for(const [sequence,text] of pieces)if(sequence>after)res.write('event: text\nid: '+sequence+'\ndata: '+JSON.stringify({sequence,text})+'\n\n'); + if(state.withdraw)res.write('event: withdraw\nid: 3\ndata: {"sequence":3}\n\n'); + const finish=()=>{if(state.complete)res.end('event: terminal\ndata: {"status":"completed","read_receipt":true}\n\n');else{state.answerOpen=res;}}; + return finish(); + } if(parts[4]==='history'){ // Execution metadata for the run: three recorded positions, and a // fourth that only the live stream delivers. Never any text. @@ -172,6 +185,33 @@ for(const mobile of [false,true])test(`the run timeline shows recorded execution assert.equal(await timeline.locator('li').count(),4);assert.equal(state.streams,1);assert.match(state.streamAfter,/^0{32}\.0{15}3\./,'the stream resumed from the last recorded position'); assert.equal(await page.getByText('a private prompt').count(),0);assert(await page.evaluate(()=>document.documentElement.scrollWidth<=innerWidth)); }); +for(const mobile of [false,true])test(`the answer a running step is writing appears as it arrives and yields to the verified result, mobile=${mobile}`,async t=>{ + const {page,state}=await fixture(t,{runs:true,answers:true,mobile});state.complete=false;await fill(page);await save(page);await page.getByText('Saved revision 1.',{exact:true}).waitFor(); + await page.getByLabel('Question',{exact:true}).fill('Write me an answer');const id=await page.getByLabel('Run identifier',{exact:true}).inputValue();await page.getByRole('button',{name:'Start run',exact:true}).click(); + const pane=page.getByRole('region',{name:'Answer being written for task-1',exact:true});await pane.waitFor(); + await pane.getByLabel('Provisional answer text for task-1',{exact:true}).waitFor(); + assert.equal(await pane.getByLabel('Provisional answer text for task-1',{exact:true}).innerText(),'Answer from the model.'); + await pane.getByText('Provisional, not the verified result',{exact:true}).waitFor(); + assert.equal(await page.getByLabel('Verified run results').count(),0,'provisional text is not a result'); + const run=state.runs.get(id);run.status={...run.status,status:'completed',active_local:false,completed_steps:['task-1'],inflight:null,inflight_steps:[]};state.complete=true; + if(state.answerOpen)state.answerOpen.end('event: terminal\ndata: {"status":"completed","read_receipt":true}\n\n'); + await page.getByLabel('Verified run results').waitFor(); + assert.equal(await page.getByRole('region',{name:'Answer being written for task-1',exact:true}).count(),0,'the verified result replaces the provisional pane'); + assert.equal(state.answerStreams,1);assert(await page.evaluate(()=>document.documentElement.scrollWidth<=innerWidth)); +}); +test('withdrawn provisional text is cleared and said',async t=>{ + const {page,state}=await fixture(t,{runs:true,answers:true});state.complete=false;state.withdraw=true;await fill(page);await save(page);await page.getByText('Saved revision 1.',{exact:true}).waitFor(); + await page.getByLabel('Question',{exact:true}).fill('Write me an answer');await page.getByRole('button',{name:'Start run',exact:true}).click(); + const pane=page.getByRole('region',{name:'Answer being written for task-1',exact:true});await pane.waitFor(); + await pane.getByText(/withdrawn; it was not the answer/).waitFor(); + assert.equal(await pane.getByLabel('Provisional answer text for task-1',{exact:true}).count(),0); +}); +test('without the text stream capability no provisional answer is offered',async t=>{ + const {page,state}=await fixture(t,{runs:true});state.complete=false;await fill(page);await save(page);await page.getByText('Saved revision 1.',{exact:true}).waitFor(); + await page.getByLabel('Question',{exact:true}).fill('No live answer');await page.getByRole('button',{name:'Start run',exact:true}).click(); + await page.getByRole('button',{name:'Cancel run',exact:true}).waitFor(); + assert.equal(await page.getByRole('region',{name:/Answer being written/}).count(),0);assert.equal(state.answerStreams,undefined); +}); test('a history entry carrying anything beyond metadata is withheld, not shown',async t=>{ const {page,state}=await fixture(t,{runs:true,history:true});state.leak=true;await fill(page);await save(page);await page.getByText('Saved revision 1.',{exact:true}).waitFor(); await page.getByLabel('Question',{exact:true}).fill('Show me the timeline');await page.getByRole('button',{name:'Start run',exact:true}).click(); diff --git a/src/agents/AgentsPage.tsx b/src/agents/AgentsPage.tsx index 2001417..ddb8adc 100644 --- a/src/agents/AgentsPage.tsx +++ b/src/agents/AgentsPage.tsx @@ -13,7 +13,7 @@ import {AgentTools} from './AgentTools'; const secureRequest={cache:'no-store',redirect:'error',credentials:'omit',referrerPolicy:'no-referrer'} as const; -type Ready={historyAvailable:boolean;approvalsAvailable:boolean;handoffRequirementsAvailable:boolean;requirementsAvailable:boolean;schemaAvailable:boolean;usageAvailable:boolean;inputsAvailable:boolean;handoffsAvailable:boolean;runsAvailable:boolean;maxParallel:number;space:string;catalog:AgentChoice[];items:SavedPlan[];next_after:string|null}; +type Ready={textStreamAvailable:boolean;historyAvailable:boolean;approvalsAvailable:boolean;handoffRequirementsAvailable:boolean;requirementsAvailable:boolean;schemaAvailable:boolean;usageAvailable:boolean;inputsAvailable:boolean;handoffsAvailable:boolean;runsAvailable:boolean;maxParallel:number;space:string;catalog:AgentChoice[];items:SavedPlan[];next_after:string|null}; function Editor({api,space,catalog,initial,onSave,onDirty,inputsAvailable,requirementsAvailable,schemaAvailable}:{api:ApiClient;space:string;catalog:AgentChoice[];initial:SavedPlan|null;onSave:(plan:SavedPlan)=>void;onDirty:()=>void;inputsAvailable:boolean;requirementsAvailable:boolean;schemaAvailable:boolean}){ const first=catalog[0]; const newTask=(id:string):AgentTask=>({task_id:id,agent_id:first?.agent_id??'',model_id:first?.default_model??'',prompt:'',depends_on:[]}); @@ -80,7 +80,7 @@ export function AgentsPage({api,enabled}:{api:ApiClient;enabled:boolean}){ if(!caps.features['agents.catalog']||!caps.features['agents.plans'])throw Error('Agent workflow configuration is not enabled on this server.'); const [catalog,page]=await Promise.all([api.request('/v1/agents/catalog',options).then(parseCatalog),api.request('/v1/agent-plans?limit=20',options).then(value=>parsePlanPage(value,status.space as string))]); const maxParallel=caps.features['agents.parallel']?parseRunPolicy(await api.request('/v1/agents/run-policy',options),status.space):1; - if(!controller.signal.aborted)setLoaded({api,data:{historyAvailable:caps.features['agents.history'],approvalsAvailable:caps.features['agents.approvals'],handoffRequirementsAvailable:caps.features['agents.handoffs.output_requirements'],requirementsAvailable:caps.features['agents.output_requirements'],schemaAvailable:caps.features['agents.output_schema'],usageAvailable:caps.features['agents.usage'],inputsAvailable:caps.features['agents.inputs'],handoffsAvailable:caps.features['agents.handoffs'],maxParallel,runsAvailable:caps.features['agents.runs'],space:status.space,catalog,...page}}); + if(!controller.signal.aborted)setLoaded({api,data:{textStreamAvailable:caps.features['agents.text_stream'],historyAvailable:caps.features['agents.history'],approvalsAvailable:caps.features['agents.approvals'],handoffRequirementsAvailable:caps.features['agents.handoffs.output_requirements'],requirementsAvailable:caps.features['agents.output_requirements'],schemaAvailable:caps.features['agents.output_schema'],usageAvailable:caps.features['agents.usage'],inputsAvailable:caps.features['agents.inputs'],handoffsAvailable:caps.features['agents.handoffs'],maxParallel,runsAvailable:caps.features['agents.runs'],space:status.space,catalog,...page}}); })().catch(error=>{if(!controller.signal.aborted)setError(error instanceof Error?error.message:'Agent configuration could not be loaded.');}).finally(()=>{if(!controller.signal.aborted)setLoading(false);}); return()=>controller.abort(); },[api,enabled,refresh]); @@ -101,7 +101,7 @@ export function AgentsPage({api,enabled}:{api:ApiClient;enabled:boolean}){ {(()=>{ const props={api,handoffRequirementsAvailable:data.handoffRequirementsAvailable,requirementsAvailable:data.requirementsAvailable,schemaAvailable:data.schemaAvailable,inputsAvailable:data.inputsAvailable,space:data.space,catalog:data.catalog,initial:selection.plan,onDirty:()=>setDirty(true),onSave:(saved:SavedPlan)=>{setDirty(false);setSelection(current=>({...current,plan:saved}));setLoaded(current=>current?.api===api?{api,data:{...current.data,items:[saved,...current.data.items.filter(item=>item.plan.workflow_id!==saved.plan.workflow_id)]}}:current);}}; return (selection.plan?isHandoffPlan(selection.plan.plan):selection.handoff)?data.handoffsAvailable?:

This server does not support handoff configuration.

:selection.plan&&isInteractivePlan(selection.plan.plan)&&!data.inputsAvailable?

This server does not support human input workflows.

:; - })()}{data.runsAvailable&&} {error&&

{error}

} + })()}{data.runsAvailable&&} {error&&

{error}

} } ; } diff --git a/src/agents/AnswerPane.tsx b/src/agents/AnswerPane.tsx new file mode 100644 index 0000000..04236d9 --- /dev/null +++ b/src/agents/AnswerPane.tsx @@ -0,0 +1,59 @@ +import {useEffect,useRef,useState} from 'react'; +import {ApiError,type ApiClient} from '../api'; +import {readAnswerStream} from './answer-stream'; + +const MAX_PREVIEW_BYTES=1024*1024; +type Phase='connecting'|'live'|'interrupted'|'waiting'|'ended'|'limited'; +// The answer a running step is writing, shown as it arrives and labelled +// provisional until the run's verified result replaces it. Mirrors the +// conversation live preview: withdrawn text is cleared, a gap is said +// rather than papered over, and the receipt decides the outcome. +export function AnswerPane({api,runId,stepId,onTerminal}:{api:ApiClient;runId:string;stepId:string;onTerminal:()=>void}){ + const [attempt,setAttempt]=useState(0),[phase,setPhase]=useState('connecting'),[text,setText]=useState(''),[missing,setMissing]=useState(false),[withdrawn,setWithdrawn]=useState(false),[retryable,setRetryable]=useState(true),[reason,setReason]=useState(''); + const cursor=useRef(0),received=useRef(''),bytes=useRef(0),owner=useRef(api),terminal=useRef(onTerminal);terminal.current=onTerminal; + useEffect(()=>{ + if(owner.current!==api){owner.current=api;cursor.current=0;received.current='';bytes.current=0;setText('');setMissing(false);setWithdrawn(false);} + const controller=new AbortController();let disposed=false,timer:ReturnType|undefined; + const heartbeat=()=>{clearTimeout(timer);timer=setTimeout(()=>controller.abort(),45000);}; + setPhase('connecting');setRetryable(true);setReason('');heartbeat(); + void(async()=>{ + try{ + const body=await api.answerStream(runId,stepId,cursor.current,controller.signal); + if(disposed){await body.cancel();return;} + for await(const event of readAnswerStream(body,cursor.current,heartbeat)){ + if(disposed)return; + if(event.kind==='text'){ + const nextBytes=bytes.current+new TextEncoder().encode(event.text).length; + if(nextBytes>MAX_PREVIEW_BYTES){setPhase('limited');return;} + cursor.current=event.sequence;bytes.current=nextBytes;received.current+=event.text;setText(received.current);setWithdrawn(false);setPhase('live'); + }else if(event.kind==='withdraw'){ + // What was written was not the answer: a tool turn followed. Clear it + // and say so, rather than leave it reading as the reply. + cursor.current=event.sequence;received.current='';bytes.current=0;setText('');setWithdrawn(true); + }else if(event.kind==='gap'){ + cursor.current=event.next-1;received.current='';bytes.current=0;setText('');setMissing(true); + }else if(event.kind==='terminal'){ + setPhase('waiting');terminal.current();return; + }else{setReason(event.reason);setPhase('ended');return;} + } + }catch(error){if(!disposed){ + setRetryable(!(error instanceof ApiError&&[401,403,404,409,422,501].includes(error.status))); + setPhase('interrupted'); + }}finally{clearTimeout(timer);controller.abort();} + })(); + return()=>{disposed=true;clearTimeout(timer);controller.abort();}; + // Not `onTerminal`: the run view hands a fresh function on every status + // poll, and reconnecting on each would open a stream per poll. + },[api,runId,stepId,attempt]); + const status=phase==='connecting'?'Connecting to the answer as it is written…':phase==='live'?'Receiving the answer as it is written':phase==='interrupted'?'Live answer interrupted. Checking the saved result.':phase==='limited'?'Preview size limit reached. The saved result carries the full answer.':phase==='ended'?(reason==='observation_window_ended'?'The observation window ended. Reconnect to keep following.':'The live answer is no longer available here. The saved result carries the answer.'):'The run has a receipt. Reading the verified result.'; + return
+
Live answer · {stepId}Provisional, not the verified result
+
{status}
+ {withdrawn&&!text&&

Text written before a tool call was withdrawn; it was not the answer.

} + {missing&&

Earlier live text is missing. Showing only the latest continuous segment.

} + {text&&owner.current===api&&
{text}
} +
Public text as the model writes it, never a token count or a tool argument. The verified result decides the outcome. + {(phase==='interrupted'&&retryable||phase==='ended'&&reason==='observation_window_ended')&&} +
+
; +} diff --git a/src/agents/RunPanel.tsx b/src/agents/RunPanel.tsx index 8155106..8b37eb3 100644 --- a/src/agents/RunPanel.tsx +++ b/src/agents/RunPanel.tsx @@ -8,6 +8,7 @@ import {isHandoffPlan,isInputTask,isInteractivePlan,type SavedPlan} from './plan import {matchRun,matchSubmission,parseRunPage,parseRunRequest,parseRunResult,parseRunStatus,runAddress,validateStart,type RunSubmission,type RunRequest,type RunStatus,type RunResult} from './runs'; import {InputPanel} from './InputPanel'; import {HistoryTimeline} from './HistoryTimeline'; +import {AnswerPane} from './AnswerPane'; import {matchInputResults,parseInputPage,type RunInput} from './inputs'; const secureRequest={cache:'no-store',redirect:'error',credentials:'omit',referrerPolicy:'no-referrer'} as const; function message(error:unknown):string{ @@ -19,7 +20,7 @@ function message(error:unknown):string{ } return error instanceof Error?error.message:'The run response is unavailable.'; } -function RunView({api,space,id,expected,onChanged,inputsAvailable,usageAvailable,approvalsAvailable,historyAvailable,approvalAttempt}:{api:ApiClient;space:string;id:string;expected?:RunSubmission;approvalAttempt:ApprovalAttemptRef;onChanged:()=>void;inputsAvailable:boolean;usageAvailable:boolean;approvalsAvailable:boolean;historyAvailable:boolean}){ +function RunView({api,space,id,expected,onChanged,inputsAvailable,usageAvailable,approvalsAvailable,historyAvailable,textStreamAvailable,approvalAttempt}:{api:ApiClient;space:string;id:string;expected?:RunSubmission;approvalAttempt:ApprovalAttemptRef;onChanged:()=>void;inputsAvailable:boolean;usageAvailable:boolean;approvalsAvailable:boolean;historyAvailable:boolean;textStreamAvailable:boolean}){ const [version,setVersion]=useState(0),[request,setRequest]=useState(null),[status,setStatus]=useState(null),[result,setResult]=useState(null),[issue,setIssue]=useState(''),[busy,setBusy]=useState(false),[cancelling,setCancelling]=useState(false); const [inputs,setInputs]=useState([]),[approvals,setApprovals]=useState[]>([]); const active=useRef(null); @@ -74,6 +75,7 @@ function RunView({api,space,id,expected,onChanged,inputsAvailable,usageAvailable {request&&status&&inputsAvailable&&inputs.length>0&&{setVersion(n=>n+1);onChanged();}}/>} {request&&status&&approvalsAvailable&&approvals.length>0&&{setVersion(n=>n+1);onChanged();}}/>} {status?.paused_steps.length&&!approvalsAvailable?

This run is paused for a tool decision. This server does not advertise the approval interface.

:null} + {request&&status&&textStreamAvailable&&!result&&(status.active_local||status.status==='running')&&status.inflight_steps.map(step=>setVersion(n=>n+1)}/>)} {result?.outcome==='handoff_limit'&&

The handoff limit was reached. These are partial results; no final answer was produced.

} {result&&
{result.tasks.map(output=>output.kind==='human_input'?

{output.task_id} · Human input

Reply used by this workflow

{output.text}
:

{output.task_id} · {output.agent_id} · {output.model_id}{result.finalTask===output.task_id?' · Final answer':''}

{output.handoff_to!==undefined&&

{output.handoff_to===null?'Agent finished':`Handed off to ${output.handoff_to}`}

}

{output.source_status==='retained'?`${output.evidence_ids.length} retained evidence references`:'No retained evidence · Treat this as ungrounded model output'} · {output.model_calls} model calls · {output.tool_calls} tool calls

{result.reusedTasks?.includes(output.task_id)&&

Saved task result reused

}{output.usage!==undefined&&}
{output.text}
)}
} {issue&&

{issue}

} @@ -94,7 +96,7 @@ function RunHistory({api,space,version,onSelect}:{api:ApiClient;space:string;ver useEffect(()=>{const controller=new AbortController();active.current=controller;setItems([]);setAfter(null);void load(controller,null);return()=>controller.abort();},[api,space,version]); return

Run history in {space}

    {items.map(run=>
    • )}
{!busy&&!items.length&&!issue&&

No runs saved yet.

}{busy&&

Loading runs…

}{after&&}{issue&&

{issue}

}
; } -export function RunPanel({api,space,plan,dirty,maxParallel,handoffsAvailable,inputsAvailable,usageAvailable,approvalsAvailable,historyAvailable=false}:{api:ApiClient;space:string;plan:SavedPlan|null;dirty:boolean;maxParallel:number;handoffsAvailable:boolean;inputsAvailable:boolean;usageAvailable:boolean;approvalsAvailable:boolean;historyAvailable?:boolean}){ +export function RunPanel({api,space,plan,dirty,maxParallel,handoffsAvailable,inputsAvailable,usageAvailable,approvalsAvailable,historyAvailable=false,textStreamAvailable=false}:{api:ApiClient;space:string;plan:SavedPlan|null;dirty:boolean;maxParallel:number;handoffsAvailable:boolean;inputsAvailable:boolean;usageAvailable:boolean;approvalsAvailable:boolean;historyAvailable?:boolean;textStreamAvailable?:boolean}){ const [runId,setRunId]=useState(()=>crypto.randomUUID()),[parallel,setParallel]=useState(1),[question,setQuestion]=useState(''),[busy,setBusy]=useState(false),[attempted,setAttempted]=useState(false),[issue,setIssue]=useState(''),[submitted,setSubmitted]=useState(null),[selected,setSelected]=useState<{id:string;version:number;expected?:RunSubmission}|null>(null),[history,setHistory]=useState(0),[lookup,setLookup]=useState(''); const active=useRef(null); const approvalAttempts=useRef({api,space,items:new Map()}); @@ -120,7 +122,7 @@ export function RunPanel({api,space,plan,dirty,maxParallel,handoffsAvailable,inp {attempted&&
A new run calls the selected models again.
} {issue&&

{issue}

}
{event.preventDefault();try{inspect(lookup);setIssue('');}catch(error){setIssue(message(error));}}}>
- {selected&&setHistory(n=>n+1)}/>} + {selected&&setHistory(n=>n+1)}/>} ; } diff --git a/src/agents/agents.css b/src/agents/agents.css index 285c905..42b0023 100644 --- a/src/agents/agents.css +++ b/src/agents/agents.css @@ -27,3 +27,10 @@ .agent-timeline-entries{list-style:none;padding:0;margin:.5rem 0;display:grid;gap:.4rem} .agent-timeline-entries li{padding:.35rem .5rem;border-left:3px solid var(--accent,#7a5c2e);background:var(--surface-2,rgba(0,0,0,.03))} .agent-timeline-position{font-variant-numeric:tabular-nums;opacity:.7;margin-right:.25rem} + +.agent-answer-live{margin:.75rem 0;border:1px solid var(--line,#d8d3c8);border-radius:.5rem;padding:.6rem .75rem;background:var(--surface-2,rgba(0,0,0,.03))} +.agent-answer-live header{display:flex;justify-content:space-between;gap:.5rem;font-size:.9rem;opacity:.85} +.agent-answer-badge{font-weight:600} +.agent-answer-status{font-size:.9rem;margin:.35rem 0} +.agent-answer-text{white-space:pre-wrap} +.agent-answer-live footer{display:flex;justify-content:space-between;align-items:center;gap:.5rem;font-size:.8rem;opacity:.8;margin-top:.4rem} diff --git a/src/agents/answer-stream.ts b/src/agents/answer-stream.ts new file mode 100644 index 0000000..ea95a93 --- /dev/null +++ b/src/agents/answer-stream.ts @@ -0,0 +1,84 @@ +// The answer a running step is writing, read as events off the host's +// text/event-stream route. Provisional by design: the terminal says the +// run has a receipt, and the verified result is what /result returns. +// Sequences are contiguous from the cursor -- a gap is the only sanctioned +// jump -- and a frame's id must name its sequence, because it is what a +// reconnect sends back. Nothing but content is ever in these frames, and a +// frame carrying more than its own fields is refused. +import {runAddress} from './runs.ts'; + +export type AnswerEvent={kind:'text';sequence:number;text:string}|{kind:'withdraw';sequence:number} + |{kind:'gap';after:number;next:number}|{kind:'terminal';status:string}|{kind:'end';reason:string}; +const MAX_FRAME=1_048_576; +const MAX_SEQUENCE=Number.MAX_SAFE_INTEGER; + +function invalid(what:string):never{throw Error('The answer stream could not be verified: '+what+'.');} +function sequence(value:unknown):number{if(typeof value!=='number'||!Number.isSafeInteger(value)||value<1||value>MAX_SEQUENCE)invalid('sequence');return value;} +function exact(row:Record,fields:string[]):void{const keys=Object.keys(row);if(keys.length!==fields.length||fields.some(field=>!Object.hasOwn(row,field)))invalid('frame fields');} +export function answerAddress(runId:string,stepId:string):string{ + if(!/^[A-Za-z0-9._:-]{1,128}$/.test(stepId)||stepId==='.'||stepId==='..')throw Error('Use letters, numbers, dots, colons, underscores or hyphens for identifiers.'); + return runAddress(runId)+'/steps/'+encodeURIComponent(stepId)+'/text/stream'; +} + +export async function* readAnswerStream(body:ReadableStream,after:number,onActivity=()=>{}):AsyncGenerator{ + const reader=body.getReader(),decoder=new TextDecoder('utf-8',{fatal:true}); + let pending='',event='',id='',data:string[]=[],size=0,cursor=after; + const sequenced=(row:Record):number=>{ + const value=sequence(row.sequence); + if(value!==cursor+1)invalid('sequence'); + if(id!==String(value))invalid('frame id does not name its sequence'); + return value; + }; + try{ + while(true){ + const {value,done}=await reader.read(); + if(done){decoder.decode();throw Error('The answer stream ended before the host ended it.');} + onActivity(); + const decoded=decoder.decode(value,{stream:true}); + let offset=0; + while(offsetMAX_FRAME)throw Error('The answer stream frame exceeds its limit.'); + if(newline<0)break; + offset=newline+1;size++; + const line=pending.endsWith('\r')?pending.slice(0,-1):pending;pending=''; + if(line===''){ + if(data.length){ + const wire:unknown=JSON.parse(data.join('\n')); + if(!wire||typeof wire!=='object'||Array.isArray(wire))invalid('frame'); + const row=wire as Record; + if(event==='error'){throw Error('The answer stream was refused: '+(typeof row.reason==='string'?row.reason:'unknown')+'.');} + if(event==='text'){ + exact(row,['sequence','text']);const value=sequenced(row); + if(typeof row.text!=='string'||!row.text.length)invalid('text'); + cursor=value;yield {kind:'text',sequence:value,text:row.text}; + }else if(event==='withdraw'){ + exact(row,['sequence']);const value=sequenced(row);cursor=value;yield {kind:'withdraw',sequence:value}; + }else if(event==='gap'){ + exact(row,['after','next_sequence']); + if(id||row.after!==cursor)invalid('gap'); + const next=sequence(row.next_sequence);if(next<=cursor+1)invalid('gap'); + const previous=cursor;cursor=next-1;yield {kind:'gap',after:previous,next}; + }else if(event==='terminal'){ + exact(row,['status','read_receipt']); + if(id||row.read_receipt!==true||typeof row.status!=='string'||!row.status)invalid('terminal without a read_receipt'); + yield {kind:'terminal',status:row.status};return; + }else if(event==='end'){ + exact(row,['reason']); + if(id||typeof row.reason!=='string'||!row.reason)invalid('end'); + yield {kind:'end',reason:row.reason};return; + }else invalid('frame kind'); + } + event='';id='';data=[];size=0; + }else if(!line.startsWith(':')){ + const colon=line.indexOf(':'),field=colon<0?line:line.slice(0,colon); + let content=colon<0?'':line.slice(colon+1);if(content.startsWith(' '))content=content.slice(1); + if(field==='event')event=content; + else if(field==='id')id=content; + else if(field==='data')data.push(content); + } + } + } + }finally{try{await reader.cancel();}finally{reader.releaseLock();}} +} diff --git a/src/api.ts b/src/api.ts index de09b54..938a7d7 100644 --- a/src/api.ts +++ b/src/api.ts @@ -3,6 +3,7 @@ import {readVideoFrame,validateVideoFrameReference,type VideoFrameReference} fro import {exportAddress,readGraphExport,type GraphExportRequest,type GraphExportFile} from './memory/knowledge-export.ts'; import {attachVoiceSocket,type VoiceEvents} from './conversations/voice/channel.ts'; import {historyAddress,historyCursor} from './agents/history.ts'; +import {answerAddress} from './agents/answer-stream.ts'; import {voiceSocketUrl} from './conversations/voice/wire.ts'; export interface ApiClient { @@ -16,6 +17,7 @@ export interface ApiClient { documentOriginal(original:ImageAttachment,signal?:AbortSignal):Promise; conversationStream(sid: string, requestId: string, after: number, signal: AbortSignal): Promise>; historyStream(runId:string,after:string|null,limit:number,signal:AbortSignal):Promise>; + answerStream(runId:string,stepId:string,after:number,signal:AbortSignal):Promise>; voiceConnection(sid:string,format:{sampleRate:number;channels:number},events:VoiceEvents,signal:AbortSignal):ReturnType; } @@ -56,6 +58,23 @@ export function createApiClient(key: string, unauthorized: () => void, base = '' const url=voiceSocketUrl(sid,base||window.location.origin); return attachVoiceSocket(new WebSocket(url),key,sid,format,events,signal); }, + async answerStream(runId,stepId,after,signal){ + if(!Number.isSafeInteger(after)||after<0)throw Error('Invalid answer stream cursor'); + const path=answerAddress(runId,stepId)+(after?'?after='+after:''); + const response=await fetch(base+path,{ + method:'GET',headers:{Authorization:'Bearer '+key,Accept:'text/event-stream',...(after?{'Last-Event-ID':String(after)}:{})},signal, + cache:'no-store',redirect:'error',credentials:'omit',referrerPolicy:'no-referrer', + }); + if(!response.ok){ + await response.body?.cancel(); + if(response.status===401)unauthorized(); + throw new ApiError(response.status,`Answer stream request failed (${response.status})`); + } + if(response.headers.get('content-type')?.split(';')[0].trim()!=='text/event-stream'||!response.body){ + await response.body?.cancel();throw Error('Invalid answer stream response'); + } + return response.body; + }, async historyStream(runId,after,limit,signal){ if(after!==null)historyCursor(after); if(!Number.isSafeInteger(limit)||limit<1||limit>100)throw Error('Invalid history page size'); diff --git a/src/capabilities.ts b/src/capabilities.ts index 1cbd08e..2839fd6 100644 --- a/src/capabilities.ts +++ b/src/capabilities.ts @@ -1,7 +1,7 @@ export const FEATURE_KEYS = ['recall', 'facts.read', 'facts.review', 'facts.close', 'facts.exclude', 'facts.include', 'events.read', 'metrics.read', 'scopes.read', 'status.read'] as const; export type Feature = typeof FEATURE_KEYS[number]; -type OptionalFeature = 'agents.history' | 'agents.handoffs.output_requirements' | 'agents.output_requirements' | 'agents.output_schema' | 'agents.approvals' | 'agents.usage' | 'documents.video.understand' | 'documents.sync' | 'agents.inputs' | 'recall.parts' | 'agents.handoffs' | 'agents.parallel' | 'agents.runs' | 'agents.catalog' | 'agents.plans' | 'episodes.forget' | 'documents.ocr.tables' | 'documents.jobs' | 'documents.files' | 'graph.knowledge_walk' | 'documents.provenance' | 'recall.graph_boost' | 'graph.knowledge_seeds' | 'graph.knowledge_paging' | 'graph.timeline' | 'graph.sources' | 'graph.export' | 'graph.path' | 'graph.knowledge' | 'entities.read' | 'graph.report' | 'images.understand' | 'models.manage' | 'episodes.attachments' | 'episodes.list' | 'episodes.read' | 'facts.links' | 'integrity.read' | 'profile.read' | 'processing.distill' | 'processing.derive' | 'jobs.read' | 'recall.conditions'; +type OptionalFeature = 'agents.text_stream' | 'agents.history' | 'agents.handoffs.output_requirements' | 'agents.output_requirements' | 'agents.output_schema' | 'agents.approvals' | 'agents.usage' | 'documents.video.understand' | 'documents.sync' | 'agents.inputs' | 'recall.parts' | 'agents.handoffs' | 'agents.parallel' | 'agents.runs' | 'agents.catalog' | 'agents.plans' | 'episodes.forget' | 'documents.ocr.tables' | 'documents.jobs' | 'documents.files' | 'graph.knowledge_walk' | 'documents.provenance' | 'recall.graph_boost' | 'graph.knowledge_seeds' | 'graph.knowledge_paging' | 'graph.timeline' | 'graph.sources' | 'graph.export' | 'graph.path' | 'graph.knowledge' | 'entities.read' | 'graph.report' | 'images.understand' | 'models.manage' | 'episodes.attachments' | 'episodes.list' | 'episodes.read' | 'facts.links' | 'integrity.read' | 'profile.read' | 'processing.distill' | 'processing.derive' | 'jobs.read' | 'recall.conditions'; export interface Capabilities { schema_version: 1; implementation: string; @@ -14,7 +14,7 @@ export function parseCapabilities(value: unknown): Capabilities { if (data.schema_version !== 1 || typeof data.implementation !== 'string' || !data.implementation.trim() || !data.features || typeof data.features !== 'object' || Array.isArray(data.features)) throw Error('Unsupported capability response'); const features = data.features as Record; - for(const key of ['agents.history','agents.approvals','agents.handoffs.output_requirements','agents.output_requirements','agents.output_schema','agents.usage','documents.video.understand','documents.sync','documents.ocr.tables','agents.inputs','recall.parts','agents.handoffs','agents.parallel','agents.runs','agents.catalog','agents.plans','episodes.forget','documents.jobs','documents.files','documents.provenance','processing.distill','processing.derive','graph.knowledge','entities.read','graph.report','graph.path','graph.export','graph.sources','graph.timeline','graph.knowledge_paging','graph.knowledge_seeds','graph.knowledge_walk','recall.graph_boost']){ + for(const key of ['agents.text_stream','agents.history','agents.approvals','agents.handoffs.output_requirements','agents.output_requirements','agents.output_schema','agents.usage','documents.video.understand','documents.sync','documents.ocr.tables','agents.inputs','recall.parts','agents.handoffs','agents.parallel','agents.runs','agents.catalog','agents.plans','episodes.forget','documents.jobs','documents.files','documents.provenance','processing.distill','processing.derive','graph.knowledge','entities.read','graph.report','graph.path','graph.export','graph.sources','graph.timeline','graph.knowledge_paging','graph.knowledge_seeds','graph.knowledge_walk','recall.graph_boost']){ if(features[key]!==undefined&&typeof features[key]!=='boolean')throw Error('Invalid processing capability flag'); } if (FEATURE_KEYS.some(key => typeof features[key] !== 'boolean')) throw Error('Incomplete or invalid capability flags'); @@ -28,5 +28,5 @@ export function parseCapabilities(value: unknown): Capabilities { if(features['recall.conditions']!==undefined&&typeof features['recall.conditions']!=='boolean')throw Error('Invalid recall conditions capability'); if(features['images.understand']!==undefined&&typeof features['images.understand']!=='boolean')throw Error('Invalid image understanding capability'); if(features['models.manage']!==undefined&&typeof features['models.manage']!=='boolean')throw Error('Invalid model management capability'); - return {schema_version:1, implementation:data.implementation, features:{'agents.history':features['agents.history']===true,'agents.approvals':features['agents.approvals']===true,'agents.handoffs.output_requirements':features['agents.handoffs.output_requirements']===true,'agents.output_requirements':features['agents.output_requirements']===true,'agents.output_schema':features['agents.output_schema']===true,'agents.usage':features['agents.usage']===true,'documents.video.understand':features['documents.video.understand']===true,'documents.sync':features['documents.sync']===true,'documents.ocr.tables':features['documents.ocr.tables']===true,'agents.inputs':features['agents.inputs']===true,'recall.parts':features['recall.parts']===true,'agents.handoffs':features['agents.handoffs']===true,'agents.parallel':features['agents.parallel']===true,'agents.runs':features['agents.runs']===true,'agents.catalog':features['agents.catalog']===true,'agents.plans':features['agents.plans']===true,'episodes.forget':features['episodes.forget']===true,'documents.jobs':features['documents.jobs']===true,'documents.files':features['documents.files']===true,'graph.knowledge_walk':features['graph.knowledge_walk']===true,'documents.provenance':features['documents.provenance']===true,'recall.graph_boost':features['recall.graph_boost']===true,'graph.knowledge_seeds':features['graph.knowledge_seeds']===true,'graph.knowledge_paging':features['graph.knowledge_paging']===true,'graph.timeline':features['graph.timeline']===true,'graph.sources':features['graph.sources']===true,'graph.export':features['graph.export']===true,'graph.path':features['graph.path']===true,'graph.report':features['graph.report']===true,'graph.knowledge':features['graph.knowledge']===true,'entities.read':features['entities.read']===true,...Object.fromEntries(FEATURE_KEYS.map(key => [key, features[key]])), 'episodes.attachments':features['episodes.attachments'] === true, 'episodes.list':features['episodes.list'] === true, 'episodes.read':features['episodes.read'] === true, 'facts.links':features['facts.links'] === true, 'integrity.read':features['integrity.read'] === true, 'profile.read':features['profile.read'] === true, 'processing.distill':features['processing.distill']===true, 'processing.derive':features['processing.derive']===true, 'jobs.read':features['jobs.read']===true, 'recall.conditions':features['recall.conditions']===true, 'models.manage':features['models.manage']===true, 'images.understand':features['images.understand']===true} as Capabilities['features']}; + return {schema_version:1, implementation:data.implementation, features:{'agents.text_stream':features['agents.text_stream']===true,'agents.history':features['agents.history']===true,'agents.approvals':features['agents.approvals']===true,'agents.handoffs.output_requirements':features['agents.handoffs.output_requirements']===true,'agents.output_requirements':features['agents.output_requirements']===true,'agents.output_schema':features['agents.output_schema']===true,'agents.usage':features['agents.usage']===true,'documents.video.understand':features['documents.video.understand']===true,'documents.sync':features['documents.sync']===true,'documents.ocr.tables':features['documents.ocr.tables']===true,'agents.inputs':features['agents.inputs']===true,'recall.parts':features['recall.parts']===true,'agents.handoffs':features['agents.handoffs']===true,'agents.parallel':features['agents.parallel']===true,'agents.runs':features['agents.runs']===true,'agents.catalog':features['agents.catalog']===true,'agents.plans':features['agents.plans']===true,'episodes.forget':features['episodes.forget']===true,'documents.jobs':features['documents.jobs']===true,'documents.files':features['documents.files']===true,'graph.knowledge_walk':features['graph.knowledge_walk']===true,'documents.provenance':features['documents.provenance']===true,'recall.graph_boost':features['recall.graph_boost']===true,'graph.knowledge_seeds':features['graph.knowledge_seeds']===true,'graph.knowledge_paging':features['graph.knowledge_paging']===true,'graph.timeline':features['graph.timeline']===true,'graph.sources':features['graph.sources']===true,'graph.export':features['graph.export']===true,'graph.path':features['graph.path']===true,'graph.report':features['graph.report']===true,'graph.knowledge':features['graph.knowledge']===true,'entities.read':features['entities.read']===true,...Object.fromEntries(FEATURE_KEYS.map(key => [key, features[key]])), 'episodes.attachments':features['episodes.attachments'] === true, 'episodes.list':features['episodes.list'] === true, 'episodes.read':features['episodes.read'] === true, 'facts.links':features['facts.links'] === true, 'integrity.read':features['integrity.read'] === true, 'profile.read':features['profile.read'] === true, 'processing.distill':features['processing.distill']===true, 'processing.derive':features['processing.derive']===true, 'jobs.read':features['jobs.read']===true, 'recall.conditions':features['recall.conditions']===true, 'models.manage':features['models.manage']===true, 'images.understand':features['images.understand']===true} as Capabilities['features']}; } diff --git a/tests/agent-answer.test.ts b/tests/agent-answer.test.ts new file mode 100644 index 0000000..20125f0 --- /dev/null +++ b/tests/agent-answer.test.ts @@ -0,0 +1,54 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {readAnswerStream,answerAddress,type AnswerEvent} from '../src/agents/answer-stream.ts'; +import {parseCapabilities} from '../src/capabilities.ts'; +import {readFileSync} from 'node:fs'; +const fixtures=JSON.parse(readFileSync(new URL('./fixtures/http-capabilities.json',import.meta.url),'utf8')) as {python:{features:Record}}; + +const frame=(event:string,data:unknown,id?:number)=>`event: ${event}\n${id===undefined?'':'id: '+id+'\n'}data: ${JSON.stringify(data)}\n\n`; +const text=(sequence:number,value:string)=>frame('text',{sequence,text:value},sequence); +function stream(...parts:string[]):ReadableStream{ + const encoder=new TextEncoder(); + return new ReadableStream({start(controller){for(const part of parts)controller.enqueue(encoder.encode(part));controller.close();}}); +} +async function collect(body:ReadableStream,after=0):Promise{ + const events:AnswerEvent[]=[]; + for await(const event of readAnswerStream(body,after))events.push(event); + return events; +} + +test('text arrives in sequence and a terminal points at the receipt',async()=>{ + const events=await collect(stream(': keep-alive\n\n',text(1,'We '),text(2,'decided.'),frame('terminal',{status:'completed',read_receipt:true}))); + assert.deepEqual(events,[{kind:'text',sequence:1,text:'We '},{kind:'text',sequence:2,text:'decided.'},{kind:'terminal',status:'completed'}]); + assert.equal(answerAddress('run-1','find'),'/v1/agent-runs/run-1/steps/find/text/stream'); +}); + +test('withdraw, gap and end are events the pane can act on',async()=>{ + const events=await collect(stream(text(1,'Looking that up.'),frame('withdraw',{sequence:2},2),frame('gap',{after:2,next_sequence:7}),text(7,'We decided.'),frame('end',{reason:'observation_window_ended'}))); + assert.deepEqual(events.map(event=>event.kind),['text','withdraw','gap','text','end']); + assert.deepEqual(events[2],{kind:'gap',after:2,next:7}); + assert.deepEqual(events[4],{kind:'end',reason:'observation_window_ended'}); +}); + +test('a resumed reader requires continuity with its cursor',async()=>{ + assert.deepEqual((await collect(stream(text(4,'more'),frame('end',{reason:'observation_window_ended'})),3)).map(event=>event.kind),['text','end']); + await assert.rejects(collect(stream(text(5,'skipped'),frame('end',{reason:'observation_window_ended'})),3),/sequence/i); +}); + +for(const [label,body,match] of [ + ['an id that does not name its sequence',frame('text',{sequence:1,text:'x'},2),/id/i], + ['a repeated sequence',text(1,'x')+text(1,'again'),/sequence/i], + ['an error frame',frame('error',{reason:'text_unavailable'}),/text_unavailable/], + ['a stream cut mid-frame',text(1,'x')+'event: text\nid: 2\n',/ended/i], + ['a kind this pane does not read',frame('other',{sequence:1},1),/kind/i], + ['a frame carrying more than its fields',frame('text',{sequence:1,text:'x',reasoning:'private'},1),/fields/i], + ['a terminal without a receipt',frame('terminal',{status:'completed',read_receipt:false}),/receipt/i], + ['a gap that does not follow the cursor',text(1,'x')+frame('gap',{after:0,next_sequence:5}),/gap/i], +] as const)test(`refused: ${label}`,async()=>{await assert.rejects(collect(stream(body)),match);}); + +test('the text stream capability is read like the others',()=>{ + const features={...fixtures.python.features};delete (features as Record)['agents.text_stream']; + assert.equal(parseCapabilities({...fixtures.python,features}).features['agents.text_stream'],false); + assert.equal(parseCapabilities({...fixtures.python,features:{...features,'agents.text_stream':true}}).features['agents.text_stream'],true); + assert.throws(()=>parseCapabilities({...fixtures.python,features:{...features,'agents.text_stream':'yes'}})); +}); diff --git a/tests/api.test.ts b/tests/api.test.ts index b018c3e..0187864 100644 --- a/tests/api.test.ts +++ b/tests/api.test.ts @@ -182,3 +182,30 @@ test('history streams send the cursor both ways and refuse non-SSE responses, re status=401;await assert.rejects(client.historyStream('run-1',null,50,controller.signal),e=>e instanceof ApiError&&e.status===401);assert.equal(denied,true); }finally{server.closeAllConnections();await new Promise(resolve=>server.close(()=>resolve()));} }); + +test('answer streams send the cursor both ways and refuse non-SSE responses, redirects and bad addresses',async()=>{ + let status=200,type='text/event-stream; charset=utf-8',denied=false,calls=0,lastEventId:string|undefined,url=''; + const server=createServer((req,res)=>{ + calls++;url=req.url??'';assert.equal(req.method,'GET');assert.equal(req.headers.authorization,'Bearer allowed');assert.equal(req.headers.accept,'text/event-stream'); + lastEventId=req.headers['last-event-id'] as string|undefined; + res.writeHead(status,{'content-type':type,location:'/unexpected'});res.end(': keep-alive\n\n'); + }); + await new Promise(resolve=>server.listen(0,'127.0.0.1',resolve)); + const address=server.address();if(!address||typeof address==='string')throw Error('missing address'); + const client=createApiClient('allowed',()=>{denied=true;},`http://127.0.0.1:${address.port}`); + const controller=new AbortController(); + try{ + const stream=await client.answerStream('run-1','find',3,controller.signal); + assert.equal(await new Response(stream).text(),': keep-alive\n\n'); + assert.equal(url,'/v1/agent-runs/run-1/steps/find/text/stream?after=3');assert.equal(lastEventId,'3'); + await client.answerStream('run-1','find',0,controller.signal); + assert.equal(url,'/v1/agent-runs/run-1/steps/find/text/stream');assert.equal(lastEventId,undefined); + for(const bad of ['../secret','..','with/slash'])await assert.rejects(client.answerStream(bad,'find',0,controller.signal)); + await assert.rejects(client.answerStream('run-1','../x',0,controller.signal)); + await assert.rejects(client.answerStream('run-1','find',-1,controller.signal),/cursor/i); + assert.equal(calls,2); + type='application/json';await assert.rejects(client.answerStream('run-1','find',0,controller.signal),/stream/i); + type='text/event-stream';status=302;await assert.rejects(client.answerStream('run-1','find',0,controller.signal)); + status=401;await assert.rejects(client.answerStream('run-1','find',0,controller.signal),e=>e instanceof ApiError&&e.status===401);assert.equal(denied,true); + }finally{server.closeAllConnections();await new Promise(resolve=>server.close(()=>resolve()));} +});