Fix onboard WebKit preview and archive board telemetry locally

apps/control-station/src/components/system/TelemetrySeries.tsx

@@ -6,19 +6,17 @@ interface TelemetrySeriesProps {
   ceiling?: number;
 }
 
-function linePoints(values: Array<number | null>, ceiling?: number): string {
+export function lineSegments(values: Array<number | null>, ceiling?: number): string[] {
-  const finite = values.filter((value): value is number =>
+  const finite = values.filter((value): value is number => value !== null && Number.isFinite(value));
-    value !== null && Number.isFinite(value),
+  if (!finite.length) return [];
-  );
+  const maximum = Math.max(ceiling ?? 0, ...finite, 1), denominator = Math.max(1, values.length - 1);
-  if (finite.length === 0) return "";
+  const segments:string[]=[];let segment:string[]=[];
-  const maximum = Math.max(ceiling ?? 0, ...finite, 1);
+  values.forEach((value,index)=>{
-  const denominator = Math.max(1, values.length - 1);
+    if(value===null||!Number.isFinite(value)){if(segment.length)segments.push(segment.join(' '));segment=[];return;}
-  return values.map((value, index) => {
+    segment.push(`${(index/denominator*100).toFixed(2)},${(36-Math.max(0,Math.min(maximum,value))/maximum*34).toFixed(2)}`);
-    const x = index / denominator * 100;
+  });
-    const normalized = value === null ? 0 : Math.max(0, Math.min(maximum, value));
+  if(segment.length)segments.push(segment.join(' '));
-    const y = 36 - normalized / maximum * 34;
+  return segments;
-    return `${x.toFixed(2)},${y.toFixed(2)}`;
-  }).join(" ");
 }
 
 export function TelemetrySeries({
@@ -28,7 +26,7 @@ export function TelemetrySeries({
   resource,
   ceiling,
 }: TelemetrySeriesProps) {
-  const points = linePoints(values, ceiling);
+  const segments = lineSegments(values, ceiling);
   return (
     <div className="system-telemetry-series">
       <div>
@@ -45,7 +43,7 @@ export function TelemetrySeries({
         aria-label={`${label}: ${value}`}
       >
         <path d="M0 37 H100" />
-        {points ? <polyline points={points} /> : null}
+        {segments.map((points,index)=><polyline key={index} points={points} />)}
       </svg>
     </div>
   );

apps/control-station/src/core/fleet/boardMonitor.ts

@@ -0,0 +1,49 @@
+import {useEffect,useState} from 'react';
+
+export interface MetricDefinition {label:string;group:string;resource:string;unit:string;reason:string|null}
+export interface BoardMonitor {
+ available:boolean;fresh:boolean;storage:string;source_id:string;
+ latest:{seq:number;at:number;boot_id:string;values:Record<string,number|null>}|null;
+ definitions:Record<string,MetricDefinition>;received_at:number;backlog:number;
+ database_bytes:number|null;database_budget_bytes:number;archive_since:number|null;
+ start:number;end:number;bucket_seconds:number;
+ series:{at:number;min:number|null;max:number|null;mean:number|null;count:number}[];
+ events:{at:number;code:string;kind:string;locations:string[]}[];
+}
+export function useBoardMonitor(vehicle:string,metric:string,window:number,end:number|null){
+ const [value,setValue]=useState<BoardMonitor|null>(null),[error,setError]=useState('');
+ useEffect(()=>{
+  let active=true;let pending:AbortController|undefined;let timer:ReturnType<typeof setTimeout>|undefined;
+  setValue(null);setError('');
+  const read=async()=>{
+   const controller=new AbortController();pending=controller;
+   const deadline=setTimeout(()=>controller.abort(),12000);
+   try {
+    const query=new URLSearchParams({metric,window:String(window),...(end===null?{}:{end:String(end)})});
+    const response=await fetch(`/api/v1/fleet/${encodeURIComponent(vehicle)}/monitor?${query}`,{credentials:'same-origin',cache:'no-store',signal:controller.signal});
+    if(!response.ok)throw new Error('Мониторинг БК недоступен.');
+    const next=await response.json() as BoardMonitor;
+    if(active){setValue(next);setError('');}
+   }catch {if(active)setError('Не удалось обновить мониторинг БК.');}
+   finally {clearTimeout(deadline);if(active)timer=setTimeout(()=>void read(),5000);}
+  };
+  void read();return()=>{active=false;pending?.abort();clearTimeout(timer);};
+ },[vehicle,metric,window,end]);
+ return {value,error,dismissError:()=>setError('')};
+}
+export function monitorSeries(value:BoardMonitor,mode:'min'|'max'|'mean') {
+ const count=Math.ceil((value.end-value.start)/value.bucket_seconds)+1;
+ const values:Array<number|null>=Array(Math.min(count,302)).fill(null);
+ const first=Math.floor(value.start/value.bucket_seconds)*value.bucket_seconds;
+ for(const point of value.series){const index=Math.round((point.at-first)/value.bucket_seconds);if(index>=0&&index<values.length)values[index]=point[mode];}
+ return values;
+}
+export function monitorValue(value:number|null|undefined,unit=''){
+ if(value==null||!Number.isFinite(value))return 'Нет измерения';
+ if(unit==='bytes'||unit==='bytes/s'){
+  const labels=unit==='bytes'?['Б','КиБ','МиБ','ГиБ']:['Б/с','КиБ/с','МиБ/с','ГиБ/с'];let index=0;
+  while(value>=1024&&index<3){value/=1024;index++;}
+  return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})} ${labels[index]}`;
+ }
+ return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})}${unit?' '+unit:''}`;
+}

apps/control-station/src/workspaces/fleet/BoardMonitorView.tsx

@@ -0,0 +1,45 @@
+import {useState} from 'react';
+import {ActivityIndicator,Button,Icon,ResourceList,ResourceRow,Select,SettingsCard,StatusBadge,TextField,ToastStack} from '@nodedc/ui-react';
+import {useBoardMonitor,monitorSeries,monitorValue} from '../../core/fleet/boardMonitor';
+import {TelemetrySeries} from '../../components/system/TelemetrySeries';
+import '../../styles/system-telemetry.css';
+import './board-monitor.css';
+
+const eventLabels:Record<string,string>={'ui-error':'Ошибка интерфейса','ui-rejection':'Ошибка асинхронной операции','ui-render-error':'Ошибка представления','system-oom':'Нехватка памяти','gpu-reset':'Сброс GPU','disk-error':'Ошибка диска','usb-disconnected':'Отключение USB','usb-error':'Ошибка USB','service-failed':'Сбой службы','ui-process-exit':'Завершение процесса окна','ui-load-failed':'Ошибка загрузки окна'};
+const time=(value:number)=>new Date(value*1000).toLocaleString('ru-RU');
+export function BoardMonitorView({vehicle,name,back}:{vehicle:string;name:string;back:()=>void}) {
+ const [metric,setMetric]=useState('cpu.usage'),[group,setGroup]=useState('CPU'),[window,setWindow]=useState(900),[end,setEnd]=useState<number|null>(null),[mode,setMode]=useState<'min'|'max'|'mean'>('max');
+ const {value,error,dismissError}=useBoardMonitor(vehicle,metric,window,end);
+ const definitions=value?.definitions??{},definition=definitions[metric];
+ const groups=[...new Set(Object.values(definitions).map(item=>item.group))];
+ const metrics=Object.entries(definitions).filter(([,item])=>item.group===group);
+ const latest=value?.latest;
+ return <div className="board-monitor">
+  <div><Button onClick={back}><Icon name="chevron-left"/>К аппарату</Button></div>
+  <SettingsCard title="Мониторинг системы БК" description={name} actions={<StatusBadge tone={value?.fresh&&!error?'success':'neutral'}>{value?.fresh&&!error?'Данные поступают':'Нет свежих данных'}</StatusBadge>}>
+   {latest&&<p>Измерено на БК: {time(latest.at)}. {value?.storage==='ready'?'Локальная история записывается.':'Запись локальной истории недоступна.'} {value?.backlog?'История догружается в Mission Core.':''}</p>}
+   {value?.database_bytes!=null&&<p>Архив на БК: {monitorValue(value.database_bytes,'bytes')} из {monitorValue(value.database_budget_bytes,'bytes')}. {value.archive_since?`Копия в Mission Core с ${time(value.archive_since)}.`:''}</p>}
+  </SettingsCard>
+  {!value?<ActivityIndicator label="Получаем мониторинг БК"/>:!value.available?<SettingsCard title="История БК пока не получена" description="Данные появятся после подключения бортового сборщика телеметрии."/>:<>
+   <SettingsCard title="История нагрузки">
+    <div className="board-monitor-controls">
+     <Select label="Ресурс" value={group} options={groups.map(label=>({value:label,label}))} onChange={next=>{setGroup(next);const first=Object.entries(definitions).find(([,item])=>item.group===next);if(first)setMetric(first[0]);}}/>
+     <Select label="Показатель" value={metric} options={metrics.map(([key,item])=>({value:key,label:`${item.resource} · ${item.label}`}))} onChange={setMetric}/>
+     <Select label="Период" value={String(window)} options={[{value:'900',label:'15 минут'},{value:'3600',label:'1 час'},{value:'21600',label:'6 часов'},{value:'86400',label:'24 часа'},{value:'604800',label:'7 дней'}]} onChange={next=>setWindow(Number(next))}/>
+     <Select label="Значение в интервале" value={mode} options={[{value:'max',label:'Максимум'},{value:'mean',label:'Среднее'},{value:'min',label:'Минимум'}]} onChange={next=>setMode(next as typeof mode)}/>
+    </div>
+    <div className="board-monitor-controls"><Button onClick={()=>setEnd(end===null?Date.now()/1000:null)}>{end===null?'Зафиксировать время':'В реальном времени'}</Button>{end!==null&&<TextField label="Конец периода" type="datetime-local" step="1" value={new Date((end-new Date(end*1000).getTimezoneOffset()*60)*1000).toISOString().slice(0,19)} onChange={event=>{const next=new Date(event.target.value).getTime()/1000;if(Number.isFinite(next))setEnd(next);}}/>}</div>
+    <TelemetrySeries label={definition?`${definition.resource} · ${definition.label}`:'Загрузка CPU'} resource="Последний интервал периода" value={monitorValue(value.series.at(-1)?.[mode],definition?.unit)} values={monitorSeries(value,mode)} ceiling={definition?.unit==='%'?100:undefined}/>
+    <p>{time(value.start)} — {time(value.end)} · интервал {value.bucket_seconds} с. Разрывы означают отсутствие измерений.</p>
+    {!value.series.length&&<p>За выбранный период измерений нет.</p>}
+   </SettingsCard>
+   <SettingsCard title={`Последние измерения · ${group}`} description={latest?time(latest.at):'Ожидаем измерения'}>
+    <ResourceList aria-label="Показатели системы">{metrics.map(([key,item])=><li key={key}><ResourceRow title={`${item.resource} · ${item.label}`} description={latest?.values[key]==null?item.reason??'Счётчик недоступен':undefined} metadata={monitorValue(latest?.values[key],item.unit)}/></li>)}</ResourceList>
+   </SettingsCard>
+   <SettingsCard title="События системы и приложения" description="За выбранный период">
+    {value.events.length?<ResourceList aria-label="События БК">{value.events.map((event,index)=><li key={`${event.at}-${index}`}><ResourceRow title={eventLabels[event.code]??'Событие системы'} description={time(event.at)}/></li>)}</ResourceList>:<p>Событий не получено.</p>}
+   </SettingsCard>
+  </>}
+  <ToastStack items={error?[{id:'monitor-error',title:error,tone:'error',durationMs:null}]:[]} onDismiss={dismissError}/>
+ </div>;
+}

apps/control-station/src/workspaces/fleet/VehiclesWorkspace.tsx

@@ -1,3 +1,4 @@
+import {BoardMonitorView} from './BoardMonitorView';
 import { useEffect, useRef, useState } from "react";
 import { ActivityIndicator, Button, ConfirmationModal, Icon, IconButton, ResourceList, ResourceRow, Select, SettingsCard, StatusBadge, TextAreaField, TextField, Window, WindowFooterActions } from "@nodedc/ui-react";
 import { fleetRequest, useFleet, type FleetPreview, type Vehicle } from "../../core/fleet/useFleet";
@@ -19,6 +20,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe
   const [error, setError] = useState("");
   const [selected, setSelected] = useState<string | null>(null);
   const [sensorOpen, setSensorOpen] = useState(false);
+  const [monitorOpen,setMonitorOpen]=useState(false);
   const [revoking, setRevoking] = useState<Vehicle | null>(null);
   const lastCreateRequest = useRef(createRequest);
   useEffect(() => { if (createRequest !== lastCreateRequest.current) { lastCreateRequest.current = createRequest; setAdding(true); setError(""); } }, [createRequest]);
@@ -40,6 +42,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe
     finally { setPending(false); }
   }
   const detail = fleet.items?.find(item => item.id === selected);
+  if(detail&&monitorOpen)return <BoardMonitorView vehicle={detail.id} name={detail.name} back={()=>setMonitorOpen(false)}/>;
   return <div className="fleet-workspace">
     {fleet.error && <p role="alert">{fleet.error}</p>}
     {!adding && error && <p role="alert">{error}</p>}
@@ -51,6 +54,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe
           <div><dt>Последняя связь</dt><dd>{detail.last_seen ? new Date(detail.last_seen * 1000).toLocaleString("ru-RU") : "Соединение ещё не получено"}</dd></div>
           {detail.host && <><div><dt>Имя БК в системе</dt><dd>{detail.host.hostname}</dd></div><div><dt>Операционная система</dt><dd>{detail.host.os}</dd></div><div><dt>Архитектура</dt><dd>{detail.host.architecture}</dd></div><div><dt>Процессоры</dt><dd>{detail.host.cpus}</dd></div><div><dt>Память</dt><dd>{detail.host.memory_kib ? `${(detail.host.memory_kib / 1024 / 1024).toFixed(1)} ГиБ` : "Нет сведений"}</dd></div></>}
         </dl>
+        <Button onClick={()=>setMonitorOpen(true)}><Icon name="activity"/>Мониторинг системы БК</Button>
         {detail.enrollment !== "revoked" && <Button onClick={() => setRevoking(detail)}>Отозвать привязку БК</Button>}
       </SettingsCard>}
       <SettingsCard title="Устройства аппарата"><VehicleSensors onDetailChange={setSensorOpen} vehicleID={detail.id} enabled={!fleet.error && detail.enrollment === "paired" && detail.connectivity === "online"} /></SettingsCard>

apps/control-station/src/workspaces/fleet/board-monitor.css

@@ -0,0 +1,3 @@
+.board-monitor { display: grid; gap: var(--nodedc-space-4); min-width: 0; }
+.board-monitor-controls { display: grid; grid-template-columns: repeat(auto-fit, minmax(220px, 1fr)); gap: var(--nodedc-space-3); align-items: end; margin-bottom: var(--nodedc-space-3); }
+.board-monitor p { font-size: var(--nodedc-font-size-sm); color: var(--nodedc-text-secondary); }

apps/control-station/test/boardMonitor.test.mjs

@@ -0,0 +1,52 @@
+import assert from 'node:assert/strict';
+import {before,after,test} from 'node:test';
+import {createServer} from 'vite';
+let server,series,segments,frames,localPreview,MEDIA_PROTOCOL;
+before(async()=>{
+ server=await createServer({appType:'custom',logLevel:'silent',server:{middlewareMode:true}});
+ ({monitorSeries:series}=await server.ssrLoadModule('/src/core/fleet/boardMonitor.ts'));
+ ({lineSegments:segments}=await server.ssrLoadModule('/src/components/system/TelemetrySeries.tsx'));
+ ({MEDIA_PROTOCOL}=await server.ssrLoadModule('@xgrids-k1/frontend/sensors/previewFrames.ts'));
+ ({localPreviewFrames:frames,localPreview}=await server.ssrLoadModule('@xgrids-k1/frontend/sensors/localPreview.ts'));
+});
+after(async()=>{await server?.close();});
+test('archive gaps break the plot instead of manufacturing zero or connecting missing time',()=>{
+ const values=series({start:0,end:20,bucket_seconds:5,series:[{at:0,max:80},{at:5,max:90},{at:20,max:10}]},'max');
+ assert.deepEqual(values,[80,90,null,null,10]);
+ const result=segments(values,100);
+ assert.equal(result.length,2);
+ assert.match(result[0],/^0.00,8.80 25.00,5.40$/);
+ assert.deepEqual(segments([null,NaN,null]),[]);
+});
+test('HTTP media framing survives every byte boundary and refuses oversized frames',()=>{
+ const output=[],receive=frames((kind,bytes)=>output.push([kind,[...bytes]]));
+ const encoded=Uint8Array.of(2,0,0,0,3,4,5,6,3,0,0,0,2,7,8);
+ for(const byte of encoded)receive(Uint8Array.of(byte));
+ assert.deepEqual(output,[[2,[4,5,6]],[3,[7,8]]]);
+ assert.throws(()=>frames(()=>{})(Uint8Array.of(2,0,1,0,0)));
+ assert.throws(()=>frames(()=>{})(Uint8Array.of(7,0,0,0,0)));
+});
+
+test('local carrier delivers registered channels and ACKs through completed responses',async()=>{
+ const encode=(kind,data)=>{const bytes=typeof data==='string'?new TextEncoder().encode(data):Uint8Array.from(data);const out=new Uint8Array(bytes.length+5);out[0]=kind;new DataView(out.buffer).setUint32(1,bytes.length);out.set(bytes,5);return out;};
+ const peer='peer_'+ 'a'.repeat(32), received=[], closed=[];let preview;
+ const device={id:'synthetic-k1',snapshot:{context:{session_id:'synthetic-session'}}};
+ const transport={
+  submit:async request=>{closed.push(request);return {state:'complete'};},
+  localPreview:{
+   open:async request=>{assert.equal(request.action_id,'offer');return new Response(encode(0,JSON.stringify({peer_id:peer,media_protocol:MEDIA_PROTOCOL})));},
+   read:async(id,after)=>{
+    assert.equal(id,peer);
+    if(after===0)return new Response(new Uint8Array([...encode(2,[8,9]),...encode(3,'camera-ready')]));
+    assert.equal(after,7);preview.close();return new Response();
+   }
+  }
+ };
+ preview=localPreview(device,transport,{},()=>assert.fail('carrier failed'));
+ preview.rrd.onmessage=event=>{received.push([...new Uint8Array(event.data)]);preview.rrd.send(JSON.stringify({ack:7}));};
+ preview.camera.onmessage=event=>received.push(event.data);
+ await preview.start();
+ assert.deepEqual(received,[[8,9],'camera-ready']);
+ assert.equal(preview.rrd.readyState,'closed');assert.equal(closed.length,1);
+ assert.equal(closed[0].action_id,'close-peer');assert.equal(closed[0].parameters.peer_id,peer);
+});

apps/node-agent/cmd/node-agent/main.go

@@ -69,12 +69,14 @@ func run() error {
 		return err
 	}
 	app := &node.Server{Store: store, Assets: assets, Origin: "http://" + *listen, Version: version, Inventory: func() node.Inventory { return node.Host("/") }}
+	app.Monitor = node.NewMonitor()
 	app.Presentation = node.NewPresentationStore(*dir)
 	pairing, err := node.OpenPairing(store, *dir, version, app.Inventory)
 	if err != nil {
 		return err
 	}
 	app.Pairing = pairing
+	pairing.Monitor = app.Monitor
 	nodeID, _ := store.Public()
 	app.Sensors, err = node.OpenSensors(*dir, nodeID)
 	if err != nil {
@@ -122,6 +124,7 @@ func run() error {
 	go func() { errs <- private.Serve(unix) }()
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
+	go app.Monitor.Run(ctx)
 	go app.Sensors.WatchUSB(ctx)
 	go pairing.Run(ctx)
 	log.Print("Mission Core Node " + version + " listening on loopback")

apps/node-agent/internal/node/local_preview.go

@@ -0,0 +1,73 @@
+package node
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+// Local WebKit requires completed bounded HTTP responses. The same plugin
+// delivery owns RRD/fMP4 framing, backpressure, resume and acquisition isolation.
+func (s *Server) localPreviewRoutes(mux *http.ServeMux) {
+	capacity := make(chan struct{}, 2)
+	for _, route := range []string{"/api/devices/preview", "/api/devices/preview/read"} {
+		mux.HandleFunc("POST "+route, func(w http.ResponseWriter, r *http.Request) {
+			if !s.authorized(w, r) {
+				return
+			}
+			var value json.RawMessage
+			if !decode(w, r, &value) {
+				return
+			}
+			if s.DeviceEnrollment == nil {
+				http.Error(w, "Preview unavailable", 503)
+				return
+			}
+			path := "/local-preview/read"
+			if r.URL.Path == "/api/devices/preview" {
+				var command SensorCommand
+				if json.Unmarshal(value, &command) != nil || command.Action != "offer" || !sensorID.MatchString(command.Session.DeviceID) || !strings.HasPrefix(command.Session.DeviceID, "k1_") {
+					http.Error(w, "Invalid preview", 400)
+					return
+				}
+				path = "/local-preview"
+			}
+			select {
+			case capacity <- struct{}{}:
+				defer func() { <-capacity }()
+			default:
+				http.Error(w, "Close another viewer", 429)
+				return
+			}
+			request, err := http.NewRequestWithContext(r.Context(), "POST", "http://k1"+path, bytes.NewReader(value))
+			if err != nil {
+				http.Error(w, "Invalid preview", 400)
+				return
+			}
+			request.Header.Set("Content-Type", "application/json")
+			request.Header.Set("X-Node-Id", s.DeviceEnrollment.nodeID)
+			client := &http.Client{Transport: s.DeviceEnrollment.client.Transport, Timeout: 5 * time.Second}
+			response, err := client.Do(request)
+			if err != nil {
+				http.Error(w, "Preview unavailable", 503)
+				return
+			}
+			defer response.Body.Close()
+			if response.StatusCode != 200 {
+				http.Error(w, "Preview unavailable", response.StatusCode)
+				return
+			}
+			data, err := io.ReadAll(io.LimitReader(response.Body, 196609))
+			if err != nil || len(data) > 196608 {
+				http.Error(w, "Preview unavailable", 503)
+				return
+			}
+			w.Header().Set("Content-Type", "application/vnd.missioncore.preview")
+			w.Header().Set("Cache-Control", "no-store")
+			_, _ = w.Write(data)
+		})
+	}
+}

apps/node-agent/internal/node/monitor.go

@@ -0,0 +1,159 @@
+package node
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+)
+
+// The collector owns sampling/storage. This bounded cache keeps monitoring I/O
+// out of the paired control heartbeat and out of acquisition lifecycles.
+type Monitor struct {
+	mu     sync.Mutex
+	client *http.Client
+	source string
+	after  int64
+	sent   int64
+	value  map[string]any
+}
+
+func NewMonitor() *Monitor {
+	return &Monitor{client: &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return (&net.Dialer{}).DialContext(ctx, "unix", "/run/mission-core-monitor/monitor.sock")
+	}}}}
+}
+func (m *Monitor) read(ctx context.Context, path string, body []byte) (map[string]any, error) {
+	method := "GET"
+	if body != nil {
+		method = "POST"
+	}
+	r, e := http.NewRequestWithContext(ctx, method, "http://monitor"+path, bytes.NewReader(body))
+	if e != nil {
+		return nil, e
+	}
+	r.Header.Set("Content-Type", "application/json")
+	response, e := m.client.Do(r)
+	if e != nil {
+		return nil, e
+	}
+	defer response.Body.Close()
+	if response.StatusCode != 200 {
+		return nil, errors.New("monitor unavailable")
+	}
+	data, e := io.ReadAll(io.LimitReader(response.Body, 262145))
+	if e != nil || len(data) > 262144 {
+		return nil, errors.New("monitor response budget")
+	}
+	var value map[string]any
+	e = json.Unmarshal(data, &value)
+	return value, e
+}
+func (m *Monitor) Run(ctx context.Context) {
+	defer m.client.CloseIdleConnections()
+	timer := time.NewTicker(2 * time.Second)
+	defer timer.Stop()
+	for {
+		status, e := m.read(ctx, "/status", nil)
+		if e == nil {
+			source, _ := status["source_id"].(string)
+			m.mu.Lock()
+			if source != m.source {
+				m.source = source
+				m.after = 0
+				m.sent = 0
+			}
+			after := m.after
+			m.mu.Unlock()
+			batch, be := m.read(ctx, "/batch?after="+strconv.FormatInt(after, 10), nil)
+			if be == nil && batch["source_id"] == source {
+				status["batch"] = batch
+			}
+			m.mu.Lock()
+			m.value = status
+			m.mu.Unlock()
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+		}
+	}
+}
+func (m *Monitor) Snapshot() map[string]any {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.value == nil {
+		return map[string]any{"storage": "unavailable"}
+	}
+	if batch, ok := m.value["batch"].(map[string]any); ok {
+		if rows, ok := batch["samples"].([]any); ok {
+			for _, raw := range rows {
+				if row, ok := raw.(map[string]any); ok {
+					if seq, ok := row["seq"].(float64); ok && int64(seq) > m.sent {
+						m.sent = int64(seq)
+					}
+				}
+			}
+		}
+	}
+	return m.value
+}
+func (m *Monitor) Acknowledge(raw json.RawMessage) {
+	var ack struct {
+		Source string `json:"source_id"`
+		After  int64  `json:"after"`
+	}
+	if json.Unmarshal(raw, &ack) != nil {
+		return
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	// A receiver may already have a later cursor from before the Node restarted.
+	// Its authenticated ACK is bounded by the collector's durable latest record.
+	latest := m.sent
+	if row, ok := m.value["latest"].(map[string]any); ok {
+		if seq, ok := row["seq"].(float64); ok {
+			latest = int64(seq)
+		}
+	}
+	if ack.Source == m.source && ack.After >= m.after && ack.After <= latest {
+		m.after = ack.After
+	}
+}
+func (s *Server) monitorRoutes(mux *http.ServeMux) {
+	mux.HandleFunc("GET /api/monitor", func(w http.ResponseWriter, r *http.Request) {
+		if !s.authorized(w, r) {
+			return
+		}
+		if s.Monitor == nil {
+			reply(w, 503, map[string]string{"error": "Мониторинг недоступен"})
+			return
+		}
+		reply(w, 200, s.Monitor.Snapshot())
+	})
+	mux.HandleFunc("POST /api/monitor/events", func(w http.ResponseWriter, r *http.Request) {
+		if !s.authorized(w, r) {
+			return
+		}
+		var value json.RawMessage
+		if !decode(w, r, &value) {
+			return
+		}
+		if len(value) > 2048 || s.Monitor == nil {
+			http.Error(w, "Monitoring unavailable", 503)
+			return
+		}
+		if _, e := s.Monitor.read(r.Context(), "/events", value); e != nil {
+			http.Error(w, "Monitoring unavailable", 503)
+			return
+		}
+		reply(w, 200, map[string]bool{"ok": true})
+	})
+}

apps/node-agent/internal/node/monitor_test.go

@@ -0,0 +1,50 @@
+package node
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestMonitorAcknowledgementsRequireSameArchiveAndDurableBounds(t *testing.T) {
+	m := NewMonitor()
+	m.source = "archive-one"
+	m.value = map[string]any{"latest": map[string]any{"seq": float64(100)}}
+	for _, raw := range []string{`{"source_id":"another","after":50}`, `{"source_id":"archive-one","after":101}`, `{"source_id":"archive-one","after":-1}`} {
+		m.Acknowledge(json.RawMessage(raw))
+		if m.after != 0 {
+			t.Fatal("accepted invalid cursor")
+		}
+	}
+	m.Acknowledge(json.RawMessage(`{"source_id":"archive-one","after":80}`))
+	if m.after != 80 {
+		t.Fatal("did not restore committed receiver cursor after restart")
+	}
+	m.Acknowledge(json.RawMessage(`{"source_id":"archive-one","after":10}`))
+	if m.after != 80 {
+		t.Fatal("regressed cursor")
+	}
+}
+
+func TestMonitorAndPreviewCannotBypassNodeSession(t *testing.T) {
+	s := newTestServer(t)
+	for _, path := range []string{"/api/devices/preview", "/api/devices/preview/read", "/api/monitor/events"} {
+		if got := call(s, "POST", path, `{}`, nil).Code; got != 401 {
+			t.Fatalf("%s: %d", path, got)
+		}
+	}
+	if got := call(s, "GET", "/api/monitor", "", nil).Code; got != 401 {
+		t.Fatal(got)
+	}
+	cookie := login(t, s)
+	if got := call(s, "POST", "/api/devices/preview", `{}`, cookie).Code; got != 503 {
+		t.Fatal(got)
+	}
+	s.Monitor = NewMonitor()
+	if got := call(s, "GET", "/api/monitor", "", cookie).Code; got != 200 {
+		t.Fatal(got)
+	}
+	// An unavailable collector remains independent from the authenticated UI.
+	if got := call(s, "GET", "/api/status", "", cookie).Code; got != 200 {
+		t.Fatal(got)
+	}
+}

apps/node-agent/internal/node/pairing.go

@@ -39,6 +39,7 @@ type PairState struct {
 	Revocations []CoreBinding `json:"revocations,omitempty"`
 }
 type Pairing struct {
+	Monitor          *Monitor
 	Sensors          *Sensors
 	DeviceEnrollment *DeviceEnrollment
 	mu               sync.Mutex

apps/node-agent/internal/node/pairing_transport.go

@@ -286,10 +286,16 @@ func (p *Pairing) channel(ctx context.Context) {
 				payload["device_enrollment"] = p.DeviceEnrollment.Status()
 				payload["enrollment_results"] = p.DeviceEnrollment.RemoteResults()
 			}
+			if p.Monitor != nil {
+				payload["monitor"] = p.Monitor.Snapshot()
+			}
 			result, status, e := p.send(ctx, *binding, "/v1/node/heartbeat", payload)
 			p.mu.Lock()
 			if p.state.Phase == "paired" && p.state.Binding != nil && p.state.Binding.BindingID == binding.BindingID {
 				if e == nil && status == 200 {
+					if p.Monitor != nil {
+						p.Monitor.Acknowledge(result["monitor_ack"])
+					}
 					p.connection = "online"
 					p.lastSeen = p.now().Unix()
 					if p.DeviceEnrollment != nil {

apps/node-agent/internal/node/server.go

@@ -13,6 +13,7 @@ import (
 )
 
 type Server struct {
+	Monitor          *Monitor
 	Store            *Store
 	Pairing          *Pairing
 	Sensors          *Sensors
@@ -81,6 +82,8 @@ func reply(w http.ResponseWriter, status int, v any) {
 
 func (s *Server) Handler() http.Handler {
 	mux := http.NewServeMux()
+	s.localPreviewRoutes(mux)
+	s.monitorRoutes(mux)
 	if s.Presentation != nil {
 		s.presentationRoutes(mux)
 	}

apps/node-agent/monitor/collector.py

@@ -0,0 +1,201 @@
+#!/usr/bin/python3
+"""Independent one-second collector, local archive and private Node read adapter."""
+
+import json
+import logging
+import os
+import re
+import signal
+import socketserver
+import threading
+import time
+from collections import deque
+from http.server import BaseHTTPRequestHandler
+from pathlib import Path
+from urllib.parse import parse_qs, urlsplit
+
+from linux_metrics import LinuxMetrics
+from storage import SCHEMA, Archive
+
+SOCKET = Path("/run/mission-core-monitor/monitor.sock")
+
+
+class Collector:
+    def __init__(self):
+        self.stop = threading.Event()
+        self.lock = threading.Lock()
+        self.latest = None
+        self.definitions = {}
+        self.events = deque(maxlen=64)
+        self.storage = "starting"
+        self.database_bytes = None
+        self.archive = None
+        self.boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
+
+    def event(self, value):
+        if (
+            not isinstance(value, dict)
+            or set(value) - {"code", "kind", "locations"}
+            or value.get("code") not in {"ui-error", "ui-rejection", "ui-render-error"}
+            or not isinstance(value.get("kind"), str)
+            or not re.fullmatch("[A-Za-z]{1,48}", value["kind"])
+            or not isinstance(value.get("locations", []), list)
+            or len(value.get("locations", [])) > 8
+            or any(
+                not isinstance(v, str)
+                or not re.fullmatch(r"[A-Za-z0-9_-]{1,96}\.js:[0-9]{1,9}:[0-9]{1,9}", v)
+                for v in value.get("locations", [])
+            )
+        ):
+            raise ValueError("invalid-event")
+        with self.lock:
+            if len(self.events) < 64:
+                self.events.append({**value, "at": time.time()})
+
+    def run(self):
+        metrics = LinuxMetrics()
+        maintenance = 0
+        while not self.stop.is_set():
+            started = time.monotonic()
+            try:
+                values = metrics.sample()
+                with self.lock:
+                    events = list(self.events)
+                sample = dict(
+                    at=time.time(), boot_id=self.boot, uptime=started, values=values, events=events
+                )
+                if self.archive is None:
+                    self.archive = Archive()
+                # Maintenance also runs while writes are paused by a disk budget.
+                if started - maintenance > 60:
+                    self.database_bytes = self.archive.maintain()
+                    maintenance = started
+                record = self.archive.append(sample, metrics.definitions)
+                with self.lock:
+                    for _ in events:
+                        self.events.popleft()
+                    self.latest = record
+                    self.definitions = metrics.definitions
+                    self.storage = "ready"
+            except Exception as error:
+                logging.warning("monitor sample failed class=%s", type(error).__name__)
+                with self.lock:
+                    self.storage = (
+                        str(error)
+                        if str(error) in {"disk-reserve", "archive-budget"}
+                        else "unavailable"
+                    )
+            self.stop.wait(max(0.05, 1 - (time.monotonic() - started)))
+
+    def status(self):
+        with self.lock:
+            return dict(
+                schema=SCHEMA,
+                source_id=self.archive.source if self.archive else None,
+                latest=self.latest,
+                definitions=self.definitions,
+                storage=self.storage,
+                database_bytes=self.database_bytes,
+                sample_interval_seconds=1,
+                retention_days=7,
+                database_budget_bytes=2 * 1024**3,
+                free_reserve_bytes=2 * 1024**3,
+            )
+
+
+class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer):
+    daemon_threads = True
+    block_on_close = False
+
+    def __init__(self, collector):
+        self.collector = collector
+        self.capacity = threading.BoundedSemaphore(6)
+        super().__init__(str(SOCKET), Handler)
+
+    def process_request(self, request, address):
+        if not self.capacity.acquire(False):
+            request.close()
+            return
+        request.settimeout(5)
+        super().process_request(request, address)
+
+    def process_request_thread(self, request, address):
+        try:
+            super().process_request_thread(request, address)
+        finally:
+            self.capacity.release()
+
+
+class Handler(BaseHTTPRequestHandler):
+    def log_message(self, *_):
+        pass
+
+    def reply(self, value, status=200):
+        payload = json.dumps(value, allow_nan=False, separators=(",", ":")).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def do_GET(self):
+        path = urlsplit(self.path)
+        try:
+            if path.path == "/status":
+                self.reply(self.server.collector.status())
+            elif path.path == "/batch":
+                after = int(parse_qs(path.query).get("after", ["0"])[0])
+                if not 0 <= after <= 2**53 - 1:
+                    raise ValueError()
+                archive = self.server.collector.archive
+                if archive is None:
+                    self.reply({"error": "Archive unavailable"}, 503)
+                else:
+                    self.reply(archive.batch(after))
+            else:
+                self.reply({"error": "Unknown request"}, 404)
+        except Exception:
+            self.reply({"error": "Archive unavailable"}, 503)
+
+    def do_POST(self):
+        try:
+            size = int(self.headers.get("Content-Length", "0"))
+            if self.path != "/events" or not 0 < size <= 2048:
+                raise ValueError()
+            self.server.collector.event(json.loads(self.rfile.read(size)))
+            self.reply({"ok": True})
+        except (ValueError, TypeError):
+            self.reply({"error": "Invalid event"}, 400)
+
+
+def main():
+    logging.basicConfig(level=logging.WARNING)
+    os.umask(0o007)
+    collector = Collector()
+    from journal_events import watch
+
+    threading.Thread(
+        target=watch, args=(collector,), name="node-system-events", daemon=True
+    ).start()
+    SOCKET.unlink(missing_ok=True)
+    server = Server(collector)
+    worker = threading.Thread(target=collector.run, name="node-system-sampler", daemon=True)
+    worker.start()
+
+    def stop(*_):
+        collector.stop.set()
+        threading.Thread(target=server.shutdown, daemon=True).start()
+
+    signal.signal(signal.SIGTERM, stop)
+    signal.signal(signal.SIGINT, stop)
+    try:
+        server.serve_forever(poll_interval=0.5)
+    finally:
+        collector.stop.set()
+        server.server_close()
+        worker.join(5)
+        SOCKET.unlink(missing_ok=True)
+
+
+if __name__ == "__main__":
+    main()

apps/node-agent/monitor/journal_events.py

@@ -0,0 +1,103 @@
+"""Classify bounded journal records; never persist free-form messages or payloads."""
+
+import json
+import re
+import subprocess
+import threading
+
+RULES = (
+    ("system-oom", r"out of memory|oom-kill|killed process"),
+    ("gpu-reset", r"GPU HANG|GPU reset|Resetting.*(gpu|chip)|i915.*(hang|reset|error)"),
+    ("disk-error", r"I/O error|EXT4-fs error|nvme.*(timeout|reset)|ata.*failed command"),
+    ("usb-disconnected", r"USB disconnect"),
+    ("usb-error", r"usb.*(error -|unable to enumerate|reset.*device)"),
+    ("service-failed", r"Failed with result|Main process exited|Failed to start"),
+    ("ui-process-exit", r"node-ui-process-terminated"),
+    ("ui-load-failed", r"node-ui-load-failed"),
+)
+
+
+def classify(record):
+    message = record.get("MESSAGE")
+    if not isinstance(message, str):
+        return None
+    code = next((code for code, pattern in RULES if re.search(pattern, message, re.I)), None)
+    if code is None:
+        return None
+    try:
+        at = int(record["__REALTIME_TIMESTAMP"]) / 1_000_000
+    except (KeyError, ValueError, TypeError):
+        return None
+    return dict(
+        code=code,
+        kind="Kernel" if record.get("_TRANSPORT") == "kernel" else "System",
+        at=at,
+        locations=[],
+    )
+
+
+def watch(collector):
+    args = [
+        "/usr/bin/journalctl",
+        "--follow",
+        "--lines=0",
+        "--output=json",
+        "--no-pager",
+        "_TRANSPORT=kernel",
+        "+",
+        "SYSLOG_IDENTIFIER=mission-core-node-ui",
+        "+",
+        "_SYSTEMD_UNIT=mission-core-node.service",
+        "+",
+        "_SYSTEMD_UNIT=mission-core-k1.service",
+        "+",
+        "_SYSTEMD_UNIT=mission-core-node-monitor.service",
+        "+",
+        "_SYSTEMD_UNIT=postgresql@16-ndc-monitor.service",
+    ]
+    for unit in (
+        "mission-core-node.service",
+        "mission-core-k1.service",
+        "mission-core-node-monitor.service",
+        "postgresql@16-ndc-monitor.service",
+    ):
+        args.extend(["+", "UNIT=" + unit])
+    while not collector.stop.is_set():
+        process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
+
+        def retire(process=process):
+            while process.poll() is None:
+                if collector.stop.wait(1):
+                    if process.poll() is None:
+                        process.terminate()
+                    return
+
+        guard = threading.Thread(target=retire, daemon=True)
+        guard.start()
+        try:
+            while not collector.stop.is_set():
+                line = process.stdout.readline(65537)
+                if not line:
+                    break
+                if len(line) > 65536:
+                    while line and not line.endswith(b"\n"):
+                        line = process.stdout.readline(65537)
+                    continue
+                try:
+                    event = classify(json.loads(line))
+                except (ValueError, TypeError):
+                    continue
+                if event:
+                    with collector.lock:
+                        if len(collector.events) < 64:
+                            collector.events.append(event)
+        finally:
+            if process.poll() is None:
+                process.terminate()
+            try:
+                process.wait(timeout=3)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+            process.stdout.close()
+        collector.stop.wait(10)

apps/node-agent/monitor/linux_metrics.py

@@ -0,0 +1,353 @@
+"""Read-only Linux counters. Missing or reset counters never manufacture zero."""
+
+import os
+import re
+import time
+from pathlib import Path
+
+
+class LinuxMetrics:
+    def __init__(self, root=Path("/"), clock=time.monotonic):
+        self.root, self.clock = Path(root), clock
+        self.previous = {}
+        self.previous_time = None
+        self.definitions = {}
+
+    def text(self, path):
+        try:
+            return (self.root / str(path).lstrip("/")).read_text()[:262144].strip()
+        except (OSError, UnicodeError):
+            return None
+
+    def number(self, path):
+        try:
+            return float(self.text(path))
+        except (ValueError, TypeError):
+            return None
+
+    def paths(self, pattern):
+        return sorted(self.root.glob(pattern))[:128]
+
+    def sample(self):
+        now = self.clock()
+        elapsed = None if self.previous_time is None else now - self.previous_time
+        self.previous_time = now
+        values, definitions, counters = {}, {}, {}
+
+        def metric(
+            key, label, group, resource, unit, value, reason="Счётчик недоступен в этой системе"
+        ):
+            values[key] = None if value is None else round(value, 3)
+            definitions[key] = dict(
+                label=label,
+                group=group,
+                resource=resource,
+                unit=unit,
+                reason=reason if value is None else None,
+            )
+
+        def delta(key, value):
+            if value is None:
+                return None
+            counters[key] = value
+            previous = self.previous.get(key)
+            return (
+                None
+                if previous is None or value < previous or not elapsed or elapsed > 10
+                else (value - previous) / elapsed
+            )
+
+        stat = self.text("proc/stat") or ""
+        cpus = 0
+        for line in stat.splitlines():
+            parts = line.split()
+            if not re.fullmatch(r"cpu\d*", parts[0]):
+                continue
+            if parts[0] != "cpu":
+                cpus += 1
+            data = [int(v) for v in parts[1:9]]
+            total = delta(parts[0] + ".total", sum(data))
+            idle = delta(parts[0] + ".idle", data[3] + data[4])
+            wait = delta(parts[0] + ".iowait", data[4])
+            for name, label, value in [
+                (
+                    "usage",
+                    "Загрузка",
+                    None if not total or idle is None else 100 * (1 - idle / total),
+                ),
+                (
+                    "iowait",
+                    "Ожидание диска",
+                    None if not total or wait is None else 100 * wait / total,
+                ),
+            ]:
+                metric(
+                    parts[0] + "." + name,
+                    label,
+                    "CPU",
+                    parts[0],
+                    "%",
+                    value,
+                    "Ожидаем два последовательных измерения",
+                )
+        if not cpus:
+            metric("cpu.usage", "Загрузка", "CPU", "CPU", "%", None)
+        memory = {}
+        for line in (self.text("proc/meminfo") or "").splitlines():
+            parts = line.replace(":", "").split()
+            if len(parts) > 1:
+                memory[parts[0]] = int(parts[1]) * 1024
+        for key, label in [
+            ("MemTotal", "Всего"),
+            ("MemAvailable", "Доступно"),
+            ("SwapTotal", "Swap всего"),
+            ("SwapFree", "Swap свободно"),
+            ("Dirty", "Ожидает записи"),
+            ("Writeback", "Записывается"),
+        ]:
+            metric("memory." + key, label, "RAM", "Память", "bytes", memory.get(key))
+        total, available = memory.get("MemTotal"), memory.get("MemAvailable")
+        metric(
+            "memory.usage",
+            "Использовано",
+            "RAM",
+            "Память",
+            "%",
+            None if not total or available is None else 100 * (1 - available / total),
+        )
+        for index, load in enumerate((self.text("proc/loadavg") or "").split()[:3]):
+            metric(
+                "load." + str(index),
+                ["За 1 минуту", "За 5 минут", "За 15 минут"][index],
+                "CPU",
+                "Load average",
+                "",
+                float(load),
+            )
+        for resource in ["cpu", "memory", "io"]:
+            for line in (self.text("proc/pressure/" + resource) or "").splitlines():
+                parts = line.split()
+                fields = dict(v.split("=") for v in parts[1:])
+                metric(
+                    "pressure." + resource + "." + parts[0],
+                    "Задержка за 10 секунд",
+                    "Ожидание ресурсов",
+                    resource + " · " + parts[0],
+                    "%",
+                    float(fields["avg10"]),
+                )
+        for interface in self.paths("sys/class/net/*"):
+            name = interface.name
+            for field, label in [
+                ("rx_bytes", "Приём"),
+                ("tx_bytes", "Передача"),
+                ("rx_errors", "Ошибки приёма"),
+                ("tx_errors", "Ошибки передачи"),
+                ("rx_dropped", "Потери приёма"),
+                ("tx_dropped", "Потери передачи"),
+            ]:
+                raw = self.number(interface.relative_to(self.root) / "statistics" / field)
+                metric(
+                    "net." + name + "." + field,
+                    label,
+                    "Сеть",
+                    name,
+                    "bytes/s" if field.endswith("bytes") else "1/s",
+                    delta("net." + name + "." + field, raw),
+                )
+            speed = self.number(interface.relative_to(self.root) / "speed")
+            metric(
+                "net." + name + ".speed",
+                "Скорость соединения",
+                "Сеть",
+                name,
+                "Mbit/s",
+                speed if speed is not None and speed > 0 else None,
+            )
+        diskstats = (self.text("proc/diskstats") or "").splitlines()
+        for line in diskstats:
+            p = line.split()
+            if (
+                len(p) < 14
+                or p[2].startswith(("loop", "ram"))
+                or not (self.root / "sys/block" / p[2]).exists()
+            ):
+                continue
+            name = p[2]
+            for index, label, unit, multiplier in [
+                (5, "Чтение", "bytes/s", 512),
+                (9, "Запись", "bytes/s", 512),
+                (3, "Операции чтения", "1/s", 1),
+                (7, "Операции записи", "1/s", 1),
+                (12, "Занятость", "%", 0.1),
+            ]:
+                key = "disk." + name + "." + str(index)
+                rate = delta(key, int(p[index]))
+                metric(key, label, "Диски", name, unit, None if rate is None else rate * multiplier)
+        if self.root == Path("/"):
+            mounts = set()
+            for line in (self.text("proc/mounts") or "").splitlines():
+                p = line.split()
+                if len(p) < 3 or not p[0].startswith("/dev/") or p[1] in mounts:
+                    continue
+                mount = p[1].replace("\\040", " ")
+                mounts.add(mount)
+                try:
+                    st = os.statvfs(mount)
+                    key = "fs." + str(len(mounts))
+                    metric(
+                        key + ".free",
+                        "Свободно",
+                        "Хранилище",
+                        mount,
+                        "bytes",
+                        st.f_bavail * st.f_frsize,
+                    )
+                    metric(
+                        key + ".usage",
+                        "Использовано",
+                        "Хранилище",
+                        mount,
+                        "%",
+                        100 * (1 - st.f_bfree / st.f_blocks) if st.f_blocks else None,
+                    )
+                    metric(
+                        key + ".inodes",
+                        "Использовано inode",
+                        "Хранилище",
+                        mount,
+                        "%",
+                        100 * (1 - st.f_ffree / st.f_files) if st.f_files else None,
+                    )
+                except OSError:
+                    metric(
+                        "fs." + str(len(mounts)) + ".free",
+                        "Свободно",
+                        "Хранилище",
+                        mount,
+                        "bytes",
+                        None,
+                    )
+        for path in self.paths("sys/class/hwmon/hwmon*/temp*_input"):
+            relative = path.relative_to(self.root)
+            name = self.text(path.parent.relative_to(self.root) / "name") or path.parent.name
+            raw = self.number(relative)
+            metric(
+                "temperature." + path.parent.name + "." + path.stem,
+                "Температура",
+                "Температура",
+                name + " · " + path.stem,
+                "°C",
+                None if raw is None else raw / 1000,
+            )
+        for card in self.paths("sys/class/drm/card[0-9]"):
+            base = card.relative_to(self.root)
+            metric(
+                "gpu." + card.name + ".usage",
+                "Загрузка",
+                "GPU",
+                card.name,
+                "%",
+                self.number(base / "device/gpu_busy_percent"),
+                "Драйвер не публикует общий счётчик загрузки GPU",
+            )
+            metric(
+                "gpu." + card.name + ".frequency",
+                "Частота",
+                "GPU",
+                card.name,
+                "MHz",
+                self.number(base / "gt_cur_freq_mhz"),
+            )
+            for field, label in [
+                ("mem_info_vram_used", "Видеопамять занята"),
+                ("mem_info_vram_total", "Видеопамять всего"),
+            ]:
+                metric(
+                    "gpu." + card.name + "." + field,
+                    label,
+                    "GPU",
+                    card.name,
+                    "bytes",
+                    self.number(base / "device" / field),
+                    "Выделенная видеопамять или её счётчик недоступны",
+                )
+        for usb in self.paths("sys/bus/usb/devices/*"):
+            if not (usb / "idVendor").exists():
+                continue
+            base = usb.relative_to(self.root)
+            name = usb.name
+            metric(
+                "usb." + name + ".speed",
+                "Скорость соединения",
+                "USB",
+                name,
+                "Mbit/s",
+                self.number(base / "speed"),
+            )
+            metric(
+                "usb." + name + ".traffic",
+                "Трафик",
+                "USB",
+                name,
+                "bytes/s",
+                None,
+                "USB-драйвер не публикует счётчик трафика порта",
+            )
+        groups = self.paths("sys/fs/cgroup/system.slice/mission-core*.service")
+        groups += self.paths(
+            "sys/fs/cgroup/system.slice/system-postgresql.slice/postgresql@16-ndc-monitor.service"
+        )
+        groups += self.paths(
+            "sys/fs/cgroup/user.slice/user-*.slice/user@*.service/app.slice/app-gnome-org.nodedc.MissionCoreNode-*.scope"
+        )
+        for path in groups:
+            base = path.relative_to(self.root)
+            name = path.name
+            key = "service." + re.sub("[^A-Za-z0-9_.-]", "_", name)
+            metric(
+                key + ".memory",
+                "Память",
+                "Приложения",
+                name,
+                "bytes",
+                self.number(base / "memory.current"),
+            )
+            metric(
+                key + ".peak",
+                "Пик памяти",
+                "Приложения",
+                name,
+                "bytes",
+                self.number(base / "memory.peak"),
+            )
+            fields = dict(
+                line.split() for line in (self.text(base / "cpu.stat") or "").splitlines()
+            )
+            rate = delta(
+                key + ".cpu", float(fields["usage_usec"]) if "usage_usec" in fields else None
+            )
+            metric(
+                key + ".cpu",
+                "Загрузка CPU",
+                "Приложения",
+                name,
+                "%",
+                None if rate is None else rate / 10000 / max(cpus, 1),
+            )
+            fields = dict(
+                line.split() for line in (self.text(base / "memory.events") or "").splitlines()
+            )
+            metric(
+                key + ".oom",
+                "Убийств из-за нехватки памяти",
+                "Приложения",
+                name,
+                "",
+                float(fields["oom_kill"]) if "oom_kill" in fields else None,
+            )
+        self.previous = counters
+        # A bounded hardware inventory protects the control-plane payload budget.
+        keys = list(values)[:384]
+        self.definitions = {key: definitions[key] for key in keys}
+        return {key: values[key] for key in keys}

apps/node-agent/monitor/schema.sql

@@ -0,0 +1,13 @@
+CREATE EXTENSION IF NOT EXISTS timescaledb;
+CREATE TABLE IF NOT EXISTS monitor_identity (singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton), source_id text NOT NULL);
+INSERT INTO monitor_identity(singleton,source_id) VALUES(true,gen_random_uuid()::text) ON CONFLICT DO NOTHING;
+CREATE TABLE IF NOT EXISTS monitor_samples (
+ seq bigint GENERATED ALWAYS AS IDENTITY,
+ observed_at timestamptz NOT NULL,
+ payload jsonb NOT NULL,
+ PRIMARY KEY(observed_at,seq)
+);
+SELECT create_hypertable('monitor_samples',by_range('observed_at',INTERVAL '1 hour'),if_not_exists=>true);
+CREATE INDEX IF NOT EXISTS monitor_samples_cursor ON monitor_samples(seq);
+CREATE TABLE IF NOT EXISTS monitor_definitions(singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton),payload jsonb NOT NULL);
+REVOKE ALL ON DATABASE mission_core_monitor FROM PUBLIC;

apps/node-agent/monitor/storage.py

@@ -0,0 +1,118 @@
+"""Bounded local Timescale archive; transport acknowledgements never delete data."""
+
+import json
+import shutil
+from contextlib import contextmanager
+from datetime import UTC, datetime
+
+SCHEMA = "missioncore.node-system-monitor/v1"
+MAX_BATCH_BYTES = 196608
+DATABASE_BUDGET = 2 * 1024**3
+
+
+class Archive:
+    def __init__(self):
+        from psycopg2.pool import ThreadedConnectionPool
+
+        self.pool = ThreadedConnectionPool(
+            1,
+            4,
+            host="/run/mission-core-monitor-db",
+            port=5433,
+            dbname="mission_core_monitor",
+            user="mission-core-monitor",
+            connect_timeout=3,
+            options="-c statement_timeout=3000",
+        )
+        try:
+            with self.connection() as db, db.cursor() as cursor:
+                cursor.execute("SELECT source_id FROM monitor_identity")
+                self.source = cursor.fetchone()[0]
+        except Exception:
+            self.pool.closeall()
+            raise
+        self.definitions = None
+        self.database_bytes = None
+
+    @contextmanager
+    def connection(self):
+        db = self.pool.getconn()
+        try:
+            yield db
+            db.commit()
+        except BaseException:
+            db.rollback()
+            raise
+        finally:
+            self.pool.putconn(db)
+
+    def append(self, sample, definitions):
+        # This reserve belongs to telemetry, never to acquisition storage.
+        if shutil.disk_usage("/var/lib/mission-core-monitor").free < 2 * 1024**3:
+            raise RuntimeError("disk-reserve")
+        if self.database_bytes is None or self.database_bytes >= DATABASE_BUDGET:
+            raise RuntimeError("archive-budget")
+        payload = json.dumps(sample, separators=(",", ":"), allow_nan=False)
+        if len(payload.encode()) > 32768:
+            raise ValueError("sample-budget")
+        with self.connection() as db, db.cursor() as cursor:
+            if definitions != self.definitions:
+                cursor.execute(
+                    "INSERT INTO monitor_definitions VALUES(true,%s) ON CONFLICT(singleton) "
+                    "DO UPDATE SET payload=excluded.payload",
+                    (json.dumps(definitions),),
+                )
+            cursor.execute(
+                "INSERT INTO monitor_samples(observed_at,payload) VALUES(%s,%s) RETURNING seq",
+                (datetime.fromtimestamp(sample["at"], UTC), payload),
+            )
+            seq = cursor.fetchone()[0]
+        self.definitions = definitions
+        return {**sample, "seq": seq}
+
+    def batch(self, after):
+        with self.connection() as db, db.cursor() as cursor:
+            cursor.execute(
+                "SELECT seq,payload FROM monitor_samples WHERE seq>%s ORDER BY seq LIMIT 32",
+                (after,),
+            )
+            rows = []
+            size = 0
+            for seq, payload in cursor.fetchall():
+                record = {**payload, "seq": seq}
+                size += len(json.dumps(record).encode())
+                if size > MAX_BATCH_BYTES:
+                    break
+                rows.append(record)
+            cursor.execute("SELECT min(seq),max(seq),min(observed_at) FROM monitor_samples")
+            first, last, oldest = cursor.fetchone()
+        return dict(
+            schema=SCHEMA,
+            source_id=self.source,
+            samples=rows,
+            first_seq=first,
+            last_seq=last,
+            retained_since=oldest.timestamp() if oldest else None,
+        )
+
+    def maintain(self):
+        with self.connection() as db, db.cursor() as cursor:
+            cursor.execute("SELECT drop_chunks('monitor_samples',INTERVAL '7 days')")
+            cursor.execute("SELECT pg_database_size(current_database())")
+            size = cursor.fetchone()[0]
+            if size >= DATABASE_BUDGET:
+                # Drop one oldest completed chunk per pass, retaining the live hour.
+                cursor.execute(
+                    "SELECT min(range_end) FROM timescaledb_information.chunks "
+                    "WHERE hypertable_name='monitor_samples' AND range_end<NOW()"
+                )
+                boundary = cursor.fetchone()[0]
+                if boundary:
+                    cursor.execute(
+                        "SELECT drop_chunks('monitor_samples',older_than=>%s::timestamptz)",
+                        (boundary,),
+                    )
+                cursor.execute("SELECT pg_database_size(current_database())")
+                size = cursor.fetchone()[0]
+        self.database_bytes = size
+        return size

apps/node-agent/packaging/build_deb.py

@@ -11,7 +11,7 @@ import sys
 
 
 ROOT = Path(__file__).resolve().parents[1]
-VERSION = "0.8.14"
+VERSION = "0.8.15"
 sys.path.insert(0, str(ROOT.parents[1] / "scripts/packaging"))
 from debian import package
 
@@ -43,7 +43,7 @@ Architecture: amd64
 Maintainer: NODE.DC local build <noreply@example.invalid>
 Section: admin
 Priority: optional
-Depends: adduser, systemd, python3, python3-gi, gir1.2-gtk-3.0, gir1.2-webkit2-4.1, pkexec, polkitd, ca-certificates, hicolor-icon-theme, network-manager, iproute2
+Depends: adduser, systemd, python3, python3-gi, gir1.2-gtk-3.0, gir1.2-webkit2-4.1, pkexec, polkitd, ca-certificates, hicolor-icon-theme, network-manager, iproute2, python3-psycopg2, postgresql-16 (>= 16.15), timescaledb-2-oss-postgresql-16 (= 2.29.2~ubuntu24.04-1615)
 Description: Mission Core onboard computer configuration
  Local graphical setup, host inventory, SSH access and persistent node identity.
 """.encode()
@@ -67,8 +67,13 @@ Description: Mission Core onboard computer configuration
         ("environment_helper.py", "usr/lib/mission-core-node/environment_helper.py", 0o644),
         ("mission-core-node-environment.service", "usr/lib/systemd/system/mission-core-node-environment.service", 0o644),
         ("60-environment.conf", "usr/share/mission-core-node/60-environment.conf", 0o644),
+        ("setup-monitor", "usr/lib/mission-core-node/setup-monitor", 0o755),
+        ("mission-core-node-monitor.service", "usr/lib/systemd/system/mission-core-node-monitor.service", 0o644),
     ]:
         files.append((path, (p / source).read_bytes(), mode))
+    for path in sorted((ROOT / 'monitor').iterdir()):
+        if path.suffix in {'.py', '.sql'}:
+            files.append(('usr/lib/mission-core-node/monitor/' + path.name, path.read_bytes(), 0o644))
     files.append(("usr/share/mission-core-node/environment-profile.json", (ROOT / "internal/node/environment-profile.json").read_bytes(), 0o644))
     for name in ("realsense_prepare.py", "realsense_iio_access.py"):
         files.append(("usr/lib/mission-core-node/" + name, (p / name).read_bytes(), 0o644))

apps/node-agent/packaging/fetch_monitor_dependencies.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Stage verified Timescale OSS Ubuntu packages, without APT/system mutation."""
+
+import argparse
+import gzip
+import hashlib
+import json
+import subprocess
+import urllib.request
+from pathlib import Path
+
+BASE = "https://packagecloud.io/timescale/timescaledb/ubuntu/"
+VERSION = "2.29.2"
+
+
+def fetch(url, limit):
+    with urllib.request.urlopen(url, timeout=30) as response:
+        data = response.read(limit + 1)
+    if len(data) > limit:
+        raise ValueError("Download budget exceeded")
+    return data
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("destination", type=Path)
+    args = parser.parse_args()
+    root = args.destination
+    root.mkdir(parents=True, exist_ok=True)
+    key = root / "timescale.asc"
+    key.write_bytes(fetch("https://packagecloud.io/timescale/timescaledb/gpgkey", 65536))
+    keyring = root / "timescale.gpg"
+    subprocess.run(
+        ["gpg", "--batch", "--yes", "--dearmor", "--output", str(keyring), str(key)], check=True
+    )
+    release = root / "InRelease"
+    release.write_bytes(fetch(BASE + "dists/noble/InRelease", 1048576))
+    subprocess.run(["gpgv", "--keyring", str(keyring.resolve()), str(release)], check=True)
+    lines = release.read_text().splitlines()
+    hashes = {}
+    inside = False
+    for line in lines:
+        if line == "SHA256:":
+            inside = True
+            continue
+        if inside and not line.startswith(" "):
+            break
+        if inside:
+            sha, size, path = line.split()
+            hashes[path] = (sha, int(size))
+    sha, size = hashes["main/binary-amd64/Packages.gz"]
+    compressed = fetch(BASE + "dists/noble/main/binary-amd64/Packages.gz", 16000000)
+    if len(compressed) != size or hashlib.sha256(compressed).hexdigest() != sha:
+        raise ValueError("Signed package index mismatch")
+    (root / "Packages.gz").write_bytes(compressed)
+    wanted = {
+        name: VERSION + "~ubuntu24.04-1615"
+        for name in ("timescaledb-2-oss-postgresql-16", "timescaledb-2-loader-postgresql-16")
+    }
+    artifacts = []
+    for block in gzip.decompress(compressed).decode().split("\n\n"):
+        fields = dict(
+            line.split(": ", 1)
+            for line in block.splitlines()
+            if ": " in line and not line.startswith(" ")
+        )
+        if (
+            fields.get("Package") not in wanted
+            or fields.get("Version") != wanted[fields["Package"]]
+        ):
+            continue
+        path = fields["Filename"]
+        if not path.startswith("pool/") or ".." in path.split("/"):
+            raise ValueError("Invalid package path")
+        data = fetch(BASE + path, 64000000)
+        if hashlib.sha256(data).hexdigest() != fields["SHA256"]:
+            raise ValueError("Package hash mismatch")
+        target = root / Path(path).name
+        target.write_bytes(data)
+        artifacts.append(
+            dict(
+                package=fields["Package"],
+                version=fields["Version"],
+                file=target.name,
+                sha256=fields["SHA256"],
+                bytes=len(data),
+                depends=fields["Depends"],
+            )
+        )
+    if len(artifacts) != len(wanted):
+        raise ValueError("Pinned package not found")
+    manifest = dict(
+        source=BASE,
+        version=VERSION,
+        key_sha256=hashlib.sha256(key.read_bytes()).hexdigest(),
+        inrelease_sha256=hashlib.sha256(release.read_bytes()).hexdigest(),
+        artifacts=artifacts,
+    )
+    (root / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n")
+    print(json.dumps(manifest, indent=2))
+
+
+if __name__ == "__main__":
+    main()

apps/node-agent/packaging/install-monitor-release

@@ -0,0 +1,56 @@
+#!/bin/sh
+# Run by the owner-facing release installer after hash verification. Scope is
+# this product's packages and its dedicated cluster; never remove another DB.
+set -eu
+test "$(id -u)" = 0
+mc_release_dir=$1
+cd "$mc_release_dir"
+sha256sum --check SHA256SUMS
+mc_pg_guard=/etc/postgresql-common/createcluster.d/60-mission-core-install.conf
+mc_created_guard=0
+cleanup() {
+  if [ "$mc_created_guard" = 1 ]; then
+    if [ "$(cat "$mc_pg_guard")" = 'create_main_cluster = false' ]; then
+      rm "$mc_pg_guard"
+    fi
+  fi
+}
+trap cleanup EXIT
+trap 'exit 130' INT
+trap 'exit 143' HUP TERM
+if ! dpkg-query -W -f='${Status}' postgresql-16 2>/dev/null | grep -qx 'install ok installed'; then
+  # postgresql-common explicitly supports this drop-in. Suppress only automatic
+  # creation of the unrelated default main cluster during the first install.
+  test ! -e "$mc_pg_guard"
+  install -d -m 0755 /etc/postgresql-common/createcluster.d
+  printf '%s\n' 'create_main_cluster = false' > "$mc_pg_guard"
+  chmod 0644 "$mc_pg_guard"
+  mc_created_guard=1
+fi
+DEBIAN_FRONTEND=noninteractive apt-get install -y --no-remove \
+  "$mc_release_dir/timescaledb-2-loader-postgresql-16_2.29.2~ubuntu24.04-1615_amd64.deb" \
+  "$mc_release_dir/timescaledb-2-oss-postgresql-16_2.29.2~ubuntu24.04-1615_amd64.deb" \
+  "$mc_release_dir/mission-core-node_0.8.15_amd64.deb" \
+  "$mc_release_dir/mission-core-xgrids-k1_0.1.14_amd64.deb"
+systemctl is-active --quiet mission-core-node.service
+systemctl is-active --quiet mission-core-k1.service
+systemctl is-active --quiet postgresql@16-ndc-monitor.service
+systemctl is-active --quiet mission-core-node-monitor.service
+runuser -u mission-core-monitor -- /usr/bin/python3 - <<'PY'
+import http.client,json,socket,time
+deadline=time.monotonic()+20
+while True:
+    c=http.client.HTTPConnection('monitor',timeout=3)
+    c.sock=socket.socket(socket.AF_UNIX);c.sock.settimeout(3)
+    try:
+        c.sock.connect('/run/mission-core-monitor/monitor.sock')
+        c.request('GET','/status');r=c.getresponse();v=json.load(r)
+        if r.status==200 and v['storage']=='ready' and v['latest']['seq']>0:break
+    except (OSError, ValueError, KeyError, TypeError, http.client.HTTPException):
+        pass
+    finally:c.close()
+    if time.monotonic()>deadline:raise SystemExit('Local telemetry did not become ready')
+    time.sleep(1)
+print('Local telemetry: ready')
+PY
+printf '%s\n' 'Mission Core Node R18: package and local archive checks passed.'

apps/node-agent/packaging/launcher.py

@@ -9,6 +9,7 @@ import re
 import socket
 import subprocess
 import threading
+import syslog
 from urllib.parse import urlsplit
 
 import gi
@@ -261,11 +262,15 @@ class NodeApplication(Gtk.Application):
         return False
 
     def load_failed(self, _view, _event, uri, _error):
+        syslog.openlog('mission-core-node-ui')
+        syslog.syslog(syslog.LOG_ERR, 'node-ui-load-failed')
         if local_url(uri):
             self.problem("Служба ноды недоступна. Повторно откройте приложение после восстановления службы.")
         return True
 
-    def process_failed(self, *_):
+    def process_failed(self, _view, reason):
+        syslog.openlog('mission-core-node-ui')
+        syslog.syslog(syslog.LOG_ERR, 'node-ui-process-terminated reason=' + str(int(reason)))
         self.problem("Окно приложения остановилось. Закройте и повторно откройте Mission Core Node. Служба борта продолжает работать отдельно.")
 
     def deny_permission(self, _view, permission):

apps/node-agent/packaging/mission-core-node-monitor.service

@@ -0,0 +1,38 @@
+[Unit]
+Description=Mission Core onboard system telemetry collector
+After=postgresql@16-ndc-monitor.service
+Wants=postgresql@16-ndc-monitor.service
+
+[Service]
+Type=simple
+User=mission-core-monitor
+Group=mission-core-node
+SupplementaryGroups=systemd-journal
+ExecStart=/usr/bin/python3 /usr/lib/mission-core-node/monitor/collector.py
+RuntimeDirectory=mission-core-monitor
+RuntimeDirectoryMode=0750
+StateDirectory=mission-core-monitor
+StateDirectoryMode=0750
+UMask=0007
+Restart=on-failure
+RestartSec=5
+MemoryHigh=96M
+MemoryMax=128M
+CPUQuota=15%
+TasksMax=24
+Nice=10
+IOSchedulingClass=idle
+NoNewPrivileges=yes
+ProtectSystem=strict
+ProtectHome=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+RestrictAddressFamilies=AF_UNIX
+CapabilityBoundingSet=
+RestrictSUIDSGID=yes
+
+[Install]
+WantedBy=multi-user.target

apps/node-agent/packaging/postinst

@@ -23,6 +23,7 @@ case "$1" in
         fi
         rm -f /run/mission-core-node-k1-upgrade-active
       fi
+      /usr/lib/mission-core-node/setup-monitor
     fi
     ;;
 esac

apps/node-agent/packaging/preinst

@@ -34,6 +34,7 @@ if [ "$1" = install ] || [ "$1" = upgrade ]; then
       (umask 077; : > /run/mission-core-node-k1-upgrade-active)
       systemctl stop mission-core-k1.service
     fi
+    systemctl stop mission-core-node-monitor.service 2>/dev/null || true
   fi
   . /etc/os-release
   if [ "${ID:-}" != ubuntu ] || [ "${VERSION_ID:-}" != 24.04 ]; then

apps/node-agent/packaging/prerm

@@ -52,6 +52,10 @@ case "$1" in
       systemctl disable mission-core-realsense.service || true
       systemctl stop mission-core-node.service
       systemctl disable mission-core-node.service
+      systemctl stop mission-core-node-monitor.service
+      systemctl disable mission-core-node-monitor.service
+      systemctl stop postgresql@16-ndc-monitor.service
+      systemctl disable postgresql@16-ndc-monitor.service
     fi
     ;;
 esac

apps/node-agent/packaging/setup-monitor

@@ -0,0 +1,54 @@
+#!/bin/sh
+# Fixed local database bootstrap; no supplied SQL, credentials or network address.
+set -eu
+test "$(id -u)" = 0
+if ! getent passwd mission-core-monitor >/dev/null; then
+ adduser --system --group --home /var/lib/mission-core-monitor --no-create-home --disabled-login mission-core-monitor
+fi
+install -d -m 0750 -o mission-core-monitor -g mission-core-node /var/lib/mission-core-monitor
+install -d -m 0755 -o postgres -g postgres /run/mission-core-monitor-db
+install -d -m 0755 /etc/tmpfiles.d
+printf '%s\n' 'd /run/mission-core-monitor-db 0755 postgres postgres -' > /etc/tmpfiles.d/mission-core-monitor.conf
+if [ ! -d /etc/postgresql/16/ndc-monitor ]; then
+ pg_createcluster 16 ndc-monitor --port=5433 --socketdir=/run/mission-core-monitor-db --datadir=/var/lib/mission-core-monitor-db --start-conf=auto -- --auth-local=peer --auth-host=reject
+fi
+install -d -m 0755 /etc/postgresql/16/ndc-monitor/conf.d
+cat > /etc/postgresql/16/ndc-monitor/conf.d/60-mission-core-monitor.conf <<'CONF'
+listen_addresses = ''
+unix_socket_directories = '/run/mission-core-monitor-db'
+shared_preload_libraries = 'timescaledb'
+shared_buffers = '32MB'
+work_mem = '2MB'
+maintenance_work_mem = '16MB'
+max_connections = 12
+max_worker_processes = 4
+max_parallel_workers = 0
+timescaledb.max_background_workers = 2
+timescaledb.telemetry_level = 'off'
+max_wal_size = '128MB'
+min_wal_size = '32MB'
+temp_file_limit = '32MB'
+statement_timeout = '5s'
+log_statement = 'none'
+log_min_error_statement = 'panic'
+CONF
+install -d -m 0755 /etc/systemd/system/postgresql@16-ndc-monitor.service.d
+cat > /etc/systemd/system/postgresql@16-ndc-monitor.service.d/60-mission-core-monitor.conf <<'CONF'
+[Service]
+ExecStartPre=+/usr/bin/install -d -m 0755 -o postgres -g postgres /run/mission-core-monitor-db
+MemoryHigh=256M
+MemoryMax=384M
+CPUQuota=25%
+TasksMax=32
+Nice=10
+CONF
+systemctl daemon-reload
+systemctl restart postgresql@16-ndc-monitor.service
+runuser -u postgres -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d postgres <<'SQL'
+SELECT 'CREATE ROLE "mission-core-monitor" LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE' WHERE NOT EXISTS(SELECT FROM pg_roles WHERE rolname='mission-core-monitor') \gexec
+SELECT 'CREATE DATABASE mission_core_monitor OWNER "mission-core-monitor"' WHERE NOT EXISTS(SELECT FROM pg_database WHERE datname='mission_core_monitor') \gexec
+SQL
+runuser -u postgres -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d mission_core_monitor -c 'CREATE EXTENSION IF NOT EXISTS timescaledb'
+runuser -u mission-core-monitor -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d mission_core_monitor -f /usr/lib/mission-core-node/monitor/schema.sql
+systemctl enable mission-core-node-monitor.service
+systemctl restart mission-core-node-monitor.service

apps/node-agent/ui/src/NodeSensors.tsx

@@ -3,5 +3,5 @@ import {createIsolatedRerunHost} from '../../../control-station/src/components/r
 import {SensorWorkspace} from '../../../../packages/sensor-ui/src/SensorWorkspace';
 import type {SensorTransport} from '../../../../packages/sensor-ui/src/contracts';
 import {request} from './api';
-const transport:SensorTransport={enrollment:{state:()=>request("/api/devices/enrollment"),submit:value=>request("/api/devices/enrollment/operations","POST",value),operation:id=>request("/api/devices/enrollment/operations/"+encodeURIComponent(id))},subscribe:(receive,unavailable)=>{const events=new EventSource('/api/devices/events');events.onmessage=e=>{try{receive(JSON.parse(e.data));}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();},inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))};
+const transport:SensorTransport={localPreview:{open:(command,signal)=>fetch("/api/devices/preview",{method:"POST",credentials:"same-origin",headers:{"Content-Type":"application/json"},body:JSON.stringify(command),signal}),read:(peer,after,signal)=>fetch("/api/devices/preview/read",{method:"POST",credentials:"same-origin",headers:{"Content-Type":"application/json"},body:JSON.stringify({peer_id:peer,after}),signal})},enrollment:{state:()=>request("/api/devices/enrollment"),submit:value=>request("/api/devices/enrollment/operations","POST",value),operation:id=>request("/api/devices/enrollment/operations/"+encodeURIComponent(id))},subscribe:(receive,unavailable)=>{const events=new EventSource('/api/devices/events');events.onmessage=e=>{try{receive(JSON.parse(e.data));}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();},inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))};
 export function NodeSensors(){return <SensorWorkspace contributions={[xgridsK1SensorUi]} createRerunHost={createIsolatedRerunHost} transport={transport}/>;}

apps/node-agent/ui/src/RuntimeBoundary.tsx

@@ -0,0 +1,20 @@
+import {Component,type ReactNode} from 'react';
+import {Button,SettingsCard} from '@nodedc/ui-react';
+
+export function runtimeDiagnostic(error:unknown,code='ui-error') {
+ const kind=error instanceof Error&&/^[A-Za-z]{1,48}$/.test(error.name)?error.name:'Error';
+ // Never send exception messages, URLs, form values or arbitrary stack text.
+ const locations=error instanceof Error?(error.stack??'').match(/[A-Za-z0-9_-]+\.js:\d+:\d+/g)?.slice(0,8)??[]:[];
+ void fetch('/api/monitor/events',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({code,kind,locations}),keepalive:true}).catch(()=>{});
+}
+export function observeRuntime(){
+ window.addEventListener('error',event=>runtimeDiagnostic(event.error));
+ window.addEventListener('unhandledrejection',event=>runtimeDiagnostic(event.reason,'ui-rejection'));
+}
+export class RuntimeBoundary extends Component<{children:ReactNode;resetKey?:string},{failed:boolean}> {
+ state={failed:false};
+ static getDerivedStateFromError(){return {failed:true};}
+ componentDidCatch(error:Error){runtimeDiagnostic(error,'ui-render-error');}
+ componentDidUpdate(previous:Readonly<{children:ReactNode;resetKey?:string}>){if(this.state.failed&&previous.resetKey!==this.props.resetKey)this.setState({failed:false});}
+ render(){return this.state.failed?<SettingsCard title="Не удалось открыть представление" description="Вернитесь к устройствам или повторите открытие."><Button onClick={()=>this.setState({failed:false})}>Открыть повторно</Button></SettingsCard>:this.props.children;}
+}

apps/node-agent/ui/src/main.tsx

@@ -18,6 +18,7 @@ import "./node.css";
 import { NodeSensors } from "./NodeSensors";
 import { Home, HomeSettings } from "./Home";
 import { usePresentation } from "./usePresentation";
+import {RuntimeBoundary,observeRuntime} from './RuntimeBoundary';
 
 function App() {
   const node = useNode();
@@ -53,7 +54,7 @@ function App() {
         contexts={value ? [{ id: "board", label: value.name, description: value.host.hostname, icon: <Icon name="activity" />, onSelect: () => openView("overview") }] : []}
         items={views.filter(item => item.root === root).map(item => ({ id: item.id, label: item.label, icon: <Icon name={item.icon} /> }))} activeId={workspace.activeView ?? undefined} onItemChange={id => openView(id as ViewId)} footer={<span>Mission Core Node · {value?.version}</span>} />}
       content={currentView && <ApplicationPanel title={currentView.label} eyebrow={currentRoot.label} expanded={workspace.contentExpanded} onExpandedChange={workspace.setContentExpanded} onClose={workspace.closeView}
-        utilityActions={[...(workspace.activeView === "ssh" ? [{ label: "Добавить доверенное устройство", icon: "plus" as const, onClick: () => setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}>{content}</ApplicationPanel>}
+        utilityActions={[...(workspace.activeView === "ssh" ? [{ label: "Добавить доверенное устройство", icon: "plus" as const, onClick: () => setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}><RuntimeBoundary resetKey={workspace.activeView??undefined}>{content}</RuntimeBoundary></ApplicationPanel>}
       stage={value ? <Home page={presentation.settings.pages.home} openView={openView} /> : <div className="node-stage" aria-busy={pending}>
         <SettingsCard className="node-entry" title={pending ? "Подключаемся к ноде" : locked ? "Вход в Mission Core Node" : "Нода недоступна"}>
           {pending ? <ActivityIndicator label="Получение сведений о ноде" /> : <p className="node-note">{locked ? "Подтвердите доступ в системном окне." : "Не удалось связаться со службой. Повторите подключение."}</p>}
@@ -64,4 +65,5 @@ function App() {
     <HomeSettings open={!!value && settingsOpen} onClose={() => setSettingsOpen(false)} presentation={presentation} />
   </>;
 }
-createRoot(document.getElementById("root")!).render(<App />);
+observeRuntime();
+createRoot(document.getElementById("root")!).render(<RuntimeBoundary><App /></RuntimeBoundary>);

docs/audits/2026-09-08-node-runtime-and-system-monitor.md

@@ -0,0 +1,100 @@
+# Node local viewer and onboard system monitoring
+
+## Approved operator job and placement
+
+Owner request: investigate the black onboard window after local K1 START, and
+add «Мониторинг системы БК» to the selected apparatus in Fleet. The selected
+board owns measurements and history; UI navigation, capture and network outages
+must not control collection. This is a detail view in the existing Fleet
+workspace, not the compute-worker contour or a new primary navigation root.
+An independent System workspace would lose the selected apparatus context; a
+modal would constrain history inspection. The owner explicitly selected the
+apparatus-detail entry. Novelty class A: domain content in admitted composition.
+
+Reuse SettingsCard, ApplicationPanel, Button, Select, StatusBadge, ResourceRow,
+Icon(activity/refresh/chevron-left), and the existing TelemetrySeries domain
+renderer. Domain cards contain current metrics; selected series shows source
+time, min/max/mean and gaps. No missing generic visual primitive is required.
+States: collecting, historical/offline, waiting for first sample, unavailable
+metric, storage failure, replication backlog, loading/query failure, empty range.
+No acquisition, provisioning, shell or host-reconfiguration action belongs here.
+
+## Reproduced diagnosis
+
+R17 installed WebKitGTK 2.52.6 on Ubuntu. Public standalone Rerun 0.36.3 starts
+and stays responsive using WebGL2. WebGPU is unavailable. RTCPeerConnection is
+undefined, and constructing it raises ReferenceError. The shared preview effect
+constructs that API synchronously after viewer readiness, outside its catch.
+Node had no React error boundary. The historical window and WebKit processes
+remained alive with an empty accessible document; Node/K1 services did not restart.
+App cgroup memory peak was about 811 MiB and OOM counters were all zero. Ten-minute
+sysstat samples cannot rule out brief peaks but do not support memory exhaustion.
+The camera consumer lease subsequently expired from queue age; that is a delivery
+symptom, not proof of GPU/RAM exhaustion.
+
+Enabling WebKit's WebRTC setting alone and supplying missing GStreamer modules
+in an isolated diagnostic environment did not expose the API. No global graphics
+workaround or new browser engine is admitted on this evidence. The local carrier
+uses authenticated loopback HTTP and the same plugin-owned bounded RRD/fMP4
+delivery, acknowledgements, recording cursor and release logic. Remote paired
+WebRTC remains unchanged. There is no second capture implementation.
+
+## Telemetry architecture and limits
+
+An independent native collector samples Linux counters every second and commits
+to a dedicated local PostgreSQL/TimescaleDB database. It uses only an OS-protected
+Unix socket, with no database TCP listener or embedded credential. Acquisition
+does not depend on that service. CPU/RAM/swap/pressure, per-interface bytes and
+errors, per-disk I/O, filesystem capacity, temperatures, observable GPU counters,
+USB topology/link speed and service cgroup usage/OOM counters are separate series.
+Unsupported counters remain null with a reason; USB negotiated speed is never
+presented as measured traffic. No USB payload, process arguments or environment
+variables are collected. UI exceptions carry only category and asset positions.
+
+Raw retention target: seven days, with a two-GiB database target and two-GiB free
+disk reserve. Capacity is checked once per minute; writes pause at the target
+and oldest completed hourly chunks are removed. This is not a strict filesystem
+quota: a minute of writes, WAL and temporary files may exceed the database target. Retention and capacity loss are visible, never disguised as zeros.
+Bounded batches replicate over the existing paired mTLS connection. Core commits
+idempotently before acknowledging; disconnects leave local collection running.
+Latest status and historical delivery cursors are distinct so a fresh snapshot
+cannot skip unreplicated history. Historical timestamps and boot identity survive
+reconnection. Core's local archive is queryable while the board is offline.
+
+Acceptance must separate fixture checks, real host collection, local viewer
+compatibility and owner hardware scans. A healthy repeated laboratory scan does
+not establish critical-scenario qualification of the K1 Bridge contour.
+
+
+## Implementation acceptance before installation
+
+A bounded native WebKitGTK probe on the owner Ubuntu loads the unmodified
+production hook through a minified Vite bundle. Its source is synthetic RRD
+(100 points) and a synthetic H.264/fMP4 camera; it cannot submit device commands.
+The 2026-09-07 22:28 UTC probe reports the error boundary present with the shell
+alive, then `hasScene=true`, `presented=true`, `cameraPresented=true`. Video time
+advances to 2.9 s. After the fixture ends, freshness expires while the scene
+remains. This establishes compatibility for these fixtures, not live K1 load.
+Local transport uses completed bounded reads, a stable viewer cursor and native
+EventTarget dispatch. Remote transport keeps WebRTC with guarded construction.
+
+Native PostgreSQL 16.15 / TimescaleDB 2.29.2 was exercised using privately
+extracted signed packages and a temporary socket-only cluster. Three real host
+samples with 184 discovered metrics were committed and replayed in order. The
+temporary cluster was stopped and removed. Installed-service acceptance is a
+separate gate.
+
+Core keeps seven-day raw data plus minute min/max/sum/count rollups. Long ranges
+use complete minute buckets, preserving extrema; the selected end is rounded
+to the last complete minute. Missing intervals break the plot. Replica writes
+run outside the control heartbeat; only a durable committed cursor is ACKed.
+The archive is a best-effort diagnostic record at one-second sampling resolution;
+unobserved subsecond peaks and unsupported hardware counters remain unproved.
+
+Packages: Node 0.8.15 and K1 0.1.14, with exact signed Timescale OSS/loader
+2.29.2~ubuntu24.04-1615. The dedicated `16/ndc-monitor` PostgreSQL cluster has no
+TCP listener. Its service memory cap is 384 MiB, collector cap 128 MiB; CPU caps
+are 25% and 15% of one core respectively. An initial PostgreSQL install suppresses
+automatic creation of an unrelated default cluster through a temporary supported
+postgresql-common configuration drop-in, removed on installer exit. Acquisition
+is protected by the existing package preinst guard and is not a test action.

packages/sensor-ui/src/contracts.ts

@@ -26,6 +26,10 @@ export interface SensorCommand {
 }
 export interface SensorOperation {state:string;error?:string;result?:unknown}
 export interface SensorTransport {
+ localPreview?: {
+  open: (command:SensorCommand, signal:AbortSignal) => Promise<Response>;
+  read: (peer:string, after:number, signal:AbortSignal) => Promise<Response>;
+ };
  enrollment?: import('./enrollment').EnrollmentTransport;
  inventory: () => Promise<SensorInventory>;
  subscribe?: (receive:(value:SensorInventory)=>void, unavailable:()=>void) => (()=>void);

plugins/xgrids-k1/frontend/src/sensors/localPreview.ts

@@ -0,0 +1,71 @@
+import {command, type Sensor, type SensorTransport} from '@mission-core/sensor-sdk';
+import {MEDIA_PROTOCOL} from './previewFrames';
+
+export interface PreviewChannel {
+ readyState:string;
+ onmessage:((event:{data:string|ArrayBuffer})=>void)|null;
+ onclose:(()=>void)|null;
+ send:(value:string)=>void;
+ close:()=>void;
+}
+
+class LocalChannel extends EventTarget implements PreviewChannel {
+ readyState='open';
+ onmessage:PreviewChannel['onmessage']=null;
+ onclose:PreviewChannel['onclose']=null;
+ constructor(private acknowledge:(value:string)=>void){
+  super();
+  this.addEventListener('message',event=>this.onmessage?.({data:(event as MessageEvent).data}));
+  this.addEventListener('close',()=>this.onclose?.());
+ }
+ send(value:string){this.acknowledge(value);}
+ close(){this.readyState='closed';}
+ receive(data:string|ArrayBuffer){if(this.readyState==='open')this.dispatchEvent(new MessageEvent('message',{data}));}
+ ended(){this.close();this.dispatchEvent(new Event('close'));}
+}
+
+/** Framing may split at any TCP boundary. Only complete bounded messages leave. */
+export function localPreviewFrames(receive:(kind:number,payload:Uint8Array)=>void) {
+ let pending=new Uint8Array(0);
+ return (input:Uint8Array)=>{
+  const joined=new Uint8Array(pending.length+input.length);joined.set(pending);joined.set(input,pending.length);
+  let offset=0;
+  while(joined.length-offset>=5){
+   const kind=joined[offset], length=new DataView(joined.buffer,joined.byteOffset+offset+1,4).getUint32(0);
+   if(kind>6||length>32768)throw new Error('Invalid local preview frame');
+   if(joined.length-offset<5+length)break;
+   receive(kind,joined.slice(offset+5,offset+5+length));offset+=5+length;
+  }
+  pending=joined.slice(offset);
+ };
+}
+
+/** Bounded completed responses work in the onboard WebKitGTK carrier. */
+export function localPreview(device:Sensor,transport:SensorTransport,parameters:Record<string,unknown>,failed:()=>void) {
+ const carrier=transport.localPreview!;
+ const abort=new AbortController();let peer='',after=Number(parameters.after)||0;
+ const makeChannel=()=>new LocalChannel(value=>{if(value==='keepalive')return;const message=JSON.parse(value);if(!Number.isSafeInteger(message.ack)||message.ack<after)throw new Error('Invalid ACK');after=message.ack;});
+ const rrd=makeChannel(),camera=makeChannel();
+ const close=()=>{if(abort.signal.aborted)return;abort.abort();rrd.close();camera.close();if(peer)void transport.submit(command(device,'close-peer',{peer_id:peer})).catch(()=>{});};
+ const start=async()=>{
+  let response=await carrier.open(command(device,'offer',parameters),abort.signal);
+  const text=new TextDecoder();
+  const parse=localPreviewFrames((kind,payload)=>{
+   if(kind===0){const hello=JSON.parse(text.decode(payload));if(peer||hello.media_protocol!==MEDIA_PROTOCOL||!/^peer_[0-9a-f]{32}$/.test(hello.peer_id))throw new Error('Invalid local preview');peer=hello.peer_id;return;}
+   if(!peer)throw new Error('Missing preview admission');
+   const channel=kind<=2||kind===6?rrd:camera;
+   if(kind>=5){channel.ended();return;}
+   channel.receive(kind%2?text.decode(payload):payload.buffer as ArrayBuffer);
+  });
+  while(!abort.signal.aborted){
+   if(!response.ok)throw new Error('Local preview unavailable');
+   const bytes=await response.arrayBuffer();
+   if(bytes.byteLength>196608)throw new Error('Local preview budget');
+   if(abort.signal.aborted)return;
+   parse(new Uint8Array(bytes));
+   if(!peer)throw new Error('Missing preview admission');
+   if(!abort.signal.aborted)response=await carrier.read(peer,after,abort.signal);
+  }
+ };
+ return {rrd,camera,start:async()=>{try{await start();}catch(error){if(!abort.signal.aborted)failed();throw error;}},close};
+}

plugins/xgrids-k1/frontend/src/sensors/useK1Preview.ts

@@ -4,6 +4,7 @@ import { perform, type Sensor, type SensorTransport } from './runtime';
 import { MEDIA_PROTOCOL, previewFrames, previewRecording } from './previewFrames';
 import { previewCamera } from './previewCamera';
 import { previewFreshness } from './previewFreshness';
+import {localPreview, type PreviewChannel} from './localPreview';
 export type PreviewStatus = {
     lidar: string;
     camera: string;
@@ -76,13 +77,28 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
             }
         } };
         const viewer = native, freshness = previewFreshness();
-        const pc = new RTCPeerConnection({ iceServers: [] });
+        if (!transport.localPreview && typeof RTCPeerConnection !== 'function') {
-        const rrd = pc.createDataChannel('rrd', { ordered: true }), camera = pc.createDataChannel('camera', { ordered: true });
+            update({lidar:'Этот браузер не поддерживает просмотр. Откройте сцену в Mission Core.',camera:'Просмотр недоступен',retry:false});
-        rrd.binaryType = 'arraybuffer';
+            return;
-        camera.binaryType = 'arraybuffer';
+        }
+        let pc:RTCPeerConnection|null=null;
+        try {if(!transport.localPreview)pc=new RTCPeerConnection({iceServers:[]});}
+        catch {update({lidar:'Не удалось открыть канал просмотра',retry:false});return;}
+        const local=transport.localPreview?localPreview(device,transport,{acquisition_id:device.control?.acquisition_id,view_id:view.id,after:view.recording.after},()=>fail()):null;
+        let rrd:PreviewChannel|RTCDataChannel, camera:PreviewChannel|RTCDataChannel;
+        try {
+            rrd = local?.rrd ?? pc!.createDataChannel('rrd', { ordered: true });
+            camera = local?.camera ?? pc!.createDataChannel('camera', { ordered: true });
+        } catch {
+            local?.close();pc?.close();
+            update({lidar:'Не удалось открыть канал просмотра',camera:'Просмотр недоступен',retry:false});return;
+        }
+        if('binaryType' in rrd)rrd.binaryType = 'arraybuffer';
+        if('binaryType' in camera)camera.binaryType = 'arraybuffer';
+        const closeConnection=()=>{local?.close();pc?.close();};
         const {channel, previousRecording} = view;
         const fail = () => { if (!active || failed)
-            return; failed = true; update({ lidar: 'Лидар: восстанавливаем связь', camera: 'Камера: ожидаем соединение', retry: true, presented: false, cameraPresented: false }); clearInterval(follow); clearInterval(keepalive); pc.close(); };
+            return; failed = true; update({ lidar: 'Лидар: восстанавливаем связь', camera: 'Камера: ожидаем соединение', retry: true, presented: false, cameraPresented: false }); clearInterval(follow); clearInterval(keepalive); closeConnection(); };
         let cameraDecoded = false, cameraFailed = false, lastVideoTime = -1, lastVideoProgress = 0, seenLidar = false;
         const cameraFail = () => { cameraFailed = true; update({ camera: 'Камера: изображение недоступно', cameraPresented: false }); camera.close(); };
         let decoder = previewCamera(video.current!, () => { cameraDecoded = true; lastVideoProgress = performance.now(); }, cameraFail);
@@ -101,7 +117,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
             fail(); };
         camera.onclose = () => { if (active && !failed)
             cameraFail(); };
-        rrd.onmessage = event => {
+        rrd.onmessage = (event:{data:string|ArrayBuffer}) => {
             if (!active)
                 return;
             try {
@@ -110,7 +126,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
                     if (value.type === 'preview-unavailable' && value.code === 'resume-expired') {
                         failed = true; clearInterval(follow); clearInterval(keepalive);
                         update({retry:false, presented:false, cameraPresented:false, lidar:'Сессия просмотра истекла. Закройте и откройте пространственную сцену.'});
-                        pc.close(); return;
+                        closeConnection(); return;
                     }
                     if (batch || value.type !== 'rrd-batch' || !Number.isSafeInteger(value.sequence) || value.sequence < 1 || !value.lidar)
                         throw new Error('Invalid RRD batch');
@@ -123,7 +139,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
                 fail();
             }
         };
-        camera.onmessage = event => {
+        camera.onmessage = (event:{data:string|ArrayBuffer}) => {
             if (!active || failed)
                 return;
             try {
@@ -176,11 +192,13 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
                     lidar: presented ? 'Лидар: данные поступают' : seenLidar ? 'Лидар: нет свежих данных' : 'Лидар: ожидаем данные',
                     camera: cameraFailed ? 'Камера: изображение недоступно' : cameraPresented ? 'Камера: изображение поступает' : cameraDecoded ? 'Камера: нет свежих кадров' : 'Камера: ожидаем изображение' });
             }, 250);
+            if(local){await local.start();return;}
+            if(!pc)return;
             await pc.setLocalDescription(await pc.createOffer());
             if (pc.iceGatheringState !== 'complete')
                 await new Promise<void>((resolve, reject) => {
                     iceTimeout = setTimeout(() => reject(new Error('ICE timeout')), 8000);
-                    pc.onicegatheringstatechange = () => { if (pc.iceGatheringState === 'complete') {
+                    pc.onicegatheringstatechange = () => { if (pc?.iceGatheringState === 'complete') {
                         clearTimeout(iceTimeout);
                         resolve();
                     } };
@@ -201,11 +219,11 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
             if (answer.media_protocol !== MEDIA_PROTOCOL) {
                 failed = true;
                 clearInterval(follow);
-                pc.close();
+                closeConnection();
                 update({ lidar: 'Для просмотра требуется обновление приложения на БК', camera: 'Камера: ожидаем обновление', retry: false });
                 return;
             }
-            pc.onconnectionstatechange = () => { if (active && ['failed', 'closed'].includes(pc.connectionState))
+            pc.onconnectionstatechange = () => { if (active && ['failed', 'closed'].includes(pc?.connectionState ?? 'closed'))
                 fail(); };
             await pc.setRemoteDescription({ type: answer.type, sdp: answer.sdp });
             keepalive = setInterval(() => { for (const c of [rrd, camera])
@@ -222,9 +240,8 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR
             camera.onmessage = null;
             camera.onclose = null;
             rrd.onclose = null;
-            pc.onconnectionstatechange = null;
+            if(pc){pc.onconnectionstatechange = null;pc.onicegatheringstatechange = null;}
-            pc.onicegatheringstatechange = null;
+            closeConnection();
-            pc.close();
             rrdFrames.close();
             cameraFrames.close();
             decoder.close();

plugins/xgrids-k1/packaging/build_deb.py

@@ -22,7 +22,7 @@ from credential_install import PROFILE_ID, validate  # noqa: E402
 from debian import package  # noqa: E402
 from runtime_payload import files as runtime_files  # noqa: E402
 
-VERSION = "0.1.13"
+VERSION = "0.1.14"
 RESOURCES = (
     "plugins/xgrids-k1/profile_loader.py",
     "plugins/xgrids-k1/plugin.manifest.json",

plugins/xgrids-k1/packaging/runtime-files.json

@@ -79,6 +79,7 @@
     "src/k1link/viewer/__init__.py",
     "src/k1link/viewer/metrics.py",
     "src/k1link/viewer/node_media.py",
+    "src/k1link/viewer/node_local_media.py",
     "src/k1link/viewer/node_rerun.py",
     "src/k1link/viewer/recorded.py",
     "src/k1link/viewer/recorded_blueprint_lifecycle.py",

src/k1link/device_plugins/xgrids_k1/node_bridge.py

@@ -382,6 +382,15 @@ def create_app(repository_root: Path):
     peers = NodeMediaPeers(bridge.rerun, bridge.service.camera_preview)
     sensor = NodeK1Sensor(bridge, peers)
 
+    from fastapi.responses import Response
+    from k1link.viewer.node_local_media import start_local, read_local
+
+    async def preview_request(request):
+        data = await request.body()
+        if len(data) > 4096:
+            raise ValueError("Preview request exceeds bound")
+        return await request.json()
+
     @asynccontextmanager
     async def lifespan(_app):
         yield
@@ -398,6 +407,30 @@ def create_app(repository_root: Path):
     async def inventory(request: Request):
         return await sensor.inventory(request.headers["X-Node-Id"])
 
+    @app.post("/local-preview")
+    async def preview(request: Request):
+        try:
+            command = await preview_request(request)
+            if datetime.fromisoformat(command["deadline_at"]) <= datetime.now(UTC):
+                raise ValueError("Preview request expired")
+            await sensor.admit_preview(command, request.headers["X-Node-Id"])
+            identifier = await start_local(peers, command['parameters'])
+            data = await read_local(peers, identifier, command['parameters'].get('after', 0))
+            return Response(data, media_type="application/vnd.missioncore.preview",
+                            headers={"Cache-Control": "no-store"})
+        except (ValueError, KeyError, TypeError):
+            return JSONResponse({"error": "Preview unavailable"}, status_code=409)
+
+    @app.post("/local-preview/read")
+    async def preview_read(request: Request):
+        try:
+            data = await preview_request(request)
+            payload = await read_local(peers, data['peer_id'], data['after'])
+            return Response(payload, media_type="application/vnd.missioncore.preview",
+                            headers={"Cache-Control": "no-store"})
+        except (ValueError, KeyError, TypeError):
+            return JSONResponse({"error": "Preview expired"}, status_code=409)
+
     @app.get("/prepare-safe")
     async def prepare_safe():
         state = await sensor.raw_state()

src/k1link/device_plugins/xgrids_k1/node_sensor.py

@@ -132,6 +132,17 @@ class NodeK1Sensor:
             "expected_state_revision": control["state_revision"],
         }
 
+    async def admit_preview(self, command, node_id):
+        item = project_sensor(await self.raw_state(), node_id)
+        if (not item or command.get("action_id") != "offer"
+                or command.get("session", {}).get("device_id") != item["id"]
+                or command["session"].get("session_id") != item["snapshot"]["context"]["session_id"]
+                or item["snapshot"]["acquisition"] != "streaming"
+                or not item["control"]["acquisition_id"]
+                or command.get("parameters", {}).get("acquisition_id") != item["control"]["acquisition_id"]):
+            raise ValueError("Acquisition is not active or changed")
+        return item
+
     async def execute(self, command, node_id):
         state = await self.raw_state()
         item = project_sensor(state, node_id)

src/k1link/fleet/monitor.py

@@ -0,0 +1,390 @@
+"""Idempotent local replica of a board-owned Timescale history, behind fleet mTLS."""
+
+import json
+import math
+import re
+import sqlite3
+import threading
+import time
+from uuid import UUID
+
+SCHEMA = "missioncore.node-system-monitor/v1"
+KEY = re.compile(r"^[A-Za-z0-9_.:-]{1,180}$")
+
+
+def checked_sample(value):
+    if not isinstance(value, dict) or set(value) != {
+        "seq",
+        "at",
+        "boot_id",
+        "uptime",
+        "values",
+        "events",
+    }:
+        raise ValueError("Invalid sample")
+    if type(value["seq"]) is not int or not 0 < value["seq"] < 2**53:
+        raise ValueError("Invalid sequence")
+    if str(UUID(value["boot_id"])) != value["boot_id"]:
+        raise ValueError("Invalid boot")
+    for name in ["at", "uptime"]:
+        if (
+            type(value[name]) not in (int, float)
+            or not math.isfinite(value[name])
+            or value[name] < 0
+        ):
+            raise ValueError("Invalid time")
+    if not isinstance(value["values"], dict) or len(value["values"]) > 384:
+        raise ValueError("Metric budget")
+    for key, number in value["values"].items():
+        if not KEY.fullmatch(key) or (
+            number is not None and (type(number) not in (int, float) or not math.isfinite(number))
+        ):
+            raise ValueError("Invalid metric")
+    if not isinstance(value["events"], list) or len(value["events"]) > 64:
+        raise ValueError("Event budget")
+    for event in value["events"]:
+        if (
+            not isinstance(event, dict)
+            or set(event) - {"code", "kind", "locations", "at"}
+            or event.get("code")
+            not in {
+                "ui-error",
+                "ui-rejection",
+                "ui-render-error",
+                "system-oom",
+                "gpu-reset",
+                "disk-error",
+                "usb-disconnected",
+                "usb-error",
+                "service-failed",
+                "ui-process-exit",
+                "ui-load-failed",
+            }
+            or not re.fullmatch("[A-Za-z]{1,48}", event.get("kind", ""))
+            or not isinstance(event.get("locations"), list)
+            or len(event["locations"]) > 8
+            or any(
+                not isinstance(v, str)
+                or not re.fullmatch(r"[A-Za-z0-9_-]{1,96}\.js:[0-9]{1,9}:[0-9]{1,9}", v)
+                for v in event["locations"]
+            )
+        ):
+            raise ValueError("Invalid event")
+        if (
+            type(event.get("at")) not in (int, float)
+            or not math.isfinite(event["at"])
+            or event["at"] < 0
+        ):
+            raise ValueError("Invalid event time")
+    return json.dumps(value, allow_nan=False, separators=(",", ":"), sort_keys=True)
+
+
+class MonitorReplica:
+    def __init__(self, root):
+        path = root / "monitor.sqlite3"
+        if path.is_symlink():
+            raise ValueError("Invalid monitor store")
+        self.db = sqlite3.connect(path, check_same_thread=False, timeout=2)
+        try:
+            path.chmod(0o600)
+            self.lock = threading.RLock()
+            self.maintenance = 0
+            self.db.execute("PRAGMA journal_mode=WAL")
+            self.db.execute("PRAGMA synchronous=FULL")
+            self.db.execute("PRAGMA max_page_count=524288")
+            self.db.execute(
+                "CREATE TABLE IF NOT EXISTS samples(node TEXT,source TEXT,seq INTEGER,at REAL,"
+                "body TEXT,PRIMARY KEY(node,source,seq))"
+            )
+            self.db.execute("CREATE INDEX IF NOT EXISTS samples_time ON samples(node,at)")
+            self.db.execute(
+                "CREATE TABLE IF NOT EXISTS minutes(node TEXT,at INTEGER,body TEXT,"
+                "PRIMARY KEY(node,at))"
+            )
+            self.db.execute(
+                "CREATE TABLE IF NOT EXISTS events(node TEXT,source TEXT,seq INTEGER,at REAL,"
+                "body TEXT,PRIMARY KEY(node,source,seq))"
+            )
+            self.db.execute("CREATE INDEX IF NOT EXISTS events_time ON events(node,at)")
+            self.db.execute(
+                "CREATE TABLE IF NOT EXISTS states(node TEXT PRIMARY KEY,source TEXT,"
+                "body TEXT,received REAL,ack INTEGER)"
+            )
+            self.db.commit()
+        except Exception:
+            self.db.close()
+            raise
+
+    def ingest(self, node, value):
+        if not isinstance(value, dict) or value.get("schema") != SCHEMA:
+            return None
+        source = value.get("source_id")
+        if not isinstance(source, str) or str(UUID(source)) != source:
+            raise ValueError("Invalid source")
+        definitions = value.get("definitions")
+        if not isinstance(definitions, dict) or len(definitions) > 384:
+            raise ValueError("Definition budget")
+        for key, definition in definitions.items():
+            if (
+                not KEY.fullmatch(key)
+                or not isinstance(definition, dict)
+                or set(definition) != {"label", "group", "resource", "unit", "reason"}
+            ):
+                raise ValueError("Invalid definition")
+            if any(
+                v is not None and (not isinstance(v, str) or len(v) > 256)
+                for v in definition.values()
+            ):
+                raise ValueError("Invalid definition")
+        if value.get("latest") is not None:
+            checked_sample(value["latest"])
+        batch = value.get("batch") or {}
+        rows = batch.get("samples", [])
+        if (
+            not isinstance(rows, list)
+            or len(rows) > 32
+            or (rows and (batch.get("source_id") != source or batch.get("schema") != SCHEMA))
+        ):
+            raise ValueError("Invalid batch")
+        serialized = [(row, checked_sample(row)) for row in rows]
+        if sum(len(text.encode()) for _, text in serialized) > 196608:
+            raise ValueError("Batch budget")
+        if any(
+            a[0]["seq"] >= b[0]["seq"] for a, b in zip(serialized, serialized[1:], strict=False)
+        ):
+            raise ValueError("Unordered batch")
+        snapshot = {
+            key: item
+            for key, item in value.items()
+            if key
+            in {
+                "schema",
+                "source_id",
+                "latest",
+                "definitions",
+                "storage",
+                "database_bytes",
+                "sample_interval_seconds",
+                "retention_days",
+                "database_budget_bytes",
+                "free_reserve_bytes",
+            }
+        }
+        snapshot.update(
+            retained_since=batch.get("retained_since"), first_seq=batch.get("first_seq")
+        )
+        with self.lock, self.db:
+            previous = self.db.execute(
+                "SELECT source,ack FROM states WHERE node=?", (node,)
+            ).fetchone()
+            ack = previous[1] if previous and previous[0] == source else 0
+            minutes = {}
+            for row, text in serialized:
+                old = self.db.execute(
+                    "SELECT body FROM samples WHERE node=? AND source=? AND seq=?",
+                    (node, source, row["seq"]),
+                ).fetchone()
+                if old and old[0] != text:
+                    raise ValueError("Conflicting history sequence")
+                self.db.execute(
+                    "INSERT OR IGNORE INTO samples VALUES(?,?,?,?,?)",
+                    (node, source, row["seq"], row["at"], text),
+                )
+                if not old:
+                    minute = int(row["at"] // 60) * 60
+                    if minute not in minutes:
+                        saved = self.db.execute(
+                            "SELECT body FROM minutes WHERE node=? AND at=?", (node, minute)
+                        ).fetchone()
+                        minutes[minute] = json.loads(saved[0]) if saved else {}
+                    for key, number in row["values"].items():
+                        if number is not None:
+                            aggregate = minutes[minute].setdefault(key, [number, number, 0, 0])
+                            aggregate[0] = min(aggregate[0], number)
+                            aggregate[1] = max(aggregate[1], number)
+                            aggregate[2] += number
+                            aggregate[3] += 1
+                    if row["events"]:
+                        self.db.execute(
+                            "INSERT INTO events VALUES(?,?,?,?,?)",
+                            (node, source, row["seq"], row["at"], json.dumps(row["events"])),
+                        )
+                ack = max(ack, row["seq"])
+            for minute, aggregate in minutes.items():
+                self.db.execute(
+                    "INSERT INTO minutes VALUES(?,?,?) ON CONFLICT(node,at) "
+                    "DO UPDATE SET body=excluded.body",
+                    (node, minute, json.dumps(aggregate)),
+                )
+            self.db.execute(
+                "INSERT INTO states VALUES(?,?,?,?,?) ON CONFLICT(node) DO UPDATE SET "
+                "source=excluded.source,body=excluded.body,received=excluded.received,ack=excluded.ack",
+                (node, source, json.dumps(snapshot), time.time(), ack),
+            )
+            if time.time() - self.maintenance > 60:
+                self.db.execute("DELETE FROM samples WHERE at<?", (time.time() - 7 * 86400,))
+                self.db.execute("DELETE FROM minutes WHERE at<?", (time.time() - 7 * 86400,))
+                self.db.execute("DELETE FROM events WHERE at<?", (time.time() - 7 * 86400,))
+                self.maintenance = time.time()
+        return dict(source_id=source, after=ack)
+
+    def query(self, node, metric="cpu.usage", window=900, end=None):
+        if not KEY.fullmatch(metric) or window not in (900, 3600, 21600, 86400, 604800):
+            raise ValueError("Invalid range")
+        end = time.time() if end is None else end
+        if not math.isfinite(end) or end < 0:
+            raise ValueError("Invalid time")
+        coarse = window >= 21600
+        width = math.ceil(window / 300 / 60) * 60 if coarse else max(1, window // 300)
+        if coarse:
+            end = math.floor(end / 60) * 60
+        start = end - window
+        path = '$.values."' + metric + '"'
+        with self.lock:
+            row = self.db.execute(
+                "SELECT body,received,ack FROM states WHERE node=?", (node,)
+            ).fetchone()
+            if not row:
+                return dict(available=False, series=[], events=[])
+            snapshot = json.loads(row[0])
+            if coarse:
+                aggregate = '$."' + metric + '"'
+                points = self.db.execute(
+                    "SELECT CAST(at/? AS INTEGER),min(json_extract(body,?)), "
+                    "max(json_extract(body,?)),sum(json_extract(body,?))*1.0/"
+                    "nullif(sum(json_extract(body,?)),0),sum(json_extract(body,?)) "
+                    "FROM minutes WHERE node=? AND at>=? AND at<? "
+                    "GROUP BY CAST(at/? AS INTEGER) ORDER BY 1",
+                    (
+                        width,
+                        aggregate + "[0]",
+                        aggregate + "[1]",
+                        aggregate + "[2]",
+                        aggregate + "[3]",
+                        aggregate + "[3]",
+                        node,
+                        start,
+                        end,
+                        width,
+                    ),
+                ).fetchall()
+            else:
+                points = self.db.execute(
+                    "SELECT CAST(at/? AS INTEGER),min(json_extract(body,?)), "
+                    "max(json_extract(body,?)),avg(json_extract(body,?)),"
+                    "count(json_extract(body,?)) FROM samples "
+                    "WHERE node=? AND at>=? AND at<=? GROUP BY CAST(at/? AS INTEGER) ORDER BY 1",
+                    (width, path, path, path, path, node, start, end, width),
+                ).fetchall()
+            event_rows = self.db.execute(
+                "SELECT body FROM events WHERE node=? AND at>=? AND at<=? "
+                "ORDER BY at DESC LIMIT 64",
+                (node, start, end),
+            ).fetchall()
+            earliest = self.db.execute(
+                "SELECT min(at) FROM samples WHERE node=?", (node,)
+            ).fetchone()[0]
+        latest = snapshot.get("latest")
+        fresh = bool(latest and -5 <= time.time() - latest["at"] < 10 and time.time() - row[1] < 15)
+        return dict(
+            available=True,
+            **snapshot,
+            fresh=fresh,
+            received_at=row[1],
+            replicated_after=row[2],
+            archive_since=earliest,
+            backlog=max(0, (latest or {}).get("seq", 0) - row[2]),
+            start=start,
+            end=end,
+            bucket_seconds=width,
+            series=[
+                dict(at=bucket * width, min=minimum, max=maximum, mean=mean, count=count)
+                for bucket, minimum, maximum, mean, count in points
+            ],
+            events=[event for record in event_rows for event in json.loads(record[0])][:64],
+        )
+
+    def close(self):
+        with self.lock:
+            self.db.close()
+
+
+class MonitorReceiver:
+    """Replica I/O never runs under the fleet/control lock, including startup.
+
+    Heartbeats may repeat a batch until this worker commits it. Replacing a
+    queued retry is safe: the sender advances only from the returned commit ACK.
+    """
+
+    def __init__(self, root):
+        self.root = root
+        self.lock = threading.Lock()
+        self.pending = {}
+        self.acks = {}
+        self.archive = None
+        self.worker = None
+        self.wake = threading.Event()
+        self.stop = threading.Event()
+
+    def submit(self, node, value):
+        if not isinstance(value, dict) or value.get("schema") != SCHEMA:
+            return None
+        with self.lock:
+            if node not in self.pending and len(self.pending) >= 32:
+                return None
+            self.pending[node] = value
+            if self.worker is None:
+                self.worker = threading.Thread(
+                    target=self.run, name="fleet-monitor-replica", daemon=True
+                )
+                self.worker.start()
+            self.wake.set()
+            ack = self.acks.get(node)
+            return ack if ack and ack["source_id"] == value.get("source_id") else None
+
+    def run(self):
+        try:
+            while not self.stop.is_set():
+                self.wake.wait(1)
+                with self.lock:
+                    pending, self.pending = self.pending, {}
+                    self.wake.clear()
+                if self.archive is None:
+                    try:
+                        self.archive = MonitorReplica(self.root)
+                    except (OSError, ValueError, sqlite3.Error):
+                        continue
+                for node, value in pending.items():
+                    try:
+                        if self.archive is None:
+                            self.archive = MonitorReplica(self.root)
+                        ack = self.archive.ingest(node, value)
+                        with self.lock:
+                            self.acks[node] = ack
+                    except (OSError, ValueError, TypeError, KeyError, sqlite3.Error):
+                        # No ACK: the board retains and retries its own archive.
+                        continue
+        finally:
+            if self.archive is not None:
+                self.archive.close()
+
+    def query(self, *args, **kwargs):
+        with self.lock:
+            if self.worker is None:
+                self.worker = threading.Thread(
+                    target=self.run, name="fleet-monitor-replica", daemon=True
+                )
+                self.worker.start()
+                self.wake.set()
+        try:
+            if self.archive is not None:
+                return self.archive.query(*args, **kwargs)
+        except sqlite3.Error:
+            pass
+        return dict(available=False, series=[], events=[])
+
+    def close(self):
+        self.stop.set()
+        self.wake.set()
+        if self.worker is not None:
+            self.worker.join(5)

src/k1link/fleet/registry.py

@@ -33,6 +33,9 @@ class FleetRegistry:
         from .device_enrollment import DeviceEnrollment
 
         self.device_enrollment = DeviceEnrollment()
+        from .monitor import MonitorReceiver
+
+        self.monitor = MonitorReceiver(root)
         self.trust = CoreTrust(root)
         path = root / "fleet.sqlite3"
         if path.is_symlink():
@@ -93,6 +96,7 @@ class FleetRegistry:
             server.server_close()
         with self.lock:
             self.db.close()
+            self.monitor.close()
 
     def listen(self, address):
         from .transport import NodeChannelServer
@@ -407,4 +411,10 @@ class FleetRegistry:
                 }
                 row["binding"]["client_pem"] = self.trust.leaf(node_id, key)
             self.save(row)
-            return 200, {"ok": True, "client_pem": row["binding"]["client_pem"], **sensor_response, **enrollment_response}
+            monitor_ack = None
+            try:
+                monitor_ack = self.monitor.submit(node_id, value.get("monitor"))
+            except (ValueError, TypeError, KeyError, sqlite3.Error):
+                # Telemetry persistence failure must not block device control.
+                pass
+            return 200, {"monitor_ack": monitor_ack, "ok": True, "client_pem": row["binding"]["client_pem"], **sensor_response, **enrollment_response}

src/k1link/viewer/node_local_media.py

@@ -0,0 +1,137 @@
+"""Local HTTP carrier for the same bounded, acknowledged Node preview delivery.
+
+The Unix-only plugin endpoint is proxied by the authenticated Node UI. It never
+starts acquisition, rewrites network settings, or creates a second recorder.
+"""
+
+import asyncio
+import json
+import time
+from uuid import UUID, uuid4
+
+from .node_media import MEDIA_PROTOCOL
+
+
+class LocalConnection:
+    def __init__(self):
+        self.queue = asyncio.Queue(maxsize=40)
+        self.buffered = 0
+        self.closed = False
+
+    def push(self, kind, payload):
+        if self.closed:
+            return
+        if isinstance(payload, str):
+            payload = payload.encode()
+        if len(payload) > 32768:
+            raise ValueError("Local preview message exceeds bound")
+        frame = bytes([kind]) + len(payload).to_bytes(4, "big") + payload
+        self.queue.put_nowait(frame)
+        self.buffered += len(frame)
+
+    async def close(self):
+        self.closed = True
+
+
+class LocalChannel:
+    def __init__(self, connection, label):
+        self.connection, self.label = connection, label
+        self.closed = False
+
+    @property
+    def readyState(self):
+        return "closed" if self.closed or self.connection.closed else "open"
+
+    @property
+    def bufferedAmount(self):
+        return self.connection.buffered
+
+    def send(self, payload):
+        kind = (1 if self.label == "rrd" else 3) + (not isinstance(payload, str))
+        self.connection.push(kind, payload)
+
+    def close(self):
+        if not self.closed:
+            self.closed = True
+            self.connection.push(5 if self.label == "camera" else 6, b"")
+
+
+def admit_local(parameters):
+    view_id, after = parameters.get("view_id"), parameters.get("after", 0)
+    if not isinstance(view_id, str) or str(UUID(view_id)) != view_id:
+        raise ValueError("Preview view identifier required")
+    if type(after) is not int or not 0 <= after <= 2**53 - 1:
+        raise ValueError("Invalid preview cursor")
+    return view_id, after
+
+
+async def start_local(peers, parameters):
+    view_id, after = admit_local(parameters)
+    for identifier, entry in list(peers.items.items()):
+        if entry.get("local") and entry["view_id"] == view_id:
+            await peers.close(identifier)
+    if len(peers.items) >= 2:
+        raise ValueError("Close another live viewer")
+    connection = LocalConnection()
+    identifier = "peer_" + uuid4().hex
+    entry = {
+        "pc": connection,
+        "seen": time.monotonic(),
+        "tasks": [],
+        "labels": {"rrd", "camera"},
+        "view_id": view_id,
+        "after": after,
+        "subscriber": None,
+        "local": True,
+        "reading": False,
+    }
+    peers.items[identifier] = entry
+    connection.push(0, json.dumps({"peer_id": identifier, "media_protocol": MEDIA_PROTOCOL}))
+    for label in ("rrd", "camera"):
+        entry["tasks"].append(
+            asyncio.create_task(peers.deliver(identifier, LocalChannel(connection, label)))
+        )
+
+    async def expiry():
+        while identifier in peers.items:
+            await asyncio.sleep(5)
+            if time.monotonic() - entry["seen"] > 30:
+                await peers.close(identifier)
+
+    entry["tasks"].append(asyncio.create_task(expiry()))
+    return identifier
+
+
+async def read_local(peers, identifier, sequence):
+    acknowledge(peers, identifier, sequence)
+    entry = peers.items[identifier]
+    if entry["reading"]:
+        raise ValueError("Concurrent preview read")
+    entry["reading"] = True
+    connection = entry["pc"]
+    try:
+        try:
+            first = await asyncio.wait_for(connection.queue.get(), 1)
+        except TimeoutError:
+            return b""
+        frames = [first]
+        size = len(first)
+        while size < 131072 and not connection.queue.empty():
+            frame = connection.queue.get_nowait()
+            frames.append(frame)
+            size += len(frame)
+        connection.buffered -= size
+        return b"".join(frames)
+    finally:
+        entry["reading"] = False
+
+
+def acknowledge(peers, identifier, sequence):
+    entry = peers.items.get(identifier)
+    if not entry or not entry.get("local"):
+        raise ValueError("Local preview expired")
+    if type(sequence) is not int or not 0 <= sequence <= 2**53 - 1:
+        raise ValueError("Invalid preview acknowledgement")
+    entry["seen"] = time.monotonic()
+    if entry["subscriber"] is not None:
+        entry["subscriber"].acknowledge(sequence)

src/k1link/viewer/node_media.py

@@ -252,9 +252,11 @@ class NodeMediaPeers:
     async def close(self, identifier):
         entry = self.items.pop(identifier, None)
         if entry:
-            for task in entry["tasks"]:
+            tasks = [task for task in entry["tasks"] if task is not asyncio.current_task()]
-                if task is not asyncio.current_task():
+            for task in tasks:
-                    task.cancel()
+                task.cancel()
+            # A reconnect must not race the previous delivery lease's finally.
+            await asyncio.gather(*tasks, return_exceptions=True)
             with suppress(Exception):
                 await entry["pc"].close()
 

src/k1link/web/fleet_api.py

@@ -51,6 +51,21 @@ class AddRequest(BaseModel):
 router = APIRouter(prefix="/api/v1/fleet", tags=["fleet"])
 
 
+@router.get("/{vehicle_id}/monitor")
+def board_monitor(vehicle_id: str, response: Response,
+                  fleet: Annotated[FleetRegistry, Depends(local_operator)],
+                  metric: str = "cpu.usage", window: int = 900, end: float | None = None):
+    response.headers["Cache-Control"] = "no-store"
+    try:
+        with fleet.lock:
+            node_id = fleet.find(vehicle_id)["node_id"]
+        return fleet.monitor.query(node_id, metric, window, end)
+    except PairingError as error:
+        raise HTTPException(404, str(error)) from None
+    except ValueError:
+        raise HTTPException(400, "Выберите доступный период и показатель.") from None
+
+
 @router.get("")
 def fleet_list(response: Response, fleet: Annotated[FleetRegistry, Depends(local_operator)]):
     response.headers["Cache-Control"] = "no-store"

tests/test_board_monitor.py

@@ -0,0 +1,168 @@
+import importlib.util
+import sqlite3
+import time
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+from k1link.fleet.monitor import SCHEMA, MonitorReceiver, MonitorReplica
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def module(name):
+    spec = importlib.util.spec_from_file_location(
+        name, ROOT / "apps/node-agent/monitor" / (name + ".py")
+    )
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+def record(seq, at, boot, value=25):
+    return dict(seq=seq, at=at, boot_id=boot, uptime=seq, values={"cpu.usage": value}, events=[])
+
+
+def envelope(source, rows, latest=None):
+    return dict(
+        schema=SCHEMA,
+        source_id=source,
+        storage="ready",
+        latest=latest or rows[-1],
+        definitions={
+            "cpu.usage": dict(label="CPU", group="CPU", resource="cpu", unit="%", reason=None)
+        },
+        batch=dict(schema=SCHEMA, source_id=source, samples=rows),
+    )
+
+
+def test_replica_retries_restart_backfill_and_latest_are_separate(tmp_path):
+    source, boot = str(uuid4()), str(uuid4())
+    now = time.time()
+    rows = [record(i, now - 40 + i, boot) for i in range(1, 41)]
+    archive = MonitorReplica(tmp_path)
+    assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32
+    result = archive.query("board")
+    assert result["latest"]["seq"] == 40 and result["backlog"] == 8
+    assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 32
+    archive.close()
+    archive = MonitorReplica(tmp_path)
+    # A lost reply cannot duplicate history; ACK comes from the durable commit.
+    assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32
+    assert archive.ingest("board", envelope(source, rows[32:]))["after"] == 40
+    assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 40
+    assert archive.query("board")["backlog"] == 0
+    # Retention on the board can leave a real gap; no zero samples are invented.
+    assert archive.ingest("board", envelope(source, [record(51, now + 11, boot)]))["after"] == 51
+    archive.close()
+
+
+def test_conflicting_or_nonfinite_batch_rolls_back_and_missing_values_are_null(tmp_path):
+    source, boot = str(uuid4()), str(uuid4())
+    now = time.time()
+    archive = MonitorReplica(tmp_path)
+    original = record(2, now - 20, boot)
+    archive.ingest("board", envelope(source, [original]))
+    bad = envelope(source, [record(1, now - 21, boot), {**original, "values": {"cpu.usage": 99}}])
+    with pytest.raises(ValueError):
+        archive.ingest("board", bad)
+    assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 1
+    with pytest.raises(ValueError):
+        archive.ingest("board", envelope(source, [record(3, now, boot, float("nan"))]))
+    archive.ingest("board", envelope(source, [record(3, now, boot, None)]))
+    point = archive.query("board")["series"][-1]
+    assert point["max"] is None and point["count"] == 0
+    archive.close()
+
+
+def test_receiver_failure_isolated_and_offline_history_reopens(tmp_path, monkeypatch):
+    source, boot = str(uuid4()), str(uuid4())
+    archive = MonitorReplica(tmp_path)
+    archive.ingest("board", envelope(source, [record(1, time.time(), boot)]))
+    archive.close()
+    receiver = MonitorReceiver(tmp_path)
+    try:
+        deadline = time.monotonic() + 2
+        while not receiver.query("board")["available"] and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert receiver.query("board")["available"]
+        monkeypatch.setattr(
+            receiver.archive,
+            "ingest",
+            lambda *_: (_ for _ in ()).throw(sqlite3.OperationalError("synthetic full disk")),
+        )
+        start = time.monotonic()
+        assert receiver.submit("board", envelope(source, [record(2, time.time(), boot)])) is None
+        assert time.monotonic() - start < 0.1
+        time.sleep(0.05)
+        assert receiver.acks.get("board") is None
+    finally:
+        receiver.close()
+
+
+def test_linux_rates_distinguish_absent_initial_reset_and_gap(tmp_path):
+    now = [100]
+
+    def write(path, text):
+        target = tmp_path / path
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(text)
+
+    def counters(cpu, idle, rx):
+        write("proc/stat", f"cpu {cpu} 0 0 {idle} 0 0 0 0\ncpu0 {cpu} 0 0 {idle} 0 0 0 0\n")
+        write("sys/class/net/eth0/statistics/rx_bytes", str(rx))
+
+    write("sys/class/net/eth0/speed", "-1")
+    write("sys/bus/usb/devices/1-2/idVendor", "0000")
+    write("sys/bus/usb/devices/1-2/speed", "480")
+    write(
+        "sys/fs/cgroup/system.slice/mission-core-node.service/memory.events", "oom 3\noom_kill 2\n"
+    )
+    metrics = module("linux_metrics").LinuxMetrics(tmp_path, lambda: now[0])
+    counters(100, 100, 1000)
+    first = metrics.sample()
+    assert first["cpu.usage"] is None and first["net.eth0.rx_bytes"] is None
+    now[0] += 1
+    counters(150, 150, 1512)
+    second = metrics.sample()
+    assert second["cpu.usage"] == 50 and second["net.eth0.rx_bytes"] == 512
+    assert second["net.eth0.speed"] is None
+    assert second["usb.1-2.speed"] == 480 and second["usb.1-2.traffic"] is None
+    assert second["service.mission-core-node.service.oom"] == 2
+    now[0] += 1
+    counters(5, 5, 100)
+    assert metrics.sample()["net.eth0.rx_bytes"] is None
+    now[0] += 11
+    counters(25, 25, 900)
+    assert metrics.sample()["cpu.usage"] is None
+
+
+def test_journal_retains_classification_without_freeform_data():
+    classify = module("journal_events").classify
+    value = classify(
+        {
+            "__REALTIME_TIMESTAMP": "123000000",
+            "_TRANSPORT": "kernel",
+            "MESSAGE": "Out of memory: Killed process 23 secret argument",
+        }
+    )
+    assert value == dict(code="system-oom", kind="Kernel", at=123, locations=[])
+    assert classify({"MESSAGE": "private Wi-Fi password"}) is None
+
+
+def test_long_range_aggregates_keep_peaks_counts_and_exclude_incomplete_minute(tmp_path):
+    source, boot = str(uuid4()), str(uuid4())
+    now = int(time.time()//60)*60
+    archive = MonitorReplica(tmp_path)
+    rows = [record(1, now-120, boot, 10), record(2, now-110, boot, 90),
+            record(3, now-60, boot, 30), record(4, now+1, boot, 100)]
+    value = envelope(source, rows)
+    archive.ingest('board', value)
+    archive.ingest('board', value)  # Duplicate delivery cannot inflate aggregates.
+    result = archive.query('board', window=21600, end=now+10)
+    assert result['end'] == now
+    assert sum(point['count'] for point in result['series']) == 3
+    assert max(point['max'] for point in result['series']) == 90
+    assert min(point['min'] for point in result['series']) == 10
+    archive.close()

tests/test_node_k1_package_lifecycle.py

@@ -34,6 +34,7 @@ case "$1" in
 esac
 ''')
     (binary / "getent").write_text("#!/bin/sh\nexit 0\n")
+    (binary / "setup-monitor").write_text("#!/bin/sh\nprintf '%s\\n' monitor-setup >> \"$TEST_EVENTS\"\n")
     (binary / "dpkg-query").write_text(
         "#!/bin/sh\necho 'install ok " + ("installed" if configured else "unpacked") + "'\n"
     )
@@ -45,6 +46,7 @@ esac
         script = tmp_path / name
         script.write_text((PACKAGING / name).read_text()
                           .replace("/run/", str(run) + "/")
+                          .replace("/usr/lib/mission-core-node/setup-monitor", str(binary / "setup-monitor"))
                           .replace("/etc/os-release", str(release)))
         subprocess.run(["/bin/sh", str(script), *args], env=env, check=True,
                        capture_output=True, text=True)

tests/test_node_local_media.py

@@ -0,0 +1,98 @@
+import asyncio
+import json
+from uuid import uuid4
+
+import pytest
+
+from k1link.viewer.node_local_media import acknowledge, admit_local, read_local, start_local
+from k1link.viewer.node_media import NodeMediaPeers
+
+
+def test_local_stream_fragments_ack_resume_and_releases_before_reconnect():
+    payload = b"RRF2" + bytes(range(256)) * 256
+    released = []
+    subscriptions = []
+
+    class Subscription:
+        pending = None
+
+        def __init__(self, after):
+            self.after = after
+            self.sent = False
+
+        def next_batch(self, **_):
+            if self.sent:
+                return b""
+            self.sent = True
+            self.pending = (self.after + 1, payload, None)
+            return self.pending
+
+        def snapshot(self, _):
+            return dict(sequence=self.after + 1, age_ms=0, points=5000)
+
+        def acknowledge(self, seq):
+            if seq == self.after + 1:
+                self.pending = None
+
+        def release(self):
+            released.append(self.after)
+
+    class Hub:
+        def subscribe(self, view_id, after):
+            sub = Subscription(after)
+            subscriptions.append((view_id, sub))
+            return sub
+
+    class Camera:
+        def snapshot(self):
+            return {"recording": {"source_end_expected": True}}
+
+    async def run():
+        peers = NodeMediaPeers(Hub(), Camera())
+        view = str(uuid4())
+        for after in (0, 1):
+            peer = await start_local(peers, dict(view_id=view, after=after))
+            if after:
+                assert released == [0]  # Lost response/reconnect retires the old lease.
+            collected = []
+            while sum(map(len, collected)) < len(payload):
+                response = await asyncio.wait_for(read_local(peers, peer, after), 1.2)
+                assert len(response) <= 196608
+                offset = 0
+                while offset < len(response):
+                    kind = response[offset]
+                    size = int.from_bytes(response[offset + 1 : offset + 5], "big")
+                    data = response[offset + 5 : offset + 5 + size]
+                    if kind == 0:
+                        assert json.loads(data)["peer_id"] == peer
+                    if kind == 2:
+                        if data.startswith(b"MCF1"):
+                            assert int.from_bytes(data[4:], "big") == len(payload)
+                        else:
+                            collected.append(data)
+                    offset += size + 5
+            assert b"".join(collected) == payload
+            acknowledge(peers, peer, after + 1)
+            assert subscriptions[-1][1].pending is None
+        await peers.close_all()
+        assert not peers.items and released == [0, 1]
+        assert subscriptions[0][0] == subscriptions[1][0] == view
+
+    asyncio.run(run())
+
+
+@pytest.mark.parametrize("after", [True, -1, 2**53, "1"])
+def test_local_cursor_rejects_noncanonical_values(after):
+    with pytest.raises(ValueError):
+        admit_local(dict(view_id=str(uuid4()), after=after))
+
+
+def test_local_carrier_does_not_exceed_peer_capacity():
+    async def run():
+        class Peers:
+            items = {"a": {}, "b": {}}
+
+        with pytest.raises(ValueError):
+            await start_local(Peers(), dict(view_id=str(uuid4()), after=0))
+
+    asyncio.run(run())