v51: stream owned inference and expose evidence-based commissioning

Author: Mantiee

V100_AUDIT.md

@@ -535,3 +535,29 @@ passed after replacing missing ML counters with explicitly unknown values. Tests
 cover archive pagination, goal attribution, incomplete tails, path/symlink rejection,
 HTTP action routes and goal-change separation while preserving verifiable replay.
 No hardware training improvement or universal goal oracle is certified.
+
+## v51: observable streaming and commissioning evidence
+
+Owned llama.cpp SSE and Ollama NDJSON transports now journal partial output and
+backend-returned local reasoning while generation is in progress. Both capture and
+stream preferences must be enabled; the upgrade enables these following the
+operator's request. Thinking modes remain unchanged. Deltas are flushed in bounded
+chunks; a lightweight read-only endpoint feeds the dashboard every two seconds
+without waiting for full hardware collection. Final token counts remain measured
+backend counters, not estimates from text. Incomplete streams are visible as
+partial output but never accepted as completed responses. Historical omitted
+reasoning, unrecorded actions and token IDs cannot be reconstructed.
+
+The readiness endpoint and dashboard distinguish worker heartbeat, reported job
+execution, verified optimizer counters, accepted weights, resolved goal outcomes,
+layout publication and observed stream transport. Unknown evidence is explicitly
+unknown. This does not bypass independent acceptance or force poor candidates into
+production. A ready worker does not prove execution, and busy inference does not
+prove training. Hardware commissioning still requires the installed user backends.
+
+Validation: full deterministic suite 1000 passed, 65 skipped, one resource-dependent
+owned-worker integration deselected and one existing PEFT warning. After adding a
+real framed-transport mock, focused streaming/helper/dashboard coverage passed
+115 tests. HTTP route checks also exercise the fast inference and readiness APIs.
+No user GPU, Windows workload or improved accepted production model was available
+for direct validation in this environment.

pyproject.toml

@@ -3,7 +3,7 @@ name = "rlms"
 authors = [
     {name = "Alex Zhang", email = "altzhang@mit.edu"},
 ]
-version = "0.1.3+v100.50"
+version = "0.1.3+v100.51"
 description = "Recursive Language Models."
 readme = "README.md"
 license = "MIT"

rlm/clients/llamacpp.py

@@ -77,9 +77,13 @@ def request(self, endpoint: str, data: dict[str, Any] | None = None) -> dict[str
             **self.activity_context,
         )
         started = time.perf_counter()
+        self.thread_state.activity_request_id = request_id
+        self.thread_state.stream_buffer = {}
         try:
             result = self.http_request(endpoint, data)
+            self.flush_stream()
         except Exception as error:
+            self.flush_stream()
             journal.write(
                 "errors", "inference-failed", {"error": type(error).__name__}, request_id=request_id
             )
@@ -134,6 +138,48 @@ def request(self, endpoint: str, data: dict[str, Any] | None = None) -> dict[str
         )
         return result
 
+    def streaming_enabled(self) -> bool:
+        if not self.activity_root:
+            return False
+        path = self.activity_root / "research/user-preferences.json"
+        if not path.exists():
+            return False
+        preferences = json.loads(path.read_text())
+        return (
+            preferences.get("stream_local_model_trace") is True
+            and preferences.get("capture_local_model_trace") is True
+        )
+
+    def stream_delta(self, channel: str, text: str) -> None:
+        self.remaining_timeout()
+        buffer = getattr(self.thread_state, "stream_buffer", {})
+        buffer[channel] = buffer.get(channel, "") + text
+        self.thread_state.stream_buffer = buffer
+        last = getattr(self.thread_state, "stream_flushed", 0)
+        if time.monotonic() - last >= 0.5 or len(buffer[channel]) >= 1024:
+            self.flush_stream()
+
+    def flush_stream(self) -> None:
+        from rlm.v100.activity import ActivityLog
+
+        journal = ActivityLog(self.activity_root, self.research_owner, self.activity_actor)
+        for channel, text in getattr(self.thread_state, "stream_buffer", {}).items():
+            if text:
+                journal.write(
+                    "steps",
+                    "inference-delta",
+                    {
+                        "channel": channel,
+                        "text": text,
+                        "model": self.model_name,
+                        "device": self.activity_context.get("device"),
+                        "partial": True,
+                    },
+                    request_id=getattr(self.thread_state, "activity_request_id", None),
+                )
+        self.thread_state.stream_buffer = {}
+        self.thread_state.stream_flushed = time.monotonic()
+
     def remaining_timeout(self) -> float:
         deadline = getattr(self, "request_deadline", None)
         if deadline is None:
@@ -145,6 +191,19 @@ def remaining_timeout(self) -> float:
 
     def http_request(self, endpoint: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
         # No environment proxy routing for local model traffic.
+        if endpoint == "/v1/chat/completions" and data is not None and self.streaming_enabled():
+            from rlm.v100.streaming import local_reply
+
+            with requests.Session() as session:
+                session.trust_env = False
+                with session.post(
+                    self.base_url + endpoint,
+                    json={**data, "stream": True, "stream_options": {"include_usage": True}},
+                    timeout=self.remaining_timeout(),
+                    stream=True,
+                ) as response:
+                    response.raise_for_status()
+                    return local_reply(response.iter_lines(chunk_size=128), self.stream_delta)
         with requests.Session() as session:
             session.trust_env = False
             response = (

rlm/v100/dashboard_layout.py

@@ -24,7 +24,7 @@
 <details><summary>Dowody treningu i ostatnie błędy</summary><pre id="evidence"></pre></details>
 <script>
 const $=id=>document.getElementById(id), fmt=x=>x===null||x===undefined?'brak potwierdzonych danych':String(x), pct=x=>typeof x==='number'&&Number.isFinite(x)?(x*100).toFixed(2)+'%':'brak danych';
-const layoutVersion='__LAYOUT_SHA__';
+const layoutVersion='__LAYOUT_SHA__';let latestSnapshot=null;
 function card(label,value){const node=document.createElement('div');node.className='card';const title=document.createElement('small');title.textContent=label;const strong=document.createElement('strong');strong.className='value';strong.textContent=fmt(value);node.append(title,strong);$('cards').append(node)}
 function workView(data){
 for(const id of ['actors','goals','events']){if(!$(id)){const node=document.createElement(id==='goals'?'pre':'div');node.id=id;if(id==='actors')node.className='grid';$('cards').after(node)}}
@@ -51,17 +51,17 @@
 const plansPanel=panel('plan-panel','Mission plan',$('cards'));let planGrid=$('plan-grid');if(!planGrid){planGrid=document.createElement('div');planGrid.id='plan-grid';planGrid.className='plan-grid';plansPanel.append(planGrid)}planGrid.replaceChildren();
 for(const [key,label] of [['long','Long-term goal'],['mid','Medium-term plan'],['short','Next steps']]){const box=document.createElement('article');box.className='plan-item';const h=document.createElement('h3');h.textContent=label;box.append(h);const p=document.createElement('p');p.textContent=data.live?.goals?.[key]?.text||'No recorded plan';box.append(p);planGrid.append(box)}
 const resources=panel('resource-panel','Compute resources',plansPanel);resources.append($('actors'));$('actors').className='resource-grid';
-const agents=panel('agent-panel','Agent activity - tasks, decisions and evidence',resources);let grid=$('agent-grid');if(!grid){const note=document.createElement('p');note.className='muted';note.textContent='Recorded public output, tool results and token counts. Counts and opted-in local model reasoning arrive after a response finishes. Model reasoning is unverified, not evidence of execution.';grid=document.createElement('div');grid.id='agent-grid';agents.append(note,grid)}
+const agents=panel('agent-panel','Agent activity - tasks, decisions and evidence',resources);let grid=$('agent-grid');if(!grid){const note=document.createElement('p');note.className='muted';note.textContent='Recorded public output, tool results and token counts. Opted-in local output and reasoning stream during inference. Measured token counts arrive at completion. Model reasoning is unverified, not evidence of execution.';grid=document.createElement('div');grid.id='agent-grid';agents.append(note,grid)}
 grid.replaceChildren();
 for(const a of data.live?.agents||[]){const node=document.createElement('article');node.className='agent-row';const head=document.createElement('header');const h=document.createElement('h3');h.textContent=a.label;const badge=document.createElement('span');badge.className='state-badge';badge.textContent=a.state||'unknown';head.append(h,badge);node.append(head);
 textField(node,'Model / device',[a.model,a.device].filter(Boolean).join(' / '));
 const u=a.usage||{};textField(node,'Last response tokens',Object.keys(u).length?`Input ${fmt(u.prompt_tokens)} | Output ${fmt(u.completion_tokens)} | Total ${fmt(u.total_tokens)} | recorded ${a.usage_at||''}`:'Not reported for this actor');
 textField(node,'Duration',a.seconds===undefined?null:a.seconds+' s');
 detailsField(node,'Assigned task',a.task,a.label+'-task');textField(node,'Tool',a.tool);
-detailsField(node,'Returned local model reasoning (unverified)',a.returned_trace,a.label+'-trace');detailsField(node,'Public conclusion and next step',a.declaration,a.label+'-output');detailsField(node,'Observed result / error',a.result,a.label+'-result');detailsField(node,'Evidence',a.source,a.label+'-source');textField(node,'Last event',typeof a.updated==='number'?new Date(a.updated*1000).toLocaleString():a.updated);grid.append(node)}
+detailsField(node,'Live partial output - latest recorded window',a.stream_output,a.label+'-live-output');detailsField(node,'Live local reasoning - latest recorded window',a.stream_reasoning,a.label+'-live-trace');detailsField(node,'Returned local model reasoning (unverified)',a.returned_trace,a.label+'-trace');detailsField(node,'Public conclusion and next step',a.declaration,a.label+'-output');detailsField(node,'Observed result / error',a.result,a.label+'-result');detailsField(node,'Evidence',a.source,a.label+'-source');textField(node,'Last event',typeof a.updated==='number'?new Date(a.updated*1000).toLocaleString():a.updated);grid.append(node)}
 if(!grid.children.length)textField(grid,'Activity','No recent recorded agent events');
 const timeline=panel('timeline-panel','Action timeline - latest 200 recorded events',agents);timeline.append($('events'));$('events').replaceChildren();
-for(const e of [...(data.live?.events||[])].reverse()){const d=document.createElement('details');d.className='event-row';d.dataset.key=(e.request_id||e.time)+'-'+e.actor+'-'+e.kind;const h=document.createElement('summary');h.textContent=[e.time,e.branch,e.actor,e.kind,e.tool].filter(Boolean).join(' | ');d.append(h);textField(d,'Returned local reasoning (unverified)',e.returned_trace);textField(d,'Task / arguments',e.task);textField(d,'Public output / result',e.summary);textField(d,'Error / status',e.detail);if(e.usage&&Object.keys(e.usage).length)textField(d,'Tokens',JSON.stringify(e.usage));textField(d,'Seconds',e.seconds);textField(d,'Evidence',e.source);$('events').append(d)}
+for(const e of [...(data.live?.events||[])].reverse()){const d=document.createElement('details');d.className='event-row';d.dataset.key=(e.request_id||e.time)+'-'+e.actor+'-'+e.kind;const h=document.createElement('summary');h.textContent=[e.time,e.branch,e.actor,e.kind,e.tool].filter(Boolean).join(' | ');d.append(h);textField(d,'Live delta',e.delta_text);textField(d,'Returned local reasoning (unverified)',e.returned_trace);textField(d,'Task / arguments',e.task);textField(d,'Public output / result',e.summary);textField(d,'Error / status',e.detail);if(e.usage&&Object.keys(e.usage).length)textField(d,'Tokens',JSON.stringify(e.usage));textField(d,'Seconds',e.seconds);textField(d,'Evidence',e.source);$('events').append(d)}
 for(const n of document.querySelectorAll('details[data-key]'))if(opened.has(n.dataset.key))n.open=true;
 }
 function archivePanel(){
@@ -73,7 +73,7 @@
 }
 function learningPanel(data){const p=panel('goal-learning-panel','Goal-directed ML pipeline',$('plan-panel'));let body=$('goal-learning-view');if(!body){body=document.createElement('div');body.id='goal-learning-view';p.append(body)}body.replaceChildren();const evidence=data.report?.mission_evidence||{},g=evidence.goal_learning||{};textField(body,'Active goal ID',g.goal_id||'No goal recorded');for(const [k,v] of Object.entries(g.learning_task||{}))textField(body,k,v);textField(body,'Recent forecasts',fmt(g.forecasts?.length));textField(body,'Recent resolved forecasts',g.recent_resolved);textField(body,'Recent Brier score',g.recent_brier);textField(body,'Admitted training examples',g.training_admission?.admitted);textField(body,'Confirmed optimizer updates',fmt(evidence.optimizer_updates_observed));textField(body,'Accepted weight updates',fmt(evidence.accepted_weight_updates_this_run));detailsField(body,'Training admission evidence',JSON.stringify(g.training_admission||{}),'goal-admission')}
 function svg(tag,attrs,text){let node=document.createElementNS('http://www.w3.org/2000/svg',tag);for(const [key,val] of Object.entries(attrs))node.setAttribute(key,val);if(text!==undefined)node.textContent=text;return node}
-function render(data){const m=data.mission||{},r=data.report||{},ev=(m.state?.phase==='official-public-baseline'?r.mission_evidence?.official_benchmark:null)||m.evaluation||{},learn=m.learning||{},gpu=data.gpu||{};
+function render(data){latestSnapshot=data;const m=data.mission||{},r=data.report||{},ev=(m.state?.phase==='official-public-baseline'?r.mission_evidence?.official_benchmark:null)||m.evaluation||{},learn=m.learning||{},gpu=data.gpu||{};
 if(data.layout?.active_sha256&&layoutVersion!==data.layout.active_sha256){location.reload();return}$('layout').textContent='HTML mastera: /workspace/dashboard/index.html | '+(data.layout?.state||'oczekiwanie')+(data.layout?.error?' | '+data.layout.error:'');
 $('fresh').textContent='Zebrano: '+new Date(data.collected_at*1000).toLocaleString()+' | Raport zbierany co 20 s | Przebieg: '+fmt(m.run);
 workView(data);learningPanel(data);archivePanel();$('warning').textContent=(data.errors||[]).join('\n');$('cards').replaceChildren();card('Misja',m.running?'działa':'zatrzymana');card('Faza',m.state?.phase);card('Cykle uczenia',learn.completed_cycles);card('Zaakceptowane aktualizacje wag',r.accepted_weight_updates_this_run);card('Potwierdzone kroki optymalizatora',r.mission_evidence?.optimizer_updates_observed);card('Kontekst',m.state?.context_window);card('GPU - wykorzystanie',gpu.utilization===undefined?null:gpu.utilization+'%');card('VRAM zajęty / razem',gpu.memory_used===undefined?null:gpu.memory_used+' / '+gpu.memory_total+' MiB');card('GPU - temperatura / moc',gpu.temperature===undefined?null:gpu.temperature+' °C / '+gpu.power+' W');
@@ -83,7 +83,9 @@
 $('reports').replaceChildren();for(const id of data.paper_reports||[]){const row=document.createElement('p');row.append(document.createTextNode(id+' '));for(const [name,label] of [['report.html','HTML i wykresy'],['report.json','JSON'],['trades.csv','CSV']]){const a=document.createElement('a');a.href='/paper/'+encodeURIComponent(id)+'/'+name;a.target='_blank';a.rel='noopener';a.textContent=label;row.append(a,document.createTextNode(' · '))}$('reports').append(row)}
 }
 async function refresh(){try{let response=await fetch('/api/status',{cache:'no-store'});if(!response.ok)throw Error('HTTP '+response.status);render(await response.json())}catch(error){$('warning').textContent='Nie można odświeżyć: '+error.message}}
-$('reload').onclick=refresh;refresh();setInterval(refresh,5000);
+async function liveInference(){if(!latestSnapshot)return;try{const response=await fetch('/api/live-inference',{cache:'no-store'});if(!response.ok)throw Error('HTTP '+response.status);const live=await response.json();agentView({...latestSnapshot,live:{...latestSnapshot.live,events:live.events,agents:[...(live.agents||[]),...(latestSnapshot.live?.agents||[]).filter(a=>a.source==='research/state/drones.sqlite3')]}})}catch(error){$('warning').textContent='Live inference unavailable: '+error.message}}
+async function readinessView(){try{const response=await fetch('/api/readiness',{cache:'no-store'});if(!response.ok)throw Error('HTTP '+response.status);const data=await response.json();const p=panel('readiness-panel','Commissioning evidence',$('goal-learning-panel')||$('plan-panel'));let body=$('readiness-view');if(!body){body=document.createElement('div');body.id='readiness-view';p.append(body)}body.replaceChildren();for(const c of data.checks||[])textField(body,c.name,`${c.state} | ${c.detail} | ${c.source||''}`)}catch(error){$('warning').textContent='Readiness unavailable: '+error.message}}
+$('reload').onclick=refresh;refresh();setInterval(refresh,5000);setInterval(liveInference,2000);setInterval(readinessView,5000);
 </script></html>"""
 
 

rlm/v100/live_status.py

@@ -75,6 +75,12 @@ def recent_events(root: Path, limit: int = 200) -> list[dict]:
                         if row.get("kind") == "local-model-reasoning"
                         else None,
                         "trace_part": payload.get("part"),
+                        "delta_channel": payload.get("channel")
+                        if row.get("kind") == "inference-delta"
+                        else None,
+                        "delta_text": payload.get("text")
+                        if row.get("kind") == "inference-delta"
+                        else None,
                         "source": str(path),
                         "device": payload.get("device"),
                         "model": payload.get("model"),
@@ -109,6 +115,14 @@ def agent_views(events: list[dict], jobs: list[dict]) -> list[dict]:
         if event.get("device"):
             view["device"] = event["device"]
             view["model"] = event.get("model")
+        if event.get("delta_text"):
+            if view.get("stream_request") != event.get("request_id"):
+                view["stream_output"], view["stream_reasoning"] = "", ""
+            view["stream_request"] = event.get("request_id")
+            key = (
+                "stream_reasoning" if event.get("delta_channel") == "reasoning" else "stream_output"
+            )
+            view[key] = (view.get(key, "") + event["delta_text"])[-16384:]
         if event.get("returned_trace"):
             if event.get("trace_part") == 0:
                 view["returned_trace"] = ""