diff --git a/omnigent/claude_native_forwarder.py b/omnigent/claude_native_forwarder.py index 2d858709cb..b364cb7eb1 100644 --- a/omnigent/claude_native_forwarder.py +++ b/omnigent/claude_native_forwarder.py @@ -854,6 +854,7 @@ async def forward_claude_transcript_to_session( poll_interval_s: float = _DEFAULT_POLL_INTERVAL_S, auth: httpx.Auth | None = None, skip_user_messages: bool = False, + start_at_offset: int | None = None, ) -> None: """ Tail Claude's JSONL transcript and mirror semantic items into AP. @@ -875,6 +876,12 @@ async def forward_claude_transcript_to_session( :param start_at_end: When ``True`` and no prior forward cursor exists, start from the current transcript end. This is used for reattach so old transcript lines are not duplicated. + Ignored when *start_at_offset* is set. + :param start_at_offset: Byte length of a resume prefix this launch + synthesized, e.g. ``5920``. Preferred over *start_at_end* on the + cold-resume path: the exact prefix is known before launch, where a + live end-offset measured after Claude boots can skip a prompt the + executor injected in the meantime. :param poll_interval_s: Seconds between transcript polls. :param auth: Optional httpx Auth that mints a fresh bearer token per request, e.g. ``_server_auth(profile)`` for a Databricks @@ -1040,6 +1047,7 @@ async def forward_claude_transcript_to_session( transcript_path=transcript_path, start_at_end=start_at_end, session_id=current_session_id, + start_at_offset=start_at_offset, ) # Forward streamed deltas BEFORE the transcript items so a # message's live chunks (incl. its ``final`` chunk) always @@ -2056,6 +2064,7 @@ async def supervise_forwarder( poll_interval_s: float = _DEFAULT_POLL_INTERVAL_S, auth: httpx.Auth | None = None, skip_user_messages: bool = False, + start_at_offset: int | None = None, ) -> None: """ Run :func:`forward_claude_transcript_to_session` under a restart supervisor. @@ -2092,6 +2101,9 @@ async def supervise_forwarder( :param agent_name: Agent/model name to stamp on mirrored output. :param start_at_end: When ``True`` and no prior forward cursor exists, start from the current transcript end. + :param start_at_offset: Byte length of a resume prefix this launch + synthesized. Forwarded verbatim; see + :func:`forward_claude_transcript_to_session`. :param poll_interval_s: Seconds between transcript polls inside the forwarder loop. Forwarded verbatim. :param auth: Optional httpx Auth that mints a fresh bearer token @@ -2114,6 +2126,7 @@ async def supervise_forwarder( poll_interval_s=poll_interval_s, auth=auth, skip_user_messages=skip_user_messages, + start_at_offset=start_at_offset, ) # The forwarder loop is ``while True`` and is not expected # to return normally. Treat any normal return as a crash @@ -3098,6 +3111,7 @@ async def _ensure_state_for_transcript( transcript_path: Path, start_at_end: bool, session_id: str, + start_at_offset: int | None = None, ) -> TranscriptForwardState: """ Return a cursor state compatible with the observed transcript. @@ -3106,9 +3120,14 @@ async def _ensure_state_for_transcript( :param state: Existing cursor state, or ``None``. :param transcript_path: Current transcript path from hooks. :param start_at_end: Whether a missing cursor should skip the - transcript's existing lines. + transcript's existing lines. Only consulted when + *start_at_offset* is ``None``. :param session_id: Omnigent session/conversation id, e.g. ``"conv_abc123"``. Used for stale-cursor diagnostics. + :param start_at_offset: Exact byte length of a prefix this launch + synthesized itself, e.g. ``5920``. Takes precedence over + *start_at_end* — see the seeding comment below for why a measured + prefix is required rather than a live ``stat``. :returns: Cursor state for ``transcript_path``. """ if state is not None and state.transcript_path == transcript_path: @@ -3131,7 +3150,22 @@ async def _ensure_state_for_transcript( await _write_forward_state_async(bridge_dir, validated) return validated byte_offset = 0 - if start_at_end: + if start_at_offset is not None: + # Cold resume: the caller wrote the prefix and measured it before + # launching Claude, so skip exactly that and nothing else. + # + # Seeding from a live ``stat`` here loses messages. Resolving + # ``transcript_path`` requires Claude to boot and fire its first hook, + # and the executor's ``inject_user_message`` waits on the same boot — + # the two are unordered, so the paste routinely wins. Whatever Claude + # wrote in that window (the user's prompt included) then sits *behind* + # the seeded cursor and is skipped for the session's lifetime: visible + # in the TUI pane, absent from the Omnigent DB, with no error anywhere. + end_offset = await asyncio.to_thread(_transcript_end_offset, transcript_path) + byte_offset = min(start_at_offset, end_offset) + elif start_at_end: + # Reattach: nothing was synthesized, so the whole existing transcript + # is content Omnigent already holds and a live end-offset is correct. byte_offset = await asyncio.to_thread(_transcript_end_offset, transcript_path) state = TranscriptForwardState( transcript_path=transcript_path, diff --git a/omnigent/runner/native/orchestration.py b/omnigent/runner/native/orchestration.py index 347487b1dc..0def231167 100644 --- a/omnigent/runner/native/orchestration.py +++ b/omnigent/runner/native/orchestration.py @@ -5561,6 +5561,30 @@ def _publish_terminal_pending( ) +def _measured_prefix_bytes(transcript_path: Path) -> int | None: + """ + Measure a just-written resume transcript so the forwarder can skip exactly it. + + Taken before Claude launches, while the file holds only the synthesized + prefix. ``None`` on any read failure, which leaves the forwarder on its + live end-offset fallback. + + :param transcript_path: Resume transcript this launch wrote, e.g. + ``Path("~/.claude/projects/-Users-me-repo/.jsonl")``. + :returns: File size in bytes, or ``None`` when it cannot be measured. + """ + try: + return transcript_path.stat().st_size + except OSError: + _logger.warning( + "Could not measure synthesized Claude resume transcript; " + "forwarder will seed from the live transcript end; transcript=%s", + transcript_path, + exc_info=True, + ) + return None + + def _native_terminal_start_error_payload(exc: BaseException, runtime_name: str) -> dict[str, str]: """ Build the structured error payload for a native terminal start failure. @@ -6153,6 +6177,12 @@ async def _auto_create_claude_terminal( # transcript that doesn't exist. See # designs/NATIVE_RUNNER_SERVER_LAUNCH.md. resume_external_session_id: str | None = None + # Byte length of the resume transcript this launch synthesized, measured + # BEFORE Claude starts. The forwarder seeds its cursor from this rather than + # from a live end-offset: resolving the transcript path needs Claude's first + # hook, and the executor's prompt inject waits on the same boot, so a + # ``stat`` taken later routinely skips the freshly-injected message. + resume_prefix_bytes: int | None = None if server_client is not None and session_external_id is not None: from omnigent.claude_native import _ensure_local_claude_resume_transcript @@ -6165,6 +6195,7 @@ async def _auto_create_claude_terminal( ) if _transcript is not None: resume_external_session_id = session_external_id + resume_prefix_bytes = _measured_prefix_bytes(_transcript) except Exception: # noqa: BLE001 — best-effort; launch fresh on failure _logger.warning( "Could not synthesize Claude resume transcript for %s; launching without --resume", @@ -6211,6 +6242,7 @@ async def _auto_create_claude_terminal( if _cloned is not None: # Resume our OWN clone (plain --resume, no --fork-session). resume_external_session_id = our_uuid + resume_prefix_bytes = _measured_prefix_bytes(_cloned) # Record the assigned id now so Omnigent reflects the clone's own # Claude session immediately, and a later relaunch resumes it # via the normal cold-resume path (this branch is gated on @@ -6273,6 +6305,7 @@ async def _auto_create_claude_terminal( ) if _built is not None: resume_external_session_id = our_uuid + resume_prefix_bytes = _measured_prefix_bytes(_built) # Record the assigned id so Omnigent reflects the clone's own Claude # session and a later relaunch resumes it via the cold-resume # path above. Best-effort, mirroring the clone branch. @@ -6586,6 +6619,7 @@ async def _supervise_bridge() -> None: bridge_dir=bridge_dir, agent_name="claude-native-ui", start_at_end=resume_external_session_id is not None, + start_at_offset=resume_prefix_bytes, auth=_runner_auth, ) finally: diff --git a/tests/test_claude_native_forwarder.py b/tests/test_claude_native_forwarder.py index 83705378be..cd409e3a07 100644 --- a/tests/test_claude_native_forwarder.py +++ b/tests/test_claude_native_forwarder.py @@ -2312,6 +2312,100 @@ async def test_forwarder_waits_for_missing_fresh_transcript_without_warning( assert "cursor fingerprint changed" not in caplog.text +@pytest.mark.asyncio +async def test_measured_prefix_seed_keeps_a_prompt_injected_during_boot( + tmp_path: Path, +) -> None: + """ + Regression: a prompt Claude records while booting must still forward. + + Cold resume writes the transcript prefix itself, then launches Claude. The + forwarder cannot seed until Claude's first hook advertises the transcript + path — and the executor's ``inject_user_message`` waits on the same boot, + so the paste routinely lands first. Seeding from a live end-offset then + puts the user's prompt BEHIND the cursor: visible in the TUI pane, absent + from the Omnigent DB, silently, for the session's lifetime. + + Passing the prefix length measured before launch makes the skip exactly the + prefix, so the boot-window records survive however late the seed runs. + """ + bridge_dir = tmp_path / "bridge" + bridge_dir.mkdir() + transcript_path = tmp_path / "session.jsonl" + # The synthesized prefix, complete before Claude starts. + transcript_path.write_text( + "".join( + json.dumps({"type": "user", "uuid": f"old{n}", "message": {"role": "user"}}) + "\n" + for n in range(3) + ), + encoding="utf-8", + ) + prefix_bytes = transcript_path.stat().st_size + # Claude boots and records the freshly-injected prompt before the forwarder + # is scheduled to seed. + with transcript_path.open("a", encoding="utf-8") as handle: + handle.write( + json.dumps( + { + "type": "user", + "uuid": "boot-window-prompt", + "message": {"role": "user", "content": "wake up and check the deploy"}, + } + ) + + "\n" + ) + + state = await forwarder._ensure_state_for_transcript( + bridge_dir=bridge_dir, + state=None, + transcript_path=transcript_path, + start_at_end=True, + session_id="conv_boot_window", + start_at_offset=prefix_bytes, + ) + + # The measured prefix wins over ``start_at_end``: the cursor sits at the + # prefix boundary, not at EOF, so the prompt is still ahead of it. + assert state.byte_offset == prefix_bytes + result = forwarder._read_transcript_items_for_state(state, "claude-native-ui", None) + forwarded = [ + block.get("text") + for item in result.items + for block in (item.data.get("content") or []) + if isinstance(block, dict) + ] + assert "wake up and check the deploy" in forwarded + + +@pytest.mark.asyncio +async def test_measured_prefix_never_seeks_past_the_transcript_end(tmp_path: Path) -> None: + """ + A prefix length larger than the file clamps to the end. + + Defensive: the measurement and the seed are separated by Claude's launch, + so a truncated or replaced transcript would otherwise leave the cursor + beyond EOF, where every later read looks like a stale-cursor reset. + """ + bridge_dir = tmp_path / "bridge" + bridge_dir.mkdir() + transcript_path = tmp_path / "session.jsonl" + transcript_path.write_text( + json.dumps({"type": "user", "uuid": "only", "message": {"role": "user"}}) + "\n", + encoding="utf-8", + ) + + state = await forwarder._ensure_state_for_transcript( + bridge_dir=bridge_dir, + state=None, + transcript_path=transcript_path, + start_at_end=True, + session_id="conv_clamp", + start_at_offset=10**9, + ) + + assert state.byte_offset == transcript_path.stat().st_size + + @pytest.mark.asyncio async def test_forwarder_skips_to_end_on_stale_byte_cursor_state(tmp_path: Path) -> None: """