fix: waits for startup by markers instead of timer

src/claude_code_api/backend.py

@@ -81,7 +81,7 @@ class BackendOptions:
     history_injection_mode: HistoryInjectionMode = "native_jsonl"
     wait_for_turn_duration: bool = False
     include_meta_user: bool = False
-    startup_delay: float = 1.0
+    startup_delay: float = 10.0
     file_wait_timeout: float = 30.0
     turn_duration_timeout: float = 5.0
 

src/claude_code_api/pty.py

@@ -282,6 +282,37 @@ class PtyClaudeProcess:
                 if overflow > 0:
                     del self._output_buffer[:overflow]
 
+    async def wait_for_output(
+        self,
+        marker: bytes,
+        *,
+        timeout: float,
+        poll: float = 0.05,
+    ) -> bool:
+        """Poll captured PTY output for `marker`; return True once seen.
+
+        Returns False on timeout. Intended for readiness signals that the
+        child writes to its PTY (e.g. claude's TUI emits ``\\x1b[?2004h``
+        — DECSET 2004 / bracketed-paste-enable — once it's ready to accept
+        pasted input). Using this beats a fixed sleep because the wait
+        completes as soon as the signal arrives, while still bounding the
+        worst case.
+        """
+        if timeout < 0:
+            msg = f"timeout must be non-negative, got {timeout!r}"
+            raise ValueError(msg)
+        if poll <= 0:
+            msg = f"poll must be positive, got {poll!r}"
+            raise ValueError(msg)
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout
+        while True:
+            if marker in self.captured_output():
+                return True
+            if loop.time() >= deadline:
+                return False
+            await asyncio.sleep(poll)
+
     async def write(self, data: str | bytes, *, newline: bool = True) -> int:
         r"""Write bytes to the child's stdin.
 

src/claude_code_api/turn.py

@@ -49,7 +49,15 @@ _TERMINAL_STOP_REASONS: frozenset[str] = frozenset(
 
 _DEFAULT_FILE_WAIT_TIMEOUT = 30.0
 _DEFAULT_TURN_DURATION_TIMEOUT = 5.0
-_DEFAULT_STARTUP_DELAY = 1.0
+_DEFAULT_STARTUP_DELAY = 10.0
+
+# DECSET 2004 — claude's TUI enables bracketed paste once it is ready to
+# accept pasted input. Writing the user prompt before this byte sequence
+# arrives causes the paste-mode markers to be parsed as raw escape codes
+# and the message is silently dropped. We poll for it in `start()` instead
+# of a fixed sleep so a slow box (e.g. a Raspberry Pi where claude's
+# initial render takes ~2-3s) doesn't lose the first turn.
+_TUI_READY_MARKER: bytes = b"\x1b[?2004h"
 
 
 ParseErrorCallback = Callable[[MessageParseError, JsonlRecord], None]
@@ -125,12 +133,24 @@ class TurnManager:
         return self._turn_count
 
     async def start(self) -> None:
-        """Spawn the PTY and let claude's TUI settle. Idempotent."""
+        """Spawn the PTY and let claude's TUI settle. Idempotent.
+
+        `startup_delay` is now a cap, not a fixed sleep: we wait for
+        claude's bracketed-paste-enable byte (DECSET 2004) to appear in
+        PTY output and return as soon as it does — typically faster than
+        `startup_delay`, but on a slow host where the marker takes longer
+        we still bound the wait. Fake PTYs in tests may not expose
+        `wait_for_output`; we fall back to a plain sleep then.
+        """
         if self._started:
             return
         await self._pty.start()
         if self._startup_delay > 0:
-            await asyncio.sleep(self._startup_delay)
+            wait = getattr(self._pty, "wait_for_output", None)
+            if wait is not None:
+                await wait(_TUI_READY_MARKER, timeout=self._startup_delay)
+            else:
+                await asyncio.sleep(self._startup_delay)
         self._started = True
 
     async def send_user_message(self, text: str) -> AsyncIterator[Event]: