Show the answer a running step is writing, provisional until the receipt

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.

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();

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<unknown>('/v1/agents/catalog',options).then(parseCatalog),api.request<unknown>('/v1/agent-plans?limit=20',options).then(value=>parsePlanPage(value,status.space as string))]);
    const maxParallel=caps.features['agents.parallel']?parseRunPolicy(await api.request<unknown>('/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?<HandoffEditor key={selection.id} {...props}/>:<p role="alert">This server does not support handoff configuration.</p>:selection.plan&&isInteractivePlan(selection.plan.plan)&&!data.inputsAvailable?<p role="alert">This server does not support human input workflows.</p>:<Editor key={selection.id} {...props}/>;
-   })()}</div>{data.runsAvailable&&<RunPanel key={data.space} api={api} space={data.space} historyAvailable={data.historyAvailable} plan={selection.plan} dirty={dirty} maxParallel={data.maxParallel} handoffsAvailable={data.handoffsAvailable} inputsAvailable={data.inputsAvailable} usageAvailable={data.usageAvailable} approvalsAvailable={data.approvalsAvailable}/>} {error&&<p role="alert">{error}</p>}
+   })()}</div>{data.runsAvailable&&<RunPanel key={data.space} api={api} space={data.space} historyAvailable={data.historyAvailable} textStreamAvailable={data.textStreamAvailable} plan={selection.plan} dirty={dirty} maxParallel={data.maxParallel} handoffsAvailable={data.handoffsAvailable} inputsAvailable={data.inputsAvailable} usageAvailable={data.usageAvailable} approvalsAvailable={data.approvalsAvailable}/>} {error&&<p role="alert">{error}</p>}
   </>}
  </main>;
 }

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<Phase>('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<typeof setTimeout>|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 <section className="agent-answer-live" aria-label={`Answer being written for ${stepId}`}>
+  <header><span className="agent-answer-badge"><span aria-hidden="true">◉</span> Live answer · {stepId}</span><span>Provisional, not the verified result</span></header>
+  <div className="agent-answer-status" role="status">{status}</div>
+  {withdrawn&&!text&&<p className="agent-notice">Text written before a tool call was withdrawn; it was not the answer.</p>}
+  {missing&&<p className="agent-notice">Earlier live text is missing. Showing only the latest continuous segment.</p>}
+  {text&&owner.current===api&&<div className="agent-output agent-answer-text" tabIndex={0} aria-label={`Provisional answer text for ${stepId}`}>{text}</div>}
+  <footer><span>Public text as the model writes it, never a token count or a tool argument. The verified result decides the outcome.</span>
+   {(phase==='interrupted'&&retryable||phase==='ended'&&reason==='observation_window_ended')&&<button type="button" onClick={()=>setAttempt(n=>n+1)}>Reconnect</button>}
+  </footer>
+ </section>;
+}