Emitter: an effectful fold step is realized as a loop in the enclosing async frame (native broker class E)

dag/gunbc/instruments/native_emission_controls.dag

@@ -147,6 +147,37 @@ fn claim_driver_filesystem_cases() -> List<NativeEmissionCase> {
   ]
 }
 
+// RULE: an effectful fold step is realized as a loop in the enclosing async frame
+// (v1.compiler.emit_rust emit_rust_effectful_fold_loop). Iterator::fold takes its step as a
+// synchronous closure, so a step that reaches an effectful callee put that callee's `.await?` inside
+// it: E0728, the shape of gunbc.auth.approval_decision_store pending_escalations. The step here
+// reads Filesystem for every element and accumulates a record, as that fold does. Every path sits
+// under a root no host provisions, so each read is a typed refusal and the counts are decided
+// everywhere; the red rows are the record's two fields and the empty list.
+type ControlReadTally {
+  refused: Int
+  seen: Int
+}
+
+fn control_tally_absent_reads(paths: List<String>) -> ControlReadTally {
+  fold(paths, init: ControlReadTally { refused: 0, seen: 0 }, f: (acc, p) =>
+    let read = Filesystem.Read(path: p)
+    ControlReadTally { refused: if read.success { acc.refused } else { acc.refused + 1 }, seen: acc.seen + 1 }
+  )
+}
+
+fn control_absent_paths() -> List<String> {
+  ["/gunbc-native-emission-controls-absent/a", "/gunbc-native-emission-controls-absent/b", "/gunbc-native-emission-controls-absent/c"]
+}
+
+fn effectful_fold_step_cases() -> List<NativeEmissionCase> {
+  [
+    NativeEmissionCase { identity: "effectful_fold_step_counts_every_refused_read", held: control_tally_absent_reads(paths: control_absent_paths()).refused == 3 },
+    NativeEmissionCase { identity: "effectful_fold_step_red_seen_counts_elements", held: control_tally_absent_reads(paths: control_absent_paths()).seen == 3 },
+    NativeEmissionCase { identity: "effectful_fold_step_red_empty_keeps_init", held: control_tally_absent_reads(paths: []).seen == 0 }
+  ]
+}
+
 data native_emission_expected_identities: List<String> = [
   "set_struct_eq_same_members_in_another_order",
   "set_struct_eq_red_bare_field_differs",
@@ -163,11 +194,14 @@ data native_emission_expected_identities: List<String> = [
   "filter_guard_receipt_blank_tail_accepts",
   "filter_guard_receipt_red_nonblank_tail_refuses",
   "filter_guard_receipt_red_empty_tag_short_circuits",
-  "claim_driver_filesystem_absent_path_read_is_refused"
+  "claim_driver_filesystem_absent_path_read_is_refused",
+  "effectful_fold_step_counts_every_refused_read",
+  "effectful_fold_step_red_seen_counts_elements",
+  "effectful_fold_step_red_empty_keeps_init"
 ]
 
 fn native_emission_cases() -> List<NativeEmissionCase> {
-  concat(concat(header_bound_cases(), filter_in_branch_condition_cases()), claim_driver_filesystem_cases())
+  concat(concat(concat(header_bound_cases(), filter_in_branch_condition_cases()), claim_driver_filesystem_cases()), effectful_fold_step_cases())
 }
 
 fn case_row(c: NativeEmissionCase) -> String {

src/v1/05_emit_rust.dag

@@ -12564,14 +12564,20 @@ fn emit_typed_collection_lambda(lambda_expr: Node, elem_type_str: String, regist
   }
 }
 
+// A unit-element accumulator rendering is an inference gap, not a type, so both fold realizations
+// leave it to the target's inference rather than annotate it.
+fn rust_fold_safe_acc_type(acc_type_str: String) -> String {
+  if acc_type_str == "Rc<Vec<()>>" || acc_type_str == "Vec<()>" || acc_type_str == "Option<()>" { "_" } else { acc_type_str }
+}
+
 fn emit_typed_fold_lambda(lambda_expr: Node, acc_type_str: String, elem_type_str: String, elem_borrowed: Bool, registry: Map<String, ItemInfo>, scope: InferScope, depth: Int, shared_types: Set<String>, emit_info: EmitGraphInfo) -> String {
   match lambda_expr.expr_data {
     ExprLambda =>
       let ps = lambda_param_names_at(texpr: lambda_expr, source_indices: scope.type_env.source_indices)
       let bd = lambda_body(texpr: lambda_expr)
       let pn = lambda_expr.children |> skip(1)
 
-      let safe_acc_type = if acc_type_str == "Rc<Vec<()>>" || acc_type_str == "Vec<()>" || acc_type_str == "Option<()>" { "_" } else { acc_type_str }
+      let safe_acc_type = rust_fold_safe_acc_type(acc_type_str: acc_type_str)
       let fallback_types = ps |> enumerate |> map(pair =>
         if pair.first == 0 { safe_acc_type } else { elem_type_str }
       )
@@ -12782,13 +12788,91 @@ fn emit_rust_fold_method_call(method_call_node: Node, fold_accumulator_type: Nod
     Present { value: a } => fold_lambda_element_unused(lambda_expr: arg_value(n: a), source_indices: scope.type_env.source_indices)
     Absent => false
   }
-  let fold_fn = match args |> skip(1) |> first {
-    Present { value: a } =>
-      emit_typed_fold_lambda(lambda_expr: arg_value(n: a), acc_type_str: lambda_acc_type_str, elem_type_str: fold_elem_type_str, elem_borrowed: elem_unused, registry: registry, scope: scope, depth: depth, shared_types: shared_types, emit_info: fold_emit_info)
-    Absent => "compile_error!(\"missing fold function argument\")"
-  }
   let iter_template = if elem_unused { replace(sharing.iter_owned, ".cloned()", "") } else { sharing.iter_owned }
-  concat(apply_type_template1(template: iter_template, arg0: recv_str), ".fold(", init_str, ", ", fold_fn, ")")
+  let iter_str = apply_type_template1(template: iter_template, arg0: recv_str)
+  let step_awaits = match args |> skip(1) |> first {
+    Present { value: a } => rust_expr_reaches_awaited_call(n: arg_value(n: a), registry: registry, source_indices: scope.type_env.source_indices)
+    Absent => false
+  }
+  if step_awaits {
+    emit_rust_effectful_fold_loop(lambda_expr: fold_lambda_node, iter_str: iter_str, init_str: init_str, acc_type_str: lambda_acc_type_str, registry: registry, scope: scope, depth: depth, shared_types: shared_types, emit_info: fold_emit_info)
+  } else {
+    let fold_fn = match args |> skip(1) |> first {
+      Present { value: a } =>
+        emit_typed_fold_lambda(lambda_expr: arg_value(n: a), acc_type_str: lambda_acc_type_str, elem_type_str: fold_elem_type_str, elem_borrowed: elem_unused, registry: registry, scope: scope, depth: depth, shared_types: shared_types, emit_info: fold_emit_info)
+      Absent => "compile_error!(\"missing fold function argument\")"
+    }
+    concat(iter_str, ".fold(", init_str, ", ", fold_fn, ")")
+  }
+}
+
+// A FOLD IS A BOUNDED LOOP, AND `Iterator::fold` IS ONLY ONE REALIZATION OF IT -- THE ONE THAT
+// PRESUMES A PURE STEP. The step is handed to fold as a closure, and a Rust closure is a separate,
+// synchronous frame: an effectful callee inside it is emitted, by the one authority every call site
+// asks (item_is_effectful_callee, and the typed service-call predicate), with `.await?`, which is
+// E0728 in that frame (four sites on the approval broker's native closure, among them
+// gunbc.auth.approval_decision_store pending_escalations). The closure is the wrong link, not the
+// await: the enclosing fn is already async BECAUSE it reaches that effect, so the step belongs in
+// the enclosing frame, where both the await and the `?` mean what they mean everywhere else in the
+// fn -- the effect's refusal propagates out of the fold exactly as it would out of a let. That is
+// the same move the tail-call lowering makes for effectful recursion (a loop over a mutable state
+// slot in the caller's frame, not a recursive async call), applied to the Loop behavior fold
+// already is. The step's parameters are bound as lets per element in declaration order, so the
+// body is emitted by the ordinary expression emitter against the same lambda scope the closure
+// realization uses; nothing about the body is re-derived here.
+fn emit_rust_effectful_fold_loop(lambda_expr: Node, iter_str: String, init_str: String, acc_type_str: String, registry: Map<String, ItemInfo>, scope: InferScope, depth: Int, shared_types: Set<String>, emit_info: EmitGraphInfo) -> String {
+  match lambda_expr.expr_data {
+    ExprLambda =>
+      let si = scope.type_env.source_indices
+      let ps = lambda_param_names_at(texpr: lambda_expr, source_indices: si)
+      let pn = lambda_expr.children |> skip(1)
+      let acc_param_type = rust_fold_safe_acc_type(acc_type_str: acc_type_str)
+      let acc_name = match ps |> first { Present { value: n } => n  Absent => "_" }
+      let elem_name = match ps |> skip(1) |> first { Present { value: n } => n  Absent => "_" }
+      let lambda_scope = lambda_scope_from_children(scope: scope, params: ps, param_nodes: pn)
+      let body_str = emit_typed_expr(texpr: lambda_body(texpr: lambda_expr), registry: registry, scope: lambda_scope, depth: depth, shared_types: shared_types, emit_info: emit_info, fuel: 1024)
+      let acc_ident = if acc_name == "_" { "_" } else { emit_ident(name: acc_name, target: Rust) }
+      let elem_ident = if elem_name == "_" { "_" } else { emit_ident(name: elem_name, target: Rust) }
+      let acc_decl = if acc_param_type == "_" || acc_param_type == "" { "" } else { concat(": ", acc_param_type) }
+      let acc_unwrap = if acc_name != "_" && set_contains(emit_info.owned_bindings, acc_name) {
+        concat("let ", acc_ident, " = v1_rt::take_owned(", acc_ident, "); ")
+      } else { "" }
+      concat(
+        "{ let mut __fold_acc", acc_decl, " = ", init_str, "; ",
+        "for __fold_elem in ", iter_str, " { ",
+        "let ", acc_ident, " = __fold_acc; ", acc_unwrap,
+        "let ", elem_ident, " = __fold_elem; ",
+        "__fold_acc = { ", body_str, " }; ",
+        "} __fold_acc }"
+      )
+    _ => emit_rust_compile_error_expr(message: "effectful fold step is not a lambda: an effectful function value has no synchronous closure realization and no loop realization")
+  }
+}
+
+// DOES EMITTING THIS EXPRESSION PRODUCE AN `.await?` -- asked of the same two authorities that write
+// one: a source call whose callee item_is_effectful_callee, and a service op through a receiver
+// is_typed_service_call_receiver admits. Nested lambdas are walked too: an await anywhere inside the
+// step forces the step into the async frame, and a nested fold realizes its own loop by this rule.
+fn rust_expr_reaches_awaited_call(n: Node, registry: Map<String, ItemInfo>, source_indices: Map<String, NewlineIndex>) -> Bool {
+  let here = match n.expr_data {
+    ExprCall { call_semantics: cs, descent_evidence: _ } =>
+      match call_semantics_target(cs: cs) {
+        SourceDeclarationCall { owner_module_path: owner, decl_name: decl } =>
+          match lookup_item_by_identity(registry: registry, id: DeclaredCallableIdentity { owner_module_path: owner, decl_name: decl }) {
+            Present { value: info } => item_is_effectful_callee(info: info)
+            Absent => false
+          }
+        _ => false
+      }
+    ExprMethodCall { method_semantics: ms } =>
+      match ms {
+        Present { value: ServiceMethodSemantics { service_name: _, op_params: _ } } =>
+          is_typed_service_call_receiver(receiver: method_receiver(texpr: n), source_indices: source_indices)
+        _ => false
+      }
+    _ => false
+  }
+  here || n.children |> any(c => rust_expr_reaches_awaited_call(n: c, registry: registry, source_indices: source_indices))
 }
 
 fn fold_lambda_element_unused(lambda_expr: Node, source_indices: Map<String, NewlineIndex>) -> Bool {