fix(record): correct live-draft waits/hovers under load + cap enrichment fan-out (#6873)

skyvern/config.py

@@ -274,6 +274,10 @@ class Settings(BaseSettings):
     # set both to record at an explicit resolution.
     BROWSER_RECORDING_WIDTH: int | None = None
     BROWSER_RECORDING_HEIGHT: int | None = None
+    # Max concurrent LLM enrichment calls per live browser-recording interpretation
+    # session. Bounds the per-action enrichment fan-out so a burst of interactions
+    # can't flood the event loop with simultaneous LLM requests.
+    RECORDING_ENRICHMENT_MAX_CONCURRENCY: int = 4
     BROWSER_POLICY_FILE: str = "/etc/chromium/policies/managed/policies.json"
     BROWSER_LOGS_ENABLED: bool = True
     BROWSER_CURSOR_VISUALIZATION: bool = False

skyvern/forge/sdk/routes/streaming/channels/exfiltration.py

@@ -43,6 +43,10 @@ class ExfiltratedEvent:
     params: dict = dataclasses.field(default_factory=dict)
     source: ExfiltratedEventSource = ExfiltratedEventSource.NOT_SPECIFIED
     timestamp: float = dataclasses.field(default_factory=lambda: time.time())  # seconds since epoch
+    # Monotonic order assigned at the earliest synchronous capture point so the
+    # interpreter can restore chronological order after async materialization
+    # (e.g. console json_value round-trips) reorders events under load.
+    capture_seq: int = -1
 
 
 OnExfiltrationEvent = t.Callable[[list[ExfiltratedEvent]], None]
@@ -82,9 +86,15 @@ class ExfiltrationChannel(CdpChannel):
         self._last_network_activity_emit = 0.0
         self._network_activity_flush_task: asyncio.Task[None] | None = None
         self._capture_paused = False
+        self._capture_seq = 0
 
         super().__init__(vnc_channel=vnc_channel)
 
+    def _next_capture_seq(self) -> int:
+        seq = self._capture_seq
+        self._capture_seq += 1
+        return seq
+
     def pause_capture(self) -> None:
         self._capture_paused = True
 
@@ -189,7 +199,7 @@ class ExfiltrationChannel(CdpChannel):
         self._recent_console_event_fingerprints[fingerprint] = now
         return True
 
-    def _emit_console_event(self, event_data: dict[str, t.Any]) -> None:
+    def _emit_console_event(self, event_data: dict[str, t.Any], capture_seq: int) -> None:
         if not self._should_emit_console_event(event_data):
             return
 
@@ -201,6 +211,7 @@ class ExfiltrationChannel(CdpChannel):
                     params=event_data,
                     source=ExfiltratedEventSource.CONSOLE,
                     timestamp=time.time(),
+                    capture_seq=capture_seq,
                 )
             ]
         )
@@ -212,9 +223,9 @@ class ExfiltrationChannel(CdpChannel):
         if event_data is None:
             return
 
-        active_channel._emit_console_event(event_data)
+        active_channel._emit_console_event(event_data, active_channel._next_capture_seq())
 
-    async def _handle_console_event_async(self, msg: ConsoleMessage) -> None:
+    async def _handle_console_event_async(self, msg: ConsoleMessage, capture_seq: int) -> None:
         """Parse Playwright console messages for exfiltrated event data."""
         event_data: dict[str, t.Any] | None = None
         try:
@@ -232,12 +243,12 @@ class ExfiltrationChannel(CdpChannel):
         if event_data is None:
             return
 
-        self._emit_console_event(event_data)
+        self._emit_console_event(event_data, capture_seq)
 
     def _handle_console_event(self, msg: ConsoleMessage) -> None:
-        self._track_event_task(self._handle_console_event_async(msg))
+        self._track_event_task(self._handle_console_event_async(msg, self._next_capture_seq()))
 
-    async def _handle_runtime_console_event_async(self, params: dict[str, t.Any]) -> None:
+    async def _handle_runtime_console_event_async(self, params: dict[str, t.Any], capture_seq: int) -> None:
         raw_args = params.get("args")
         if not isinstance(raw_args, list):
             return
@@ -246,14 +257,16 @@ class ExfiltrationChannel(CdpChannel):
         if event_data is None:
             return
 
-        self._emit_console_event(event_data)
+        self._emit_console_event(event_data, capture_seq)
 
     async def _attach_page_cdp_console_capture(self, page: Page) -> CDPSession | None:
         cdp_session = await page.context.new_cdp_session(page)
         await cdp_session.send("Runtime.enable")
         cdp_session.on(
             "Runtime.consoleAPICalled",
-            lambda params: self._track_event_task(self._handle_runtime_console_event_async(params)),
+            lambda params: self._track_event_task(
+                self._handle_runtime_console_event_async(params, self._next_capture_seq())
+            ),
         )
         return cdp_session
 
@@ -335,6 +348,7 @@ class ExfiltrationChannel(CdpChannel):
                 params=params,
                 source=ExfiltratedEventSource.CDP,
                 timestamp=time.time(),
+                capture_seq=self._next_capture_seq(),
             ),
         ]
 

skyvern/services/browser_recording/interpretation.py

@@ -13,6 +13,7 @@ from skyvern.client.types.workflow_definition_yaml_blocks_item import (
     WorkflowDefinitionYamlBlocksItem_GotoUrl,
     WorkflowDefinitionYamlBlocksItem_Wait,
 )
+from skyvern.config import settings
 from skyvern.forge.sdk.routes.streaming.channels import exfiltration as streaming_exfiltration
 from skyvern.forge.sdk.routes.streaming.channels.exfiltration import ExfiltratedEventSource
 from skyvern.services.browser_recording.service import (
@@ -39,6 +40,10 @@ LOG = structlog.get_logger(__name__)
 
 INTERPRETATION_DEBOUNCE_SECONDS = 0.4
 INTERPRETATION_MAX_WAIT_SECONDS = 2.0
+# events_to_actions is pure CPU and runs on the event loop; log when a single pass
+# is slow enough to risk starving the raw-event/WebSocket path so we can decide
+# whether to offload it to a thread.
+EVENTS_TO_ACTIONS_SLOW_MS = 50.0
 SIGNIFICANT_CONSOLE_EVENT_TYPES = {
     "blur",
     "change",
@@ -258,6 +263,7 @@ class RecordingInterpretationSession:
         self._debounce_task: asyncio.Task[None] | None = None
         self._pending_since: float | None = None
         self._enrichment_tasks: set[asyncio.Task[None]] = set()
+        self._enrichment_semaphore = asyncio.Semaphore(settings.RECORDING_ENRICHMENT_MAX_CONCURRENCY)
         self._interpret_lock = asyncio.Lock()
         self._action_machines: list[sm.StateMachine] = [
             sm.Click(),
@@ -292,6 +298,14 @@ class RecordingInterpretationSession:
 
         self.events.extend(reified_events)
 
+        # Async materialization (e.g. console json_value round-trips) can append
+        # events out of true order under load. Re-sort only the not-yet-interpreted
+        # tail by capture order; never touch the processed prefix, whose emitted
+        # actions are tracked by index. sorted() is stable, so legacy events without
+        # a capture_seq (-1) keep their arrival order.
+        tail_start = self._processed_event_count
+        self.events[tail_start:] = sorted(self.events[tail_start:], key=lambda event: event.capture_seq)
+
         if not any(event_should_trigger_interpretation(event) for event in reified_events):
             return
 
@@ -350,11 +364,21 @@ class RecordingInterpretationSession:
 
             if self._processed_event_count < len(self.events):
                 new_events = self.events[self._processed_event_count :]
+                started_at = time.perf_counter()
                 self._all_actions = processor.events_to_actions(
                     new_events,
                     machines=self._action_machines,
                     initial_actions=self._all_actions,
                 )
+                elapsed_ms = (time.perf_counter() - started_at) * 1000.0
+                if elapsed_ms >= EVENTS_TO_ACTIONS_SLOW_MS:
+                    LOG.warning(
+                        "Slow events_to_actions pass blocked the event loop",
+                        browser_session_id=self.browser_session_id,
+                        elapsed_ms=round(elapsed_ms, 1),
+                        new_event_count=len(new_events),
+                        total_action_count=len(self._all_actions),
+                    )
                 self._processed_event_count = len(self.events)
 
             new_actions = self._all_actions[self.emitted_action_count :]
@@ -444,7 +468,8 @@ class RecordingInterpretationSession:
         enriched: RecordingDraftStep | None = None
 
         try:
-            block = await processor.create_action_block(action)
+            async with self._enrichment_semaphore:
+                block = await processor.create_action_block(action)
             enriched = _draft_step_from_block(
                 browser_session_id=self.browser_session_id,
                 action_index=action_index,

skyvern/services/browser_recording/state_machines/wait.py

@@ -37,6 +37,12 @@ class StateMachineWait(StateMachine):
             maxlen=MAX_PAGE_ACTIVITY_TIMESTAMPS
         )
 
+        # client_ms - server_ms, measured from console events (which carry both
+        # clocks). Used to project server-stamped CDP activity timestamps into the
+        # client clock so the gap and the busy-span are compared in one domain.
+        # A property of the connection, not the wait window — never cleared on reset.
+        self.client_server_offset_ms: float | None = None
+
         self.reset()
 
     def tick(self, event: ExfiltratedEvent, current_actions: list[Action]) -> ActionWait | None:
@@ -44,12 +50,15 @@ class StateMachineWait(StateMachine):
             if event.event_name == PAGE_ACTIVITY_EVENT_NAME or event.event_name.startswith(
                 PAGE_ACTIVITY_EVENT_NAME_PREFIX
             ):
-                self.page_activity_timestamps_ms.append(event.timestamp * 1000.0)
+                offset_ms = self.client_server_offset_ms or 0.0
+                self.page_activity_timestamps_ms.append(event.timestamp * 1000.0 + offset_ms)
             return None
 
         if event.source != "console":
             return None
 
+        self.client_server_offset_ms = event.params.timestamp - event.timestamp * 1000.0
+
         if self.last_event_timestamp is not None:
             gap_ms = int(event.params.timestamp - self.last_event_timestamp)
 

skyvern/services/browser_recording/types.py

@@ -238,6 +238,9 @@ class ExfiltratedCdpEvent(BaseModel):
     params: ExfiltratedEventCdpParams
     source: Literal["cdp"]
     timestamp: float
+    # Monotonic server-receipt order, assigned before async materialization can
+    # reorder events. -1 for payloads that predate the field (legacy/batch replay).
+    capture_seq: int = -1
 
 
 class ExfiltratedConsoleEvent(BaseModel):
@@ -246,6 +249,7 @@ class ExfiltratedConsoleEvent(BaseModel):
     params: ExfiltratedEventConsoleParams
     source: Literal["console"]
     timestamp: float
+    capture_seq: int = -1
 
 
 ExfiltratedEvent = ExfiltratedCdpEvent | ExfiltratedConsoleEvent

tests/unit/forge/sdk/routes/streaming/test_exfiltration_channel.py

@@ -123,7 +123,7 @@ class TestExfiltrationChannelEvents:
         message.args = []
         message.text = f"[EXFIL] {json.dumps(event_data)}"
 
-        await channel._handle_console_event_async(message)
+        await channel._handle_console_event_async(message, 0)
 
         on_event.assert_called_once()
         assert on_event.call_args.args[0][0].params == event_data
@@ -134,7 +134,7 @@ class TestExfiltrationChannelEvents:
         started = asyncio.Event()
         release = asyncio.Event()
 
-        async def handle_console_event(_: object) -> None:
+        async def handle_console_event(_: object, __: int) -> None:
             started.set()
             await release.wait()
 
@@ -163,7 +163,7 @@ class TestExfiltrationChannelEvents:
         message.args = [marker, payload]
         message.text = "[EXFIL] JSHandle@object"
 
-        await channel._handle_console_event_async(message)
+        await channel._handle_console_event_async(message, 0)
 
         on_event.assert_called_once()
         assert on_event.call_args.args[0][0].params == event_data
@@ -180,14 +180,15 @@ class TestExfiltrationChannelEvents:
         message.text = f"[EXFIL] {json.dumps(event_data)}"
 
         channel._handle_binding_event({"page": page}, event_data)
-        await channel._handle_console_event_async(message)
+        await channel._handle_console_event_async(message, 1)
         await channel._handle_runtime_console_event_async(
             {
                 "args": [
                     {"type": "string", "value": "[EXFIL]"},
                     {"type": "string", "value": json.dumps(event_data)},
                 ]
-            }
+            },
+            2,
         )
 
         on_event.assert_called_once()

tests/unit/services/test_browser_recording.py

@@ -68,12 +68,15 @@ def make_console_event(
 
     params = {**default_params, **params}
 
+    # params.timestamp is the client clock (Date.now(), ms); the outer event
+    # timestamp is the server clock (time.time(), seconds). Mirror production so
+    # the Wait machine's client/server offset is ~0 for zero-skew fixtures.
     return ExfiltratedConsoleEvent(
         kind="exfiltrated-event",
         source="console",
         event_name="user_interaction",
         params=params,
-        timestamp=timestamp,
+        timestamp=timestamp / 1000.0,
     )
 
 
@@ -609,6 +612,56 @@ def test_collapse_consecutive_waits_merges_durations() -> None:
     assert wait_action.duration_ms == 11000
 
 
+def make_skewed_console_event(
+    event_type: str,
+    target: dict[str, t.Any],
+    client_ms: float,
+    server_skew_seconds: float,
+) -> ExfiltratedConsoleEvent:
+    """A console event whose server clock is offset from the client clock."""
+    params: dict[str, t.Any] = {
+        "type": event_type,
+        "target": target,
+        "timestamp": client_ms,
+        "url": "https://example.com",
+        "activeElement": {"tagName": "BUTTON"},
+        "window": {"height": 800, "width": 1200, "scrollX": 0, "scrollY": 0},
+        "mousePosition": {"xp": 0.5, "yp": 0.5},
+    }
+    return ExfiltratedConsoleEvent(
+        kind="exfiltrated-event",
+        source="console",
+        event_name="user_interaction",
+        params=params,
+        timestamp=client_ms / 1000.0 + server_skew_seconds,
+    )
+
+
+def test_wait_offset_projection_cancels_client_server_clock_skew() -> None:
+    # Server clock runs 60s ahead of the client clock. The Wait machine must
+    # project the server-stamped CDP activity back into the client clock so the
+    # busy span is measured correctly; otherwise the activity falls outside the
+    # client-clock gap and the (real) wait is wrongly suppressed.
+    skew = 60.0
+    target = dict(id="button-1", skyId="sky-123", tagName="BUTTON", text=["Click me"])
+
+    events = [
+        make_skewed_console_event("click", target, client_ms=1000.0, server_skew_seconds=skew),
+        # real activity at client 4s/7s -> server-stamped 64s/67s
+        make_cdp_event("net:activity", timestamp_seconds=4.0 + skew, params={"count": 12}),
+        make_cdp_event("net:activity", timestamp_seconds=7.0 + skew, params={"count": 3}),
+        make_skewed_console_event("focus", target, client_ms=9000.0, server_skew_seconds=skew),
+    ]
+
+    processor = Processor(PBS_ID, ORG_ID, WP_ID)
+    actions = processor.events_to_actions(events)
+
+    assert [action.kind for action in actions] == [ActionKind.CLICK, ActionKind.WAIT]
+    wait_action = actions[1]
+    assert isinstance(wait_action, ActionWait)
+    assert wait_action.duration_ms == 6000
+
+
 def test_wait_ignores_activity_outside_the_idle_gap() -> None:
     target = dict(id="button-1", skyId="sky-123", tagName="BUTTON", text=["Click me"])
 

tests/unit/test_browser_recording_interpretation.py

@@ -6,6 +6,7 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 from skyvern.client.types.workflow_definition_yaml_blocks_item import WorkflowDefinitionYamlBlocksItem_Wait
+from skyvern.forge import app
 from skyvern.forge.sdk.routes.streaming.channels.exfiltration import ExfiltratedEvent as StreamingExfiltratedEvent
 from skyvern.forge.sdk.routes.streaming.channels.exfiltration import (
     ExfiltratedEventSource as StreamingExfiltratedEventSource,
@@ -22,6 +23,7 @@ from skyvern.services.browser_recording.types import (
     ExfiltratedConsoleEvent,
     Mouse,
     RecordingDraftStep,
+    RecordingDraftStepStatus,
 )
 
 ORG_ID = "org_123"
@@ -196,20 +198,27 @@ async def test_processor_process_uses_draft_steps_without_compressed_chunks() ->
     assert parameters == []
 
 
-def _click_streaming_event(*, timestamp: float = 1234.0) -> StreamingExfiltratedEvent:
+def _click_streaming_event(
+    *,
+    timestamp: float = 1234.0,
+    capture_seq: int = -1,
+    sky_id: str = "sky-1",
+    target_id: str = "submit",
+) -> StreamingExfiltratedEvent:
     return StreamingExfiltratedEvent(
         event_name="user_interaction",
         source=StreamingExfiltratedEventSource.CONSOLE,
         timestamp=timestamp,
+        capture_seq=capture_seq,
         params={
             "type": "click",
             "url": "https://example.com",
             "timestamp": timestamp,
             "target": {
                 "tagName": "BUTTON",
-                "id": "submit",
+                "id": target_id,
                 "text": ["Submit"],
-                "skyId": "sky-1",
+                "skyId": sky_id,
             },
             "mousePosition": {"xp": 0.5, "yp": 0.5},
             "activeElement": {"tagName": "BUTTON"},
@@ -223,6 +232,30 @@ def _click_streaming_event(*, timestamp: float = 1234.0) -> StreamingExfiltrated
     )
 
 
+@pytest.mark.asyncio
+async def test_ingest_events_sorts_unprocessed_tail_by_capture_seq() -> None:
+    session = RecordingInterpretationSession(
+        browser_session_id=PBS_ID,
+        organization_id=ORG_ID,
+        workflow_permanent_id=WP_ID,
+        on_update=lambda _: None,
+        debounce_seconds=60,
+    )
+
+    # Events arrive out of capture order (later capture_seq first), as can happen
+    # when a console event's async materialization completes after a later event.
+    session.ingest_events(
+        [
+            _click_streaming_event(timestamp=1003.0, capture_seq=3, sky_id="sky-c", target_id="c"),
+            _click_streaming_event(timestamp=1001.0, capture_seq=1, sky_id="sky-a", target_id="a"),
+            _click_streaming_event(timestamp=1002.0, capture_seq=2, sky_id="sky-b", target_id="b"),
+        ]
+    )
+
+    assert [event.capture_seq for event in session.events] == [1, 2, 3]
+    session.cancel()
+
+
 @pytest.mark.asyncio
 async def test_recording_interpretation_session_reschedules_debounce_on_new_events(
     monkeypatch: pytest.MonkeyPatch,
@@ -352,6 +385,43 @@ async def test_recording_interpretation_session_advances_past_unhandled_actions(
     assert len(session.steps) == 1
 
 
+@pytest.mark.asyncio
+async def test_enrichment_calls_are_capped_by_semaphore(monkeypatch: pytest.MonkeyPatch) -> None:
+    in_flight = 0
+    max_in_flight = 0
+
+    async def fake_llm(*args: object, **kwargs: object) -> dict[str, object]:
+        nonlocal in_flight, max_in_flight
+        in_flight += 1
+        max_in_flight = max(max_in_flight, in_flight)
+        await asyncio.sleep(0.02)
+        in_flight -= 1
+        return {"block_label": "click_x", "title": "Click X", "prompt": "Click X."}
+
+    monkeypatch.setattr(app, "LLM_API_HANDLER", fake_llm)
+
+    session = RecordingInterpretationSession(
+        browser_session_id=PBS_ID,
+        organization_id=ORG_ID,
+        workflow_permanent_id=WP_ID,
+        on_update=lambda _: None,
+        debounce_seconds=0.01,
+        max_wait_seconds=0.05,
+    )
+    session._enrichment_semaphore = asyncio.Semaphore(2)
+
+    events = [
+        _click_streaming_event(timestamp=1000.0 + i, capture_seq=i, sky_id=f"sky-{i}", target_id=f"t{i}")
+        for i in range(8)
+    ]
+    session.ingest_events(events)
+    steps = await session.flush()
+
+    assert len(steps) == 8
+    assert all(step.status == RecordingDraftStepStatus.READY for step in steps)
+    assert max_in_flight == 2
+
+
 def test_emit_snapshot_replays_current_revision_without_incrementing() -> None:
     updates: list[int] = []