From 7a525a6d2610a8de65724bf4a279267ae1e28a33 Mon Sep 17 00:00:00 2001 From: Cindy Li Date: Sun, 28 Jun 2026 05:06:17 -0400 Subject: [PATCH] fix(record): correct live-draft waits/hovers under load + cap enrichment fan-out (#6873) --- skyvern/config.py | 4 + .../routes/streaming/channels/exfiltration.py | 30 ++++++-- .../browser_recording/interpretation.py | 27 ++++++- .../browser_recording/state_machines/wait.py | 11 ++- skyvern/services/browser_recording/types.py | 4 + .../streaming/test_exfiltration_channel.py | 11 +-- tests/unit/services/test_browser_recording.py | 55 +++++++++++++- .../test_browser_recording_interpretation.py | 76 ++++++++++++++++++- 8 files changed, 199 insertions(+), 19 deletions(-) diff --git a/skyvern/config.py b/skyvern/config.py index a86b1da25..cd7013f45 100644 --- a/skyvern/config.py +++ b/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 diff --git a/skyvern/forge/sdk/routes/streaming/channels/exfiltration.py b/skyvern/forge/sdk/routes/streaming/channels/exfiltration.py index 7ad4b3efc..343648061 100644 --- a/skyvern/forge/sdk/routes/streaming/channels/exfiltration.py +++ b/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(), ), ] diff --git a/skyvern/services/browser_recording/interpretation.py b/skyvern/services/browser_recording/interpretation.py index 4893b57f0..25f322bba 100644 --- a/skyvern/services/browser_recording/interpretation.py +++ b/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, diff --git a/skyvern/services/browser_recording/state_machines/wait.py b/skyvern/services/browser_recording/state_machines/wait.py index 7a0a0896b..febf4a015 100644 --- a/skyvern/services/browser_recording/state_machines/wait.py +++ b/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) diff --git a/skyvern/services/browser_recording/types.py b/skyvern/services/browser_recording/types.py index d37c1b845..d08364752 100644 --- a/skyvern/services/browser_recording/types.py +++ b/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 diff --git a/tests/unit/forge/sdk/routes/streaming/test_exfiltration_channel.py b/tests/unit/forge/sdk/routes/streaming/test_exfiltration_channel.py index 5b49d4a90..8dd3d4073 100644 --- a/tests/unit/forge/sdk/routes/streaming/test_exfiltration_channel.py +++ b/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() diff --git a/tests/unit/services/test_browser_recording.py b/tests/unit/services/test_browser_recording.py index f27c8375f..69990b3c2 100644 --- a/tests/unit/services/test_browser_recording.py +++ b/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"]) diff --git a/tests/unit/test_browser_recording_interpretation.py b/tests/unit/test_browser_recording_interpretation.py index fb4c64b60..b2eb66b2c 100644 --- a/tests/unit/test_browser_recording_interpretation.py +++ b/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] = []