1	import Base
2	import ./world.bend as W
3	import ./event_runtime.bend as Ev
4	import ./ordinary_event_family.bend as Composition
5	import ./event_family_routing.bend as Route
6	import ./event_stream.bend as Leaf
7	
8	# Advanced affine family state containing actual typed leaves F and owners H,
9	# with a shared Data publication journal and frame marker. It owns no second World.
10	# Canonical provisioning couples routes; raw callbacks remain trusted setup.
11	type State<P:Data,F:Type,H:Type> is Type:
12	  State{family:F,owners:H,journal:List<&2,Ev.Batch<P>>,frameStart:Nat}
13	# Create typed leaves from the canonical Provision domains/capacity, retaining
14	# actual World and supplied H owners with empty journal/tick zero. Does not register
15	# readers or automatically synchronize an existing World publication log.
16	def setup(~S:Data,~C:Type,~R:Type,~P:Data,~D:Data,~F:Type,~H:Type,~make:D -> Nat -> F,~sync:F -> Nat -> List<&2,P> -> F,~frame:F -> Nat -> F,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,~inventory:F -> F & List<&2,String>,world:W.World<S,C,R,P>,provision:Composition.Provision<S,P,D,F,make,sync,frame,retain,inventory>,owners:H,capacity:Nat) -> Route.Runtime<S,C,R,P,State<P,F,H>>:
17	  match provision:
18	    case Composition.Provision{domains}:Route.Runtime{world,State{make(domains,capacity),owners,[],0n},0n}
19	def synchronize_world(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,state:State<P,F,H>,+tick:Nat,observed:W.World<S,C,R,P> & List<&2,P>) -> W.World<S,C,R,P> & State<P,F,H>:
20	  match state:
21	    case State{family,owners,+journal,frameStart}:
22	      (world,records)=observed
23	      +published=List.drop(&2,P,records,Ev.size(~P,journal))
24	      (world,State{sync(family,tick,published),owners,Ev.sync_checked(~P,Nat.is_lt(0n,Ev.count(~P,published)),journal,tick,published),frameStart})
25	# Advanced publication synchronization: thread actual World/state and distribute
26	# World log suffix beyond journal length through the generated sync callback.
27	# Append nonempty suffix batches at the supplied tick. Caller must preserve the
28	# log/journal prefix invariant; this does not execute a barrier or advance readers.
29	def synchronize(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,world:W.World<S,C,R,P>,state:State<P,F,H>,tick:Nat) -> W.World<S,C,R,P> & State<P,F,H>:
30	  synchronize_world(~S,~C,~R,~P,~F,~H,~sync,state,tick,W.events_view(~S,~C,~R,~P,world))
31	def kept_batch(~P:Data,keep:Bool,tick:Nat,records:List<&2,P>,journal:List<&2,Ev.Batch<P>>) -> List<&2,Ev.Batch<P>>:
32	  match keep:
33	    case False{}:journal
34	    case True{}:Ev.Batch{tick,records} <> journal
35	def kept_batch_returned(~P:Data,~F:Type,tick:Nat,+records:List<&2,P>,returned:F & List<&2,Ev.Batch<P>>) -> F & List<&2,Ev.Batch<P>>:
36	  match returned:
37	    case (family,journal):(family,kept_batch(~P,Nat.is_lt(0n,Ev.count(~P,records)),tick,records,journal))
38	def kept_batch_head(~P:Data,~F:Type,rest:F -> F & List<&2,Ev.Batch<P>>,tick:Nat,returned:F & List<&2,P>) -> F & List<&2,Ev.Batch<P>>:
39	  match returned:
40	    case (family,records):kept_batch_returned(~P,~F,tick,records,rest(family))
41	def retained_journal(~P:Data,~F:Type,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,journal:List<&2,Ev.Batch<P>>,family:F) -> F & List<&2,Ev.Batch<P>>:
42	  match journal:
43	    case []:(family,[])
44	    case Ev.Batch{+tick,records} <> rest:kept_batch_head(~P,~F,(family => retained_journal(~P,~F,~retain,rest,family)),tick,retain(family,tick,records))
45	def framed_journal(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,world:W.World<S,C,R,P>,owners:H,+tick:Nat,returned:F & List<&2,Ev.Batch<P>>) -> Route.Runtime<S,C,R,P,State<P,F,H>>:
46	  match returned:
47	    case (family,+journal):Route.Runtime{W.events_replace(~S,~C,~R,~P,world,Ev.values(~P,journal)),State{family,owners,journal,tick},tick}
48	# Apply each generated family frame/retention operation at current routing tick,
49	# retain only still-represented journal records, and replace World event log with
50	# those retained values. Return actual family/owners/World; no gameplay dispatch
51	# or implicit affine payload-log retirement is added to this Data layer.
52	def frame(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~frame:F -> Nat -> F,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,runtime:Route.Runtime<S,C,R,P,State<P,F,H>>) -> Route.Runtime<S,C,R,P,State<P,F,H>>:
53	  match runtime:
54	    case Route.Runtime{world,State{family,owners,journal,_},+tick}:framed_journal(~S,~C,~R,~P,~F,~H,world,owners,tick,retained_journal(~P,~F,~retain,journal,frame(family,tick)))
55	def inventory_returned(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,world:W.World<S,C,R,P>,owners:H,journal:List<&2,Ev.Batch<P>>,frameStart:Nat,tick:Nat,returned:F & List<&2,String>) -> Route.Runtime<S,C,R,P,State<P,F,H>> & List<&2,String>:
56	  match returned:
57	    case (family,names):(Route.Runtime{world,State{family,owners,journal,frameStart},tick},names)
58	def inventory_optional(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,returned:Route.Runtime<S,C,R,P,State<P,F,H>> & List<&2,String>) -> Route.Runtime<S,C,R,P,State<P,F,H>> & Maybe<&2,List<&2,String>>:
59	  match returned:
60	    case (runtime,names):(runtime,Some{names})
61	# Return actual Runtime and optional detached family inventory. Disabled bypasses
62	# the inventory callback; enabled threads actual F through it. No synchronization,
63	# trim, frame or reader acknowledgement is performed by this observation.
64	def inventory_enabled(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~inventory:F -> F & List<&2,String>,enabled:Bool,runtime:Route.Runtime<S,C,R,P,State<P,F,H>>) -> Route.Runtime<S,C,R,P,State<P,F,H>> & Maybe<&2,List<&2,String>>:
65	  match enabled runtime:
66	    case False{} runtime:(runtime,None{})
67	    case True{} Route.Runtime{world,State{family,owners,journal,frameStart},tick}:inventory_optional(~S,~C,~R,~P,~F,~H,inventory_returned(~S,~C,~R,~P,~F,~H,world,owners,journal,frameStart,tick,inventory(family)))
68	
69	def selected_state(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,owners:H,journal:List<&2,Ev.Batch<P>>,frameStart:Nat,returned:F & Leaf.Stream<S,E>) -> State<P,F,H> & Leaf.Stream<S,E>:
70	  match returned:
71	    case (family,stream):(State{family,owners,journal,frameStart},stream)
72	# Trusted selector adapter extracting a copied Data stream while retaining actual
73	# family/owners/journal. Selector coherence comes from canonical provision, not
74	# a universal ownership or route-validity check in this wrapper.
75	def take(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,~take:F -> F & Leaf.Stream<S,E>,state:State<P,F,H>) -> State<P,F,H> & Leaf.Stream<S,E>:
76	  match state:
77	    case State{family,owners,journal,frameStart}:selected_state(~S,~E,~P,~F,~H,owners,journal,frameStart,take(family))
78	# Trusted selector adapter reinserting a Data stream into actual family state,
79	# preserving owners/journal/frame marker. Does not validate arbitrary stream origin
80	# or manufacture a reader capability.
81	def put(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,~put:F -> Leaf.Stream<S,E> -> F,state:State<P,F,H>,stream:Leaf.Stream<S,E>) -> State<P,F,H>:
82	  match state:
83	    case State{family,owners,journal,frameStart}:State{put(family,stream),owners,journal,frameStart}