path: root/src/selection_pipe.zig

//! `|`: pipe every editor selection through one shell command. First the
//! native runner and its boundary-owned values: native shells call `Job.copy`
//! before starting a worker, then `runOne` on that worker, so no subprocess or
//! borrowed core memory reaches the editor/event-loop thread. Then the editor
//! side: the prompt, the request the core emits, and the atomic edit its
//! answer makes.
const std = @import("std");
const filesystem = @import("fs.zig");
const pardes = @import("pardes.zig");
const edit = @import("edit.zig");
const modal = @import("modal.zig");
const panes = @import("panes.zig");
const config = @import("config.zig");
const Pane = panes.Pane;
const Key = pardes.Key;
const Pardes = pardes.Pardes;

/// 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, filesystem.localPath(request.cwd) orelse 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);
    }
};

/// WHY a filter produced nothing, carried home so somebody can be told.
///
/// Until this existed the runner read the command's stderr into memory and
/// then FREED IT UNREAD — the one artifact that explains a failure, discarded
/// two lines after it arrived — and every caller answered a failed filter with
/// a bare `return`. `| trr a-z A-Z` (a typo), `| grep nomatch` (exit 1),
/// `| jq .` on bad JSON: all of them did nothing, said nothing, and left the
/// text alone with no way to find out why.
pub const Failure = struct {
    pub const Kind = enum {
        /// The command never started: no `/bin/sh`, a cwd that is gone, a NUL
        /// in the command, a fork that failed.
        spawn,
        /// Still running at `command_timeout_seconds`.
        timeout,
        /// Past `max_stdout_bytes` / `max_stderr_bytes` / the job total.
        too_large,
        /// Ran, and exited nonzero. `code` says which.
        exit,
        /// Killed by a signal.
        signal,
        /// A read, a write or an allocation failed under us.
        io,
    };

    kind: Kind = .io,
    /// Which selection this was, so a report over several cursors can say.
    index: u32 = 0,
    /// The exit status, when `kind` is `.exit`.
    code: u8 = 0,
    /// stderr exactly as the command wrote it, OWNED by the response. Empty
    /// when the command said nothing, which is why `kind` and `code` exist.
    stderr: []u8 = &.{},
};

/// One worker answer. `outputs` owns each slice; a failure owns an empty list
/// and, usually, the command's own account of itself in `failure`.
pub const Response = struct {
    id: u32,
    success: bool,
    outputs: [][]u8,
    failure: ?Failure = null,

    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);
        if (response.failure) |f| if (f.stderr.len > 0) gpa.free(f.stderr);
        response.* = undefined;
    }
};

/// What one invocation came to: the bytes, or the reason there are none.
pub const Outcome = union(enum) { ok: []u8, failed: Failure };

/// The in-flight set a host keeps while pipes run off its loop.
///
/// tty.zig and gui.zig each had this verbatim — same `finish` walk, same
/// `cancelAll`, same 16 — differing only in whether `add` asserted or returned
/// a bool. It lives here beside the Job it tracks so a third host (the AppKit
/// shell, which had no pipe support at all) does not have to grow a fourth.
///
/// Bounded on purpose: a filter is a user gesture, and sixteen concurrent ones
/// is already more than anybody means. `add` returning false is the host's cue
/// to answer the request as failed rather than to queue it.
pub const Tasks = struct {
    pub const capacity = 16;

    pub const Task = struct {
        id: u32,
        future: std.Io.Future(anyerror!void),
    };

    items: [capacity]Task = undefined,
    len: usize = 0,

    pub fn full(tasks: *const Tasks) bool {
        return tasks.len == tasks.items.len;
    }

    pub fn add(tasks: *Tasks, task: Task) bool {
        if (tasks.full()) return false;
        tasks.items[tasks.len] = task;
        tasks.len += 1;
        return true;
    }

    /// Join the one that answered and drop it, preserving order so `cancelAll`
    /// stays deterministic.
    pub fn finish(tasks: *Tasks, io: std.Io, id: u32) void {
        for (tasks.items[0..tasks.len], 0..) |*task, i| if (task.id == id) {
            task.future.await(io) catch {};
            tasks.len -= 1;
            std.mem.copyForwards(Task, tasks.items[i..tasks.len], tasks.items[i + 1 .. tasks.len + 1]);
            return;
        };
    }

    pub fn cancelAll(tasks: *Tasks, io: std.Io) void {
        for (tasks.items[0..tasks.len]) |*task| task.future.cancel(io) catch {};
        tasks.len = 0;
    }
};

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,
) Outcome {
    const fail = struct {
        fn k(kind: Failure.Kind) Outcome {
            return .{ .failed = .{ .kind = kind } };
        }
    };
    if (std.mem.indexOfScalar(u8, command, 0) != null or
        std.mem.indexOfScalar(u8, cwd, 0) != null) return fail.k(.spawn);

    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 fail.k(.spawn);

    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 fail.k(.spawn);
    };
    // 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 fail.k(.too_large);
    } else |err| switch (err) {
        error.EndOfStream => {},
        // The deadline is the only one of these a human is likely to cause,
        // and it is the one they are least able to guess at: ten seconds of
        // nothing used to be followed by nothing.
        error.Timeout => return fail.k(.timeout),
        else => return fail.k(.io),
    }
    if (stdout_reader.buffered().len > max_stdout_bytes or
        stderr_reader.buffered().len > max_stderr_bytes) return fail.k(.too_large);
    multi_reader.checkAnyError() catch return fail.k(.io);
    const term = child.wait(io) catch return fail.k(.io);
    writer.?.join();
    writer = null;

    const stdout = multi_reader.toOwnedSlice(0) catch return fail.k(.io);
    const stderr = multi_reader.toOwnedSlice(1) catch {
        gpa.free(stdout);
        return fail.k(.io);
    };
    // stderr is NOT freed here any more. It is the command's own account of
    // what went wrong, and it goes home with the failure.
    errdefer gpa.free(stderr);

    // A STDIN WRITE THAT ENDED EARLY IS NOT A FAILURE. `writer_context.ok` was
    // part of this condition, so `| head -1` over a selection bigger than the
    // pipe buffer failed — the command closed stdin after the line it wanted,
    // the write got EPIPE, and a filter that had done exactly its job reported
    // nothing. helix joins its input task and ignores the result for this
    // reason; the exit status is the whole verdict.
    switch (term) {
        .exited => |code| if (code != 0) {
            gpa.free(stdout);
            return .{ .failed = .{ .kind = .exit, .code = code, .stderr = stderr } };
        },
        else => {
            gpa.free(stdout);
            return .{ .failed = .{ .kind = .signal, .stderr = stderr } };
        },
    }
    gpa.free(stderr);
    return .{ .ok = 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 = switch (runOne(gpa, io, job.command, job.cwd, input)) {
            .ok => |bytes| bytes,
            .failed => |f| {
                // WHICH selection, because with several cursors "it failed" is
                // not enough to go looking with.
                var owned = f;
                owned.index = @intCast(i);
                response.failure = owned;
                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);
        response.failure = .{ .kind = .too_large, .index = @intCast(i) };
        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 selection filters use the physical directory of an explicit OS mount" {
    const gpa = std.testing.allocator;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    var directory_buf: [4096]u8 = undefined;
    const directory = directory_buf[0..try tmp.dir.realPath(std.testing.io, &directory_buf)];
    var declared_buf: [4102]u8 = undefined;
    const declared = try std.fmt.bufPrint(&declared_buf, "/n/os{s}", .{directory});
    const job = try Job.copy(gpa, .{ .id = 1, .command = "pwd", .cwd = declared, .inputs = &.{.{ .bytes = "" }} });
    defer job.deinit(gpa);
    try std.testing.expectEqualStrings(directory, job.cwd);
    switch (runOne(gpa, std.testing.io, job.command, job.cwd, "")) {
        .ok => |bytes| {
            defer gpa.free(bytes);
            try std.testing.expectEqualStrings(directory, std.mem.trimEnd(u8, bytes, "\n"));
        },
        .failed => |failure| {
            gpa.free(failure.stderr);
            return error.FilterFailed;
        },
    }
}

test "native pipe runner preserves stdin/stdout bytes and reports how it failed" {
    const gpa = std.testing.allocator;
    const io = std.testing.io;
    const output = switch (runOne(gpa, io, "printf 'prefix:'; cat; printf '\\n'", "/tmp", "a\x00b\n")) {
        .ok => |bytes| bytes,
        .failed => return error.PipeCommandFailed,
    };
    defer gpa.free(output);
    try std.testing.expectEqualSlices(u8, "prefix:a\x00b\n\n", output);

    // A NONZERO EXIT COMES HOME WITH ITS REASON. The status and the command's
    // own stderr are the whole of what a human needs to fix a typo'd filter,
    // and both used to be freed on the floor.
    switch (runOne(gpa, io, "echo trouble >&2; exit 7", "/tmp", "")) {
        .ok => |bytes| {
            gpa.free(bytes);
            return error.PipeShouldHaveFailed;
        },
        .failed => |f| {
            defer gpa.free(f.stderr);
            try std.testing.expectEqual(Failure.Kind.exit, f.kind);
            try std.testing.expectEqual(@as(u8, 7), f.code);
            try std.testing.expectEqualSlices(u8, "trouble\n", f.stderr);
        },
    }

    // ...and a command that stops reading its stdin SUCCEEDS. `| head -1` over
    // a selection bigger than the pipe buffer takes the line it wanted and
    // closes the pipe; the write gets EPIPE, and that used to fail the filter
    // even though it had done exactly its job.
    const big = try gpa.alloc(u8, 512 * 1024);
    defer gpa.free(big);
    @memset(big, 'x');
    big[0] = 'a';
    big[1] = '\n';
    switch (runOne(gpa, io, "head -1", "/tmp", big)) {
        .ok => |bytes| {
            defer gpa.free(bytes);
            try std.testing.expectEqualSlices(u8, "a\n", bytes);
        },
        .failed => return error.EarlyStdinCloseShouldNotFail,
    }
}

// ---- the pipe inside the editor: the prompt, the request, the atomic edit ----

fn nextPipeEffect(p: *Pardes) ?u32 {
    while (p.nextEffect()) |effect| switch (effect) {
        .pipe => |request| return request.id,
        else => {},
    };
    return null;
}

/// The tag-tail marker each behaviour arms with. One function so the prompt
/// that is DRAWN and the command that is PARSED can never disagree.
pub fn pipeMarker(how: modal.Normal.PipeBehavior) []const u8 {
    return switch (how) {
        .replace => config.pipe_marker,
        .ignore => config.pipe_marker_to,
        .insert => config.pipe_marker_insert,
        .append => config.pipe_marker_append,
    };
}

pub const PendingPipe = struct {
    id: u32,
    pane: usize,
    serial: u32,
    revision: u32,
    command: []u8,
    cwd: []u8,
    inputs: []Input,
    ranges: [Pane.max_selections]modal.Selection,
    primary: u8,
    explicit: bool,
    how: modal.Normal.PipeBehavior,
    nranges: u8,

    pub 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) Request {
        return .{
            .id = wait.id,
            .command = wait.command,
            .cwd = wait.cwd,
            .inputs = wait.inputs,
        };
    }
};

pub fn submitPipe(p: *Pardes, id: usize) void {
    const pane = p.panes[id] orelse return;
    const f = pane.file orelse return;
    if (!panes.Output.fileTraits(f.output).saves) return;
    const prompt = switch (pane.prompt) {
        .pipe => |pipe| pipe,
        else => return,
    };
    const tail = pane.tagSlice();
    const armed = tail[@min(prompt.at, tail.len)..];
    const marker = pipeMarker(prompt.how);
    if (!std.mem.startsWith(u8, armed, marker)) return;
    const command = armed[marker.len..];
    if (command.len == 0) return;

    var ranges: [Pane.max_selections]modal.Selection = undefined;
    const got = pane.ranges(f.content, 0, &ranges);
    // `!`/`A-!` take no stdin and run ONCE — see `PendingPipe.nranges`.
    const ninputs = if (prompt.how.pipes()) got.n else 1;
    const inputs = p.gpa.alloc(Input, ninputs) catch return;
    var made: usize = 0;
    for (ranges[0..ninputs], 0..) |range, i| {
        const bytes = if (prompt.how.pipes()) bytes: {
            const lo = @min(range.anchor, range.head);
            const hi = @max(range.anchor, range.head);
            if (hi > f.content.len) break;
            break :bytes f.content[lo..hi];
        } else "";
        const copy = p.gpa.dupe(u8, bytes) catch break;
        inputs[i] = .{ .bytes = copy };
        made += 1;
    }
    if (made != ninputs) {
        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, Pardes.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,
        .how = prompt.how,
        .nranges = @intCast(got.n),
    };
    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) ?Request {
    if (p.pipe_wait) |*wait| {
        if (wait.id == id) return wait.request();
    }
    return null;
}

fn pipeFailed(p: *Pardes, wait: *const PendingPipe, failure: ?Failure) void {
    var out: std.Io.Writer.Allocating = .init(p.gpa);
    defer out.deinit();
    const w = &out.writer;
    w.print("| {s}\n\n", .{wait.command}) catch return;
    if (failure) |fail| {
        if (wait.inputs.len > 1)
            w.print("selection {d} of {d}: ", .{ fail.index + 1, wait.inputs.len }) catch return;
        switch (fail.kind) {
            .exit => w.print("exit status {d}\n", .{fail.code}) catch return,
            .signal => w.writeAll("killed by a signal\n") catch return,
            .timeout => w.print(
                "still running after {d} seconds, and stopped\n",
                .{command_timeout_seconds},
            ) catch return,
            .too_large => w.writeAll("produced more output than a filter may return\n") catch return,
            .spawn => w.writeAll("could not be started\n") catch return,
            .io => w.writeAll("could not be read\n") catch return,
        }
        if (fail.stderr.len > 0) w.print("\n{s}", .{fail.stderr}) catch return;
    } else {
        // No diagnosis at all: nobody ran it. The detached daemon, the
        // browser and the board all leave `pipe` null on purpose.
        w.writeAll("this session cannot run filters\n") catch return;
    }
    const content = out.toOwnedSlice() catch return;
    const was = p.active;
    panes.Output.openErrors(p, wait.pane, content) catch |err| {
        p.gpa.free(content);
        // Nowhere to put the report is itself worth one line.
        p.reportError(wait.pane, "pipe", err);
        return;
    };
    if (p.panes[was] != null) p.active = was;
}

fn pipeCut(wait: *const PendingPipe, content: []const u8, i: usize) ?struct { lo: usize, hi: usize } {
    const range = wait.ranges[i];
    const lo = @min(range.anchor, range.head);
    const hi = @max(range.anchor, range.head);
    if (hi > content.len) return null;
    return switch (wait.how) {
        .replace, .ignore => .{ .lo = lo, .hi = hi },
        .insert => .{ .lo = lo, .hi = lo },
        .append => .{ .lo = hi, .hi = hi },
    };
}

fn pipeOutput(
    wait: *const PendingPipe,
    outputs: []const []const u8,
    content: []const u8,
    i: usize,
) []const u8 {
    var out = if (wait.how.pipes()) outputs[i] else outputs[0];
    const range = wait.ranges[i];
    const lo = @min(range.anchor, range.head);
    const hi = @max(range.anchor, range.head);
    const selected = if (hi <= content.len) content[lo..hi] else "";
    if (!std.mem.endsWith(u8, selected, "\n") and std.mem.endsWith(u8, out, "\n")) {
        out = out[0 .. out.len - 1];
        if (std.mem.endsWith(u8, out, "\r")) out = out[0 .. out.len - 1];
    }
    return out;
}

pub fn pipeResponse(
    p: *Pardes,
    id: u32,
    success: bool,
    outputs: []const []const u8,
    failure: ?Failure,
) void {
    // A SUPERSEDED OR UNKNOWN id is the one silence worth keeping: it is
    // the answer to a question nobody is still asking.
    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 pipeFailed(p, &wait, failure);

    const pane = p.panes[wait.pane] orelse return;
    if (pane.serial != wait.serial) return;
    const f = if (pane.file) |*file| file else return;
    if (!panes.Output.fileTraits(f.output).saves) return;
    if (f.revision != wait.revision)
        return p.reportError(wait.pane, "pipe", error.FileChangedWhileFiltering);

    if (wait.how == .ignore) return;

    const n = wait.nranges;
    var total_output: usize = 0;
    var removed: usize = 0;
    var previous_end: usize = 0;
    for (0..n) |i| {
        const cut = pipeCut(&wait, f.content, i) orelse return;
        const output = pipeOutput(&wait, outputs, f.content, i);
        if (output.len > max_stdout_bytes) return;
        total_output = std.math.add(usize, total_output, output.len) catch return;
        if (total_output > max_total_stdout_bytes) return;
        if (cut.lo < previous_end) return;
        removed = std.math.add(usize, removed, cut.hi - cut.lo) catch return;
        previous_end = cut.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: [Pane.max_selections]modal.Selection = undefined;
    var read_at: usize = 0;
    var write_at: usize = 0;
    for (0..n) |i| {
        const range = wait.ranges[i];
        const cut = pipeCut(&wait, f.content, i).?;
        const output = pipeOutput(&wait, outputs, f.content, i);
        @memcpy(replacement[write_at .. write_at + (cut.lo - read_at)], f.content[read_at..cut.lo]);
        write_at += cut.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 = cut.hi;
    }
    @memcpy(replacement[write_at..], f.content[read_at..]);

    // One async request is one history transaction, even at 64 cursors.
    edit.pushUndo(p, pane);
    panes.File.setContent(p, f, replacement);
    pane.setRanges(f.content, new_ranges[0..n], &.{}, wait.primary, wait.explicit);
    pane.select = false;
    pane.ensureCursorVisible();
}

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.setTestFile("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.prompt == .pipe);
    try std.testing.expect(std.mem.endsWith(u8, pane.tagSlice(), 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.prompt != .pipe);
    const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
    const request = pipeRequest(p, 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.prompt != .pipe);
    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.setTestFile("aa bb cc\n");
    const ranges = [_]modal.Selection{
        .{ .anchor = 0, .head = 2 },
        .{ .anchor = 6, .head = 8 },
    };
    pane.setRanges(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 = pipeRequest(p, 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);

    var denied = std.testing.FailingAllocator.init(gpa, .{ .fail_index = 0 });
    p.scratch.deinit();
    p.scratch = .init(denied.allocator());
    const outputs: []const []const u8 = &.{ "AA\n", "" };
    p.update(.{ .pipe_resp = .{ .id = id, .success = true, .outputs = outputs } });
    try std.testing.expectEqualSlices(u8, "AA bb \n", pane.file.?.content);
    try std.testing.expectEqual(@as(usize, 1), pane.file.?.history.undo_len);
    try std.testing.expectEqual(@as(u8, 1), pane.nsel);
    try std.testing.expectEqual(@as(i32, 1), pane.cur_col);
    try std.testing.expectEqual(@as(i32, 6), pane.sels[0].col);
    try std.testing.expect(!denied.has_induced_failure);

    p.update(.{ .key = .{ .cp = 'u' } });
    try std.testing.expectEqualSlices(u8, "aa bb cc\n", pane.file.?.content);
}

test "the four shell behaviours put their output where helix puts it" {
    const gpa = std.testing.allocator;
    const Case = struct {
        key: Key,
        outputs: []const []const u8,
        want: []const u8,
        want_effect: bool = true,
    };
    // One selection, `bb`, in `aa bb cc\n` — bytes 3..5.
    for ([_]Case{
        // `|` replaces the selection with what the command wrote.
        .{ .key = .{ .cp = '|' }, .outputs = &.{"XX"}, .want = "aa XX cc\n" },
        // `!` inserts BEFORE it, and the selection survives to its right.
        .{ .key = .{ .cp = '!' }, .outputs = &.{"XX"}, .want = "aa XXbb cc\n" },
        // `A-!` appends AFTER it.
        .{ .key = .{ .cp = '!', .alt = true }, .outputs = &.{"XX"}, .want = "aa bbXX cc\n" },
        // `A-|` runs the command and leaves the text completely alone: the
        // point is the side effect, not the output.
        .{ .key = .{ .cp = '|', .alt = true }, .outputs = &.{"XX"}, .want = "aa bb cc\n" },
    }) |case| {
        const p = try Pardes.init(gpa, .{ .tty_only = true, .cols = 100, .rows = 30 });
        defer p.deinit();
        while (p.nextEffect()) |_| {}
        const pane = try p.setTestFile("aa bb cc\n");
        pane.setRanges(pane.file.?.content, &.{.{ .anchor = 3, .head = 5 }}, &.{}, 0, true);

        p.update(.{ .key = case.key });
        for ("cmd") |c| p.update(.{ .key = .{ .cp = c, .text = &.{c} } });
        p.update(.{ .key = .{ .cp = Key.enter } });
        const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
        p.update(.{ .pipe_resp = .{ .id = id, .success = true, .outputs = case.outputs } });
        try std.testing.expectEqualSlices(u8, case.want, pane.file.?.content);
    }
}

test "a command with no stdin runs once and every cursor gets that one answer" {
    const gpa = std.testing.allocator;
    const p = try Pardes.init(gpa, .{ .tty_only = true, .cols = 100, .rows = 30 });
    defer p.deinit();
    while (p.nextEffect()) |_| {}
    const pane = try p.setTestFile("aa bb cc\n");
    const ranges = [_]modal.Selection{
        .{ .anchor = 0, .head = 2 },
        .{ .anchor = 6, .head = 8 },
    };
    pane.setRanges(pane.file.?.content, &ranges, &.{}, 0, true);

    p.update(.{ .key = .{ .cp = '!' } }); // insert-output: no stdin
    for ("date") |c| p.update(.{ .key = .{ .cp = c, .text = &.{c} } });
    p.update(.{ .key = .{ .cp = Key.enter } });
    const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;
    const request = pipeRequest(p, id) orelse return error.MissingPipeRequest;

    try std.testing.expectEqual(@as(usize, 1), request.inputs.len);
    try std.testing.expectEqualSlices(u8, "", request.inputs[0].bytes);

    p.update(.{ .pipe_resp = .{ .id = id, .success = true, .outputs = &.{"T"} } });
    try std.testing.expectEqualSlices(u8, "Taa bb Tcc\n", pane.file.?.content);
    try std.testing.expectEqual(@as(usize, 1), pane.file.?.history.undo_len);
}

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.setTestFile("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.?.history.undo_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;
    panes.File.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.?.history.undo_len);
}

test "a failed filter opens an errors buffer carrying the command's own words" {
    const gpa = std.testing.allocator;
    const p = try Pardes.init(gpa, .{ .tty_only = true, .cols = 100, .rows = 30 });
    defer p.deinit();
    while (p.nextEffect()) |_| {}
    const pane = try p.setTestFile("abc\n");
    pane.cur_col = 2;
    pane.vsel = .{ .active = true, .row = 0, .col = 0, .explicit = true };

    p.update(.{ .key = .{ .cp = '|' } });
    for ("trr") |c| p.update(.{ .key = .{ .cp = c, .text = &.{c} } });
    p.update(.{ .key = .{ .cp = Key.enter } });
    const id = nextPipeEffect(p) orelse return error.MissingPipeEffect;

    // The shape a real runner brings home for `| trr`: nonzero exit, and the
    // shell's own sentence about it.
    var stderr = "sh: line 1: trr: command not found\n".*;
    p.update(.{ .pipe_resp = .{
        .id = id,
        .success = false,
        .outputs = &.{},
        .failure = .{ .kind = .exit, .code = 127, .stderr = &stderr },
    } });

    // The text is untouched — a failed filter is not an edit...
    try std.testing.expectEqualSlices(u8, "abc\n", pane.file.?.content);
    try std.testing.expectEqual(@as(usize, 0), pane.file.?.history.undo_len);
    // ...and the cursor did not go anywhere, so `|` again edits the same file.
    try std.testing.expectEqual(@as(usize, 0), p.active);

    var found: ?[]const u8 = null;
    for (p.panes) |slot| {
        const q = slot orelse continue;
        const qf = q.file orelse continue;
        const o = qf.output orelse continue;
        if (std.meta.activeTag(o.from) == .errors) found = qf.content;
    }
    const report = found orelse return error.NoErrorsBuffer;
    try std.testing.expect(std.mem.indexOf(u8, report, "| trr") != null);
    try std.testing.expect(std.mem.indexOf(u8, report, "exit status 127") != null);
    try std.testing.expect(std.mem.indexOf(u8, report, "command not found") != null);
}

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.setTestFile("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.setTestFile("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.?.history.undo_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.?.history.undo_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.setTestFile("x");

    pane.file.?.output = .{ .from = .search };
    p.update(.{ .key = .{ .cp = '|' } });
    try std.testing.expect(pane.prompt != .pipe);
    pane.file.?.output = null;
    pane.mode = .insert;
    p.update(.{ .key = .{ .cp = '|', .text = "|" } });
    try std.testing.expect(pane.prompt != .pipe);
    try std.testing.expectEqualSlices(u8, "|x", pane.file.?.content);
}