ControlProtocol(fix): Add SKIPPING state for unexpected %begin blocks

why: Hook commands trigger additional %begin/%end blocks that
desync the command queue, causing protocol errors and timeouts.
what:
- Add ParserState.SKIPPING to handle unexpected %begin blocks
- Skip block content instead of marking connection DEAD
- Return to IDLE when skipped block ends with %end/%error
- Update protocol test to verify new SKIPPING behavior

Author: tony

src/libtmux/_internal/engines/control_protocol.py

@@ -61,6 +61,7 @@ class ParserState(enum.Enum):
 
     IDLE = enum.auto()
     IN_COMMAND = enum.auto()
+    SKIPPING = enum.auto()  # Skipping unexpected %begin/%end block
     DEAD = enum.auto()
 
 
@@ -214,10 +215,17 @@ def _handle_percent_line(self, line: str) -> None:
     def _handle_plain_line(self, line: str) -> None:
         if self.state is ParserState.IN_COMMAND and self._current:
             self._current.stdout.append(line)
+        elif self.state is ParserState.SKIPPING:
+            # Ignore output from skipped blocks (hook command output)
+            pass
         else:
             logger.debug("Unexpected plain line outside command: %r", line)
 
     def _on_begin(self, parts: list[str]) -> None:
+        if self.state is ParserState.SKIPPING:
+            # Nested %begin while skipping - ignore
+            logger.debug("Nested %%begin while skipping: %s", parts)
+            return
         if self.state is not ParserState.IDLE:
             self._protocol_error("nested %begin")
             return
@@ -233,7 +241,12 @@ def _on_begin(self, parts: list[str]) -> None:
         try:
             ctx = self._pending.popleft()
         except IndexError:
-            self._protocol_error(f"no pending command for %begin id={cmd_id}")
+            # No pending command - this is likely from a hook action.
+            # Skip this block instead of killing the connection.
+            logger.debug(
+                "Unexpected %%begin id=%d (hook execution?), skipping block", cmd_id
+            )
+            self.state = ParserState.SKIPPING
             return
 
         ctx.cmd_id = cmd_id
@@ -244,6 +257,11 @@ def _on_begin(self, parts: list[str]) -> None:
         self.state = ParserState.IN_COMMAND
 
     def _on_end_or_error(self, tag: str, parts: list[str]) -> None:
+        if self.state is ParserState.SKIPPING:
+            # End of skipped block - return to idle
+            logger.debug("Skipped block ended with %s", tag)
+            self.state = ParserState.IDLE
+            return
         if self.state is not ParserState.IN_COMMAND or self._current is None:
             self._protocol_error(f"unexpected {tag}")
             return

tests/test_engine_protocol.py

@@ -92,11 +92,6 @@ def test_control_protocol_notifications() -> None:
         line="%end 123 1 0",
         expected_reason="unexpected %end",
     ),
-    ProtocolErrorFixture(
-        test_id="no_pending_begin",
-        line="%begin 999 1 0",
-        expected_reason="no pending command for %begin",
-    ),
 ]
 
 
@@ -111,6 +106,27 @@ def test_control_protocol_errors(case: ProtocolErrorFixture) -> None:
     assert case.expected_reason in stats.last_error
 
 
+def test_control_protocol_skips_unexpected_begin() -> None:
+    """Unexpected %begin (e.g., from hook execution) should enter SKIPPING state.
+
+    This is not a fatal error - hooks can trigger additional %begin/%end blocks
+    that have no matching registered command. The protocol skips these blocks
+    instead of marking the connection dead.
+    """
+    proto = ControlProtocol()
+    proto.feed_line("%begin 999 1 0")
+    assert proto.state is ParserState.SKIPPING
+    # Output during skipped block is ignored
+    proto.feed_line("some hook output")
+    assert proto.state is ParserState.SKIPPING
+    # End of skipped block returns to IDLE
+    proto.feed_line("%end 999 1 0")
+    assert proto.state is ParserState.IDLE
+    # Connection is still usable
+    stats = proto.get_stats(restarts=0)
+    assert stats.last_error is None
+
+
 NOTIFICATION_FIXTURES: list[NotificationFixture] = [
     NotificationFixture(
         test_id="layout_change",