diff --git a/src/hub-event-stream.mjs b/src/hub-event-stream.mjs
new file mode 100644
index 0000000..666b95a
--- /dev/null
+++ b/src/hub-event-stream.mjs
@@ -0,0 +1,119 @@
+/** One resumable, server-only Hub telemetry connection. Only invalidation hints
+ * cross this boundary; canonical task detail remains the lifecycle authority. */
+const identifier = value => typeof value === 'string' && value.length <= 128 &&
+  /^[A-Za-z0-9][A-Za-z0-9._:@/+\-]*$/.test(value);
+const cursorValue = value => typeof value === 'string' && /^\d{1,40}$/.test(value);
+
+export function createHubEventStream({
+  url, token = '', fetchFn = globalThis.fetch, onChange = () => {},
+  retryMs = 2000, connectTimeoutMs = 10000, idleTimeoutMs = 45000,
+  maxFrameBytes = 256 * 1024,
+} = {}) {
+  let active = false, generation = 0, cursor = null, controller = null, currentReader = null;
+  let retryTimer = null, connectTimer = null, idleTimer = null;
+  let reconnectMs = retryMs, failures = 0;
+  const clearConnectionTimers = () => {
+    clearTimeout(connectTimer); clearTimeout(idleTimer);
+    connectTimer = idleTimer = null;
+  };
+  const notify = (hint, current) => {
+    if (active && generation === current) onChange(hint);
+  };
+
+  async function connect(current) {
+    const abort = new AbortController(); controller = abort;
+    connectTimer = setTimeout(() => abort.abort(), connectTimeoutMs);
+    connectTimer.unref?.();
+    let reader;
+    try {
+      const response = await fetchFn(url, {
+        headers: { accept: 'text/event-stream', ...(token ? { authorization: `Bearer ${token}` } : {}),
+          ...(cursor ? { 'last-event-id': cursor } : {}) },
+        redirect: 'error', signal: abort.signal,
+      });
+      if (!active || generation !== current) { await response.body?.cancel?.(); return; }
+      clearTimeout(connectTimer); connectTimer = null;
+      if (!response.ok || !response.headers?.get('content-type')?.includes('text/event-stream') || !response.body?.getReader) {
+        await response.body?.cancel?.(); throw new Error('Hub stream unavailable');
+      }
+      failures = 0;
+      // Connect before resnapshotting: subsequent events fence any in-flight
+      // canonical read, including the first connection's head-to-read window.
+      notify({ type: 'resync' }, current);
+      const decoder = new TextDecoder(); let pending = '', event = '', data = [], id = null, frameBytes = 0;
+      const dispatch = () => {
+        if (event === 'reset' || event === 'snapshot') {
+          notify({ type: 'resync' }, current);
+        } else if (event === 'agent_event' && data.length) {
+          let envelope;
+          try { envelope = JSON.parse(data.join('\n')); } catch { throw new Error('Invalid Hub stream frame'); }
+          if (envelope?.schemaVersion === 1 && envelope.kind === 'diagnostic' && envelope.payload?.code === 'telemetry_gap') {
+            notify({ type: 'resync' }, current);
+          } else if (envelope?.schemaVersion === 1 && envelope.kind === 'task_updated' && envelope.payload?.resync === true) {
+            // Canonical public IDs can exceed telemetry's bounded vocabulary.
+            // A global hint preserves those tasks through a canonical resnapshot.
+            notify({ type: 'resync' }, current);
+          } else if (envelope?.schemaVersion === 1 && identifier(envelope.taskId)) {
+            const type = ['task_updated', 'lifecycle'].includes(envelope.kind) ? 'task' :
+              ['tool', 'usage', 'context', 'progress', 'diagnostic'].includes(envelope.kind) ? 'telemetry' : null;
+            if (type) notify({ type, taskId: envelope.taskId,
+              parentTaskId: identifier(envelope.parentTaskId) ? envelope.parentTaskId : null }, current);
+          }
+        }
+        if (id !== null && cursorValue(id)) cursor = id;
+        event = ''; data = []; id = null; frameBytes = 0;
+      };
+      const line = value => {
+        if (value === '') { dispatch(); return; }
+        frameBytes += value.length;
+        if (frameBytes > maxFrameBytes) throw new Error('Hub stream frame too large');
+        if (value[0] === ':') return;
+        const colon = value.indexOf(':');
+        const field = colon < 0 ? value : value.slice(0, colon);
+        let content = colon < 0 ? '' : value.slice(colon + 1);
+        if (content[0] === ' ') content = content.slice(1);
+        if (field === 'event') event = content;
+        else if (field === 'data') data.push(content);
+        else if (field === 'id') id = content;
+        else if (field === 'retry' && /^\d+$/.test(content)) reconnectMs = Math.max(250, Math.min(30000, Number(content)));
+      };
+      reader = currentReader = response.body.getReader();
+      abort.signal.addEventListener('abort', () => { void reader.cancel().catch(() => {}); }, { once: true });
+      while (active && generation === current && !abort.signal.aborted) {
+        clearTimeout(idleTimer);
+        idleTimer = setTimeout(() => abort.abort(), idleTimeoutMs); idleTimer.unref?.();
+        const { done, value } = await reader.read();
+        if (done || !active || generation !== current || abort.signal.aborted) break;
+        pending += decoder.decode(value, { stream: true });
+        let end;
+        while ((end = pending.indexOf('\n')) >= 0) {
+          const next = pending.slice(0, end).replace(/\r$/, ''); pending = pending.slice(end + 1);
+          line(next);
+        }
+        // Bound even an unterminated line; complete frames are discarded as read.
+        if (pending.length + frameBytes > maxFrameBytes) throw new Error('Hub stream frame too large');
+      }
+    } catch { failures++; /* Polling retains recovery; never export upstream errors. */ }
+    finally {
+      abort.abort();
+      try { await reader?.cancel(); } catch { /* Already aborted. */ }
+      if (generation === current) {
+        clearConnectionTimers(); controller = currentReader = null;
+        if (active) {
+          const delay = Math.min(30000, reconnectMs * 2 ** Math.min(failures, 4));
+          retryTimer = setTimeout(() => { retryTimer = null; void connect(current); }, delay);
+          retryTimer.unref?.();
+        }
+      }
+    }
+  }
+
+  return {
+    start() { if (!active) { active = true; generation++; void connect(generation); } },
+    stop() {
+      active = false; generation++; controller?.abort(); controller = null;
+      void currentReader?.cancel().catch(() => {}); currentReader = null;
+      clearTimeout(retryTimer); retryTimer = null; clearConnectionTimers();
+    },
+  };
+}
diff --git a/src/hub-inventory.mjs b/src/hub-inventory.mjs
index d85b925..0ea5a28 100644
--- a/src/hub-inventory.mjs
+++ b/src/hub-inventory.mjs
@@ -5,6 +5,34 @@ import {timestampMs,safeActivityUrl} from './activity.mjs';
 import {taskText} from './task-text.mjs';
 import {classifyOccupancy} from './occupancy.mjs';
 import {normalizePr} from './pr.mjs';
+import {attachHubTelemetry} from './hub-telemetry.mjs';
+import {createHubEventStream} from './hub-event-stream.mjs';
+
+// Hub's dashboard is a list of roots, not a lifecycle filter for their children.
+// Promote only explicitly linked, canonically fresh active sessions; cached
+// history remains untouched so a disappearing root cannot leave ghosts behind.
+function includeActiveFamilies(tasks, now) {
+  const active = cottage => cottage.conversationTarget?.recordKind==='external_session' &&
+    cottage.freshness?.isStale===false && ['working','blocked','idle'].includes(cottage.status) &&
+    ['running','queued','pending','needs_input','awaiting_input','awaiting_review'].includes(cottage.executionStatus);
+  for(const cottage of tasks.values()) {
+    if(cottage.inventoryScope!=='history'||!cottage.parent||!active(cottage))continue;
+    const seen=new Set([cottage.id]);let ancestor=tasks.get(cottage.parent),anchored=false,valid=false;
+    while(ancestor) {
+      if(seen.has(ancestor.id))break;
+      seen.add(ancestor.id);
+      if(ancestor.inventoryScope==='dashboard')anchored=true;
+      else if(!active(ancestor))break;
+      if(!ancestor.parent){valid=anchored;break;}
+      ancestor=tasks.get(ancestor.parent);
+    }
+    if(valid) {
+      const family={...cottage,inventoryScope:'family'};
+      family.occupancy=classifyOccupancy(family,now);
+      tasks.set(cottage.id,family);
+    }
+  }
+}
 
 export function toInventoryCottage(task, now=Date.now()) {
   const context=typeof task.context==='object' && task.context ? task.context : {};
@@ -27,6 +55,10 @@ export function toInventoryCottage(task, now=Date.now()) {
   },now);
   const request=taskText(task.originalRequest);
   cottage.taskId=task.id;
+  // The inventory's execution backend is explicit harness evidence; a model or
+  // the caller who dispatched the task is not a substitute for it.
+  cottage.harness=typeof task.harness==='string'&&task.harness.trim()?task.harness.trim().slice(0,120):
+    typeof task.backend==='string'&&task.backend.trim()?task.backend.trim().slice(0,120):null;
   // The shared inventory owns lifecycle evidence. Old local context and prose
   // must never override a fresher reconciliation (or invent a missing PR).
   const finalizationState=task.finalization?.status;
@@ -54,8 +86,14 @@ export function toInventoryCottage(task, now=Date.now()) {
   cottage.branch=task.branch || '';
   cottage.updatedAt=timestampMs(task.updatedAt) || cottage.updatedAt;
   cottage.inventorySource=task.source || null;
+  // The rollout observer writes cumulative usage to the shared event stream,
+  // not the legacy task counters (whose storage defaults are zero).
+  if(task.recordKind==='external_session'&&task.source==='codex_rollout') {
+    if(cottage.tokens===0)cottage.tokens=null;
+    if(cottage.cost===0)cottage.cost=null;
+  }
   cottage.dispatchedBy=task.provenance?.caller || task.provenance?.platform || task.platform || task.provider || 'hub';
-  cottage.freshness=task.freshness || {lastSeenAt:task.lastSeenAt || null,isStale:task.isStale===true};
+  cottage.freshness=task.freshness || {lastSeenAt:task.lastSeenAt || null,isStale:typeof task.isStale==='boolean'?task.isStale:null};
   cottage.resultLinks=resultLinks.map(({kind,url})=>({kind,url}));
   cottage.artifacts=resultLinks.map(link=>({url:link.url,title:taskText(link.label || link.title) || (link.kind==='pull_request'?'Pull request':'Result')}));
   cottage.conversationTarget={...cottage.conversationTarget,
@@ -75,7 +113,8 @@ export function toInventoryCottage(task, now=Date.now()) {
 
 export function createHubInventoryReader({
   baseUrl=process.env.AUTOHUB_HUB_BASE || '',token=process.env.AUTOHUB_HUB_TOKEN || process.env.COTTAGE_HUB_TOKEN || '',
-  fetchFn=globalThis.fetch,now=Date.now,ttl=5000,pageSize=500,historyTtl=60000,timeoutMs=45000,
+  fetchFn=globalThis.fetch,now=Date.now,ttl=5000,pageSize=500,historyTtl=60000,timeoutMs=45000,telemetryTimeoutMs=3000,
+  streamFactory=createHubEventStream,streamCoalesceMs=100,
 }={}) {
   let base=null;
   if(baseUrl) {
@@ -86,21 +125,81 @@ export function createHubInventoryReader({
     } catch {throw new Error('AUTOHUB_HUB_BASE must be an HTTP(S) URL without credentials.');}
   }
   let previous=null,checkedAt=0,inFlight=null,history=null,historyCheckedAt=0;
-  async function get(path) {
+  let revision=0,appliedRevision=0,fullRevision=0,appliedFullRevision=0,canonicalRevision=0;
+  const changedTasks=new Map();
+  let stream=null,onChange=null,notifyTimer=null;
+  function scheduleChange() {
+    if(onChange&&notifyTimer===null) {
+      notifyTimer=setTimeout(()=>{notifyTimer=null;onChange?.();},streamCoalesceMs);
+      notifyTimer.unref?.();
+    }
+  }
+  function invalidate(hint) {
+    const next=++revision;
+    if(hint.type==='resync'){fullRevision=next;canonicalRevision=next;}
+    else if(hint.type==='task'&&typeof hint.taskId==='string') {
+      canonicalRevision=next;
+      changedTasks.set(hint.taskId,next);
+      if(hint.parentTaskId)changedTasks.set(hint.parentTaskId,next);
+      // A replay burst is cheaper and more complete as one canonical rebuild.
+      if(changedTasks.size>200){fullRevision=next;changedTasks.clear();}
+    }
+    scheduleChange();
+  }
+  async function get(path,{allowMissing=false}={}) {
     const response=await fetchFn(new URL(path,base),{headers:{accept:'application/json',...(token?{authorization:`Bearer ${token}`}:{})},redirect:'error',signal:AbortSignal.timeout(timeoutMs)});
+    if(allowMissing&&response.status===404)return null;
     if(!response.ok)throw new Error('Hub inventory temporarily unavailable');
     return response.json();
   }
+  async function telemetry() {
+    const controller=new AbortController();let timer;
+    try {
+      return await Promise.race([
+        (async()=>{
+          const response=await fetchFn(new URL('agent-events/snapshot?limit=200',base),{headers:{accept:'application/json',...(token?{authorization:`Bearer ${token}`}:{})},redirect:'error',signal:controller.signal});
+          if(!response.ok)return null;
+          return await response.json();
+        })(),
+        new Promise(resolve=>{timer=setTimeout(()=>{controller.abort();resolve(null);},Math.max(1,telemetryTimeoutMs));}),
+      ]);
+    }catch{return null;}finally{clearTimeout(timer);}
+  }
   async function refresh() {
     try {
+      const readingRevision=revision,readingFullRevision=fullRevision;
+      const changed=[...changedTasks].filter(([,value])=>value<=readingRevision).map(([id])=>id);
+      const telemetryRead=telemetry();
       const tasks=new Map();
-      let nextHistory=null;
+      let nextHistory=null,historyPaginated=false;
       // The Hub owns Live + recent membership. History alone cannot reproduce
       // it from PR links or task clocks, and includes old external children.
       // Read dashboard last so its newer rows win if a task changes mid-scan.
       for(const scope of ['history','dashboard']) {
-        if(scope==='history'&&history&&now()-historyCheckedAt<historyTtl){
+        if(scope==='history'&&history&&readingFullRevision===appliedFullRevision&&now()-historyCheckedAt<historyTtl){
           for(const [id,cottage] of history)tasks.set(id,cottage);
+          // Notifications supply identity only. Hydrate the current canonical
+          // row (and missing ancestry), including archive/delete/unlink changes.
+          const queue=[...changed],seen=new Set();
+          while(queue.length) {
+            const batch=queue.splice(0,4).filter(id=>!seen.has(id));
+            for(const id of batch)seen.add(id);
+            const details=await Promise.all(batch.map(async id=>({id,task:await get(`tasks/${encodeURIComponent(id)}`,{allowMissing:true})})));
+            for(const {id,task} of details) {
+              if(task&&task.id!==id)throw new Error();
+              if(!task||task.archived===true){tasks.delete(id);continue;}
+              const cottage={...toInventoryCottage(task,now()),inventoryScope:'history'};
+              cottage.occupancy=classifyOccupancy(cottage,now());
+              const cached=tasks.get(id);
+              // Version, not updatedAt: a parent-link change may preserve clocks.
+              if(!cached||cottage.version>=cached.version)tasks.set(id,cottage);
+              if(cottage.parent&&!tasks.has(cottage.parent)&&!seen.has(cottage.parent)) {
+                if(seen.size+queue.length>=200)throw new Error();
+                queue.push(cottage.parent);
+              }
+            }
+          }
+          nextHistory=new Map(tasks);
           continue;
         }
         const seen=new Set();let cursor='';
@@ -111,7 +210,9 @@ export function createHubInventoryReader({
           if(!Array.isArray(page.tasks))throw new Error();
           for(const task of page.tasks) {
             if(typeof task?.id!=='string'||!task.id)throw new Error();
-            const cottage={...toInventoryCottage(task,now()),inventoryScope:scope};
+            if(task.archived===true){tasks.delete(task.id);continue;}
+            const incoming=toInventoryCottage(task,now()),cached=tasks.get(task.id);
+            const cottage={...(cached&&cached.version>incoming.version?cached:incoming),inventoryScope:scope};
             cottage.occupancy=classifyOccupancy(cottage,now());
             tasks.set(task.id,cottage);
           }
@@ -119,18 +220,44 @@ export function createHubInventoryReader({
           if(page.has_more && (typeof cursor!=='string'||!cursor||seen.has(cursor)))throw new Error();
           if(cursor)seen.add(cursor);
         }while(cursor);
-        if(scope==='history')nextHistory=new Map(tasks);
+        if(scope==='history'){nextHistory=new Map(tasks);historyPaginated=true;}
+      }
+      if(nextHistory){
+        history=nextHistory;
+        if(historyPaginated)historyCheckedAt=now();
       }
-      if(nextHistory){history=nextHistory;historyCheckedAt=now();}
-      previous={ok:true,agents:[...tasks.values()],keys:new Set(tasks.keys()),links:new Map([...tasks.keys()].map(id=>[id,new Set([id])]))};
-      checkedAt=now();return previous;
+      includeActiveFamilies(tasks,now());
+      const toolSnapshot=await telemetryRead;
+      previous={ok:true,agents:[...tasks.values()].map(cottage=>attachHubTelemetry(cottage,toolSnapshot,now())),keys:new Set(tasks.keys()),links:new Map([...tasks.keys()].map(id=>[id,new Set([id])]))};
+      checkedAt=now();appliedRevision=readingRevision;appliedFullRevision=readingFullRevision;
+      for(const [id,value] of changedTasks)if(value<=readingRevision)changedTasks.delete(id);
+      return previous;
     }catch {throw new Error('Hub inventory temporarily unavailable');}
   }
   return {configured:Boolean(base),
+    start(callback) {
+      if(!base||stream)return;
+      onChange=callback;
+      stream=streamFactory({url:new URL('agent-events/stream',base),token,fetchFn,onChange:invalidate});
+      stream.start();
+    },
+    stop() {
+      stream?.stop();stream=null;onChange=null;
+      clearTimeout(notifyTimer);notifyTimer=null;
+    },
     async read(){
       if(!base)return {ok:true,missing:true,agents:[],keys:new Set(),links:new Map()};
-      if(previous && now()-checkedAt<ttl)return previous;
-      if(!inFlight)inFlight=refresh().finally(()=>{inFlight=null;});
+      if(previous && appliedRevision===revision && now()-checkedAt<ttl)return previous;
+      if(!inFlight)inFlight=(async()=>{
+        // An event arriving during canonical hydration cannot be marked served
+        // by the older response. Catch up once, then publish a coherent pass
+        // even when canonical writes continue; dirty versions and IDs survive
+        // for the next scan rather than requiring global stream quiescence.
+        let passes=0;
+        do {await refresh();passes++;}while(appliedRevision<canonicalRevision&&passes<2);
+        if(appliedRevision!==revision)scheduleChange();
+        return previous;
+      })().finally(()=>{inFlight=null;});
       return inFlight;
     },
     async detail(id){
@@ -140,7 +267,12 @@ export function createHubInventoryReader({
         if(task.id!==id)throw new Error();
         const cottage=toInventoryCottage(task,now());
         const scope=previous?.agents.find(agent=>agent.id===id)?.inventoryScope;
-        if(scope)cottage.inventoryScope=scope;
+        if(scope)cottage.inventoryScope=scope==='family'?'history':scope;
+        if(scope==='family') {
+          const family=new Map(previous.agents.map(agent=>[agent.id,agent]));
+          family.set(id,cottage);includeActiveFamilies(family,now());
+          cottage.inventoryScope=family.get(id).inventoryScope;
+        }
         cottage.occupancy=classifyOccupancy(cottage,now());
         return cottage;
       }catch {throw new Error('Hub task detail temporarily unavailable');}
diff --git a/src/hub-telemetry.mjs b/src/hub-telemetry.mjs
new file mode 100644
index 0000000..275ac41
--- /dev/null
+++ b/src/hub-telemetry.mjs
@@ -0,0 +1,70 @@
+/** Small public telemetry facts only. Hub task inventory remains lifecycle authority. */
+const states=new Set(['started','succeeded','failed','cancelled']);
+const identifier=value=>typeof value==='string'&&value.length>0&&value.length<=120&&/^[a-zA-Z0-9_.:/@+-]+$/.test(value);
+const tokenFields=['inputTokens','outputTokens','cacheReadTokens','cacheWriteTokens','totalTokens'];
+const measured=value=>typeof value==='number'&&Number.isFinite(value)&&value>=0?value:null;
+function scopedUsage(scope,row,snapshot) {
+  const usage=scope.usage;
+  // An observed alias belongs to its managed task. Its retained meter details
+  // are diagnostic, not a second billable usage source.
+  if(row.ownedBy||usage?.supersededBy)return {usage:null,usageCoverage:{available:false,reason:'superseded'}};
+  if(!usage||typeof usage!=='object')return {usage:null,usageCoverage:{available:false,reason:'missing_usage'}};
+  const tokens=Object.fromEntries(tokenFields.map(key=>[key,measured(usage.tokens?.[key])]));
+  const sources=Array.isArray(usage.cost?.sources)?usage.cost.sources.filter(source=>['vendor','estimate','unmetered'].includes(source)):[];
+  const usd=sources.some(source=>['vendor','estimate'].includes(source))?measured(usage.cost?.usd):null;
+  if(!Object.values(tokens).some(value=>value!==null)&&usd===null)return {usage:null,usageCoverage:{available:false,reason:'missing_usage'}};
+  const incompleteFields=tokenFields.filter(key=>Array.isArray(usage.incompleteFields)&&usage.incompleteFields.includes(key)||snapshot.coverage?.complete===false&&tokens[key]!==null);
+  const lowerBound=['input','output','cacheRead','cacheWrite','total'].filter(key=>Array.isArray(usage.lowerBound)&&usage.lowerBound.includes(key));
+  return {usage:{source:'hub:agent-events',scope:scope.scope.startsWith('s:')?'session':'attempt',scopeId:scope.scope,
+    tokens,incompleteFields,lowerBound,
+    cost:{usd,sources,complete:usage.cost?.complete===true&&snapshot.coverage?.complete!==false},
+    // No usage timestamp is exported by Hub's projection. A later tool event
+    // must not make old cumulative usage look newly measured.
+    measuredAt:null,coverageComplete:snapshot.coverage?.complete===true},
+    usageCoverage:{available:true,reason:incompleteFields.length||lowerBound.length||snapshot.coverage?.complete===false?'partial':null}};
+}
+export function attachHubTelemetry(cottage,snapshot,now=Date.now()) {
+  let result={...cottage,toolActivity:null,toolCoverage:{available:false,reason:'unavailable'},
+    usage:null,usageCoverage:{available:false,reason:'unavailable'},contextUsage:null};
+  if(!snapshot||!Array.isArray(snapshot.tasks))return result;
+  const rows=snapshot.tasks.filter(row=>row?.taskId===cottage.taskId);
+  const missing=reason=>({...result,toolCoverage:{available:false,reason},
+    ...(result.usageCoverage.reason==='unavailable'?{usageCoverage:{available:false,reason}}:{})});
+  if(rows.length!==1)return missing(rows.length?'ambiguous_task':'missing_task');
+  const row=rows[0];let scope;
+  if(row.currentAttempt!=null) {
+    if(typeof row.currentAttempt!=='string'||!Array.isArray(row.attempts))return missing('ambiguous_scope');
+    // A retained event window may name an attempt whose scope has no events.
+    // Absence is not an identity conflict and must never enable session fallback.
+    if(!row.attempts.length)return missing('missing_scope');
+    const matches=row.attempts.filter(s=>s?.scope===row.currentAttempt&&s.current===true);
+    if(matches.length!==1||row.attempts.some(s=>s?.current===true&&s.scope!==row.currentAttempt))return missing('ambiguous_scope');
+    scope=matches[0];
+  } else {
+    // Never fall back from an ambiguous attempt to an observed session.
+    if(Array.isArray(row.attempts)&&row.attempts.length)return missing('ambiguous_scope');
+    if(!cottage.sessionId||!Array.isArray(row.observed))return missing('missing_scope');
+    const matches=row.observed.filter(s=>Array.isArray(s?.sessionIds)&&s.sessionIds.includes(cottage.sessionId));
+    if(!matches.length)return missing('missing_scope');
+    if(matches.length!==1||matches[0].sessionIds.length!==1||matches[0].scope!==`s:${cottage.sessionId}`)return missing('ambiguous_scope');
+    scope=matches[0];
+  }
+  if(scope.staleAttempt===true)return missing('stale_scope');
+  result={...result,...scopedUsage(scope,row,snapshot)};
+  const context=scope.context;
+  const contextAt=typeof context?.at==='string'?Date.parse(context.at):NaN;
+  if(!row.ownedBy&&!scope.usage?.supersededBy&&Number.isFinite(contextAt)&&contextAt>0&&contextAt<=now+30000){
+    const usedTokens=measured(context.usedTokens),windowTokens=measured(context.windowTokens);
+    if(usedTokens!==null||windowTokens!==null)result.contextUsage={source:'hub:agent-events',scope:scope.scope.startsWith('s:')?'session':'attempt',
+      usedTokens,windowTokens,at:contextAt,usedRatio:usedTokens!==null&&windowTokens>0?usedTokens/windowTokens:null,
+      stale:cottage.freshness?.isStale===true||now-contextAt>120000};
+  }
+  const call=scope.tool?.current?{...scope.tool.current,state:'started'}:scope.tool?.last;
+  if(!call)return missing('missing_tool');
+  const at=typeof call.at==='string'?Date.parse(call.at):NaN;
+  if(!identifier(call.name)||!states.has(call.state)||!Number.isFinite(at)||at<=0||at>now+30000||
+    (call.callId!=null&&!identifier(call.callId)))return missing('invalid_tool');
+  const stale=cottage.freshness?.isStale===true||now-at>120000;
+  return {...result,toolActivity:{name:call.name,state:call.state,at,callId:call.callId??null,source:'hub:agent-events',stale},
+    toolCoverage:{available:true,reason:snapshot.coverage?.complete===false?'partial':null}};
+}
diff --git a/test/hub-event-stream.test.mjs b/test/hub-event-stream.test.mjs
new file mode 100644
index 0000000..8c08779
--- /dev/null
+++ b/test/hub-event-stream.test.mjs
@@ -0,0 +1,110 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {createHubEventStream} from '../src/hub-event-stream.mjs';
+
+const waitFor=async predicate=>{
+  const until=Date.now()+2000;
+  while(!predicate()){
+    if(Date.now()>until)throw new Error('Timed out waiting for stream');
+    await new Promise(resolve=>setTimeout(resolve,2));
+  }
+};
+const envelope=(kind,taskId='child',extra={})=>({schemaVersion:1,kind,taskId,parentTaskId:'root',...extra});
+const frame=(id,value,event='agent_event')=>`id: ${id}\nevent: ${event}\ndata: ${JSON.stringify(value)}\n\n`;
+function connection(){
+  let controller,cancelled=false;
+  const body=new ReadableStream({start(value){controller=value;},cancel(){cancelled=true;}});
+  return {response:{ok:true,headers:new Headers({'content-type':'text/event-stream; charset=utf-8'}),body},
+    send(text){controller.enqueue(new TextEncoder().encode(text));},close(){controller.close();},get cancelled(){return cancelled;}};
+}
+
+test('stream sends server-only credentials and emits bounded identity hints from split SSE frames',async t=>{
+  const socket=connection(),seen=[],requests=[];
+  const stream=createHubEventStream({url:new URL('http://hub.test/v1/agent-events/stream'),token:'private-token',
+    fetchFn:async(url,options)=>{requests.push({url,options});return socket.response;},onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();stream.start();await waitFor(()=>seen.length===1);
+  const value=frame('1',envelope('task_updated','child',{payload:{version:3,thinking:'private thought',arguments:'private arguments'}})).replaceAll('\n','\r\n');
+  socket.send(value.slice(0,15));socket.send(value.slice(15));
+  socket.send(frame('2',envelope('tool'))+frame('3',envelope('usage',null)));
+  await waitFor(()=>seen.length===3);
+  assert.deepEqual(seen,[{type:'resync'},{type:'task',taskId:'child',parentTaskId:'root'},{type:'telemetry',taskId:'child',parentTaskId:'root'}]);
+  assert.equal(requests.length,1);assert.equal(requests[0].options.headers.authorization,'Bearer private-token');
+  assert.equal(requests[0].options.redirect,'error');assert.equal(JSON.stringify(seen).includes('private'),false);
+  stream.stop();await waitFor(()=>socket.cancelled);assert.equal(requests[0].options.signal.aborted,true);
+});
+
+test('heartbeats advance the reconnect cursor and each connection resyncs canonical inventory',async t=>{
+  const sockets=[connection(),connection()],requests=[],seen=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',retryMs:5,
+    fetchFn:async(url,options)=>{requests.push(options);return sockets[requests.length-1].response;},onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>seen.length===1);
+  sockets[0].send(frame('12',{cursor:'12'},'heartbeat'));sockets[0].close();
+  await waitFor(()=>requests.length===2);
+  assert.equal(requests[1].headers['last-event-id'],'12');
+  assert.deepEqual(seen,[{type:'resync'},{type:'resync'}]);
+});
+
+test('reset and telemetry gaps force resync without exporting diagnostic text',async t=>{
+  const sockets=[connection(),connection()],requests=[],seen=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',retryMs:5,
+    fetchFn:async(url,options)=>{requests.push(options);return sockets[requests.length-1].response;},onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>seen.length===1);
+  sockets[0].send(frame('7',envelope('diagnostic',null,{payload:{code:'telemetry_gap',message:'private'}})));
+  sockets[0].send(frame('10',{reason:'cursor_pruned'},'reset'));sockets[0].close();
+  await waitFor(()=>requests.length===2);
+  assert.equal(requests[1].headers['last-event-id'],'10');
+  assert.deepEqual(seen,Array.from({length:4},()=>({type:'resync'})));
+});
+
+test('malformed frames cannot advance the replay cursor and oversized frames are bounded',async t=>{
+  const sockets=[connection(),connection(),connection()],requests=[],seen=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',retryMs:2,maxFrameBytes:256,
+    fetchFn:async(url,options)=>{requests.push(options);return sockets[requests.length-1].response;},onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>seen.length===1);
+  sockets[0].send(frame('5',{cursor:'5'},'heartbeat')+'id: 6\nevent: agent_event\ndata: broken\n\n');
+  await waitFor(()=>requests.length===2);
+  assert.equal(requests[1].headers['last-event-id'],'5');
+  sockets[1].send('data: '+'x'.repeat(257));await waitFor(()=>requests.length===3);
+  assert.equal(requests[2].headers['last-event-id'],'5');
+  assert.equal(seen.every(hint=>hint.type==='resync'),true);
+});
+
+test('global inventory hints resync canonical IDs outside the telemetry vocabulary',async t=>{
+  const socket=connection(),seen=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',
+    fetchFn:async()=>socket.response,onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>seen.length===1);
+  socket.send(frame('1',envelope('task_updated',null,{parentTaskId:null,payload:{version:null,resync:true,private:'not exported'}})));
+  socket.send(frame('2',envelope('usage',null,{parentTaskId:null,payload:{resync:true}})));
+  await waitFor(()=>seen.length===2);
+  assert.deepEqual(seen,[{type:'resync'},{type:'resync'}]);
+});
+
+test('many complete frames in one transport chunk are processed without an aggregate-size rejection',async t=>{
+  const socket=connection(),seen=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',maxFrameBytes:256,
+    fetchFn:async()=>socket.response,onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>seen.length===1);
+  socket.send(Array.from({length:100},(_,index)=>frame(String(index+1),envelope('tool'))).join(''));
+  await waitFor(()=>seen.length===101);
+  assert.equal(seen.slice(1).every(hint=>hint.type==='telemetry'),true);
+});
+
+test('stop fences a late handshake and cancels retry timers',async()=>{
+  const socket=connection(),seen=[];let release,requests=0;
+  const waiting=new Promise(resolve=>{release=resolve;});
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',retryMs:2,
+    fetchFn:async()=>{requests++;await waiting;return socket.response;},onChange:hint=>seen.push(hint)});
+  stream.start();stream.stop();release();await waitFor(()=>socket.cancelled);
+  await new Promise(resolve=>setTimeout(resolve,10));
+  assert.equal(requests,1);assert.deepEqual(seen,[]);
+});
+
+test('a silent connected stream times out and reconnects instead of freezing inventory recovery',async t=>{
+  const sockets=[connection(),connection()],seen=[],requests=[];
+  const stream=createHubEventStream({url:'http://hub.test/v1/agent-events/stream',idleTimeoutMs:15,retryMs:2,
+    fetchFn:async(url,options)=>{requests.push(options);return sockets[requests.length-1].response;},onChange:hint=>seen.push(hint)});
+  t.after(()=>stream.stop());stream.start();await waitFor(()=>requests.length===2);
+  assert.equal(requests[0].signal.aborted,true);assert.equal(sockets[0].cancelled,true);
+  assert.deepEqual(seen,[{type:'resync'},{type:'resync'}]);
+});
diff --git a/test/hub-inventory.test.mjs b/test/hub-inventory.test.mjs
index 2793d5d..6699869 100644
--- a/test/hub-inventory.test.mjs
+++ b/test/hub-inventory.test.mjs
@@ -6,6 +6,166 @@ import {elapsedMs,normalizeCottage} from '../src/feed-client.mjs';
 import {prCi,normalizePr} from '../src/pr.mjs';
 const now=Date.parse('2026-09-22T06:00:00Z');
 const task=(id,extra={})=>({id,recordKind:'logical_task',task:'Short task',originalRequest:'Full request',status:'running',updatedAt:new Date(now).toISOString(),projectPath:'/projects/autohub',worktreePath:'/projects/autohub/.worktrees/fix',branch:'fix/cursor',source:'codex_app_server',provider:'openai',backend:'codex',sessionId:'session',currentActivity:'Testing pagination',controlTargetId:null,capabilities:{canInspect:true,canRespond:false,canSteer:false},...extra});
+
+function streamingFixture(t){
+  const records=new Map([['root',task('root',{version:1})],['other-root',task('other-root',{version:1})]]);
+  const dashboard=new Set(['root','other-root']),requests=[];
+  let emit,starts=0,stops=0,changes=0,detailHook=null;
+  const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>now,streamCoalesceMs:1,
+    streamFactory:options=>{emit=options.onChange;return {start(){starts++;},stop(){stops++;}};},
+    fetchFn:async url=>{
+      const u=new URL(url),scope=u.searchParams.get('scope');requests.push(u.pathname+u.search);
+      if(u.pathname.endsWith('/snapshot'))return {ok:true,json:async()=>({tasks:[],coverage:{complete:true}})};
+      if(u.pathname==='/v1/tasks')return {ok:true,json:async()=>({tasks:[...records.values()].filter(row=>!row.archived&&(scope!=='dashboard'||dashboard.has(row.id))).map(row=>structuredClone(row)),has_more:false})};
+      const id=decodeURIComponent(u.pathname.slice('/v1/tasks/'.length));
+      if(detailHook){const result=await detailHook(id);if(result)return result;}
+      const row=records.get(id);
+      return row?{ok:true,json:async()=>structuredClone(row)}:{ok:false,status:404};
+    }});
+  reader.start(()=>{changes++;});t.after(()=>reader.stop());
+  return {reader,records,dashboard,requests,emit:hint=>emit(hint),get changes(){return changes;},get starts(){return starts;},get stops(){return stops;},
+    setDetailHook(value){detailHook=value;},
+    historyReads:()=>requests.filter(value=>value.includes('scope=history')).length,
+    child:(id='child',extra={})=>task(id,{recordKind:'external_session',parentId:'root',source:'codex_rollout',sessionId:`session-${id}`,isStale:false,version:1,...extra})};
+}
+
+test('shared task hints hydrate new children, parent changes, completions and archives without waiting for history TTL',async t=>{
+  const f=streamingFixture(t);await f.reader.read();
+  f.records.set('child',f.child());f.emit({type:'task',taskId:'child',parentTaskId:'root'});
+  let value=await f.reader.read(),child=value.agents.find(a=>a.id==='child');
+  assert.equal(child.status,'working');assert.equal(child.occupancy,'live');assert.equal(child.inventoryScope,'family');assert.equal(child.parent,'root');
+  f.records.set('child',f.child('child',{parentId:'other-root',version:2}));
+  f.emit({type:'task',taskId:'child'});value=await f.reader.read();
+  assert.equal(value.agents.find(a=>a.id==='child').parent,'other-root','same updatedAt does not hide a newer parent version');
+  f.records.set('child',f.child('child',{parentId:'other-root',status:'completed',version:3}));
+  f.emit({type:'task',taskId:'child'});value=await f.reader.read();
+  child=value.agents.find(a=>a.id==='child');assert.equal(child.status,'done');assert.equal(child.inventoryScope,'history');assert.equal(child.occupancy,'settled');
+  f.records.set('child',f.child('child',{archived:true,version:4}));f.emit({type:'task',taskId:'child'});
+  assert.equal((await f.reader.read()).agents.some(a=>a.id==='child'),false);
+  assert.equal(f.historyReads(),1,'targeted changes never repage warm history');
+});
+
+test('new explicit ancestry is hydrated while missing, unlinked and deleted parents remain conservative',async t=>{
+  const f=streamingFixture(t);await f.reader.read();
+  f.records.set('middle',f.child('middle'));
+  f.records.set('child',f.child('child',{parentId:'middle'}));f.emit({type:'task',taskId:'child'});
+  const value=await f.reader.read();assert.deepEqual(value.agents.filter(a=>a.inventoryScope==='family').map(a=>a.id).sort(),['child','middle']);
+  f.records.set('child',f.child('child',{parentId:null,version:2}));f.emit({type:'task',taskId:'child'});
+  assert.equal((await f.reader.read()).agents.find(a=>a.id==='child').inventoryScope,'history');
+  f.records.set('child',f.child('child',{parentId:'missing',version:3}));f.emit({type:'task',taskId:'child'});
+  assert.equal((await f.reader.read()).agents.find(a=>a.id==='child').inventoryScope,'history');
+  f.records.delete('child');f.emit({type:'task',taskId:'child'});
+  assert.equal((await f.reader.read()).agents.some(a=>a.id==='child'),false);
+});
+
+test('a lifecycle notification during hydration reruns before resolving shared readers',async t=>{
+  const f=streamingFixture(t);f.records.set('child',f.child());await f.reader.read();
+  let release,entered;const gate=new Promise(resolve=>{release=resolve;}),started=new Promise(resolve=>{entered=resolve;});let calls=0;
+  f.setDetailHook(async id=>{
+    if(id!=='child'||++calls!==1)return null;
+    const old=structuredClone(f.records.get(id));entered();await gate;return {ok:true,json:async()=>old};
+  });
+  f.emit({type:'task',taskId:'child'});const reading=f.reader.read();await started;
+  f.records.set('child',f.child('child',{status:'completed',version:2}));f.emit({type:'task',taskId:'child'});
+  const joined=f.reader.read();release();
+  assert.equal((await reading).agents.find(a=>a.id==='child').status,'done');
+  assert.equal((await joined).agents.find(a=>a.id==='child').status,'done');assert.equal(calls,2);
+  assert.equal(f.historyReads(),1);
+});
+
+test('sustained canonical changes publish a coherent pass without waiting for global quiescence',async t=>{
+  let emit,changing=false,version=1,dashboardReads=0,wakeups=0,enteredThird,releaseThird;
+  const thirdStarted=new Promise(resolve=>{enteredThird=resolve;});
+  const thirdGate=new Promise(resolve=>{releaseThird=resolve;});
+  const child=()=>task('child',{recordKind:'external_session',parentId:'root',isStale:false,version,
+    status:version%2?'running':'completed'});
+  const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>now,streamCoalesceMs:1,
+    streamFactory:options=>{emit=options.onChange;return {start(){},stop(){}};},
+    fetchFn:async url=>{
+      const u=new URL(url);
+      if(u.pathname.endsWith('/snapshot'))return {ok:false};
+      if(u.pathname.endsWith('/child')){const row=child();return {ok:true,json:async()=>row};}
+      const scope=u.searchParams.get('scope');
+      if(scope==='dashboard'&&changing){
+        dashboardReads++;
+        if(dashboardReads===3){enteredThird();await thirdGate;}
+        if(changing){version++;emit({type:'task',taskId:'child'});}
+      }
+      return {ok:true,json:async()=>({tasks:scope==='history'?[task('root'),child()]:[task('root')],has_more:false})};
+    }});
+  reader.start(()=>{wakeups++;});t.after(()=>reader.stop());await reader.read();
+  changing=true;emit({type:'task',taskId:'child'});
+  const reading=reader.read();
+  const result=await Promise.race([reading.then(value=>({published:true,value})),thirdStarted.then(()=>({published:false}))]);
+  changing=false;releaseThird();await reading;
+  assert.equal(result.published,true,'two successful passes must be published while a newer task change remains pending');
+  assert.equal(dashboardReads,2);
+  const published=result.value.agents.find(a=>a.id==='child');
+  assert.equal(published.version,2);assert.equal(published.status,'done');
+  assert.equal(version,3,'a newer canonical change still exists at publication');
+  const caughtUp=(await reader.read()).agents.find(a=>a.id==='child');
+  assert.equal(caughtUp.version,3);assert.equal(caughtUp.status,'working');
+  await new Promise(resolve=>setTimeout(resolve,5));assert.ok(wakeups>0,'pending changes retain a subsequent scan wakeup');
+});
+
+test('failed targeted hydration retains dirty identities for recovery instead of accepting an old cached row',async t=>{
+  const f=streamingFixture(t);await f.reader.read();f.records.set('child',f.child());f.emit({type:'task',taskId:'child'});
+  f.setDetailHook(async()=>{throw new Error('private upstream error');});
+  await assert.rejects(f.reader.read(),/^Error: Hub inventory temporarily unavailable$/);
+  f.setDetailHook(null);
+  assert.equal((await f.reader.read()).agents.find(a=>a.id==='child').status,'working');
+});
+
+test('telemetry invalidations refresh enrichment without historical paging or lifecycle inference',async t=>{
+  const f=streamingFixture(t);f.records.set('child',f.child('child',{status:'completed'}));await f.reader.read();
+  const before=f.requests.length;for(let i=0;i<100;i++)f.emit({type:'telemetry',taskId:'child'});
+  const next=await f.reader.read();assert.equal(next.agents.find(a=>a.id==='child').status,'done');
+  assert.equal(f.historyReads(),1);assert.equal(f.requests.slice(before).some(path=>path==='/v1/tasks/child'),false);
+  assert.equal(f.requests.filter(path=>path.includes('/snapshot')).length,2);
+  await new Promise(resolve=>setTimeout(resolve,5));assert.equal(f.changes,1,'bursts produce one scan wakeup');
+});
+
+test('reconnect or reset invalidates full history and start/stop own a single subscription',async t=>{
+  const f=streamingFixture(t);await f.reader.read();f.reader.start(()=>{});assert.equal(f.starts,1);
+  f.records.set('quiet-new-child',f.child('quiet-new-child'));f.emit({type:'resync'});
+  assert.equal((await f.reader.read()).agents.find(a=>a.id==='quiet-new-child').status,'working');assert.equal(f.historyReads(),2);
+  f.records.delete('quiet-new-child');f.emit({type:'resync'});
+  assert.equal((await f.reader.read()).agents.some(a=>a.id==='quiet-new-child'),false);assert.equal(f.historyReads(),3);
+  f.reader.stop();f.reader.stop();assert.equal(f.stops,1);
+});
+
+test('Codex rollout turn boundaries never become task or persistent session starts',()=>{
+  const start=now-60000,sessionStart=now-3600000;
+  const record=task('turn',{recordKind:'external_session',source:'codex_rollout',startedAt:new Date(start).toISOString()});
+  const observed=toInventoryCottage(record,now);
+  assert.equal(observed.turnStartedAt,start);assert.equal(observed.taskStartedAt,null);assert.equal(observed.sessionStartedAt,null);
+  const knownSession=toInventoryCottage({...record,context:{sessionStartedAt:new Date(sessionStart).toISOString()}},now);
+  assert.equal(knownSession.sessionStartedAt,sessionStart);assert.equal(knownSession.turnStartedAt,start);
+  const contextSource=toInventoryCottage({...record,source:undefined,context:{source:'codex_rollout',lifecycle:{sessionStartedAt:new Date(sessionStart).toISOString()}}},now);
+  assert.equal(contextSource.turnStartedAt,start);assert.equal(contextSource.sessionStartedAt,sessionStart);
+  const missing=toInventoryCottage({...record,startedAt:null,queuedAt:new Date(start).toISOString()},now);
+  assert.equal(missing.turnStartedAt,null);assert.equal(missing.sessionStartedAt,null,'queue time is not a turn or session boundary');
+  const other=toInventoryCottage({...record,source:'codex_app_server'},now);
+  assert.equal(other.sessionStartedAt,start);assert.equal(other.turnStartedAt,null,'other adapters retain their existing clock semantics');
+});
+test('inventory preserves known legacy counters and absent measurements stay unavailable',()=>{
+ const absent=toInventoryCottage(task('missing'),now);
+ assert.equal(absent.tokens,null);assert.equal(absent.cost,null);
+ const zero=toInventoryCottage(task('zero',{inputTokens:0,totalCost:0}),now);
+ assert.equal(zero.tokens,0);assert.equal(zero.cost,0);
+ const known=toInventoryCottage(task('known',{inputTokens:100,outputTokens:20,cacheReadTokens:50,totalCost:0.12}),now);
+ assert.equal(known.tokens,170);assert.equal(known.cost,0.12);
+ const observed=toInventoryCottage(task('observed',{recordKind:'external_session',source:'codex_rollout',inputTokens:0,outputTokens:0,totalCost:0}),now);
+ assert.equal(observed.tokens,null);assert.equal(observed.cost,null);
+ const prior=toInventoryCottage(task('prior',{recordKind:'external_session',source:'codex_rollout',inputTokens:100,totalCost:0.2}),now);
+ assert.equal(prior.tokens,100);assert.equal(prior.cost,0.2);
+});
+test('execution harness uses explicit backend evidence, never provider or model inference',()=>{
+  assert.equal(toInventoryCottage(task('codex'),now).harness,'codex');
+  assert.equal(toInventoryCottage(task('claude',{backend:'claude-code'}),now).harness,'claude-code');
+  assert.equal(toInventoryCottage(task('explicit',{harness:'custom-runner'}),now).harness,'custom-runner');
+  assert.equal(toInventoryCottage(task('missing',{backend:null,model:'gpt',provider:'openai'}),now).harness,null);
+});
 test('canonical model hints preserve known providers without inventing a model family',()=>{
   assert.equal(toInventoryCottage(task('codex'),now).model,'gpt');
   assert.equal(toInventoryCottage(task('backend',{provider:null}),now).model,'gpt');
@@ -17,6 +177,7 @@ test('canonical model hints preserve known providers without inventing a model f
 test('inventory adapter follows all cursor pages and preserves canonical parent/control identities',async()=>{
   const calls=[];
   const reader=createHubInventoryReader({baseUrl:'http://hub.test',token:'secret',now:()=>now,ttl:0,fetchFn:async(url,options)=>{
+    if(new URL(url).pathname.endsWith('/snapshot'))return {ok:false};
     calls.push({url:String(url),options});
     return {ok:true,json:async()=>new URL(url).searchParams.get('cursor') ? {tasks:[task('child',{recordKind:'external_session',parentId:'canonical-parent'})],has_more:false,next_cursor:null} : {tasks:[task('canonical-parent')],has_more:true,next_cursor:'page-2'}};
   }});
@@ -75,10 +236,39 @@ test('canonical tasks can settle or disappear without old PRs resurrecting them'
   phase=2;await feed.scan();assert.equal(feed.snapshot().agents.length,0);
 });
 
+test('fresh active external children join their dashboard family without promoting unrelated history',async()=>{
+  let clock=now,rootVisible=true,childDone=false;
+  const child=(id,parentId,extra={})=>task(id,{recordKind:'external_session',parentId,isStale:false,...extra});
+  const root=task('root'),cyclicRoot=task('cyclic-root',{parentId:'cyclic-child'});
+  const rows=[root,child('child','root'),child('grandchild','child'),
+    child('stale','root',{isStale:true}),child('unknown','root',{isStale:undefined}),
+    child('done','root',{status:'completed'}),child('orphan','missing'),
+    child('cycle-a','cycle-b'),child('cycle-b','cycle-a'),cyclicRoot,child('cyclic-child','cyclic-root'),
+    task('logical',{parentId:'root',isStale:false})];
+  const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>clock,ttl:0,fetchFn:async url=>{
+    const u=new URL(url);
+    if(u.pathname.endsWith('/snapshot'))return {ok:false};
+    if(u.pathname.endsWith('/child'))return {ok:true,json:async()=>child('child','root',{status:childDone?'completed':'running'})};
+    return {ok:true,json:async()=>({tasks:u.searchParams.get('scope')==='dashboard'?(rootVisible?[root,cyclicRoot]:[]):rows,has_more:false})};
+  }});
+  const first=await reader.read();
+  assert.deepEqual(first.agents.filter(a=>a.inventoryScope==='family').map(a=>a.id).sort(),['child','grandchild']);
+  assert.equal(first.agents.find(a=>a.id==='child').occupancy,'live');
+  assert.equal(first.agents.find(a=>a.id==='child').executionStatus,'running');
+  assert.equal((await reader.detail('child')).inventoryScope,'family');
+  childDone=true;
+  assert.equal((await reader.detail('child')).inventoryScope,'history','terminal detail cannot inherit active family membership');
+  rootVisible=false;clock+=5000;
+  const next=await reader.read();
+  assert.equal(next.agents.some(a=>a.inventoryScope==='family'),false,'cached history must not retain prior family promotion');
+  assert.equal(next.agents.find(a=>a.id==='child').occupancy,'settled');
+});
+
 test('dashboard keeps refreshing while expensive historical pages are cached',async()=>{
   let clock=now;const scopes=[];
   const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>clock,ttl:0,fetchFn:async url=>{
-    const u=new URL(url),scope=u.searchParams.get('scope');scopes.push(scope);
+    const u=new URL(url);if(u.pathname.endsWith('/snapshot'))return {ok:false};
+    const scope=u.searchParams.get('scope');scopes.push(scope);
     assert.equal(u.searchParams.get('limit'),'500');
     return {ok:true,json:async()=>({tasks:[task(scope==='history'?'old':'current')],has_more:false})};
   }});
@@ -87,6 +277,21 @@ test('dashboard keeps refreshing while expensive historical pages are cached',as
   clock+=60000;await reader.read();
   assert.deepEqual(scopes,['history','dashboard','dashboard','history','dashboard']);
 });
+test('a cached history pass crossing its TTL never renews the history pagination clock',async()=>{
+  let clock=now,advanceDashboard=false,historyReads=0;
+  const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>clock,ttl:0,fetchFn:async url=>{
+    const u=new URL(url);if(u.pathname.endsWith('/snapshot'))return {ok:false};
+    const scope=u.searchParams.get('scope');
+    if(scope==='history')historyReads++;
+    if(scope==='dashboard'&&advanceDashboard)clock+=2;
+    return {ok:true,json:async()=>({tasks:[task('root')],has_more:false})};
+  }});
+  await reader.read();assert.equal(historyReads,1);
+  clock=now+59999;advanceDashboard=true;
+  await reader.read();assert.equal(historyReads,1,'the pass began with valid cached history');
+  clock=now+60002;await reader.read();
+  assert.equal(historyReads,2,'crossing the TTL during dashboard hydration cannot extend old history');
+});
 test('configured hub mode never calls standalone scanner or SQLite reader',async()=>{
   let scans=0,reads=0;
   const inventory={configured:true,read:async()=>({ok:true,agents:[toInventoryCottage(task('canonical'),now)],keys:new Set(),links:new Map()})};
diff --git a/test/hub-telemetry.test.mjs b/test/hub-telemetry.test.mjs
new file mode 100644
index 0000000..9df54cc
--- /dev/null
+++ b/test/hub-telemetry.test.mjs
@@ -0,0 +1,157 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import {attachHubTelemetry} from '../src/hub-telemetry.mjs';
+import {createHubInventoryReader} from '../src/hub-inventory.mjs';
+import {normalizeCottage} from '../src/feed-client.mjs';
+const now=Date.parse('2026-11-05T12:00:00Z');
+const cottage={taskId:'task',sessionId:'session',status:'working',freshness:{isStale:false}};
+const call={name:'read_file',at:new Date(now).toISOString(),callId:'call-1'};
+const scope={scope:'a:one',current:true,tool:{current:call}};
+const row={taskId:'task',currentAttempt:'a:one',attempts:[scope],observed:[]};
+const snapshot={tasks:[row],coverage:{complete:true},secret:'private'};
+test('only explicit safe fields survive; canonical data stays intact',()=>{
+ const value=attachHubTelemetry(cottage,{...snapshot,tasks:[{...row,attempts:[{...scope,tool:{current:{...call,arguments:'secret',output:'secret'}}}]}]},now);
+ assert.deepEqual(value.toolActivity,{name:'read_file',state:'started',at:now,callId:'call-1',source:'hub:agent-events',stale:false});
+ assert.equal(value.status,'working');assert.equal(JSON.stringify(value).includes('secret'),false);
+ assert.equal(JSON.stringify(value).includes('attempts'),false);
+});
+test('foreign, duplicate, conflicting and old attempt scopes cannot supply tools',()=>{
+ for(const tasks of [[],[{...row,taskId:'foreign'}],[row,row],[{...row,currentAttempt:'a:new'}],[{...row,attempts:[scope,{...scope,scope:'a:old'}]}],[{...row,currentAttempt:null}]]){
+  assert.equal(attachHubTelemetry(cottage,{tasks},now).toolActivity,null);
+ }
+});
+test('observed scopes require exact known unambiguous session identity',()=>{
+ const observed={scope:'s:session',sessionIds:['session'],tool:{current:call}};
+ const sample={taskId:'task',currentAttempt:null,attempts:[],observed:[observed]};
+ assert.equal(attachHubTelemetry(cottage,{tasks:[sample]},now).toolActivity.name,'read_file');
+ for(const scopes of [[{...observed,sessionIds:['foreign']}],[observed,observed],[{...observed,sessionIds:['session','other']}],[{...observed,scope:'s:other'}]])assert.equal(attachHubTelemetry(cottage,{tasks:[{...sample,observed:scopes}]},now).toolActivity,null);
+ assert.equal(attachHubTelemetry({...cottage,sessionId:null},{tasks:[sample]},now).toolActivity,null);
+});
+test('absent retained scopes are missing while conflicting identities remain ambiguous',()=>{
+ const observed={scope:'s:session',sessionIds:['session'],tool:{current:call}};
+ const absent=[{...row,attempts:[],observed:[observed]},
+  {...row,currentAttempt:null,attempts:[],observed:[]},
+  {...row,currentAttempt:null,attempts:[],observed:[{...observed,scope:'s:foreign',sessionIds:['foreign']}]}];
+ for(const value of absent){
+  const result=attachHubTelemetry(cottage,{tasks:[value]},now);
+  assert.equal(result.toolActivity,null);
+  assert.equal(result.toolCoverage.reason,'missing_scope');
+ }
+ const ambiguous=[{...row,attempts:[scope,scope]},
+  {...row,attempts:[scope,{...scope,scope:'a:other'}]},
+  {...row,currentAttempt:null,attempts:[],observed:[observed,observed]},
+  {...row,currentAttempt:null,attempts:[],observed:[{...observed,sessionIds:['session','other']}]},
+  {...row,currentAttempt:null,attempts:[],observed:[{...observed,scope:'s:other'}]}];
+ for(const value of ambiguous){
+  const result=attachHubTelemetry(cottage,{tasks:[value]},now);
+  assert.equal(result.toolActivity,null);
+  assert.equal(result.toolCoverage.reason,'ambiguous_scope');
+ }
+});
+test('malformed names, times, states and call ids degrade only telemetry',()=>{
+ for(const invalid of [{name:'cat secret args'},{name:'x'.repeat(121)},{at:'invalid'},{at:new Date(now+60000).toISOString()},{callId:'secret string'},{state:'completed'}]){
+  const last={...call,state:'succeeded',...invalid};
+  assert.equal(attachHubTelemetry(cottage,{tasks:[{...row,attempts:[{...scope,tool:{last}}]}]},now).toolActivity,null);
+ }
+ assert.equal(attachHubTelemetry(cottage,null,now).toolActivity,null);
+ assert.equal(attachHubTelemetry(cottage,snapshot,now+121000).toolActivity.stale,true);
+ for(const state of ['succeeded','failed','cancelled'])assert.equal(attachHubTelemetry(cottage,{tasks:[{...row,attempts:[{...scope,tool:{last:{...call,state}}}]}]},now).toolActivity.state,state);
+});
+test('one authenticated optional snapshot is TTL coalesced; failures and unresponsive fetches are bounded',async()=>{
+ for(const failure of ['404','throw','hang','badjson']){
+  let count=0;
+  const reader=createHubInventoryReader({baseUrl:'http://hub.test',token:'private-token',now:()=>now,telemetryTimeoutMs:20,fetchFn:async(url,options)=>{
+   if(new URL(url).pathname.endsWith('/snapshot')){
+    count++;assert.equal(new URL(url).searchParams.get('limit'),'200');assert.equal(options.headers.authorization,'Bearer private-token');
+    if(failure==='throw')throw Error('private');if(failure==='hang')return new Promise(()=>{});
+    if(failure==='badjson')return {ok:true,json:async()=>{throw Error('private');}};
+    return {ok:false};
+   }
+   return {ok:true,json:async()=>({tasks:[{id:'task',status:'running',sessionId:'session'}],has_more:false})};
+  }});
+  const start=Date.now();const [a,b]=await Promise.all([reader.read(),reader.read()]);
+  assert.equal(a,b);assert.equal(a.agents.length,1);assert.equal(a.agents[0].toolActivity,null);assert.equal(count,1);await reader.read();assert.equal(count,1);assert.ok(Date.now()-start<1000);
+ }
+});
+test('default telemetry budget accepts a cold snapshot without a real-time wait',async t=>{
+ t.mock.timers.enable({apis:['setTimeout']});
+ const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>now,fetchFn:async url=>{
+  if(new URL(url).pathname.endsWith('/snapshot'))return new Promise(resolve=>setTimeout(()=>resolve({ok:true,json:async()=>snapshot}),1800));
+  return {ok:true,json:async()=>({tasks:[{id:'task',status:'running',sessionId:'session'}],has_more:false})};
+ }});
+ const pending=reader.read();
+ t.mock.timers.tick(1800);
+ const result=await pending;
+ assert.equal(result.agents[0].toolActivity?.name,'read_file');
+});
+test('fresh telemetry reaches cached historical session cottages on subsequent inventory refreshes',async()=>{
+ let clock=now,historyReads=0,telemetryReads=0;
+ const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>clock,ttl:0,fetchFn:async url=>{
+  const u=new URL(url);
+  if(u.pathname.endsWith('/snapshot')){telemetryReads++;return {ok:true,json:async()=>({tasks:[{taskId:'historical',currentAttempt:null,attempts:[],observed:[{scope:'s:session',sessionIds:['session'],tool:{current:call}}]}]})};}
+  if(u.searchParams.get('scope')==='history'){historyReads++;return {ok:true,json:async()=>({tasks:[{id:'historical',sessionId:'session',status:'completed'}],has_more:false})};}
+  return {ok:true,json:async()=>({tasks:[],has_more:false})};
+ }});
+ assert.equal((await reader.read()).agents[0].toolActivity.name,'read_file');clock+=5000;
+ assert.equal((await reader.read()).agents[0].toolActivity.name,'read_file');assert.equal(historyReads,1);assert.equal(telemetryReads,2);
+});
+
+const sessionUsage={tokens:{inputTokens:200,outputTokens:10,cacheReadTokens:120,cacheWriteTokens:null,totalTokens:210},
+ incompleteFields:[],lowerBound:[],unavailable:['cost'],cost:{usd:null,sources:[],bySource:{},complete:false},arrivalOrdered:false,
+ meters:[{producer:'codex_rollout',sessionId:'session',meter:'session',basis:'latest_state',inputTokens:200,totalTokens:210,private:'secret'}]};
+const observedUsage={scope:'s:session',sessionIds:['session'],usage:sessionUsage,
+ context:{usedTokens:100,windowTokens:200000,at:new Date(now-1000).toISOString(),usedRatio:0.0005},
+ lastObservedAt:new Date(now).toISOString(),tool:{current:null,last:null}};
+const usageRow={taskId:'task',currentAttempt:null,attempts:[],observed:[observedUsage]};
+const usageSnapshot={tasks:[usageRow],coverage:{complete:true}};
+test('observed cumulative session usage survives without tool facts and does not become run spend',()=>{
+ const value=attachHubTelemetry({...cottage,tokens:999,cost:2},usageSnapshot,now);
+ assert.equal(value.toolCoverage.reason,'missing_tool');
+ assert.equal(value.usageCoverage.available,true);assert.equal(value.usageCoverage.reason,null);
+ assert.equal(value.usage.scope,'session');assert.equal(value.usage.scopeId,'s:session');
+ assert.equal(value.usage.tokens.totalTokens,210,'cached input is already included, not added a second time');
+ assert.equal(value.usage.cost.usd,null);assert.equal(value.usage.measuredAt,null,'latest tool observation is not a usage timestamp');
+ assert.equal(value.tokens,999);assert.equal(value.cost,2,'canonical known counters remain intact');
+ assert.deepEqual(value.contextUsage,{source:'hub:agent-events',scope:'session',usedTokens:100,windowTokens:200000,at:now-1000,usedRatio:0.0005,stale:false});
+ assert.equal(JSON.stringify(value).includes('secret'),false);assert.equal('meters' in value.usage,false);
+});
+test('usage and context use the same strict identity fence as tools',()=>{
+ for(const tasks of [[],[usageRow,usageRow],[{...usageRow,taskId:'foreign'}],
+  [{...usageRow,currentAttempt:'a:new'}],
+  [{...usageRow,observed:[observedUsage,{...observedUsage}]}],
+  [{...usageRow,observed:[{...observedUsage,sessionIds:['session','other']}]}],
+  [{...usageRow,observed:[{...observedUsage,scope:'s:other'}]}]]){
+  const value=attachHubTelemetry(cottage,{tasks},now);
+  assert.equal(value.usage,null);assert.equal(value.contextUsage,null);
+ }
+});
+test('managed ownership suppresses duplicate session usage including retained diagnostic meters',()=>{
+ for(const row of [{...usageRow,ownedBy:'managed'},
+  {...usageRow,observed:[{...observedUsage,usage:{...sessionUsage,supersededBy:'managed',superseded:sessionUsage}}]}]){
+  const value=attachHubTelemetry(cottage,{tasks:[row]},now);
+  assert.equal(value.usage,null);assert.equal(value.usageCoverage.reason,'superseded');assert.equal(value.contextUsage,null);
+ }
+});
+test('unknown, partial, malformed, and explicitly zero measurements remain distinct',()=>{
+ const partial=attachHubTelemetry(cottage,{...usageSnapshot,coverage:{complete:false}},now);
+ assert.equal(partial.usageCoverage.reason,'partial');assert.equal(partial.usage.coverageComplete,false);
+ assert.ok(partial.usage.incompleteFields.includes('totalTokens'));
+ const usage={...sessionUsage,tokens:{inputTokens:0,outputTokens:-1,totalTokens:null,cacheReadTokens:'10',cacheWriteTokens:Infinity},
+  cost:{usd:0,sources:['vendor'],complete:true}};
+ const value=attachHubTelemetry(cottage,{...usageSnapshot,tasks:[{...usageRow,observed:[{...observedUsage,usage}]}]},now);
+ assert.deepEqual(value.usage.tokens,{inputTokens:0,outputTokens:null,cacheReadTokens:null,cacheWriteTokens:null,totalTokens:null});
+ assert.equal(value.usage.cost.usd,0);assert.equal(value.usage.cost.complete,true);
+ const stale=attachHubTelemetry(cottage,usageSnapshot,now+121000);
+ assert.equal(stale.contextUsage.stale,true);assert.equal(stale.usage.measuredAt,null);
+});
+test('real observer shape reaches the browser without turning missing legacy measurements into zero',async()=>{
+ const task={id:'task',sessionId:'session',recordKind:'external_session',source:'codex_rollout',backend:'codex',
+  status:'running',inputTokens:0,outputTokens:0,cacheReadTokens:0,cacheWriteTokens:0,totalCost:0};
+ const reader=createHubInventoryReader({baseUrl:'http://hub.test',now:()=>now,fetchFn:async url=>({ok:true,json:async()=>
+  new URL(url).pathname.endsWith('/snapshot')?usageSnapshot:{tasks:[task],has_more:false}})});
+ const agent=(await reader.read()).agents[0];
+ const value=normalizeCottage(agent,0,{town:x=>x,model:x=>x,occupancy:()=> 'live',now});
+ assert.equal(value.tokens,null);assert.equal(value.cost,null);
+ assert.equal(value.usage.tokens.totalTokens,210);assert.equal(value.usage.cost.usd,null);
+ assert.equal(value.usage.scope,'session');assert.equal(value.contextUsage.usedTokens,100);
+});