1	import Base
2	import ./world.bend as W
3	import ./system.bend as Sy
4	
5	# One event domain; World owns payloads, this owner owns batch and reader metadata.
6	# Ordered detached Data publications grouped at a runtime tick. This is not an
7	# affine payload owner or an acknowledgement from a reader.
8	type Batch<E: Data> is Data:
9	  Batch{tick: Nat,values: List<&2,E>}
10	# Activated reader metadata: last consumed cursor and activation tick. Merely
11	# allocating a Reader token does not add a Position or hold old batches.
12	type Position is Data:
13	  Position{id: U32,cursor: Nat,registeredAt: Nat}
14	# Actual affine token binding runtime namespace and reader id to one registry id.
15	# Raw construction is trusted setup, not proof of registration/provenance.
16	type Reader<S: Data> is Type:
17	  Reader{namespace: U32,id: U32,registryId: U32}
18	# Sole actual World plus Data batch/reader metadata and retention accounting.
19	# This lower-level runtime handles Data E; affine Type payload logs use separate
20	# owned-event APIs. Raw metadata construction and World callbacks remain trusted.
21	type Runtime<S: Data,C: Type,R: Type,E: Data> is Type:
22	  Runtime{world: W.World<S,C,R,E>,namespace: U32,batches: List<&2,Batch<E>>,positions: List<&2,Position>,nextReader: U32,tick: Nat,frameStart: Nat,boundary: Nat,capacity: Nat,droppedThrough: Nat}
23	# Unread Data values and lag flag supplied to a runner with actual possibly-
24	# affine Args. Callback consumption does not imply Args recovery on failure.
25	type Input<E: Data,A: Type> is Type:
26	  Input{values: List<&2,E>,lagged: Bool,args: A}
27	# Detached unread Data in publication order plus bounded-retention lag status.
28	# Reading alone does not advance a cursor or protect messages from trimming.
29	type Reading<E: Data> is Data:
30	  Reading{values: List<&2,E>,lagged: Bool}
31	type Trimmed<E: Data> is Data:
32	  Trimmed{batches: List<&2,Batch<E>>,dropped: Nat}
33	
34	def count_loop(~E: Data,items: List<&2,E>,+acc: Nat) -> Nat:
35	  match items:
36	    case []: acc
37	    case _ <> rest: count_loop(~E,rest,1n+acc)
38	
39	def count(~E: Data,items: List<&2,E>) -> Nat: count_loop(~E,items,0n)
40	
41	def append_loop(~E: Data,reversed: List<&2,E>,+output: List<&2,E>) -> List<&2,E>:
42	  match reversed:
43	    case []: output
44	    case head <> rest: append_loop(~E,rest,head <> output)
45	
46	def append(~E: Data,+left: List<&2,E>,+right: List<&2,E>) -> List<&2,E>:
47	  match left right:
48	    case Nil{} right: right
49	    case left Nil{}: left
50	    case left right: append_loop(~E,List.reverse(&2,E,left),right)
51	
52	def values(~E: Data,batches: List<&2,Batch<E>>) -> List<&2,E>:
53	  match batches:
54	    case []: []
55	    case Batch{_,items} <> rest: append(~E,items,values(~E,rest))
56	
57	def size_loop(~E: Data,batches: List<&2,Batch<E>>,+acc: Nat) -> Nat:
58	  match batches:
59	    case []: acc
60	    case Batch{_,items} <> rest: size_loop(~E,rest,count_loop(~E,items,acc))
61	
62	def size(~E: Data,batches: List<&2,Batch<E>>) -> Nat:
63	  size_loop(~E,batches,0n)
64	
65	def hold_boundary(positions: List<&2,Position>,+boundary: Nat) -> Nat:
66	  match positions:
67	    case []: boundary
68	    case Position{_,cursor,_} <> rest: hold_boundary(rest,Nat.min(boundary,cursor))
69	
70	def past_checked(~E: Data,remove: Bool,+tick: Nat,+items: List<&2,E>,+rest: List<&2,Batch<E>>,+dropped: Nat,ih: Nat -> Trimmed<E>) -> Trimmed<E>:
71	  match remove:
72	    case False{}: Trimmed{Batch{tick,items} <> rest,dropped}
73	    case True{}: ih(Nat.max(dropped,tick))
74	
75	def past_window(~E: Data,batches: List<&2,Batch<E>>,+boundary: Nat,+dropped: Nat) -> Trimmed<E>:
76	  match batches:
77	    case []: Trimmed{[],dropped}
78	    case Batch{+tick,+items} <> +rest: past_checked(~E,Nat.is_le(tick,boundary),tick,items,rest,dropped,gone => past_window(~E,rest,boundary,gone))
79	
80	def capacity_checked(~E: Data,overflow: Bool,+tick: Nat,+items: List<&2,E>,+rest: List<&2,Batch<E>>,+dropped: Nat,ih: Nat -> Trimmed<E>) -> Trimmed<E>:
81	  match overflow:
82	    case False{}: Trimmed{Batch{tick,items} <> rest,dropped}
83	    case True{}: ih(Nat.max(dropped,tick))
84	
85	def at_capacity(~E: Data,batches: List<&2,Batch<E>>,+capacity: Nat,+dropped: Nat) -> Trimmed<E>:
86	  match batches:
87	    case []: Trimmed{[],dropped}
88	    case Batch{+tick,+items} <> +rest: capacity_checked(~E,Nat.is_lt(capacity,size(~E,Batch{tick,items} <> rest)),tick,items,rest,dropped,gone => at_capacity(~E,rest,capacity,gone))
89	
90	def trim_capacity(~E: Data,trimmed: Trimmed<E>,+capacity: Nat) -> Trimmed<E>:
91	  match trimmed:
92	    case Trimmed{batches,dropped}: at_capacity(~E,batches,capacity,dropped)
93	
94	def trim_result(~S: Data,~C: Type,~R: Type,~E: Data,world: W.World<S,C,R,E>,+namespace: U32,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,trimmed: Trimmed<E>) -> Runtime<S,C,R,E>:
95	  match trimmed:
96	    case Trimmed{+batches,dropped}: Runtime{W.events_replace(~S,~C,~R,~E,world,values(~E,batches)),namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped}
97	
98	def trim(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Runtime<S,C,R,E>) -> Runtime<S,C,R,E>:
99	  match runtime:
100	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: trim_result(~S,~C,~R,~E,world,namespace,positions,nextReader,tick,frameStart,boundary,capacity,trim_capacity(~E,past_window(~E,batches,hold_boundary(positions,boundary),dropped),capacity))
101	
102	# Advance frame markers and apply trim, returning actual Runtime/World. This
103	# bounds normal visibility while preserving unread holds only within hard capacity.
104	# Does not run a system, apply pending structural commands or acknowledge readers.
105	def frame(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Runtime<S,C,R,E>) -> Runtime<S,C,R,E>:
106	  match runtime:
107	    case Runtime{world,namespace,batches,positions,nextReader,+tick,+frameStart,_,capacity,dropped}: trim(~S,~C,~R,~E,Runtime{world,namespace,batches,positions,nextReader,tick,tick,frameStart,capacity,dropped})
108	
109	def start_named(~S: Data,~C: Type,~R: Type,~E: Data,+capacity: Nat,named: W.World<S,C,R,E> & U32) -> Runtime<S,C,R,E>:
110	  (world,namespace) = named
111	  Runtime{world,namespace,[],[],1,0n,0n,0n,capacity,0n}
112	
113	# Start runtime metadata for the actual World namespace and supplied capacity.
114	# Initial batches/readers/tick are empty/zero; this does not consume World events
115	# or automatically synchronize an already populated log.
116	def start(~S: Data,~C: Type,~R: Type,~E: Data,world: W.World<S,C,R,E>,+capacity: Nat) -> Runtime<S,C,R,E>:
117	  start_named(~S,~C,~R,~E,capacity,W.namespace(~S,~C,~R,~E,world))
118	
119	
120	def update_checked(found: Bool,dispose: Bool,+id: U32,+old: Nat,+registered: Nat,+rest: List<&2,Position>,+cursor: Nat) -> List<&2,Position>:
121	  match found dispose:
122	    case False{} _: Position{id,old,registered} <> rest
123	    case True{} False{}: Position{id,cursor,registered} <> rest
124	    case True{} True{}: rest
125	
126	def update(positions: List<&2,Position>,+id: U32,+cursor: Nat,+dispose: Bool) -> List<&2,Position>:
127	  match positions:
128	    case []: []
129	    case Position{+actual,+old,+registered} <> rest: update_checked(U32.is_eq(actual,id),dispose,actual,old,registered,update(rest,id,cursor,dispose),cursor)
130	
131	
132	def choose_position(found: Bool,+position: Position,fallback: Maybe<Position>) -> Maybe<Position>:
133	  match found:
134	    case False{}: fallback
135	    case True{}: Some{position}
136	
137	def lookup_checked(position: Position,+id: U32,fallback: Maybe<Position>) -> Maybe<Position>:
138	  match position:
139	    case Position{+actual,+cursor,+registered}: choose_position(U32.is_eq(actual,id),Position{actual,cursor,registered},fallback)
140	
141	def lookup(positions: List<&2,Position>,+id: U32) -> Maybe<Position>:
142	  match positions:
143	    case []: None{}
144	    case +position <> rest: lookup_checked(position,id,lookup(rest,id))
145	
146	
147	def since_checked(~E: Data,visible: Bool,+items: List<&2,E>,+rest: List<&2,E>) -> List<&2,E>:
148	  match visible:
149	    case False{}: rest
150	    case True{}: append(~E,items,rest)
151	
152	def since(~E: Data,batches: List<&2,Batch<E>>,+cursor: Nat) -> List<&2,E>:
153	  match batches:
154	    case []: []
155	    case Batch{+tick,+items} <> rest: since_checked(~E,Nat.is_lt(cursor,tick),items,since(~E,rest,cursor))
156	
157	
158	def reading(~E: Data,position: Position,+batches: List<&2,Batch<E>>,+dropped: Nat) -> Reading<E>:
159	  match position:
160	    case Position{_,+cursor,+registered}: Reading{since(~E,batches,cursor),Nat.is_lt(Nat.max(cursor,registered),dropped)}
161	
162	def sync_checked(~E: Data,nonempty: Bool,+batches: List<&2,Batch<E>>,+tick: Nat,+published: List<&2,E>) -> List<&2,Batch<E>>:
163	  match nonempty:
164	    case False{}: batches
165	    case True{}: append(~Batch<E>,batches,[Batch{tick,published}])
166	
167	def sync_values(~S: Data,~C: Type,~R: Type,~E: Data,observed: W.World<S,C,R,E> & List<&2,E>,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Runtime<S,C,R,E>:
168	  (world,+all) = observed
169	  +published = List.drop(&2,E,all,size(~E,batches))
170	  Runtime{world,namespace,sync_checked(~E,Nat.is_lt(0n,count(~E,published)),batches,tick,published),positions,nextReader,tick,frameStart,boundary,capacity,dropped}
171	
172	def with_world_result(~S: Data,~C: Type,~R: Type,~E: Data,~O: Type,returned: W.World<S,C,R,E> & O,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Runtime<S,C,R,E> & O:
173	  (world,output) = returned
174	  (sync_values(~S,~C,~R,~E,W.events_view(~S,~C,~R,~E,world),namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped),output)
175	
176	# Advanced action bridge: consume/return actual World and Out, advance runtime
177	# tick, then synchronize newly appended World events by the existing retained-log
178	# prefix length. Action must respect that synchronization invariant; this is not
179	# a read-only guarantee, transaction wrapper or implicit structural barrier.
180	def with_world(~S: Data,~C: Type,~R: Type,~E: Data,~O: Type,effect: W.World<S,C,R,E> -> W.World<S,C,R,E> & O,runtime: Runtime<S,C,R,E>) -> Runtime<S,C,R,E> & O:
181	  match runtime:
182	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: with_world_result(~S,~C,~R,~E,~O,effect(world),namespace,batches,positions,nextReader,1n+tick,frameStart,boundary,capacity,dropped)
183	
184	# Reader registration result. Success returns actual Runtime, Reader and registry;
185	# failure returns Runtime without dropping World. No callback executes here.
186	type Registered<S: Data,C: Type,R: Type,E: Data,A: Type,Error: Data,O: Type,-runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>> is Type:
187	  Registered{runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>}
188	  RegisterFailed{runtime: Runtime<S,C,R,E>}
189	# Succeeded returns actual Out; Failed carries callback Error without consumed
190	# Args; Rejected returns unconsumed Args. Higher-level semantic errors encoded
191	# inside Out must be inspected separately from transport success.
192	type Completion<Error: Data,O: Type,A: Type> is Type:
193	  Succeeded{output: O}
194	  Failed{error: Error}
195	  Rejected{args: A}
196	# Every reader run returns actual Runtime, Reader and registry with Completion.
197	# No duplicate World or runner owner is retained in detached observations.
198	type RunResult<S: Data,C: Type,R: Type,E: Data,A: Type,Error: Data,O: Type,-runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>> is Type:
199	  Ran{runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,completion: Completion<Error,O,A>}
200	
201	def registered_identity(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,world: W.World<S,C,R,E>,identity: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner> & U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Registered<S,C,R,E,A,Error,O,runner>:
202	  (registry,+registryId) = identity
203	  Registered{Runtime{world,namespace,batches,positions,(nextReader+1 : U32),tick,frameStart,boundary,capacity,dropped},Reader{namespace,nextReader,registryId},registry}
204	
205	def registered_result(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,result: Sy.RegistrationResult<S,C,R,E,Input<E,A>,Error,O,runner>,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Registered<S,C,R,E,A,Error,O,runner>:
206	  match result:
207	    case Sy.RegistrationFailed{world}: RegisterFailed{Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped}}
208	    case Sy.Registered{world,registry}: registered_identity(~S,~C,~R,~E,~A,~Error,~O,~runner,world,Sy.identity(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,registry),namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
209	
210	# Register a runner with supplied name/access on the actual World, allocate its
211	# reader token on success and return Runtime on refusal. This does not run the
212	# body or activate a reader position; unused registrations do not retain backlog.
213	# Name/access and runner coherence are trusted lower-level provisioning.
214	def register(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,+name: String,+access: List<&2,String>) -> Registered<S,C,R,E,A,Error,O,runner>:
215	  match runtime:
216	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: registered_result(~S,~C,~R,~E,~A,~Error,~O,~runner,Sy.register(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,world,name,access),namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
217	
218	def activation(position: Maybe<Position>,+positions: List<&2,Position>,+id: U32,+registered: Nat) -> List<&2,Position>:
219	  match position:
220	    case None{}: Position{id,0n,registered} <> positions
221	    case Some{_}: positions
222	
223	def activate(+positions: List<&2,Position>,+id: U32,+registered: Nat) -> List<&2,Position>:
224	  activation(lookup(positions,id),positions,id,registered)
225	
226	def run_outcome(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,outcome: Sy.Outcome<W.World<S,C,R,E>,Error,O>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,+readerId: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> RunResult<S,C,R,E,A,Error,O,runner>:
227	  match outcome:
228	    case Sy.Failed{world,error}: Ran{Runtime{world,namespace,batches,activate(positions,readerId,Nat.sub(tick,1n)),nextReader,tick,frameStart,boundary,capacity,dropped},reader,registry,Failed{error}}
229	    case Sy.Succeeded{world,output}: Ran{sync_values(~S,~C,~R,~E,W.events_view(~S,~C,~R,~E,world),namespace,batches,update(activate(positions,readerId,Nat.sub(tick,1n)),readerId,tick,False{}),nextReader,tick,frameStart,boundary,capacity,dropped),reader,registry,Succeeded{output}}
230	
231	def reject_input(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,input: Input<E,A>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>) -> RunResult<S,C,R,E,A,Error,O,runner>:
232	  match input:
233	    case Input{_,_,args}: Ran{runtime,reader,registry,Rejected{args}}
234	
235	def run_registered_result(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,result: Sy.RunResult<S,C,R,E,Input<E,A>,Error,O,runner>,reader: Reader<S>,+readerId: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> RunResult<S,C,R,E,A,Error,O,runner>:
236	  match result:
237	    case Sy.RegistrationRejected{registry,world,args}: reject_input(~S,~C,~R,~E,~A,~Error,~O,~runner,args,Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},reader,registry)
238	    case Sy.Completed{registry,outcome}: run_outcome(~S,~C,~R,~E,~A,~Error,~O,~runner,outcome,reader,registry,readerId,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
239	
240	def run_reading(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,reading: Reading<E>,world: W.World<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,args: A,+readerId: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> RunResult<S,C,R,E,A,Error,O,runner>:
241	  match reading:
242	    case Reading{items,lagged}: run_registered_result(~S,~C,~R,~E,~A,~Error,~O,~runner,Sy.run(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,registry,world,Input{items,lagged,args}),reader,readerId,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
243	
244	def run_position(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,position: Maybe<Position>,world: W.World<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,args: A,+readerId: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> RunResult<S,C,R,E,A,Error,O,runner>:
245	  match position:
246	    case None{}: run_reading(~S,~C,~R,~E,~A,~Error,~O,~runner,reading(~E,Position{readerId,0n,tick},batches,dropped),world,reader,registry,args,readerId,namespace,batches,positions,nextReader,1n+tick,frameStart,boundary,capacity,dropped)
247	    case Some{position}: run_reading(~S,~C,~R,~E,~A,~Error,~O,~runner,reading(~E,position,batches,dropped),world,reader,registry,args,readerId,namespace,batches,positions,nextReader,1n+tick,frameStart,boundary,capacity,dropped)
248	
249	def run_checked(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,allowed: Bool,world: W.World<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,args: A,+readerId: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> RunResult<S,C,R,E,A,Error,O,runner>:
250	  match allowed:
251	    case False{}: Ran{Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},reader,registry,Rejected{args}}
252	    case True{}: run_position(~S,~C,~R,~E,~A,~Error,~O,~runner,lookup(positions,readerId),world,reader,registry,args,readerId,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
253	
254	def run_identity(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,actual: U32,args: A) -> RunResult<S,C,R,E,A,Error,O,runner>:
255	  match runtime reader:
256	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped} Reader{+readerNamespace,+readerId,+expected}: run_checked(~S,~C,~R,~E,~A,~Error,~O,~runner,Bool.and(U32.is_eq(namespace,readerNamespace),U32.is_eq(expected,actual)),world,Reader{readerNamespace,readerId,expected},registry,args,readerId,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
257	
258	def run_identified(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,identified: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner> & U32,args: A) -> RunResult<S,C,R,E,A,Error,O,runner>:
259	  (registry,actual) = identified
260	  run_identity(~S,~C,~R,~E,~A,~Error,~O,~runner,runtime,reader,registry,actual,args)
261	
262	# Validate token namespace/registry identity, then invoke the actual registered
263	# runner with ordered unread Data, lag flag and Args. Admission rejection returns
264	# Args without body execution; registry checks remain authoritative in Sy.run.
265	# Success advances reader cursor and synchronizes returned publications. Callback
266	# failure preserves unread position, activating an absent reader under the failed-
267	# reader rule; returned World is whatever the callback returned, not automatically
268	# rolled back by this runtime. Actual transaction behavior belongs to the runner.
269	# Runtime ticks and reader cursors are distinct from system lifecycle cursors.
270	def run(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,args: A) -> RunResult<S,C,R,E,A,Error,O,runner>:
271	  run_identified(~S,~C,~R,~E,~A,~Error,~O,~runner,runtime,reader,Sy.identity(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,registry),args)
272	
273	# A condition skip discards stream backlog without invoking gameplay.
274	def skip_checked(~S: Data,~C: Type,~R: Type,~E: Data,allowed: Bool,runtime: Runtime<S,C,R,E>,reader: Reader<S>,+id: U32) -> Runtime<S,C,R,E> & Reader<S>:
275	  match allowed runtime:
276	    case False{} runtime: (runtime,reader)
277	    case True{} Runtime{world,namespace,batches,positions,nextReader,+tick,frameStart,boundary,capacity,dropped}: (Runtime{world,namespace,batches,update(positions,id,tick,False{}),nextReader,tick,frameStart,boundary,capacity,dropped},reader)
278	
279	# Condition-false lifecycle operation: for a matching namespace, advance an
280	# existing position to current tick without invoking gameplay. Does not activate
281	# an absent reader; foreign namespace leaves Runtime/Reader unchanged. No trim
282	# or registry disposal is performed.
283	def skip(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Runtime<S,C,R,E>,reader: Reader<S>) -> Runtime<S,C,R,E> & Reader<S>:
284	  match runtime reader:
285	    case Runtime{world,+namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped} Reader{+readerNamespace,+id,+expected}: skip_checked(~S,~C,~R,~E,U32.is_eq(namespace,readerNamespace),Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},Reader{readerNamespace,id,expected},id)
286	
287	# Detached active-reader id/cursor/activation tick with unread count and lag flag.
288	# Registry ids are separate; statuses do not grant reading or mutation authority.
289	type ReaderStatus is Data:
290	  ReaderStatus{id: U32,cursor: Nat,registeredAt: Nat,unread: Nat,lagged: Bool}
291	# Detached visible batch values, retained World log and activated-reader statuses
292	# with tick/boundary. This is observation, not a restorable runtime checkpoint.
293	type Snapshot<E: Data> is Data:
294	  Snapshot{visible: List<&2,E>,retained: List<&2,E>,readers: List<&2,ReaderStatus>,tick: Nat,boundary: Nat}
295	
296	def reader_status(~E: Data,position: Position,+batches: List<&2,Batch<E>>,+dropped: Nat) -> ReaderStatus:
297	  match position:
298	    case Position{+id,+cursor,+registered}: ReaderStatus{id,cursor,registered,count(~E,since(~E,batches,cursor)),Nat.is_lt(Nat.max(cursor,registered),dropped)}
299	
300	def reader_statuses(~E: Data,positions: List<&2,Position>,+batches: List<&2,Batch<E>>,+dropped: Nat) -> List<&2,ReaderStatus>:
301	  match positions:
302	    case []: []
303	    case position <> rest: reader_status(~E,position,batches,dropped) <> reader_statuses(~E,rest,batches,dropped)
304	
305	def inspect_values(~S: Data,~C: Type,~R: Type,~E: Data,observed: W.World<S,C,R,E> & List<&2,E>,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Runtime<S,C,R,E> & Snapshot<E>:
306	  (world,retained) = observed
307	  (Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},Snapshot{since(~E,batches,boundary),retained,reader_statuses(~E,List.reverse(&2,Position,positions),batches,dropped),tick,boundary})
308	
309	# Return Runtime and detached Snapshot without synchronization, trimming, frame
310	# advance or reader acknowledgement. visible derives from batches after boundary;
311	# retained is the actual World event log, so unsynchronized publications may differ.
312	# Reader statuses follow reversed position order, not an inferred registry catalog.
313	def inspect(~S: Data,~C: Type,~R: Type,~E: Data,runtime: Runtime<S,C,R,E>) -> Runtime<S,C,R,E> & Snapshot<E>:
314	  match runtime:
315	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped}: inspect_values(~S,~C,~R,~E,W.events_view(~S,~C,~R,~E,world),namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
316	
317	# Success returns Runtime after registration/position removal. Rejection returns
318	# Runtime plus actual Reader and registry; no payload destructor is invented.
319	type Disposal<S: Data,C: Type,R: Type,E: Data,A: Type,Error: Data,O: Type,-runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>> is Type:
320	  Disposed{runtime: Runtime<S,C,R,E>}
321	  DisposalRejected{runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>}
322	
323	def disposal_result(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,result: Sy.Disposal<S,C,R,E,Input<E,A>,Error,O,runner>,reader: Reader<S>,+id: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Disposal<S,C,R,E,A,Error,O,runner>:
324	  match result:
325	    case Sy.Disposed{world}: Disposed{Runtime{world,namespace,batches,update(positions,id,tick,True{}),nextReader,tick,frameStart,boundary,capacity,dropped}}
326	    case Sy.DisposalRejected{world,registry}: DisposalRejected{Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},reader,registry}
327	
328	def disposal_checked(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,allowed: Bool,world: W.World<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,+id: U32,+namespace: U32,+batches: List<&2,Batch<E>>,+positions: List<&2,Position>,+nextReader: U32,+tick: Nat,+frameStart: Nat,+boundary: Nat,+capacity: Nat,+dropped: Nat) -> Disposal<S,C,R,E,A,Error,O,runner>:
329	  match allowed:
330	    case False{}: DisposalRejected{Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},reader,registry}
331	    case True{}: disposal_result(~S,~C,~R,~E,~A,~Error,~O,~runner,Sy.dispose(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,registry,world),reader,id,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
332	
333	def disposal_identity(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>,actual: U32) -> Disposal<S,C,R,E,A,Error,O,runner>:
334	  match runtime reader:
335	    case Runtime{world,+namespace,+batches,+positions,+nextReader,+tick,+frameStart,+boundary,+capacity,+dropped} Reader{+readerNamespace,+id,+expected}: disposal_checked(~S,~C,~R,~E,~A,~Error,~O,~runner,Bool.and(U32.is_eq(namespace,readerNamespace),U32.is_eq(actual,expected)),world,Reader{readerNamespace,id,expected},registry,id,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped)
336	
337	def disposal_identified(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,identified: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner> & U32) -> Disposal<S,C,R,E,A,Error,O,runner>:
338	  (registry,actual) = identified
339	  disposal_identity(~S,~C,~R,~E,~A,~Error,~O,~runner,runtime,reader,registry,actual)
340	
341	# Validate token/registry identity and dispose through Sy.dispose, removing the
342	# reader position only after success. Return actual owners on refusal. Does not
343	# immediately trim retained batches or dispose unrelated readers/log domains.
344	def dispose(~S: Data,~C: Type,~R: Type,~E: Data,~A: Type,~Error: Data,~O: Type,~runner: W.World<S,C,R,E> -> Input<E,A> -> Sy.Outcome<W.World<S,C,R,E>,Error,O>,runtime: Runtime<S,C,R,E>,reader: Reader<S>,registry: Sy.Registry<S,C,R,E,Input<E,A>,Error,O,runner>) -> Disposal<S,C,R,E,A,Error,O,runner>:
345	  disposal_identified(~S,~C,~R,~E,~A,~Error,~O,~runner,runtime,reader,Sy.identity(~S,~C,~R,~E,~Input<E,A>,~Error,~O,~runner,registry))
346	
347	# Gameplay receives only a closed capability over an abstract affine owner.
348	# Closed read capability over opaque H returning detached Reading<E>. The bound
349	# operation defines lifecycle behavior; raw adapters remain trusted setup.
350	type ReadAccess<H: Type,E: Data,-observe: H -> H & Reading<E>> is Data:
351	  ReadAccess{}
352	# Closed Data-publication capability threading opaque H. This type alone does
353	# not promise transaction staging or rollback of arbitrary host side effects.
354	type WriteAccess<H: Type,E: Data,-publish: H -> E -> H> is Data:
355	  WriteAccess{}
356	
357	# Invoke the declared closed observation, returning actual H plus Reading.
358	# Adds no validation or acknowledgement beyond the supplied operation.
359	def all(~H: Type,~E: Data,~observe: H -> H & Reading<E>,_: ReadAccess<H,E,observe>,owner: H) -> H & Reading<E>:
360	  observe(owner)
361	
362	# Invoke the declared publication operation with Data event and return H.
363	# Publication timing/transaction behavior belongs to that bound provider.
364	def emit(~H: Type,~E: Data,~publish: H -> E -> H,_: WriteAccess<H,E,publish>,owner: H,event: E) -> H:
365	  publish(owner,event)
366	
367	# Project copied unread values and lag from Input while retaining actual Args
368	# inside the returned Input owner. Does not rerun or advance the reader.
369	def input_read(~E: Data,~A: Type,input: Input<E,A>) -> Input<E,A> & Reading<E>:
370	  match input:
371	    case Input{+items,+lagged,args}: (Input{items,lagged,args},Reading{items,lagged})