pipe selections through shell commands

7 files changed, 794 insertions, 7 deletions

diff --git a/src/config.zig b/src/config.zig
index 3eedf707..a14ed797 100644
--- a/src/config.zig
+++ b/src/config.zig
@@ -356,6 +356,9 @@ pub const command_line: []const Chord = &.{.{ .cp = ':' }};
 pub const search: []const Chord = &.{.{ .cp = '/' }};
 pub const search_next: []const Chord = &.{.{ .cp = 'n' }};
 pub const search_prev: []const Chord = &.{.{ .cp = 'N' }};
+/// Helix `|`: in body normal mode, pipe every file selection through one
+/// command typed in the pane's visible tag-tail input.
+pub const pipe_selection: []const Chord = &.{.{ .cp = '|' }};
 /// Enter on an armed search input runs it; `escape` (PART 3) abandons it.
 pub const search_submit: []const Chord = &.{.{ .cp = Key.enter }};
 
@@ -574,6 +577,10 @@ pub const symbol_marker = " WsSymbols /";
 /// the one input that previews as you type (pardes.zig, previewSelRegex).
 pub const select_marker = " Select /";
 pub const split_marker = " Split /";
+/// The `|` prompt is not an executable tag word: the marker only makes the
+/// pending shell filter visible, and everything after it is preserved as the
+/// exact command passed to `/bin/sh -c`.
+pub const pipe_marker = " |";
 
 /// Output-buffer names (acme's +Errors). Cosmetic now, and deliberately so: a
 /// buffer is DERIVED from the command that opened it (output_pane.traits), and
diff --git a/src/file_pane.zig b/src/file_pane.zig
index b164a7b3..2f2a0acb 100644
--- a/src/file_pane.zig
+++ b/src/file_pane.zig
@@ -94,6 +94,7 @@ pub fn open(p: *Pardes, id: usize, path: []const u8, line: usize) !*Pane {
 pub fn setContent(p: *Pardes, f: *File, new: []u8) void {
     p.gpa.free(f.content);
     f.content = new;
+    f.revision +%= 1;
     if (f.line_starts.len > 0) p.gpa.free(f.line_starts);
     f.line_starts = &.{};
     // the highlights go too, and not just because they are stale: their byte
diff --git a/src/gui/gui.zig b/src/gui/gui.zig
index 5744b3f2..6bd2c715 100644
--- a/src/gui/gui.zig
+++ b/src/gui/gui.zig
@@ -23,6 +23,7 @@ const look = @import("../look.zig");
 const deck = @import("deck.zig");
 const crt = @import("crt.zig");
 const fonts = @import("fonts.zig"); // the Font builtin's half of the seam
+const selection_pipe = @import("../selection_pipe.zig");
 
 const is_emscripten = builtin.os.tag == .emscripten;
 const temp_file = if (is_emscripten) struct {} else @import("../temp_file.zig");
@@ -578,6 +579,8 @@ const Msg = union(enum) {
     eof: struct { pane: u8, gen: u32, fd: c_int },
     /// a language query finished on its own thread (see lspThread)
     lsp: struct { id: u32, rows: []u8 },
+    /// a selection-filter worker finished; every stdout is gpa-owned
+    pipe: selection_pipe.Response,
     /// something happened in a watched directory (see watchThread)
     files_changed,
 };
@@ -610,6 +613,11 @@ const LspJob = struct {
     }
 };
 
+const PipeTask = struct {
+    id: u32,
+    future: std.Io.Future(anyerror!void),
+};
+
 const Queue = struct {
     gpa: std.mem.Allocator,
     sdl_wake: bool, // wake a blocking SDL_WaitEventTimeout on cross-thread push
@@ -630,6 +638,10 @@ const Queue = struct {
             switch (m) {
                 .output => |o| q.gpa.free(o.bytes),
                 .lsp => |l| q.gpa.free(l.rows),
+                .pipe => |response_value| {
+                    var response = response_value;
+                    response.deinit(q.gpa);
+                },
                 .eof, .files_changed => {},
             }
             return;
@@ -639,6 +651,10 @@ const Queue = struct {
             switch (m) {
                 .output => |o| q.gpa.free(o.bytes),
                 .lsp => |l| q.gpa.free(l.rows),
+                .pipe => |response_value| {
+                    var response = response_value;
+                    response.deinit(q.gpa);
+                },
                 .eof, .files_changed => {},
             }
             return;
@@ -667,6 +683,10 @@ const Queue = struct {
         for (q.items.items) |m| switch (m) {
             .output => |o| q.gpa.free(o.bytes),
             .lsp => |l| q.gpa.free(l.rows),
+            .pipe => |response_value| {
+                var response = response_value;
+                response.deinit(q.gpa);
+            },
             .eof, .files_changed => {},
         };
         q.items.deinit(q.gpa);
@@ -831,6 +851,35 @@ fn spawnLsp(core: *pardes.Pardes, gpa: std.mem.Allocator, q: *Queue, e: anytype)
     th.detach();
 }
 
+fn pipeThread(io: std.Io, gpa: std.mem.Allocator, job: *selection_pipe.Job, q: *Queue) anyerror!void {
+    defer job.deinit(gpa);
+    const response = selection_pipe.runJob(gpa, io, job);
+    q.push(.{ .pipe = response });
+}
+
+/// Copy every borrowed core byte before the tracked worker starts. The queue
+/// owns the response and already has close-time disposal for a late answer.
+fn spawnPipe(
+    core: *pardes.Pardes,
+    io: std.Io,
+    gpa: std.mem.Allocator,
+    q: *Queue,
+    tasks: *std.ArrayList(PipeTask),
+    request: anytype,
+) void {
+    const view = core.pipeRequest(request.id) orelse return;
+    const job = selection_pipe.Job.copy(gpa, view) catch return;
+    tasks.ensureUnusedCapacity(gpa, 1) catch {
+        job.deinit(gpa);
+        return;
+    };
+    const future = io.concurrent(pipeThread, .{ io, gpa, job, q }) catch {
+        job.deinit(gpa);
+        return;
+    };
+    tasks.appendAssumeCapacity(.{ .id = request.id, .future = future });
+}
+
 // ---- the renderer state ----
 
 const Gui = struct {
@@ -1192,6 +1241,11 @@ fn runNative(init: std.process.Init, opts_in: pardes.Options) !void {
     };
     var queue: Queue = .{ .gpa = gpa, .sdl_wake = true };
     defer queue.close();
+    var pipe_tasks: std.ArrayList(PipeTask) = .empty;
+    defer {
+        for (pipe_tasks.items) |*task| task.future.cancel(io) catch {};
+        pipe_tasks.deinit(gpa);
+    }
     // One inotify instance for every watched pane, opened here — before any
     // thread exists — so the pre-loop drain below can already mark the file a
     // positional path argument opened. -1 off linux: watchPane goes quiet and
@@ -1205,7 +1259,7 @@ fn runNative(init: std.process.Init, opts_in: pardes.Options) !void {
 
     // initial spawns BEFORE any worker thread exists: forkpty from a
     // multithreaded process can wedge the child before exec (see tty.zig).
-    drainEffects(core, &ptys, &gens, gpa, &queue, &g, inotify_fd, &watches, false);
+    drainEffects(core, &ptys, &gens, io, gpa, &queue, &pipe_tasks, &g, inotify_fd, &watches, false);
     for (&ptys, 0..) |*slot, id| if (slot.*) |pt| spawnReader(gpa, pt, @intCast(id), gens[id], &queue);
     // ...and the one file watcher. Started even with nothing marked yet: the fd
     // already exists and an unwatched inotify instance just parks in read(2) —
@@ -1252,6 +1306,20 @@ fn runNative(init: std.process.Init, opts_in: pardes.Options) !void {
                 core.update(.{ .lsp_resp = .{ .id = l.id, .rows = l.rows } });
                 gpa.free(l.rows);
             },
+            .pipe => |response_value| {
+                var response = response_value;
+                core.update(.{ .pipe_resp = .{
+                    .id = response.id,
+                    .success = response.success,
+                    .outputs = response.outputs,
+                } });
+                response.deinit(gpa);
+                for (pipe_tasks.items, 0..) |*task, i| if (task.id == response_value.id) {
+                    task.future.await(io) catch {};
+                    _ = pipe_tasks.orderedRemove(i);
+                    break;
+                };
+            },
             // Coalesced on purpose: a burst of writes (a formatter, a build, a
             // `git checkout`) collapses into ONE pass below, so it cannot queue
             // a reload — or an undo entry — per write.
@@ -1262,7 +1330,7 @@ fn runNative(init: std.process.Init, opts_in: pardes.Options) !void {
         // 3. steamdeck: poll gamepad axes into virtual cursor / wheel events
         pollGamepad(&g, core);
         // 4. effects
-        drainEffects(core, &ptys, &gens, gpa, &queue, &g, inotify_fd, &watches, true);
+        drainEffects(core, &ptys, &gens, io, gpa, &queue, &pipe_tasks, &g, inotify_fd, &watches, true);
         if (core.quit) break;
         // Restore builtin: swap in a core rebuilt from the dump; kill the live
         // shells (their detached readers wake on child death; gens bumped so
@@ -1669,11 +1737,16 @@ fn runGrid(init: std.process.Init, opts_in: pardes.Options) !void {
     };
     var queue: Queue = .{ .gpa = gpa, .sdl_wake = false };
     defer queue.close();
+    var pipe_tasks: std.ArrayList(PipeTask) = .empty;
+    defer {
+        for (pipe_tasks.items) |*task| task.future.cancel(io) catch {};
+        pipe_tasks.deinit(gpa);
+    }
     // no inotify here on purpose: this mode's whole contract is one frame per
     // scripted input event, and a reload that arrives on its own clock would
     // put a frame in the stream nothing asked for. -1 makes watchPane a no-op.
     var watches: [pardes.MAX_PANES]?Watch = @splat(null);
-    drainEffects(core, &ptys, &gens, gpa, &queue, null, -1, &watches, false);
+    drainEffects(core, &ptys, &gens, io, gpa, &queue, &pipe_tasks, null, -1, &watches, false);
     for (&ptys, 0..) |*slot, id| if (slot.*) |pt| spawnReader(gpa, pt, @intCast(id), gens[id], &queue);
 
     setStdinRaw() catch {}; // stdin may be a pipe, not a pty — best effort
@@ -1712,10 +1785,25 @@ fn runGrid(init: std.process.Init, opts_in: pardes.Options) !void {
                 gpa.free(l.rows);
                 n_events += 1;
             },
+            .pipe => |response_value| {
+                var response = response_value;
+                core.update(.{ .pipe_resp = .{
+                    .id = response.id,
+                    .success = response.success,
+                    .outputs = response.outputs,
+                } });
+                response.deinit(gpa);
+                n_events += 1;
+                for (pipe_tasks.items, 0..) |*task, i| if (task.id == response_value.id) {
+                    task.future.await(io) catch {};
+                    _ = pipe_tasks.orderedRemove(i);
+                    break;
+                };
+            },
             .files_changed => {}, // unreachable: no watcher thread in this mode
         };
         msgs.deinit(gpa);
-        drainEffects(core, &ptys, &gens, gpa, &queue, null, -1, &watches, true);
+        drainEffects(core, &ptys, &gens, io, gpa, &queue, &pipe_tasks, null, -1, &watches, true);
         pollCwds(core, &ptys);
         if (n_events == 0) continue; // idle tick: nothing changed, no frame
         _ = frame_arena.reset(.retain_capacity);
@@ -2423,8 +2511,10 @@ fn drainEffects(
     core: *pardes.Pardes,
     ptys: *[pardes.MAX_PANES]?Pty,
     gens: *[pardes.MAX_PANES]u32,
+    io: std.Io,
     gpa: std.mem.Allocator,
     queue: *Queue,
+    pipe_tasks: *std.ArrayList(PipeTask),
     g: ?*Gui, // null in grid test mode (no SDL: clipboard effects are no-ops)
     inotify_fd: c_int,
     watches: *[pardes.MAX_PANES]?Watch,
@@ -2507,6 +2597,7 @@ fn drainEffects(
             _ = c.SDL_SetClipboardText(z.ptr);
         },
         .lsp => |e| if (threads_ok) spawnLsp(core, gpa, queue, e),
+        .pipe => |e| if (threads_ok) spawnPipe(core, io, gpa, queue, pipe_tasks, e),
         .watch => |w| {
             // starting, the path and the on-disk bytes are read off the core
             // (same split as save_file); stopping, the pane is already gone
@@ -2557,7 +2648,7 @@ fn drainEffectsWeb(core: *pardes.Pardes, gpa: std.mem.Allocator, g: *Gui) void {
         .open_link => |url| openLinkWeb(gpa, url.slice()),
         // nothing to spawn/write/resize/save/dump into, and no filesystem to
         // watch — all no-ops
-        .spawn, .write, .resize_pty, .save_file, .new_file, .write_dump, .lsp, .watch, .quit => {},
+        .spawn, .write, .resize_pty, .save_file, .new_file, .write_dump, .lsp, .pipe, .watch, .quit => {},
     };
 }
 
diff --git a/src/pardes.zig b/src/pardes.zig
index 23611990..25b1ecb3 100644
--- a/src/pardes.zig
+++ b/src/pardes.zig
@@ -30,6 +30,7 @@ const term_pane = @import("term_pane.zig");
 const file_pane = @import("file_pane.zig");
 const output_pane = @import("output_pane.zig");
 const builtins = @import("builtins.zig");
+const selection_pipe = @import("selection_pipe.zig");
 pub const config = @import("config.zig");
 pub const pdf_enabled = @import("pardes_config").mupdf;
 const pdf_impl = if (pdf_enabled) @import("mupdf") else struct {};
@@ -1071,6 +1072,160 @@ test "PDF builtins and leader paths follow the MuPDF feature gate" {
     }
 }
 
+fn nextPipeEffect(p: *Pardes) ?u32 {
+    while (p.nextEffect()) |effect| switch (effect) {
+        .pipe => |request| return request.id,
+        else => {},
+    };
+    return null;
+}
+
+test "selection pipe prompt submits exact request and Escape cancels" {
+    const gpa = std.testing.allocator;
+    const p = try Pardes.init(gpa, .{ .tty_only = true });
+    defer p.deinit();
+    while (p.nextEffect()) |_| {}
+    const pane = try p.hxOpenFileContent("one\ntwo\n");
+    pane.cur_row = 0;
+    pane.cur_col = 2;
+    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };
+
+    p.update(.{ .key = .{ .cp = '|' } });
+    try std.testing.expect(pane.tag_edit and pane.pipe_edit);
+    try std.testing.expect(std.mem.endsWith(u8, pane.tag_tail.items, config.pipe_marker));
+    try std.testing.expect(nextPipeEffect(p) == null);
+    p.update(.{ .key = .{ .cp = 't', .text = "tr a-z A-Z" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    try std.testing.expect(!pane.tag_edit and !pane.pipe_edit);
+    const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    const request = p.pipeRequest(id) orelse return error.MissingPipeRequest;
+    try std.testing.expectEqualStrings("tr a-z A-Z", request.command);
+    try std.testing.expectEqualStrings("/", request.cwd);
+    try std.testing.expectEqual(@as(usize, 1), request.inputs.len);
+    try std.testing.expectEqualSlices(u8, "one", request.inputs[0].bytes);
+
+    const before = pane.file.?.content;
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.escape } });
+    try std.testing.expect(!pane.tag_edit and !pane.pipe_edit);
+    try std.testing.expectEqualSlices(u8, before, pane.file.?.content);
+    try std.testing.expect(nextPipeEffect(p) == null);
+}
+
+test "selection pipe replaces all ranges atomically and undo restores them" {
+    const gpa = std.testing.allocator;
+    const p = try Pardes.init(gpa, .{ .tty_only = true });
+    defer p.deinit();
+    while (p.nextEffect()) |_| {}
+    const pane = try p.hxOpenFileContent("aa bb cc\n");
+    const pl = try p.paneCursorLines(pane);
+    const ranges = [_]modal.HxRange{
+        .{ .anchor = 0, .head = 2 },
+        .{ .anchor = 6, .head = 8 },
+    };
+    Pardes.setPaneRanges(pane, pl, pane.file.?.content, &ranges, &.{}, 0, true);
+
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    const request = p.pipeRequest(id) orelse return error.MissingPipeRequest;
+    try std.testing.expectEqual(@as(usize, 2), request.inputs.len);
+    try std.testing.expectEqualSlices(u8, "aa", request.inputs[0].bytes);
+    try std.testing.expectEqualSlices(u8, "cc", request.inputs[1].bytes);
+
+    const outputs: []const []const u8 = &.{ "AA\n", "" };
+    p.update(.{ .pipe_resp = .{ .id = id, .success = true, .outputs = outputs } });
+    try std.testing.expectEqualSlices(u8, "AA\n bb \n", pane.file.?.content);
+    try std.testing.expectEqual(@as(usize, 1), pane.file.?.undo.items.len);
+    try std.testing.expectEqual(@as(u8, 1), pane.nsel);
+
+    p.update(.{ .key = .{ .cp = 'u' } });
+    try std.testing.expectEqualSlices(u8, "aa bb cc\n", pane.file.?.content);
+}
+
+test "selection pipe failure and stale completion never mutate the file" {
+    const gpa = std.testing.allocator;
+    const p = try Pardes.init(gpa, .{ .tty_only = true });
+    defer p.deinit();
+    while (p.nextEffect()) |_| {}
+    const pane = try p.hxOpenFileContent("abc\n");
+    pane.cur_col = 2;
+    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };
+
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const failed_id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    p.update(.{ .pipe_resp = .{ .id = failed_id, .success = false, .outputs = &.{} } });
+    try std.testing.expectEqualSlices(u8, "abc\n", pane.file.?.content);
+    try std.testing.expectEqual(@as(usize, 0), pane.file.?.undo.items.len);
+
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const stale_id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    file_pane.setContent(p, &pane.file.?, try gpa.dupe(u8, "changed\n"));
+    p.update(.{ .pipe_resp = .{ .id = stale_id, .success = true, .outputs = &.{"ABC"} } });
+    try std.testing.expectEqualSlices(u8, "changed\n", pane.file.?.content);
+    try std.testing.expectEqual(@as(usize, 0), pane.file.?.undo.items.len);
+}
+
+test "selection pipe rejects a reused pane slot and a superseded request" {
+    const gpa = std.testing.allocator;
+    const p = try Pardes.init(gpa, .{ .tty_only = true });
+    defer p.deinit();
+    while (p.nextEffect()) |_| {}
+    var pane = try p.hxOpenFileContent("old\n");
+    pane.cur_col = 2;
+    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };
+
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const replaced_id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    pane = try p.hxOpenFileContent("new\n"); // same slot, different serial
+    p.update(.{ .pipe_resp = .{ .id = replaced_id, .success = true, .outputs = &.{"OLD"} } });
+    try std.testing.expectEqualSlices(u8, "new\n", pane.file.?.content);
+    try std.testing.expectEqual(@as(usize, 0), pane.file.?.undo.items.len);
+
+    pane.cur_col = 2;
+    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "cat" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const first_id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    p.update(.{ .key = .{ .cp = '|' } });
+    p.update(.{ .key = .{ .cp = 'c', .text = "tr a-z A-Z" } });
+    p.update(.{ .key = .{ .cp = Key.enter } });
+    const latest_id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
+    try std.testing.expect(first_id != latest_id);
+
+    p.update(.{ .pipe_resp = .{ .id = first_id, .success = true, .outputs = &.{"BAD"} } });
+    try std.testing.expectEqualSlices(u8, "new\n", pane.file.?.content);
+    p.update(.{ .pipe_resp = .{ .id = latest_id, .success = true, .outputs = &.{"NEW"} } });
+    try std.testing.expectEqualSlices(u8, "NEW\n", pane.file.?.content);
+    try std.testing.expectEqual(@as(usize, 1), pane.file.?.undo.items.len);
+}
+
+test "selection pipe binding is file-normal-only" {
+    const gpa = std.testing.allocator;
+    const p = try Pardes.init(gpa, .{ .tty_only = true });
+    defer p.deinit();
+    while (p.nextEffect()) |_| {}
+    const pane = try p.hxOpenFileContent("x");
+
+    pane.file.?.output = .{ .from = .search };
+    p.update(.{ .key = .{ .cp = '|' } });
+    try std.testing.expect(!pane.pipe_edit);
+    pane.file.?.output = null;
+    pane.mode = .insert;
+    p.update(.{ .key = .{ .cp = '|', .text = "|" } });
+    try std.testing.expect(!pane.pipe_edit);
+    try std.testing.expectEqualSlices(u8, "|x", pane.file.?.content);
+}
+
 test "startup config runs builtin lines in order and isolates bad lines" {
     const p = try Pardes.init(std.testing.allocator, .{
         .startup_config =
@@ -2003,6 +2158,10 @@ pub const Event = union(enum) {
     /// stale answer (the pane was closed, or a newer query superseded it) and
     /// is dropped.
     lsp_resp: struct { id: u32, rows: []const u8 },
+    /// A selection-pipe worker finished. Every output is borrowed for this
+    /// update only; success is atomic, so a failed/nonzero invocation carries
+    /// no usable outputs and changes nothing.
+    pipe_resp: struct { id: u32, success: bool, outputs: []const []const u8 },
     /// a file the shell was asked to watch changed on disk; `bytes` are its
     /// new contents, borrowed for this call exactly like `output`. A shell
     /// with no filesystem (the browser) or no watcher simply never sends one —
@@ -2043,6 +2202,9 @@ pub const Effect = union(enum) {
     /// core (like save_file) and must SNAPSHOT them before the worker starts —
     /// the core keeps editing while this is in flight.
     lsp: struct { id: u32, kind: lsp.Kind, pane: u8, offset: u32, arg: Buf(128) },
+    /// Snapshot the matching request with pipeRequest(id), then run it away
+    /// from the UI/event loop and answer with pipe_resp.
+    pipe: struct { id: u32 },
     /// start (`on`) or stop watching this pane's file on disk. Starting, the
     /// shell reads the path off the core exactly like save_file does; stopping
     /// carries nothing, because by the time an `off` is drained the pane is
@@ -2169,6 +2331,10 @@ pub const FileSnap = struct {
 pub const File = struct {
     path: []u8,
     content: []u8,
+    /// Monotonic content identity for asynchronous edits. Every content swap
+    /// goes through file_pane.setContent, which bumps this; a pipe completion
+    /// accepted against another revision would overwrite intervening work.
+    revision: u32 = 0,
     /// set = this is an OUTPUT buffer (acme's +Errors): a file pane with no
     /// file behind it, showing text the core produced itself. It records the
     /// COMMAND that opened it, and output_pane.zig's one table turns that into
@@ -2396,6 +2562,11 @@ pub const Pane = struct {
     /// output_pane.resultsFrom is how `]d` asks.
     search_edit: bool = false,
     search_at: u16 = 0,
+    /// `|` reuses the visible tag-tail input but submits a shell-filter
+    /// request rather than a search. pipe_at is a tail byte offset exactly
+    /// like search_at; Enter/Esc restore the tail to it.
+    pipe_edit: bool = false,
+    pipe_at: u16 = 0,
     search_pane: ?usize = null,
     search_row: ?usize = null,
     /// The selection an `s`/`S` input was armed on, as gap offsets over the
@@ -2725,6 +2896,36 @@ pub const Options = struct {
     rows: u16 = 24,
 };
 
+const PendingPipe = struct {
+    id: u32,
+    pane: usize,
+    serial: u32,
+    revision: u32,
+    command: []u8,
+    cwd: []u8,
+    inputs: []selection_pipe.Input,
+    ranges: [MAX_SELS]modal.HxRange,
+    primary: u8,
+    explicit: bool,
+
+    fn deinit(wait: *PendingPipe, gpa: std.mem.Allocator) void {
+        gpa.free(wait.command);
+        gpa.free(wait.cwd);
+        for (wait.inputs) |input| gpa.free(@constCast(input.bytes));
+        gpa.free(wait.inputs);
+        wait.* = undefined;
+    }
+
+    fn request(wait: *const PendingPipe) selection_pipe.Request {
+        return .{
+            .id = wait.id,
+            .command = wait.command,
+            .cwd = wait.cwd,
+            .inputs = wait.inputs,
+        };
+    }
+};
+
 pub const Pardes = struct {
     gpa: std.mem.Allocator,
     opts: Options,
@@ -2825,6 +3026,12 @@ pub const Pardes = struct {
     lsp_seq: u32 = 0,
     lsp_wait: ?struct { id: u32, kind: lsp.Kind, pane: usize, arg: Effect.Buf(128) } = null,
 
+    /// One current shell-filter request. A newer submit frees and supersedes
+    /// it; old worker answers then fail the id check. The request itself owns
+    /// every byte a shell snapshots while draining the id-only effect.
+    pipe_seq: u32 = 0,
+    pipe_wait: ?PendingPipe = null,
+
     /// pending effects, drained by the shell after each update. Growable: one
     /// update can emit unbounded writes (a big selection executed is chunked
     /// into 64-byte write effects) — a fixed ring overflows and panics.
@@ -2910,6 +3117,7 @@ pub const Pardes = struct {
         if (p.dump_out) |d| gpa.free(d);
         if (p.last_dump) |d| gpa.free(d);
         if (p.chord_arg) |a| gpa.free(a);
+        if (p.pipe_wait) |*wait| wait.deinit(gpa);
         p.effects.deinit(gpa);
         p.scratch.deinit();
         gpa.free(p.surface.cells);
@@ -3172,6 +3380,7 @@ pub const Pardes = struct {
             },
             .eof => |e| p.removePane(e.pane),
             .lsp_resp => |r| p.lspResponse(r.id, r.rows),
+            .pipe_resp => |r| p.pipeResponse(r.id, r.success, r.outputs),
             .file_changed => |fc| p.fileChanged(fc.pane, fc.bytes),
             .key => |key| p.handleKey(key),
             .mouse => |m| p.handleMouse(m),
@@ -3301,6 +3510,7 @@ pub const Pardes = struct {
         pane.tag_edit = false;
         pane.tag_sel = false;
         pane.search_edit = false; // an abandoned search input stays as tag text
+        pane.pipe_edit = false;
         pane.nsel_snap = 0; // ...and its s/S preview keeps whatever it previewed
     }
 
@@ -3761,6 +3971,17 @@ pub const Pardes = struct {
             } else p.enterTty(p.active);
             return;
         }
+        // `|` owns the same visible one-line tag input as search, but Enter
+        // snapshots an asynchronous shell filter. Escape is a pure cancel:
+        // restore the old tail and never emit a request.
+        if (pane.pipe_edit and (hit(key, config.search_submit) or hit(key, config.escape))) {
+            if (hit(key, config.search_submit)) p.submitPipe(p.active);
+            pane.tag_tail.shrinkRetainingCapacity(@min(pane.pipe_at, pane.tag_tail.items.len));
+            exitTagEdit(pane);
+            pane.mode = .normal;
+            pane.pending = 0;
+            return;
+        }
         // a search input in flight (`/` or Find): Enter searches, Esc abandons;
         // both restore the tag tail and hand focus back to the body.
         if (pane.search_edit and (hit(key, config.search_submit) or hit(key, config.escape))) {
@@ -5055,6 +5276,7 @@ pub const Pardes = struct {
             hit(key, config.comment_toggle) or
             hit(key, config.undo) or hit(key, config.redo) or
             hit(key, config.command_line) or hit(key, config.search) or
+            hit(key, config.pipe_selection) or
             hit(key, config.search_next) or hit(key, config.search_prev) or
             hit(key, config.leader);
     }
@@ -5424,6 +5646,9 @@ pub const Pardes = struct {
             if (pane.tag_edit) pane.mode = .normal;
             return;
         }
+        // `|` — Helix's pipe command. Only a real, savable file can be
+        // replaced; output buffers, terminals, images and PDFs stay inert.
+        if (hit(key, config.pipe_selection)) return p.startPipe(pane);
         // search: `/` types a pattern into the tag, n/N walk the results. Same
         // keys on every kind of pane; a terminal with no search armed falls
         // back to n/N as a motion over the lookable tokens in its output.
@@ -5436,6 +5661,151 @@ pub const Pardes = struct {
         }
     }
 
+    // ---- selection pipe (`|`): visible prompt, async shell, atomic edit ----
+
+    fn startPipe(p: *Pardes, pane: *Pane) void {
+        const f = pane.file orelse return;
+        if (!output_pane.fileTraits(f.output).saves) return;
+        if (!pane.tag_init) {
+            pane.tag_tail.appendSlice(p.gpa, curTail(pane)) catch return;
+            pane.tag_init = true;
+        }
+        pane.pipe_at = @intCast(pane.tag_tail.items.len);
+        pane.tag_tail.appendSlice(p.gpa, config.pipe_marker) catch return;
+        pane.pipe_edit = true;
+        pane.tag_edit = true;
+        pane.tag_sel = false;
+        pane.mode = .insert;
+        pane.pending = 0;
+        pane.tag_col = @intCast((p.tagPrefix(pane) catch return).len + pane.tag_tail.items.len);
+    }
+
+    /// Snapshot command/cwd/ranges/selection bytes before emitting the id-only
+    /// effect. Every allocation is owned by pipe_wait, so the frontend can
+    /// copy it synchronously and the core can keep editing immediately after.
+    fn submitPipe(p: *Pardes, id: usize) void {
+        const pane = p.panes[id] orelse return;
+        const f = pane.file orelse return;
+        if (!output_pane.fileTraits(f.output).saves) return;
+        const tail = pane.tag_tail.items;
+        const armed = tail[@min(pane.pipe_at, tail.len)..];
+        if (!std.mem.startsWith(u8, armed, config.pipe_marker)) return;
+        const command = armed[config.pipe_marker.len..];
+        if (command.len == 0) return;
+
+        var ranges: [MAX_SELS]modal.HxRange = undefined;
+        const got = paneRanges(pane, f.content, 0, &ranges);
+        const inputs = p.gpa.alloc(selection_pipe.Input, got.n) catch return;
+        var made: usize = 0;
+        for (ranges[0..got.n], 0..) |range, i| {
+            const lo = @min(range.anchor, range.head);
+            const hi = @max(range.anchor, range.head);
+            if (hi > f.content.len) break;
+            const copy = p.gpa.dupe(u8, f.content[lo..hi]) catch break;
+            inputs[i] = .{ .bytes = copy };
+            made += 1;
+        }
+        if (made != got.n) {
+            for (inputs[0..made]) |input| p.gpa.free(@constCast(input.bytes));
+            p.gpa.free(inputs);
+            return;
+        }
+        const command_copy = p.gpa.dupe(u8, command) catch {
+            for (inputs) |input| p.gpa.free(@constCast(input.bytes));
+            p.gpa.free(inputs);
+            return;
+        };
+        const cwd_copy = p.gpa.dupe(u8, paneDir(pane)) catch {
+            p.gpa.free(command_copy);
+            for (inputs) |input| p.gpa.free(@constCast(input.bytes));
+            p.gpa.free(inputs);
+            return;
+        };
+
+        p.pipe_seq +%= 1;
+        if (p.pipe_wait) |*old| old.deinit(p.gpa);
+        p.pipe_wait = .{
+            .id = p.pipe_seq,
+            .pane = id,
+            .serial = pane.serial,
+            .revision = f.revision,
+            .command = command_copy,
+            .cwd = cwd_copy,
+            .inputs = inputs,
+            .ranges = ranges,
+            .primary = @intCast(got.pri),
+            .explicit = (pane.vsel.active and pane.vsel.explicit) or pane.msel.active,
+        };
+        p.emit(.{ .pipe = .{ .id = p.pipe_seq } });
+    }
+
+    /// Frontends call this while draining a .pipe effect, then immediately
+    /// copy the borrowed view into a worker-owned selection_pipe.Job.
+    pub fn pipeRequest(p: *const Pardes, id: u32) ?selection_pipe.Request {
+        if (p.pipe_wait) |*wait| {
+            if (wait.id == id) return wait.request();
+        }
+        return null;
+    }
+
+    fn pipeResponse(p: *Pardes, id: u32, success: bool, outputs: []const []const u8) void {
+        if (p.pipe_wait == null or p.pipe_wait.?.id != id) return;
+        var wait = p.pipe_wait.?;
+        p.pipe_wait = null;
+        defer wait.deinit(p.gpa);
+        if (!success or outputs.len != wait.inputs.len) return;
+
+        const pane = p.panes[wait.pane] orelse return;
+        if (pane.serial != wait.serial) return;
+        const f = if (pane.file) |*file| file else return;
+        if (!output_pane.fileTraits(f.output).saves or f.revision != wait.revision) return;
+
+        var total_output: usize = 0;
+        var removed: usize = 0;
+        var previous_end: usize = 0;
+        for (outputs, wait.ranges[0..outputs.len]) |output, range| {
+            if (output.len > selection_pipe.max_stdout_bytes) return;
+            total_output = std.math.add(usize, total_output, output.len) catch return;
+            if (total_output > selection_pipe.max_total_stdout_bytes) return;
+            const lo = @min(range.anchor, range.head);
+            const hi = @max(range.anchor, range.head);
+            if (lo < previous_end or hi > f.content.len) return;
+            removed = std.math.add(usize, removed, hi - lo) catch return;
+            previous_end = hi;
+        }
+        const kept = f.content.len - removed;
+        const final_len = std.math.add(usize, kept, total_output) catch return;
+        const replacement = p.gpa.alloc(u8, final_len) catch return;
+
+        var new_ranges: [MAX_SELS]modal.HxRange = undefined;
+        var read_at: usize = 0;
+        var write_at: usize = 0;
+        for (outputs, wait.ranges[0..outputs.len], 0..) |output, range, i| {
+            const lo = @min(range.anchor, range.head);
+            const hi = @max(range.anchor, range.head);
+            @memcpy(replacement[write_at .. write_at + (lo - read_at)], f.content[read_at..lo]);
+            write_at += lo - read_at;
+            const out_start = write_at;
+            @memcpy(replacement[write_at .. write_at + output.len], output);
+            write_at += output.len;
+            const out_end = write_at;
+            new_ranges[i] = if (range.anchor > range.head)
+                .{ .anchor = out_end, .head = out_start }
+            else
+                .{ .anchor = out_start, .head = out_end };
+            read_at = hi;
+        }
+        @memcpy(replacement[write_at..], f.content[read_at..]);
+
+        // One async request is one history transaction, even at 64 cursors.
+        p.pushUndo(pane);
+        file_pane.setContent(p, f, replacement);
+        const pl = p.paneCursorLines(pane) catch return;
+        setPaneRanges(pane, pl, f.content, new_ranges[0..outputs.len], &.{}, wait.primary, wait.explicit);
+        pane.select = false;
+        pane.ensureCursorVisible();
+    }
+
     // ---- search (`/`, Find, n/N): results into an output buffer, look the rows ----
 
     const Search = enum { text, find, grep };
diff --git a/src/selection_pipe.zig b/src/selection_pipe.zig
new file mode 100644
index 00000000..65c4c7fd
--- /dev/null
+++ b/src/selection_pipe.zig
@@ -0,0 +1,214 @@
+//! Native runner and boundary-owned values for `|`: pipe every editor
+//! selection through one shell command. The core imports only the plain value
+//! types below; native shells call `Job.copy` before starting a worker, then
+//! `runOne` on that worker. No subprocess or borrowed core memory reaches the
+//! editor/event-loop thread.
+const std = @import("std");
+
+/// A deliberately finite answer. One selection cannot retain more than 1 MiB
+/// and a multi-selection command cannot retain more than 4 MiB in total.
+/// stderr is diagnostic-only and has a smaller independent ceiling. Crossing
+/// any ceiling fails the whole atomic request.
+pub const max_stdout_bytes: usize = 1024 * 1024;
+pub const max_total_stdout_bytes: usize = 4 * 1024 * 1024;
+pub const max_stderr_bytes: usize = 64 * 1024;
+pub const command_timeout_seconds: u64 = 10;
+
+pub const Input = struct { bytes: []const u8 };
+
+/// Borrowed view exposed by the core while an effect is being drained.
+pub const Request = struct {
+    id: u32,
+    command: []const u8,
+    cwd: []const u8,
+    inputs: []const Input,
+};
+
+/// Worker-owned snapshot. `copy` is intentionally called synchronously while
+/// draining the effect: the worker can start after arbitrary later edits and
+/// still owns exactly the command, directory, and selection bytes submitted.
+pub const Job = struct {
+    id: u32,
+    command: []u8,
+    cwd: []u8,
+    inputs: [][]u8,
+
+    pub fn copy(gpa: std.mem.Allocator, request: Request) !*Job {
+        const job = try gpa.create(Job);
+        errdefer gpa.destroy(job);
+        job.* = .{
+            .id = request.id,
+            .command = try gpa.dupe(u8, request.command),
+            .cwd = &.{},
+            .inputs = &.{},
+        };
+        errdefer gpa.free(job.command);
+        job.cwd = try gpa.dupe(u8, request.cwd);
+        errdefer gpa.free(job.cwd);
+        job.inputs = try gpa.alloc([]u8, request.inputs.len);
+        errdefer gpa.free(job.inputs);
+        var made: usize = 0;
+        errdefer for (job.inputs[0..made]) |input| gpa.free(input);
+        for (request.inputs, 0..) |input, i| {
+            job.inputs[i] = try gpa.dupe(u8, input.bytes);
+            made += 1;
+        }
+        return job;
+    }
+
+    pub fn deinit(job: *Job, gpa: std.mem.Allocator) void {
+        gpa.free(job.command);
+        gpa.free(job.cwd);
+        for (job.inputs) |input| gpa.free(input);
+        gpa.free(job.inputs);
+        gpa.destroy(job);
+    }
+};
+
+/// One worker answer. `outputs` owns each slice; a failure owns an empty list.
+pub const Response = struct {
+    id: u32,
+    success: bool,
+    outputs: [][]u8,
+
+    pub fn deinit(response: *Response, gpa: std.mem.Allocator) void {
+        for (response.outputs) |output| gpa.free(output);
+        if (response.outputs.len > 0) gpa.free(response.outputs);
+        response.* = undefined;
+    }
+};
+
+const WriterContext = struct {
+    io: std.Io,
+    file: std.Io.File,
+    input: []const u8,
+    ok: bool = false,
+};
+
+fn writeInput(context: *WriterContext) void {
+    defer context.file.close(context.io);
+    context.file.writeStreamingAll(context.io, context.input) catch return;
+    context.ok = true;
+}
+
+/// Run one POSIX shell command with exact stdin, concurrently draining stdout
+/// and stderr so a producer cannot deadlock against a full pipe. The caller is
+/// already a worker. Nonzero exit, signal, timeout, IO failure, or either
+/// output ceiling is reported as `null`; only an exited-zero command transfers
+/// ownership of stdout to the caller.
+pub fn runOne(
+    gpa: std.mem.Allocator,
+    io: std.Io,
+    command: []const u8,
+    cwd: []const u8,
+    input: []const u8,
+) ?[]u8 {
+    if (std.mem.indexOfScalar(u8, command, 0) != null or
+        std.mem.indexOfScalar(u8, cwd, 0) != null) return null;
+
+    var child = std.process.spawn(io, .{
+        .argv = &.{ "/bin/sh", "-c", command },
+        .cwd = if (cwd.len == 0) .inherit else .{ .path = cwd },
+        .stdin = .pipe,
+        .stdout = .pipe,
+        .stderr = .pipe,
+    }) catch return null;
+
+    var writer_context: WriterContext = .{
+        .io = io,
+        .file = child.stdin.?,
+        .input = input,
+    };
+    child.stdin = null; // writer_context owns and closes this endpoint
+    var writer: ?std.Thread = std.Thread.spawn(.{}, writeInput, .{&writer_context}) catch {
+        writer_context.file.close(io);
+        child.kill(io);
+        return null;
+    };
+    // On every early return kill first, unblocking a command which never read
+    // stdin, then join the short-lived writer before its borrowed input dies.
+    defer if (writer) |thread| thread.join();
+    defer child.kill(io);
+
+    var multi_reader_buffer: std.Io.File.MultiReader.Buffer(2) = undefined;
+    var multi_reader: std.Io.File.MultiReader = undefined;
+    multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ child.stdout.?, child.stderr.? });
+    defer multi_reader.deinit();
+
+    const stdout_reader = multi_reader.reader(0);
+    const stderr_reader = multi_reader.reader(1);
+    const deadline: std.Io.Timeout = (std.Io.Timeout{ .duration = .{
+        .clock = .awake,
+        .raw = .fromSeconds(command_timeout_seconds),
+    } }).toDeadline(io);
+    while (multi_reader.fill(64, deadline)) |_| {
+        if (stdout_reader.buffered().len > max_stdout_bytes or
+            stderr_reader.buffered().len > max_stderr_bytes) return null;
+    } else |err| switch (err) {
+        error.EndOfStream => {},
+        else => return null,
+    }
+    if (stdout_reader.buffered().len > max_stdout_bytes or
+        stderr_reader.buffered().len > max_stderr_bytes) return null;
+    multi_reader.checkAnyError() catch return null;
+    const term = child.wait(io) catch return null;
+    writer.?.join();
+    writer = null;
+
+    const stdout = multi_reader.toOwnedSlice(0) catch return null;
+    const stderr = multi_reader.toOwnedSlice(1) catch {
+        gpa.free(stdout);
+        return null;
+    };
+    defer gpa.free(stderr);
+    const exited_zero = switch (term) {
+        .exited => |code| code == 0,
+        else => false,
+    };
+    if (!writer_context.ok or !exited_zero) {
+        gpa.free(stdout);
+        return null;
+    }
+    return stdout;
+}
+
+/// Invoke the command independently for every selection. The response is all
+/// or nothing: one failure frees every earlier stdout and returns a failed,
+/// empty answer for the core to ignore.
+pub fn runJob(gpa: std.mem.Allocator, io: std.Io, job: *const Job) Response {
+    var response: Response = .{ .id = job.id, .success = false, .outputs = &.{} };
+    const outputs = gpa.alloc([]u8, job.inputs.len) catch return response;
+    var made: usize = 0;
+    var total: usize = 0;
+    for (job.inputs, 0..) |input, i| {
+        const output = runOne(gpa, io, job.command, job.cwd, input) orelse break;
+        if (std.math.add(usize, total, output.len) catch null) |next_total| {
+            if (next_total <= max_total_stdout_bytes) {
+                outputs[i] = output;
+                total = next_total;
+                made += 1;
+                continue;
+            }
+        }
+        gpa.free(output);
+        break;
+    }
+    if (made != job.inputs.len) {
+        for (outputs[0..made]) |output| gpa.free(output);
+        gpa.free(outputs);
+        return response;
+    }
+    response.success = true;
+    response.outputs = outputs;
+    return response;
+}
+
+test "native pipe runner preserves stdin/stdout bytes and rejects failure" {
+    const gpa = std.testing.allocator;
+    const io = std.testing.io;
+    const output = runOne(gpa, io, "printf 'prefix:'; cat; printf '\\n'", "/tmp", "a\x00b\n") orelse
+        return error.PipeCommandFailed;
+    defer gpa.free(output);
+    try std.testing.expectEqualSlices(u8, "prefix:a\x00b\n\n", output);
+    try std.testing.expect(runOne(gpa, io, "printf ignored; exit 7", "/tmp", "") == null);
+}
diff --git a/src/tty/tty.zig b/src/tty/tty.zig
index 90477d06..908f17d7 100644
--- a/src/tty/tty.zig
+++ b/src/tty/tty.zig
@@ -12,6 +12,7 @@ const pardes = @import("../pardes.zig");
 const tracy = @import("../tracy.zig");
 const look = @import("../look.zig");
 const temp_file = @import("../temp_file.zig");
+const selection_pipe = @import("../selection_pipe.zig");
 
 extern "c" fn forkpty(amaster: *c_int, name: ?[*:0]u8, termp: ?*const anyopaque, winp: ?*const posix.winsize) c_int;
 extern "c" fn execv(path: [*:0]const u8, argv: [*:null]const ?[*:0]const u8) c_int;
@@ -38,6 +39,8 @@ pub const Command = struct {
         paste: []const u8,
         /// a language query finished on a worker; rows are gpa-owned
         lsp_done: struct { id: u32, rows: []u8 },
+        /// a selection-filter worker finished; every stdout is gpa-owned
+        pipe_done: selection_pipe.Response,
         /// something happened in a watched directory (see watchFiles)
         files_changed,
     } = .nop;
@@ -86,6 +89,11 @@ const LspJob = struct {
     }
 };
 
+const PipeTask = struct {
+    id: u32,
+    future: std.Io.Future(anyerror!void),
+};
+
 const Pty = struct {
     file: std.Io.File,
     pid: posix.pid_t,
@@ -387,6 +395,9 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
     var gens: [pardes.MAX_PANES]u32 = @splat(0);
     // the single in-flight language query (see the .lsp effect)
     var lsp_task: ?std.Io.Future(anyerror!void) = null;
+    // Selection filters may overlap: a second submit supersedes the first in
+    // core without synchronously canceling a possibly slow shell command.
+    var pipe_tasks: std.ArrayList(PipeTask) = .empty;
     // One inotify instance for every watched pane, opened here — before any
     // thread exists — so the pre-loop effect drain below can already mark the
     // file a positional path argument opened. -1 off linux: watchPane goes
@@ -411,6 +422,8 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             t.cancel(io) catch {};
             lsp_task = null;
         }
+        for (pipe_tasks.items) |*task| task.future.cancel(io) catch {};
+        pipe_tasks.deinit(gpa);
         // same contract as the pty readers: cancel unblocks the watcher's
         // read, and only then is the fd safe to close
         if (watch_task) |*t| {
@@ -425,13 +438,17 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             .pty_read => |pr| gpa.free(pr.bytes),
             .paste => |b| gpa.free(@constCast(b)),
             .lsp_done => |d| gpa.free(d.rows),
+            .pipe_done => |response_value| {
+                var response = response_value;
+                response.deinit(gpa);
+            },
             else => {},
         };
     }
 
     // Perform the initial spawns BEFORE any worker thread exists: forkpty from
     // a multithreaded process can wedge the child before exec.
-    drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, inotify_fd, &watches, false);
+    drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, &pipe_tasks, inotify_fd, &watches, false);
 
     try loop.start();
     defer loop.stop();
@@ -589,6 +606,20 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
                         lsp_task = null;
                     }
                 },
+                .pipe_done => |response_value| {
+                    var response = response_value;
+                    core.update(.{ .pipe_resp = .{
+                        .id = response.id,
+                        .success = response.success,
+                        .outputs = response.outputs,
+                    } });
+                    response.deinit(gpa);
+                    for (pipe_tasks.items, 0..) |*task, i| if (task.id == response_value.id) {
+                        task.future.await(io) catch {};
+                        _ = pipe_tasks.orderedRemove(i);
+                        break;
+                    };
+                },
             }
             batch += 1;
             if (stop or output or native_pdf_page_changed or batch >= 64) break;
@@ -619,7 +650,7 @@ pub fn run(init: std.process.Init, opts: pardes.Options) !void {
             }
         }
 
-        drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, inotify_fd, &watches, true);
+        drainEffects(core, &ptys, &gens, io, gpa, &loop, &vx, &tty, &lsp_task, &pipe_tasks, inotify_fd, &watches, true);
         if (core.quit) break;
 
         // Restore builtin: swap in a core rebuilt from the dump; the live
@@ -781,6 +812,7 @@ fn drainEffects(
     vx: *vaxis.Vaxis,
     tty: *vaxis.Tty,
     lsp_task: *?std.Io.Future(anyerror!void),
+    pipe_tasks: *std.ArrayList(PipeTask),
     inotify_fd: c_int,
     watches: *[pardes.MAX_PANES]?Watch,
     threads_ok: bool,
@@ -910,6 +942,20 @@ fn drainEffects(
                 continue;
             };
         },
+        .pipe => |request| {
+            if (!threads_ok) continue;
+            const view = core.pipeRequest(request.id) orelse continue;
+            const job = selection_pipe.Job.copy(gpa, view) catch continue;
+            pipe_tasks.ensureUnusedCapacity(gpa, 1) catch {
+                job.deinit(gpa);
+                continue;
+            };
+            const future = io.concurrent(pipeWorker, .{ io, gpa, job, loop }) catch {
+                job.deinit(gpa);
+                continue;
+            };
+            pipe_tasks.appendAssumeCapacity(.{ .id = request.id, .future = future });
+        },
         .watch => |w| {
             // starting, the path and the on-disk bytes are read off the core
             // (same split as save_file); stopping, the pane is already gone
@@ -949,6 +995,17 @@ fn lspWorker(gpa: std.mem.Allocator, job: *LspJob, loop: *Loop) anyerror!void {
     return;
 }
 
+fn pipeWorker(
+    io: std.Io,
+    gpa: std.mem.Allocator,
+    job: *selection_pipe.Job,
+    loop: *Loop,
+) anyerror!void {
+    defer job.deinit(gpa);
+    var response = selection_pipe.runJob(gpa, io, job);
+    loop.postEvent(.{ .pipe_done = response }) catch response.deinit(gpa);
+}
+
 /// Mark or unmark one pane's file (`path` null = unmark). Linux only: on any
 /// other OS this returns silently, the core never receives a file_changed
 /// event, and the feature is simply off — a state the core has to tolerate
diff --git a/src/web.zig b/src/web.zig
index 6ceca904..42359329 100644
--- a/src/web.zig
+++ b/src/web.zig
@@ -188,6 +188,37 @@ export fn pardes_lsp_response(id: u32, len_arg: u32) void {
     s.core.update(.{ .lsp_resp = .{ .id = id, .rows = input[0..len] } });
 }
 
+/// Complete host-handled effect kind 12. On success `input` is
+/// `[count:u32][count * output_len:u32][output bytes...]`, little-endian.
+/// A static browser has no process host and simply never calls this; an
+/// embedding such as Electron can run the request without another WASM ABI.
+export fn pardes_pipe_response(id: u32, success_arg: u32, len_arg: u32) void {
+    const s = &(state orelse return);
+    if (success_arg == 0) {
+        s.core.update(.{ .pipe_resp = .{ .id = id, .success = false, .outputs = &.{} } });
+        return;
+    }
+    const len: usize = @min(len_arg, input.len);
+    if (len < 4) return;
+    const count: usize = std.mem.readInt(u32, input[0..4], .little);
+    if (count == 0 or count > pardes.MAX_SELS) return;
+    const header_len = std.math.add(usize, 4, std.math.mul(usize, count, 4) catch return) catch return;
+    if (header_len > len) return;
+    var outputs: [pardes.MAX_SELS][]const u8 = undefined;
+    var at = header_len;
+    for (0..count) |i| {
+        const off = 4 + i * 4;
+        const encoded_len: *const [4]u8 = @ptrCast(&input[off]);
+        const output_len: usize = std.mem.readInt(u32, encoded_len, .little);
+        const end = std.math.add(usize, at, output_len) catch return;
+        if (end > len) return;
+        outputs[i] = input[at..end];
+        at = end;
+    }
+    if (at != len) return;
+    s.core.update(.{ .pipe_resp = .{ .id = id, .success = true, .outputs = outputs[0..count] } });
+}
+
 export fn pardes_file_changed(pane: u32, len_arg: u32) void {
     const s = &(state orelse return);
     if (pane >= pardes.MAX_PANES) return;
@@ -369,6 +400,22 @@ export fn pardes_effect_next() u32 {
             putEffect(s, e.arg.slice());
             break :blk 8;
         },
+        // Process-capable embeddings receive exact command/cwd/input bytes:
+        // [command_len][cwd_len][count][input lengths...][all payloads...].
+        // The ordinary static browser surfaces this as `pardes-io` and stays
+        // inert because it has no process authority.
+        .pipe => |e| blk: {
+            effect_aux0 = e.id;
+            const request = s.core.pipeRequest(e.id) orelse break :blk 12;
+            putU32(s, @intCast(request.command.len));
+            putU32(s, @intCast(request.cwd.len));
+            putU32(s, @intCast(request.inputs.len));
+            for (request.inputs) |selection| putU32(s, @intCast(selection.bytes.len));
+            appendEffect(s, request.command);
+            appendEffect(s, request.cwd);
+            for (request.inputs) |selection| appendEffect(s, selection.bytes);
+            break :blk 12;
+        },
         .watch => |e| blk: {
             effect_aux0 = e.pane;
             effect_aux1 = @intFromBool(e.on);