feat(cli): serial-read --grep + vserial-start live-tail hint

Two small onboarding wins for AI callers hunting a specific line in a
noisy serial backlog:

1. serial-read --grep <regex>
   Server-side re.search filter applied BEFORE the tail/paging window,
   so non-matching entries do not consume the byte budget. Cursor
   semantics unchanged: 'next' still points into the raw ring so a
   caller can drop the filter and keep paging. Invalid regex returns
   success=false + invalid_grep=true + a human-readable error instead
   of a 500. Direct-mode falls back to a local re.search over the
   tail so the flag behaves the same either way.

2. vserial-start epilog with copy-pasteable live-tail recipes
   For continuous 'wait for PANIC' style workflows the right tool is a
   PTY + standard unix ('timeout 60 grep -m1 PANIC /tmp/fpb-tty…'),
   not a bespoke --follow flag. The updated epilog spells that out so
   a first-time AI doesn't reach for polling loops.

Tests:
  - core/serial_read.py: 7 new cases (tail filter, no-match, paging,
    budget accounting, invalid regex envelope, overflow-with-filter,
    grep=None regression guard).
  - routes/logs: grep filters before budget, invalid regex returns
    structured error (not 500).
  - cli/fpb_cli: proxy forwards grep, invalid_grep surfaces on the
    CLI JSON envelope so callers can branch on success.

Coverage 86.0% (gate 85%), lint clean, verified against live server:
valid regex returns only matching entries, invalid regex returns the
structured error envelope.

Author: FASTSHIFT

Tools/WebServer/app/routes/logs.py

@@ -123,6 +123,8 @@ def api_serial_read():
       max_bytes  hard cap on returned data bytes; default 4096
       tail       return only the newest N bytes (ignores ``since``)
       drop       "1"/"true" to skip the backlog and just advance the cursor
+      grep       optional regex; only entries matching (``re.search``) are
+                 considered before tail/paging is applied
 
     Returns data + next cursor + pending_bytes/entries + buffer_overflowed so a
     reader can page the whole backlog without ever over-reading its context.
@@ -134,6 +136,7 @@ def api_serial_read():
     tail = request.args.get("tail", 0, type=int)
     drop_raw = (request.args.get("drop", "") or "").lower()
     drop = drop_raw in ("1", "true", "yes")
+    grep = request.args.get("grep", None, type=str) or None
 
     device = state.device
     snapshot = list(device.raw_serial_log)
@@ -146,8 +149,9 @@ def api_serial_read():
         max_bytes=max_bytes,
         tail=tail,
         drop=drop,
+        grep=grep,
     )
-    result["success"] = True
+    result["success"] = not result.get("invalid_grep", False)
     return jsonify(result)
 
 

Tools/WebServer/cli/arg_parser.py

@@ -388,6 +388,15 @@ def build_parser(prog: str) -> argparse.ArgumentParser:
         help="Skip the buffered backlog and just advance the cursor "
         "(adb logcat -c style); returns the new 'next'.",
     )
+    serial_read_parser.add_argument(
+        "--grep",
+        type=str,
+        default=None,
+        help="Server-side regex filter (re.search); only matching entries "
+        "count against --tail / --max-bytes. Great for spotting PANIC/assert "
+        "in a big backlog without dragging the whole thing into context. "
+        "Invalid regex returns success=false with invalid_grep=true.",
+    )
 
     # file-list command (requires device)
     file_list_parser = subparsers.add_parser(
@@ -457,6 +466,18 @@ def build_parser(prog: str) -> argparse.ArgumentParser:
     vserial_start_parser = subparsers.add_parser(
         "vserial-start",
         help="Create virtual serial passthrough on the server (requires server)",
+        epilog=(
+            "Live tail with standard tools once the PTY exists:\n"
+            "  timeout 30 cat /tmp/fpb-ttyACM0                    "
+            "# adb-logcat-like tail\n"
+            "  timeout 60 grep --line-buffered -m1 PANIC "
+            "/tmp/fpb-ttyACM0   # exit on first hit\n"
+            "  tail -c 4096 <(timeout 5 cat /tmp/fpb-ttyACM0)     "
+            "# newest 4KB\n"
+            "For programmatic cursor-based reads with a byte budget and\n"
+            "server-side grep, use 'fpbinject serial-read --grep ...'."
+        ),
+        formatter_class=argparse.RawDescriptionHelpFormatter,
     )
     vserial_start_parser.add_argument(
         "--symlink",

Tools/WebServer/cli/fpb_cli.py

@@ -959,12 +959,15 @@ def serial_read(
         max_bytes: int = 4096,
         tail: int = 0,
         drop: bool = False,
+        grep: Optional[str] = None,
     ) -> None:
         """Read serial output, context-safe (adb-logcat style).
 
         Bounded by ``max_bytes`` so a large backlog never blows up the
         consumer's context. Defaults to a tail read; page the rest with
-        ``--since <next>`` or skip it with ``--drop``.
+        ``--since <next>`` or skip it with ``--drop``. Pass ``grep`` for
+        a server-side regex filter (proxy mode only — direct mode falls
+        back to a local ``re.search`` over the freshly-read tail).
         """
         try:
             if self._proxy:
@@ -978,6 +981,7 @@ def serial_read(
                     max_bytes=max_bytes,
                     tail=effective_tail,
                     drop=drop,
+                    grep=grep,
                 )
                 self._emit_serial_window(win)
                 return
@@ -1009,6 +1013,26 @@ def serial_read(
                 truncated = True
 
             log_lines = [ln for ln in new_data.split("\n") if ln.strip()][-lines:]
+            # Direct mode has no server-side ring; apply grep locally over
+            # the freshly-read tail so the CLI flag behaves the same either
+            # way. Invalid regex surfaces as a structured error.
+            if grep:
+                import re as _re
+
+                try:
+                    pat = _re.compile(grep)
+                except _re.error as re_err:
+                    self.output_json(
+                        {
+                            "success": False,
+                            "invalid_grep": True,
+                            "error": f"invalid grep pattern: {re_err}",
+                        }
+                    )
+                    return
+                log_lines = [ln for ln in log_lines if pat.search(ln)]
+                new_data = "\n".join(log_lines)
+                data_bytes = new_data.encode("utf-8")
             out = {
                 "success": True,
                 "new_data": new_data,
@@ -1030,12 +1054,15 @@ def _emit_serial_window(self, win: dict) -> None:
         """Emit a context-safe windowed read result (proxy mode).
 
         Builds a bounded JSON payload and attaches an actionable ``hint`` so
-        the caller knows how to page or skip the remaining backlog.
+        the caller knows how to page or skip the remaining backlog. Passes
+        through ``invalid_grep`` / ``error`` if the server rejected the
+        regex, so callers can branch on ``success`` without re-parsing text.
         """
         data = win.get("data", "")
         pending = win.get("pending_bytes", 0)
+        # Server flips success to false on invalid grep; propagate faithfully.
         out = {
-            "success": True,
+            "success": win.get("success", True),
             "data": data,
             "next": win.get("next", 0),
             "returned_bytes": win.get("returned_bytes", len(data.encode("utf-8"))),
@@ -1044,6 +1071,9 @@ def _emit_serial_window(self, win: dict) -> None:
             "truncated": win.get("truncated", False),
             "buffer_overflowed": win.get("buffer_overflowed", False),
         }
+        if win.get("invalid_grep"):
+            out["invalid_grep"] = True
+            out["error"] = win.get("error", "invalid grep pattern")
         hints = []
         if pending and pending > 0:
             hints.append(
@@ -1421,6 +1451,7 @@ def main():
                 args.max_bytes,
                 args.tail,
                 args.drop,
+                grep=getattr(args, "grep", None),
             )
         elif args.command == "doctor":
             cli.doctor(args.start_size, args.max_size, args.timeout, args.trials)

Tools/WebServer/cli/server_proxy.py

@@ -444,17 +444,25 @@ def serial_read_window(
         max_bytes: int = 4096,
         tail: int = 0,
         drop: bool = False,
+        grep: Optional[str] = None,
     ) -> dict:
         """Context-safe windowed serial read via /api/serial/read.
 
-        Returns dict with data / next / pending_bytes / pending_entries /
-        buffer_overflowed. Never returns more than ``max_bytes`` of data.
+        ``grep`` is an optional server-side regex (``re.search``) applied
+        BEFORE the tail/paging window, so only matching entries consume
+        the byte budget. Returns dict with data / next / pending_bytes /
+        pending_entries / buffer_overflowed. Never returns more than
+        ``max_bytes`` of data.
         """
+        from urllib.parse import quote
+
         q = f"/api/serial/read?since={since}&max_bytes={max_bytes}"
         if tail and tail > 0:
             q += f"&tail={tail}"
         if drop:
             q += "&drop=1"
+        if grep:
+            q += f"&grep={quote(grep, safe='')}"
         return self._get(q)
 
     # ------------------------------------------------------------------

Tools/WebServer/core/serial_read.py

@@ -30,7 +30,8 @@
 The logic here is pure (no Flask, no device) so it can be unit tested directly.
 """
 
-from typing import Dict, List
+import re
+from typing import Dict, List, Optional
 
 # Default byte budget for a single read. Small on purpose: a bare read must
 # never dump the whole backlog.
@@ -57,6 +58,15 @@ def _empty_result(next_id: int, overflowed: bool = False) -> Dict:
     }
 
 
+def _invalid_grep_result(next_id: int, message: str) -> Dict:
+    """Return an error envelope when the caller-supplied regex fails to
+    compile. Keeps the schema stable so callers can branch on ``error``."""
+    r = _empty_result(next_id)
+    r["error"] = f"invalid grep pattern: {message}"
+    r["invalid_grep"] = True
+    return r
+
+
 def compute_read(
     entries: List[dict],
     next_id: int,
@@ -65,6 +75,7 @@ def compute_read(
     max_bytes: int = DEFAULT_MAX_BYTES,
     tail: int = 0,
     drop: bool = False,
+    grep: Optional[str] = None,
 ) -> Dict:
     """Compute a context-safe windowed read over a ring buffer snapshot.
 
@@ -75,6 +86,12 @@ def compute_read(
         max_bytes: Hard cap on returned ``data`` bytes. <= 0 means DEFAULT.
         tail: If > 0, cap the returned window to the newest ``tail`` bytes.
         drop: If True, return no data and advance the cursor to ``next_id``.
+        grep: Optional regex; only entries whose ``data`` matches (``re.search``)
+            are considered. Filtering is applied FIRST, then the tail/paging
+            window is computed over the filtered list. ``since``/``next``
+            cursors still refer to the underlying buffer ids so callers can
+            keep paging after they drop the filter. Invalid regex returns an
+            error envelope (``invalid_grep=True``) instead of raising.
 
     Returns:
         dict: data, next, returned_bytes, pending_bytes, pending_entries,
@@ -87,21 +104,34 @@ def compute_read(
     if drop:
         return _empty_result(next_id)
 
-    # Overflow: a since-cursor older than the oldest retained id means the ring
-    # evicted data the caller had not read yet. Only meaningful for paging.
+    # Compile and apply the grep filter up-front (before overflow/tail/page
+    # calculations) so every downstream branch operates on the filtered list.
+    # We keep the original ids so the returned cursor remains meaningful.
+    filtered = entries
+    if grep:
+        try:
+            pat = re.compile(grep)
+        except re.error as e:
+            return _invalid_grep_result(next_id, str(e))
+        filtered = [e for e in entries if pat.search(e.get("data", ""))]
+
+    # Overflow: a since-cursor older than the oldest retained id in the RAW
+    # buffer means the ring evicted data the caller had not read yet -- the
+    # dropped entry may well have been a match, so we must still flag this
+    # even when a grep filter is active. Only meaningful for paging.
     overflowed = (
         not tail and since > 0 and bool(entries) and since < _earliest_id(entries)
     )
 
     # -------------------- tail mode (newest window) --------------------
     if tail and tail > 0:
         budget = min(tail, max_bytes)
-        total_all = _nbytes("".join(e.get("data", "") for e in entries))
+        total_all = _nbytes("".join(e.get("data", "") for e in filtered))
         # Walk entries from newest to oldest, accumulating whole entries until
         # the budget is reached; then byte-trim the oldest included entry.
         picked: List[str] = []
         acc = 0
-        for e in reversed(entries):
+        for e in reversed(filtered):
             d = e.get("data", "")
             picked.append(d)
             acc += _nbytes(d)
@@ -125,7 +155,7 @@ def compute_read(
         }
 
     # -------------------- paging mode (since, forward) --------------------
-    working = [e for e in entries if e.get("id", 0) >= since]
+    working = [e for e in filtered if e.get("id", 0) >= since]
     if not working:
         return _empty_result(next_id, overflowed)
 
@@ -159,7 +189,7 @@ def compute_read(
         truncated = True
         cursor = working[0].get("id", since)
 
-    remaining = [e for e in entries if e.get("id", 0) >= cursor]
+    remaining = [e for e in filtered if e.get("id", 0) >= cursor]
     pending_bytes = _nbytes("".join(e.get("data", "") for e in remaining))
     pending_entries = len(remaining)
     if pending_bytes > 0:

Tools/WebServer/tests/test_cli_coexistence.py

@@ -454,6 +454,7 @@ def do_GET(self):
             max_bytes = int(qs.get("max_bytes", [4096])[0])
             tail = int(qs.get("tail", [0])[0])
             drop = (qs.get("drop", [""])[0] or "").lower() in ("1", "true", "yes")
+            grep = qs.get("grep", [None])[0]
             next_id = (
                 max(e["id"] for e in self.log_entries) + 1 if self.log_entries else 0
             )
@@ -464,8 +465,10 @@ def do_GET(self):
                 max_bytes=max_bytes,
                 tail=tail,
                 drop=drop,
+                grep=grep,
             )
-            result["success"] = True
+            # Match the real route: success flips false on invalid grep.
+            result["success"] = not result.get("invalid_grep", False)
             body = json.dumps(result).encode()
             self.send_response(200)
             self.send_header("Content-Type", "application/json")
@@ -599,6 +602,43 @@ def test_serial_read_incremental_workflow(self):
         _CursorMockHandler.log_entries.pop()
         cli.cleanup()
 
+    def test_serial_read_grep_filters_via_proxy(self):
+        """serial_read(grep=...) forwards to /api/serial/read; only matches
+        come back, and the JSON envelope carries the same schema."""
+        _CursorMockHandler.log_entries.extend(
+            [
+                {"id": 3, "data": "boot: ok\n"},
+                {"id": 4, "data": "PANIC: null deref\n"},
+                {"id": 5, "data": "boot: heartbeat\n"},
+            ]
+        )
+        try:
+            cli = self._make_cli()
+            buf = io.StringIO()
+            with redirect_stdout(buf):
+                cli.serial_read(grep="PANIC")
+            result = json.loads(buf.getvalue())
+            self.assertTrue(result["success"])
+            self.assertIn("PANIC", result["data"])
+            self.assertNotIn("boot:", result["data"])
+            cli.cleanup()
+        finally:
+            del _CursorMockHandler.log_entries[3:]
+
+    def test_serial_read_invalid_grep_surfaces_error(self):
+        """Invalid regex must round-trip as success=false + invalid_grep=true
+        so JSON callers can branch without regex-matching the error string."""
+        cli = self._make_cli()
+        buf = io.StringIO()
+        with redirect_stdout(buf):
+            cli.serial_read(grep="[unclosed")
+        result = json.loads(buf.getvalue())
+        self.assertFalse(result["success"])
+        self.assertTrue(result.get("invalid_grep"))
+        self.assertIn("invalid grep pattern", result.get("error", ""))
+        self.assertEqual(result["data"], "")
+        cli.cleanup()
+
 
 class TestFPBCLIIsRemoteUrl(unittest.TestCase):
     """Test the URL locality classifier."""

Tools/WebServer/tests/test_routes.py

@@ -1647,6 +1647,42 @@ def test_serial_read_window_overflow_flag(self):
         self.assertTrue(data["success"])
         self.assertTrue(data["buffer_overflowed"])
 
+    def test_serial_read_window_grep_filters_before_budget(self):
+        """?grep=... keeps only matching entries; non-matches don't count."""
+        state.device.raw_serial_log = [
+            {"id": 0, "data": "boot: idle\n"},
+            {"id": 1, "data": "PANIC: null deref\n"},
+            {"id": 2, "data": "boot: ok\n"},
+            {"id": 3, "data": "assert: overflow\n"},
+        ]
+        state.device.raw_log_next_id = 4
+
+        response = self.client.get(
+            "/api/serial/read?tail=4096&max_bytes=4096&grep=PANIC%7Cassert"
+        )
+        data = json.loads(response.data)
+
+        self.assertTrue(data["success"])
+        self.assertIn("PANIC", data["data"])
+        self.assertIn("assert", data["data"])
+        self.assertNotIn("boot:", data["data"])
+
+    def test_serial_read_window_grep_invalid_regex(self):
+        """Invalid regex returns a structured error, not a 500."""
+        state.device.raw_serial_log = [{"id": 0, "data": "hi\n"}]
+        state.device.raw_log_next_id = 1
+
+        response = self.client.get("/api/serial/read?tail=4096&grep=%5Bunclosed")
+        self.assertEqual(response.status_code, 200)
+        data = json.loads(response.data)
+
+        # success flips to false when the pattern is invalid so JSON callers
+        # can branch on it; the error text tells the human what went wrong.
+        self.assertFalse(data["success"])
+        self.assertTrue(data.get("invalid_grep"))
+        self.assertIn("invalid grep pattern", data.get("error", ""))
+        self.assertEqual(data["data"], "")
+
     def test_serial_send_no_data(self):
         """Test serial send without data"""
         response = self.client.post(