1	import Base
2	import ./change_tick.bend as Tick
3	import ./machine_world.bend as MachineWorld
4	import ./ordinary_app_execution_run.bend as ExecuteRun
5	import ./machine_handler_engine.bend as HandlerEngine
6	import ./machine_handlers.bend as Handlers
7	import ./machine.bend as Machine
8	import ./ordinary_machine_declaration.bend as MachineDecl
9	import ./ordinary_app_resource_field_declaration.bend as FieldDecl
10	import ./ordinary_resource_field_system.bend as FieldSystem
11	import ./transaction.bend as Transaction
12	import ./execution_context.bend as Ctx
13	import ./trace_facts.bend as Facts
14	import ./execution_trace.bend as Trace
15	import ./ordinary_app_schedule.bend as Single
16	import ./ordinary_app_schema.bend as App
17	import ./ordinary_app_composer.bend as Build
18	import ./ordinary_app_recipe.bend as Tree
19	import ./ordinary_app_registration_fields.bend as Records
20	import ./ordinary_app_observation_data.bend as DTO
21	import ./ordinary_schema.bend as Schema
22	import ./world.bend as W
23	import ./system.bend as Sys
24	import ./schedule.bend as Sch
25	import ./schedule_provision.bend as SP
26	
27	# A detached batch is returned only after its actual ECS run has completed.
28	# Execution returns a retained application result and a detached trace batch.
29	# Inspect result for Finished/Failed/Rejected/Missing; tracing is Disabled or
30	# Ready according to the actual run. Subscribers process the copied batch
31	# afterward; external printing/file IO is outside ECS rollback.
32	# Example: examples/scheduled-health/application.bend, ran().
33	type Run<A:Type,S:Data,Error:Data> is Type:
34	  Run{result:Single.RunResult<A,S,Error>,batch:Trace.Batch<Error>}
35	type Journal<Error:Data> is Data:
36	  Journal{catalog:List<&2,Trace.Identity>,events:List<&2,Trace.TraceEvent<Error>>}
37	def catalog(~S:Data,fields:List<&2,Records.Fields<S>>) -> List<&2,Trace.Identity>:
38	  match fields:
39	    case []:[]
40	    case Records.Fields{namespace,id,name,_,_,_,_,_} <> rest:Trace.Identity{namespace,id,Some{name}} <> catalog(~S,rest)
41	def named(found:Bool,name:Maybe<&2,String>,next:Unit -> Maybe<&2,String>) -> Maybe<&2,String>:
42	  match found:
43	    case True{}:name
44	    case False{}:next(Unit{})
45	def registration_name(items:List<&2,Trace.Identity>,+id:U32) -> Maybe<&2,String>:
46	  match items:
47	    case []:None{}
48	    case Trace.Identity{_,candidate,name} <> rest:named(U32.is_eq(candidate,id),name,_ => registration_name(rest,id))
49	def identity_clock(~S:Data,~C:Type,~R:Type,~E:Data,namespace:U32,id:U32,name:Maybe<&2,String>,observed:W.World<S,C,R,E> & Tick.Tick) -> W.World<S,C,R,E> & (Trace.Identity & Tick.Tick):
50	  match observed:
51	    case (world,clock):(world,(Trace.Identity{namespace,id,name},clock))
52	def identity_namespace(~S:Data,~C:Type,~R:Type,~E:Data,id:U32,name:Maybe<&2,String>,observed:W.World<S,C,R,E> & U32) -> W.World<S,C,R,E> & (Trace.Identity & Tick.Tick):
53	  match observed:
54	    case (world,namespace):identity_clock(~S,~C,~R,~E,namespace,id,name,W.clock(~S,~C,~R,~E,world))
55	def identity(~S:Data,~C:Type,~R:Type,~E:Data,~Error:Data,journal:Journal<Error>,world:W.World<S,C,R,E>,+id:U32) -> W.World<S,C,R,E> & (Trace.Identity & Tick.Tick):
56	  match journal:
57	    case Journal{catalog,_}:identity_namespace(~S,~C,~R,~E,id,registration_name(catalog,id),W.namespace(~S,~C,~R,~E,world))
58	def record(~Error:Data,event:Trace.TraceEvent<Error>,journal:Journal<Error>) -> Journal<Error>:
59	  match journal:
60	    case Journal{catalog,events}:Journal{catalog,event <> events}
61	def succeeded(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,owners:H,fact:Trace.Identity,before:Tick.Tick,journal:Journal<Error>,observed:W.World<S,C,R,E> & Tick.Tick) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
62	  match observed:
63	    case (world,after):((owners,record(~Error,Trace.SystemCompleted{fact,before,after,Trace.SystemOk{}},journal)),Sys.Succeeded{world,Unit{}})
64	def failed(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,owners:H,fact:Trace.Identity,before:Tick.Tick,journal:Journal<Error>,+error:Error,observed:W.World<S,C,R,E> & Tick.Tick) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
65	  match observed:
66	    case (world,after):((owners,record(~Error,Trace.SystemCompleted{fact,before,after,Trace.SystemError{error}},journal)),Sys.Failed{world,error})
67	def record_operations(~Error:Data,facts:List<&2,Facts.Fact>,+identity:Trace.Identity,journal:Journal<Error>) -> Journal<Error>:
68	  match facts:
69	    case []:journal
70	    case fact <> rest:record_operations(~Error,rest,identity,record(~Error,Trace.OperationObserved{identity,fact},journal))
71	def record_barrier_operations(~Error:Data,facts:List<&2,Facts.Fact>,journal:Journal<Error>) -> Journal<Error>:
72	  match facts:
73	    case []:journal
74	    case fact <> rest:record_barrier_operations(~Error,rest,record(~Error,Trace.BarrierOperationObserved{fact},journal))
75	def barrier_returned(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,owners:H,journal:Journal<Error>,result:W.World<S,C,R,E> & Facts.State) -> (H & Journal<Error>) & W.World<S,C,R,E>:
76	  match result:
77	    case (world,trace):((owners,record_barrier_operations(~Error,Facts.facts(trace),journal)),world)
78	def barrier_recorded(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,owners:H & Journal<Error>,world:W.World<S,C,R,E>) -> (H & Journal<Error>) & W.World<S,C,R,E>:
79	  match owners:
80	    case (owners,journal):barrier_returned(~S,~C,~R,~E,~H,~Error,owners,journal,W.barrier_traced(~S,~C,~R,~E,world,Facts.On{[]}))
81	def dispatched(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,+fact:Trace.Identity,before:Tick.Tick,journal:Journal<Error>,result:H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
82	  match result:
83	    case (owners,Sys.Succeeded{Ctx.Context{world,trace},_}):succeeded(~S,~C,~R,~E,~H,~Error,owners,fact,before,record_operations(~Error,Facts.facts(trace),fact,journal),W.clock(~S,~C,~R,~E,world))
84	    case (owners,Sys.Failed{Ctx.Context{world,trace},error}):failed(~S,~C,~R,~E,~H,~Error,owners,fact,before,record_operations(~Error,Facts.facts(trace),fact,journal),error,W.clock(~S,~C,~R,~E,world))
85	def identified(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,owners:H,journal:Journal<Error>,+id:U32,result:W.World<S,C,R,E> & (Trace.Identity & Tick.Tick)) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
86	  match result:
87	    case (world,(fact,before)):dispatched(~S,~C,~R,~E,~H,~Error,fact,before,journal,dispatch(owners,Ctx.Context{world,Facts.On{[]}},id))
88	def dispatch_id(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,owners:H & Journal<Error>,world:W.World<S,C,R,E>,+id:U32) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
89	  match owners:
90	    case (owners,+journal):identified(~S,~C,~R,~E,~H,~Error,~dispatch,owners,journal,id,identity(~S,~C,~R,~E,~Error,journal,world,id))
91	def dispatch_recorded(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,owners:H & Journal<Error>,world:W.World<S,C,R,E>,id:U32) -> (H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
92	  dispatch_id(~S,~C,~R,~E,~H,~Error,~dispatch,owners,world,id)
93	# Sch.Ran supplies position only. The completion/error comes from the actual
94	# dispatch result above, never from a guessed interpretation of Ran.
95	def ran_chunk_more(~Error:Data,event:Trace.TraceEvent<Error>,chunk:List<&2,Trace.TraceEvent<Error>> & List<&2,Trace.TraceEvent<Error>>) -> List<&2,Trace.TraceEvent<Error>> & List<&2,Trace.TraceEvent<Error>>:
96	  match chunk:
97	    case (prefix,remaining):(event <> prefix,remaining)
98	def ran_chunk(~Error:Data,journal:List<&2,Trace.TraceEvent<Error>>) -> List<&2,Trace.TraceEvent<Error>> & List<&2,Trace.TraceEvent<Error>>:
99	  match journal:
100	    case []:([],[])
101	    case Trace.OperationObserved{identity,fact} <> rest:ran_chunk_more(~Error,Trace.OperationObserved{identity,fact},ran_chunk(~Error,rest))
102	    case event <> rest:([event],rest)
103	def barrier_chunk(~Error:Data,journal:List<&2,Trace.TraceEvent<Error>>) -> List<&2,Trace.TraceEvent<Error>> & List<&2,Trace.TraceEvent<Error>>:
104	  match journal:
105	    case []:([],[])
106	    case Trace.BarrierOperationObserved{fact} <> rest:ran_chunk_more(~Error,Trace.BarrierOperationObserved{fact},barrier_chunk(~Error,rest))
107	    case event <> rest:([],event <> rest)
108	def merged_chunk(~Error:Data,next:List<&2,Trace.TraceEvent<Error>> -> List<&2,Trace.TraceEvent<Error>>,chunk:List<&2,Trace.TraceEvent<Error>> & List<&2,Trace.TraceEvent<Error>>) -> List<&2,Trace.TraceEvent<Error>>:
109	  match chunk:
110	    case (prefix,remaining):List.append(&2,Trace.TraceEvent<Error>,prefix,next(remaining))
111	def merge(~Error:Data,observations:List<&2,Sch.Observation>,journal:List<&2,Trace.TraceEvent<Error>>) -> List<&2,Trace.TraceEvent<Error>>:
112	  match observations journal:
113	    case [], journal:journal
114	    case Sch.Entered{name} <> rest, journal:Trace.PhaseEntered{name} <> merge(~Error,rest,journal)
115	    case Sch.Skipped{id} <> rest, journal:Trace.SystemSkipped{id} <> merge(~Error,rest,journal)
116	    case Sch.Applied{} <> rest, journal:merged_chunk(~Error,(remaining=>Trace.DeferredApplied{} <> merge(~Error,rest,remaining)),barrier_chunk(~Error,journal))
117	    case Sch.Ran{_} <> rest, journal:merged_chunk(~Error,(remaining=>merge(~Error,rest,remaining)),ran_chunk(~Error,journal))
118	def batch(~Error:Data,namespace:U32,name:String,observations:List<&2,Sch.Observation>,journal:Journal<Error>,outcome:Trace.Completion<Error>) -> Trace.Batch<Error>:
119	  match journal:
120	    case Journal{_,events}:Trace.BendvyReady{Trace.ScheduleStarted{namespace,name} <> List.append(&2,Trace.TraceEvent<Error>,merge(~Error,observations,List.reverse(&2,Trace.TraceEvent<Error>,events)),[Trace.ScheduleEnded{outcome}])}
121	def restored(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,schema:Schema.Schema<S,C,R>,world:W.World<S,C,R,E>,owners:H,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,debug:Bool,plan:SP.Plan<S>,namespace:U32,name:String,steps:List<&2,Sch.Step>,needs:List<&2,SP.Requirement<S>>) -> Single.Application<S,C,R,E,H,Rows,observe,fold>:
122	  Single.Application{App.Application{schema,Build.Registered{world,owners,recipe,debug}},plan,namespace,name,steps,needs}
123	def disabled_outcome(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,result:H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>) -> H & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
124	  match result:
125	    case (owners,Sys.Succeeded{Ctx.Context{world,_},output}):(owners,Sys.Succeeded{world,output})
126	    case (owners,Sys.Failed{Ctx.Context{world,_},error}):(owners,Sys.Failed{world,error})
127	def dispatch_disabled(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,owners:H,world:W.World<S,C,R,E>,id:U32) -> H & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
128	  disabled_outcome(~S,~C,~R,~E,~H,~Error,dispatch(owners,Ctx.Context{world,Facts.Off{}},id))
129	# Disabled runs use the old ordinary run directly: no identity/clock scan,
130	# completion journal, trace-event construction or subscriber delivery occurs.
131	# Transition-handler failure identifying its exit/transition/enter phase
132	# and error text. Enter failure follows an already committed transition;
133	# inspect the phase instead of assuming whole-transition rollback.
134	type HandlerFailure is Data:
135	  HandlerFailure{phase:Handlers.Phase,error:String}
136	# Transition-handler execution retains the App, all handler owners, the
137	# captured machine snapshot, optional phase failure and detached trace batch.
138	# The captured snapshot is not a rollback command or a copy of affine owners.
139	type HandlerRun<App:Type,Owners:Type,Snapshot:Data> is Type:
140	  HandlerRun{app:App,owners:Owners,captured:Snapshot,failure:Maybe<&2,HandlerFailure>,batch:Trace.Batch<String>}
141	def operation_events(facts:List<&2,Facts.Fact>) -> List<&2,Trace.TraceEvent<String>>:
142	  match facts:
143	    case []:[]
144	    case fact <> rest:Trace.ExecutionOperationObserved{fact} <> operation_events(rest)
145	def operation_batch(state:Facts.State) -> Trace.Batch<String>:
146	  match state:
147	    case Facts.Off{}:Trace.Disabled{}
148	    case Facts.On{facts}:Trace.BendvyReady{operation_events(List.reverse(&2,Facts.Fact,facts))}
149	def trace_state(debug:Bool) -> Facts.State:
150	  match debug:
151	    case False{}:Facts.Off{}
152	    case True{}:Facts.On{[]}
153	def handlers_returned(~S:Data,~C:Type,~R:Type,~E:Data,~Ops:Type -> Type,~Value:Data,~declaration:FieldDecl.Declaration<S,C,R,E,Ops,Value>,~K:Data,~MachineV:Data,~take:R -> R & Machine.Slot<S,K,MachineV>,~put:R -> Machine.Slot<S,K,MachineV> -> R,~machine:MachineDecl.Declaration<S,K,MachineV,R,take>,~same:MachineV -> MachineV -> Bool,~publish:W.World<S,C,R,E> -> Machine.Transition<MachineV> -> W.World<S,C,R,E>,~body:@-H:Type -> @-caps:Ops(H) -> Handlers.Invocation<MachineV> -> H -> H & (Transaction.Outcome<String> & Unit),~Owners:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,schema:Schema.Schema<S,C,R>,previous:Owners,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,debug:Bool,plan:SP.Plan<S>,namespace:U32,name:String,steps:List<&2,Sch.Step>,needs:List<&2,SP.Requirement<S>>,captured:Machine.Snapshot<MachineV>,result:HandlerEngine.HandlerResult<Ctx.Context<W.World<S,C,R,E>>,Handlers.Entry<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>,Handlers.Phase>) -> HandlerRun<Single.Application<S,C,R,E,Owners,Rows,observe,fold>,Handlers.Owners<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>,Machine.Snapshot<MachineV>>:
154	  match result:
155	    case HandlerEngine.Completed{HandlerEngine.Owners{exit,transition,enter},Ctx.Context{world,trace}}:HandlerRun{restored(~S,~C,~R,~E,~Owners,~Rows,~observe,~fold,~String,schema,world,previous,recipe,debug,plan,namespace,name,steps,needs),Handlers.Owners{exit,transition,enter},captured,None{},operation_batch(trace)}
156	    case HandlerEngine.Stopped{HandlerEngine.Owners{exit,transition,enter},Ctx.Context{world,trace},phase,error}:HandlerRun{restored(~S,~C,~R,~E,~Owners,~Rows,~observe,~fold,~String,schema,world,previous,recipe,debug,plan,namespace,name,steps,needs),Handlers.Owners{exit,transition,enter},captured,Some{HandlerFailure{phase,error}},operation_batch(trace)}
157	def handlers_read(~S:Data,~C:Type,~R:Type,~E:Data,~Ops:Type -> Type,~Value:Data,~declaration:FieldDecl.Declaration<S,C,R,E,Ops,Value>,~K:Data,~MachineV:Data,~take:R -> R & Machine.Slot<S,K,MachineV>,~put:R -> Machine.Slot<S,K,MachineV> -> R,~machine:MachineDecl.Declaration<S,K,MachineV,R,take>,~same:MachineV -> MachineV -> Bool,~publish:W.World<S,C,R,E> -> Machine.Transition<MachineV> -> W.World<S,C,R,E>,~body:@-H:Type -> @-caps:Ops(H) -> Handlers.Invocation<MachineV> -> H -> H & (Transaction.Outcome<String> & Unit),~Owners:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,schema:Schema.Schema<S,C,R>,previous:Owners,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,+debug:Bool,plan:SP.Plan<S>,namespace:U32,name:String,steps:List<&2,Sch.Step>,needs:List<&2,SP.Requirement<S>>,owners:Handlers.Owners<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>,read:W.World<S,C,R,E> & Machine.Slot<S,K,MachineV>) -> HandlerRun<Single.Application<S,C,R,E,Owners,Rows,observe,fold>,Handlers.Owners<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>,Machine.Snapshot<MachineV>>:
158	  match read:
159	    case (world,+slot):handlers_returned(~S,~C,~R,~E,~Ops,~Value,~declaration,~K,~MachineV,~take,~put,~machine,~same,~publish,~body,~Owners,~Rows,~observe,~fold,schema,previous,recipe,debug,plan,namespace,name,steps,needs,Machine.snapshot(~S,~K,~MachineV,slot),ExecuteRun.handlers_resource_field(~S,~C,~R,~E,~Ops,~Value,~declaration,~K,~MachineV,~take,~put,~machine,~same,~publish,~body,owners,Ctx.Context{world,trace_state(debug)},Machine.snapshot(~S,~K,~MachineV,slot)))
160	def run_handlers_resource_field(~S:Data,~C:Type,~R:Type,~E:Data,~Ops:Type -> Type,~Value:Data,~declaration:FieldDecl.Declaration<S,C,R,E,Ops,Value>,~K:Data,~MachineV:Data,~take:R -> R & Machine.Slot<S,K,MachineV>,~put:R -> Machine.Slot<S,K,MachineV> -> R,~machine:MachineDecl.Declaration<S,K,MachineV,R,take>,~same:MachineV -> MachineV -> Bool,~publish:W.World<S,C,R,E> -> Machine.Transition<MachineV> -> W.World<S,C,R,E>,~body:@-H:Type -> @-caps:Ops(H) -> Handlers.Invocation<MachineV> -> H -> H & (Transaction.Outcome<String> & Unit),~Owners:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:Owners -> Owners & List<&2,Records.Fields<S>>,app:Single.Application<S,C,R,E,Owners,Rows,observe,fold>,owners:Handlers.Owners<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>) -> HandlerRun<Single.Application<S,C,R,E,Owners,Rows,observe,fold>,Handlers.Owners<S,C,R,E,MachineV,FieldSystem.runner(~S,~C,~R,~E,~Ops,~Handlers.Invocation<MachineV>,~String,~Unit,~FieldDecl.ordinary(~S,~C,~R,~E,~Ops,~Value,declaration),~body)>,Machine.Snapshot<MachineV>>:
161	  match app:
162	    case Single.Application{App.Application{schema,Build.Registered{world,previous,recipe,debug}},plan,namespace,name,steps,needs}:handlers_read(~S,~C,~R,~E,~Ops,~Value,~declaration,~K,~MachineV,~take,~put,~machine,~same,~publish,~body,~Owners,~Rows,~observe,~fold,schema,previous,recipe,debug,plan,namespace,name,steps,needs,owners,MachineWorld.take(~S,~K,~MachineV,~C,~R,~E,~take,world))
163	
164	type ExecutionState<Error:Data> is Data:
165	  Off{}
166	  On{journal:Journal<Error>}
167	def disabled_dispatched(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,result:H & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>) -> (H & ExecutionState<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
168	  match result:
169	    case (owners,outcome):((owners,Off{}),outcome)
170	def enabled_dispatched(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,result:(H & Journal<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>) -> (H & ExecutionState<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
171	  match result:
172	    case ((owners,journal),outcome):((owners,On{journal}),outcome)
173	def dispatch_state(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,owner:H & ExecutionState<Error>,world:W.World<S,C,R,E>,id:U32) -> (H & ExecutionState<Error>) & Sys.Outcome<W.World<S,C,R,E>,Error,Unit>:
174	  match owner:
175	    case (owners,Off{}):disabled_dispatched(~S,~C,~R,~E,~H,~Error,dispatch_disabled(~S,~C,~R,~E,~H,~Error,~dispatch,owners,world,id))
176	    case (owners,On{journal}):enabled_dispatched(~S,~C,~R,~E,~H,~Error,dispatch_recorded(~S,~C,~R,~E,~H,~Error,~dispatch,(owners,journal),world,id))
177	def enabled_barrier(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,result:(H & Journal<Error>) & W.World<S,C,R,E>) -> (H & ExecutionState<Error>) & W.World<S,C,R,E>:
178	  match result:
179	    case ((owners,journal),world):((owners,On{journal}),world)
180	def barrier_state(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Error:Data,owner:H & ExecutionState<Error>,world:W.World<S,C,R,E>) -> (H & ExecutionState<Error>) & W.World<S,C,R,E>:
181	  match owner:
182	    case (owners,Off{}):((owners,Off{}),W.barrier(~S,~C,~R,~E,world))
183	    case (owners,On{journal}):enabled_barrier(~S,~C,~R,~E,~H,~Error,barrier_recorded(~S,~C,~R,~E,~H,~Error,(owners,journal),world))
184	def executed(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,schema:Schema.Schema<S,C,R>,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,plan:SP.Plan<S>,needs:List<&2,SP.Requirement<S>>,result:Sch.RunResult<S,C,R,E,H & ExecutionState<Error>,Error>) -> Run<Single.Application<S,C,R,E,H,Rows,observe,fold>,S,Error>:
185	  match result:
186	    case Sch.Finished{Sch.Schedule{namespace,name,steps,(owners,Off{})},world,observations}:Run{Single.Finished{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,False{},plan,namespace,name,steps,needs),observations},Trace.Disabled{}}
187	    case Sch.Finished{Sch.Schedule{+namespace,+name,steps,(owners,On{journal})},world,+observations}:Run{Single.Finished{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,True{},plan,namespace,name,steps,needs),observations},batch(~Error,namespace,name,observations,journal,Trace.EndOk{})}
188	    case Sch.Failed{Sch.Schedule{namespace,name,steps,(owners,Off{})},world,error,observations}:Run{Single.Failed{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,False{},plan,namespace,name,steps,needs),error,observations},Trace.Disabled{}}
189	    case Sch.Failed{Sch.Schedule{+namespace,+name,steps,(owners,On{journal})},world,+error,+observations}:Run{Single.Failed{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,True{},plan,namespace,name,steps,needs),error,observations},batch(~Error,namespace,name,observations,journal,Trace.EndFailed{error})}
190	    case Sch.Rejected{Sch.Schedule{namespace,name,steps,(owners,Off{})},world}:Run{Single.Rejected{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,False{},plan,namespace,name,steps,needs)},Trace.Disabled{}}
191	    case Sch.Rejected{Sch.Schedule{+namespace,+name,steps,(owners,On{journal})},world}:Run{Single.Rejected{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,True{},plan,namespace,name,steps,needs)},Trace.BendvyReady{[Trace.ScheduleAdmissionRefused{namespace,name,Trace.EndRejected{}}]}}
192	def provisioned(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,schema:Schema.Schema<S,C,R>,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,plan:SP.Plan<S>,result:SP.ProvisionResult<S,C,R,E,H & ExecutionState<Error>,Error>) -> Run<Single.Application<S,C,R,E,H,Rows,observe,fold>,S,Error>:
193	  match result:
194	    case SP.Executed{result,needs}:executed(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,recipe,plan,needs,result)
195	    case SP.Missing{SP.Provisioned{Sch.Schedule{namespace,name,steps,(owners,Off{})},needs},world,missing}:Run{Single.Missing{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,False{},plan,namespace,name,steps,needs),missing},Trace.Disabled{}}
196	    case SP.Missing{SP.Provisioned{Sch.Schedule{+namespace,+name,steps,(owners,On{journal})},needs},world,+missing}:Run{Single.Missing{restored(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,world,owners,recipe,True{},plan,namespace,name,steps,needs),missing},Trace.BendvyReady{[Trace.ScheduleAdmissionRefused{namespace,name,Trace.EndMissing{}}]}}
197	def start(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,~condition:W.World<S,C,R,E> -> U32 -> W.World<S,C,R,E> & Bool,~provision:W.World<S,C,R,E> -> W.World<S,C,R,E> & List<&2,SP.Requirement<S>>,schema:Schema.Schema<S,C,R>,world:W.World<S,C,R,E>,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,plan:SP.Plan<S>,namespace:U32,name:String,steps:List<&2,Sch.Step>,needs:List<&2,SP.Requirement<S>>,owners:H,state:ExecutionState<Error>) -> Run<Single.Application<S,C,R,E,H,Rows,observe,fold>,S,Error>:
198	  provisioned(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,schema,recipe,plan,SP.run_with_barrier(~S,~C,~R,~E,~(H & ExecutionState<Error>),~Error,~dispatch_state(~S,~C,~R,~E,~H,~Error,~dispatch),~condition,~barrier_state(~S,~C,~R,~E,~H,~Error),~provision,SP.Provisioned{Sch.Schedule{namespace,name,steps,(owners,state)},needs},world))
199	def prepared(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,~condition:W.World<S,C,R,E> -> U32 -> W.World<S,C,R,E> & Bool,~provision:W.World<S,C,R,E> -> W.World<S,C,R,E> & List<&2,SP.Requirement<S>>,schema:Schema.Schema<S,C,R>,world:W.World<S,C,R,E>,recipe:Tree.Recipe<W.World<S,C,R,E>,Rows,observe>,plan:SP.Plan<S>,namespace:U32,name:String,steps:List<&2,Sch.Step>,needs:List<&2,SP.Requirement<S>>,value:H & List<&2,Records.Fields<S>>) -> Run<Single.Application<S,C,R,E,H,Rows,observe,fold>,S,Error>:
200	  match value:
201	    case (owners,fields):start(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,~dispatch,~condition,~provision,schema,world,recipe,plan,namespace,name,steps,needs,owners,On{Journal{catalog(~S,fields),[]}})
202	# Advanced traced schedule execution with supplied dispatch/condition/provision
203	# callbacks. Returns Run on admission refusal, missing requirements, success
204	# or system failure, retaining App ownership throughout. Disabled debug omits
205	# trace collection; enabled debug derives identities and operations from actual
206	# execution. Supplied callbacks are trusted world-transforming adapters;
207	# arbitrary callback mutations/host IO are not automatically transactional.
208	# Prefer ordinary-app-run.run for registration-derived dispatch.
209	def run(~S:Data,~C:Type,~R:Type,~E:Data,~H:Type,~Rows:Data,~observe:W.World<S,C,R,E> -> W.World<S,C,R,E> & DTO.Observation<Rows>,~fold:H -> H & List<&2,Records.Fields<S>>,~Error:Data,~dispatch:H -> Ctx.Context<W.World<S,C,R,E>> -> U32 -> H & Sys.Outcome<Ctx.Context<W.World<S,C,R,E>>,Error,Unit>,~condition:W.World<S,C,R,E> -> U32 -> W.World<S,C,R,E> & Bool,~provision:W.World<S,C,R,E> -> W.World<S,C,R,E> & List<&2,SP.Requirement<S>>,value:Single.Application<S,C,R,E,H,Rows,observe,fold>) -> Run<Single.Application<S,C,R,E,H,Rows,observe,fold>,S,Error>:
210	  match value:
211	    case Single.Application{App.Application{schema,Build.Registered{world,owners,recipe,False{}}},plan,namespace,name,steps,needs}:start(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,~dispatch,~condition,~provision,schema,world,recipe,plan,namespace,name,steps,needs,owners,Off{})
212	    case Single.Application{App.Application{schema,Build.Registered{world,owners,recipe,True{}}},plan,namespace,name,steps,needs}:prepared(~S,~C,~R,~E,~H,~Rows,~observe,~fold,~Error,~dispatch,~condition,~provision,schema,world,recipe,plan,namespace,name,steps,needs,fold(owners))