From 2b6be521e7b991008dd7dc1268cde1229ad5516c Mon Sep 17 00:00:00 2001 From: Mark Sturman Date: Sun, 13 Sep 2026 15:06:56 -0500 Subject: [PATCH] Show the answer a running step is writing, provisional until the receipt When the host advertises agents.text_stream, each step a run is working on shows a Live answer pane: the answer as the model writes it, read off the host's text/event-stream route with the same rules as the SDK -- sequences contiguous from the cursor, an id naming its sequence, a gap the only sanctioned jump, foreign fields refused, a terminal only with a read receipt. Text written before a tool call is withdrawn and the pane says so; a gap is said rather than papered over; 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. It does not depend on the run view's callback identity, so a status poll cannot reopen the stream. Fourteen unit tests over the reader and capability, an API client test for the request shape, and four browser journeys: the pane arrives and yields to the verified result on desktop and mobile, withdrawn text is cleared and said, and no pane without the capability. 32 native journeys and 523 unit tests pass; typecheck and build are clean. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_012aNJobunWMPFEa8mjsFQqY --- README.md | 9 ++++ scripts/test-agents.cjs | 44 ++++++++++++++++++- src/agents/AgentsPage.tsx | 6 +-- src/agents/AnswerPane.tsx | 59 ++++++++++++++++++++++++++ src/agents/RunPanel.tsx | 8 ++-- src/agents/agents.css | 7 ++++ src/agents/answer-stream.ts | 84 +++++++++++++++++++++++++++++++++++++ src/api.ts | 19 +++++++++ src/capabilities.ts | 6 +-- tests/agent-answer.test.ts | 54 ++++++++++++++++++++++++ tests/api.test.ts | 27 ++++++++++++ 11 files changed, 312 insertions(+), 11 deletions(-) create mode 100644 src/agents/AnswerPane.tsx create mode 100644 src/agents/answer-stream.ts create mode 100644 tests/agent-answer.test.ts 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()));} +});