event_runtime.bend
bendvy-ecs-core@0.1.0.0 · 371 lines · 0 definitions
1import Base
2import bendvy-ecs-core@0.1.0.0/world.bend as W
3import bendvy-ecs-core@0.1.0.0/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.
8type 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.
12type 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.
16type 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.
21type 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.
25type 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.
29type Reading<E: Data> is Data:
30 Reading{values: List<&2,E>,lagged: Bool}
31type Trimmed<E: Data> is Data:
32 Trimmed{batches: List<&2,Batch<E>>,dropped: Nat}
33
34def 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
39def count(~E: Data,items: List<&2,E>) -> Nat: count_loop(~E,items,0n)
40
41def 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
46def 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
52def 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
57def 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
62def size(~E: Data,batches: List<&2,Batch<E>>) -> Nat:
63 size_loop(~E,batches,0n)
64
65def 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
70def 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
75def 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
80def 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
85def 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
90def 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
94def 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
98def 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.
105def 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
109def 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.
116def 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
120def 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
126def 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
132def choose_position(found: Bool,+position: Position,fallback: Maybe<Position>) -> Maybe<Position>:
133 match found:
134 case False{}: fallback
135 case True{}: Some{position}
136
137def 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
141def 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
147def 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
152def 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
158def 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
162def 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
167def 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
172def 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.
180def 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.
186type 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.
192type 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.
198type 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
201def 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
205def 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.
214def 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
218def 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
223def activate(+positions: List<&2,Position>,+id: U32,+registered: Nat) -> List<&2,Position>:
224 activation(lookup(positions,id),positions,id,registered)
225
226def 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
231def 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
235def 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
240def 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
244def 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
249def 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
254def 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
258def 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.
270def 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.
274def 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.
283def 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.
289type 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.
293type Snapshot<E: Data> is Data:
294 Snapshot{visible: List<&2,E>,retained: List<&2,E>,readers: List<&2,ReaderStatus>,tick: Nat,boundary: Nat}
295
296def 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
300def 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
305def 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.
313def 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.
319type 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
323def 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
328def 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
333def 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
337def 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.
344def 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.
350type 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.
354type 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.
359def 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.
364def 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.
369def 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})