diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 431cec13..32ae5a49 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -170,6 +170,7 @@ jobs:
       - run: zig build -Doptimize=ReleaseFast
       - run: python3 ./tests/memory_soak
       - name: worker pool retention
+        timeout-minutes: 5
         run: ./zig-out/bin/zphp run tests/workers/memory.php
 
   php-compat:
diff --git a/README.md b/README.md
index 6be2c2cb..60775f2c 100644
--- a/README.md
+++ b/README.md
@@ -117,7 +117,44 @@ A task is a named callable: a function name, `'Class::method'`, or `[$class, $me
 
 `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.
 
-Channels and submitting closures are planned.
+### Channels
+
+`Zphp\Channel` is a bounded queue that workers and the main thread share. A channel passed to a task binds to the same queue on the other side, so a producer and its consumers can run on different threads without sharing PHP memory.
+
+```php
+$jobs = new Zphp\Channel(capacity: 16);
+$results = new Zphp\Channel(capacity: 256);
+
+$consumers = [];
+for ($i = 0; $i < 4; $i++) {
+    $consumers[] = $pool->submit('resize_images', [$jobs, $results]);
+}
+foreach (glob('uploads/*.jpg') as $path) {
+    $jobs->send($path);
+}
+$jobs->close();
+foreach ($consumers as $future) {
+    $future->await();
+}
+$results->close();
+foreach ($results as $thumbnail) {
+    echo $thumbnail, "\n";
+}
+```
+
+```php
+// worker.php
+function resize_images(Zphp\Channel $jobs, Zphp\Channel $results): void
+{
+    foreach ($jobs as $path) {
+        $results->send(resize($path));
+    }
+}
+```
+
+`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/runtime/value.zig b/src/runtime/value.zig
index dd4cfd8d..b0e6d3a7 100644
--- a/src/runtime/value.zig
+++ b/src/runtime/value.zig
@@ -803,6 +803,7 @@ pub const NativeHandle = struct {
         xml_writer,
         pool,
         future,
+        channel,
         _,
     };
 
diff --git a/src/runtime/vm.zig b/src/runtime/vm.zig
index d2498333..b85ef2bb 100644
--- a/src/runtime/vm.zig
+++ b/src/runtime/vm.zig
@@ -1452,6 +1452,7 @@ pub const VM = struct {
         try @import("../stdlib/intl.zig").register(vm, allocator);
         try @import("../stdlib/gmp.zig").register(vm, allocator);
         try @import("../stdlib/workers.zig").register(vm, allocator);
+        try @import("../stdlib/channel.zig").register(vm, allocator);
         try @import("../stdlib/bcmath.zig").register(vm, allocator);
         try @import("../stdlib/gd.zig").register(vm, allocator);
         try @import("../stdlib/soap.zig").register(vm, allocator);
@@ -2432,6 +2433,7 @@ pub const VM = struct {
         @import("../stdlib/intl.zig").cleanupResources(self.objects);
         @import("../stdlib/gmp.zig").cleanupResources(self.objects);
         @import("../stdlib/workers.zig").cleanupResources(self.objects);
+        @import("../stdlib/channel.zig").cleanupResources(self.objects);
         extension.cleanupResources(self.objects);
         @import("../stdlib/gd.zig").cleanupResources(self.objects);
         @import("../stdlib/ftp.zig").cleanupResources(self.objects);
diff --git a/src/stdlib/channel.zig b/src/stdlib/channel.zig
new file mode 100644
index 00000000..e1ae608a
--- /dev/null
+++ b/src/stdlib/channel.zig
@@ -0,0 +1,404 @@
+// a Zphp\Channel is a bounded queue of serialized values shared between
+// threads by identity: every VM that holds one binds its own wrapper object
+// to the same Channel, and a wrapper crosses to a worker as the channel's id
+const std = @import("std");
+const value_mod = @import("../runtime/value.zig");
+const Value = value_mod.Value;
+const PhpObject = value_mod.PhpObject;
+const vm_mod = @import("../runtime/vm.zig");
+const VM = vm_mod.VM;
+const NativeContext = vm_mod.NativeContext;
+const ClassDef = vm_mod.ClassDef;
+const NativeResult = @import("../runtime/native_result.zig").NativeResult;
+const RuntimeError = error{ RuntimeError, OutOfMemory };
+const workers = @import("workers.zig");
+const platform = @import("../platform.zig");
+
+pub const channel_class = "Zphp\\Channel";
+const channel_exception = "Zphp\\ChannelException";
+const timeout_exception = "Zphp\\TimeoutException";
+
+// ---------------------------------------------------------------------------
+// the shared queue
+
+const Wait = union(enum) { none, forever, until: i128 };
+
+fn waitFor(cond: *std.Thread.Condition, mutex: *std.Thread.Mutex, wait: Wait) bool {
+    switch (wait) {
+        .none => return false,
+        .forever => {
+            cond.wait(mutex);
+            return true;
+        },
+        .until => |deadline| {
+            const now = std.time.nanoTimestamp();
+            if (now >= deadline) return false;
+            cond.timedWait(mutex, @intCast(deadline - now)) catch {};
+            return true;
+        },
+    }
+}
+
+pub const Channel = struct {
+    allocator: std.mem.Allocator,
+    id: u64,
+    items: []workers.Payload,
+    head: usize = 0,
+    len: usize = 0,
+    closed: bool = false,
+    mutex: std.Thread.Mutex = .{},
+    not_empty: std.Thread.Condition = .{},
+    not_full: std.Thread.Condition = .{},
+    // one reference per wrapper object, across every vm in the process
+    refs: std.atomic.Value(u32) = std.atomic.Value(u32).init(1),
+
+    const SendResult = enum { ok, full, closed, timeout };
+    const RecvResult = union(enum) { value: workers.Payload, empty, closed, timeout };
+
+    fn send(ch: *Channel, payload: workers.Payload, wait: Wait) SendResult {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        while (!ch.closed and ch.len == ch.items.len) {
+            if (!waitFor(&ch.not_full, &ch.mutex, wait)) return if (wait == .none) .full else .timeout;
+        }
+        if (ch.closed) return .closed;
+        ch.items[(ch.head + ch.len) % ch.items.len] = payload;
+        ch.len += 1;
+        ch.not_empty.signal();
+        return .ok;
+    }
+
+    fn recv(ch: *Channel, wait: Wait) RecvResult {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        while (ch.len == 0 and !ch.closed) {
+            if (!waitFor(&ch.not_empty, &ch.mutex, wait)) return if (wait == .none) .empty else .timeout;
+        }
+        if (ch.len == 0) return .closed;
+        const payload = ch.items[ch.head];
+        ch.head = (ch.head + 1) % ch.items.len;
+        ch.len -= 1;
+        ch.not_full.signal();
+        return .{ .value = payload };
+    }
+
+    // buffered values stay receivable; senders and empty receivers are released
+    fn close(ch: *Channel) void {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        ch.closed = true;
+        ch.not_empty.broadcast();
+        ch.not_full.broadcast();
+    }
+
+    fn count(ch: *Channel) usize {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        return ch.len;
+    }
+
+    fn isClosed(ch: *Channel) bool {
+        ch.mutex.lock();
+        defer ch.mutex.unlock();
+        return ch.closed;
+    }
+
+    pub fn retain(ch: *Channel) void {
+        _ = ch.refs.fetchAdd(1, .acq_rel);
+    }
+
+    // the last wrapper takes the channel out of the registry under its lock,
+    // so a lookup racing the release either retains a live channel or misses
+    pub fn release(ch: *Channel) void {
+        registry_mutex.lock();
+        if (ch.refs.fetchSub(1, .acq_rel) != 1) {
+            registry_mutex.unlock();
+            return;
+        }
+        _ = registry.remove(ch.id);
+        registry_mutex.unlock();
+        ch.free();
+    }
+
+    fn free(ch: *Channel) void {
+        var i: usize = 0;
+        while (i < ch.len) : (i += 1) ch.items[(ch.head + i) % ch.items.len].free(ch.allocator);
+        ch.allocator.free(ch.items);
+        ch.allocator.destroy(ch);
+    }
+};
+
+// ---------------------------------------------------------------------------
+// the process-wide registry that resolves an id back to its channel
+
+var registry_mutex: std.Thread.Mutex = .{};
+var registry: std.AutoHashMapUnmanaged(u64, *Channel) = .{};
+var next_id: u64 = 1;
+const registry_allocator = std.heap.page_allocator;
+
+fn create(allocator: std.mem.Allocator, capacity: usize) !*Channel {
+    const ch = try allocator.create(Channel);
+    errdefer allocator.destroy(ch);
+    const items = try allocator.alloc(workers.Payload, capacity);
+    errdefer allocator.free(items);
+    registry_mutex.lock();
+    defer registry_mutex.unlock();
+    ch.* = .{ .allocator = allocator, .id = next_id, .items = items };
+    try registry.put(registry_allocator, ch.id, ch);
+    next_id += 1;
+    return ch;
+}
+
+fn lookup(id: u64) ?*Channel {
+    registry_mutex.lock();
+    defer registry_mutex.unlock();
+    const ch = registry.get(id) orelse return null;
+    ch.retain();
+    return ch;
+}
+
+// ---------------------------------------------------------------------------
+// php surface
+
+fn getThis(ctx: *NativeContext) ?*PhpObject {
+    const v = ctx.vm.currentFrame().vars.get("$this") orelse return null;
+    if (v != .object) return null;
+    return v.object;
+}
+
+fn throwNamed(ctx: *NativeContext, class_name: []const u8, comptime fmt: []const u8, args: anytype) RuntimeError {
+    const msg = try std.fmt.allocPrint(ctx.allocator, fmt, args);
+    try ctx.vm.strings.append(ctx.allocator, msg);
+    try ctx.vm.setPendingException(class_name, msg);
+    return error.RuntimeError;
+}
+
+fn channelOf(ctx: *NativeContext, obj: *PhpObject) RuntimeError!*Channel {
+    return obj.native.get(Channel, .channel) orelse throwNamed(ctx, channel_exception, "the channel is not open", .{});
+}
+
+fn bind(obj: *PhpObject, ch: *Channel) void {
+    obj.native = .{ .kind = .channel, .ptr = @intFromPtr(ch) };
+}
+
+fn waitArg(args: []const Value, index: usize) Wait {
+    if (index >= args.len or args[index] == .null) return .forever;
+    const ns = workers.optionalSeconds(args[index]) orelse return .forever;
+    return .{ .until = std.time.nanoTimestamp() + @as(i128, ns) };
+}
+
+fn flushOutput(vm: *VM) void {
+    if (vm.output.items.len == 0) return;
+    platform.writeStdout(vm.output.items);
+    vm.output.clearRetainingCapacity();
+}
+
+fn channelConstruct(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    var capacity: usize = 1;
+    if (args.len >= 1 and args[0] != .null) {
+        if (args[0] != .int or args[0].int < 1) return throwNamed(ctx, channel_exception, "capacity must be a positive integer", .{});
+        capacity = @intCast(args[0].int);
+    }
+    const ch = create(workers.transferAllocator(ctx.allocator), capacity) catch return throwNamed(ctx, channel_exception, "could not create the channel", .{});
+    bind(obj, ch);
+    return NativeResult.scalar(.null);
+}
+
+fn sendValue(ctx: *NativeContext, args: []const Value, wait: Wait) RuntimeError!Channel.SendResult {
+    const obj = getThis(ctx) orelse return .closed;
+    const ch = try channelOf(ctx, obj);
+    if (args.len < 1) return throwNamed(ctx, channel_exception, "send() needs a value", .{});
+    const payload = try workers.pack(ctx, args[0], "value", ch.allocator);
+    flushOutput(ctx.vm);
+    const result = ch.send(payload, wait);
+    if (result != .ok) payload.free(ch.allocator);
+    return result;
+}
+
+fn channelSend(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    switch (try sendValue(ctx, args, waitArg(args, 1))) {
+        .ok, .full => return NativeResult.scalar(.null),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+        .timeout => return throwNamed(ctx, timeout_exception, "the channel did not accept the value in time", .{}),
+    }
+}
+
+fn channelTrySend(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    switch (try sendValue(ctx, args, .none)) {
+        .ok => return NativeResult.scalar(.{ .bool = true }),
+        .full, .timeout => return NativeResult.scalar(.{ .bool = false }),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+    }
+}
+
+fn unpackValue(ctx: *NativeContext, ch: *Channel, payload: workers.Payload) RuntimeError!Value {
+    return workers.unpack(ctx, payload, ch.allocator) orelse throwNamed(ctx, channel_exception, "the value did not transfer", .{});
+}
+
+fn receive(ctx: *NativeContext, ch: *Channel, wait: Wait) Channel.RecvResult {
+    flushOutput(ctx.vm);
+    return ch.recv(wait);
+}
+
+fn channelRecv(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    switch (receive(ctx, ch, waitArg(args, 0))) {
+        .value => |payload| return NativeResult.transfer(try unpackValue(ctx, ch, payload)),
+        .closed => return throwNamed(ctx, channel_exception, "the channel is closed", .{}),
+        .empty, .timeout => return throwNamed(ctx, timeout_exception, "no value arrived in time", .{}),
+    }
+}
+
+fn channelClose(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    ch.close();
+    return NativeResult.scalar(.null);
+}
+
+fn channelIsClosed(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .bool = ch.isClosed() });
+}
+
+fn channelCount(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.count()) });
+}
+
+fn channelCapacity(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.items.len) });
+}
+
+fn channelId(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    return NativeResult.scalar(.{ .int = @intCast(ch.id) });
+}
+
+// ---------------------------------------------------------------------------
+// iteration: foreach pulls values until the channel is closed and drained.
+// the pending value and its position live on the wrapper, so each consumer
+// iterates independently
+
+fn channelRewind(_: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    return NativeResult.scalar(.null);
+}
+
+fn channelValid(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.{ .bool = false });
+    const ch = try channelOf(ctx, obj);
+    if (obj.get("__ready") == .bool and obj.get("__ready").bool) return NativeResult.scalar(.{ .bool = true });
+    switch (receive(ctx, ch, .forever)) {
+        .value => |payload| {
+            const v = try unpackValue(ctx, ch, payload);
+            try obj.set(ctx.allocator, "__current", v);
+            if (v == .string) v.string.release();
+            try obj.set(ctx.allocator, "__ready", .{ .bool = true });
+            return NativeResult.scalar(.{ .bool = true });
+        },
+        else => return NativeResult.scalar(.{ .bool = false }),
+    }
+}
+
+fn channelCurrent(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    return NativeResult.share(obj.get("__current"));
+}
+
+fn channelKey(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const key = obj.get("__key");
+    return NativeResult.scalar(.{ .int = if (key == .int) key.int else 0 });
+}
+
+fn channelNext(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const key = obj.get("__key");
+    try obj.set(ctx.allocator, "__key", .{ .int = if (key == .int) key.int + 1 else 1 });
+    try obj.set(ctx.allocator, "__ready", .{ .bool = false });
+    try obj.set(ctx.allocator, "__current", .null);
+    return NativeResult.scalar(.null);
+}
+
+// ---------------------------------------------------------------------------
+// transfer: a channel serializes as its id and binds to the same channel
+// wherever it is unserialized in this process
+
+fn channelSerialize(ctx: *NativeContext, _: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    const ch = try channelOf(ctx, obj);
+    const arr = try ctx.createArray();
+    try arr.set(ctx.allocator, .{ .string = Value.String.borrowed("id") }, .{ .int = @intCast(ch.id) });
+    return NativeResult.borrowed(.{ .array = arr });
+}
+
+fn channelUnserialize(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResult {
+    const obj = getThis(ctx) orelse return NativeResult.scalar(.null);
+    if (obj.native.kind != .none) return NativeResult.scalar(.null);
+    const id: u64 = blk: {
+        if (args.len < 1 or args[0] != .array) break :blk 0;
+        const v = args[0].array.get(.{ .string = Value.String.borrowed("id") });
+        break :blk if (v == .int and v.int > 0) @intCast(v.int) else 0;
+    };
+    const ch = lookup(id) orelse return throwNamed(ctx, channel_exception, "channel {d} does not exist in this process", .{id});
+    bind(obj, ch);
+    return NativeResult.scalar(.null);
+}
+
+// ---------------------------------------------------------------------------
+// lifetimes
+
+fn cleanupChannel(obj: *PhpObject) bool {
+    const ch = obj.native.get(Channel, .channel) orelse return true;
+    obj.native = .{};
+    ch.release();
+    return true;
+}
+
+pub fn cleanupResources(objects: std.ArrayListUnmanaged(*PhpObject)) void {
+    for (objects.items) |obj| {
+        if (obj.pooled) continue;
+        if (obj.native.kind == .channel) _ = cleanupChannel(obj);
+    }
+}
+
+// ---------------------------------------------------------------------------
+// registration
+
+const methods = [_]struct { name: []const u8, arity: u8, native: vm_mod.NativeFn }{
+    .{ .name = "__construct", .arity = 1, .native = channelConstruct },
+    .{ .name = "send", .arity = 2, .native = channelSend },
+    .{ .name = "trySend", .arity = 1, .native = channelTrySend },
+    .{ .name = "recv", .arity = 1, .native = channelRecv },
+    .{ .name = "close", .arity = 0, .native = channelClose },
+    .{ .name = "isClosed", .arity = 0, .native = channelIsClosed },
+    .{ .name = "count", .arity = 0, .native = channelCount },
+    .{ .name = "capacity", .arity = 0, .native = channelCapacity },
+    .{ .name = "id", .arity = 0, .native = channelId },
+    .{ .name = "rewind", .arity = 0, .native = channelRewind },
+    .{ .name = "valid", .arity = 0, .native = channelValid },
+    .{ .name = "current", .arity = 0, .native = channelCurrent },
+    .{ .name = "key", .arity = 0, .native = channelKey },
+    .{ .name = "next", .arity = 0, .native = channelNext },
+    .{ .name = "__serialize", .arity = 0, .native = channelSerialize },
+    .{ .name = "__unserialize", .arity = 1, .native = channelUnserialize },
+};
+
+pub fn register(vm: *VM, a: std.mem.Allocator) !void {
+    var def = ClassDef{ .name = channel_class, .is_final = true, .native_cleanup = cleanupChannel };
+    try def.interfaces.append(a, "Iterator");
+    try def.interfaces.append(a, "Countable");
+    inline for (methods) |m| {
+        try def.methods.put(a, m.name, .{ .name = m.name, .arity = m.arity });
+        try vm.native_fns.put(a, channel_class ++ "::" ++ m.name, m.native);
+    }
+    try vm.classes.put(a, channel_class, def);
+    try vm.classes.put(a, channel_exception, ClassDef{ .name = channel_exception, .parent = "Exception" });
+}
diff --git a/src/stdlib/native_params.zig b/src/stdlib/native_params.zig
index 38b06f9d..480fd10e 100644
--- a/src/stdlib/native_params.zig
+++ b/src/stdlib/native_params.zig
@@ -12,6 +12,10 @@ pub const map = std.StaticStringMap([]const []const u8).initComptime(.{
     .{ "Zphp\\Pool::collect", &.{"$timeout"} },
     .{ "Zphp\\Pool::shutdown", &.{"$timeout"} },
     .{ "Zphp\\Future::await", &.{"$timeout"} },
+    .{ "Zphp\\Channel::__construct", &.{"$capacity"} },
+    .{ "Zphp\\Channel::send", &.{ "$value", "$timeout" } },
+    .{ "Zphp\\Channel::trySend", &.{"$value"} },
+    .{ "Zphp\\Channel::recv", &.{"$timeout"} },
     .{ "substr", &.{ "$string", "$offset", "$length" } },
     .{ "str_replace", &.{ "$search", "$replace", "$subject", "$count" } },
     .{ "str_ireplace", &.{ "$search", "$replace", "$subject", "$count" } },
diff --git a/src/stdlib/serialize.zig b/src/stdlib/serialize.zig
index a0a87eda..eb408b2d 100644
--- a/src/stdlib/serialize.zig
+++ b/src/stdlib/serialize.zig
@@ -180,6 +180,22 @@ pub fn serializeToString(ctx: *NativeContext, val: Value) RuntimeError!NativeRes
 }
 
 // a string result carries one reference the caller must release or store
+// the object keeps its class name, so it must outlive this call: the class
+// table's own key does, and only an unknown class needs a copy. a script that
+// unserializes objects in a loop would otherwise grow the request arena by a
+// class name per object
+fn keptClassName(ctx: *NativeContext, class_allowed: bool, orig_class: []const u8) ![]const u8 {
+    if (!class_allowed) return "__PHP_Incomplete_Class";
+    return ctx.vm.classes.getKey(orig_class) orelse try ctx.createString(orig_class);
+}
+
+// a declared property is stored by slot and the name is not kept; a dynamic
+// one is stored by name and needs request-lifetime bytes
+fn keptPropertyName(ctx: *NativeContext, obj: *PhpObject, name: []const u8) ![]const u8 {
+    if (obj.getSlotIndex(name) != null) return name;
+    return ctx.createString(name);
+}
+
 pub fn unserializeFromString(ctx: *NativeContext, s: []const u8) ?Value {
     var uctx = UnserCtx{};
     defer uctx.deinit(ctx.allocator);
@@ -540,7 +556,9 @@ fn native_unserialize(ctx: *NativeContext, args: []const Value) RuntimeError!Nat
         if (md == .int and md.int >= 0) uctx.max_depth = @intCast(md.int);
     }
     const result = unserializeValue(ctx, &uctx, s, 0) catch {
-        if (uctx.threw) return error.RuntimeError;
+        // an exception from __unserialize, __wakeup, or Serializable::unserialize
+        // propagates; only a parse failure is a warning
+        if (uctx.threw or ctx.vm.pending_exception != null) return error.RuntimeError;
         // emit PHP's parse-failure warning. depth-exceeded gets its own
         // message that names the option + ini setting; generic parse errors
         // get 'Error at offset N of M bytes'
@@ -711,7 +729,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
             if (colon1 + 2 + name_len + 1 >= s.len) return error.RuntimeError;
             const orig_class = s[colon1 + 2 .. colon1 + 2 + name_len];
             const class_allowed = uctx.classAllowed(orig_class) and ctx.vm.classes.contains(orig_class);
-            const class_name = try ctx.createString(if (class_allowed) orig_class else "__PHP_Incomplete_Class");
+            const class_name = try keptClassName(ctx, class_allowed, orig_class);
             var p = colon1 + 2 + name_len + 2;
             const count_end = std.mem.indexOfPos(u8, s, p, ":") orelse return error.RuntimeError;
             const prop_count = std.fmt.parseInt(usize, s[p..count_end], 10) catch return error.RuntimeError;
@@ -787,10 +805,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
                 } else if (fixed_data) |arr| {
                     if (key_result.value == .int) try arr.set(ctx.allocator, .{ .int = key_result.value.int }, val_result.value);
                 } else if (key_result.value == .string) {
-                    // property names are kept by the object as given, so
-                    // they need request-lifetime bytes rather than the
-                    // counted key that dies with this iteration
-                    const stripped = try ctx.createString(stripVisibilityPrefix(key_result.value.string.bytes()));
+                    const stripped = try keptPropertyName(ctx, obj, stripVisibilityPrefix(key_result.value.string.bytes()));
                     // when restoring into a kept class, assigning an
                     // __PHP_Incomplete_Class value to a typed property whose
                     // declared type isn't compatible is a TypeError in PHP
@@ -836,7 +851,7 @@ fn unserializeValue(ctx: *NativeContext, uctx: *UnserCtx, s: []const u8, pos: us
             if (colon1 + 2 + name_len + 1 >= s.len) return error.RuntimeError;
             const orig_class = s[colon1 + 2 .. colon1 + 2 + name_len];
             const class_allowed = uctx.classAllowed(orig_class) and ctx.vm.classes.contains(orig_class);
-            const class_name = try ctx.createString(if (class_allowed) orig_class else "__PHP_Incomplete_Class");
+            const class_name = try keptClassName(ctx, class_allowed, orig_class);
             var p = colon1 + 2 + name_len + 2;
             const len_end = std.mem.indexOfPos(u8, s, p, ":") orelse return error.RuntimeError;
             const data_len = std.fmt.parseInt(usize, s[p..len_end], 10) catch return error.RuntimeError;
diff --git a/src/stdlib/workers.zig b/src/stdlib/workers.zig
index b6d75303..a9f16a3d 100644
--- a/src/stdlib/workers.zig
+++ b/src/stdlib/workers.zig
@@ -3,6 +3,7 @@
 // Zphp\Future carries the result or the exception back. values cross as
 // serialized bytes and are materialized in the receiving VM
 const std = @import("std");
+const builtin = @import("builtin");
 const value_mod = @import("../runtime/value.zig");
 const Value = value_mod.Value;
 const PhpObject = value_mod.PhpObject;
@@ -18,6 +19,7 @@ const serialize = @import("serialize.zig");
 const network = @import("network.zig");
 const platform = @import("../platform.zig");
 const extension = @import("../extension.zig");
+const channel = @import("channel.zig");
 
 const pool_class = "Zphp\\Pool";
 const future_class = "Zphp\\Future";
@@ -30,6 +32,18 @@ const transfer_exception = "Zphp\\TransferException";
 
 const default_queue: usize = 1024;
 
+// everything the pool touches, worker VM heaps included. the release build's
+// smp_allocator keeps a freelist per thread slot, reclaims another slot's
+// list only when its own runs dry, and hands every thread slot zero until
+// contention moves it, so a producer that allocates on one thread for a
+// consumer that frees on another keeps mapping fresh slabs, and two busy
+// worker VMs on a four-cpu box scatter their frees across slots and grew
+// rss by up to 90 MB over a soak that other runs finished flat. libc malloc
+// balances both. the Debug build keeps its leak-checking allocator
+pub fn transferAllocator(vm_allocator: std.mem.Allocator) std.mem.Allocator {
+    return if (builtin.mode == .Debug) vm_allocator else std.heap.c_allocator;
+}
+
 // ---------------------------------------------------------------------------
 // tasks
 
@@ -47,9 +61,9 @@ const Task = struct {
     id: u64,
     pool: *Pool,
     callable: []u8,
-    args: []u8,
+    args: Payload,
     state: TaskState = .queued,
-    result: ?[]u8 = null,
+    result: ?Payload = null,
     failure: ?Failure = null,
     fatal: ?[]u8 = null,
     cancel_requested: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
@@ -68,8 +82,8 @@ const Task = struct {
         const pool = t.pool;
         const a = pool.allocator;
         a.free(t.callable);
-        a.free(t.args);
-        if (t.result) |r| a.free(r);
+        t.args.free(a);
+        if (t.result) |r| r.free(a);
         if (t.failure) |f| {
             a.free(f.class_name);
             a.free(f.message);
@@ -214,14 +228,20 @@ const Pool = struct {
         if (pool.refs.fetchSub(1, .acq_rel) == 1) pool.free();
     }
 
-    fn create(allocator: std.mem.Allocator, owner: *VM, workers: usize, bootstrap_path: ?[]const u8, queue_size: usize) !*Pool {
+    fn create(vm_allocator: std.mem.Allocator, owner: *VM, workers: usize, bootstrap_path: ?[]const u8, queue_size: usize) !*Pool {
+        const allocator = transferAllocator(vm_allocator);
         const pool = try allocator.create(Pool);
         errdefer allocator.destroy(pool);
         const items = try allocator.alloc(*Task, queue_size);
         errdefer allocator.free(items);
         const slots = try allocator.alloc(Worker, workers);
         errdefer allocator.free(slots);
+        // the wake is a level: one byte per pending completion, written
+        // without blocking so a backlog nobody collects can never stall a
+        // worker, and read without blocking so a missing byte is not waited on
         const wake = try platform.socketPair();
+        try platform.setNonBlocking(wake[0], true);
+        try platform.setNonBlocking(wake[1], true);
         pool.* = .{
             .allocator = allocator,
             .owner = owner,
@@ -271,8 +291,20 @@ const Pool = struct {
         return id;
     }
 
+    // a future awaited directly may take delivery between the task settling
+    // and this call; then the completion list and the wake must not see it,
+    // or the byte nobody reads accumulates until the socket buffer stalls
+    // every worker
     fn complete(pool: *Pool, task: *Task) void {
         pool.mutex.lock();
+        task.mutex.lock();
+        const delivered = task.delivered;
+        task.mutex.unlock();
+        if (delivered) {
+            pool.mutex.unlock();
+            task.release();
+            return;
+        }
         pool.completed.append(pool.allocator, task) catch {};
         pool.changed.broadcast();
         pool.mutex.unlock();
@@ -500,7 +532,9 @@ fn runTask(vm: *VM, task: *Task) void {
 
 fn execute(ctx: *NativeContext, task: *Task) void {
     const callable = hold(serialize.unserializeFromString(ctx, task.callable) orelse return settleFatal(task, "the callable did not transfer"));
-    const args_value = hold(serialize.unserializeFromString(ctx, task.args) orelse return settleFatal(task, "the arguments 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"));
     var args: [64]Value = undefined;
     var count: usize = 0;
     if (args_value == .array) {
@@ -520,15 +554,13 @@ fn execute(ctx: *NativeContext, task: *Task) void {
     // native result: strings need one more retain, containers are ours
     const owned = NativeResult.share(result).value;
     defer ctx.vm.releaseValue(owned);
-    const bytes = serialize.serializeToString(ctx, result) catch {
-        settleFatal(task, "the result could not be transferred back");
+    const payload = pack(ctx, result, "result", task.pool.allocator) catch {
+        settleFailure(ctx.vm, task, callable);
         return;
     };
-    defer bytes.value.string.release();
-    const copy = task.pool.allocator.dupe(u8, bytes.value.string.bytes()) catch return settleFatal(task, "out of memory");
     task.mutex.lock();
     defer task.mutex.unlock();
-    task.result = copy;
+    task.result = payload;
     task.state = .done;
     task.finished.broadcast();
 }
@@ -604,14 +636,31 @@ fn settleFailure(vm: *VM, task: *Task, callable: Value) void {
 }
 
 // ---------------------------------------------------------------------------
-// transfer rules, checked in the caller so the error names the path
+// transfer: a value crosses threads as serialized bytes plus a reference to
+// every channel inside it, so a channel that only the bytes name stays alive
+// until the receiver has bound its own wrapper
+
+pub const Payload = struct {
+    bytes: []u8,
+    channels: []*channel.Channel,
+
+    pub const empty = Payload{ .bytes = &.{}, .channels = &.{} };
+
+    pub fn free(p: Payload, a: std.mem.Allocator) void {
+        for (p.channels) |ch| ch.release();
+        a.free(p.channels);
+        a.free(p.bytes);
+    }
+};
 
+// the transfer rules are checked in the sender so the error names the path
 const TransferCheck = struct {
     ctx: *NativeContext,
     path: std.ArrayListUnmanaged(u8) = .{},
+    channels: std.ArrayListUnmanaged(*channel.Channel) = .{},
 
     fn refuse(self: *TransferCheck, what: []const u8) RuntimeError {
-        const msg = try std.fmt.allocPrint(self.ctx.allocator, "{s} cannot be transferred to a worker (at {s})", .{ what, self.path.items });
+        const msg = try std.fmt.allocPrint(self.ctx.allocator, "{s} cannot be transferred between threads (at {s})", .{ what, self.path.items });
         try self.ctx.vm.strings.append(self.ctx.allocator, msg);
         try self.ctx.vm.setPendingException(transfer_exception, msg);
         return error.RuntimeError;
@@ -635,6 +684,7 @@ const TransferCheck = struct {
                 }
             },
             .object => |obj| {
+                if (obj.native.get(channel.Channel, .channel)) |ch| return self.channels.append(self.ctx.allocator, ch);
                 if (obj.native.kind != .none) return self.refuse("an object backed by a native handle");
                 if (std.mem.eql(u8, obj.class_name, pool_class) or std.mem.eql(u8, obj.class_name, future_class)) return self.refuse("a pool or future");
                 const mark = self.path.items.len;
@@ -653,17 +703,30 @@ const TransferCheck = struct {
     }
 };
 
-fn checkTransferable(ctx: *NativeContext, v: Value, root: []const u8) RuntimeError!void {
+fn serializedCopy(ctx: *NativeContext, v: Value, allocator: std.mem.Allocator) RuntimeError![]u8 {
+    const bytes = try serialize.serializeToString(ctx, v);
+    defer bytes.value.string.release();
+    return allocator.dupe(u8, bytes.value.string.bytes());
+}
+
+pub fn pack(ctx: *NativeContext, v: Value, root: []const u8, allocator: std.mem.Allocator) RuntimeError!Payload {
     var tc = TransferCheck{ .ctx = ctx };
     defer tc.path.deinit(ctx.allocator);
+    defer tc.channels.deinit(ctx.allocator);
     try tc.path.appendSlice(ctx.allocator, root);
     try tc.check(v);
+    const bytes = try serializedCopy(ctx, v, allocator);
+    errdefer allocator.free(bytes);
+    const channels = try allocator.dupe(*channel.Channel, tc.channels.items);
+    for (channels) |ch| ch.retain();
+    return .{ .bytes = bytes, .channels = channels };
 }
 
-fn serializedCopy(ctx: *NativeContext, v: Value, allocator: std.mem.Allocator) RuntimeError![]u8 {
-    const bytes = try serialize.serializeToString(ctx, v);
-    defer bytes.value.string.release();
-    return allocator.dupe(u8, bytes.value.string.bytes());
+// materializes the value in this vm and drops the payload; the wrappers the
+// unserializer bound hold their own channel references
+pub fn unpack(ctx: *NativeContext, payload: Payload, allocator: std.mem.Allocator) ?Value {
+    defer payload.free(allocator);
+    return serialize.unserializeFromString(ctx, payload.bytes);
 }
 
 // ---------------------------------------------------------------------------
@@ -692,7 +755,7 @@ fn taskOf(obj: *PhpObject) ?*Task {
     return obj.native.get(Task, .future);
 }
 
-fn optionalSeconds(v: Value) ?u64 {
+pub fn optionalSeconds(v: Value) ?u64 {
     return switch (v) {
         .int => |i| if (i < 0) null else @as(u64, @intCast(i)) * std.time.ns_per_s,
         .float => |f| if (f < 0) null else @as(u64, @intFromFloat(f * @as(f64, @floatFromInt(std.time.ns_per_s)))),
@@ -747,7 +810,8 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
     if (!valid) return throwNamed(ctx, transfer_exception, "tasks are 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", .{});
-    try checkTransferable(ctx, task_args, "args");
+    const packed_args = try pack(ctx, task_args, "args", pool.allocator);
+    errdefer packed_args.free(pool.allocator);
     pool.mutex.lock();
     const closed = pool.shutting_down;
     pool.mutex.unlock();
@@ -755,26 +819,24 @@ fn submitTask(ctx: *NativeContext, args: []const Value, block: bool) RuntimeErro
 
     const task = try pool.allocator.create(Task);
     errdefer pool.allocator.destroy(task);
-    task.* = .{ .id = pool.nextId(), .pool = pool, .callable = &.{}, .args = &.{} };
+    task.* = .{ .id = pool.nextId(), .pool = pool, .callable = &.{}, .args = packed_args };
     pool.retain();
     errdefer pool.release();
     task.callable = try serializedCopy(ctx, callable, pool.allocator);
     errdefer pool.allocator.free(task.callable);
-    task.args = try serializedCopy(ctx, task_args, pool.allocator);
-    errdefer pool.allocator.free(task.args);
 
     switch (pool.queue.push(task, block)) {
         .ok => {},
         .full => {
             pool.allocator.free(task.callable);
-            pool.allocator.free(task.args);
+            packed_args.free(pool.allocator);
             pool.allocator.destroy(task);
             pool.release();
             return NativeResult.scalar(.null);
         },
         .closed => {
             pool.allocator.free(task.callable);
-            pool.allocator.free(task.args);
+            packed_args.free(pool.allocator);
             pool.allocator.destroy(task);
             pool.release();
             return throwNamed(ctx, pool_exception, "the pool is shutting down", .{});
@@ -854,13 +916,12 @@ fn futureAwait(ctx: *NativeContext, args: []const Value) RuntimeError!NativeResu
     switch (state) {
         .done => {
             // materialized once; later awaits read the value kept on the future
-            if (task.result) |bytes| {
-                const v = serialize.unserializeFromString(ctx, bytes) orelse return throwNamed(ctx, task_exception, "the result did not transfer", .{});
+            if (task.result) |payload| {
+                task.result = null;
+                const v = unpack(ctx, payload, task.pool.allocator) orelse return throwNamed(ctx, task_exception, "the result did not transfer", .{});
                 // the store takes its own reference; the one unserialize handed over goes
                 try obj.set(ctx.allocator, "__result", v);
                 if (v == .string) v.string.release();
-                task.pool.allocator.free(bytes);
-                task.result = null;
             }
             return NativeResult.share(obj.get("__result"));
         },
diff --git a/tests/unserialize_magic_throws.php b/tests/unserialize_magic_throws.php
new file mode 100644
index 00000000..59c8d4aa
--- /dev/null
+++ b/tests/unserialize_magic_throws.php
@@ -0,0 +1,16 @@
+<?php
+// an exception thrown while unserialize() rebuilds an object propagates to
+// the caller; only a malformed payload is a warning plus false
+class Rebuilt {
+    public function __serialize(): array { return ['n' => 1]; }
+    public function __unserialize(array $data): void { throw new RuntimeException("rebuild refused " . $data['n']); }
+}
+class Woken {
+    public $n = 2;
+    public function __wakeup(): void { throw new LogicException("wake refused"); }
+}
+try { unserialize(serialize(new Rebuilt)); echo "no exception\n"; } catch (RuntimeException $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+try { unserialize(serialize(new Woken)); echo "no exception\n"; } catch (LogicException $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+var_dump(@unserialize('a:1:{i:0;s:3:"ab"}'));
+$nested = unserialize(serialize(['ok' => 1]));
+var_dump($nested);
diff --git a/tests/workers/basic.expected b/tests/workers/basic.expected
index 19eee832..504ddf5a 100644
--- a/tests/workers/basic.expected
+++ b/tests/workers/basic.expected
@@ -5,7 +5,7 @@ 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]
-transfer: an object backed by a native handle cannot be transferred to a worker (at args[0])
+transfer: an object backed by a native handle cannot be transferred between threads (at args[0])
 int(2470)
 timeout
 bool(false)
diff --git a/tests/workers/channel.expected b/tests/workers/channel.expected
new file mode 100644
index 00000000..cecd7012
--- /dev/null
+++ b/tests/workers/channel.expected
@@ -0,0 +1,48 @@
+int(3)
+int(0)
+bool(false)
+bool(true)
+bool(true)
+int(3)
+bool(false)
+string(1) "a"
+array(1) {
+}
+NULL
+recv timeout: no value arrived in time
+bool(true)
+string(4) "last"
+recv closed: the channel is closed
+send closed: the channel is closed
+trySend closed: the channel is closed
+bad capacity: capacity must be a positive integer
+transfer: a Closure cannot be transferred between threads (at value)
+Error: Trying to clone an uncloneable object of class Zphp\Channel
+send timeout: the channel did not accept the value in time
+bool(false)
+string(1) "x"
+bool(true)
+string(1) "y"
+0 => 10
+1 => 20
+2 => 30
+bool(true)
+bool(true)
+stale: channel 999999 does not exist in this process
+string(14) "through a copy"
+int(20)
+int(2870)
+int(2)
+int(50)
+int(1)
+int(50)
+string(11) "produced 50"
+string(5) "x,y,z"
+string(7) "timeout"
+string(8) "got late"
+string(6) "closed"
+bool(true)
+string(17) "hello from worker"
+int(3)
+result: a Closure cannot be transferred between threads (at result)
+end
diff --git a/tests/workers/channel.php b/tests/workers/channel.php
new file mode 100644
index 00000000..8f90ccf1
--- /dev/null
+++ b/tests/workers/channel.php
@@ -0,0 +1,92 @@
+<?php
+// single thread: a bounded queue with value semantics for what crosses it
+$ch = new Zphp\Channel(capacity: 3);
+var_dump($ch->capacity(), count($ch), $ch->isClosed(), $ch instanceof Iterator, $ch instanceof Countable);
+$ch->send("a"); $ch->send(["k" => 2]); $ch->send(null);
+var_dump(count($ch), $ch->trySend(4));
+var_dump($ch->recv(), $ch->recv(), $ch->recv());
+try { $ch->recv(0.05); } catch (Zphp\TimeoutException $e) { echo "recv timeout: ", $e->getMessage(), "\n"; }
+$ch->send("last");
+$ch->close();
+var_dump($ch->isClosed(), $ch->recv());
+try { $ch->recv(); } catch (Zphp\ChannelException $e) { echo "recv closed: ", $e->getMessage(), "\n"; }
+try { $ch->send(1); } catch (Zphp\ChannelException $e) { echo "send closed: ", $e->getMessage(), "\n"; }
+try { $ch->trySend(1); } catch (Zphp\ChannelException $e) { echo "trySend closed: ", $e->getMessage(), "\n"; }
+try { new Zphp\Channel(0); } catch (Zphp\ChannelException $e) { echo "bad capacity: ", $e->getMessage(), "\n"; }
+try { (new Zphp\Channel)->send(fn() => 1); } catch (Zphp\TransferException $e) { echo "transfer: ", $e->getMessage(), "\n"; }
+try { clone $ch; } catch (Error $e) { echo get_class($e), ": ", $e->getMessage(), "\n"; }
+
+// a full channel refuses a non-blocking send and times out a bounded one
+$full = new Zphp\Channel(1);
+$full->send("x");
+try { $full->send("y", 0.05); } catch (Zphp\TimeoutException $e) { echo "send timeout: ", $e->getMessage(), "\n"; }
+var_dump($full->trySend("y"), $full->recv(), $full->trySend("y"), $full->recv());
+
+// iteration drains a closed channel, keys count from zero per consumer
+$it = new Zphp\Channel(5);
+foreach ([1, 2, 3] as $n) $it->send($n * 10);
+$it->close();
+foreach ($it as $k => $v) echo "$k => $v\n";
+
+// a channel crosses as its id and binds to the same channel; a stale id is an error
+$again = unserialize(serialize($it));
+var_dump($again->isClosed(), $again->id() === $it->id());
+try { unserialize('O:12:"Zphp\Channel":1:{s:2:"id";i:999999;}'); } catch (Zphp\ChannelException $e) { echo "stale: ", $e->getMessage(), "\n"; }
+$carrier = new Zphp\Channel(1);
+$inner = new Zphp\Channel(1);
+$carrier->send(['inner' => $inner]);
+unset($inner);
+$got = $carrier->recv();
+$got['inner']->send("through a copy");
+var_dump($got['inner']->recv());
+
+// across threads
+$pool = new Zphp\Pool(workers: 2, bootstrap: __DIR__ . "/worker.php");
+
+// fan out jobs to two consumers, collect on a second channel
+$jobs = new Zphp\Channel(4);
+$results = new Zphp\Channel(100);
+$a = $pool->submit('consume', [$jobs, $results]);
+$b = $pool->submit('consume', [$jobs, $results]);
+for ($i = 1; $i <= 20; $i++) $jobs->send($i);
+$jobs->close();
+var_dump($a->await() + $b->await());
+$results->close();
+$sum = 0; $workers = []; foreach ($results as $r) { $sum += $r['sq']; $workers[$r['worker']] = true; }
+var_dump($sum, count($workers));
+
+// a worker produces, main iterates, capacity one keeps them in step
+$stream = new Zphp\Channel(1);
+$p = $pool->submit('produce', [$stream, 50]);
+$seen = []; foreach ($stream as $k => $v) $seen[$k] = $v;
+var_dump(count($seen), $seen[0], $seen[49], $p->await());
+
+// a pipeline of three stages
+$s1 = new Zphp\Channel(2); $s2 = new Zphp\Channel(2); $s3 = new Zphp\Channel(2);
+$pool->submit('forward', [$s1, $s2]); $pool->submit('forward', [$s2, $s3]);
+foreach (["x", "y", "z"] as $v) $s1->send($v);
+$s1->close();
+var_dump(implode(",", iterator_to_array($s3, false)));
+
+// a worker blocked in recv sees the timeout, then a value, then the close
+$c = new Zphp\Channel(1);
+var_dump($pool->submit('wait_recv', [$c, 0.05])->await());
+$w = $pool->submit('wait_recv', [$c, 5]); usleep(50000); $c->send("late"); var_dump($w->await());
+$w = $pool->submit('wait_recv', [$c, 5]); usleep(50000); $c->close(); var_dump($w->await());
+
+// a channel created in a worker comes back in a result and outlives the worker's wrapper
+$r = $pool->submit('make_channel')->await();
+var_dump($r['ch'] instanceof Zphp\Channel, $r['ch']->recv());
+
+// main drops its wrapper while a worker still consumes
+$d = new Zphp\Channel(2);
+$f = $pool->submit('slow_drain', [$d]);
+$d->send(1); $d->send(2); $d->send(3); $d->close();
+unset($d);
+var_dump($f->await());
+
+// a result that cannot cross fails the task with the path
+try { $pool->submit('bad_result')->await(); } catch (Zphp\TransferException $e) { echo "result: ", $e->getMessage(), "\n"; }
+
+$pool->shutdown();
+echo "end\n";
diff --git a/tests/workers/memory.php b/tests/workers/memory.php
index 86738125..dd526139 100644
--- a/tests/workers/memory.php
+++ b/tests/workers/memory.php
@@ -12,6 +12,13 @@ function big(int $n): array { return array_fill(0, $n, str_repeat("x", 64)); }
     for ($i = 0; $i < 200; $i++) { $pool->submit('str_repeat', ['x', 8000]); }
     while ($pool->collect(5.0)) {}
 }
+$stream = new Zphp\Channel(8);
+for ($batch = 0; $batch < 10; $batch++) {
+    $f = $pool->submit('produce_big', [$stream, 300]);
+    for ($i = 0; $i < 300; $i++) { $stream->recv(); }
+    $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"); }
 $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 55b34d0a..d5f35449 100755
--- a/tests/workers/run
+++ b/tests/workers/run
@@ -19,7 +19,7 @@ check() {
     fi
 }
 
-for script in basic edge; do
+for script in basic edge channel; 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 9b0d1163..f3032925 100644
--- a/tests/workers/worker.php
+++ b/tests/workers/worker.php
@@ -12,3 +12,21 @@ function until_cancelled(): string { while (!Zphp\Task::cancelled()) usleep(1000
 function counter(): int { static $n = 0; return ++$n; }
 function big(int $n): array { return array_fill(0, $n, str_repeat("x", 64)); }
 function echoes(string $s): void { echo $s, "\n"; }
+function consume(Zphp\Channel $jobs, Zphp\Channel $results): int {
+    $n = 0;
+    foreach ($jobs as $job) { $results->send(['job' => $job, 'worker' => Zphp\Task::worker(), 'sq' => $job * $job]); $n++; }
+    return $n;
+}
+function produce(Zphp\Channel $out, int $count): string {
+    for ($i = 1; $i <= $count; $i++) $out->send($i);
+    $out->close();
+    return "produced $count";
+}
+function wait_recv(Zphp\Channel $ch, float $t): string {
+    try { return "got " . $ch->recv($t); } catch (Zphp\TimeoutException $e) { return "timeout"; } catch (Zphp\ChannelException $e) { return "closed"; }
+}
+function make_channel(): array { $c = new Zphp\Channel(3); $c->send("hello from worker"); return ['ch' => $c]; }
+function slow_drain(Zphp\Channel $ch): int { $n = 0; foreach ($ch as $v) { usleep(20000); $n++; } return $n; }
+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)); }