test(compile): acceptance suites — framing items 1-33 dispatch-driven, 35 two-replica

tests/compile_mode_acceptance.rs (34 tests): spawn shape + header +
exit markers; read-only under dispatch; child-boundary stderr merge
in emission order; stdin EOF; group kill/leader-exit/escalation/
ledger bites incl. the redirected TERM-ignoring survivor and the
pipe-holding-descendant tick-latency bound; starter-rule parsing
with 0-based normalization and severity posture; sub-1 fail-closed;
severity override + malformed-rule containers; unterminated final
line; RET/n-p/M-g n/M-g p/C-x ` navigation pins with the diag
fallback; recompile + q-target discipline; supersede baseline; all
seven undo/redo chords table-driven; M-x undo after a completed run
recovering via buffer.after-edit; no-hook shrink and same-length
newline-moving replace with anchor epochs; ANSI SGR/CR with
rendered-cell attachment proof surviving RET-then-M-,; killed-buffer
teardown; grep locations panel, kill-mid-search + masking
prevention, root retention; shell-command M-!; round-trip pins.

tests/compile_mode_crdt_acceptance.rs: a chord-triggered full run
converges byte-identically on two replicas (mid-session generated-
buffer snapshot adoption), and a synthetic accepted replica edit
triggers the immediate recovery marker, converging across the
causal-reorder seam.

Fixes found by the suite: compile.lua's CR handling now scans the
current line start from the buffer (the REPL discipline) instead of
using the per-batch parse position — a same-batch CR previously let
a progress line overwrite earlier output; malformed Lua patterns
are rejected (and counted) at validation time via a probe match.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VoiEyuPjoBhvwACf8HAnLB

builtin/runtime/compile.lua

@@ -92,6 +92,10 @@ pmacs.compile.rules = {
 local function rule_is_valid(rule)
   if type(rule) ~= "table" then return false end
   if type(rule.pattern) ~= "string" then return false end
+  -- Probe the pattern against the empty string so a malformed Lua
+  -- pattern is caught (and counted in the status note) here at
+  -- validation time, not silently at match time.
+  if not pcall(string.match, "", rule.pattern) then return false end
   if type(rule.file) ~= "number" or rule.file < 1 then return false end
   if type(rule.line) ~= "number" or rule.line < 1 then return false end
   if rule.col ~= nil and (type(rule.col) ~= "number" or rule.col < 1) then return false end
@@ -321,9 +325,29 @@ local function emit_text(slot, text)
   add_style_span(slot, pos, pos + #text)
 end
 
--- The current unterminated line spans [parse_line_start, buf:len());
--- CR/BS/erase are confined to it by construction (lines are parsed
--- and left behind the moment their newline lands).
+-- Byte offset where the line containing `out_pos` starts. Scanned
+-- from the buffer (the REPL's `_current_line_start` discipline) —
+-- NOT `parse_line_start`, which only advances once per batch: a CR
+-- arriving in the same batch as earlier completed lines must rewind
+-- to the start of the CURRENT line, not to the batch's first line
+-- (using the stale value let a progress line overwrite everything
+-- emitted earlier in the batch).
+local function current_line_start(slot)
+  local pos = math.min(slot.out_pos, slot.buf:len())
+  local prefix = slot.buf:slice(0, pos)
+  local start = 0
+  local search = 1
+  while true do
+    local idx = prefix:find("\n", search, true)
+    if not idx then return start end
+    start = idx
+    search = idx + 1
+  end
+end
+
+-- The current unterminated line runs from its scanned start to
+-- buf:len() — no newline ever exists past out_pos (output is
+-- append-only except CR/BS rewinds within the current line).
 local function apply_events(slot, events)
   local buf = slot.buf
   for _, ev in ipairs(events) do
@@ -333,9 +357,9 @@ local function apply_events(slot, events)
     elseif kind == "set_style" then
       slot.cur_style = ev.style
     elseif kind == "carriage_return" then
-      slot.out_pos = slot.parse_line_start
+      slot.out_pos = current_line_start(slot)
     elseif kind == "backspace" then
-      if slot.out_pos > slot.parse_line_start then
+      if slot.out_pos > current_line_start(slot) then
         slot.out_pos = slot.out_pos - 1
       end
     elseif kind == "erase_to_eol" then
@@ -344,11 +368,12 @@ local function apply_events(slot, events)
         buf:delete(slot.out_pos, len, { bypass_intercept = true })
       end
     elseif kind == "erase_line" then
+      local ls = current_line_start(slot)
       local len = buf:len()
-      if slot.parse_line_start < len then
-        buf:delete(slot.parse_line_start, len, { bypass_intercept = true })
+      if ls < len then
+        buf:delete(ls, len, { bypass_intercept = true })
       end
-      slot.out_pos = slot.parse_line_start
+      slot.out_pos = ls
     end
     -- alt-screen suppression happens inside the parser; titles and
     -- shell-integration markers are irrelevant to a compile buffer.

tests/compile_mode_crdt_acceptance.rs

@@ -0,0 +1,252 @@
+// compile_mode_crdt_acceptance.rs --- compile-mode over the wire.
+
+//! Compile-mode two-replica acceptance (docs/compile-mode-framing.md,
+//! item 35): a full compile run's generated buffer converges
+//! byte-identically on a mirror replica, and a synthetic accepted
+//! replica edit to that buffer triggers the immediate recovery
+//! marker (the `buffer.after-edit` path fires for accepted `CrdtOp`s)
+//! and still converges on both replicas — even though the
+//! hook-produced marker may queue before the source edit's
+//! rebroadcast (the established causal-reordering seam).
+//!
+//! All compile-buffer writes are daemon-side Lua bypass edits —
+//! ordinary daemon-peer CRDT ops with no optimistic involvement —
+//! so convergence here pins the whole streaming pipeline (header,
+//! parsed output, exit marker) as replicable state.
+
+#![cfg(feature = "crdt")]
+
+use std::time::Duration;
+
+use pmacs::crdt::CrdtState;
+use pmacs::protocol::{FrontendEvent, FrontendId, Key, KeyEvent, Modifiers};
+use pmacs::rope::CrdtOp as RopeCrdtOp;
+use pmacs::transport::write_message;
+
+mod common;
+use common::daemon::{TestDaemon, attach_multi};
+
+fn read_initial_snapshot(
+    stream: &mut std::os::unix::net::UnixStream,
+) -> (pmacs::buffer::BufferId, Vec<u8>) {
+    match pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(stream)
+        .expect("read initial BufferSnapshot")
+    {
+        pmacs::protocol::InstanceMessage::BufferSnapshot {
+            buffer_id,
+            crdt_snapshot,
+        } => (buffer_id, crdt_snapshot),
+        other => panic!("expected initial BufferSnapshot, got {other:?}"),
+    }
+}
+
+struct Replica {
+    stream: std::os::unix::net::UnixStream,
+    state: CrdtState,
+    fid: FrontendId,
+    buffer_id: pmacs::buffer::BufferId,
+}
+
+fn attach_replica(daemon: &TestDaemon) -> Replica {
+    let (hello, mut stream) = attach_multi(daemon);
+    let fid = hello.assigned_frontend_id;
+    let (buffer_id, snap) = read_initial_snapshot(&mut stream);
+    let state = CrdtState::new(fid.0).expect("CrdtState::new");
+    state.import_snapshot(&snap).expect("import_snapshot");
+    Replica {
+        stream,
+        state,
+        fid,
+        buffer_id,
+    }
+}
+
+fn send_key(replica: &mut Replica, key: Key, mods: Modifiers) {
+    write_message(
+        &mut replica.stream,
+        &FrontendEvent::Key(KeyEvent {
+            frontend_id: replica.fid,
+            key,
+            mods,
+            timestamp_ns: 0,
+        }),
+    )
+    .expect("send Key");
+}
+
+/// Mutate the local replica, export the delta, and ship it as an
+/// optimistic `FrontendEvent::CrdtOp` (the `m10_11` idiom).
+fn send_optimistic_op<F>(replica: &mut Replica, mutate: F)
+where
+    F: FnOnce(&CrdtState),
+{
+    let v = replica.state.version();
+    mutate(&replica.state);
+    let op_bytes = replica
+        .state
+        .export_updates_since(&v)
+        .expect("export updates after local mutation");
+    write_message(
+        &mut replica.stream,
+        &FrontendEvent::CrdtOp {
+            frontend_id: replica.fid,
+            buffer_id: replica.buffer_id,
+            op: RopeCrdtOp {
+                peer_id: replica.fid.0,
+                bytes: op_bytes,
+            },
+        },
+    )
+    .expect("write CrdtOp");
+}
+
+/// Read until a `BufferSnapshot` for a buffer other than the current
+/// one arrives (the compile run creates *compilation* mid-session;
+/// the daemon broadcasts a snapshot for the newly-CRDT-backed buffer
+/// and via the active-buffer-follow path). Re-seats the replica's
+/// mirror on that buffer.
+fn adopt_next_buffer(replica: &mut Replica, what: &str) {
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    loop {
+        assert!(
+            std::time::Instant::now() < deadline,
+            "timeout adopting the new buffer snapshot for {what}"
+        );
+        replica
+            .stream
+            .set_read_timeout(Some(Duration::from_millis(100)))
+            .ok();
+        match pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(
+            &mut replica.stream,
+        ) {
+            Ok(pmacs::protocol::InstanceMessage::BufferSnapshot {
+                buffer_id,
+                crdt_snapshot,
+            }) if buffer_id != replica.buffer_id => {
+                let state = CrdtState::new(replica.fid.0).expect("CrdtState::new");
+                state
+                    .import_snapshot(&crdt_snapshot)
+                    .expect("import new-buffer snapshot");
+                replica.state = state;
+                replica.buffer_id = buffer_id;
+                return;
+            }
+            Ok(_) | Err(_) => {}
+        }
+    }
+}
+
+/// Pump broadcast messages, importing every `CrdtOp` for the tracked
+/// buffer, until `pred(text)` holds.
+fn pump_until_text<P: Fn(&str) -> bool>(
+    replica: &mut Replica,
+    timeout: Duration,
+    what: &str,
+    pred: P,
+) -> String {
+    let deadline = std::time::Instant::now() + timeout;
+    let mut text = replica.state.materialize_string();
+    loop {
+        if pred(&text) {
+            return text;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "pump timeout waiting for {what}; text={text:?}"
+        );
+        replica
+            .stream
+            .set_read_timeout(Some(Duration::from_millis(100)))
+            .ok();
+        match pmacs::transport::read_message::<pmacs::protocol::InstanceMessage>(
+            &mut replica.stream,
+        ) {
+            Ok(pmacs::protocol::InstanceMessage::CrdtOp { buffer_id: b, op })
+                if b == replica.buffer_id =>
+            {
+                let _ = replica.state.import_updates(&op.bytes);
+                text = replica.state.materialize_string();
+            }
+            Ok(_) | Err(_) => {}
+        }
+    }
+}
+
+const DESYNC: &str = "[output desynced by external edit]";
+
+#[test]
+fn compile_run_converges_and_replica_edit_triggers_recovery() {
+    // Fixture: the compile command lives in a shared tempdir; the
+    // init.lua binds a chord that runs it (typing an M-x prompt over
+    // the wire would test the minibuffer, not compile-mode).
+    let dir = tempfile::tempdir().expect("tempdir");
+    let script = dir.path().join("fix.sh");
+    std::fs::write(&script, "printf 'x.c:1:1: error: boom\\ndone\\n'\n").unwrap();
+    let init = format!(
+        r#"
+        pmacs.command.define {{
+            name = "test.compile",
+            description = "compile-mode CRDT fixture trigger",
+            fn = function()
+                pmacs.compile.run("sh {script}", {{ cwd = "{dir}" }})
+            end,
+        }}
+        pmacs.keymap.bind {{ scope = "global", sequence = "C-c 9", command = "test.compile" }}
+        "#,
+        script = script.display(),
+        dir = dir.path().display(),
+    );
+    let daemon = TestDaemon::spawn_with_config(&init);
+    let mut source = attach_replica(&daemon);
+    let mut observer = attach_replica(&daemon);
+    let initial = source.buffer_id;
+
+    // Trigger the run from the source replica (round-tripped keys).
+    send_key(&mut source, Key::Char('c'), Modifiers::CTRL);
+    send_key(&mut source, Key::Char('9'), Modifiers::NONE);
+
+    // Both replicas adopt the freshly-created *compilation* buffer.
+    adopt_next_buffer(&mut source, "source");
+    adopt_next_buffer(&mut observer, "observer");
+    assert_ne!(source.buffer_id, initial, "a new buffer was created");
+    assert_eq!(
+        source.buffer_id, observer.buffer_id,
+        "both replicas mirror the same generated buffer"
+    );
+
+    // The full run — header, streamed output, exit marker — reaches
+    // both mirrors byte-identically.
+    let done = |t: &str| t.contains("[compile exited with code 0]");
+    let src_text = pump_until_text(&mut source, Duration::from_secs(15), "source run", done);
+    let obs_text = pump_until_text(&mut observer, Duration::from_secs(15), "observer run", done);
+    assert_eq!(src_text, obs_text, "byte-identical convergence");
+    assert!(src_text.contains("x.c:1:1: error: boom"), "output replicated");
+    assert!(src_text.starts_with("$ sh "), "header replicated");
+
+    // Synthetic accepted replica edit to the generated buffer: the
+    // daemon applies it, buffer.after-edit fires, and compile.lua's
+    // revision guard appends the recovery marker immediately. The
+    // marker (a daemon-peer op) may broadcast before the source
+    // edit's own rebroadcast — the causal-reordering seam — and both
+    // replicas must still converge.
+    send_optimistic_op(&mut source, |r| {
+        r.insert(0, "Z").expect("replica edit");
+    });
+    let recovered = |t: &str| t.contains(DESYNC) && t.starts_with('Z');
+    let src_text = pump_until_text(
+        &mut source,
+        Duration::from_secs(10),
+        "source recovery marker",
+        recovered,
+    );
+    let obs_text = pump_until_text(
+        &mut observer,
+        Duration::from_secs(10),
+        "observer recovery marker",
+        recovered,
+    );
+    assert_eq!(
+        src_text, obs_text,
+        "post-recovery convergence across the reorder seam"
+    );
+}