diff --git a/README.md b/README.md
index 60775f2c..f4bc0dff 100644
--- a/README.md
+++ b/README.md
@@ -113,7 +113,19 @@ foreach ($futures as $future) {
 $pool->shutdown();
 ```
 
-A task is a named callable: a function name, `'Class::method'`, or `[$class, $method]`, defined by the bootstrap or built in. Arguments and results are copied between interpreters, so they must be null, bool, int, float, string, arrays of those, or objects of classes both sides define. Closures, generators, and handle-backed objects such as PDO connections are refused when submitted, with the offending path in the message.
+A task is a closure or a named callable: a function name, `'Class::method'`, or `[$class, $method]`, defined by the bootstrap or built in. Arguments and results are copied between interpreters, so they must be null, bool, int, float, string, arrays of those, or objects of classes both sides define. Generators and handle-backed objects such as PDO connections are refused when submitted, with the offending path in the message.
+
+A closure travels as its compiled code plus its captures: `use` variables, the variables an arrow function reads from the submitting scope, and `$this` when it has one. Each worker loads the code once and runs every later submit of the same closure against it. Captures follow the transfer rules above, and a closure that captures by reference is refused, since nothing can be shared between threads.
+
+```php
+$scale = 0.5;
+$thumbnails = [];
+foreach ($paths as $path) {
+    $thumbnails[] = $pool->submit(function (string $path) use ($scale) {
+        return resize($path, $scale);
+    }, [$path]);
+}
+```
 
 `await()` returns the result, or rethrows the task's exception as the same class when the caller has it. A queued task can be cancelled; a running one sees `Zphp\Task::cancelled()` and stops when it chooses, since nothing is ever killed. The queue is bounded: `submit()` blocks when it is full and `trySubmit()` returns null instead. `collect()` hands back completed futures in completion order, and `readiness()` is a stream that becomes readable when one is waiting, for use with `stream_select()`. `shutdown()` stops accepting work, cancels what is queued, and waits for running tasks; the pool's destructor does the same.
 
@@ -154,7 +166,6 @@ function resize_images(Zphp\Channel $jobs, Zphp\Channel $results): void
 
 `send()` blocks while the channel is full and `recv()` blocks while it is empty; both take an optional timeout in seconds and throw `Zphp\TimeoutException` when it passes. `trySend()` returns false instead of waiting. `close()` lets buffered values drain and then ends every `foreach`, while `send()` and `recv()` on a closed channel throw `Zphp\ChannelException`. Values follow the same transfer rules as task arguments, and a channel can carry other channels. A channel stays alive while any thread holds it or a value in flight names it.
 
-Submitting closures is planned.
 
 ## Related projects
 
diff --git a/src/pipeline/compiler.zig b/src/pipeline/compiler.zig
index 4da79c07..d1d37e0d 100644
--- a/src/pipeline/compiler.zig
+++ b/src/pipeline/compiler.zig
@@ -17,14 +17,25 @@ const compiler_class = @import("compiler_class.zig");
 const Allocator = std.mem.Allocator;
 const Error = Allocator.Error || error{CompileError};
 
-var global_closure_counter: u32 = 0;
+// closure and anonymous class names carry a process-wide id: worker threads
+// compile their own scripts and receive closures compiled on other threads,
+// so two compile units must never mint the same name
+var next_closure_id = std.atomic.Value(u32).init(0);
 
 pub fn closureCounter() u32 {
-    return global_closure_counter;
+    return next_closure_id.load(.monotonic);
 }
 
+// raises the counter past ids baked into loaded bytecode; never lowers it
 pub fn setClosureCounter(value: u32) void {
-    global_closure_counter = value;
+    var current = next_closure_id.load(.monotonic);
+    while (value > current) {
+        current = next_closure_id.cmpxchgWeak(current, value, .monotonic, .monotonic) orelse return;
+    }
+}
+
+pub fn allocClosureId() u32 {
+    return next_closure_id.fetchAdd(1, .monotonic);
 }
 
 pub const TypeHint = struct {
@@ -112,7 +123,10 @@ pub fn compileWithPath(ast: *const Ast, allocator: Allocator, file_path: []const
         .break_jumps = .{},
         .continue_jumps = .{},
         .file_path = file_path,
-        .closure_count = global_closure_counter,
+        // seeds the closures-so-far heuristic behind the locals-only frame
+        // fast path exactly as the old process counter did; names come from
+        // allocClosureId
+        .closure_count = closureCounter(),
     };
     errdefer {
         c.chunk.deinit(allocator);
@@ -166,7 +180,6 @@ pub fn compileWithPath(ast: *const Ast, allocator: Allocator, file_path: []const
     const slot_names = try c.buildSlotNames();
     const local_count = c.next_slot;
     c.local_slots.deinit(allocator);
-    global_closure_counter = c.closure_count;
     const strict = detectStrictTypes(ast.source);
     for (c.functions.items) |*f| f.strict_types = strict;
     return .{ .chunk = c.chunk, .functions = c.functions, .string_allocs = c.string_allocs, .allocator = allocator, .local_count = local_count, .slot_names = slot_names, .type_hints = c.type_hints, .function_attrs = c.function_attrs, .new_defaults = c.new_defaults, .deferred_exprs = c.deferred_exprs, .source = ast.source, .file_path = file_path, .strict_types = strict };
diff --git a/src/pipeline/compiler_class.zig b/src/pipeline/compiler_class.zig
index 478df633..0e4fb059 100644
--- a/src/pipeline/compiler_class.zig
+++ b/src/pipeline/compiler_class.zig
@@ -1,4 +1,5 @@
 const std = @import("std");
+const compiler = @import("compiler.zig");
 const Compiler = @import("compiler.zig").Compiler;
 const TypeHint = @import("compiler.zig").TypeHint;
 const FunctionAttrEntry = @import("compiler.zig").FunctionAttrEntry;
@@ -891,7 +892,7 @@ pub fn compileFunction(self: *Compiler, node: Ast.Node) Error!void {
 }
 
 pub fn compileClosure(self: *Compiler, node: Ast.Node) Error!void {
-    const id = self.closure_count;
+    const id = compiler.allocClosureId();
     self.closure_count += 1;
 
     var name_buf: [32]u8 = undefined;
@@ -1899,7 +1900,7 @@ pub fn compileClassDecl(self: *Compiler, node: Ast.Node) Error!void {
 }
 
 pub fn compileAnonymousClass(self: *Compiler, node: Ast.Node) Error!void {
-    const anon_name = try std.fmt.allocPrint(self.allocator, "class@anonymous_{d}", .{self.closure_count});
+    const anon_name = try std.fmt.allocPrint(self.allocator, "class@anonymous_{d}", .{compiler.allocClosureId()});
     try self.string_allocs.append(self.allocator, anon_name);
     self.closure_count += 1;
 
diff --git a/src/runtime/vm.zig b/src/runtime/vm.zig
index b85ef2bb..8ac7f6e2 100644
--- a/src/runtime/vm.zig
+++ b/src/runtime/vm.zig
@@ -534,6 +534,7 @@ pub const VM = struct {
     fiber_suspend_value: Value = .null,
     captures: std.ArrayListUnmanaged(CaptureEntry) = .{},
     capture_index: std.StringHashMapUnmanaged(CaptureRange) = .{},
+    interned_names: std.StringHashMapUnmanaged(void) = .{},
     cycle_closures: std.AutoArrayHashMapUnmanaged(*Value.String.Owner, i64) = .{},
     collecting_cycles: bool = false,
     closure_instance_count: u32 = 0,
@@ -2637,6 +2638,7 @@ pub const VM = struct {
         self.persistent_strings.deinit(self.allocator);
         self.captures.deinit(self.allocator);
         self.capture_index.deinit(self.allocator);
+        self.interned_names.deinit(self.allocator);
         self.free_closure_names.deinit(self.allocator);
         self.php_constants.deinit(self.allocator);
         self.user_constants.deinit(self.allocator);
@@ -2829,6 +2831,7 @@ pub const VM = struct {
         self.cycle_array_candidates.clearRetainingCapacity();
         self.captures.clearRetainingCapacity();
         self.capture_index.clearRetainingCapacity();
+        self.interned_names.clearRetainingCapacity();
         self.ob_stack.clearRetainingCapacity();
         self.request_vars.clearRetainingCapacity();
         self.response_code = 200;
@@ -13132,6 +13135,47 @@ pub const VM = struct {
         return .{ .string = .{ .ptr = name.ptr, .len = name.len, .owner = owner } };
     }
 
+    // a closure instance for a function this vm knows, with captures given
+    // up front the way closure_use adds them one call at a time; a closure
+    // that arrived from another thread is rebuilt through here
+    pub fn bindClosureInstance(self: *VM, compile_name: []const u8, names: []const []const u8, values: []const Value) !Value {
+        const func = self.functions.get(compile_name) orelse return error.RuntimeError;
+        const instance = try self.newClosureInstance(compile_name, func);
+        const name = instance.string.bytes();
+        for (names, values) |var_name, val| {
+            const kept = try self.internName(var_name);
+            retainValue(val);
+            try self.captures.append(self.allocator, .{ .closure_name = name, .var_name = kept, .value = val });
+            self.capture_index.getPtr(name).?.len += 1;
+        }
+        return instance;
+    }
+
+    // request-lifetime bytes for a name that recurs across calls, kept once
+    pub fn internName(self: *VM, name: []const u8) ![]const u8 {
+        if (self.interned_names.getKey(name)) |kept| return kept;
+        const kept = try self.allocator.dupe(u8, name);
+        errdefer self.allocator.free(kept);
+        try self.strings.append(self.allocator, kept);
+        try self.interned_names.put(self.allocator, kept, {});
+        return kept;
+    }
+
+    // a variable of the frame below the current one: what a native sees as
+    // its caller's scope. reference cells win, then the slot, then the map
+    pub fn callerVar(self: *VM, name: []const u8) ?Value {
+        if (self.frame_count < 2) return null;
+        const frame = &self.frames[self.frame_count - 2];
+        if (frame.ref_slots.get(name)) |cell| return cell.*;
+        const slot_names = if (frame.slot_names.len > 0) frame.slot_names else if (frame.func) |f| f.slot_names else self.global_slot_names;
+        for (slot_names, 0..) |sn, si| {
+            if (!std.mem.eql(u8, sn, name)) continue;
+            if (si < frame.locals.len and frame.locals[si] != .null) return frame.locals[si];
+            break;
+        }
+        return frame.vars.get(name);
+    }
+
     pub fn retainClosureByName(self: *VM, name: []const u8) void {
         const range = self.capture_index.getPtr(name) orelse return;
         if (range.owner) |owner| owner.refcount += 1;
diff --git a/src/stdlib/workers.zig b/src/stdlib/workers.zig
index a9f16a3d..88967ced 100644
--- a/src/stdlib/workers.zig
+++ b/src/stdlib/workers.zig
@@ -20,6 +20,9 @@ const network = @import("network.zig");
 const platform = @import("../platform.zig");
 const extension = @import("../extension.zig");
 const channel = @import("channel.zig");
+const bytecode_format = @import("../bytecode_format.zig");
+const CompileResult = @import("../pipeline/compiler.zig").CompileResult;
+const ObjFunction = @import("../pipeline/bytecode.zig").ObjFunction;
 
 const pool_class = "Zphp\\Pool";
 const future_class = "Zphp\\Future";
@@ -57,10 +60,26 @@ const Failure = struct {
     line: i64,
 };
 
+// a closure crossing threads: its compiled function, with every closure it
+// creates, as bytecode the receiving vm loads once per compile name, plus
+// its captures as a payload
+const ClosureTransfer = struct {
+    compile_name: []u8,
+    // owned by the pool's closure_code cache, which outlives every task
+    code: []const u8,
+    captures: Payload,
+
+    fn free(t: ClosureTransfer, a: std.mem.Allocator) void {
+        a.free(t.compile_name);
+        t.captures.free(a);
+    }
+};
+
 const Task = struct {
     id: u64,
     pool: *Pool,
     callable: []u8,
+    closure: ?ClosureTransfer = null,
     args: Payload,
     state: TaskState = .queued,
     result: ?Payload = null,
@@ -82,6 +101,7 @@ const Task = struct {
         const pool = t.pool;
         const a = pool.allocator;
         a.free(t.callable);
+        if (t.closure) |c| c.free(a);
         t.args.free(a);
         if (t.result) |r| r.free(a);
         if (t.failure) |f| {
@@ -194,6 +214,20 @@ const Worker = struct {
     pool: *Pool,
     index: usize,
     thread: ?std.Thread = null,
+    // closure code this worker's vm has loaded, by compile name
+    loaded: std.StringHashMapUnmanaged(*CompileResult) = .{},
+
+    // after the vm is gone: nothing references the chunks any more
+    fn freeLoaded(w: *Worker) void {
+        const a = w.pool.allocator;
+        var it = w.loaded.iterator();
+        while (it.next()) |entry| {
+            entry.value_ptr.*.deinit();
+            a.destroy(entry.value_ptr.*);
+            a.free(entry.key_ptr.*);
+        }
+        w.loaded.deinit(a);
+    }
 };
 
 const StartState = enum { starting, running, failed };
@@ -216,6 +250,8 @@ const Pool = struct {
     next_id: u64 = 1,
     wake: [2]std.posix.socket_t,
     readiness: ?*PhpObject = null,
+    // serialized closure code by compile name, built on first submit
+    closure_code: std.StringHashMapUnmanaged([]u8) = .{},
     // the php object plus every live task hold the pool; a future can outlive
     // the pool that made it
     refs: std.atomic.Value(u32) = std.atomic.Value(u32).init(1),
@@ -413,6 +449,12 @@ const Pool = struct {
     }
 
     fn free(pool: *Pool) void {
+        var codes = pool.closure_code.iterator();
+        while (codes.next()) |entry| {
+            pool.allocator.free(entry.key_ptr.*);
+            pool.allocator.free(entry.value_ptr.*);
+        }
+        pool.closure_code.deinit(pool.allocator);
         pool.completed.deinit(pool.allocator);
         if (pool.readiness == null) platform.closeSocket(platform.socketToInt(pool.wake[0]));
         platform.closeSocket(platform.socketToInt(pool.wake[1]));
@@ -446,6 +488,7 @@ fn workerMain(w: *Worker) void {
         r.deinit();
         pool.allocator.destroy(r);
     };
+    defer w.freeLoaded();
     defer {
         vm.deinit();
         pool.allocator.destroy(vm);
@@ -456,7 +499,7 @@ fn workerMain(w: *Worker) void {
         boot_result = bootstrap(vm, pool, path) orelse return;
     }
     pool.reportStart(null);
-    while (pool.queue.pop()) |task| runTask(vm, task);
+    while (pool.queue.pop()) |task| runTask(w, vm, task);
 }
 
 fn bootstrap(vm: *VM, pool: *Pool, path: []const u8) ?*@import("../pipeline/compiler.zig").CompileResult {
@@ -499,7 +542,7 @@ fn flushOutput(vm: *VM) void {
     vm.output.clearRetainingCapacity();
 }
 
-fn runTask(vm: *VM, task: *Task) void {
+fn runTask(w: *Worker, vm: *VM, task: *Task) void {
     const pool = task.pool;
     task.mutex.lock();
     if (task.state == .cancelled) {
@@ -517,7 +560,7 @@ fn runTask(vm: *VM, task: *Task) void {
     defer current_task = null;
     extension.beginRequest(vm) catch {};
     var ctx = vm.makeContext(task_class);
-    execute(&ctx, task);
+    execute(&ctx, w, task);
     extension.endRequest(vm);
     vm.pending_exception = null;
     vm.error_msg = null;
@@ -530,8 +573,14 @@ fn runTask(vm: *VM, task: *Task) void {
     pool.complete(task);
 }
 
-fn execute(ctx: *NativeContext, task: *Task) void {
-    const callable = hold(serialize.unserializeFromString(ctx, task.callable) orelse return settleFatal(task, "the callable did not transfer"));
+fn execute(ctx: *NativeContext, w: *Worker, task: *Task) void {
+    const callable: Value = if (task.closure) |*transfer|
+        materializeClosure(ctx, w, transfer) catch {
+            settleFailure(ctx.vm, task, .null);
+            return;
+        }
+    else
+        hold(serialize.unserializeFromString(ctx, task.callable) orelse return settleFatal(task, "the callable did not transfer"));
     const args_payload = task.args;
     task.args = Payload.empty;
     const args_value = hold(unpack(ctx, args_payload, task.pool.allocator) orelse return settleFatal(task, "the arguments did not transfer"));
@@ -729,6 +778,154 @@ pub fn unpack(ctx: *NativeContext, payload: Payload, allocator: std.mem.Allocato
     return serialize.unserializeFromString(ctx, payload.bytes);
 }
 
+// ---------------------------------------------------------------------------
+// closures: the function travels as bytecode, the captures as values
+
+fn isCompileName(name: []const u8) bool {
+    if (!std.mem.startsWith(u8, name, "__closure_")) return false;
+    const rest = name["__closure_".len..];
+    if (rest.len == 0) return false;
+    for (rest) |c| if (c < '0' or c > '9') return false;
+    return true;
+}
+
+// every closure function the body can create, transitively, by the compile
+// names its chunks hold as constants
+fn collectClosureFunctions(ctx: *NativeContext, root: *const ObjFunction, out: *std.ArrayListUnmanaged(ObjFunction)) RuntimeError!void {
+    var seen: std.StringHashMapUnmanaged(void) = .{};
+    defer seen.deinit(ctx.allocator);
+    var queue: std.ArrayListUnmanaged(*const ObjFunction) = .{};
+    defer queue.deinit(ctx.allocator);
+    try queue.append(ctx.allocator, root);
+    try seen.put(ctx.allocator, root.name, {});
+    while (queue.items.len > 0) {
+        const func = queue.pop().?;
+        try out.append(ctx.allocator, func.*);
+        for (func.chunk.constants.items) |constant| {
+            if (constant != .string) continue;
+            const name = constant.string.bytes();
+            if (!isCompileName(name) or seen.contains(name)) continue;
+            const nested = ctx.vm.functions.get(name) orelse continue;
+            try seen.put(ctx.allocator, name, {});
+            try queue.append(ctx.allocator, nested);
+        }
+    }
+}
+
+fn serializeClosureCode(ctx: *NativeContext, func: *const ObjFunction, allocator: std.mem.Allocator) RuntimeError![]u8 {
+    const origin = ctx.vm.chunk_to_result.get(@intFromPtr(&func.chunk)) orelse return throwNamed(ctx, transfer_exception, "the closure has no compiled unit to travel with (at callable)", .{});
+    var functions: std.ArrayListUnmanaged(ObjFunction) = .{};
+    defer functions.deinit(ctx.allocator);
+    try collectClosureFunctions(ctx, func, &functions);
+    var unit = CompileResult{
+        .chunk = .{},
+        .functions = functions,
+        .string_allocs = .{},
+        .allocator = ctx.allocator,
+        .new_defaults = origin.new_defaults,
+        .deferred_exprs = origin.deferred_exprs,
+        .source = origin.source,
+        .file_path = origin.file_path,
+        .strict_types = origin.strict_types,
+    };
+    defer unit.type_hints.deinit(ctx.allocator);
+    defer unit.function_attrs.deinit(ctx.allocator);
+    for (functions.items) |f| {
+        for (origin.type_hints.items) |th| if (std.mem.eql(u8, th.name, f.name)) try unit.type_hints.append(ctx.allocator, th);
+        for (origin.function_attrs.items) |fa| if (std.mem.eql(u8, fa.name, f.name)) try unit.function_attrs.append(ctx.allocator, fa);
+    }
+    const bytes = bytecode_format.serialize(ctx.allocator, &unit) catch return throwNamed(ctx, transfer_exception, "the closure could not be serialized (at callable)", .{});
+    defer ctx.allocator.free(bytes);
+    return allocator.dupe(u8, bytes);
+}
+
+fn closureCode(ctx: *NativeContext, pool: *Pool, func: *const ObjFunction) RuntimeError![]const u8 {
+    pool.mutex.lock();
+    const cached = pool.closure_code.get(func.name);
+    pool.mutex.unlock();
+    if (cached) |code| return code;
+    const code = try serializeClosureCode(ctx, func, pool.allocator);
+    errdefer pool.allocator.free(code);
+    const key = try pool.allocator.dupe(u8, func.name);
+    errdefer pool.allocator.free(key);
+    pool.mutex.lock();
+    defer pool.mutex.unlock();
+    try pool.closure_code.put(pool.allocator, key, code);
+    return code;
+}
+
+fn arrayHasKey(arr: *PhpArray, key: []const u8) bool {
+    return arr.get(.{ .string = Value.String.borrowed(key) }) != .null;
+}
+
+// the captures as a php array keyed by variable name: use variables, $this,
+// the scope markers, and for an arrow function the caller's variables its
+// body names, since the vm resolves those at call time from a frame that
+// will not exist on the other thread
+fn captureArray(ctx: *NativeContext, instance: []const u8, func: *const ObjFunction) RuntimeError!*PhpArray {
+    const arr = try ctx.createArray();
+    VM.arrayRetain(arr);
+    errdefer ctx.vm.releaseValue(.{ .array = arr });
+    if (ctx.vm.getCaptureRange(instance)) |range| {
+        if (range.has_refs) return throwNamed(ctx, transfer_exception, "a closure capturing by reference cannot be transferred between threads (at callable)", .{});
+        for (ctx.vm.captures.items[range.start .. range.start + range.len]) |cap| {
+            try arr.set(ctx.allocator, .{ .string = Value.String.borrowed(cap.var_name) }, cap.value);
+        }
+    }
+    if (func.is_arrow) {
+        for (func.slot_names) |name| {
+            var is_param = false;
+            for (func.params) |p| if (std.mem.eql(u8, p, name)) {
+                is_param = true;
+            };
+            if (is_param or arrayHasKey(arr, name)) continue;
+            const v = ctx.vm.callerVar(name) orelse continue;
+            try arr.set(ctx.allocator, .{ .string = Value.String.borrowed(name) }, v);
+        }
+    }
+    return arr;
+}
+
+fn packClosure(ctx: *NativeContext, pool: *Pool, instance: []const u8) RuntimeError!ClosureTransfer {
+    const func = ctx.vm.functions.get(instance) orelse return throwNamed(ctx, transfer_exception, "the closure is not callable (at callable)", .{});
+    const code = try closureCode(ctx, pool, func);
+    const arr = try captureArray(ctx, instance, func);
+    defer ctx.vm.releaseValue(.{ .array = arr });
+    const captures = try pack(ctx, .{ .array = arr }, "closure", pool.allocator);
+    errdefer captures.free(pool.allocator);
+    const compile_name = try pool.allocator.dupe(u8, func.name);
+    return .{ .compile_name = compile_name, .code = code, .captures = captures };
+}
+
+// loads the code once per worker, then binds a fresh instance with the
+// captures; the instance carries one reference the caller releases
+fn materializeClosure(ctx: *NativeContext, w: *Worker, transfer: *ClosureTransfer) RuntimeError!Value {
+    const pool = w.pool;
+    if (!w.loaded.contains(transfer.compile_name)) {
+        const result = try pool.allocator.create(CompileResult);
+        errdefer pool.allocator.destroy(result);
+        result.* = bytecode_format.deserialize(pool.allocator, transfer.code) catch return throwNamed(ctx, transfer_exception, "the closure code did not load in the worker", .{});
+        errdefer result.deinit();
+        try ctx.vm.registerResultFunctions(result);
+        try w.loaded.put(pool.allocator, try pool.allocator.dupe(u8, transfer.compile_name), result);
+    }
+    const payload = transfer.captures;
+    transfer.captures = Payload.empty;
+    const caps = hold(unpack(ctx, payload, pool.allocator) orelse return throwNamed(ctx, transfer_exception, "the closure captures did not transfer", .{}));
+    defer ctx.vm.releaseValue(caps);
+    if (caps != .array) return throwNamed(ctx, transfer_exception, "the closure captures did not transfer", .{});
+    var names: std.ArrayListUnmanaged([]const u8) = .{};
+    defer names.deinit(ctx.allocator);
+    var values: std.ArrayListUnmanaged(Value) = .{};
+    defer values.deinit(ctx.allocator);
+    for (caps.array.entries.items) |entry| {
+        if (entry.key != .string) continue;
+        try names.append(ctx.allocator, entry.key.string.bytes());
+        try values.append(ctx.allocator, if (entry.ref) |cell| cell.* else entry.value);
+    }
+    return ctx.vm.bindClosureInstance(transfer.compile_name, names.items, values.items) catch return throwNamed(ctx, transfer_exception, "the closure could not be bound in the worker", .{});
+}
+
 // ---------------------------------------------------------------------------
 // php surface
 
@@ -802,12 +999,13 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
     flushOutput(ctx.vm);
     if (args.len < 1) return throwNamed(ctx, pool_exception, "submit() needs a callable", .{});
     const callable = args[0];
-    const valid = switch (callable) {
-        .string => |s| !std.mem.startsWith(u8, s.bytes(), "__closure"),
+    const is_closure = callable == .string and std.mem.startsWith(u8, callable.string.bytes(), "__closure_");
+    const valid = is_closure or switch (callable) {
+        .string => true,
         .array => |arr| arr.entries.items.len == 2 and arr.entries.items[0].value == .string and arr.entries.items[1].value == .string,
         else => false,
     };
-    if (!valid) return throwNamed(ctx, transfer_exception, "tasks are named callables: a function name, 'Class::method', or [class, method]", .{});
+    if (!valid) return throwNamed(ctx, transfer_exception, "tasks are closures or named callables: a function name, 'Class::method', or [class, method]", .{});
     const task_args: Value = if (args.len >= 2) args[1] else .{ .array = try ctx.createArray() };
     if (task_args != .array) return throwNamed(ctx, pool_exception, "arguments must be an array", .{});
     const packed_args = try pack(ctx, task_args, "args", pool.allocator);
@@ -822,23 +1020,22 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
     task.* = .{ .id = pool.nextId(), .pool = pool, .callable = &.{}, .args = packed_args };
     pool.retain();
     errdefer pool.release();
-    task.callable = try serializedCopy(ctx, callable, pool.allocator);
+    if (is_closure) {
+        task.closure = try packClosure(ctx, pool, callable.string.bytes());
+    } else {
+        task.callable = try serializedCopy(ctx, callable, pool.allocator);
+    }
     errdefer pool.allocator.free(task.callable);
+    errdefer if (task.closure) |c| c.free(pool.allocator);
 
     switch (pool.queue.push(task, block)) {
         .ok => {},
         .full => {
-            pool.allocator.free(task.callable);
-            packed_args.free(pool.allocator);
-            pool.allocator.destroy(task);
-            pool.release();
+            discardTask(task);
             return NativeResult.scalar(.null);
         },
         .closed => {
-            pool.allocator.free(task.callable);
-            packed_args.free(pool.allocator);
-            pool.allocator.destroy(task);
-            pool.release();
+            discardTask(task);
             return throwNamed(ctx, pool_exception, "the pool is shutting down", .{});
         },
     }
@@ -848,6 +1045,16 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
     return NativeResult.borrowed(.{ .object = future });
 }
 
+// a task the queue refused, never seen by a worker or a future
+fn discardTask(task: *Task) void {
+    const pool = task.pool;
+    pool.allocator.free(task.callable);
+    if (task.closure) |c| c.free(pool.allocator);
+    task.args.free(pool.allocator);
+    pool.allocator.destroy(task);
+    pool.release();
+}
+
 fn poolSubmit(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
     return submitTask(ctx, args, true);
 }
diff --git a/tests/workers/basic.expected b/tests/workers/basic.expected
index 504ddf5a..63e08f1b 100644
--- a/tests/workers/basic.expected
+++ b/tests/workers/basic.expected
@@ -4,7 +4,7 @@ int(6)
 int(9)
 InvalidArgumentException: bad input 7
 Error: Call to undefined function missing_fn()
-transfer: tasks are named callables: a function name, 'Class::method', or [class, method]
+int(1)
 transfer: an object backed by a native handle cannot be transferred between threads (at args[0])
 int(2470)
 timeout
diff --git a/tests/workers/basic.php b/tests/workers/basic.php
index 51f95729..40a8f029 100644
--- a/tests/workers/basic.php
+++ b/tests/workers/basic.php
@@ -7,7 +7,7 @@
 var_dump($pool->submit(['Jobs', 'sum'], [[4, 5]])->await());
 try { $pool->submit('boom', ['bad input'])->await(); } catch (InvalidArgumentException $e) { echo get_class($e), ": ", $e->getMessage(), " ", $e->getCode(), "\n"; }
 try { $pool->submit('missing_fn')->await(); } catch (Throwable $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
-try { $pool->submit(fn() => 1); } catch (Zphp\TransferException $e) { echo "transfer: ", $e->getMessage(), "\n"; }
+var_dump($pool->submit(fn() => 1)->await());
 try { $pool->submit('square', [new PDO('sqlite::memory:')]); } catch (Zphp\TransferException $e) { echo "transfer: ", $e->getMessage(), "\n"; }
 $futures = [];
 for ($i = 0; $i < 20; $i++) $futures[] = $pool->submit('square', [$i]);
diff --git a/tests/workers/closures.expected b/tests/workers/closures.expected
new file mode 100644
index 00000000..d2217b8e
--- /dev/null
+++ b/tests/workers/closures.expected
@@ -0,0 +1,16 @@
+int(42)
+int(104)
+bool(true)
+string(14) "15 / secret 10"
+int(44)
+array(3) {
+}
+int(8575)
+DomainException: inside 3
+coerced
+refused: a closure capturing by reference cannot be transferred between threads (at callable)
+result: a Closure cannot be transferred between threads (at result)
+capture: an object backed by a native handle cannot be transferred between threads (at closure[$pdo])
+args: a Closure cannot be transferred between threads (at args[0])
+int(42)
+end
diff --git a/tests/workers/closures.php b/tests/workers/closures.php
new file mode 100644
index 00000000..833a2d27
--- /dev/null
+++ b/tests/workers/closures.php
@@ -0,0 +1,40 @@
+<?php
+require __DIR__ . '/worker.php';
+$pool = new Zphp\Pool(2, __DIR__ . '/worker.php');
+
+// use captures by value, arrow functions capturing the caller's scope, static closures
+$factor = 7;
+var_dump($pool->submit(function (int $x) use ($factor) { return $x * $factor; }, [6])->await());
+$offset = 100;
+var_dump($pool->submit(fn(int $x) => $x + $offset + helper(1), [1])->await());
+var_dump($pool->submit(static fn() => Zphp\Task::worker() >= 0)->await());
+
+// $this and scope travel with the closure
+$c = new Counter(10);
+$bound = (function (int $n) { return $this->bump($n) . " / " . $this->secret(); })->bindTo($c, Counter::class);
+var_dump($pool->submit($bound, [5])->await());
+var_dump($pool->submit((new Maker(4))->job(), [11])->await());
+
+// closures that create closures
+$nested = function (array $xs) { $sq = fn($v) => $v * $v; return array_map($sq, array_map(function ($v) { return $v + 1; }, $xs)); };
+var_dump($pool->submit($nested, [[1, 2, 3]])->await());
+
+// the same closure submitted many times loads once per worker
+$futures = []; for ($i = 0; $i < 50; $i++) $futures[] = $pool->submit(function (int $i) use ($factor) { return $i * $factor; }, [$i]);
+$sum = 0; foreach ($futures as $f) $sum += $f->await(); var_dump($sum);
+
+// exceptions, type coercion, and refusals
+try { $pool->submit(function () { throw new DomainException("inside", 3); })->await(); } catch (DomainException $e) { echo get_class($e), ": ", $e->getMessage(), " ", $e->getCode(), "\n"; }
+try { $pool->submit(function (string $s) { return $s; }, [5])->await(); echo "coerced\n"; } catch (TypeError $e) { echo "TypeError\n"; }
+$byref = 1; try { $pool->submit(function () use (&$byref) { return $byref; }); } catch (Zphp\TransferException $e) { echo "refused: ", $e->getMessage(), "\n"; }
+try { $pool->submit(function () { return fn() => 1; })->await(); } catch (Zphp\TransferException $e) { echo "result: ", $e->getMessage(), "\n"; }
+$pdo = new PDO('sqlite::memory:'); try { $pool->submit(function () use ($pdo) { return 1; }); } catch (Zphp\TransferException $e) { echo "capture: ", $e->getMessage(), "\n"; }
+try { $pool->submit(function () { return 1; }, [fn() => 2]); } catch (Zphp\TransferException $e) { echo "args: ", $e->getMessage(), "\n"; }
+
+// a closure consuming a channel
+$ch = new Zphp\Channel(4);
+$w = $pool->submit(function (Zphp\Channel $ch) use ($factor) { $n = 0; foreach ($ch as $v) $n += $v * $factor; return $n; }, [$ch]);
+foreach ([1, 2, 3] as $v) $ch->send($v); $ch->close(); var_dump($w->await());
+
+$pool->shutdown();
+echo "end\n";
diff --git a/tests/workers/memory.php b/tests/workers/memory.php
index dd526139..4e83bacb 100644
--- a/tests/workers/memory.php
+++ b/tests/workers/memory.php
@@ -19,6 +19,8 @@ function big(int $n): array { return array_fill(0, $n, str_repeat("x", 64)); }
     $f->await();
 }
 for ($i = 0; $i < 2000; $i++) { $carrier = new Zphp\Channel(1); $carrier->send(['inner' => new Zphp\Channel(1)]); $carrier->recv()['inner']->trySend("x"); }
+$scale = 3;
+for ($i = 0; $i < 3000; $i++) { $pool->submit(function (int $n) use ($scale) { return big($n * $scale); }, [30])->await(); }
 $growth = memory_get_usage() - $before;
 echo $growth < 16 * 1024 * 1024 ? "memory bounded\n" : "memory grew by $growth\n";
 exit($growth < 16 * 1024 * 1024 ? 0 : 1);
diff --git a/tests/workers/run b/tests/workers/run
index d5f35449..93d818da 100755
--- a/tests/workers/run
+++ b/tests/workers/run
@@ -19,7 +19,7 @@ check() {
     fi
 }
 
-for script in basic edge channel; do
+for script in basic edge channel closures; do
     actual="$("$ZPHP" run "$SCRIPT_DIR/$script.php" 2>&1 | tr -d '\r' | grep -v '^error(gpa)' | grep -v '^\s' | grep -v '^/')" || true
     check "$script.php" "$(tr -d '\r' < "$SCRIPT_DIR/$script.expected")" "$actual"
 done
diff --git a/tests/workers/worker.php b/tests/workers/worker.php
index f3032925..4f785845 100644
--- a/tests/workers/worker.php
+++ b/tests/workers/worker.php
@@ -30,3 +30,10 @@ function slow_drain(Zphp\Channel $ch): int { $n = 0; foreach ($ch as $v) { uslee
 function forward(Zphp\Channel $in, Zphp\Channel $out): void { foreach ($in as $v) $out->send($v); $out->close(); }
 function bad_result(): Closure { return fn() => 1; }
 function produce_big(Zphp\Channel $out, int $n): void { for ($i = 0; $i < $n; $i++) $out->send(big(100)); }
+class Counter {
+    public function __construct(public int $base = 0) {}
+    public function bump(int $n): int { return $this->base + $n; }
+    private function secret(): string { return "secret " . $this->base; }
+}
+class Maker { public function __construct(private int $k) {} public function job(): Closure { return function (int $x) { return $x * $this->k; }; } }
+function helper(int $x): int { return $x * 3; }