SKY-10863: credential-aware scouting + credential-typed synthesis in code-block mode (#6521)

skyvern/forge/prompts/skyvern/workflow_knowledge_base.txt

@@ -104,7 +104,7 @@ The interaction blocks differ by how specifically you describe the step — from
 
 - navigation — the general, goal-level block. State the outcome you want as a navigation_goal (e.g. "search for {{ query }} and open the first result") and the agent works out and performs the actions to get there. Use it when the step is best expressed as where you want to end up, stated as intent — never as a click-by-click script.
 - action — one specific UI action you name directly: a single click, type, or select on the current page (e.g. "click the Add to Cart button"). If you can name the exact action, prefer action over navigation.
-- login — authentication. Handles stored credentials and 2FA/TOTP. Whenever the step is logging in, this is the right block; do NOT hand-build a navigation login or avoid login to save steps.
+- login — authentication. Handles stored credentials and 2FA/TOTP. Whenever the step is logging in, this is the right block; do NOT hand-build a navigation login or avoid login to save steps. When the active block authoring policy makes the login block unavailable, bind the saved credential as a credential_id workflow parameter on the replacement block instead — never inline secret values.
 - file_download — the step's purpose is to download a file (detects when the download completes).
 - extraction — capturing structured data, validated against data_schema.
 

skyvern/forge/sdk/copilot/agent.py

@@ -157,6 +157,11 @@ Code block runtime facts:
 - Use deterministic, bounded Playwright calls such as `await page.goto(...)`,
   `await page.click(...)`, `await page.fill(...)`, `await page.press(...)`,
   `await page.wait_for_load_state(...)`, and `await page.evaluate(...)`.
+- A workflow parameter bound to a saved credential (`workflow_parameter_type:
+  credential_id`) resolves at runtime to a credential object: read
+  `<key>.username`, `<key>.password`, and `<key>.totp` (a fresh one-time code
+  generated at block start). Scout saved-credential fields with the
+  `fill_credential_field` tool; never type or embed literal secret values.
 - For extraction blocks, return precise JSON-safe structured data and the
   visible evidence text used to confirm it. Do not return only booleans for
   visible records, products, totals, confirmations, or identifiers.

skyvern/forge/sdk/copilot/build_phase.py

@@ -65,6 +65,7 @@ _BROWSER_PRIMITIVE_TOOLS: frozenset[str] = frozenset(
         "evaluate",
         "click",
         "type_text",
+        "fill_credential_field",
         "scroll",
         "select_option",
         "press_key",

skyvern/forge/sdk/copilot/code_block_synthesis.py

@@ -20,14 +20,24 @@ from skyvern.forge.sdk.copilot.composition_evidence import SCOUT_INTERACTION_EVI
 _MAX_STEPS = 20
 _INDENT = "    "
 
+CREDENTIAL_FILL_TOOL_NAME = "fill_credential_field"
+_CREDENTIAL_FIELDS = frozenset({"username", "password", "totp"})
+
 _SYNTHESIZED_BLOCK_LABEL = "scout_synthesized_browser_steps"
 
 # Names the code-block executor reserves in its exec() namespace (block.py build_safe_vars
 # plus the injected `page`). A parameter key colliding with one of these is silently dropped
 # at bind time, so the synthesized fill would stringify the builtin instead of the user value.
+# "username"/"password"/"totp"/"totp_identifier" are reserved too: CodeBlock.execute also
+# injects a bound credential's fields under those bare names, so a plain parameter named
+# `password` would resolve to the credential's secret value instead of the user input.
 _RESERVED_PARAM_NAMES = frozenset(
     {
         "page",
+        "username",
+        "password",
+        "totp",
+        "totp_identifier",
         "print",
         "len",
         "range",
@@ -184,10 +194,7 @@ def _locator_expr(
     return ""
 
 
-def _param_key(interaction: Mapping[str, Any], used: set[str]) -> str:
-    name = str(interaction.get("accessible_name") or "").strip()
-    role = str(interaction.get("role") or "").strip()
-    base = _safe_param_base(name or role or "value")
+def _unique_key(base: str, used: set[str]) -> str:
     candidate = base
     suffix = 2
     while candidate in used:
@@ -197,6 +204,17 @@ def _param_key(interaction: Mapping[str, Any], used: set[str]) -> str:
     return candidate
 
 
+def _param_key(interaction: Mapping[str, Any], used: set[str]) -> str:
+    name = str(interaction.get("accessible_name") or "").strip()
+    role = str(interaction.get("role") or "").strip()
+    return _unique_key(_safe_param_base(name or role or "value"), used)
+
+
+def _credential_param_key(interaction: Mapping[str, Any], used: set[str]) -> str:
+    name = str(interaction.get("credential_name") or "").strip()
+    return _unique_key(_safe_param_base(name or "credential"), used)
+
+
 def synthesize_code_block(trajectory: Sequence[Mapping[str, Any]]) -> SynthesizedCodeBlock | None:
     """Deterministically synthesize a code block from a scout trajectory, or None if empty."""
     if not trajectory:
@@ -206,6 +224,7 @@ def synthesize_code_block(trajectory: Sequence[Mapping[str, Any]]) -> Synthesize
     notes: list[str] = []
     parameters: list[dict[str, str]] = []
     used_param_keys: set[str] = set()
+    credential_param_keys: dict[str, str] = {}
 
     entry_url = ""
     entry_index = -1
@@ -252,6 +271,18 @@ def synthesize_code_block(trajectory: Sequence[Mapping[str, Any]]) -> Synthesize
             param_key = _param_key(interaction, used_param_keys)
             parameters.append({"key": param_key})
             lines.append(f"{_INDENT}await {locator}.fill(str({param_key}))")
+        elif tool_name == CREDENTIAL_FILL_TOOL_NAME:
+            credential_id = str(interaction.get("credential_id") or "").strip()
+            credential_field = str(interaction.get("credential_field") or "").strip()
+            if not credential_id or credential_field not in _CREDENTIAL_FIELDS:
+                notes.append("dropped a credential fill with no usable credential reference")
+                continue
+            credential_param_key = credential_param_keys.get(credential_id)
+            if credential_param_key is None:
+                credential_param_key = _credential_param_key(interaction, used_param_keys)
+                credential_param_keys[credential_id] = credential_param_key
+                parameters.append({"key": credential_param_key, "credential_id": credential_id})
+            lines.append(f"{_INDENT}await {locator}.fill({credential_param_key}.{credential_field})")
         elif tool_name == "select_option":
             value = str(interaction.get("value") or "").strip()
             if not value:
@@ -367,7 +398,8 @@ def render_synthesized_offer_text(
     trajectory: Sequence[Mapping[str, Any]] | None = None,
 ) -> str:
     """Render the offer body the copilot sees for a synthesized block (pure)."""
-    param_keys = [p.get("key", "") for p in synthesized.parameters if p.get("key")]
+    param_keys = [p.get("key", "") for p in synthesized.parameters if p.get("key") and not p.get("credential_id")]
+    credential_parameters = [p for p in synthesized.parameters if p.get("key") and p.get("credential_id")]
     parts = [
         "SYNTHESIZED CODE BLOCK (offered once). The page interactions you scouted were compiled into a "
         "deterministic Playwright snippet. Persist it VERBATIM as a `code` block labeled "
@@ -379,6 +411,16 @@ def render_synthesized_offer_text(
     ]
     if param_keys:
         parts.append("Workflow parameters referenced (bind these): " + ", ".join(param_keys) + ".")
+    if credential_parameters:
+        bindings = ", ".join(f"`{p['key']}` -> `{p['credential_id']}`" for p in credential_parameters)
+        parts.append(
+            "Credential parameters referenced: "
+            + bindings
+            + ". Bind each as a workflow parameter with `workflow_parameter_type: credential_id` and the "
+            "credential ID in `default_value`; at runtime the key resolves to a credential object whose "
+            "`.username` / `.password` / `.totp` attributes the snippet reads (`.totp` is a fresh one-time "
+            "code generated when the block starts). Never replace these attribute reads with literal values."
+        )
     if synthesized.notes:
         parts.append("Synthesis notes: " + "; ".join(synthesized.notes) + ".")
     if trajectory:

skyvern/forge/sdk/copilot/mcp_adapter.py

@@ -41,6 +41,7 @@ from skyvern.forge.sdk.copilot.runtime import (
     mcp_to_copilot,
 )
 from skyvern.forge.sdk.copilot.screenshot_utils import enqueue_screenshot_from_result
+from skyvern.forge.sdk.copilot.secret_scrub import scrub_secrets_from_structure
 
 if TYPE_CHECKING:
     from skyvern.forge.sdk.copilot.context import CopilotContext
@@ -286,7 +287,7 @@ class SkyvernOverlayMCPServer(MCPServer):
                 error=str(e),
                 exc_info=True,
             )
-            err = {"ok": False, "error": f"{tool_name} failed: {e}"}
+            err = scrub_secrets_from_structure(copilot_ctx, {"ok": False, "error": f"{tool_name} failed: {e}"})
             record_tool_step_result_for_ctx(copilot_ctx, tool_name, arguments, err)
             return _copilot_to_call_tool_result(err)
 
@@ -300,6 +301,9 @@ class SkyvernOverlayMCPServer(MCPServer):
                 raw_mcp["error"] = " ".join(text_parts) if text_parts else "Unknown MCP error"
             else:
                 raw_mcp["error"] = raw_mcp.get("error") or "Unknown MCP error"
+        # Scrub before the post hook so evidence the hooks record from raw_mcp
+        # (flow evidence, scout observations) is scrubbed too.
+        raw_mcp = scrub_secrets_from_structure(copilot_ctx, raw_mcp)
         copilot_result = mcp_to_copilot(raw_mcp)
 
         if overlay.post_hook:

skyvern/forge/sdk/copilot/output_policy.py

@@ -11,7 +11,12 @@ from skyvern.forge.sdk.copilot.output_utils import (
     looks_like_workflow_delivery_claim,
     looks_like_workflow_yaml_in_chat,
 )
-from skyvern.forge.sdk.copilot.request_policy import RAW_SECRET_PATTERNS, RequestPolicy, contains_email_password_pair
+from skyvern.forge.sdk.copilot.request_policy import (
+    RAW_SECRET_PATTERNS,
+    SECRET_KEYWORD_ASSIGNMENT_PATTERN,
+    RequestPolicy,
+    contains_email_password_pair,
+)
 from skyvern.forge.sdk.copilot.workflow_credential_utils import (
     block_credential_ids,
     credential_params,
@@ -25,6 +30,18 @@ from skyvern.forge.sdk.copilot.workflow_credential_utils import (
 WORKFLOW_PRESENT_SENTINEL = object()
 _CREDENTIAL_ID_RE = re.compile(r"\bcred_[A-Za-z0-9][A-Za-z0-9_-]*\b")
 _PLACEHOLDER_MARKERS = ("{{", "{%", "[REDACTED_SECRET]")
+# RHS of a secret-keyword assignment that references a bound value instead of carrying one:
+# a `parameters`-rooted lookup (quoted-key subscript / .get / attribute), or an attribute
+# chain ending in a credential field (`cred.password`), optionally wrapped in str(...).
+# Fully anchored — only closing punctuation may follow, so a literal appended to a
+# reference (`cred.password+"hunter2"`) or a dotted literal (a JWT) never passes.
+_SANCTIONED_SECRET_REFERENCE_RE = re.compile(
+    r"^(?:str\()?"
+    r"(?:parameters(?:\[(?:'[^']*'|\"[^\"]*\")\]|\.get\((?:'[^']*'|\"[^\"]*\")\)|(?:\.[A-Za-z_][A-Za-z0-9_]*)+)"
+    r"|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*\.(?:username|password|totp))"
+    r"[)\]\},;.'\"]*$"
+)
+_SECRET_ASSIGNMENT_RHS_RE = re.compile(r"[:=]\s*(\S+)\s*$")
 _UNVALIDATED_PROPOSAL_AFFORDANCE_RE = re.compile(
     r"\baccept\b(?=[\s\S]{0,120}\bsav(?:e|ed|ing)\b)(?=[\s\S]{0,160}\b(?:reject|discard)\b)",
     re.IGNORECASE,
@@ -369,11 +386,22 @@ def _contains_raw_secret(value: Any) -> bool:
             return True
         for pattern in RAW_SECRET_PATTERNS:
             for match in pattern.finditer(text):
-                if not any(marker in match.group(0) for marker in _PLACEHOLDER_MARKERS):
-                    return True
+                matched = match.group(0)
+                if any(marker in matched for marker in _PLACEHOLDER_MARKERS):
+                    continue
+                if pattern is SECRET_KEYWORD_ASSIGNMENT_PATTERN and _is_sanctioned_secret_reference(matched):
+                    continue
+                return True
     return False
 
 
+def _is_sanctioned_secret_reference(matched: str) -> bool:
+    rhs_match = _SECRET_ASSIGNMENT_RHS_RE.search(matched)
+    if rhs_match is None:
+        return False
+    return bool(_SANCTIONED_SECRET_REFERENCE_RE.match(rhs_match.group(1)))
+
+
 def _contains_internal_tool_instruction(user_response: str | None) -> bool:
     if not isinstance(user_response, str):
         return False

skyvern/forge/sdk/copilot/request_policy.py

@@ -94,8 +94,11 @@ _WORKFLOW_CREDENTIAL_INPUTS_UNBOUND_QUESTION = (
     "I couldn't find the required credentials for the existing workflow. "
     f"Please add them via the Credentials UI and I can try again. {_CREDENTIALS_UI_DIRECTIONS}"
 )
+SECRET_KEYWORD_ASSIGNMENT_PATTERN = re.compile(
+    r"\b(?:password|passcode|api[_ -]?key|secret|token|bearer|authorization)\s*[:=]\s*\S+", re.I
+)
 _RAW_SECRET_PATTERNS = (
-    re.compile(r"\b(?:password|passcode|api[_ -]?key|secret|token|bearer|authorization)\s*[:=]\s*\S+", re.I),
+    SECRET_KEYWORD_ASSIGNMENT_PATTERN,
     re.compile(
         r"\b(?:otp|totp|mfa|2fa|verification|auth(?:entication)? code)(?:\s+code)?\s*(?:is|[:=])?\s*\d{6,8}\b",
         re.I,

skyvern/forge/sdk/copilot/runtime.py

@@ -116,6 +116,10 @@ class ScoutedInteraction(TypedDict):
     role: NotRequired[str]
     accessible_name: NotRequired[str]
     trajectory_index: NotRequired[int]
+    # Credential fills carry references and metadata only — never secret values.
+    credential_id: NotRequired[str]
+    credential_field: NotRequired[str]
+    credential_name: NotRequired[str]
 
 
 @dataclass
@@ -242,6 +246,10 @@ class AgentContext:
     synthesized_block_offered: bool = False
     # Source page of an in-flight scout action, captured before it may navigate away.
     pending_scout_source_url: 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.
+    secret_scrub_values: list[str] = field(default_factory=list)
 
     # Set by tool gates / loop guards / tool-side error branches when a tool
     # dispatch is blocked. The finalization shim in agent.py reads this at

skyvern/forge/sdk/copilot/secret_scrub.py

@@ -0,0 +1,102 @@
+"""Exact-string scrubbing of filled credential values from tool results.
+
+A value filled in one turn persists in the page's input.value on the cross-turn
+debug session, so registered values are also kept in a session-keyed registry to
+stay scrubbed on readbacks in later turns whose per-turn context is empty.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import structlog
+
+from skyvern.forge.sdk.copilot.output_utils import is_valid_image_base64
+
+if TYPE_CHECKING:
+    from skyvern.forge.sdk.copilot.runtime import AgentContext
+
+LOG = structlog.get_logger()
+
+REDACTED_SECRET_PLACEHOLDER = "[REDACTED_SECRET]"
+
+_SESSION_SCRUB_VALUES: dict[str, list[str]] = {}
+# No deterministic per-session teardown exists, so FIFO-evict to bound worker memory.
+_MAX_SCRUB_SESSIONS = 1024
+
+
+def _session_id(ctx: AgentContext) -> str | None:
+    session_id = getattr(ctx, "browser_session_id", None)
+    return session_id if isinstance(session_id, str) and session_id else None
+
+
+def register_secret_scrub_value(ctx: AgentContext, value: str | None) -> None:
+    if not isinstance(value, str) or not value:
+        return
+    values = getattr(ctx, "secret_scrub_values", None)
+    if isinstance(values, list) and value not in values:
+        values.append(value)
+    session_id = _session_id(ctx)
+    if session_id is not None:
+        new_session = session_id not in _SESSION_SCRUB_VALUES
+        session_values = _SESSION_SCRUB_VALUES.setdefault(session_id, [])
+        if value not in session_values:
+            session_values.append(value)
+        if new_session:
+            while len(_SESSION_SCRUB_VALUES) > _MAX_SCRUB_SESSIONS:
+                _SESSION_SCRUB_VALUES.pop(next(iter(_SESSION_SCRUB_VALUES)))
+
+
+def clear_session_scrub_values(session_id: str | None) -> None:
+    if isinstance(session_id, str):
+        _SESSION_SCRUB_VALUES.pop(session_id, None)
+
+
+def _registered_scrub_values(ctx: AgentContext) -> list[str]:
+    merged: list[str] = []
+    values = getattr(ctx, "secret_scrub_values", None)
+    if isinstance(values, list):
+        merged.extend(value for value in values if isinstance(value, str) and value)
+    session_id = _session_id(ctx)
+    if session_id is not None:
+        merged.extend(value for value in _SESSION_SCRUB_VALUES.get(session_id, []) if isinstance(value, str) and value)
+    # Longest first so an overlapping shorter value never splits a longer one.
+    return sorted(set(merged), key=len, reverse=True)
+
+
+def scrub_secrets_from_text(ctx: AgentContext, text: str) -> str:
+    for value in _registered_scrub_values(ctx):
+        text = text.replace(value, REDACTED_SECRET_PLACEHOLDER)
+    return text
+
+
+def scrub_secrets_from_structure(ctx: AgentContext, obj: Any) -> Any:
+    values = _registered_scrub_values(ctx)
+    if not values:
+        return obj
+    replacements = 0
+
+    def walk(node: Any) -> Any:
+        nonlocal replacements
+        if isinstance(node, str):
+            # A short alphanumeric secret (an OTP code) can occur inside image
+            # base64 by coincidence; replacing it would corrupt the image.
+            if is_valid_image_base64(node):
+                return node
+            for value in values:
+                if value in node:
+                    replacements += node.count(value)
+                    node = node.replace(value, REDACTED_SECRET_PLACEHOLDER)
+            return node
+        if isinstance(node, dict):
+            return {walk(key): walk(item) for key, item in node.items()}
+        if isinstance(node, list):
+            return [walk(item) for item in node]
+        if isinstance(node, tuple):
+            return tuple(walk(item) for item in node)
+        return node
+
+    scrubbed = walk(obj)
+    if replacements:
+        LOG.info("Scrubbed registered secret values from a tool result", replacements=replacements)
+    return scrubbed

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

@@ -35,10 +35,12 @@ from skyvern.forge.sdk.copilot.output_utils import (
     sanitize_tool_result_for_llm,
 )
 from skyvern.forge.sdk.copilot.screenshot_utils import enqueue_screenshot_from_result
+from skyvern.forge.sdk.copilot.secret_scrub import scrub_secrets_from_structure
 from skyvern.forge.sdk.copilot.tracing_setup import copilot_span
 from skyvern.forge.sdk.routes.workflow_copilot import _process_workflow_yaml as _process_workflow_yaml
 
 from ._shared import _COMPOSITION_STRIPPED_HTML_MAX_CHARS as _COMPOSITION_STRIPPED_HTML_MAX_CHARS
+from ._shared import _CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS as _CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS
 from ._shared import _FAILED_BLOCK_STATUSES as _FAILED_BLOCK_STATUSES
 from ._shared import BLOCK_RUNNING_TOOLS as BLOCK_RUNNING_TOOLS
 from ._shared import COPILOT_FINAL_REPLY_RESERVE_SECONDS as COPILOT_FINAL_REPLY_RESERVE_SECONDS
@@ -113,6 +115,11 @@ from .composition_capture import (
 )
 from .composition_capture import _normalized_inspect_url as _normalized_inspect_url
 from .composition_capture import _same_inspect_target as _same_inspect_target
+from .credential_fill import _credential_fill_policy_error as _credential_fill_policy_error
+from .credential_fill import (
+    _fill_credential_field_impl,
+)
+from .credential_fill import _resolve_credential_fill_value as _resolve_credential_fill_value
 from .credentials import _credential_id_misbinding_error_message as _credential_id_misbinding_error_message
 from .credentials import _credential_id_misbinding_findings as _credential_id_misbinding_findings
 from .credentials import _credential_reference_validation_error as _credential_reference_validation_error
@@ -792,7 +799,7 @@ async def discover_workflow_entrypoint_tool(
     buttons, run JavaScript, or submit forms.
     """
     result = await _discover_workflow_entrypoint_impl(ctx.context, site_or_url, intent_hint)
-    return json.dumps(result)
+    return json.dumps(scrub_secrets_from_structure(ctx.context, result))
 
 
 @function_tool(name_override="inspect_page_for_composition", strict_mode=False)
@@ -827,7 +834,38 @@ async def inspect_page_for_composition_tool(
     report the observed anti-bot blocker rather than retrying the same flow.
     """
     result = await _inspect_page_for_composition_impl(ctx.context, target_url)
-    return json.dumps(result)
+    return json.dumps(scrub_secrets_from_structure(ctx.context, result))
+
+
+@function_tool(name_override="fill_credential_field", strict_mode=False)
+async def fill_credential_field_tool(
+    ctx: RunContextWrapper,
+    selector: str,
+    credential_id: str,
+    field: str,
+) -> str:
+    """Fill ONE field of a SAVED credential into the live debug browser during code-only scouting.
+
+    The secret value is resolved server-side from the stored credential and never
+    enters the conversation; the result reports only `typed_length`. Use this
+    instead of `type_text` whenever a login form field should receive a saved
+    credential's username, password, or one-time TOTP code — `type_text` cannot
+    type secrets and you never have the values.
+
+    `selector` must be a CSS selector for the exact input field (no comma-union
+    fallbacks — inspect the page first and target the proven field).
+    `credential_id` must be a credential from the request policy's
+    `resolved_credentials`. `field` is one of `username`, `password`, `totp`
+    (`totp` generates a fresh code at call time).
+
+    This tool only fills; it never clicks or submits. Each successful fill is
+    recorded as a scouted interaction, so the SYNTHESIZED CODE BLOCK will bind
+    the credential as a `credential_id` workflow parameter and reference it as
+    `<parameter_key>.username` / `.password` / `.totp` — keep that attribute
+    form when persisting code blocks.
+    """
+    result = await _fill_credential_field_impl(ctx.context, selector, credential_id, field)
+    return json.dumps(scrub_secrets_from_structure(ctx.context, result))
 
 
 NATIVE_TOOLS = [
@@ -838,4 +876,5 @@ NATIVE_TOOLS = [
     update_and_run_blocks_tool,
     discover_workflow_entrypoint_tool,
     inspect_page_for_composition_tool,
+    fill_credential_field_tool,
 ]

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

@@ -193,6 +193,8 @@ def _block_data_payload(extracted_data: Any, block_type: str | None) -> Any:
 
 BLOCK_RUNNING_TOOLS = frozenset({"run_blocks_and_collect_debug", "update_and_run_blocks"})
 
+_CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS = BLOCK_RUNNING_TOOLS | {"fill_credential_field"}
+
 
 WORKFLOW_MUTATION_TOOLS = frozenset({"update_workflow", "update_and_run_blocks"})
 

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

@@ -28,6 +28,7 @@ from skyvern.forge.sdk.workflow.models.workflow import WorkflowRun, WorkflowRunS
 from skyvern.schemas.workflows import BlockType
 
 from ._shared import (
+    _CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS,
     _DATA_PRODUCING_BLOCK_TYPES,
     _FAILED_BLOCK_STATUSES,
     BLOCK_RUNNING_TOOLS,
@@ -931,9 +932,11 @@ def _tool_loop_error(ctx: AgentContext, tool_name: str, arguments: dict[str, Any
 
     # Consecutive same-name guard: false-positives on the intended iterative
     # build (one new block per update_and_run_blocks). Block-running tools
-    # rely on the progress-aware checks below instead.
+    # rely on the progress-aware checks below instead. fill_credential_field is
+    # exempt because a username+password+TOTP form legitimately needs three
+    # consecutive calls; its failed-step guard above stays argument-aware.
     tracker = getattr(ctx, "consecutive_tool_tracker", None)
-    if isinstance(tracker, list) and tool_name not in BLOCK_RUNNING_TOOLS:
+    if isinstance(tracker, list) and tool_name not in _CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS:
         detected = detect_tool_loop(tracker, tool_name)
         if detected is not None:
             return _emit_tool_blocker_signal(

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

@@ -0,0 +1,249 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pyotp
+import structlog
+
+from skyvern.cli.core.session_manager import get_page
+from skyvern.forge import app
+from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
+from skyvern.forge.sdk.copilot.loop_detection import record_tool_step_result_for_ctx
+from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
+from skyvern.forge.sdk.copilot.runtime import AgentContext, ensure_browser_session, mcp_browser_context
+from skyvern.forge.sdk.copilot.secret_scrub import (
+    REDACTED_SECRET_PLACEHOLDER,
+    register_secret_scrub_value,
+    scrub_secrets_from_text,
+)
+from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential
+from skyvern.forge.sdk.services.credentials import parse_totp_secret
+
+from .banned_blocks import _copilot_block_authoring_policy
+from .blockers import _tool_loop_error
+from .credentials import _missing_credential_reference_tool_error
+from .guardrails import _authority_tool_error
+from .mcp_hooks import _verify_scout_type_landed
+from .scouting import (
+    _capture_scout_source_url,
+    _clear_pending_browser_interaction_observation,
+    _consume_scout_source_url,
+    _live_working_page_url,
+    _mark_pending_browser_interaction_observation,
+    _record_scouted_interaction,
+    _register_scout_interaction_observation,
+    _resolve_scout_role_name,
+)
+
+LOG = structlog.get_logger()
+
+_CREDENTIAL_FILL_FIELDS = frozenset({"username", "password", "totp"})
+_CREDENTIAL_FILL_TIMEOUT_MS = 15000
+
+
+def _scrub_secret_from_text(text: str, secret_value: str) -> str:
+    if not secret_value:
+        return text
+    return text.replace(secret_value, REDACTED_SECRET_PLACEHOLDER)
+
+
+def _credential_fill_policy_error(copilot_ctx: AgentContext, credential_id: str) -> str | None:
+    if _copilot_block_authoring_policy(copilot_ctx) != BlockAuthoringPolicy.CODE_ONLY_BROWSER:
+        return (
+            "fill_credential_field is only available in code-only browser authoring mode. "
+            "Author a `login` block bound to the credential parameter instead."
+        )
+    policy = getattr(copilot_ctx, "request_policy", None)
+    if not isinstance(policy, RequestPolicy) or not policy.allow_run_blocks:
+        return (
+            "Saved-credential scouting is not authorized for this request. "
+            "Ask the user for the required credential or clarification before filling credential fields."
+        )
+    resolved_ids = {
+        credential.credential_id
+        for credential in policy.resolved_credentials
+        if isinstance(getattr(credential, "credential_id", None), str)
+    }
+    if credential_id not in resolved_ids:
+        return (
+            f"The credential `{credential_id}` is not in the credentials resolved for this request, so it "
+            "cannot be filled into the live browser. Only credentials the user referenced (listed under "
+            "`resolved_credentials` in the request policy) may be scouted. Ask the user which saved "
+            "credential to use, or bind the credential as an untested draft parameter without running it."
+        )
+    return None
+
+
+async def _resolve_credential_fill_value(
+    copilot_ctx: AgentContext,
+    credential_id: str,
+    field: str,
+) -> tuple[str | None, str, str | None]:
+    """Resolve (secret_value, credential_name, error) for one credential field, server-side only."""
+    try:
+        db_credential = await app.DATABASE.credentials.get_credential(
+            credential_id, organization_id=copilot_ctx.organization_id
+        )
+    except Exception:
+        LOG.warning(
+            "fill_credential_field could not read the credential record",
+            credential_id=credential_id,
+            organization_id=copilot_ctx.organization_id,
+            exc_info=True,
+        )
+        return None, "", f"Could not read credential `{credential_id}`. Ask the user to verify it exists."
+    if db_credential is None:
+        return None, "", _missing_credential_reference_tool_error([credential_id])
+
+    vault_type = db_credential.vault_type or CredentialVaultType.BITWARDEN
+    credential_service = app.CREDENTIAL_VAULT_SERVICES.get(vault_type)
+    if credential_service is None:
+        return None, "", f"The credential vault for `{credential_id}` is not configured on this deployment."
+    try:
+        credential_item = await credential_service.get_credential_item(db_credential)
+    except Exception as exc:
+        LOG.warning(
+            "fill_credential_field could not fetch the credential from the vault",
+            credential_id=credential_id,
+            vault_type=str(vault_type),
+            exc_info=True,
+        )
+        return None, "", f"Could not fetch credential `{credential_id}` from the vault: {type(exc).__name__}."
+    credential = credential_item.credential
+    if not isinstance(credential, PasswordCredential):
+        return None, "", f"Credential `{credential_id}` is not a username/password credential."
+
+    if field == "username":
+        value: str | None = credential.username
+    elif field == "password":
+        value = credential.password
+        register_secret_scrub_value(copilot_ctx, value)
+    else:
+        if not credential.totp:
+            return None, "", f"Credential `{credential_id}` has no TOTP secret configured."
+        try:
+            value = pyotp.TOTP(parse_totp_secret(credential.totp)).now()
+        except Exception:
+            LOG.warning(
+                "fill_credential_field could not generate a TOTP code",
+                credential_id=credential_id,
+                exc_info=True,
+            )
+            return None, "", f"Could not generate a TOTP code for credential `{credential_id}`."
+        register_secret_scrub_value(copilot_ctx, value)
+    if not value:
+        return None, "", f"Credential `{credential_id}` has no `{field}` value."
+    return value, credential_item.name, None
+
+
+async def _fill_credential_field_impl(
+    copilot_ctx: AgentContext,
+    selector: str,
+    credential_id: str,
+    field: str,
+) -> dict[str, Any]:
+    arguments = {"selector": selector, "credential_id": credential_id, "field": field}
+
+    def finish(result: dict[str, Any]) -> dict[str, Any]:
+        record_tool_step_result_for_ctx(copilot_ctx, "fill_credential_field", arguments, result)
+        return result
+
+    loop_error = _tool_loop_error(copilot_ctx, "fill_credential_field", arguments)
+    if loop_error:
+        return {"ok": False, "error": loop_error}
+    authority_error = _authority_tool_error(copilot_ctx, "fill_credential_field")
+    if authority_error:
+        return finish({"ok": False, "error": authority_error})
+
+    selector = (selector or "").strip()
+    field = (field or "").strip().lower()
+    credential_id = (credential_id or "").strip()
+    if not selector:
+        return finish({"ok": False, "error": "fill_credential_field requires a CSS selector for the input field."})
+    if field not in _CREDENTIAL_FILL_FIELDS:
+        return finish({"ok": False, "error": "fill_credential_field `field` must be one of: username, password, totp."})
+    policy_error = _credential_fill_policy_error(copilot_ctx, credential_id)
+    if policy_error:
+        LOG.info(
+            "copilot fill_credential_field rejected tool-side",
+            credential_id=credential_id,
+            field=field,
+            organization_id=copilot_ctx.organization_id,
+        )
+        return finish({"ok": False, "error": policy_error})
+
+    value, credential_name, resolve_error = await _resolve_credential_fill_value(copilot_ctx, credential_id, field)
+    if resolve_error or value is None:
+        return finish({"ok": False, "error": resolve_error or "Could not resolve the credential value."})
+
+    session_error = await ensure_browser_session(copilot_ctx)
+    if session_error:
+        return finish(session_error)
+    await _capture_scout_source_url(copilot_ctx)
+    try:
+        async with mcp_browser_context(copilot_ctx):
+            page, _ = await get_page(session_id=copilot_ctx.browser_session_id)
+            await page.fill(selector, value, mode="direct", timeout=_CREDENTIAL_FILL_TIMEOUT_MS)
+    except Exception as exc:
+        error_text = scrub_secrets_from_text(copilot_ctx, _scrub_secret_from_text(str(exc), value))
+        LOG.info(
+            "copilot fill_credential_field fill failed",
+            selector=selector,
+            credential_id=credential_id,
+            field=field,
+            error_type=type(exc).__name__,
+        )
+        return finish(
+            {
+                "ok": False,
+                "error": (
+                    f"fill_credential_field could not fill {selector!r}: {error_text} "
+                    "Verify the selector matches a single visible, editable input on the current page "
+                    "(inspect the page again if needed), then retry."
+                ),
+            }
+        )
+
+    _clear_pending_browser_interaction_observation(copilot_ctx)
+    source_url = _consume_scout_source_url(copilot_ctx)
+    landing_failure = await _verify_scout_type_landed(copilot_ctx, selector=selector, typed_length=len(value))
+    if landing_failure is not None:
+        return finish(landing_failure)
+    url = await _live_working_page_url(copilot_ctx) or ""
+    _mark_pending_browser_interaction_observation(copilot_ctx, tool_name="fill_credential_field", url=url)
+    role, accessible_name = await _resolve_scout_role_name(copilot_ctx, selector)
+    _record_scouted_interaction(
+        copilot_ctx,
+        tool_name="fill_credential_field",
+        selector=selector,
+        source_url=source_url,
+        typed_length=len(value),
+        role=role,
+        accessible_name=accessible_name,
+        credential_id=credential_id,
+        credential_field=field,
+        credential_name=credential_name,
+    )
+    observation_step = _register_scout_interaction_observation(
+        copilot_ctx, tool_name="fill_credential_field", selector=selector, source_url=source_url, url=url
+    )
+    data: dict[str, Any] = {
+        "selector": selector,
+        "credential_id": credential_id,
+        "field": field,
+        "typed_length": len(value),
+        "url": url,
+    }
+    result: dict[str, Any] = {"ok": True, "data": data}
+    if observation_step is not None:
+        result["observation_step"] = observation_step
+        data["observation_step"] = observation_step
+    LOG.info(
+        "copilot fill_credential_field filled a saved credential field",
+        selector=selector,
+        credential_id=credential_id,
+        field=field,
+        typed_length=len(value),
+        url=url or None,
+    )
+    return finish(result)

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

@@ -105,6 +105,7 @@ async def _get_block_schema_post_hook(
                 "Use concrete selectors and text anchors found during exploration. If only intent targeting is available, inspect the page again before mutating.",
                 "Call update_and_run_blocks with a connected runnable set of real code blocks instead of validating dummy or probe blocks.",
                 "Keep block outputs JSON-safe and include visible evidence text when extracting records, products, totals, confirmations, or identifiers.",
+                "For saved credentials: bind the credential as a workflow parameter with workflow_parameter_type credential_id and the credential ID in default_value. At runtime the parameter key resolves to a credential object — read <key>.username, <key>.password, and <key>.totp (a fresh one-time code generated when the block starts). Never put literal secret values in code; scout credential fields with fill_credential_field.",
             ]
     return result
 

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

@@ -193,6 +193,9 @@ def _record_scouted_interaction(
     typed_length: int = 0,
     role: str = "",
     accessible_name: str = "",
+    credential_id: str = "",
+    credential_field: str = "",
+    credential_name: str = "",
 ) -> None:
     selector = selector.strip()
     # press_key may be page-level, so it is recorded by key even with no selector; other tools require one.
@@ -213,6 +216,12 @@ def _record_scouted_interaction(
         artifact["role"] = role
     if accessible_name:
         artifact["accessible_name"] = accessible_name
+    if credential_id:
+        artifact["credential_id"] = credential_id
+    if credential_field:
+        artifact["credential_field"] = credential_field
+    if credential_name:
+        artifact["credential_name"] = credential_name
     interactions = [
         item
         for item in ctx.scouted_interactions
@@ -220,6 +229,7 @@ def _record_scouted_interaction(
             item.get("tool_name") == artifact["tool_name"]
             and item.get("selector") == artifact.get("selector")
             and item.get("source_url") == artifact.get("source_url")
+            and item.get("credential_field") == artifact.get("credential_field")
         )
     ]
     interactions.append(artifact)

tests/unit/test_copilot_code_block_synthesis.py

@@ -648,6 +648,158 @@ class TestSynthesizedArtifactMetadata:
         assert "```json" not in text
 
 
+class TestCredentialFillSynthesis:
+    """A scouted fill_credential_field compiles into an attribute read on a
+    credential-bound parameter — references only, never values."""
+
+    def _credential_fill(self, **overrides: Any) -> dict[str, Any]:
+        fields = {
+            "selector": "#userName",
+            "source_url": "https://authenticationtest.com/simpleFormAuth/",
+            "typed_length": 24,
+            "credential_id": "cred_123",
+            "credential_field": "username",
+            "credential_name": "authtest simple",
+        }
+        fields.update(overrides)
+        return _interaction("fill_credential_field", **fields)
+
+    def test_emits_attribute_fill_and_credential_parameter(self) -> None:
+        result = synthesize_code_block([self._credential_fill()])
+        assert result is not None
+        assert 'await page.locator("#userName").fill(authtest_simple.username)' in result.code
+        assert result.parameters == [{"key": "authtest_simple", "credential_id": "cred_123"}]
+
+    def test_same_credential_shares_one_parameter(self) -> None:
+        result = synthesize_code_block(
+            [
+                self._credential_fill(),
+                self._credential_fill(selector="#passwordInput", credential_field="password", typed_length=12),
+            ]
+        )
+        assert result is not None
+        assert 'await page.locator("#userName").fill(authtest_simple.username)' in result.code
+        assert 'await page.locator("#passwordInput").fill(authtest_simple.password)' in result.code
+        assert result.parameters == [{"key": "authtest_simple", "credential_id": "cred_123"}]
+
+    def test_totp_field_reads_totp_attribute(self) -> None:
+        result = synthesize_code_block(
+            [self._credential_fill(selector="#totpCode", credential_field="totp", typed_length=6)]
+        )
+        assert result is not None
+        assert 'await page.locator("#totpCode").fill(authtest_simple.totp)' in result.code
+
+    def test_missing_credential_reference_is_dropped_with_note(self) -> None:
+        result = synthesize_code_block(
+            [
+                self._credential_fill(credential_id=""),
+                _interaction("click", selector="#next", source_url="https://example.com/login"),
+            ]
+        )
+        assert result is not None
+        assert ".fill(" not in result.code
+        assert result.parameters == []
+        assert any("credential" in note for note in result.notes)
+
+    def test_unknown_credential_field_is_dropped(self) -> None:
+        result = synthesize_code_block(
+            [
+                self._credential_fill(credential_field="cvv"),
+                _interaction("click", selector="#next", source_url="https://example.com/login"),
+            ]
+        )
+        assert result is not None
+        assert ".fill(" not in result.code
+        assert result.parameters == []
+
+    def test_param_key_defaults_when_credential_name_missing(self) -> None:
+        result = synthesize_code_block([self._credential_fill(credential_name="")])
+        assert result is not None
+        assert ".fill(credential.username)" in result.code
+        assert result.parameters == [{"key": "credential", "credential_id": "cred_123"}]
+
+    def test_credential_param_key_does_not_collide_with_typed_param(self) -> None:
+        result = synthesize_code_block(
+            [
+                _interaction(
+                    "type_text",
+                    selector="#company",
+                    source_url="https://example.com/form",
+                    typed_length=6,
+                    role="textbox",
+                    accessible_name="authtest simple",
+                ),
+                self._credential_fill(),
+            ]
+        )
+        assert result is not None
+        assert result.parameters[0] == {"key": "authtest_simple"}
+        assert result.parameters[1] == {"key": "authtest_simple_2", "credential_id": "cred_123"}
+        assert ".fill(authtest_simple_2.username)" in result.code
+
+    def test_offer_text_carries_credential_binding_contract(self) -> None:
+        trajectory = [self._credential_fill()]
+        synthesized = synthesize_code_block(trajectory)
+        assert synthesized is not None
+        text = render_synthesized_offer_text(synthesized, trajectory)
+        assert "`authtest_simple` -> `cred_123`" in text
+        assert "workflow_parameter_type: credential_id" in text
+        assert "default_value" in text
+        assert ".username` / `.password` / `.totp`" in text
+        assert "authtest_simple" not in [p.get("key") for p in synthesized.parameters if "credential_id" not in p]
+
+    def test_credential_parameters_excluded_from_plain_bind_line(self) -> None:
+        trajectory = [
+            _interaction(
+                "type_text",
+                selector="#q",
+                source_url="https://example.com/",
+                typed_length=4,
+                role="textbox",
+                accessible_name="Search",
+            ),
+            self._credential_fill(),
+        ]
+        synthesized = synthesize_code_block(trajectory)
+        assert synthesized is not None
+        text = render_synthesized_offer_text(synthesized, trajectory)
+        assert "Workflow parameters referenced (bind these): search." in text
+        assert "Credential parameters referenced" in text
+
+    def test_plain_param_never_takes_a_bare_credential_field_name(self) -> None:
+        # CodeBlock.execute injects a bound credential's fields under the bare
+        # names username/password/totp, so a plain typed parameter must not
+        # claim those keys or it would resolve to the secret value at runtime.
+        result = synthesize_code_block(
+            [
+                _interaction(
+                    "type_text",
+                    selector="#confirm",
+                    source_url="https://example.com/form",
+                    typed_length=8,
+                    role="textbox",
+                    accessible_name="Password",
+                ),
+                self._credential_fill(),
+            ]
+        )
+        assert result is not None
+        assert result.parameters[0] == {"key": "password_field"}
+        assert "fill(str(password_field))" in result.code
+        assert {"key": "password"} not in result.parameters
+
+    def test_synthesized_credential_code_is_valid_python(self) -> None:
+        result = synthesize_code_block(
+            [
+                self._credential_fill(),
+                self._credential_fill(selector="#passwordInput", credential_field="password"),
+            ]
+        )
+        assert result is not None
+        wrapped = "async def _block(page, authtest_simple):\n" + result.code
+        ast.parse(wrapped)
+
+
 def test_code_block_preflight_restores_recursion_limit() -> None:
     before = sys.getrecursionlimit()
     preflight_code_block("await page.locator('button[type=submit]').first.click(timeout=5000)\n")

tests/unit/test_copilot_fill_credential_field.py

@@ -0,0 +1,396 @@
+"""Tests for the copilot `fill_credential_field` scouting tool.
+
+OSS-synced: only example.* / authenticationtest.com fixtures. Secret values in
+fixtures are fake and exist to assert they never surface in any tool result,
+recorded interaction, or error string.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+from typing import Any, AsyncIterator
+from unittest.mock import AsyncMock
+
+import pytest
+
+from skyvern.forge import app
+from skyvern.forge.sdk.copilot import tools as tools_module
+from skyvern.forge.sdk.copilot.build_phase import _BROWSER_PRIMITIVE_TOOLS
+from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
+from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
+from skyvern.forge.sdk.copilot.tools import credential_fill as credential_fill_module
+from skyvern.forge.sdk.copilot.tools import scouting as scouting_module
+from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential
+
+_FAKE_PASSWORD = "fake-test-password-7x9"
+_FAKE_USERNAME = "qa.user@example.test"
+_FAKE_TOTP_SEED = "JBSWY3DPEHPK3PXP"
+
+
+def _resolved_credential(credential_id: str = "cred_123") -> SimpleNamespace:
+    return SimpleNamespace(credential_id=credential_id, name="authtest simple")
+
+
+def _policy(**overrides: Any) -> RequestPolicy:
+    policy = RequestPolicy(resolved_credentials=[_resolved_credential()])
+    for key, value in overrides.items():
+        setattr(policy, key, value)
+    return policy
+
+
+def _ctx(**overrides: Any) -> SimpleNamespace:
+    ns = SimpleNamespace(
+        organization_id="o_1",
+        request_policy=_policy(),
+        block_authoring_policy=BlockAuthoringPolicy.CODE_ONLY_BROWSER,
+        browser_session_id="pbs_1",
+        scouted_interactions=[],
+        scout_trajectory=[],
+        observed_browser_urls=[],
+        pending_scout_source_url=None,
+        pending_browser_interaction_observation=None,
+        discovery_mcp_server=None,
+        secret_scrub_values=[],
+    )
+    for key, value in overrides.items():
+        setattr(ns, key, value)
+    return ns
+
+
+class TestCredentialFillPolicyGate:
+    def test_rejects_outside_code_only_mode(self) -> None:
+        ctx = _ctx(block_authoring_policy=BlockAuthoringPolicy.STANDARD)
+        error = tools_module._credential_fill_policy_error(ctx, "cred_123")
+        assert error is not None
+        assert "login" in error
+
+    def test_rejects_without_request_policy(self) -> None:
+        ctx = _ctx(request_policy=None)
+        assert tools_module._credential_fill_policy_error(ctx, "cred_123") is not None
+
+    def test_rejects_when_run_blocks_not_allowed(self) -> None:
+        ctx = _ctx(request_policy=_policy(allow_run_blocks=False))
+        assert tools_module._credential_fill_policy_error(ctx, "cred_123") is not None
+
+    def test_rejects_credential_outside_resolved_set(self) -> None:
+        ctx = _ctx()
+        error = tools_module._credential_fill_policy_error(ctx, "cred_999")
+        assert error is not None
+        assert "cred_999" in error
+        assert "resolved" in error
+
+    def test_discovered_credential_is_not_run_authorized(self) -> None:
+        policy = _policy()
+        policy.discovered_credentials = [_resolved_credential("cred_discovered")]
+        ctx = _ctx(request_policy=policy)
+        assert tools_module._credential_fill_policy_error(ctx, "cred_discovered") is not None
+
+    def test_allows_resolved_credential_in_code_only_mode(self) -> None:
+        ctx = _ctx()
+        assert tools_module._credential_fill_policy_error(ctx, "cred_123") is None
+
+
+class TestResolveCredentialFillValue:
+    def _wire_vault(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        credential: Any,
+        *,
+        name: str = "authtest simple",
+    ) -> None:
+        db_credential = SimpleNamespace(vault_type=CredentialVaultType.BITWARDEN)
+        monkeypatch.setattr(
+            app.DATABASE,
+            "credentials",
+            SimpleNamespace(get_credential=AsyncMock(return_value=db_credential)),
+            raising=False,
+        )
+        vault = SimpleNamespace(
+            get_credential_item=AsyncMock(return_value=SimpleNamespace(name=name, credential=credential))
+        )
+        # `app` is an AppHolder proxy without __delattr__; patch the underlying instance
+        # so monkeypatch teardown can delete the attribute it set.
+        app_instance = object.__getattribute__(app, "_inst")
+        monkeypatch.setattr(
+            app_instance, "CREDENTIAL_VAULT_SERVICES", {CredentialVaultType.BITWARDEN: vault}, raising=False
+        )
+
+    @pytest.mark.asyncio
+    async def test_resolves_username_and_password(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(monkeypatch, PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=None))
+        value, name, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "username")
+        assert (value, name, error) == (_FAKE_USERNAME, "authtest simple", None)
+
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "password")
+        assert (value, error) == (_FAKE_PASSWORD, None)
+
+    @pytest.mark.asyncio
+    async def test_totp_mints_fresh_code_not_the_seed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(
+            monkeypatch,
+            PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=_FAKE_TOTP_SEED),
+        )
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "totp")
+        assert error is None
+        assert value is not None
+        assert value.isdigit()
+        assert len(value) == 6
+        assert value != _FAKE_TOTP_SEED
+
+    @pytest.mark.asyncio
+    async def test_password_resolve_registers_scrub_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(monkeypatch, PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=None))
+        ctx = _ctx()
+        value, _, error = await tools_module._resolve_credential_fill_value(ctx, "cred_123", "password")
+        assert (value, error) == (_FAKE_PASSWORD, None)
+        assert ctx.secret_scrub_values == [_FAKE_PASSWORD]
+
+    @pytest.mark.asyncio
+    async def test_username_resolve_does_not_register_scrub_value(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(monkeypatch, PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=None))
+        ctx = _ctx()
+        value, _, error = await tools_module._resolve_credential_fill_value(ctx, "cred_123", "username")
+        assert (value, error) == (_FAKE_USERNAME, None)
+        assert ctx.secret_scrub_values == []
+
+    @pytest.mark.asyncio
+    async def test_minted_otp_is_registered_at_mint_time(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(
+            monkeypatch,
+            PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=_FAKE_TOTP_SEED),
+        )
+        ctx = _ctx()
+        value, _, error = await tools_module._resolve_credential_fill_value(ctx, "cred_123", "totp")
+        assert error is None
+        assert ctx.secret_scrub_values == [value]
+        assert _FAKE_TOTP_SEED not in ctx.secret_scrub_values
+
+    @pytest.mark.asyncio
+    async def test_totp_without_seed_errors(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(monkeypatch, PasswordCredential(username=_FAKE_USERNAME, password=_FAKE_PASSWORD, totp=None))
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "totp")
+        assert value is None
+        assert error is not None
+        assert "TOTP" in error
+
+    @pytest.mark.asyncio
+    async def test_missing_credential_errors(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(
+            app.DATABASE,
+            "credentials",
+            SimpleNamespace(get_credential=AsyncMock(return_value=None)),
+            raising=False,
+        )
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "username")
+        assert value is None
+        assert error is not None
+        assert "cred_123" in error
+
+    @pytest.mark.asyncio
+    async def test_vault_exception_error_carries_no_secret_text(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        db_credential = SimpleNamespace(vault_type=CredentialVaultType.BITWARDEN)
+        monkeypatch.setattr(
+            app.DATABASE,
+            "credentials",
+            SimpleNamespace(get_credential=AsyncMock(return_value=db_credential)),
+            raising=False,
+        )
+        vault = SimpleNamespace(get_credential_item=AsyncMock(side_effect=RuntimeError(f"vault said {_FAKE_PASSWORD}")))
+        app_instance = object.__getattribute__(app, "_inst")
+        monkeypatch.setattr(
+            app_instance, "CREDENTIAL_VAULT_SERVICES", {CredentialVaultType.BITWARDEN: vault}, raising=False
+        )
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "password")
+        assert value is None
+        assert error is not None
+        assert _FAKE_PASSWORD not in error
+
+
+class _FakePage:
+    def __init__(self, fill_error: Exception | None = None) -> None:
+        self.fill_calls: list[tuple[Any, ...]] = []
+        self.fill_kwargs: list[dict[str, Any]] = []
+        self._fill_error = fill_error
+
+    async def fill(self, *args: Any, **kwargs: Any) -> None:
+        self.fill_calls.append(args)
+        self.fill_kwargs.append(kwargs)
+        if self._fill_error is not None:
+            raise self._fill_error
+
+
+def _wire_impl(
+    monkeypatch: pytest.MonkeyPatch,
+    page: _FakePage,
+    *,
+    secret_value: str = _FAKE_PASSWORD,
+    credential_name: str = "authtest simple",
+) -> None:
+    async def fake_resolve(_ctx: Any, _credential_id: str, _field: str) -> tuple[str, str, None]:
+        return secret_value, credential_name, None
+
+    async def fake_ensure(_ctx: Any) -> None:
+        return None
+
+    @asynccontextmanager
+    async def fake_browser_context(_ctx: Any) -> AsyncIterator[None]:
+        yield
+
+    async def fake_get_page(session_id: str | None = None) -> tuple[_FakePage, None]:
+        return page, None
+
+    async def fake_verify(*_args: Any, **_kwargs: Any) -> None:
+        return None
+
+    async def fake_url(_ctx: Any) -> str:
+        return "https://authenticationtest.com/simpleFormAuth/"
+
+    async def fake_role_name(*_args: Any, **_kwargs: Any) -> tuple[str, str]:
+        return "textbox", "Password"
+
+    monkeypatch.setattr(credential_fill_module, "_resolve_credential_fill_value", fake_resolve)
+    monkeypatch.setattr(credential_fill_module, "ensure_browser_session", fake_ensure)
+    monkeypatch.setattr(credential_fill_module, "mcp_browser_context", fake_browser_context)
+    monkeypatch.setattr(credential_fill_module, "get_page", fake_get_page)
+    monkeypatch.setattr(credential_fill_module, "_verify_scout_type_landed", fake_verify)
+    monkeypatch.setattr(credential_fill_module, "_live_working_page_url", fake_url)
+    monkeypatch.setattr(scouting_module, "_live_working_page_url", fake_url)
+    monkeypatch.setattr(credential_fill_module, "_resolve_scout_role_name", fake_role_name)
+    monkeypatch.setattr(credential_fill_module, "_tool_loop_error", lambda *a, **k: None)
+    monkeypatch.setattr(credential_fill_module, "_authority_tool_error", lambda *a, **k: None)
+    monkeypatch.setattr(credential_fill_module, "record_tool_step_result_for_ctx", lambda *a, **k: None)
+    monkeypatch.setattr(credential_fill_module, "_register_scout_interaction_observation", lambda *a, **k: 3)
+
+
+class TestFillCredentialFieldImpl:
+    @pytest.mark.asyncio
+    async def test_happy_path_fills_and_records_value_free(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+        ctx = _ctx()
+
+        result = await tools_module._fill_credential_field_impl(ctx, "#passwordInput", "cred_123", "password")
+
+        assert result["ok"] is True
+        assert page.fill_calls == [("#passwordInput", _FAKE_PASSWORD)]
+        assert page.fill_kwargs[0]["mode"] == "direct"
+        assert result["data"]["typed_length"] == len(_FAKE_PASSWORD)
+        assert result["data"]["credential_id"] == "cred_123"
+        assert result["data"]["field"] == "password"
+        assert result["data"]["observation_step"] == 3
+        assert _FAKE_PASSWORD not in json.dumps(result)
+
+        assert len(ctx.scouted_interactions) == 1
+        recorded = ctx.scouted_interactions[0]
+        assert recorded["tool_name"] == "fill_credential_field"
+        assert recorded["credential_id"] == "cred_123"
+        assert recorded["credential_field"] == "password"
+        assert recorded["credential_name"] == "authtest simple"
+        assert recorded["typed_length"] == len(_FAKE_PASSWORD)
+        assert _FAKE_PASSWORD not in json.dumps(recorded)
+        assert _FAKE_PASSWORD not in json.dumps(ctx.scout_trajectory)
+
+    @pytest.mark.asyncio
+    async def test_fill_error_text_is_scrubbed(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage(fill_error=RuntimeError(f"could not type {_FAKE_PASSWORD} into element"))
+        _wire_impl(monkeypatch, page)
+
+        result = await tools_module._fill_credential_field_impl(_ctx(), "#passwordInput", "cred_123", "password")
+
+        assert result["ok"] is False
+        assert _FAKE_PASSWORD not in result["error"]
+        assert "[REDACTED_SECRET]" in result["error"]
+
+    @pytest.mark.asyncio
+    async def test_rejects_unknown_field(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+
+        result = await tools_module._fill_credential_field_impl(_ctx(), "#cvv", "cred_123", "cvv")
+
+        assert result["ok"] is False
+        assert "username, password, totp" in result["error"]
+        assert page.fill_calls == []
+
+    @pytest.mark.asyncio
+    async def test_rejects_empty_selector(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+
+        result = await tools_module._fill_credential_field_impl(_ctx(), "   ", "cred_123", "password")
+
+        assert result["ok"] is False
+        assert page.fill_calls == []
+
+    @pytest.mark.asyncio
+    async def test_unresolved_credential_never_reaches_vault_or_page(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+        resolver = AsyncMock()
+        monkeypatch.setattr(credential_fill_module, "_resolve_credential_fill_value", resolver)
+
+        result = await tools_module._fill_credential_field_impl(_ctx(), "#passwordInput", "cred_999", "password")
+
+        assert result["ok"] is False
+        assert "cred_999" in result["error"]
+        resolver.assert_not_awaited()
+        assert page.fill_calls == []
+
+    @pytest.mark.asyncio
+    async def test_standard_mode_is_rejected(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+        ctx = _ctx(block_authoring_policy=BlockAuthoringPolicy.STANDARD)
+
+        result = await tools_module._fill_credential_field_impl(ctx, "#passwordInput", "cred_123", "password")
+
+        assert result["ok"] is False
+        assert page.fill_calls == []
+
+    @pytest.mark.asyncio
+    async def test_readback_failure_surfaces_and_skips_recording(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        page = _FakePage()
+        _wire_impl(monkeypatch, page)
+
+        async def failing_verify(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
+            return {"ok": False, "error": "field is still empty"}
+
+        monkeypatch.setattr(credential_fill_module, "_verify_scout_type_landed", failing_verify)
+        ctx = _ctx()
+
+        result = await tools_module._fill_credential_field_impl(ctx, "#passwordInput", "cred_123", "password")
+
+        assert result == {"ok": False, "error": "field is still empty"}
+        assert ctx.scouted_interactions == []
+
+
+class TestConsecutiveLoopGuardExemption:
+    def test_three_consecutive_fills_are_not_loop_blocked(self) -> None:
+        ctx = _ctx(consecutive_tool_tracker=[])
+        for field in ("username", "password", "totp"):
+            error = tools_module._tool_loop_error(
+                ctx, "fill_credential_field", {"selector": f"#{field}", "field": field}
+            )
+            assert error is None
+        assert ctx.consecutive_tool_tracker == []
+
+    def test_exemption_set_membership(self) -> None:
+        assert "fill_credential_field" in tools_module._CONSECUTIVE_LOOP_GUARD_EXEMPT_TOOLS
+
+
+class TestToolRegistration:
+    def test_tool_is_registered_native(self) -> None:
+        names = [tool.name for tool in tools_module.NATIVE_TOOLS]
+        assert "fill_credential_field" in names
+
+    def test_tool_description_states_value_free_contract(self) -> None:
+        tool = next(t for t in tools_module.NATIVE_TOOLS if t.name == "fill_credential_field")
+        description = tool.description or ""
+        assert "server-side" in description
+        assert "never" in description
+        assert "type_text" in description
+
+    def test_tool_is_phase_gated_as_browser_primitive(self) -> None:
+        assert "fill_credential_field" in _BROWSER_PRIMITIVE_TOOLS

tests/unit/test_copilot_output_policy.py

@@ -315,6 +315,119 @@ def test_rejects_raw_secret_in_structured_tool_arguments() -> None:
     assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
 
 
+class TestSanctionedSecretReferenceIdiom:
+    """`password = parameters[...]` / attribute-chain reads are references to bound
+    values, not leaks — the code-only authoring idiom must pass the syntactic
+    backstop while literal values keep blocking."""
+
+    def test_allows_parameters_subscript_assignment_in_tool_arguments(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": 'code: |\n  password = parameters["login_credential"]'},
+        )
+
+        assert verdict.allowed
+
+    def test_allows_credential_attribute_read_in_workflow_yaml(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            workflow_yaml=(
+                "code: |\n"
+                "  username = login_credential.username\n"
+                "  password = login_credential.password\n"
+                '  await page.locator("#passwordInput").fill(login_credential.password)\n'
+            ),
+        )
+
+        assert verdict.allowed
+
+    def test_allows_str_wrapped_parameter_reference(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": 'code: |\n  token = str(parameters.get("api_token"))'},
+        )
+
+        assert verdict.allowed
+
+    def test_still_rejects_literal_appended_to_reference(self) -> None:
+        for rhs in (
+            'login_credential.password+"hunter2"',
+            'str(login_credential.password)+"hunter2"',
+            'parameters["cred"]or"hunter2"',
+        ):
+            verdict = evaluate_output_policy(
+                request_policy=_policy(),
+                tool_arguments={"workflow_yaml": f"code: |\n  password = {rhs}"},
+            )
+
+            assert not verdict.allowed, rhs
+            assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_allows_keyword_argument_reference_with_closing_paren(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": 'code: |\n  await do_login(password=parameters["cred"])'},
+        )
+
+        assert verdict.allowed
+
+    def test_still_rejects_jwt_shaped_literal_assignment(self) -> None:
+        # A dotted literal (JWT-shaped) must not pass as an "attribute chain".
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "code: |\n  token = eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjMifQ.sig"},
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_still_rejects_dotted_literal_not_ending_in_credential_field(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "code: |\n  password = admin.password123"},
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_still_rejects_quoted_literal_assignment(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": 'code: |\n  password = "hunter2"'},
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_still_rejects_bare_literal_assignment(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "password: hunter2"},
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_still_rejects_email_password_pair_next_to_reference_idiom(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={
+                "workflow_yaml": 'code: |\n  password = parameters["cred"]\nnotes: qa.user@example.test:FakePass123!'
+            },
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
+    def test_chat_surface_request_policy_detection_is_unchanged(self) -> None:
+        # The pre-agent chat-surface detector stays conservative: even the code
+        # idiom pasted into chat still counts as a raw-secret signal there.
+        from skyvern.forge.sdk.copilot.request_policy import _raw_secret_detected
+
+        assert _raw_secret_detected("password: hunter2")
+        assert _raw_secret_detected('password = parameters["cred"]')
+
+
 def test_rejects_bulk_colon_delimited_credentials_in_tool_arguments() -> None:
     verdict = evaluate_output_policy(
         request_policy=_policy(),

tests/unit/test_copilot_secret_scrub.py

@@ -0,0 +1,305 @@
+"""Per-turn secret scrub set: registration plus exact-string scrubbing of page-readback tool results.
+
+OSS-synced: only example.* / authenticationtest.com fixtures with fake secret values.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+from collections.abc import Iterator
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+
+from skyvern.forge.sdk.copilot import mcp_adapter, secret_scrub
+from skyvern.forge.sdk.copilot.mcp_adapter import SchemaOverlay, SkyvernOverlayMCPServer
+from skyvern.forge.sdk.copilot.runtime import AgentContext
+from skyvern.forge.sdk.copilot.secret_scrub import (
+    REDACTED_SECRET_PLACEHOLDER,
+    clear_session_scrub_values,
+    register_secret_scrub_value,
+    scrub_secrets_from_structure,
+    scrub_secrets_from_text,
+)
+
+_FAKE_PASSWORD = "fake-pa55w0rd-7x9"
+_FAKE_OTP = "392817"
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_scrub_registry() -> Iterator[None]:
+    secret_scrub._SESSION_SCRUB_VALUES.clear()
+    yield
+    secret_scrub._SESSION_SCRUB_VALUES.clear()
+
+
+def _agent_ctx(browser_session_id: str = "pbs_1") -> AgentContext:
+    return AgentContext(
+        organization_id="o_1",
+        workflow_id="w_1",
+        workflow_permanent_id="wpid_1",
+        workflow_yaml="",
+        browser_session_id=browser_session_id,
+        stream=MagicMock(),
+    )
+
+
+class TestRegistration:
+    def test_registers_and_dedupes(self) -> None:
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, _FAKE_PASSWORD)
+        register_secret_scrub_value(ctx, _FAKE_PASSWORD)
+        register_secret_scrub_value(ctx, _FAKE_OTP)
+        assert ctx.secret_scrub_values == [_FAKE_PASSWORD, _FAKE_OTP]
+
+    def test_ignores_empty_and_non_string(self) -> None:
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, "")
+        register_secret_scrub_value(ctx, None)
+        assert ctx.secret_scrub_values == []
+
+    def test_tolerates_context_without_scrub_list(self) -> None:
+        register_secret_scrub_value(object(), _FAKE_PASSWORD)  # type: ignore[arg-type]
+
+
+class TestScrubStructure:
+    def test_replaces_in_nested_dicts_lists_and_keys(self) -> None:
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, _FAKE_PASSWORD)
+        register_secret_scrub_value(ctx, _FAKE_OTP)
+        result = scrub_secrets_from_structure(
+            ctx,
+            {
+                "data": {
+                    "result": [f"input value is {_FAKE_PASSWORD}", {_FAKE_OTP: "totp input"}],
+                    "html": f"<input value='{_FAKE_PASSWORD}'><input value='{_FAKE_OTP}'>",
+                },
+                "count": 3,
+            },
+        )
+        dumped = json.dumps(result)
+        assert _FAKE_PASSWORD not in dumped
+        assert _FAKE_OTP not in dumped
+        assert REDACTED_SECRET_PLACEHOLDER in dumped
+        assert result["count"] == 3
+
+    def test_no_registered_values_returns_object_unchanged(self) -> None:
+        ctx = _agent_ctx()
+        payload = {"data": {"result": f"value {_FAKE_PASSWORD}"}}
+        assert scrub_secrets_from_structure(ctx, payload) is payload
+
+    def test_overlapping_values_scrub_longest_first(self) -> None:
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, "abc")
+        register_secret_scrub_value(ctx, "abcdef")
+        assert scrub_secrets_from_text(ctx, "xabcdefy") == f"x{REDACTED_SECRET_PLACEHOLDER}y"
+
+    def test_image_base64_is_not_corrupted(self) -> None:
+        png_b64 = base64.b64encode(b"\x89PNG\r\n\x1a\n" + b"q" * 200).decode()
+        embedded = png_b64[20:30]
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, embedded)
+        result = scrub_secrets_from_structure(
+            ctx,
+            {"data": {"screenshot_base64": png_b64, "note": f"code {embedded} typed"}},
+        )
+        assert result["data"]["screenshot_base64"] == png_b64
+        assert embedded not in result["data"]["note"]
+
+
+class _FakeRawResult:
+    def __init__(self, payload: dict[str, Any], is_error: bool = False) -> None:
+        self.structured_content = payload
+        self.is_error = is_error
+        self.content: list[Any] = []
+
+
+class _FakeClient:
+    def __init__(self, payload: dict[str, Any] | Exception) -> None:
+        self._payload = payload
+
+    async def call_tool(self, name: str, args: dict[str, Any], raise_on_error: bool = False) -> _FakeRawResult:
+        if isinstance(self._payload, Exception):
+            raise self._payload
+        return _FakeRawResult(self._payload)
+
+
+def _evaluate_readback_payload() -> dict[str, Any]:
+    """An evaluate-shaped DOM readback of a credential form after a scout fill."""
+    return {
+        "ok": True,
+        "data": {
+            "url": "https://authenticationtest.com/totpChallenge/",
+            "title": "TOTP Challenge",
+            "result": {
+                "inputs": [
+                    {"name": "password", "selector": "#password", "type": "password"},
+                    {"name": "totp", "selector": "#totpmfa", "type": "text"},
+                ],
+                "rows": [{"cells": [{"text": "password"}, {"value": _FAKE_PASSWORD}, {"value": _FAKE_OTP}]}],
+                "html": f"<input id='password' value='{_FAKE_PASSWORD}'><input id='totpmfa' value='{_FAKE_OTP}'>",
+            },
+        },
+    }
+
+
+def _make_server(
+    ctx: AgentContext, payload: dict[str, Any] | Exception, overlay: SchemaOverlay
+) -> SkyvernOverlayMCPServer:
+    server = SkyvernOverlayMCPServer(
+        transport=MagicMock(),
+        overlays={"evaluate": overlay},
+        alias_map={},
+        allowlist=frozenset(),
+        context_provider=lambda: ctx,
+    )
+    server._client = _FakeClient(payload)
+    return server
+
+
+class TestAdapterScrubChokepoint:
+    @pytest.mark.asyncio
+    async def test_post_fill_evaluate_readback_is_redacted_in_result_record_and_loop_context(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from skyvern.forge.sdk.copilot.tools.mcp_hooks import _evaluate_post_hook, _evaluate_pre_hook
+
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, _FAKE_PASSWORD)
+        register_secret_scrub_value(ctx, _FAKE_OTP)
+
+        recorded: list[dict[str, Any]] = []
+        monkeypatch.setattr(
+            mcp_adapter,
+            "record_tool_step_result_for_ctx",
+            lambda _ctx, _tool, _args, result: recorded.append(dict(result)),
+        )
+
+        overlay = SchemaOverlay(pre_hook=_evaluate_pre_hook, post_hook=_evaluate_post_hook)
+        server = _make_server(ctx, _evaluate_readback_payload(), overlay)
+
+        result = await server.call_tool("evaluate", {"expression": "scan()"})
+
+        tool_text = result.content[0].text
+        assert _FAKE_PASSWORD not in tool_text
+        assert _FAKE_OTP not in tool_text
+        assert REDACTED_SECRET_PLACEHOLDER in tool_text
+
+        assert recorded, "tool result was not recorded"
+        recorded_text = json.dumps(recorded)
+        assert _FAKE_PASSWORD not in recorded_text
+        assert _FAKE_OTP not in recorded_text
+        assert REDACTED_SECRET_PLACEHOLDER in recorded_text
+
+        loop_context_text = json.dumps(
+            {
+                "flow_evidence": ctx.flow_evidence,
+                "composition_page_evidence": ctx.composition_page_evidence,
+                "scouted_interactions": ctx.scouted_interactions,
+                "scout_trajectory": ctx.scout_trajectory,
+            }
+        )
+        assert _FAKE_PASSWORD not in loop_context_text
+        assert _FAKE_OTP not in loop_context_text
+        assert ctx.flow_evidence, "evaluate evidence was not recorded into the loop context"
+        assert REDACTED_SECRET_PLACEHOLDER in loop_context_text
+
+    @pytest.mark.asyncio
+    async def test_no_fill_this_turn_leaves_result_unscrubbed(self) -> None:
+        ctx = _agent_ctx()
+        server = _make_server(ctx, _evaluate_readback_payload(), SchemaOverlay())
+
+        result = await server.call_tool("evaluate", {"expression": "scan()"})
+
+        tool_text = result.content[0].text
+        assert _FAKE_PASSWORD in tool_text
+        assert _FAKE_OTP in tool_text
+        assert REDACTED_SECRET_PLACEHOLDER not in tool_text
+
+    @pytest.mark.asyncio
+    async def test_tool_exception_text_is_redacted(self) -> None:
+        ctx = _agent_ctx()
+        register_secret_scrub_value(ctx, _FAKE_PASSWORD)
+        server = _make_server(
+            ctx, RuntimeError(f"locator resolved to <input value='{_FAKE_PASSWORD}'>"), SchemaOverlay()
+        )
+
+        result = await server.call_tool("evaluate", {"expression": "scan()"})
+
+        tool_text = result.content[0].text
+        assert result.isError is True
+        assert _FAKE_PASSWORD not in tool_text
+        assert REDACTED_SECRET_PLACEHOLDER in tool_text
+
+
+class TestCrossTurnSessionScrub:
+    def test_session_registry_survives_a_fresh_turn_context(self) -> None:
+        turn1 = _agent_ctx()
+        register_secret_scrub_value(turn1, _FAKE_PASSWORD)
+
+        turn2 = _agent_ctx()
+        assert turn2.secret_scrub_values == []
+        assert scrub_secrets_from_text(turn2, f"value {_FAKE_PASSWORD}") == f"value {REDACTED_SECRET_PLACEHOLDER}"
+
+    def test_registry_is_scoped_per_browser_session(self) -> None:
+        turn1 = _agent_ctx("pbs_1")
+        register_secret_scrub_value(turn1, _FAKE_PASSWORD)
+
+        other_session = _agent_ctx("pbs_2")
+        assert scrub_secrets_from_text(other_session, f"value {_FAKE_PASSWORD}") == f"value {_FAKE_PASSWORD}"
+
+    def test_clear_session_scrub_values_drops_the_session(self) -> None:
+        turn1 = _agent_ctx()
+        register_secret_scrub_value(turn1, _FAKE_PASSWORD)
+        clear_session_scrub_values("pbs_1")
+
+        turn2 = _agent_ctx()
+        assert scrub_secrets_from_text(turn2, f"value {_FAKE_PASSWORD}") == f"value {_FAKE_PASSWORD}"
+
+    def test_session_registry_is_bounded_and_evicts_oldest(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(secret_scrub, "_MAX_SCRUB_SESSIONS", 3)
+        for i in range(5):
+            register_secret_scrub_value(_agent_ctx(f"pbs_{i}"), _FAKE_PASSWORD)
+
+        assert len(secret_scrub._SESSION_SCRUB_VALUES) == 3
+        assert set(secret_scrub._SESSION_SCRUB_VALUES) == {"pbs_2", "pbs_3", "pbs_4"}
+        # The evicted oldest session no longer scrubs; the newest still does.
+        assert scrub_secrets_from_text(_agent_ctx("pbs_0"), _FAKE_PASSWORD) == _FAKE_PASSWORD
+        assert scrub_secrets_from_text(_agent_ctx("pbs_4"), _FAKE_PASSWORD) == REDACTED_SECRET_PLACEHOLDER
+
+    @pytest.mark.asyncio
+    async def test_readback_in_later_turn_is_redacted_in_result_and_record(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from skyvern.forge.sdk.copilot.tools.mcp_hooks import _evaluate_post_hook, _evaluate_pre_hook
+
+        turn1 = _agent_ctx()
+        register_secret_scrub_value(turn1, _FAKE_PASSWORD)
+        register_secret_scrub_value(turn1, _FAKE_OTP)
+
+        turn2 = _agent_ctx()
+        assert turn2.secret_scrub_values == []
+
+        recorded: list[dict[str, Any]] = []
+        monkeypatch.setattr(
+            mcp_adapter,
+            "record_tool_step_result_for_ctx",
+            lambda _ctx, _tool, _args, result: recorded.append(dict(result)),
+        )
+
+        overlay = SchemaOverlay(pre_hook=_evaluate_pre_hook, post_hook=_evaluate_post_hook)
+        server = _make_server(turn2, _evaluate_readback_payload(), overlay)
+
+        result = await server.call_tool("evaluate", {"expression": "scan()"})
+
+        tool_text = result.content[0].text
+        assert _FAKE_PASSWORD not in tool_text
+        assert _FAKE_OTP not in tool_text
+        assert REDACTED_SECRET_PLACEHOLDER in tool_text
+
+        assert recorded, "tool result was not recorded"
+        recorded_text = json.dumps(recorded)
+        assert _FAKE_PASSWORD not in recorded_text
+        assert _FAKE_OTP not in recorded_text