1 import Base 2 import ./world.bend as W 3 import ./world_resource_owner.bend as Resource 4 import ./event_runtime.bend as Ev 5 import ./capabilities.bend as Cap 6 import ./owned_event_log.bend as Log 7 8 # Trusted schema storage: user resources and each affine payload log have one owner. 9 # Data IDs use the existing event runtime; its cursor/lag/retention policy is unchanged. 10 type Resources is Type: 11 Resources{resource:R,log:Log.Log

} 12 def World(~S:Data,~C:Type,~R:Type,~P:Type) -> Type:W.World,Nat> 13 def Runtime(~S:Data,~C:Type,~R:Type,~P:Type) -> Type:Ev.Runtime,Nat> 14 15 # Canonical World resource splitting preserves every non-resource field. 16 def resources_returned(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,rest:Resource.Rest,Nat>,returned:Resources & Out) -> World(~S,~C,~R,~P) & Out: 17 match returned: 18 case (resources,output):(Resource.put(~S,~C,~Resources,~Nat,rest,resources),output) 19 def resources_taken(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,effect:Resources -> Resources & Out,taken:Resource.Rest,Nat> & Resources) -> World(~S,~C,~R,~P) & Out: 20 match taken: 21 case (rest,resources):resources_returned(~S,~C,~R,~P,~Out,rest,effect(resources)) 22 def with_resources(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,effect:Resources -> Resources & Out,world:World(~S,~C,~R,~P)) -> World(~S,~C,~R,~P) & Out: 23 resources_taken(~S,~C,~R,~P,~Out,effect,Resource.take(~S,~C,~Resources,~Nat,world)) 24 25 def appended(~R:Type,~P:Type,resource:R,result:Log.Log

& List<&2,Nat>) -> Resources & List<&2,Nat>: 26 match result: 27 case (log,ids):(Resources{resource,log},ids) 28 def append_resources(~R:Type,~P:Type,values:List

,resources:Resources) -> Resources & List<&2,Nat>: 29 match resources: 30 case Resources{resource,log}:appended(~R,~P,resource,Log.append_many(~P,log,values)) 31 def publish_ids(~S:Data,~C:Type,~R:Type,~P:Type,result:World(~S,~C,~R,~P) & List<&2,Nat>) -> World(~S,~C,~R,~P): 32 match result: 33 case (world,ids):W.event_publish(~S,~C,~Resources,~Nat,world,ids) 34 # Called only after successful canonical transaction finish: owners and IDs publish 35 # in the same FIFO order. Failure/abort transport remains the ordinary adapter's job. 36 def commit_owned(~S:Data,~C:Type,~R:Type,~P:Type,world:World(~S,~C,~R,~P),values:List

) -> World(~S,~C,~R,~P): 37 publish_ids(~S,~C,~R,~P,with_resources(~S,~C,~R,~P,~List<&2,Nat>,resources => append_resources(~R,~P,values,resources),world)) 38 39 def retained(~R:Type,~P:Type,resource:R,result:Log.Log

& List>) -> Resources & List>: 40 match result: 41 case (log,retired):(Resources{resource,log},retired) 42 def retain_resources(~R:Type,~P:Type,ids:List<&2,Nat>,resources:Resources) -> Resources & List>: 43 match resources: 44 case Resources{resource,log}:retained(~R,~P,resource,Log.retain(~P,log,ids)) 45 def drain_resources(~R:Type,~P:Type,resources:Resources) -> Resources & List>: 46 match resources: 47 case Resources{resource,log}:retained(~R,~P,resource,Log.drain(~P,log)) 48 def runtime_returned(~S:Data,~C:Type,~R:Type,~P:Type,~Out:Type,namespace:U32,batches:List<&2,Ev.Batch>,positions:List<&2,Ev.Position>,nextReader:U32,tick:Nat,frameStart:Nat,boundary:Nat,capacity:Nat,dropped:Nat,result:World(~S,~C,~R,~P) & Out) -> Runtime(~S,~C,~R,~P) & Out: 49 match result: 50 case (world,output):(Ev.Runtime{world,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped},output) 51 def framed(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List>: 52 match runtime: 53 case Ev.Runtime{world,namespace,+batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped}: 54 runtime_returned(~S,~C,~R,~P,~List>,namespace,batches,positions,nextReader,tick,frameStart,boundary,capacity,dropped,with_resources(~S,~C,~R,~P,~List>,resources => retain_resources(~R,~P,Ev.values(~Nat,batches),resources),world)) 55 # Existing Ev.frame chooses surviving complete batches. Return every other payload 56 # owner once; the caller chooses reuse or release, with no automatic IO finalizer. 57 def frame(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List>: 58 framed(~S,~C,~R,~P,Ev.frame(~S,~C,~Resources,~Nat,runtime)) 59 def cleared_events(~S:Data,~C:Type,~R:Type,~P:Type,result:World(~S,~C,~R,~P) & List>) -> World(~S,~C,~R,~P) & List>: 60 match result: 61 case (world,rows):(W.events_replace(~S,~C,~Resources,~Nat,world,[]),rows) 62 # Shutdown clears only queued IDs/batches and log rows. Preserve reader/runtime 63 # metadata and Log.next so old IDs cannot alias a later appended payload. 64 def drain(~S:Data,~C:Type,~R:Type,~P:Type,runtime:Runtime(~S,~C,~R,~P)) -> Runtime(~S,~C,~R,~P) & List>: 65 match runtime: 66 case Ev.Runtime{world,namespace,_,positions,nextReader,tick,frameStart,boundary,capacity,dropped}: 67 runtime_returned(~S,~C,~R,~P,~List>,namespace,[],positions,nextReader,tick,frameStart,boundary,capacity,dropped,cleared_events(~S,~C,~R,~P,with_resources(~S,~C,~R,~P,~List>,drain_resources(~R,~P),world))) 68 69 def read_returned(~R:Type,~P:Type,~Args:Type,~Observation:Type,resource:R,result:Log.Log

& Log.ReadResult) -> Resources & Log.ReadResult: 70 match result: 71 case (log,reading):(Resources{resource,log},reading) 72 def read_resources(~R:Type,~P:Type,~Request:Type,~Projection:Data,~Args:Type,~Observation:Type,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request -> H -> Args -> H & Observation,resources:Resources,id:Nat,args:Args) -> Resources & Log.ReadResult: 73 match resources: 74 case Resources{resource,log}:read_returned(~R,~P,~Args,~Observation,resource,Log.read(~P,~Request,~Projection,~Args,~Observation,~at,~body,log,id,args)) 75 # The projection is trusted schema provisioning. The closed body receives only an 76 # abstract payload handle and declared scoped read request; it never gets World. 77 def read(~S:Data,~C:Type,~R:Type,~P:Type,~Request:Type,~Projection:Data,~Args:Type,~Observation:Type,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request -> H -> Args -> H & Observation,world:World(~S,~C,~R,~P),id:Nat,args:Args) -> World(~S,~C,~R,~P) & Log.ReadResult: 78 with_resources(~S,~C,~R,~P,~Log.ReadResult,resources => read_resources(~R,~P,~Request,~Projection,~Args,~Observation,~at,~body,resources,id,args),world)