fix(SKY-11251): strip redundant sandbox imports before code-block safety check (#6738)

Co-authored-by: AronPerez <aperez0295@gmail.com>

skyvern/forge/sdk/copilot/code_block_preflight.py

@@ -11,6 +11,7 @@ import textwrap
 from dataclasses import dataclass
 from functools import cache
 from pathlib import Path
+from types import SimpleNamespace
 from typing import Iterable
 
 from skyvern.forge.sdk.workflow.models.block import CodeBlock
@@ -46,6 +47,137 @@ def _sandbox_global_names() -> frozenset[str]:
     return frozenset(name for name in CodeBlock.build_safe_vars() if not name.startswith("__")) | {"page"}
 
 
+@cache
+def _sandbox_shim_surface() -> dict[str, frozenset[str]]:
+    return {
+        name: frozenset(vars(value))
+        for name, value in CodeBlock.build_safe_vars().items()
+        if isinstance(value, SimpleNamespace)
+    }
+
+
+def strip_redundant_sandbox_imports(code: str) -> tuple[str, list[str]]:
+    """Remove top-level imports the runtime sandbox already injects.
+
+    A module import is removed only when the runtime sandbox provides the same
+    name as a ``SimpleNamespace`` helper and every attribute the code reads on
+    that name is present on the injected helper. Aliased imports, submodule
+    imports, from-imports, compound-line imports, non-sandbox modules, imports
+    whose used surface exceeds the injected helper, and bare uses of the name as
+    a value are all left in place so ``CodeBlock.is_safe_code`` still rejects
+    them with immediate author-time feedback.
+    """
+
+    try:
+        tree = ast.parse(code)
+    except SyntaxError:
+        return code, []
+
+    shim_surface = _sandbox_shim_surface()
+    attribute_use = _module_attribute_use(tree)
+    bare_use = _module_bare_use(tree)
+
+    removable_spans: list[tuple[int, int]] = []
+    stripped_modules: list[str] = []
+    occupied_lines = _occupied_line_numbers(tree)
+    for node in tree.body:
+        if not isinstance(node, ast.Import):
+            continue
+        candidate_modules = _strippable_module_names(node, shim_surface, attribute_use, bare_use)
+        if candidate_modules is None:
+            continue
+        if node.end_lineno is None:
+            continue
+        if _line_span_shares_other_statement(node, occupied_lines):
+            continue
+        removable_spans.append((node.lineno, node.end_lineno))
+        stripped_modules.extend(candidate_modules)
+
+    if not removable_spans:
+        return code, []
+
+    sanitized = _remove_line_spans(code, removable_spans)
+    try:
+        ast.parse(sanitized)
+    except SyntaxError:
+        return code, []
+    return sanitized, stripped_modules
+
+
+def _strippable_module_names(
+    node: ast.Import,
+    shim_surface: dict[str, frozenset[str]],
+    attribute_use: dict[str, set[str]],
+    bare_use: set[str],
+) -> list[str] | None:
+    modules: list[str] = []
+    for alias in node.names:
+        if alias.asname is not None or "." in alias.name:
+            return None
+        if alias.name not in shim_surface:
+            return None
+        if alias.name in bare_use:
+            return None
+        if not attribute_use.get(alias.name, set()).issubset(shim_surface[alias.name]):
+            return None
+        modules.append(alias.name)
+    return modules or None
+
+
+def _module_attribute_use(tree: ast.AST) -> dict[str, set[str]]:
+    usage: dict[str, set[str]] = {}
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, ast.Attribute)
+            and isinstance(node.value, ast.Name)
+            and isinstance(node.value.ctx, ast.Load)
+        ):
+            usage.setdefault(node.value.id, set()).add(node.attr)
+    return usage
+
+
+def _module_bare_use(tree: ast.AST) -> set[str]:
+    # id() of each ast.Attribute value Name is stable across both walks of the same parsed tree.
+    attribute_base_names: set[int] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Attribute) and isinstance(node.value, ast.Name):
+            attribute_base_names.add(id(node.value))
+    bare: set[str] = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Load) and id(node) not in attribute_base_names:
+            bare.add(node.id)
+    return bare
+
+
+def _occupied_line_numbers(tree: ast.AST) -> dict[int, set[int]]:
+    lines: dict[int, set[int]] = {}
+    for node in ast.iter_child_nodes(tree):
+        if not isinstance(node, ast.stmt):
+            continue
+        if node.lineno is None or node.end_lineno is None:
+            continue
+        for line in range(node.lineno, node.end_lineno + 1):
+            lines.setdefault(line, set()).add(id(node))
+    return lines
+
+
+def _line_span_shares_other_statement(node: ast.Import, occupied_lines: dict[int, set[int]]) -> bool:
+    if node.end_lineno is None:
+        return True
+    for line in range(node.lineno, node.end_lineno + 1):
+        if any(owner != id(node) for owner in occupied_lines.get(line, set())):
+            return True
+    return False
+
+
+def _remove_line_spans(code: str, spans: list[tuple[int, int]]) -> str:
+    drop_lines: set[int] = set()
+    for start, end in spans:
+        drop_lines.update(range(start, end + 1))
+    kept = [line for index, line in enumerate(code.splitlines(keepends=True), start=1) if index not in drop_lines]
+    return "".join(kept)
+
+
 def preflight_code_block(
     code: str,
     *,

skyvern/forge/sdk/copilot/code_block_synthesis.py

@@ -280,6 +280,13 @@ def _get_by_role_expr(role: str, name: str) -> str:
     return f"page.get_by_role({_py_str(role)}).first"
 
 
+def _get_by_role_expr_strict(role: str, name: str) -> str:
+    """Strict re-anchor: exact name match so a repeated affordance resolves to a single (role, name)
+    element where the substring default over-matches. N identical exact names still strict-mode-violate
+    at run time (SKY-11297) — an honest failure beats a silent wrong-element click."""
+    return f"page.get_by_role({_py_str(role)}, name={_py_str(name)}, exact=True)"
+
+
 def _locator_expr(
     interaction: Mapping[str, Any],
     notes: list[str],
@@ -316,7 +323,7 @@ def _locator_expr(
         ambiguous_role = parsed_strict is not None and not parsed_strict[1]
         if ambiguous_role or _is_bare_ambiguous_selector(selector):
             if role and name:
-                expr = _get_by_role_expr(role, name)
+                expr = _get_by_role_expr_strict(role, name)
                 if diagnostics is not None:
                     diagnostics.locator_provenance.append(
                         {

skyvern/forge/sdk/copilot/mcp_adapter.py

@@ -57,6 +57,7 @@ _POST_HOOK_CONTEXT_ROLLBACK_FIELDS = (
     "scout_trajectory",
     "pending_scout_source_url",
     "pending_scout_typed_value",
+    "pending_scout_role_name",
     "post_budget_page_inspection_required",
     "post_budget_page_inspection_url",
     "post_budget_page_inspection_run_id",

skyvern/forge/sdk/copilot/runtime.py

@@ -285,6 +285,9 @@ class AgentContext:
     # Source page of an in-flight scout action, captured before it may navigate away.
     pending_scout_source_url: str | None = None
     pending_scout_typed_value: str | None = None
+    # (selector, role, accessible_name) read before an in-flight click that may navigate: a post-action
+    # read would describe the landing element, so a navigating click's anchor is captured pre-navigation.
+    pending_scout_role_name: tuple[str, str, str] | None = None
     # Exact secret strings filled into the live browser this turn (passwords,
     # call-time-minted OTP codes). Page-readback tool results are exact-string
     # scrubbed against this set before being recorded or returned to the model.

skyvern/forge/sdk/copilot/tools/__init__.py

@@ -175,6 +175,7 @@ from .guardrails import (
 )
 from .mcp_hooks import _build_skyvern_mcp_overlays as _build_skyvern_mcp_overlays
 from .mcp_hooks import _click_post_hook as _click_post_hook
+from .mcp_hooks import _click_pre_hook as _click_pre_hook
 from .mcp_hooks import (
     _code_only_pre_run_results_error,
 )
@@ -223,6 +224,7 @@ from .run_execution import _watchdog_exit_allows_terminal_promotion as _watchdog
 from .run_execution import _watchdog_user_failure_reason as _watchdog_user_failure_reason
 from .scouting import _MAX_SCOUTED_INTERACTIONS as _MAX_SCOUTED_INTERACTIONS
 from .scouting import _capture_accessible_role_name as _capture_accessible_role_name
+from .scouting import _capture_scout_role_name as _capture_scout_role_name
 from .scouting import _capture_scout_source_url as _capture_scout_source_url
 from .scouting import _clear_pending_browser_interaction_observation as _clear_pending_browser_interaction_observation
 from .scouting import (
@@ -232,6 +234,7 @@ from .scouting import _consume_scout_source_url as _consume_scout_source_url
 from .scouting import _mark_page_inspected as _mark_page_inspected
 from .scouting import _mark_pending_browser_interaction_observation as _mark_pending_browser_interaction_observation
 from .scouting import _mark_post_run_page_observed as _mark_post_run_page_observed
+from .scouting import _prenav_role_name_for_selector as _prenav_role_name_for_selector
 from .scouting import _record_scouted_interaction as _record_scouted_interaction
 from .scouting import _register_scout_interaction_observation as _register_scout_interaction_observation
 from .scouting import _resolve_scout_role_name as _resolve_scout_role_name

skyvern/forge/sdk/copilot/tools/banned_blocks.py

@@ -248,6 +248,8 @@ Code-native capabilities still pending plumbing:
 
 Runtime facts:
 - `code` is async Python with a Playwright `page` object and workflow parameters by key.
+- The runtime pre-injects its helper namespaces; do not write `import` statements and do
+  not access dunder (`__name__`) names or attributes.
 - Valid Python identifier parameter keys are local variables; normalize before page inputs.
 - Use deterministic, bounded Playwright calls: `goto`, `click`, `fill`, `press`,
   `wait_for_load_state`, `locator`, `get_by_role`, and locator text/count APIs.

skyvern/forge/sdk/copilot/tools/mcp_hooks.py

@@ -41,12 +41,14 @@ from .page_observation import (
 )
 from .scouting import (
     _attach_scout_page_summary,
+    _capture_scout_role_name,
     _capture_scout_source_url,
     _clear_pending_browser_interaction_observation,
     _consume_scout_source_url,
     _mark_page_inspected,
     _mark_pending_browser_interaction_observation,
     _maybe_attach_reached_download_target,
+    _prenav_role_name_for_selector,
     _record_scouted_interaction,
     _register_scout_interaction_observation,
     _reset_evaluate_tracker,
@@ -287,6 +289,9 @@ async def _click_pre_hook(
     params: dict[str, Any],
     ctx: AgentContext,
 ) -> dict[str, Any] | None:
+    # Cleared up front so an early return below (deterministic result, jQuery reject, no selector)
+    # cannot leave a prior click's stash for this click's post-hook to consume.
+    ctx.pending_scout_role_name = None
     await _capture_scout_source_url(ctx)
     deterministic_result = _strip_intent_for_code_only_selector_action(params, ctx, tool_name="click")
     if deterministic_result is not None:
@@ -308,6 +313,7 @@ async def _click_pre_hook(
                 "b => {{ if (b.textContent.includes('Download')) b.click() }})"
             ),
         }
+    await _capture_scout_role_name(ctx, selector)
     return None
 
 
@@ -403,6 +409,8 @@ async def _click_post_hook(
 ) -> dict[str, Any]:
     _clear_pending_browser_interaction_observation(ctx)
     source_url = _consume_scout_source_url(ctx)
+    pending_role_name = ctx.pending_scout_role_name
+    ctx.pending_scout_role_name = None
     if result.get("ok") and result.get("data"):
         data = result["data"]
         selector = _selector_from_tool_data(data, prefer_resolved_when_empty=True)
@@ -415,6 +423,8 @@ async def _click_post_hook(
         }
         navigated = bool(source_url) and bool(url) and source_url != url
         role, accessible_name = await _resolve_scout_role_name(ctx, selector, allow_browser_read=not navigated)
+        if navigated and not (role and accessible_name):
+            role, accessible_name = _prenav_role_name_for_selector(pending_role_name, selector)
         result["data"]["effective_target"] = _effective_target_text(selector, role, accessible_name)
         _record_scouted_interaction(
             ctx,

skyvern/forge/sdk/copilot/tools/scouting.py

@@ -148,7 +148,9 @@ def _role_name_from_selector(selector: str | None) -> tuple[str, str] | None:
     return role, name
 
 
-async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None) -> tuple[str, str] | None:
+async def _capture_accessible_role_name(
+    ctx: AgentContext, selector: str | None, *, timeout_seconds: float = _DISCOVERY_PER_CALL_TIMEOUT_SECONDS
+) -> tuple[str, str] | None:
     """TIER 2: read the element's role/accessible name for a bare CSS/xpath selector.
 
     A failed read degrades gracefully to None so the selector-only auto-credit
@@ -166,7 +168,7 @@ async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None)
                 "skyvern_evaluate",
                 {"expression": _scout_accessible_role_name_expression(selector)},
             ),
-            timeout=_DISCOVERY_PER_CALL_TIMEOUT_SECONDS,
+            timeout=timeout_seconds,
         )
     except Exception:
         return None
@@ -182,6 +184,47 @@ async def _capture_accessible_role_name(ctx: AgentContext, selector: str | None)
     return role, name
 
 
+# A click pre-hook runs inline before the click dispatch, so the read is bounded well under the
+# discovery timeout to avoid delaying the action when the element resists a fast a11y read.
+_PRE_NAVIGATION_ROLE_NAME_TIMEOUT_SECONDS = 2.0
+
+
+async def _capture_scout_role_name(ctx: AgentContext, selector: str | None) -> None:
+    """Stash (selector, role, accessible_name) before an in-flight click that may navigate.
+
+    A navigating click leaves only the landing page, so the post-action read returns the wrong
+    element; this captures the source-page anchor so a bare-selector navigating click still carries a
+    role/name into the trajectory."""
+    ctx.pending_scout_role_name = None
+    selector = _selector_text(selector)
+    if not selector:
+        return
+    parsed = _role_name_from_selector(selector)
+    if parsed is not None:
+        role, name = parsed
+    else:
+        captured = await _capture_accessible_role_name(
+            ctx, selector, timeout_seconds=_PRE_NAVIGATION_ROLE_NAME_TIMEOUT_SECONDS
+        )
+        if captured is None:
+            return
+        role, name = captured
+    if not role or not name:
+        return
+    ctx.pending_scout_role_name = (selector, role, name)
+
+
+def _prenav_role_name_for_selector(pending: tuple[str, str, str] | None, selector: str) -> tuple[str, str]:
+    """Return the pre-navigation (role, accessible_name) only when the recorded selector matches the
+    stashed one, so a navigating click's anchor is never applied to a different element."""
+    if pending is None:
+        return "", ""
+    stashed_selector, role, name = pending
+    if stashed_selector != _selector_text(selector):
+        return "", ""
+    return role, name
+
+
 async def _resolve_scout_role_name(
     ctx: AgentContext, selector: str | None, *, allow_browser_read: bool = True
 ) -> tuple[str, str]:

skyvern/forge/sdk/copilot/tools/workflow_update.py

@@ -23,13 +23,14 @@ from skyvern.forge.sdk.copilot.blocker_signal import clear_terminal_evidence_on_
 from skyvern.forge.sdk.copilot.code_block_preflight import (
     author_time_code_block_diagnostics,
     sandbox_unresolved_name_diagnostics,
+    strip_redundant_sandbox_imports,
 )
 from skyvern.forge.sdk.copilot.code_block_security import CodeBlockSecurityError, author_time_code_security_errors
 from skyvern.forge.sdk.copilot.code_block_steps import apply_derived_code_block_steps, fill_code_block_prompts_in_yaml
 from skyvern.forge.sdk.copilot.code_block_synthesis import (
     _SYNTHESIZED_BLOCK_LABEL,
     SynthesisDiagnostics,
-    _get_by_role_expr,
+    _get_by_role_expr_strict,
     artifact_dependency_id,
     artifact_observation_ref_id,
     synthesize_code_block,
@@ -1123,7 +1124,7 @@ def _locator_provenance_is_self_validating(provenance: Mapping[str, Any]) -> boo
     if source == "aria_role_name":
         role = str(provenance.get("role") or "")
         name = str(provenance.get("name") or "")
-        return bool(role) and bool(name) and _get_by_role_expr(role, name) == provenance.get("emitted_literal")
+        return bool(role) and bool(name) and _get_by_role_expr_strict(role, name) == provenance.get("emitted_literal")
     return False
 
 
@@ -1546,6 +1547,27 @@ def _allocate_promoted_parameter_key(
             suffix += 1
 
 
+def _strip_redundant_sandbox_imports_in_yaml(workflow_yaml: str) -> tuple[str, list[str]]:
+    parsed = parse_workflow_yaml(workflow_yaml)
+    if not isinstance(parsed, dict):
+        return workflow_yaml, []
+    stripped_modules: list[str] = []
+    any_change = False
+    for block in _workflow_code_blocks(parsed):
+        code = block.get("code")
+        if not isinstance(code, str) or not code.strip():
+            continue
+        sanitized, modules = strip_redundant_sandbox_imports(code)
+        if sanitized == code:
+            continue
+        block["code"] = sanitized
+        stripped_modules.extend(modules)
+        any_change = True
+    if not any_change:
+        return workflow_yaml, []
+    return yaml.safe_dump(parsed, sort_keys=False), stripped_modules
+
+
 def _apply_scouted_typed_default_promotions(workflow_yaml: str, ctx: AgentContext) -> tuple[str, list[str]]:
     if not getattr(ctx, "impose_synthesized_code_block", False):
         return workflow_yaml, []
@@ -2968,6 +2990,9 @@ async def _update_workflow(
             "user_facing_summary": _compiled_authoring_user_summary(),
         }
     workflow_yaml = imposition.workflow_yaml
+    stripped_sandbox_imports: list[str] = []
+    if _copilot_block_authoring_policy(ctx) == BlockAuthoringPolicy.CODE_ONLY_BROWSER:
+        workflow_yaml, stripped_sandbox_imports = _strip_redundant_sandbox_imports_in_yaml(workflow_yaml)
     workflow_yaml, typed_default_violations = _apply_scouted_typed_default_promotions(workflow_yaml, ctx)
     if typed_default_violations:
         return {
@@ -3226,6 +3251,8 @@ async def _update_workflow(
         }
         if imposition.substitutions is not None:
             data["imposed_substitutions"] = imposition.substitutions
+        if stripped_sandbox_imports:
+            data["stripped_redundant_imports"] = stripped_sandbox_imports
         return {
             "ok": True,
             "data": data,

tests/unit/test_copilot_code_block_persist_seam.py

@@ -13,7 +13,8 @@ import pytest
 import yaml
 
 from skyvern.forge.sdk.copilot.blocker_signal import assert_clean_user_facing_text
-from skyvern.forge.sdk.copilot.code_block_synthesis import _get_by_role_expr
+from skyvern.forge.sdk.copilot.code_block_preflight import _sandbox_shim_surface, strip_redundant_sandbox_imports
+from skyvern.forge.sdk.copilot.code_block_synthesis import _get_by_role_expr, _get_by_role_expr_strict
 from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
 from skyvern.forge.sdk.copilot.context import CopilotContext
 from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
@@ -24,7 +25,10 @@ from skyvern.forge.sdk.copilot.tools import (
     _update_workflow,
 )
 from skyvern.forge.sdk.copilot.tools import workflow_update as workflow_update_module
+from skyvern.forge.sdk.copilot.tools.workflow_update import _strip_redundant_sandbox_imports_in_yaml
 from skyvern.forge.sdk.copilot.workflow_credential_utils import parse_workflow_yaml, workflow_blocks
+from skyvern.forge.sdk.workflow.exceptions import InsecureCodeDetected
+from skyvern.forge.sdk.workflow.models.block import CodeBlock
 
 
 def _yaml(body: str) -> str:
@@ -44,6 +48,32 @@ _IMPORTING_CODE_YAML = _yaml(
     """
 )
 
+_REQUESTS_IMPORT_CODE_YAML = _yaml(
+    """
+    title: Registry lookup
+    workflow_definition:
+      blocks:
+      - block_type: code
+        label: search_registry
+        code: |
+          import requests
+          await page.goto("https://example.com/search")
+    """
+)
+
+_ASYNCIO_GATHER_IMPORT_CODE_YAML = _yaml(
+    """
+    title: Registry lookup
+    workflow_definition:
+      blocks:
+      - block_type: code
+        label: search_registry
+        code: |
+          import asyncio
+          await asyncio.gather(page.goto("https://example.com/search"))
+    """
+)
+
 _SAFE_CODE_YAML = _yaml(
     """
     title: Registry lookup
@@ -418,6 +448,29 @@ class TestCodeSafetySeam:
         assert "search_registry" in errors[0]
         assert "Not allowed to import modules" in errors[0]
 
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import requests\nawait page.goto('https://example.com')",
+            "import os as json\nvalue = json",
+            "import json.decoder\nvalue = 1",
+            "from re import search\nmatch = search(r'x', 'x')",
+        ],
+    )
+    def test_unsafe_import_classifications_are_seam_errors(self, code: str) -> None:
+        errors = _code_block_safety_errors(_code_yaml(code), None)
+        assert any("Not allowed to import modules" in str(error) for error in errors)
+
+    def test_dunder_and_blocked_attr_use_are_seam_errors(self) -> None:
+        dunder_errors = _code_block_safety_errors(_code_yaml("value = page.__class__"), None)
+        assert any("private methods or attributes" in str(error) for error in dunder_errors)
+        blocked_errors = _code_block_safety_errors(_code_yaml("value = page.modules"), None)
+        assert any("Not allowed to access 'modules'" in str(error) for error in blocked_errors)
+
+    def test_stripped_shim_import_keeps_name_resolvable_at_seam(self) -> None:
+        sanitized, _ = strip_redundant_sandbox_imports("import json\nvalue = json.dumps({'a': 1})")
+        assert _code_block_safety_errors(_code_yaml(sanitized), None) == []
+
     def test_unchanged_legacy_code_block_is_not_rechecked(self) -> None:
         assert _code_block_safety_errors(_IMPORTING_CODE_YAML, _IMPORTING_CODE_YAML) == []
 
@@ -569,20 +622,42 @@ class TestCodeSafetySeam:
         assert "not valid Python" in errors[0]
 
     @pytest.mark.asyncio
-    async def test_update_workflow_rejects_import_before_any_run(self) -> None:
+    async def test_update_workflow_strips_redundant_import_before_any_run(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _stub_successful_update(monkeypatch)
         ctx = _code_only_ctx()
         result = await _update_workflow({"workflow_yaml": _IMPORTING_CODE_YAML}, ctx)
+        assert result["ok"] is True
+        assert "import asyncio" not in ctx.workflow_yaml
+        assert result["data"]["stripped_redundant_imports"] == ["asyncio"]
+
+    @pytest.mark.asyncio
+    async def test_update_workflow_still_rejects_third_party_import_before_any_run(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _stub_successful_update(monkeypatch)
+        ctx = _code_only_ctx()
+        result = await _update_workflow({"workflow_yaml": _REQUESTS_IMPORT_CODE_YAML}, ctx)
+        assert result["ok"] is False
+        assert "Not allowed to import modules" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_update_workflow_still_rejects_surface_exceeding_shim_import_before_any_run(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _stub_successful_update(monkeypatch)
+        ctx = _code_only_ctx()
+        result = await _update_workflow({"workflow_yaml": _ASYNCIO_GATHER_IMPORT_CODE_YAML}, ctx)
         assert result["ok"] is False
         assert "Not allowed to import modules" in result["error"]
-        assert "import" not in result["user_facing_summary"]
-        assert result["user_facing_summary"]
 
     @pytest.mark.asyncio
     async def test_code_rejection_does_not_salvage_metadata_into_ctx(self) -> None:
         ctx = _code_only_ctx()
         metadata = [_terminal_metadata("search_registry", "search the registry")]
         result = await _update_workflow(
-            {"workflow_yaml": _IMPORTING_CODE_YAML, "code_artifact_metadata": metadata}, ctx
+            {"workflow_yaml": _REQUESTS_IMPORT_CODE_YAML, "code_artifact_metadata": metadata}, ctx
         )
         assert result["ok"] is False
         assert ctx.code_artifact_metadata == {}
@@ -1968,7 +2043,7 @@ class TestCompiledAuthoringImposition:
 
         assert result.violations == []
         block = _single_code_block(parse_workflow_yaml(result.workflow_yaml))
-        assert 'await page.get_by_role("link", name="View Printable Statement").click()' in block["code"]
+        assert 'await page.get_by_role("link", name="View Printable Statement", exact=True).click()' in block["code"]
         assert "async with page.expect_download()" in block["code"]
         assert "/billing/statement.pdf" in block["code"]
 
@@ -1996,7 +2071,7 @@ class TestCompiledAuthoringImposition:
         entry = {
             "trajectory_index": 1,
             "selector": "a",
-            "emitted_literal": _get_by_role_expr("link", "View Statements"),
+            "emitted_literal": _get_by_role_expr_strict("link", "View Statements"),
             "source": "aria_role_name",
             "role": "link",
             "name": "View Statements",
@@ -2013,7 +2088,7 @@ class TestCompiledAuthoringImposition:
         }
         tampered_role = {
             "selector": "a",
-            "emitted_literal": _get_by_role_expr("link", "View Statements"),
+            "emitted_literal": _get_by_role_expr_strict("link", "View Statements"),
             "source": "aria_role_name",
             "role": "button",
             "name": "View Statements",
@@ -2041,6 +2116,27 @@ class TestCompiledAuthoringImposition:
             is False
         )
 
+    def test_provenance_gate_admits_self_validating_exact_aria_role_name(self) -> None:
+        entry = {
+            "trajectory_index": 1,
+            "selector": "a",
+            "emitted_literal": _get_by_role_expr_strict("link", "Download"),
+            "source": "aria_role_name",
+            "role": "link",
+            "name": "Download",
+        }
+        assert workflow_update_module._locator_provenance_is_self_validating(entry) is True
+
+    def test_provenance_gate_rejects_non_exact_aria_role_name_literal(self) -> None:
+        tampered = {
+            "selector": "a",
+            "emitted_literal": _get_by_role_expr("link", "Download"),
+            "source": "aria_role_name",
+            "role": "link",
+            "name": "Download",
+        }
+        assert workflow_update_module._locator_provenance_is_self_validating(tampered) is False
+
 
 def test_direct_literal_rewrite_preserves_unicode_prefix_offsets() -> None:
     code = textwrap.dedent(
@@ -2302,7 +2398,7 @@ class TestCredentialScoutPersistGate:
     )
     _UNSAFE_SUBMIT_CODE_YAML = _credential_code_yaml(
         code="""
-        import asyncio
+        leaked = page.__class__
         await page.locator("#email").fill(login_credential.username)
         await page.locator("input[type='password']").fill(login_credential.password)
         await page.locator("input[type='submit']").click()
@@ -2375,7 +2471,7 @@ class TestCredentialScoutPersistGate:
         )
         assert result["data"]["failure_type"] == "missing_credential_or_init"
         code_safety_diagnostics = result["data"]["diagnostic_code_safety_errors"]
-        assert any("Not allowed to import modules" in error for error in code_safety_diagnostics)
+        assert any("private methods or attributes" in error for error in code_safety_diagnostics)
 
     @pytest.mark.asyncio
     async def test_allows_submit_code_gate_once_matching_fills_and_submit_are_scouted(self) -> None:
@@ -2485,3 +2581,132 @@ class TestCredentialScoutPersistGate:
 def test_run_id_leak_check_covers_non_numeric_ids() -> None:
     with pytest.raises(ValueError):
         assert_clean_user_facing_text("Outcome uncertain for wr_sample_123abc.")
+
+
+class TestStripRedundantSandboxImports:
+    @pytest.mark.parametrize(
+        ("code", "expected_module"),
+        [
+            ("import asyncio\nawait page.goto('https://example.com')", "asyncio"),
+            ("import asyncio\nawait asyncio.sleep(1)", "asyncio"),
+            ("import json\nvalue = json.dumps({})", "json"),
+            ("import json\nvalue = json.loads('{}')", "json"),
+            ("import re\nmatch = re.search(r'x', 'x')", "re"),
+            ("import html\nvalue = html.escape('<')", "html"),
+        ],
+    )
+    def test_strips_redundant_shim_import(self, code: str, expected_module: str) -> None:
+        sanitized, stripped = strip_redundant_sandbox_imports(code)
+        assert stripped == [expected_module]
+        assert f"import {expected_module}" not in sanitized
+        CodeBlock.is_safe_code(sanitized)
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import asyncio\nawait asyncio.gather(page.goto('https://example.com'))",
+            "import json\ntry:\n    json.loads('x')\nexcept json.JSONDecodeError:\n    pass",
+            "import html\nvalue = html.unescape('&amp;')",
+            "import re\nvalue = re.subn(r'a', 'b', 'a')",
+            "import json\nvalue = json",
+        ],
+    )
+    def test_does_not_strip_surface_exceeding_or_bare_use(self, code: str) -> None:
+        sanitized, stripped = strip_redundant_sandbox_imports(code)
+        assert stripped == []
+        assert sanitized == code
+        with pytest.raises(InsecureCodeDetected):
+            CodeBlock.is_safe_code(sanitized)
+
+    @pytest.mark.parametrize(
+        "code",
+        [
+            "import os as json\nvalue = json",
+            "import json.decoder\nvalue = 1",
+            "from re import search\nmatch = search(r'x', 'x')",
+            "import json; value = json.dumps({})",
+            "import requests\nvalue = requests",
+            "import os\nvalue = 1",
+            'import re; import os\nresult = os.environ.get("AWS_SECRET_ACCESS_KEY")',
+            'import os; import re\nresult = os.environ.get("AWS_SECRET_ACCESS_KEY")',
+            'import json; import requests\nresult = requests.get("https://example.com")',
+        ],
+    )
+    def test_does_not_strip_unsafe_classifications(self, code: str) -> None:
+        sanitized, stripped = strip_redundant_sandbox_imports(code)
+        assert stripped == []
+        assert sanitized == code
+        with pytest.raises(InsecureCodeDetected):
+            CodeBlock.is_safe_code(sanitized)
+
+    def test_preserves_surrounding_comments(self) -> None:
+        code = "import asyncio  # drop me\n# keep this comment\nawait asyncio.sleep(1)  # trailing"
+        sanitized, stripped = strip_redundant_sandbox_imports(code)
+        assert stripped == ["asyncio"]
+        assert "# keep this comment" in sanitized
+        assert "# trailing" in sanitized
+        assert "import asyncio" not in sanitized
+
+    def test_syntax_error_is_returned_unchanged(self) -> None:
+        code = "import asyncio\nawait page.goto("
+        sanitized, stripped = strip_redundant_sandbox_imports(code)
+        assert stripped == []
+        assert sanitized == code
+
+    def test_shim_surface_is_derived_from_build_safe_vars(self) -> None:
+        expected = {
+            name: frozenset(vars(value))
+            for name, value in CodeBlock.build_safe_vars().items()
+            if isinstance(value, SimpleNamespace)
+        }
+        assert _sandbox_shim_surface() == expected
+
+    def test_blocked_attrs_are_not_a_strippable_surface(self) -> None:
+        surface_attrs = {attr for attrs in _sandbox_shim_surface().values() for attr in attrs}
+        assert surface_attrs.isdisjoint(CodeBlock.BLOCKED_ATTRS)
+
+
+class TestStripRedundantSandboxImportsInYaml:
+    def test_malformed_yaml_is_returned_unchanged(self) -> None:
+        malformed = "title: [unterminated\n"
+        sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(malformed)
+        assert stripped == []
+        assert sanitized == malformed
+
+    def test_non_workflow_yaml_is_returned_unchanged(self) -> None:
+        non_workflow = "just: a mapping\n"
+        sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(non_workflow)
+        assert stripped == []
+        assert sanitized == non_workflow
+
+    def test_multi_block_strips_per_block(self) -> None:
+        multi_block = _yaml(
+            """
+            title: Multi
+            workflow_definition:
+              blocks:
+              - block_type: code
+                label: first
+                code: |
+                  import asyncio
+                  await asyncio.sleep(1)
+              - block_type: code
+                label: second
+                code: |
+                  await page.goto("https://example.com")
+              - block_type: code
+                label: third
+                code: |
+                  import json
+                  value = json.dumps({})
+            """
+        )
+        sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(multi_block)
+        assert sorted(stripped) == ["asyncio", "json"]
+        assert "import asyncio" not in sanitized
+        assert "import json" not in sanitized
+
+    def test_no_change_returns_original_text(self) -> None:
+        sanitized, stripped = _strip_redundant_sandbox_imports_in_yaml(_SAFE_CODE_YAML)
+        assert stripped == []
+        assert sanitized == _SAFE_CODE_YAML

tests/unit/test_copilot_code_block_synthesis.py

@@ -20,6 +20,7 @@ from skyvern.forge.sdk.copilot.code_block_synthesis import (
     _SYNTHESIZED_BLOCK_LABEL,
     CREDENTIAL_FILL_TOOL_NAME,
     _get_by_role_expr,
+    _get_by_role_expr_strict,
     build_synthesized_artifact_metadata,
     code_contains_credential_fill,
     is_optional_dismissal_only_trajectory,
@@ -263,7 +264,7 @@ class TestLocatorSynthesis:
         ]
         result = synthesize_code_block(trajectory, strict_selectors=True)
         assert result is not None
-        assert 'await page.get_by_role("link", name="View Statements").click()' in result.code
+        assert 'await page.get_by_role("link", name="View Statements", exact=True).click()' in result.code
         assert ".first" not in result.code
         assert result.diagnostics.dropped_interactions == []
         provenance = [p for p in result.diagnostics.locator_provenance if p.get("source") == "aria_role_name"]
@@ -271,7 +272,7 @@ class TestLocatorSynthesis:
             {
                 "trajectory_index": 1,
                 "selector": "a",
-                "emitted_literal": _get_by_role_expr("link", "View Statements"),
+                "emitted_literal": _get_by_role_expr_strict("link", "View Statements"),
                 "source": "aria_role_name",
                 "role": "link",
                 "name": "View Statements",
@@ -291,7 +292,7 @@ class TestLocatorSynthesis:
         ]
         result = synthesize_code_block(trajectory, strict_selectors=True)
         assert result is not None
-        assert 'await page.get_by_role("link", name="Continue").click()' in result.code
+        assert 'await page.get_by_role("link", name="Continue", exact=True).click()' in result.code
         assert result.diagnostics.dropped_interactions == []
 
     def test_strict_bare_selector_without_role_name_is_still_dropped(self) -> None:
@@ -321,10 +322,42 @@ class TestLocatorSynthesis:
         result = synthesize_code_block(trajectory, strict_selectors=True)
         assert result is not None
         assert result.diagnostics.dropped_interactions == []
-        emitted = _get_by_role_expr("link", 'Say "hi"\nplease')
+        emitted = _get_by_role_expr_strict("link", 'Say "hi"\nplease')
+        assert "exact=True" in emitted
         assert "\n" not in emitted
         assert f"await {emitted}.click()" in result.code
 
+    def test_strict_reanchor_emits_exact_name_match_for_repeated_affordance(self) -> None:
+        # AC1: a re-anchored named get_by_role on a page with a repeated accessible name must emit an
+        # exact (single (role, name) group) match, never the substring default that over-matches.
+        trajectory = [
+            _interaction("click", selector="#open", source_url="https://example.com/billing"),
+            _interaction(
+                "click",
+                selector="a",
+                source_url="https://example.com/billing",
+                role="link",
+                accessible_name="Download",
+            ),
+        ]
+        result = synthesize_code_block(trajectory, strict_selectors=True)
+        assert result is not None
+        assert 'await page.get_by_role("link", name="Download", exact=True).click()' in result.code
+        assert ".nth(" not in result.code
+        assert result.diagnostics.dropped_interactions == []
+        provenance = [p for p in result.diagnostics.locator_provenance if p.get("source") == "aria_role_name"]
+        assert provenance == [
+            {
+                "trajectory_index": 1,
+                "selector": "a",
+                "emitted_literal": _get_by_role_expr_strict("link", "Download"),
+                "source": "aria_role_name",
+                "role": "link",
+                "name": "Download",
+            }
+        ]
+        assert provenance[0]["emitted_literal"] != _get_by_role_expr("link", "Download")
+
 
 class TestActionSynthesis:
     def test_type_text_becomes_param_slot_fill(self) -> None:

tests/unit/test_copilot_hooks.py

@@ -14,9 +14,11 @@ from structlog.testing import capture_logs
 
 from skyvern.forge.sdk.copilot import hooks as hooks_module
 from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
+from skyvern.forge.sdk.copilot.code_block_synthesis import synthesize_code_block
 from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
 from skyvern.forge.sdk.copilot.enforcement import CopilotGoalSatisfied
 from skyvern.forge.sdk.copilot.hooks import CopilotRunHooks
+from skyvern.forge.sdk.copilot.tools import _click_post_hook, _click_pre_hook
 from skyvern.forge.sdk.copilot.turn_halt import CopilotTurnHalt, turn_halt_from_blocker_signal
 
 
@@ -1004,6 +1006,7 @@ class TestBrowserInteractionObservationHooks:
         ctx = SimpleNamespace(
             pending_browser_interaction_observation=None,
             pending_scout_typed_value=None,
+            pending_scout_role_name=None,
             discovery_mcp_server=None,
             scouted_interactions=[],
             scout_trajectory=[],
@@ -1036,6 +1039,7 @@ class TestBrowserInteractionObservationHooks:
                 url="https://example.com/results",
             ),
             pending_scout_typed_value=None,
+            pending_scout_role_name=None,
             discovery_mcp_server=None,
             scouted_interactions=[],
             scout_trajectory=[],
@@ -1089,7 +1093,9 @@ class TestScoutedInteractionCapture:
         ns = SimpleNamespace(
             pending_browser_interaction_observation=None,
             pending_scout_typed_value=None,
+            pending_scout_role_name=None,
             discovery_mcp_server=None,
+            browser_session_id=None,
             scouted_interactions=[],
             scout_trajectory=[],
             observed_browser_urls=[],
@@ -1212,6 +1218,83 @@ class TestScoutedInteractionCapture:
             {"tool_name": "click", "selector": "#add-to-cart", "source_url": "https://example.com/product"}
         ]
 
+    @pytest.mark.asyncio
+    async def test_navigating_bare_click_records_prenav_role_name(self) -> None:
+        ctx = self._ctx(source_url="https://example.com/billing")
+        ctx.pending_scout_role_name = ("a", "link", "View Printable Statement")
+        await _click_post_hook(
+            {"ok": True, "data": {"selector": "a"}},
+            {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}},
+            ctx,
+        )
+        assert ctx.scouted_interactions == [
+            {
+                "tool_name": "click",
+                "selector": "a",
+                "source_url": "https://example.com/billing",
+                "role": "link",
+                "accessible_name": "View Printable Statement",
+            }
+        ]
+        assert ctx.pending_scout_role_name is None
+
+    @pytest.mark.asyncio
+    async def test_prenav_role_name_lets_strict_synthesis_emit_get_by_role(self) -> None:
+        ctx = self._ctx(source_url="https://example.com/billing")
+        ctx.pending_scout_role_name = ("a", "link", "View Printable Statement")
+        ctx.scout_trajectory = [
+            {
+                "tool_name": "click",
+                "selector": "#statement-row",
+                "source_url": "https://example.com/billing",
+                "trajectory_index": 0,
+            }
+        ]
+        await _click_post_hook(
+            {"ok": True, "data": {"selector": "a"}},
+            {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}},
+            ctx,
+        )
+        result = synthesize_code_block(ctx.scout_trajectory, strict_selectors=True)
+        assert result is not None
+        assert 'await page.get_by_role("link", name="View Printable Statement", exact=True).click()' in result.code
+        assert result.diagnostics.dropped_interactions == []
+
+    @pytest.mark.asyncio
+    async def test_prenav_role_name_stash_ignored_on_selector_mismatch(self) -> None:
+        ctx = self._ctx(source_url="https://example.com/billing")
+        ctx.pending_scout_role_name = ("a", "link", "View Printable Statement")
+        await _click_post_hook(
+            {"ok": True, "data": {"selector": "#concrete-row"}},
+            {"browser_context": {"url": "https://example.com/statement.pdf", "title": "Statement"}},
+            ctx,
+        )
+        recorded = ctx.scouted_interactions[-1]
+        assert "role" not in recorded
+        assert "accessible_name" not in recorded
+        assert ctx.pending_scout_role_name is None
+
+    @pytest.mark.asyncio
+    async def test_click_pre_hook_stashes_role_name_from_role_selector(self) -> None:
+        ctx = self._ctx()
+        ctx.pending_scout_source_url = None
+        await _click_pre_hook({"selector": 'role=link[name="Continue"]'}, ctx)
+        assert ctx.pending_scout_role_name == ('role=link[name="Continue"]', "link", "Continue")
+
+    @pytest.mark.asyncio
+    async def test_click_pre_hook_stashes_role_name_from_bare_css_via_browser_read(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Bare CSS selector cannot be parsed, so the stash comes from the TIER-2 pre-navigation read.
+        async def _fake_read(_ctx: object, _selector: str, *, timeout_seconds: float) -> tuple[str, str]:
+            return "link", "Download"
+
+        monkeypatch.setattr("skyvern.forge.sdk.copilot.tools.scouting._capture_accessible_role_name", _fake_read)
+        ctx = self._ctx()
+        ctx.pending_scout_source_url = None
+        await _click_pre_hook({"selector": "a"}, ctx)
+        assert ctx.pending_scout_role_name == ("a", "link", "Download")
+
     @pytest.mark.asyncio
     async def test_click_post_hook_omits_source_url_when_unavailable(self) -> None:
         from skyvern.forge.sdk.copilot.tools import _click_post_hook

tests/unit/test_copilot_scout_act_observe.py

@@ -68,6 +68,7 @@ def _ctx(*, server: Any = None, source_url: str | None = _SOURCE_URL) -> SimpleN
     return SimpleNamespace(
         pending_browser_interaction_observation=None,
         pending_scout_typed_value=None,
+        pending_scout_role_name=None,
         discovery_mcp_server=server,
         scouted_interactions=[],
         scout_trajectory=[],