pipe selections through shell commands

1 files changed, 96 insertions, 5 deletions

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 => {},
     };
 }