feat: transfer owned captures into structured tasks

HANDOFF.md

@@ -2,6 +2,15 @@
 
 ## Current direction
 
+The owner's 2026-11-10 request selects R41's admitted-file startup witness.
+[Plan259](docs/impl/259-task-owned-captures.md) supplies owned captures in a
+literal `spawn` and file-bearing results, allowing the same opened file and
+bounded window to move through worker reads. Shared views remain read-only;
+ordinary stored/pipeline lambdas remain Copy-only. Provider owners cover file
+identity, source/alias invalidation, cleanup, imports and cache replay. LLM
+integration, native full-answer comparison and startup measurements remain
+consumer-owned; media and voice require no adoption in this provider change.
+
 The owner's 2026-10-30 request check selects R109's concrete byte-access guard
 residual. [Plan257](docs/impl/257-byte-access-guard-fusion.md) fuses fixed-width
 read/write guards and updates both proof consumers together. Current LLM module

crates/align_driver/src/lib.rs

@@ -7585,7 +7585,7 @@ fn referenced_program_calls(
                 Rvalue::FnAddr { target, .. } => {
                     targets.insert(target.clone());
                 }
-                Rvalue::Closure { lifted, .. } => {
+                Rvalue::Closure { lifted, .. } | Rvalue::SpawnTask { lifted, .. } => {
                     targets.insert(lifted.clone());
                 }
                 Rvalue::ParMapParallel { func, stages, .. } => {

crates/align_mir/src/lib.rs

@@ -1107,13 +1107,12 @@ pub enum Rvalue {
     ArenaBegin,
     /// Open a `task_group`; the value is its handle (a `*TaskGroup`).
     TgBegin,
-    /// Publish a task (`spawn`): snapshot the closure's captures into a fresh env in the
-    /// task-group region, allocate the result slot there, and register the task. Yields the slot
-    /// pointer (the `Task<R>` handle). `tg` = the task-group handle, `closure` = the `{fn, env}`
-    /// value, `capture_tys` give the env layout (empty = non-capturing), `r` = the result scalar.
+    /// Transfer captures directly into a fresh task-group environment and publish one invocation.
+    /// No reusable Fn value carries owned inputs. Yields the result slot (`Task<R>` handle).
     SpawnTask {
         tg: Operand,
-        closure: Operand,
+        lifted: ProgramCall,
+        captures: Vec<Operand>,
         capture_tys: Vec<Ty>,
         r: Ty,
         fallible: bool,
@@ -10625,24 +10624,31 @@ fn lower_expr_recursive(b: &mut Builder, e: &hir::Expr) -> Operand {
                     unreachable!("spawn result is a Task")
                 };
                 let r_ty = align_sema::scalar_to_ty(s);
-                let capture_tys: Vec<Ty> = match &closure.kind {
-                    hir::ExprKind::Closure { captures, .. } => {
-                        captures.iter().map(|c| c.ty).collect()
+                let (lifted, captures) = match &closure.kind {
+                    hir::ExprKind::Closure { lifted, captures } => (lifted.as_str(), captures.as_slice()),
+                    hir::ExprKind::FnValue(lifted) => (lifted.as_str(), &[][..]),
+                    _ => {
+                        b.terminate(Term::Unreachable);
+                        return Operand::Const(Const::Unit);
                     }
-                    _ => Vec::new(),
                 };
-                lower_required_binding!(
-                    b,
-                    clos = lower_expr(b, closure),
-                    Operand::Const(Const::Unit)
-                );
+                let capture_tys = captures.iter().map(|capture| capture.ty).collect();
+                let mut operands = Vec::with_capacity(captures.len());
+                for capture in captures {
+                    operands.push(lower_required!(b, lower_expr(b, capture), Operand::Const(Const::Unit)));
+                }
+                // Snapshot every input before retiring any source; publication may start a worker.
+                for capture in captures {
+                    null_moved_source(b, capture);
+                }
                 let tg = Operand::Value(*b.task_groups.last().expect("spawn outside a task_group"));
                 let v = b.fresh_value(e.ty);
                 b.push(Stmt::Let(
                     v,
                     Rvalue::SpawnTask {
                         tg,
-                        closure: clos,
+                        lifted: ProgramCall::from_validated(lifted),
+                        captures: operands,
                         capture_tys,
                         r: r_ty,
                         fallible: *fallible,

crates/align_mir/src/print.rs

@@ -403,8 +403,9 @@ fn rvalue_str(rv: &Rvalue) -> String {
         }
         Rvalue::ArenaBegin => "arena_begin".to_string(),
         Rvalue::TgBegin => "tg_begin".to_string(),
-        Rvalue::SpawnTask { closure, fallible, .. } => {
-            format!("spawn_task{} {}", if *fallible { " fallible" } else { "" }, operand_str(closure))
+        Rvalue::SpawnTask { lifted, captures, fallible, .. } => {
+            format!("spawn_task{} {}({})", if *fallible { " fallible" } else { "" }, lifted,
+                captures.iter().map(operand_str).collect::<Vec<_>>().join(", "))
         }
         Rvalue::TgWaitResult { tg, fallible } => {
             format!("tg_wait{} {}", if *fallible { " fallible" } else { "" }, operand_str(tg))

crates/align_mir/src/producer.rs

@@ -8701,7 +8701,7 @@ fn xml_local_call_components(program: &Program) -> Result<Vec<Vec<usize>>, Produ
                     Some(target)
                 }
                 Rvalue::CallWithCleanup(call) => Some(&call.target),
-                Rvalue::Closure { lifted, .. } => Some(lifted),
+                Rvalue::Closure { lifted, .. } | Rvalue::SpawnTask { lifted, .. } => Some(lifted),
                 _ => None,
             };
             if let Some(target) = target.and_then(|target| indices.get(target).copied()) {
@@ -9253,7 +9253,7 @@ fn builder_boolean_slots(function: &Function) -> Vec<bool> {
                     crate::loop_facts::DropStateCall::None => {}
                 }
                 if let Rvalue::Use(operand) = value { address(&mut eligible, operand); }
-                if let Rvalue::Closure { captures, .. } = value {
+                if let Rvalue::Closure { captures, .. } | Rvalue::SpawnTask { captures, .. } = value {
                     for capture in captures { address(&mut eligible, capture); }
                 }
             }
@@ -10206,6 +10206,16 @@ fn validate_resource_rvalues_component(
                         })
                 };
                 let call_arguments_valid = match rvalue {
+                    Rvalue::SpawnTask { lifted, captures, capture_tys, .. } => {
+                        // Direct task inputs are ordinary owning parameters. A matching type
+                        // alone cannot turn a shared borrowed origin into a transferable owner.
+                        xml_direct_call_facts(program, lifted, local_contracts).is_some_and(|facts| {
+                            facts.params == *capture_tys
+                                && facts.modes.iter().all(|mode| *mode == align_ast::ParamMode::ByValue)
+                                && operands_match_modes(captures, &facts.modes, &facts.params, program)
+                                && xml_call_arguments_valid(captures, &facts.params, &facts.modes, *value)
+                        })
+                    }
                     Rvalue::Call(DirectCall::Program(target), args) => {
                         xml_direct_call_facts(program, target, local_contracts).map_or_else(
                             || {
@@ -10325,6 +10335,9 @@ fn validate_resource_rvalues_component(
                 };
                 if !call_arguments_valid {
                     let detail = match rvalue {
+                        Rvalue::SpawnTask { lifted, .. } => {
+                            format!("owned-leaf argument provenance mismatch at task '{lifted}'")
+                        }
                         Rvalue::Call(DirectCall::Program(target), _) => {
                             format!("owned-leaf argument provenance mismatch at call to '{target}'")
                         }

crates/align_mir/src/validate_hir.rs

@@ -13283,7 +13283,13 @@ impl<'a> BodyValidator<'a> {
             .zip(captures)
         {
             let Some(flow) = self.expr_flow(capture) else { return false };
-            if !self.body_ty_matches(flow.ty, *expected) || !self.ty_copy_ok(flow.ty, context) {
+            let owned = spawn.is_some() && align_sema::task_owned_capture_type(
+                flow.ty, &self.program.structs, &self.program.tuples,
+                &self.program.enums, &self.program.tagged_types,
+            );
+            if !self.body_ty_matches(flow.ty, *expected)
+                || !(self.ty_copy_ok(flow.ty, context) || (owned && self.body_ty_ok(flow.ty)))
+            {
                 return false;
             }
         }

crates/align_mir/src/validate_hir_tests.rs

@@ -21940,6 +21940,46 @@ fn deflate_codec_hir_signature_matrix() -> Result<(), String> {
     Ok(())
 }
 
+#[test]
+fn task_owned_capture_hir_consumption_is_replayed() -> Result<(), String> {
+    let base = checked_source_program("fn main(){owner:=\"owned\".clone();task_group{task:=spawn(fn{owner});wait();task.get();()};()}\n");
+    for per_unit in [false, true] {
+        lower_program_checked(&base, per_unit, None).map_err(|error| format!("{error:?}"))?;
+    }
+    for mutation in 0..4 {
+        let mut bad = base.clone();
+        let main = bad.fns.iter_mut().find(|f| f.name == "main").ok_or("main")?;
+        let group = main.body.stmts.iter_mut().find_map(|stmt| match stmt {
+            hir::Stmt::Expr(hir::Expr { kind: hir::ExprKind::TaskGroup(group), .. }) => Some(group),
+            _ => None,
+        }).ok_or("group")?;
+        let spawn = group.stmts.iter().find_map(|stmt| match stmt {
+            hir::Stmt::Let { init, .. } if matches!(init.kind, hir::ExprKind::Spawn { .. }) => Some(init.clone()),
+            _ => None,
+        }).ok_or("spawn")?;
+        let hir::ExprKind::Spawn { closure, .. } = &spawn.kind else { return Err("spawn kind".into()) };
+        let hir::ExprKind::Closure { captures, .. } = &closure.kind else { return Err("closure".into()) };
+        match mutation {
+            0 => group.stmts.insert(1, hir::Stmt::Expr(captures.first().ok_or("capture")?.clone())),
+            1 => group.stmts.insert(1, hir::Stmt::Expr(spawn)),
+            2 => group.stmts.insert(0, hir::Stmt::Expr(*closure.clone())),
+            _ => {
+                let statement = group.stmts.iter_mut().find_map(|stmt| match stmt {
+                    hir::Stmt::Let { init, .. } if matches!(init.kind, hir::ExprKind::Spawn { .. }) => Some(init),
+                    _ => None,
+                }).ok_or("spawn mutation")?;
+                let hir::ExprKind::Spawn { closure, .. } = &mut statement.kind else { return Err("spawn kind".into()) };
+                let hir::ExprKind::Closure { captures, .. } = &mut closure.kind else { return Err("closure".into()) };
+                captures.first_mut().ok_or("capture")?.ty = Ty::Bool;
+            }
+        }
+        for per_unit in [false, true] {
+            assert!(lower_program_checked(&bad, per_unit, None).is_err(), "{mutation}/{per_unit}");
+        }
+    }
+    Ok(())
+}
+
 #[test]
 fn task_owned_hir_wait_move_and_payload_replay() -> Result<(), String> {
     let valid=checked_source_program("fn main() {task_group {t:=spawn(fn {\"owned\".clone()});wait();t.get();()};()}\n");

crates/align_sema/src/callback_access.rs

@@ -940,6 +940,12 @@ impl MoveCheck<'_> {
         self.check_callback_requirements(lifted, captures, closure.span);
         let mut roots = BorrowRoots::new();
         let mut map = |root: &hir::MutableAccessRoot| -> BorrowRoots {
+            let (hir::MutableAccessRoot::Contained(index) | hir::MutableAccessRoot::Storage(index)) = root;
+            if captures.get(*index as usize).is_some_and(|capture| super::task_owned_capture_type(
+                capture.ty, self.structs, self.tuples, self.enums, self.tagged_types,
+            )) {
+                return BorrowRoots::new();
+            }
             match root {
                 hir::MutableAccessRoot::Contained(i) => captures
                     .get(*i as usize)