1	import Base
2	import ./world.bend as W
3	import ./system.bend as Sys
4	import ./event_runtime.bend as Ev
5	import ./event_stream.bend as Stream
6	import ./event_family_routing.bend as Route
7	import ./ordinary_event_declaration.bend as Event
8	import ./ordinary_app_event.bend as OrdinaryEvent
9	import ./ordinary_app_registration_fields.bend as Records
10	import ./ordinary_query_declaration.bend as Query
11	import ./ordinary_app_observation_data.bend as DTO
12	import ./ordinary_app_composer.bend as Build
13	import ./ordinary_app_schema.bend as SchemaApp
14	import ./ordinary_app_event_family_runtime.bend as App
15	import ./ordinary_app_records.bend as Leaf
16	import ./ordinary_app_recipe.bend as Tree
17	import ./ordinary_schema.bend as Schema
18	# One retained operational declaration and the actual returned reader/Registry.
19	# Debug folds only registration fields and never reads or advances the Reader.
20	# Registered App reader for one typed Data event stream E in family F.
21	# Retain the actual declaration, reader and Registry returned after each run;
22	# registration metadata alone cannot replace their runtime authority. Different
23	# readers of the same stream keep independent positions. F may also contain
24	# affine owners, but this reader's event payload E is Data, not an owned Type log.
25	type Owner<S:Data,E:Data,Reader:Type,Registry:Type> is Type:
26	  Owner{declaration:Event.Declaration<S,E>,reader:Reader,registry:Registry}
27	# Advanced App metadata projection: return this owner with its registration
28	# fields and event-read clause. The reported Registry cursor is not a read of
29	# the event stream's position. Debug collection does not deliver, advance or
30	# trim events; the actual reader and Registry are returned inside the owner.
31	def fields(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~Reader:Type,~A:Type,~Error:Data,~Out:Type,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,owner:Owner<S,E,Reader,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>) -> Owner<S,E,Reader,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>> & List<&2,Records.Fields<S>>:
32	  match owner:
33	    case Owner{Event.Declaration{Event.Domain{+key},+mode},reader,Sys.Registry{+namespace,+id,+name,+access,+cursor}}:(Owner{Event.Declaration{Event.Domain{key},mode},reader,Sys.Registry{namespace,id,name,access,cursor}},[Records.Fields{namespace,id,name,access,cursor,key,[Query.Clause{key,OrdinaryEvent.query_mode(mode)}],Records.EventKind{}}])
34	# Advanced App observer returning the same World with an empty Unit observation.
35	# Event inventory comes from registration metadata; this function does not read
36	# payloads or advance a stream cursor merely to display the App.
37	def observe(~S:Data,~C:Type,~R:Type,~P:Data,world:W.World<S,C,R,P>) -> W.World<S,C,R,P> & DTO.Observation<Unit>:(world,DTO.Observation{Unit{},[]})
38	# Reader registration result. Registered returns the runtime plus its actual
39	# reader/Registry owner; RegistrationFailed returns the runtime without a new
40	# owner. Thread the returned runtime on either branch instead of retaining an
41	# old World or copying the family owners.
42	type Registration<S:Data,C:Type,R:Type,P:Data,E:Data,F:Type,A:Type,Error:Data,Out:Type,-take:F -> F & Stream.Stream<S,E>,-put:F -> Stream.Stream<S,E> -> F,-runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>> is Type:
43	  Registered{runtime:Route.Runtime<S,C,R,P,F>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>}
44	  RegistrationFailed{runtime:Route.Runtime<S,C,R,P,F>}
45	def registered(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,declaration:Event.Declaration<S,E>,result:Route.Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
46	  match result:
47	    case Route.RegistrationFailed{runtime}:RegistrationFailed{runtime}
48	    case Route.Registered{runtime,reader,registry}:Registered{runtime,Owner{declaration,reader,registry}}
49	def declared(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,family:F,stream:Stream.Stream<S,E>,tick:Nat,name:String,+declaration:Event.Declaration<S,E>,named:W.World<S,C,R,P> & U32) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
50	  registered(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,declaration,Route.registration_named(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,family,stream,tick,declaration,name,named))
51	def taken(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,world:W.World<S,C,R,P>,tick:Nat,name:String,returned:F & Stream.Stream<S,E>) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
52	  match returned:
53	    case (family,+stream):declared(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,family,stream,tick,name,Stream.read_declaration(~S,~E,stream),W.namespace(~S,~C,~R,~P,world))
54	# Register a reader against the stream selected by the closed take/put route.
55	# Derives read authority from that stream's operational declaration and uses the
56	# ordinary Registry registration rules. It allocates a reader identity without
57	# delivering events. Refusal, including exhausted reader IDs or failed Registry
58	# registration, returns RegistrationFailed with the returned runtime.
59	# The runner receives event-runtime.Input{values, lagged, args}. This entry is
60	# for Data streams; affine payload publication/reading uses the separate
61	# ordinary-owned-event-* APIs. Qualified family setup and retention/refusal
62	# scenarios are linked from docs/parity/typed-data-events.md.
63	def register(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,runtime:Route.Runtime<S,C,R,P,F>,name:String) -> Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
64	  match runtime:
65	    case Route.Runtime{world,family,tick}:taken(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,world,tick,name,take(family))
66	# Advanced type alias for the retained reader and Registry for this exact
67	# stream route and runner. Use it in App owner aliases, not as a numeric ID token.
68	def OwnerType(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>) -> Type:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>
69	# App type after installing this family reader alongside prior owners.
70	# Keeps the schema, runtime family and combined metadata observer; this leaf adds
71	# Unit to observation rows rather than copying event payloads into debug output.
72	# Use install to perform registration and construct the extended application.
73	def Application(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>) -> Type:App.Application<S,C,R,P,F,Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,DTO.Product<Rows,Unit>,Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner))>
74	# Advanced composition entry for an already-registered reader owner.
75	# Consumes the composer and owner, returning their combined owners/observer.
76	# Does not register a second reader or deliver events. Ordinary setup uses install.
77	def append(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,value:Build.Registered<S,W.World<S,C,R,P>,Owners,Rows,prior,fold>) -> Build.Registered<S,W.World<S,C,R,P>,Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,DTO.Product<Rows,Unit>,Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner))>:
78	  match value:
79	    case Build.Registered{world,owners,_,debug}:Build.appended(~S,~W.World<S,C,R,P>,~Owners,~Rows,~prior,~fold,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~Unit,~observe(~S,~C,~R,~P),~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner),world,owners,owner,debug)
80	def appended(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,schema:Schema.Schema<S,C,R>,family:F,tick:Nat,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,value:Build.Registered<S,W.World<S,C,R,P>,Owners,Rows,prior,fold>) -> Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold):
81	  App.attached(~S,~C,~R,~P,~F,~(Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>),~DTO.Product<Rows,Unit>,~Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),~Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner)),family,tick,SchemaApp.attached(~S,~C,~R,~P,~(Owners & Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>),~DTO.Product<Rows,Unit>,~Tree.observe_pair(~W.World<S,C,R,P>,~Rows,~Unit,~prior,~observe(~S,~C,~R,~P)),~Leaf.fold_pair(~S,~Owners,~Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,~fold,~fields(~S,~C,~R,~P,~E,~Route.Reader<S,E,F,take,put>,~A,~Error,~Out,~runner)),schema,append(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,owner,value)))
82	def installed(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,~Output:Type,done:Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold) -> Output,failed:App.Application<S,C,R,P,F,Owners,Rows,prior,fold> -> Output,schema:Schema.Schema<S,C,R>,owners:Owners,recipe:Tree.Recipe<W.World<S,C,R,P>,Rows,prior>,debug:Bool,result:Registration<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> Output:
83	  match result:
84	    case RegistrationFailed{runtime}:failed(App.Application{schema,runtime,owners,recipe,debug})
85	    case Registered{Route.Runtime{world,family,tick},owner}:done(appended(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,schema,family,tick,owner,Build.Registered{world,owners,recipe,debug}))
86	# Register and append a typed Data-family reader to an existing family App.
87	# Consumes value and calls done with the extended App on success, or failed
88	# with the returned previous App on refusal. Schema, family, prior owners and
89	# debug setting are threaded through both branches. name follows Registry rules.
90	# This is setup only: it neither runs the reader nor advances its delivery cursor.
91	def install(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~Owners:Type,~Rows:Data,~prior:W.World<S,C,R,P> -> W.World<S,C,R,P> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,~Output:Type,done:Application(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold) -> Output,failed:App.Application<S,C,R,P,F,Owners,Rows,prior,fold> -> Output,value:App.Application<S,C,R,P,F,Owners,Rows,prior,fold>,name:String) -> Output:
92	  match value:
93	    case App.Application{schema,runtime,owners,recipe,debug}:installed(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~Owners,~Rows,~prior,~fold,~Output,done,failed,schema,owners,recipe,debug,register(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,runtime,name))
94	# Returned runtime and reader owner together with event-runtime.Completion.
95	# Succeeded carries Out, Failed carries the runner's error, and Rejected returns
96	# Args when execution authority is refused. Always retain the returned runtime
97	# and owner; failure does not mean the input owners can be reused separately.
98	type RunResult<S:Data,C:Type,R:Type,P:Data,E:Data,F:Type,A:Type,Error:Data,Out:Type,-take:F -> F & Stream.Stream<S,E>,-put:F -> Stream.Stream<S,E> -> F,-runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>> is Type:
99	  Ran{runtime:Route.Runtime<S,C,R,P,F>,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,completion:Ev.Completion<Error,Out,A>}
100	def ran(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,declaration:Event.Declaration<S,E>,result:Route.RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>) -> RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
101	  match result:
102	    case Route.Ran{runtime,reader,registry,completion}:Ran{runtime,Owner{declaration,reader,registry},completion}
103	# Run this registered reader against its typed family stream.
104	# The runner receives retained unread values in publication order, a lagged flag
105	# and args. On Sys.Succeeded the reader advances and synchronize threads committed
106	# World/family publication into the returned runtime. On Sys.Failed its stream
107	# position does not advance; the runner's returned World is retained. This wrapper
108	# does not itself undo arbitrary callback or host-IO effects.
109	# A routing/registration rejection returns Rejected{args} without consuming the
110	# read position. The completion and actual reader/Registry owners are returned
111	# on every branch. Stream retention/capacity still bound later availability;
112	# lagged reports dropped unread history, so failure is not an unlimited retry
113	# retention promise. Supply the matching family synchronization route; merely
114	# observing fields/metadata is not a run. See event-family-routing and the
115	# qualified scenarios in docs/parity/typed-data-events.md.
116	def run(~S:Data,~C:Type,~R:Type,~P:Data,~E:Data,~F:Type,~A:Type,~Error:Data,~Out:Type,~take:F -> F & Stream.Stream<S,E>,~put:F -> Stream.Stream<S,E> -> F,~runner:W.World<S,C,R,P> -> Ev.Input<E,A> -> Sys.Outcome<W.World<S,C,R,P>,Error,Out>,~synchronize:W.World<S,C,R,P> -> F -> Nat -> W.World<S,C,R,P> & F,owner:Owner<S,E,Route.Reader<S,E,F,take,put>,Sys.Registry<S,C,R,P,Ev.Input<E,A>,Error,Out,runner>>,runtime:Route.Runtime<S,C,R,P,F>,args:A) -> RunResult<S,C,R,P,E,F,A,Error,Out,take,put,runner>:
117	  match owner:
118	    case Owner{declaration,reader,registry}:ran(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,declaration,Route.run(~S,~C,~R,~P,~E,~F,~A,~Error,~Out,~take,~put,~runner,~synchronize,runtime,reader,registry,args))