Index runtime hook subscriptions and share scoped sync counts

headless/README.md

@@ -277,7 +277,8 @@ Import from `/react` for React only, or the re-export in `/react-flow`.
   its own lease. Controller replacements keep independent token slots. Preview
   geometry remains shared across views; active ownership and controls belong to
   the owning view. Host bindings still handle pointer capture and events.
-- `useOperation(id)` observes both the logical record and its frozen request.
+- `useOperation(id)` observes both the logical record and its frozen request;
+  request-only changes are included without observing unrelated Root fields.
 - `useAsset(id)` resolves a releasable display handle, returns metadata plus
   loading/ready/error/url/retry, and releases on ref/owner change or unmount.
   Ready means the display handle resolved; it does not prove upload ACK, image
@@ -542,3 +543,23 @@ subscription removes its bucket and releases the upstream listener. A controller
 identity change (engine/view/target/incarnation) creates a private lease slot, so
 old cleanup cannot consume the replacement controller's token. This does not
 change the one shared gesture owner, its history unit or durable task lifecycle.
+
+
+Operation, asset metadata and sync-count channels share one upstream editor
+listener per engine. Their dependency tuples contain only the tables they read
+(and the loading flag for counts). Table identities use weak object IDs, so a
+channel caches a primitive version signature and selected values rather than an
+old Root or full table. Changed tables recheck subscribed IDs; unchanged tables
+skip those reads. Read caches and last-notified values stay separate so an early
+`getSnapshot()` cannot consume another observer's notification.
+
+Sync counts build one scalar index per operations/requests/loading version:
+project totals plus canvas→counts. Requests spanning multiple pending operations
+count once per matching canvas. Operations with no canvas remain in project
+counts. Unknown canvas queries return the shared zero count without growing the
+index. This preserves the existing blocked/unknown definitions; table changes
+still require a traversal, not a per-record ledger write-set update.
+
+Asset display effects keep their previous ownership, abort/release and retry
+rules. Durable-only completion/error/close is still read from the independent
+durability source, so saving and synchronization remain distinct.

src/lib/canvas-bindings/react.ts

@@ -3,12 +3,13 @@ import {createContext,createElement,useCallback,useContext,useEffect,useMemo,use
 import {createSelectorStore,type ExternalSource} from './selectors.ts';
 import {createElementSources} from './element-sources.ts';
 import {createSessionSources,type GestureSelection} from './session-sources.ts';
+import {createRuntimeSources,type OperationView} from './runtime-sources.ts';
+export type {OperationView} from './runtime-sources.ts';
 import type {ProjectEngine} from '../canvas-runtime/workspace.ts';
 import type {EngineSnapshot} from '../canvas-runtime/contracts.ts';
 import type {Target,Id,Data,Geometry,ElementRecord,Availability,Result} from '../canvas-core/types.ts';
 import type {ElementLookup} from '../canvas-core/queries.ts';
 import {equal,fault} from '../canvas-core/values.ts';
-import type {OperationRecord,FrozenRequest} from '../canvas-runtime/ledger.ts';
 import type {DisplayHandle} from '../canvas-runtime/resources.ts';
 
 export type CanvasBinding={readonly engine:ProjectEngine;readonly viewId?:Id};
@@ -68,24 +69,22 @@ export function useGestureBindings(target:Target){
  },[engine,viewId,id,incarnation]);
  useEffect(()=>()=>controls.cancel(),[controls]);return useMemo(()=>Object.freeze({...state,...controls}),[state,controls]);
 }
-export type OperationView={readonly operation:OperationRecord;readonly request?:FrozenRequest}|undefined;
-export function useOperation(id:Id):OperationView{return useCanvas(useCallback(root=>{const operation=root.runtime.ledger.operations.get(id);return operation?{operation,...(operation.requestId?{request:root.runtime.ledger.requests.get(operation.requestId)}:{})}:undefined;},[id]),equal);}
+const runtimeSources=new WeakMap<ProjectEngine,ReturnType<typeof createRuntimeSources>>();
+function runtimeSourcesFor(engine:ProjectEngine){let sources=runtimeSources.get(engine);if(!sources){sources=createRuntimeSources({getSnapshot:engine.getSnapshot,subscribe:notify=>engine.getDurability().closed?()=>{}:engine.editor.subscribe(notify)});runtimeSources.set(engine,sources);}return sources;}
+export function useOperation(id:Id):OperationView{const {engine}=useCanvasBinding(),source=useMemo(()=>runtimeSourcesFor(engine).operation(id),[engine,id]);return useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot);}
 type Durability=ReturnType<ProjectEngine['getDurability']>;
 const syncDurability=(value:Durability)=>({dirty:value.rootRevision>value.durableRootRevision,error:value.error,closing:value.closing,closed:value.closed});
 const inactiveOwner=(value:Durability)=>value.closing||value.closed;
 function useDurability<T>(selector:(value:Durability)=>T,isEqual:(a:T,b:T)=>boolean=Object.is){const {engine}=useCanvasBinding();const source=useMemo(()=>({getSnapshot:engine.getDurability,subscribe:engine.subscribeDurability}),[engine]);return useSelected(source,selector,isEqual);}
 export function useSyncStatus(canvasId?:Id){
- const durable=useDurability(syncDurability,equal),counts=useCanvas(useCallback(root=>{
-  const operations=[...root.runtime.ledger.operations.values()].filter(row=>row.phase!=='resolved'&&(canvasId===undefined||row.intent.canvasIds.includes(canvasId))),ids=new Set(operations.map(row=>row.intent.id)),requests=[...root.runtime.ledger.requests.values()].filter(row=>row.phase!=='resolved'&&row.operationIds.some(id=>ids.has(id)));
-  const blocked=operations.filter(row=>row.phase==='blocked'&&row.blockers.some(reason=>reason!=='waiting_digest')).length+requests.filter(row=>row.error&&row.retryAt===undefined).length,unknown=requests.filter(row=>row.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(row.phase)).length;
-  return {pending:operations.length,blocked,unknown,loading:!!root.runtime.loading};
- },[canvasId]),equal);
+ const {engine}=useCanvasBinding(),source=useMemo(()=>runtimeSourcesFor(engine).counts(canvasId),[engine,canvasId]);
+ const durable=useDurability(syncDurability,equal),counts=useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot);
  const dirty=durable.dirty;
  return useMemo(()=>Object.freeze({...counts,dirty,error:durable.error,status:durable.closed?'closed' as const:durable.error||counts.blocked?'blocked' as const:durable.closing?'closing' as const:dirty?'saving' as const:counts.pending?'syncing' as const:counts.loading?'loading' as const:'synced' as const}),[counts,dirty,durable.error,durable.closing,durable.closed]);
 }
 export function useAsset(id:Id){
  const {engine}=useCanvasBinding(),inactive=useDurability(inactiveOwner);
- const metadata=useCanvas(useCallback(root=>({record:root.runtime.resources.get(id),binding:root.runtime.assetBindings?.get(id)}),[id]),equal);
+ const source=useMemo(()=>runtimeSourcesFor(engine).asset(id),[engine,id]),metadata=useSyncExternalStore(source.subscribe,source.getSnapshot,source.getSnapshot);
  const key=JSON.stringify([metadata.record?.operationId,metadata.record?.retirement?.purgedAt,metadata.binding?.authority,metadata.binding?.digest]);
  const [attempt,setAttempt]=useState(0),[state,setState]=useState<{engine:ProjectEngine;id:Id;key:string;status:'loading'|'ready'|'error';url?:string;error?:unknown}>();
  useEffect(()=>{if(inactive){setState({engine,id,key,status:'error',error:fault('engine_closing')});return;}const controller=new AbortController();let handle:DisplayHandle|undefined;setState({engine,id,key,status:'loading'});void engine.displayResource(id,{signal:controller.signal}).then(value=>{if(controller.signal.aborted){value.release();return;}handle=value;setState({engine,id,key,status:'ready',url:value.url});},error=>{if(!controller.signal.aborted)setState({engine,id,key,status:'error',error});});return()=>{controller.abort();handle?.release();};},[engine,id,key,attempt,inactive]);

src/lib/canvas-bindings/runtime-sources.ts

@@ -0,0 +1,79 @@
+import type {Id} from '../canvas-core/types.ts';
+import type {EngineSnapshot} from '../canvas-runtime/contracts.ts';
+import type {OperationRecord,FrozenRequest} from '../canvas-runtime/ledger.ts';
+import type {ResourceRecord} from '../canvas-runtime/resources.ts';
+import type {SharedAssetBinding} from '../canvas-runtime/asset-depot.ts';
+import {equal} from '../canvas-core/values.ts';
+import type {ExternalSource} from './selectors.ts';
+
+export type OperationView={readonly operation:OperationRecord;readonly request?:FrozenRequest}|undefined;
+export type AssetMetadata={readonly record:ResourceRecord|undefined;readonly binding:SharedAssetBinding|undefined};
+export type SyncCounts={readonly pending:number;readonly blocked:number;readonly unknown:number;readonly loading:boolean};
+type Dependencies=readonly(object|boolean|undefined)[];
+
+/** Read-only table projections: one upstream listener, no per-hook Root/table snapshots. */
+export function createRuntimeSources(source:ExternalSource<EngineSnapshot>){
+ type Group={update(root:EngineSnapshot,notifications:Set<()=>void>):void};
+ const active=new Set<Group>(),identities=new WeakMap<object,number>();let nextIdentity=0,stop:(()=>void)|undefined;
+ function signature(values:Dependencies){return values.map(value=>{if(value===undefined)return'u';if(typeof value==='boolean')return value?'t':'f';let id=identities.get(value);if(id===undefined){id=++nextIdentity;identities.set(value,id);}return`o${id}`;}).join(':');}
+ function changed(){
+  if(!active.size)return;const root=source.getSnapshot(),notifications=new Set<()=>void>();
+  for(const group of active)group.update(root,notifications);
+  let failed=false,error:unknown;for(const notify of notifications)try{notify();}catch(caught){if(!failed){failed=true;error=caught;}}if(failed)throw error;
+ }
+ function group<K,V>(dependencies:(root:EngineSnapshot)=>Dependencies,read:(root:EngineSnapshot,key:K)=>V){
+  // Direct reads may precede notification, so read caches cannot replace observed baselines.
+  type Bucket={cached:V;cachedSignature:string;observed:V;listeners:Set<()=>void>};
+  const buckets=new Map<K,Bucket>(),failed=new Set<K>();let observedSignature:string|undefined;
+  function readCached(root:EngineSnapshot,key:K,bucket:Bucket,stamp:string){
+   if(bucket.cachedSignature!==stamp){const value=read(root,key);if(!equal(bucket.cached,value))bucket.cached=value;bucket.cachedSignature=stamp;}
+   return bucket.cached;
+  }
+  const owner:Group={update(root,notifications){
+   const stamp=signature(dependencies(root)),all=stamp!==observedSignature;if(!all&&!failed.size)return;observedSignature=stamp;
+   for(const key of all?buckets.keys():[...failed]){
+    const bucket=buckets.get(key);if(!bucket)continue;
+    try{const value=readCached(root,key,bucket,stamp);if(!failed.has(key)&&equal(bucket.observed,value))continue;bucket.observed=value;failed.delete(key);}catch{failed.add(key);}
+    for(const notify of bucket.listeners)notifications.add(notify);
+   }
+  }};
+  return{forKey(key:K):ExternalSource<V>{
+   let initialized=false,local:V,localSignature:string;
+   function getSnapshot(){
+    const root=source.getSnapshot(),stamp=signature(dependencies(root)),bucket=buckets.get(key);if(bucket)return readCached(root,key,bucket,stamp);
+    if(!initialized||stamp!==localSignature){const value=read(root,key);if(!initialized||!equal(local,value))local=value;localSignature=stamp;initialized=true;}
+    return local;
+   }
+   return Object.freeze({getSnapshot,subscribe(notify:()=>void){
+    let bucket=buckets.get(key);if(!bucket){const value=getSnapshot();if(!buckets.size)observedSignature=signature(dependencies(source.getSnapshot()));bucket={cached:value,cachedSignature:signature(dependencies(source.getSnapshot())),observed:value,listeners:new Set()};buckets.set(key,bucket);active.add(owner);}
+    const listener=()=>notify();bucket.listeners.add(listener);
+    const remove=()=>{bucket!.listeners.delete(listener);if(!bucket!.listeners.size){buckets.delete(key);failed.delete(key);}if(!buckets.size){active.delete(owner);observedSignature=undefined;}};
+    if(!stop)try{stop=source.subscribe(changed);}catch(error){remove();throw error;}
+    let live=true;return()=>{if(!live)return;live=false;remove();if(!active.size){stop?.();stop=undefined;}};
+   }});
+  }};
+ }
+ const operations=group<Id,OperationView>(root=>[root.runtime.ledger.operations,root.runtime.ledger.requests],(root,id)=>{const operation=root.runtime.ledger.operations.get(id);return operation?Object.freeze({operation,...(operation.requestId?{request:root.runtime.ledger.requests.get(operation.requestId)}:{})}):undefined;});
+ const assets=group<Id,AssetMetadata>(root=>[root.runtime.resources,root.runtime.assetBindings],(root,id)=>Object.freeze({record:root.runtime.resources.get(id),binding:root.runtime.assetBindings?.get(id)}));
+ let countStamp:string|undefined,projectCounts:SyncCounts,emptyCounts:SyncCounts,canvasCounts=new Map<Id,SyncCounts>();
+ const countDependencies=(root:EngineSnapshot):Dependencies=>[root.runtime.ledger.operations,root.runtime.ledger.requests,!!root.runtime.loading];
+ const counts=group<Id|undefined,SyncCounts>(countDependencies,(root,canvasId)=>{
+  const stamp=signature(countDependencies(root));
+  if(stamp!==countStamp){
+   const loading=!!root.runtime.loading,empty=()=>({pending:0,blocked:0,unknown:0,loading}),project=empty(),byCanvas=new Map<Id,ReturnType<typeof empty>>();
+   const scope=(id:Id)=>{let row=byCanvas.get(id);if(!row){row=empty();byCanvas.set(id,row);}return row;};
+   for(const operation of root.runtime.ledger.operations.values())if(operation.phase!=='resolved'){
+    const ids=operation.intent.canvasIds.length>1?new Set(operation.intent.canvasIds):operation.intent.canvasIds,blocked=operation.phase==='blocked'&&operation.blockers.some(reason=>reason!=='waiting_digest');project.pending++;if(blocked)project.blocked++;
+    for(const id of ids){const row=scope(id);row.pending++;if(blocked)row.blocked++;}
+   }
+   for(const request of root.runtime.ledger.requests.values())if(request.phase!=='resolved'){
+    let matched=false;const ids=new Set<Id>();for(const id of request.operationIds){const operation=root.runtime.ledger.operations.get(id);if(operation&&operation.phase!=='resolved'){matched=true;for(const canvas of operation.intent.canvasIds)ids.add(canvas);}}
+    if(!matched)continue;const blocked=!!request.error&&request.retryAt===undefined,unknown=request.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(request.phase);
+    if(blocked)project.blocked++;if(unknown)project.unknown++;for(const id of ids){const row=scope(id);if(blocked)row.blocked++;if(unknown)row.unknown++;}
+   }
+   projectCounts=Object.freeze(project);emptyCounts=Object.freeze(empty());canvasCounts=new Map([...byCanvas].map(([id,row])=>[id,Object.freeze(row)]));countStamp=stamp;
+  }
+  return canvasId===undefined?projectCounts:canvasCounts.get(canvasId)??emptyCounts;
+ });
+ return Object.freeze({operation:operations.forKey,asset:assets.forKey,counts:(canvasId?:Id)=>counts.forKey(canvasId)});
+}

tests/headless/package.integration.mjs

@@ -78,10 +78,10 @@ try {
     import React from 'react';import {renderToStaticMarkup} from 'react-dom/server';
     import {createReactRenderers,CanvasRendererCommandsContext,useCanvasRendererCommands} from '@endoleap/headless-canvas/react-flow';
     import {createRegistry,builtinComponents} from '@endoleap/headless-canvas/core';
-    import {CanvasProvider,useCanvas,useSyncStatus,useElement,useMode,useGestureBindings} from '@endoleap/headless-canvas/react';
-    const snapshot={rootRevision:0,runtime:{ledger:{operations:new Map(),requests:new Map()}},document:{projectId:'react-consumer'}};const durable={rootRevision:0,durableRootRevision:0,closed:false,closing:false};const hookEngine={getSnapshot:()=>snapshot,getDurability:()=>durable,subscribeDurability:()=>()=>{},editor:{subscribe:()=>()=>{},getQueries:()=>({lookupElement:()=>({status:'absent'})})},sessions:{getSnapshot:()=>({views:new Map([['ssr',{modes:new Map()}]]),gesture:null}),subscribe:()=>()=>{}}};
-    function HookView(){const id=useCanvas(value=>value.document.projectId),status=useSyncStatus(),elementStatus=useElement('missing',view=>view.status),mode=useMode({id:'missing',incarnation:1}),gesture=useGestureBindings({id:'missing',incarnation:1});return React.createElement('span',null,id+':'+status.status+':'+elementStatus+':'+mode.mode+':'+gesture.active);}
-    assert.equal(renderToStaticMarkup(React.createElement(CanvasProvider,{engine:hookEngine,viewId:'ssr'},React.createElement(HookView))),'<span>react-consumer:synced:absent:null:false</span>');
+    import {CanvasProvider,useCanvas,useSyncStatus,useElement,useMode,useGestureBindings,useOperation,useAsset} from '@endoleap/headless-canvas/react';
+    const snapshot={rootRevision:0,runtime:{ledger:{operations:new Map(),requests:new Map()},resources:new Map()},document:{projectId:'react-consumer'}};const durable={rootRevision:0,durableRootRevision:0,closed:false,closing:false};const hookEngine={getSnapshot:()=>snapshot,getDurability:()=>durable,subscribeDurability:()=>()=>{},editor:{subscribe:()=>()=>{},getQueries:()=>({lookupElement:()=>({status:'absent'})})},sessions:{getSnapshot:()=>({views:new Map([['ssr',{modes:new Map()}]]),gesture:null}),subscribe:()=>()=>{}}};
+    function HookView(){const id=useCanvas(value=>value.document.projectId),status=useSyncStatus(),elementStatus=useElement('missing',view=>view.status),mode=useMode({id:'missing',incarnation:1}),gesture=useGestureBindings({id:'missing',incarnation:1}),operation=useOperation('missing'),asset=useAsset('missing');return React.createElement('span',null,id+':'+status.status+':'+elementStatus+':'+mode.mode+':'+gesture.active+':'+(operation===undefined)+':'+asset.status);}
+    assert.equal(renderToStaticMarkup(React.createElement(CanvasProvider,{engine:hookEngine,viewId:'ssr'},React.createElement(HookView))),'<span>react-consumer:synced:absent:null:false:true:loading</span>');
     const registry=createRegistry(builtinComponents());let dispatched=0,seen;const commands={dispatch(){dispatched++;return{status:'applied',revision:1,operationIds:[]};},transaction(){throw Error('unused');}};
     function View(){seen=useCanvasRendererCommands();return React.createElement('span',null,'peer renderer');}
     const renderers=createReactRenderers([{componentId:'text',dataVersion:1,structure:'node',rendererId:'fixture',render:View}],registry);
@@ -92,9 +92,9 @@ try {
   await writeFile(join(packageRoot,'react-consumer-types.mts'),`
     import {createElement} from 'react';import type {NodeProps,EdgeProps} from '@xyflow/react';
     import {createReactRenderers,type ReactFlowProjectionNode,type ReactFlowProjectionEdge} from '@endoleap/headless-canvas/react-flow';
-    import {CanvasProvider,useCanvas,useElement,useActions,useMode,useGestureBindings,useOperation,useAsset,useSyncStatus} from '@endoleap/headless-canvas/react';
+    import {CanvasProvider,useCanvas,useElement,useActions,useMode,useGestureBindings,useOperation,useAsset,useSyncStatus,type OperationView} from '@endoleap/headless-canvas/react';
     import type {ProjectEngine} from '@endoleap/headless-canvas/runtime';
-    function Hooks(){const ref={id:'a',incarnation:1};const title=useCanvas(root=>root.document.projectId);useElement(ref);const selectedId:string|undefined=useElement(ref,view=>view.status==='present'?view.element.id:undefined);useElement(ref,view=>({status:view.status}),(a,b)=>a.status===b.status);void selectedId;const actions=useActions(ref);useMode(ref);useGestureBindings(ref);useOperation('operation');useAsset('resource');useSyncStatus('canvas');return createElement('button',{onClick:()=>actions.rename(title,{label:'Rename'})},title);}
+    function Hooks(){const ref={id:'a',incarnation:1};const title=useCanvas(root=>root.document.projectId);useElement(ref);const selectedId:string|undefined=useElement(ref,view=>view.status==='present'?view.element.id:undefined);useElement(ref,view=>({status:view.status}),(a,b)=>a.status===b.status);void selectedId;const actions=useActions(ref);useMode(ref);useGestureBindings(ref);const operation:OperationView=useOperation('operation');void operation;useAsset('resource');useSyncStatus('canvas');return createElement('button',{onClick:()=>actions.rename(title,{label:'Rename'})},title);}
     export function BoundHooks({engine}:{engine:ProjectEngine}){return createElement(CanvasProvider,{engine,viewId:'view'},createElement(Hooks));}
     function Card({data}:NodeProps<ReactFlowProjectionNode>){return createElement('div',null,data.title);}
     function Link({data}:EdgeProps<ReactFlowProjectionEdge>){return createElement('span',null,data?.__headlessRecord?.id);}

tests/headless/runtime-sources.test.mjs

@@ -0,0 +1,51 @@
+import {test} from 'node:test';import assert from 'node:assert/strict';
+import {createRuntimeSources} from '../../src/lib/canvas-bindings/runtime-sources.ts';
+class Tracked extends Map{gets=0;scans=0;broken=false;get(id){this.gets++;if(this.broken)throw Error('table unavailable');return super.get(id);}values(){this.scans++;return super.values();}reset(){this.gets=0;this.scans=0;}}
+const operation=(id,canvasIds=['a'],phase='pending',blockers=[])=>({intent:{id,canvasIds},phase,blockers});
+function fixture(count=3){let root={document:{},runtime:{ledger:{operations:new Tracked(Array.from({length:count},(_,index)=>[String(index),operation(String(index))])),requests:new Tracked()},resources:new Tracked(Array.from({length:count},(_,index)=>[String(index),{id:String(index),name:'image',operationId:'upload-'+index}])),assetBindings:new Tracked()}},broken=false;const listeners=new Set();const source={getSnapshot:()=>root,subscribe(fn){if(broken)throw Error('subscribe failed');listeners.add(fn);return()=>listeners.delete(fn);}};return{source,get:()=>root,get size(){return listeners.size;},broken(value){broken=value;},emit(next){root=next;for(const fn of [...listeners])fn();}};}
+
+test('2000 asset/operation subscribers share one listener and unrelated document cuts read no tables',()=>{
+ const f=fixture(1000),sources=createRuntimeSources(f.source),seen=[],stops=[];for(let index=0;index<1000;index++){const id=String(index);stops.push(sources.asset(id).subscribe(()=>seen.push('asset'+id)),sources.operation(id).subscribe(()=>seen.push('op'+id)));}assert.equal(f.size,1);const {resources,ledger}=f.get().runtime;resources.reset();ledger.operations.reset();ledger.requests.reset();f.emit({...f.get(),document:{edited:true}});assert.equal(resources.gets+ledger.operations.gets+ledger.requests.gets,0);assert.deepEqual(seen,[]);
+ const next=new Tracked(resources);next.set('300',{...resources.get('300'),name:'new image'});f.emit({...f.get(),runtime:{...f.get().runtime,resources:next}});assert.deepEqual(seen,['asset300']);assert.equal(ledger.operations.gets+ledger.requests.gets,0);for(const stop of stops)stop();assert.equal(f.size,0);
+});
+
+test('request-only updates notify the linked operation while unrelated operation views stay stable',()=>{
+ const f=fixture(),ops=new Tracked(f.get().runtime.ledger.operations),requests=new Tracked([['request',{id:'request',phase:'waiting',outcome:'accepted'}]]);ops.set('0',{...ops.get('0'),requestId:'request'});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests}}});const sources=createRuntimeSources(f.source),a=sources.operation('0'),b=sources.operation('1'),seen=[],before=b.getSnapshot(),stops=[a.subscribe(()=>seen.push('0')),b.subscribe(()=>seen.push('1'))];const changed=new Tracked(requests);changed.set('request',{id:'request',phase:'waiting',outcome:'unresolved',error:'transport'});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests:changed}}});assert.deepEqual(seen,['0']);assert.equal(a.getSnapshot().request.error,'transport');assert.strictEqual(b.getSnapshot(),before);for(const stop of stops)stop();
+});
+
+test('early snapshot reads cannot consume a binding notification before other observers receive it',()=>{
+ const f=fixture();let channel;const earlier=f.source.subscribe(()=>channel?.getSnapshot()),sources=createRuntimeSources(f.source);channel=sources.asset('shared');let seen=0;const stop=channel.subscribe(()=>seen++),binding={authority:'scope',digest:'a'.repeat(64),asset:{id:'shared'}};f.emit({...f.get(),runtime:{...f.get().runtime,assetBindings:new Tracked([['shared',binding]])}});assert.equal(seen,1);assert.equal(channel.getSnapshot().binding.digest,binding.digest);stop();earlier();assert.equal(f.size,0);
+});
+
+test('sync counts index shared multi-canvas requests once and handles project-only operations',()=>{
+ const f=fixture(0),ops=new Tracked([['one',operation('one',['a','b','a'])],['two',operation('two',['b'],'blocked',['permission'])],['digest',operation('digest',['a'],'blocked',['waiting_digest'])],['done',operation('done',['a'],'resolved')],['project',operation('project',[])]]),requests=new Tracked([['r1',{phase:'waiting',outcome:'unresolved',error:'transport',operationIds:['one','two']}],['r2',{phase:'waiting',outcome:'unresolved',error:'old',operationIds:['done']}],['r3',{phase:'waiting',outcome:'unresolved',operationIds:['project']}],['r4',{phase:'waiting',outcome:'unresolved',error:'retry',retryAt:100,operationIds:['digest']}]]);f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations:ops,requests},loading:{version:6}}});const sources=createRuntimeSources(f.source);
+ assert.deepEqual(sources.counts().getSnapshot(),{pending:4,blocked:2,unknown:3,loading:true});assert.deepEqual(sources.counts('a').getSnapshot(),{pending:2,blocked:1,unknown:2,loading:true});assert.deepEqual(sources.counts('b').getSnapshot(),{pending:2,blocked:2,unknown:1,loading:true});for(let index=0;index<1000;index++)assert.deepEqual(sources.counts('missing-'+index).getSnapshot(),{pending:0,blocked:0,unknown:0,loading:true});assert.equal(ops.scans,1);assert.equal(requests.scans,1);
+ const channel=sources.counts('a'),stop=channel.subscribe(()=>{});ops.reset();requests.reset();f.emit({...f.get(),document:{unrelated:true}});channel.getSnapshot();assert.equal(ops.scans+requests.scans,0);f.emit({...f.get(),runtime:{...f.get().runtime,loading:undefined}});assert.equal(channel.getSnapshot().loading,false);assert.equal(ops.scans,1);assert.equal(requests.scans,1);stop();
+});
+
+test('cloned equal tables preserve view identity and unsubscribed channels still read current facts',()=>{
+ const f=fixture(),sources=createRuntimeSources(f.source),channel=sources.asset('0'),before=channel.getSnapshot();let calls=0;const stop=channel.subscribe(()=>calls++);f.emit({...f.get(),runtime:{...f.get().runtime,resources:new Tracked([...f.get().runtime.resources].map(([id,row])=>[id,structuredClone(row)]))}});assert.equal(calls,0);assert.strictEqual(channel.getSnapshot(),before);stop();const resources=new Tracked(f.get().runtime.resources);resources.delete('0');f.emit({...f.get(),runtime:{...f.get().runtime,resources}});assert.equal(channel.getSnapshot().record,undefined);assert.equal(f.size,0);
+});
+
+test('lookup failures retry without poisoning cached facts, and failed subscription retains no live group',()=>{
+ const f=fixture(),sources=createRuntimeSources(f.source),channel=sources.asset('0');f.broken(true);assert.throws(()=>channel.subscribe(()=>{}),/subscribe failed/);f.broken(false);let calls=0;const stop=channel.subscribe(()=>calls++),before=channel.getSnapshot(),resources=new Tracked(f.get().runtime.resources);resources.broken=true;f.emit({...f.get(),runtime:{...f.get().runtime,resources}});assert.equal(calls,1);assert.throws(channel.getSnapshot,/table unavailable/);resources.broken=false;f.emit({...f.get(),document:{metadata:true}});assert.equal(calls,2);assert.strictEqual(channel.getSnapshot(),before);stop();stop();assert.equal(f.size,0);
+});
+
+test('observer failures do not starve another collection and late captured callbacks after cleanup are safe',()=>{
+ const f=fixture(),sources=createRuntimeSources(f.source);let other=0;const a=sources.asset('0').subscribe(()=>{throw Error('observer failed');}),b=sources.operation('0').subscribe(()=>other++),resources=new Tracked(f.get().runtime.resources),ops=new Tracked(f.get().runtime.ledger.operations);resources.set('0',{id:'0',operationId:'upload',name:'changed'});ops.set('0',operation('0',['a'],'blocked',['permission']));assert.throws(()=>f.emit({...f.get(),runtime:{...f.get().runtime,resources,ledger:{...f.get().runtime.ledger,operations:ops}}}),/observer failed/);assert.equal(other,1);a();b();assert.equal(f.size,0);
+ let stop;const earlier=f.source.subscribe(()=>stop());stop=sources.counts().subscribe(()=>{throw Error('detached observer');});assert.doesNotThrow(()=>f.emit({...f.get(),runtime:{...f.get().runtime,loading:{}}}));earlier();assert.equal(f.size,0);
+});
+
+test('request status changes update scoped sync counts and unrelated requests preserve cached count identity',()=>{
+ const f=fixture(1),request={phase:'waiting',outcome:'accepted',operationIds:['0']};f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests:new Tracked([['r',request]])}}});const sources=createRuntimeSources(f.source),channel=sources.counts('a');let calls=0;const stop=channel.subscribe(()=>calls++);assert.deepEqual(channel.getSnapshot(),{pending:1,blocked:0,unknown:0,loading:false});const requests=new Tracked([['r',{...request,outcome:'unresolved',error:'transport'}]]);f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests}}});assert.equal(calls,1);const counted=channel.getSnapshot();assert.deepEqual(counted,{pending:1,blocked:1,unknown:1,loading:false});const unrelated=new Tracked(requests);unrelated.set('other',{phase:'waiting',outcome:'unresolved',error:'other',operationIds:['missing']});f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{...f.get().runtime.ledger,requests:unrelated}}});assert.equal(calls,1);assert.strictEqual(sources.counts('a').getSnapshot(),counted);stop();
+});
+
+test('indexed counts preserve the previous filter definition over mixed request and operation states',()=>{
+ const f=fixture(0),sources=createRuntimeSources(f.source);let seed=7;const random=n=>{seed=(Math.imul(seed,1664525)+1013904223)>>>0;return seed%n;};
+ for(let sample=0;sample<40;sample++){
+  const operations=new Tracked(Array.from({length:24},(_,index)=>{const id='op'+index,canvases=Array.from({length:random(4)},()=>['a','b','c'][random(3)]);return[id,operation(id,canvases,['pending','blocked','frozen','resolved'][random(4)],random(2)?['waiting_digest']:['permission'])];})),requests=new Tracked(Array.from({length:12},(_,index)=>['r'+index,{phase:['prepared','dispatch-marked','waiting','resolved'][random(4)],outcome:[undefined,'unresolved','accepted','applied'][random(4)],error:random(2)?'transport':undefined,retryAt:random(2)?100:undefined,operationIds:Array.from({length:random(5)},()=>'op'+random(26))}]));f.emit({...f.get(),runtime:{...f.get().runtime,ledger:{operations,requests},loading:sample%2?{}:undefined}});
+  for(const scope of [undefined,'a','b','c','missing']){
+   const selected=[...operations.values()].filter(row=>row.phase!=='resolved'&&(scope===undefined||row.intent.canvasIds.includes(scope))),ids=new Set(selected.map(row=>row.intent.id)),visible=[...requests.values()].filter(row=>row.phase!=='resolved'&&row.operationIds.some(id=>ids.has(id))),expected={pending:selected.length,blocked:selected.filter(row=>row.phase==='blocked'&&row.blockers.some(reason=>reason!=='waiting_digest')).length+visible.filter(row=>row.error&&row.retryAt===undefined).length,unknown:visible.filter(row=>row.outcome==='unresolved'&&['dispatch-marked','waiting'].includes(row.phase)).length,loading:!!f.get().runtime.loading};assert.deepEqual(sources.counts(scope).getSnapshot(),expected,`sample ${sample}, scope ${scope}`);
+  }
+ }
+});