broker(pty): replace TerminalQueryParser with alacritty EventListener

.trajectories/completed/2026-05/traj_piik8r6zu3i7.json

@@ -0,0 +1,89 @@
+{
+  "id": "traj_piik8r6zu3i7",
+  "version": 1,
+  "task": {
+    "title": "Issue 867: RelayEventListener"
+  },
+  "status": "completed",
+  "startedAt": "2026-05-18T01:56:18.236Z",
+  "completedAt": "2026-05-18T02:01:49.991Z",
+  "agents": [
+    {
+      "name": "default",
+      "role": "lead",
+      "joinedAt": "2026-05-18T01:58:24.265Z"
+    }
+  ],
+  "chapters": [
+    {
+      "id": "chap_xnbleth5fk5g",
+      "title": "Work",
+      "agentName": "default",
+      "startedAt": "2026-05-18T01:58:24.265Z",
+      "endedAt": "2026-05-18T02:01:49.991Z",
+      "events": [
+        {
+          "ts": 1779069504266,
+          "type": "decision",
+          "content": "Drain writeback channel from a dedicated std::thread, not tokio task: Drain writeback channel from a dedicated std::thread, not tokio task",
+          "raw": {
+            "question": "Drain writeback channel from a dedicated std::thread, not tokio task",
+            "chosen": "Drain writeback channel from a dedicated std::thread, not tokio task",
+            "alternatives": [],
+            "reasoning": "The reader thread is a std::thread (not tokio) and it's the only place that calls processor.advance which fires send_event. Keeping the drainer as a std::thread that takes self.writer.lock() (parking_lot Mutex) avoids importing the tokio runtime into PtySession::spawn and matches the existing reader thread shape. Also avoids holding any async tasks at spawn time when no runtime is guaranteed."
+          },
+          "significance": "high"
+        },
+        {
+          "ts": 1779069510157,
+          "type": "decision",
+          "content": "Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel: Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel",
+          "raw": {
+            "question": "Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel",
+            "chosen": "Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel",
+            "alternatives": [],
+            "reasoning": "send_event must be non-blocking (called while holding processor+term locks). std mpsc Sender::send never blocks. A dedicated std::thread drains the receiver and writes to the PTY writer, decoupling potential write_all blocking from the parser's hot path. Avoids tokio runtime requirement in PtySession::spawn."
+          },
+          "significance": "high"
+        },
+        {
+          "ts": 1779069704857,
+          "type": "decision",
+          "content": "Made Snapshot::from_term generic over EventListener: Made Snapshot::from_term generic over EventListener",
+          "raw": {
+            "question": "Made Snapshot::from_term generic over EventListener",
+            "chosen": "Made Snapshot::from_term generic over EventListener",
+            "alternatives": [],
+            "reasoning": "Changing PtySession's term field from Term<VoidListener> to Term<RelayEventListener> broke Snapshot::capture which calls pty.with_term(Self::from_term). Generic over L: EventListener keeps offline tests on VoidListener and live capture on RelayEventListener without an extra wrapper."
+          },
+          "significance": "high"
+        },
+        {
+          "ts": 1779069705573,
+          "type": "decision",
+          "content": "Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs): Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)",
+          "raw": {
+            "question": "Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)",
+            "chosen": "Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)",
+            "alternatives": [],
+            "reasoning": "Issue scope said pty_worker.rs, but TerminalQueryParser was also used by run_wrap (wrap.rs) and exported via helpers + asserted in main.rs tests. Both used the same PtySession, so the listener now handles them. Leaving them would have broken the import-removal in main.rs."
+          },
+          "significance": "high"
+        }
+      ]
+    }
+  ],
+  "retrospective": {
+    "summary": "Replaced hand-rolled TerminalQueryParser with alacritty RelayEventListener. Listener owns a std::sync::mpsc::Sender; alacritty's send_event(Event::PtyWrite) pushes query responses (DSR/DA1/DA2/CPR) onto the channel, drained by a dedicated thread that writes to the PTY. CPR responses now reflect real cursor position instead of hardcoded 1;1. Deleted the parser and call sites in pty_worker.rs, wrap.rs, and main.rs; made Snapshot::from_term generic over EventListener. All 615 tests pass; clippy clean.",
+    "approach": "Standard approach",
+    "confidence": 0.85
+  },
+  "commits": [],
+  "filesChanged": [],
+  "projectId": "relay",
+  "tags": [],
+  "_trace": {
+    "startRef": "5fc8a131561feedfe990fc265f224101f6f267c4",
+    "endRef": "5fc8a131561feedfe990fc265f224101f6f267c4"
+  }
+}

.trajectories/completed/2026-05/traj_piik8r6zu3i7.md

@@ -0,0 +1,51 @@
+# Trajectory: Issue 867: RelayEventListener
+
+> **Status:** ✅ Completed
+> **Confidence:** 85%
+> **Started:** May 17, 2026 at 09:56 PM
+> **Completed:** May 17, 2026 at 10:01 PM
+
+---
+
+## Summary
+
+Replaced hand-rolled TerminalQueryParser with alacritty RelayEventListener. Listener owns a std::sync::mpsc::Sender; alacritty's send_event(Event::PtyWrite) pushes query responses (DSR/DA1/DA2/CPR) onto the channel, drained by a dedicated thread that writes to the PTY. CPR responses now reflect real cursor position instead of hardcoded 1;1. Deleted the parser and call sites in pty_worker.rs, wrap.rs, and main.rs; made Snapshot::from_term generic over EventListener. All 615 tests pass; clippy clean.
+
+**Approach:** Standard approach
+
+---
+
+## Key Decisions
+
+### Drain writeback channel from a dedicated std::thread, not tokio task
+
+- **Chose:** Drain writeback channel from a dedicated std::thread, not tokio task
+- **Reasoning:** The reader thread is a std::thread (not tokio) and it's the only place that calls processor.advance which fires send_event. Keeping the drainer as a std::thread that takes self.writer.lock() (parking_lot Mutex) avoids importing the tokio runtime into PtySession::spawn and matches the existing reader thread shape. Also avoids holding any async tasks at spawn time when no runtime is guaranteed.
+
+### Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel
+
+- **Chose:** Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel
+- **Reasoning:** send_event must be non-blocking (called while holding processor+term locks). std mpsc Sender::send never blocks. A dedicated std::thread drains the receiver and writes to the PTY writer, decoupling potential write_all blocking from the parser's hot path. Avoids tokio runtime requirement in PtySession::spawn.
+
+### Made Snapshot::from_term generic over EventListener
+
+- **Chose:** Made Snapshot::from_term generic over EventListener
+- **Reasoning:** Changing PtySession's term field from Term<VoidListener> to Term<RelayEventListener> broke Snapshot::capture which calls pty.with_term(Self::from_term). Generic over L: EventListener keeps offline tests on VoidListener and live capture on RelayEventListener without an extra wrapper.
+
+### Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)
+
+- **Chose:** Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)
+- **Reasoning:** Issue scope said pty_worker.rs, but TerminalQueryParser was also used by run_wrap (wrap.rs) and exported via helpers + asserted in main.rs tests. Both used the same PtySession, so the listener now handles them. Leaving them would have broken the import-removal in main.rs.
+
+---
+
+## Chapters
+
+### 1. Work
+
+_Agent: default_
+
+- Drain writeback channel from a dedicated std::thread, not tokio task: Drain writeback channel from a dedicated std::thread, not tokio task
+- Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel: Use std::sync::mpsc::Sender (sync, unbounded-ish) for writeback channel
+- Made Snapshot::from_term generic over EventListener: Made Snapshot::from_term generic over EventListener
+- Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs): Also removed parser call sites in wrap.rs and main.rs (not just pty_worker.rs)

.trajectories/index.json

@@ -1,6 +1,6 @@
 {
   "version": 1,
-  "lastUpdated": "2026-05-17T14:33:32.468Z",
+  "lastUpdated": "2026-05-18T02:01:50.133Z",
   "trajectories": {
     "traj_9gq96irkj00s": {
       "title": "Update relay to use published relaycast Rust reclaim fix",
@@ -591,6 +591,13 @@
       "startedAt": "2026-05-17T14:19:10.603Z",
       "completedAt": "2026-05-17T14:33:32.293Z",
       "path": ".trajectories/completed/2026-05/traj_cbmwd07phhm2.json"
+    },
+    "traj_piik8r6zu3i7": {
+      "title": "Issue 867: RelayEventListener",
+      "status": "completed",
+      "startedAt": "2026-05-18T01:56:18.236Z",
+      "completedAt": "2026-05-18T02:01:49.991Z",
+      "path": ".trajectories/completed/2026-05/traj_piik8r6zu3i7.json"
     }
   }
 }

src/helpers.rs

@@ -216,89 +216,6 @@ pub(crate) fn check_echo_in_output(output: &str, expected: &str) -> bool {
     clean.contains(expected)
 }
 
-#[derive(Debug, Clone, Copy, Default)]
-pub(crate) enum TerminalQueryState {
-    #[default]
-    Idle,
-    Esc,
-    Csi,
-    CsiQmark,
-    Csi6,
-    CsiQmark6,
-    /// CSI 0 — could be DA1 param (ESC [ 0 c)
-    Csi0,
-    /// CSI 5 — could be DSR (ESC [ 5 n)
-    Csi5,
-    /// CSI > — DA2 prefix (ESC [ > c)
-    CsiGt,
-}
-
-#[derive(Debug, Default)]
-pub(crate) struct TerminalQueryParser {
-    pub(crate) state: TerminalQueryState,
-}
-
-impl TerminalQueryParser {
-    pub(crate) fn feed(&mut self, chunk: &[u8]) -> Vec<&'static [u8]> {
-        const ESC: u8 = 0x1b;
-        const CSI: u8 = b'[';
-        const QMARK: u8 = b'?';
-        const SIX: u8 = b'6';
-        const N: u8 = b'n';
-
-        let mut out = Vec::new();
-        for byte in chunk {
-            self.state = match (self.state, *byte) {
-                (_, ESC) => TerminalQueryState::Esc,
-                (TerminalQueryState::Esc, CSI) => TerminalQueryState::Csi,
-                (TerminalQueryState::Csi, QMARK) => TerminalQueryState::CsiQmark,
-                (TerminalQueryState::Csi, b'>') => TerminalQueryState::CsiGt,
-                (TerminalQueryState::Csi, SIX) => TerminalQueryState::Csi6,
-                (TerminalQueryState::Csi, b'0') => TerminalQueryState::Csi0,
-                (TerminalQueryState::Csi, b'5') => TerminalQueryState::Csi5,
-                (TerminalQueryState::CsiQmark, SIX) => TerminalQueryState::CsiQmark6,
-                // CSI 6 n → Cursor Position Report (CPR)
-                (TerminalQueryState::Csi6, N) => {
-                    out.push(b"\x1b[1;1R".as_slice());
-                    TerminalQueryState::Idle
-                }
-                (TerminalQueryState::CsiQmark6, N) => {
-                    out.push(b"\x1b[?1;1R".as_slice());
-                    TerminalQueryState::Idle
-                }
-                // CSI c → DA1 (Device Attributes primary, no params)
-                (TerminalQueryState::Csi, b'c') => {
-                    out.push(b"\x1b[?1;2c".as_slice()); // VT100 with AVO
-                    TerminalQueryState::Idle
-                }
-                // CSI 0 c → DA1 with explicit 0 param
-                (TerminalQueryState::Csi0, b'c') => {
-                    out.push(b"\x1b[?1;2c".as_slice());
-                    TerminalQueryState::Idle
-                }
-                // CSI > c → DA2 (Device Attributes secondary)
-                (TerminalQueryState::CsiGt, b'c') => {
-                    out.push(b"\x1b[>1;10;0c".as_slice()); // VT100, version 10
-                    TerminalQueryState::Idle
-                }
-                // CSI 5 n → DSR (Device Status Report)
-                (TerminalQueryState::Csi5, N) => {
-                    out.push(b"\x1b[0n".as_slice()); // terminal OK
-                    TerminalQueryState::Idle
-                }
-                _ => TerminalQueryState::Idle,
-            };
-        }
-        out
-    }
-}
-
-#[cfg(test)]
-pub(crate) fn terminal_query_responses(chunk: &[u8]) -> Vec<&'static [u8]> {
-    let mut parser = TerminalQueryParser::default();
-    parser.feed(chunk)
-}
-
 fn workspace_context_label(
     workspace_id: Option<&str>,
     workspace_alias: Option<&str>,
@@ -1196,46 +1113,6 @@ mod tests {
         assert_eq!(detector.detect_activity(&output, expected_echo), None);
     }
 
-    #[test]
-    fn terminal_query_da1_no_param() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[c");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[?1;2c");
-    }
-
-    #[test]
-    fn terminal_query_da1_with_zero() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[0c");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[?1;2c");
-    }
-
-    #[test]
-    fn terminal_query_da2() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[>c");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[>1;10;0c");
-    }
-
-    #[test]
-    fn terminal_query_dsr() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[5n");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[0n");
-    }
-
-    #[test]
-    fn terminal_query_cpr_still_works() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[6n");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[1;1R");
-    }
-
     // ==================== detect_codex_model_prompt tests ====================
 
     #[test]
@@ -1469,55 +1346,6 @@ mod tests {
         assert!(!has_allow);
     }
 
-    // ==================== terminal query parser edge cases ====================
-
-    #[test]
-    fn terminal_query_multiple_queries_in_one_chunk() {
-        let mut parser = TerminalQueryParser::default();
-        // DA1 + CPR + DSR all in one chunk
-        let responses = parser.feed(b"\x1b[c\x1b[6n\x1b[5n");
-        assert_eq!(responses.len(), 3);
-        assert_eq!(responses[0], b"\x1b[?1;2c"); // DA1
-        assert_eq!(responses[1], b"\x1b[1;1R"); // CPR
-        assert_eq!(responses[2], b"\x1b[0n"); // DSR
-    }
-
-    #[test]
-    fn terminal_query_interleaved_with_text() {
-        let mut parser = TerminalQueryParser::default();
-        // Normal text + DA1 query + more text
-        let responses = parser.feed(b"Hello\x1b[cWorld");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[?1;2c");
-    }
-
-    #[test]
-    fn terminal_query_split_across_chunks() {
-        let mut parser = TerminalQueryParser::default();
-        // ESC in first chunk, [6n in second
-        let r1 = parser.feed(b"\x1b");
-        assert_eq!(r1.len(), 0);
-        let r2 = parser.feed(b"[6n");
-        assert_eq!(r2.len(), 1);
-        assert_eq!(r2[0], b"\x1b[1;1R");
-    }
-
-    #[test]
-    fn terminal_query_incomplete_sequence_reset() {
-        let mut parser = TerminalQueryParser::default();
-        // ESC [ but then a regular char (not a query) — should reset
-        let responses = parser.feed(b"\x1b[A");
-        assert_eq!(responses.len(), 0, "cursor up should not generate response");
-    }
-
-    #[test]
-    fn terminal_query_qmark_cpr() {
-        let mut parser = TerminalQueryParser::default();
-        let responses = parser.feed(b"\x1b[?6n");
-        assert_eq!(responses.len(), 1);
-        assert_eq!(responses[0], b"\x1b[?1;1R");
-    }
-
     // ==================== throttle edge cases ====================
 
     #[test]