fix(SKY-8920): cap extract-* prompt sizes to reduce Gemini TPM 429s (#5502)

alembic/versions/2026_04_14_2233-c9005bafa5ec_restore_include_extracted_text_to_tasks.py

@@ -0,0 +1,30 @@
+"""restore include_extracted_text to tasks
+
+Revision ID: c9005bafa5ec
+Revises: 12f6731887f4
+Create Date: 2026-04-14T22:33:36.939859+00:00
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "c9005bafa5ec"
+down_revision: Union[str, None] = "12f6731887f4"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "tasks",
+        sa.Column("include_extracted_text", sa.Boolean(), server_default=sa.true(), nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("tasks", "include_extracted_text")

skyvern/core/script_generations/real_skyvern_page_ai.py

@@ -24,6 +24,7 @@ from skyvern.services import script_service
 from skyvern.services.otp_service import poll_otp_value
 from skyvern.utils.css_selector import compute_selector_options
 from skyvern.utils.prompt_engine import load_prompt_with_elements
+from skyvern.utils.prompt_truncation import truncate_extraction_schema
 from skyvern.webeye.actions import handler_utils
 from skyvern.webeye.actions.actions import (
     ActionStatus,
@@ -901,6 +902,7 @@ class RealSkyvernPageAi(SkyvernPageAi):
         intention: str | None = None,
         data: str | dict[str, Any] | None = None,
         skip_refresh: bool = False,
+        include_extracted_text: bool = True,
     ) -> dict[str, Any] | list | str | None:
         """Extract information from the page using AI."""
 
@@ -916,15 +918,17 @@ class RealSkyvernPageAi(SkyvernPageAi):
         # Render the prompt FIRST so the cache key hashes the exact string
         # that will be sent to the LLM (captures economy-tree swaps and 2/3
         # truncation inside load_prompt_with_elements).
+        extracted_text_for_prompt = self.scraped_page.extracted_text if include_extracted_text else None
+
         extract_information_prompt = load_prompt_with_elements(
             element_tree_builder=self.scraped_page,
             prompt_engine=prompt_engine,
             template_name="extract-information",
             html_need_skyvern_attrs=False,
             data_extraction_goal=prompt,
-            extracted_information_schema=schema,
+            extracted_information_schema=truncate_extraction_schema(schema),
             current_url=self.scraped_page.url,
-            extracted_text=self.scraped_page.extracted_text,
+            extracted_text=extracted_text_for_prompt,
             error_code_mapping_str=(json.dumps(error_code_mapping) if error_code_mapping else None),
             local_datetime=local_datetime_str,
         )

skyvern/core/script_generations/skyvern_page_ai.py

@@ -66,6 +66,7 @@ class SkyvernPageAi(Protocol):
         intention: str | None = None,
         data: str | dict[str, Any] | None = None,
         skip_refresh: bool = False,
+        include_extracted_text: bool = True,
     ) -> dict[str, Any] | list | str | None:
         """Extract information from the page using AI."""
         ...

skyvern/forge/agent.py

@@ -113,7 +113,14 @@ from skyvern.services.otp_service import (
     try_generate_totp_from_credential,
 )
 from skyvern.utils.image_resizer import Resolution
-from skyvern.utils.prompt_engine import MaxStepsReasonResponse, load_prompt_with_elements
+from skyvern.utils.prompt_engine import (
+    PROMPT_HARD_CEILING_TOKENS,
+    MaxStepsReasonResponse,
+    enforce_prompt_ceiling,
+    load_prompt_with_elements,
+)
+from skyvern.utils.prompt_truncation import truncate_extraction_schema
+from skyvern.utils.token_counter import count_tokens
 from skyvern.webeye.actions.action_types import ActionType
 from skyvern.webeye.actions.actions import (
     Action,
@@ -145,6 +152,15 @@ from skyvern.webeye.utils.page import SkyvernFrame
 LOG = structlog.get_logger()
 
 EXTRACT_ACTION_TEMPLATE = "extract-action"
+
+
+class _PromptCeilingExceeded(Exception):
+    """Internal signal: the cached split-template prompt blew past the
+    PROMPT_HARD_CEILING_TOKENS budget. Raised inside the cached extract-action
+    branch to trigger the fall-through to load_prompt_with_elements, which
+    applies the per-template fallback drop chain."""
+
+
 EXTRACT_ACTION_PROMPT_NAME = "extract-actions"
 EXTRACT_ACTION_CACHE_KEY_PREFIX = f"{EXTRACT_ACTION_TEMPLATE}-static"
 
@@ -308,6 +324,7 @@ class ForgeAgent:
             browser_address=workflow_run.browser_address,
             browser_session_id=workflow_run.browser_session_id,
             download_timeout=task_block.download_timeout,
+            include_extracted_text=task_block.include_extracted_text,
         )
         LOG.info(
             "Created a new task for workflow run",
@@ -376,6 +393,7 @@ class ForgeAgent:
             extra_http_headers=task_request.extra_http_headers,
             browser_session_id=task_request.browser_session_id,
             browser_address=task_request.browser_address,
+            include_extracted_text=task_request.include_extracted_text,
         )
         LOG.info(
             "Created new task",
@@ -3224,6 +3242,16 @@ class ForgeAgent:
 
                 combined_prompt = f"{static_prompt.rstrip()}\n\n{dynamic_prompt.lstrip()}"
 
+                if count_tokens(combined_prompt) > PROMPT_HARD_CEILING_TOKENS:
+                    # The cached split-template path renders static+dynamic separately,
+                    # so the load_prompt_with_elements ceiling logic never sees this
+                    # prompt. Raise the dedicated sentinel to trigger the except below,
+                    # which falls through to the full load_prompt_with_elements render
+                    # where the fallback drop chain can apply.
+                    raise _PromptCeilingExceeded(
+                        f"cached extract-action prompt exceeded {PROMPT_HARD_CEILING_TOKENS} tokens"
+                    )
+
                 LOG.info(
                     "Using cached prompt",
                     task_id=task.task_id,
@@ -5039,13 +5067,20 @@ class ForgeAgent:
     @staticmethod
     async def create_extract_action(task: Task, step: Step, scraped_page: ScrapedPage) -> ExtractAction:
         context = skyvern_context.ensure_context()
+        capped_schema = truncate_extraction_schema(task.extracted_information_schema)
         # generate reasoning by prompt llm to think briefly what data to extract
-        prompt = prompt_engine.load_prompt(
-            "data-extraction-summary",
-            data_extraction_goal=task.data_extraction_goal,
-            data_extraction_schema=task.extracted_information_schema,
-            current_url=scraped_page.url,
-            local_datetime=datetime.now(context.tz_info).isoformat(),
+        summary_kwargs: dict[str, Any] = {
+            "data_extraction_goal": task.data_extraction_goal,
+            "data_extraction_schema": capped_schema,
+            "current_url": scraped_page.url,
+            "local_datetime": datetime.now(context.tz_info).isoformat(),
+        }
+        prompt = prompt_engine.load_prompt("data-extraction-summary", **summary_kwargs)
+        prompt = enforce_prompt_ceiling(
+            prompt,
+            prompt_engine=prompt_engine,
+            template_name="data-extraction-summary",
+            kwargs=summary_kwargs,
         )
 
         # Cache the summary LLM call — the inputs (goal, schema, URL) are

skyvern/forge/prompts/skyvern/extract-information.j2

@@ -27,8 +27,10 @@ Clickable elements from `{{ current_url }}`:
 ```
 
 Current URL: {{ current_url }}
+{%- if extracted_text %}
 
 Text extracted from the webpage: {{ extracted_text }}
+{%- endif %}
 
 User Navigation Payload: {{ navigation_payload }}
 

skyvern/forge/sdk/db/models.py

@@ -107,6 +107,7 @@ class TaskModel(Base):
     max_steps_per_run = Column(Integer, nullable=True)
     application = Column(String, nullable=True)
     include_action_history_in_verification = Column(Boolean, default=False, nullable=True)
+    include_extracted_text = Column(Boolean, default=True, nullable=False, server_default=sqlalchemy.true())
     queued_at = Column(DateTime, nullable=True)
     started_at = Column(DateTime, nullable=True)
     finished_at = Column(DateTime, nullable=True)

skyvern/forge/sdk/db/repositories/tasks.py

@@ -65,6 +65,7 @@ class TasksRepository(BaseRepository):
         browser_session_id: str | None = None,
         browser_address: str | None = None,
         download_timeout: float | None = None,
+        include_extracted_text: bool = True,
     ) -> Task:
         # Sanitize text fields to remove NUL bytes and control characters
         # that PostgreSQL cannot store in text columns
@@ -108,6 +109,7 @@ class TasksRepository(BaseRepository):
                 browser_session_id=browser_session_id,
                 browser_address=browser_address,
                 download_timeout=download_timeout,
+                include_extracted_text=include_extracted_text,
             )
             session.add(new_task)
             await session.commit()

skyvern/forge/sdk/db/utils.py

@@ -225,6 +225,7 @@ def convert_to_task(task_obj: TaskModel, debug_enabled: bool = False, workflow_p
         complete_criterion=task_obj.complete_criterion,
         terminate_criterion=task_obj.terminate_criterion,
         include_action_history_in_verification=task_obj.include_action_history_in_verification,
+        include_extracted_text=task_obj.include_extracted_text,
         webhook_callback_url=task_obj.webhook_callback_url,
         webhook_failure_reason=task_obj.webhook_failure_reason,
         totp_verification_url=task_obj.totp_verification_url,
@@ -698,6 +699,7 @@ def convert_to_workflow_run_block(
         block.terminate_criterion = task.terminate_criterion
         block.complete_criterion = task.complete_criterion
         block.include_action_history_in_verification = task.include_action_history_in_verification
+        block.include_extracted_text = task.include_extracted_text
 
     return block
 

skyvern/forge/sdk/schemas/tasks.py

@@ -118,6 +118,10 @@ class TaskBase(BaseModel):
         description="The maximum time to wait for downloads to complete, in seconds. If not set, defaults to BROWSER_DOWNLOAD_TIMEOUT seconds.",
         examples=[15.0],
     )
+    include_extracted_text: bool = Field(
+        default=True,
+        description="If False, omit the scraped page text dump from the extract-information prompt. ExtractionBlock opts out; everything else keeps the default.",
+    )
 
 
 class TaskRequest(TaskBase):

skyvern/forge/sdk/schemas/workflow_runs.py

@@ -39,6 +39,7 @@ class WorkflowRunBlock(BaseModel):
     created_at: datetime
     modified_at: datetime
     include_action_history_in_verification: bool | None = False
+    include_extracted_text: bool = True
     duration: float | None = None
 
     # for loop block

skyvern/forge/sdk/workflow/models/block.py

@@ -729,6 +729,7 @@ class BaseTaskBlock(Block):
     complete_verification: bool = True
     include_action_history_in_verification: bool = False
     download_timeout: float | None = None  # minutes
+    include_extracted_text: bool = True
 
     def get_all_parameters(
         self,
@@ -4676,6 +4677,7 @@ class ExtractionBlock(BaseTaskBlock):
     block_type: Literal[BlockType.EXTRACTION] = BlockType.EXTRACTION  # type: ignore
 
     data_extraction_goal: str
+    include_extracted_text: bool = False
 
 
 class LoginBlock(BaseTaskBlock):

skyvern/library/skyvern_browser_page_ai.py

@@ -167,12 +167,14 @@ class SdkSkyvernPageAi(SkyvernPageAi):
         intention: str | None = None,
         data: str | dict[str, Any] | None = None,
         skip_refresh: bool = False,
+        include_extracted_text: bool = True,
     ) -> dict[str, Any] | list | str | None:
         """Extract information from the page using AI via API call.
 
-        Note: skip_refresh is accepted for Protocol compatibility but not forwarded
-        to the API. The server-side always refreshes when called via the SDK HTTP path.
-        The optimization only takes effect on the direct RealSkyvernPageAI path (MCP local browser).
+        Note: skip_refresh and include_extracted_text are accepted for Protocol
+        compatibility but not forwarded to the API. The server-side controls
+        both via the Task record on the SDK HTTP path. The optimizations only
+        take effect on the direct RealSkyvernPageAI path (MCP local browser).
         """
 
         LOG.info("AI extract", prompt=prompt, workflow_run_id=self._browser.workflow_run_id)

skyvern/utils/prompt_engine.py

@@ -38,6 +38,21 @@ class MaxStepsReasonResponse(BaseModel):
     failure_categories: list[dict] = []
 
 
+PROMPT_HARD_CEILING_TOKENS = 180_000
+
+CEILING_FALLBACK_KEYS_BY_TEMPLATE: dict[str, list[str]] = {
+    "extract-information": [
+        "previous_extracted_information",
+        "extracted_information_schema",
+        "extracted_text",
+    ],
+    "extract-action": ["action_history", "navigation_payload_str"],
+    "extract-action-dynamic": ["action_history", "navigation_payload_str"],
+    "extract-action-static": [],
+    "data-extraction-summary": ["data_extraction_schema"],
+}
+
+
 def load_prompt_with_elements(
     element_tree_builder: ElementTreeBuilder,
     prompt_engine: PromptEngine,
@@ -54,10 +69,8 @@ def load_prompt_with_elements(
     token_count = count_tokens(prompt)
     if token_count > DEFAULT_MAX_TOKENS and element_tree_builder.support_economy_elements_tree():
         # get rid of all the secondary elements like SVG, etc
-        economy_elements_tree = element_tree_builder.build_economy_elements_tree(
-            html_need_skyvern_attrs=html_need_skyvern_attrs
-        )
-        prompt = prompt_engine.load_prompt(template_name, elements=economy_elements_tree, **kwargs)
+        elements = element_tree_builder.build_economy_elements_tree(html_need_skyvern_attrs=html_need_skyvern_attrs)
+        prompt = prompt_engine.load_prompt(template_name, elements=elements, **kwargs)
         economy_token_count = count_tokens(prompt)
         LOG.warning(
             "Prompt is longer than the max tokens. Going to use the economy elements tree.",
@@ -69,11 +82,11 @@ def load_prompt_with_elements(
         if economy_token_count > DEFAULT_MAX_TOKENS:
             # !!! HACK alert
             # dump the last 1/3 of the html context and keep the first 2/3 of the html context
-            economy_elements_tree_dumped = element_tree_builder.build_economy_elements_tree(
+            elements = element_tree_builder.build_economy_elements_tree(
                 html_need_skyvern_attrs=html_need_skyvern_attrs,
                 percent_to_keep=2 / 3,
             )
-            prompt = prompt_engine.load_prompt(template_name, elements=economy_elements_tree_dumped, **kwargs)
+            prompt = prompt_engine.load_prompt(template_name, elements=elements, **kwargs)
             token_count_after_dump = count_tokens(prompt)
             LOG.warning(
                 "Prompt is still longer than the max tokens. Will only keep the first 2/3 of the html context.",
@@ -83,4 +96,57 @@ def load_prompt_with_elements(
                 token_count_after_dump=token_count_after_dump,
                 max_tokens=DEFAULT_MAX_TOKENS,
             )
+
+    return enforce_prompt_ceiling(
+        prompt,
+        prompt_engine=prompt_engine,
+        template_name=template_name,
+        kwargs=kwargs,
+        elements=elements,
+    )
+
+
+def enforce_prompt_ceiling(
+    prompt: str,
+    *,
+    prompt_engine: PromptEngine,
+    template_name: str,
+    kwargs: dict[str, Any],
+    elements: Any | None = None,
+) -> str:
+    """Drop fallback-chain keys in priority order until the prompt fits.
+
+    Use this at any call site that builds a prompt via prompt_engine.load_prompt
+    directly, so the 180k hard ceiling is enforced regardless of whether the
+    caller went through load_prompt_with_elements.
+    """
+    final_token_count = count_tokens(prompt)
+    if final_token_count <= PROMPT_HARD_CEILING_TOKENS:
+        return prompt
+    fallback_keys = CEILING_FALLBACK_KEYS_BY_TEMPLATE.get(template_name, [])
+    working_kwargs = dict(kwargs)
+    for drop_key in fallback_keys:
+        if working_kwargs.get(drop_key) is None:
+            continue
+        LOG.warning(
+            "Prompt exceeds hard ceiling; dropping fallback key",
+            template_name=template_name,
+            drop_key=drop_key,
+            final_token_count=final_token_count,
+            hard_ceiling=PROMPT_HARD_CEILING_TOKENS,
+        )
+        working_kwargs[drop_key] = None
+        if elements is None:
+            prompt = prompt_engine.load_prompt(template_name, **working_kwargs)
+        else:
+            prompt = prompt_engine.load_prompt(template_name, elements=elements, **working_kwargs)
+        final_token_count = count_tokens(prompt)
+        if final_token_count <= PROMPT_HARD_CEILING_TOKENS:
+            return prompt
+    LOG.error(
+        "Prompt still exceeds hard ceiling after all fallback drops",
+        template_name=template_name,
+        final_token_count=final_token_count,
+        hard_ceiling=PROMPT_HARD_CEILING_TOKENS,
+    )
     return prompt

skyvern/utils/prompt_truncation.py

@@ -0,0 +1,149 @@
+"""Helpers for capping prompt inputs before templating.
+
+Separate from prompt_engine.py so per-field caps can be applied at the call
+boundary without reaching into the element-tree truncation logic. Reuses
+skyvern.utils.token_counter.count_tokens for consistent measurement.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import structlog
+
+from skyvern.utils.token_counter import count_tokens, decode_tokens, encode_tokens
+
+LOG = structlog.get_logger()
+
+PREVIOUS_EXTRACTED_INFO_MAX_TOKENS = 20_000
+EXTRACTION_SCHEMA_MAX_TOKENS = 10_000
+
+
+def _crop_string(value: str, max_tokens: int) -> str:
+    if max_tokens <= 0:
+        return ""
+    tokens = encode_tokens(value)
+    if len(tokens) <= max_tokens:
+        return value
+    return decode_tokens(tokens[-max_tokens:])
+
+
+def _crop_list(value: list[Any], max_tokens: int) -> list[Any]:
+    """Greedy reverse-iteration: keep the most recent items that fit and stop
+    at the first overshoot. Earlier items are dropped even if they would
+    individually fit — recency over coverage is the intent."""
+    result: list[Any] = []
+    remaining = max_tokens
+    for item in reversed(value):
+        rendered = item if isinstance(item, str) else json.dumps(item, default=str)
+        cost = count_tokens(rendered)
+        if cost > remaining:
+            break
+        result.append(item)
+        remaining -= cost
+    result.reverse()
+    return result
+
+
+def _crop_dict(value: dict[str, Any], max_tokens: int) -> dict[str, Any]:
+    if not value:
+        return value
+    per_key = max(1, max_tokens // len(value))
+    cropped: dict[str, Any] = {}
+    for key, inner in value.items():
+        inner_str = inner if isinstance(inner, str) else json.dumps(inner, default=str)
+        if count_tokens(inner_str) <= per_key:
+            # Preserve the original type when the value already fits — only
+            # values that overshoot the per-key budget get coerced to a
+            # truncated string.
+            cropped[key] = inner
+        else:
+            cropped[key] = _crop_string(inner_str, per_key)
+    return cropped
+
+
+def truncate_previous_extracted_information(
+    value: Any,
+    max_tokens: int = PREVIOUS_EXTRACTED_INFO_MAX_TOKENS,
+) -> Any:
+    """Cap the prompt contribution of `previous_extracted_information`."""
+    if value is None:
+        return None
+
+    rendered_before = value if isinstance(value, str) else json.dumps(value, default=str)
+    before_tokens = count_tokens(rendered_before)
+
+    if isinstance(value, str):
+        result: Any = _crop_string(value, max_tokens)
+    elif isinstance(value, list):
+        cropped = _crop_list(value, max_tokens)
+        result = cropped if cropped else _crop_string(json.dumps(value, default=str), max_tokens)
+    elif isinstance(value, dict):
+        result = _crop_dict(value, max_tokens)
+    else:
+        result = _crop_string(json.dumps(value, default=str), max_tokens)
+
+    rendered_after = result if isinstance(result, str) else json.dumps(result, default=str)
+    after_tokens = count_tokens(rendered_after)
+    if after_tokens < before_tokens:
+        LOG.warning(
+            "Truncated previous_extracted_information",
+            value_kind=type(value).__name__,
+            before_tokens=before_tokens,
+            after_tokens=after_tokens,
+            max_tokens=max_tokens,
+        )
+    return result
+
+
+def truncate_extraction_schema(
+    schema: Any,
+    max_tokens: int = EXTRACTION_SCHEMA_MAX_TOKENS,
+) -> Any:
+    """Cap a customer-provided JSONSchema when it blows the prompt budget.
+
+    Passes through unchanged when under budget. Otherwise replaces the body
+    with a placeholder that preserves top-level shape (object/array) and tells
+    the LLM to fall back to general extraction.
+    """
+    if schema is None:
+        return None
+
+    top_level: Any = None
+    if isinstance(schema, str):
+        before_tokens = count_tokens(schema)
+        if before_tokens <= max_tokens:
+            return schema
+        # Truncating a JSON string tail-first breaks brace/bracket pairing; try
+        # to recover the top-level type from a parse attempt, otherwise fall
+        # through to the object placeholder below.
+        try:
+            parsed = json.loads(schema)
+            if isinstance(parsed, dict):
+                top_level = parsed.get("type")
+        except (ValueError, TypeError):
+            pass
+    else:
+        rendered = json.dumps(schema, default=str)
+        before_tokens = count_tokens(rendered)
+        if before_tokens <= max_tokens:
+            return schema
+        if isinstance(schema, dict):
+            top_level = schema.get("type")
+
+    placeholder = {
+        "type": top_level if top_level in ("object", "array") else "object",
+        "_skyvern_schema_truncated": True,
+        "_skyvern_schema_hint": (
+            "The full extraction schema exceeded the prompt budget and was replaced with this placeholder. "
+            "Extract all relevant information from the page as structured data; do not assume specific field names."
+        ),
+    }
+    LOG.warning(
+        "Truncated extraction schema to placeholder",
+        before_tokens=before_tokens,
+        max_tokens=max_tokens,
+        placeholder_top_level=placeholder["type"],
+    )
+    return placeholder

skyvern/utils/token_counter.py

@@ -5,3 +5,11 @@ _ENCODING = tiktoken.encoding_for_model("gpt-4o")
 
 def count_tokens(text: str) -> int:
     return len(_ENCODING.encode(text))
+
+
+def encode_tokens(text: str) -> list[int]:
+    return _ENCODING.encode(text)
+
+
+def decode_tokens(tokens: list[int]) -> str:
+    return _ENCODING.decode(tokens)

skyvern/webeye/actions/handler.py

@@ -84,6 +84,7 @@ from skyvern.utils.prompt_engine import (
     CheckPhoneNumberFormatResponse,
     load_prompt_with_elements,
 )
+from skyvern.utils.prompt_truncation import truncate_extraction_schema, truncate_previous_extracted_information
 from skyvern.webeye.actions import actions, handler_utils
 from skyvern.webeye.actions.action_types import ActionType
 from skyvern.webeye.actions.actions import (
@@ -4269,6 +4270,11 @@ async def extract_information_for_navigation_goal(
     # same value (avoids stale hits when date-relative extraction goals cross midnight).
     local_datetime_str = datetime.now(context.tz_info).isoformat()
 
+    extracted_text_for_prompt = scraped_page_refreshed.extracted_text if task.include_extracted_text else None
+
+    previous_info_capped = truncate_previous_extracted_information(task.extracted_information)
+    capped_schema = truncate_extraction_schema(task.extracted_information_schema)
+
     # Render the prompt FIRST so the cache key hashes the exact string that
     # will be sent to the LLM (captures economy-tree swaps and 2/3 truncation
     # inside load_prompt_with_elements).
@@ -4279,11 +4285,11 @@ async def extract_information_for_navigation_goal(
         html_need_skyvern_attrs=False,
         navigation_goal=task.navigation_goal,
         navigation_payload=task.navigation_payload,
-        previous_extracted_information=task.extracted_information,
+        previous_extracted_information=previous_info_capped,
         data_extraction_goal=task.data_extraction_goal,
-        extracted_information_schema=task.extracted_information_schema,
+        extracted_information_schema=capped_schema,
         current_url=scraped_page_refreshed.url,
-        extracted_text=scraped_page_refreshed.extracted_text,
+        extracted_text=extracted_text_for_prompt,
         error_code_mapping_str=(json.dumps(task.error_code_mapping) if task.error_code_mapping else None),
         local_datetime=local_datetime_str,
     )

tests/unit/test_data_extraction_summary_schema_cap.py

@@ -0,0 +1,79 @@
+"""Tests for the extraction-schema cap at the data-extraction-summary call site (SKY-8920 Phase D)."""
+
+from __future__ import annotations
+
+
+def _run_create_extract_action(monkeypatch, extracted_information_schema):
+    import asyncio
+    from unittest.mock import MagicMock
+
+    from skyvern.forge import agent as agent_module
+
+    captured: dict = {}
+
+    original_load_prompt = agent_module.prompt_engine.load_prompt
+
+    def capturing_load_prompt(template_name, **kwargs):
+        if template_name == "data-extraction-summary":
+            captured.update(kwargs)
+        return original_load_prompt(template_name, **kwargs)
+
+    async def fake_handler(*, prompt, step, prompt_name):
+        captured["prompt"] = prompt
+        return {"summary": "ok"}
+
+    monkeypatch.setattr(agent_module.prompt_engine, "load_prompt", capturing_load_prompt)
+    monkeypatch.setattr(agent_module.app, "EXTRACTION_LLM_API_HANDLER", fake_handler)
+    monkeypatch.setattr(
+        agent_module.skyvern_context,
+        "ensure_context",
+        lambda: MagicMock(tz_info=None, workflow_run_id="wr_test"),
+    )
+    monkeypatch.setattr(
+        agent_module.extraction_cache,
+        "compute_cache_key",
+        lambda **_: None,
+    )
+    monkeypatch.setattr(
+        agent_module.extraction_cache,
+        "lookup",
+        lambda *a, **k: None,
+    )
+
+    task = MagicMock()
+    task.data_extraction_goal = "Extract documents"
+    task.extracted_information_schema = extracted_information_schema
+    task.task_id = "tsk_test"
+    task.workflow_run_id = "wr_test"
+    task.organization_id = "o_test"
+
+    step = MagicMock(step_id="stp_test", order=0)
+    scraped_page = MagicMock(url="https://example.test")
+    # Avoid attribute errors from AsyncMock
+    step.step_id = "stp_test"
+    step.order = 0
+
+    asyncio.run(agent_module.ForgeAgent.create_extract_action(task=task, step=step, scraped_page=scraped_page))
+    return captured
+
+
+def test_create_extract_action_caps_huge_schema(monkeypatch) -> None:
+    huge_schema = {
+        "type": "object",
+        "properties": {f"field_{i}": {"type": "string", "description": "lorem ipsum " * 40} for i in range(1000)},
+    }
+
+    captured = _run_create_extract_action(monkeypatch, huge_schema)
+
+    schema_passed = captured["data_extraction_schema"]
+    assert isinstance(schema_passed, dict)
+    assert schema_passed.get("_skyvern_schema_truncated") is True
+    assert schema_passed.get("type") == "object"
+
+
+def test_create_extract_action_passes_small_schema_unchanged(monkeypatch) -> None:
+    small_schema = {"type": "object", "properties": {"title": {"type": "string"}}}
+
+    captured = _run_create_extract_action(monkeypatch, small_schema)
+
+    assert captured["data_extraction_schema"] == small_schema

tests/unit/test_extract_actions_ceiling.py

@@ -0,0 +1,70 @@
+"""Tests for the 180k ceiling applied to extract-action templates (SKY-8920 Phase E)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+
+def _make_element_tree_builder() -> MagicMock:
+    builder = MagicMock()
+    builder.build_element_tree = MagicMock(return_value="<a>link</a>")
+    builder.support_economy_elements_tree = MagicMock(return_value=False)
+    return builder
+
+
+def test_extract_action_ceiling_drops_action_history_on_overshoot() -> None:
+    from skyvern.forge.prompts import prompt_engine as engine_module
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS, load_prompt_with_elements
+    from skyvern.utils.token_counter import count_tokens
+
+    oversized_history = "\n".join(f"UNIQUE_ACTION_BLOCK_{i}_" + ("lorem ipsum " * 200) for i in range(3000))
+
+    rendered = load_prompt_with_elements(
+        element_tree_builder=_make_element_tree_builder(),
+        prompt_engine=engine_module,
+        template_name="extract-action",
+        navigation_goal="Log in to the site",
+        navigation_payload_str="{}",
+        starting_url="https://example.test",
+        current_url="https://example.test",
+        data_extraction_goal=None,
+        action_history=oversized_history,
+        error_code_mapping_str=None,
+        local_datetime="2026-04-14T12:00:00",
+        verification_code_check=False,
+        complete_criterion=None,
+        terminate_criterion=None,
+        parse_select_feature_enabled=False,
+        has_magic_link_page=False,
+    )
+
+    assert count_tokens(rendered) <= PROMPT_HARD_CEILING_TOKENS
+    assert "UNIQUE_ACTION_BLOCK_0_" not in rendered
+
+
+def test_extract_action_small_prompt_passes_through() -> None:
+    from skyvern.forge.prompts import prompt_engine as engine_module
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS, load_prompt_with_elements
+    from skyvern.utils.token_counter import count_tokens
+
+    rendered = load_prompt_with_elements(
+        element_tree_builder=_make_element_tree_builder(),
+        prompt_engine=engine_module,
+        template_name="extract-action",
+        navigation_goal="Log in to the site",
+        navigation_payload_str="{}",
+        starting_url="https://example.test",
+        current_url="https://example.test",
+        data_extraction_goal=None,
+        action_history="small history",
+        error_code_mapping_str=None,
+        local_datetime="2026-04-14T12:00:00",
+        verification_code_check=False,
+        complete_criterion=None,
+        terminate_criterion=None,
+        parse_select_feature_enabled=False,
+        has_magic_link_page=False,
+    )
+
+    assert "small history" in rendered
+    assert count_tokens(rendered) <= PROMPT_HARD_CEILING_TOKENS

tests/unit/test_extract_information_previous_info_cap.py

@@ -0,0 +1,155 @@
+"""Tests for previous_extracted_information capping (SKY-8920 Phase B + D)."""
+
+from __future__ import annotations
+
+
+def _make_scraped_page(refreshed_extracted_text: str = "small"):
+    from unittest.mock import AsyncMock, MagicMock
+
+    refreshed = MagicMock()
+    refreshed.extracted_text = refreshed_extracted_text
+    refreshed.url = "https://example.test"
+    refreshed.screenshots = []
+    refreshed.build_element_tree = MagicMock(return_value="<a>link</a>")
+    refreshed.support_economy_elements_tree = MagicMock(return_value=False)
+
+    scraped_page = MagicMock()
+    scraped_page.refresh = AsyncMock(return_value=refreshed)
+    scraped_page.screenshots = []
+    return scraped_page
+
+
+def _capture_handler_kwargs(monkeypatch, previous_extracted_information):
+    import asyncio
+    from unittest.mock import AsyncMock, MagicMock
+
+    from skyvern.webeye.actions import handler
+
+    captured: dict = {}
+
+    def fake_load_prompt_with_elements(**kwargs):
+        captured.update(kwargs)
+        return "rendered-prompt"
+
+    async def fake_handler_call(**kwargs):
+        captured["prompt"] = kwargs.get("prompt")
+        return {}
+
+    monkeypatch.setattr(handler, "load_prompt_with_elements", fake_load_prompt_with_elements)
+    monkeypatch.setattr(handler, "ensure_context", lambda: MagicMock(tz_info=None))
+    monkeypatch.setattr(handler.service_utils, "is_cua_task", AsyncMock(return_value=False))
+    monkeypatch.setattr(
+        handler.LLMAPIHandlerFactory,
+        "get_override_llm_api_handler",
+        lambda llm_key, default: fake_handler_call,
+    )
+    monkeypatch.setattr(handler.extraction_cache, "compute_cache_key", lambda **_: None)
+
+    scraped_page = _make_scraped_page()
+
+    task = MagicMock()
+    task.navigation_goal = None
+    task.navigation_payload = None
+    task.extracted_information = previous_extracted_information
+    task.data_extraction_goal = "Extract documents"
+    task.extracted_information_schema = {"type": "object"}
+    task.error_code_mapping = None
+    task.llm_key = None
+    task.workflow_run_id = None
+    task.task_id = "tsk_test"
+    task.include_extracted_text = True
+
+    asyncio.run(handler.extract_information_for_navigation_goal(task=task, step=MagicMock(), scraped_page=scraped_page))
+
+    return captured
+
+
+def test_handler_truncates_huge_previous_extracted_information(monkeypatch) -> None:
+    import json
+
+    from skyvern.utils.token_counter import count_tokens
+
+    huge_prev = [{"iter": i, "blob": "x" * 2_000} for i in range(500)]
+
+    captured = _capture_handler_kwargs(monkeypatch, previous_extracted_information=huge_prev)
+
+    capped = captured["previous_extracted_information"]
+    assert capped is not None
+    assert isinstance(capped, list)
+    # Recent iterations survive; early ones are dropped.
+    assert capped[-1]["iter"] == 499
+    assert capped[0]["iter"] != 0
+    # Capped result fits inside the 20k-token budget.
+    assert count_tokens(json.dumps(capped)) <= 20_500
+
+
+def test_handler_passes_small_previous_extracted_information_unchanged(monkeypatch) -> None:
+    small_prev = [{"iter": 0, "blob": "small"}]
+    captured = _capture_handler_kwargs(monkeypatch, previous_extracted_information=small_prev)
+    assert captured["previous_extracted_information"] == small_prev
+
+
+def _capture_handler_schema(monkeypatch, extracted_information_schema):
+    import asyncio
+    from unittest.mock import AsyncMock, MagicMock
+
+    from skyvern.webeye.actions import handler
+
+    captured: dict = {}
+
+    def fake_load_prompt_with_elements(**kwargs):
+        captured.update(kwargs)
+        return "rendered-prompt"
+
+    async def fake_handler_call(**kwargs):
+        captured["prompt"] = kwargs.get("prompt")
+        return {}
+
+    monkeypatch.setattr(handler, "load_prompt_with_elements", fake_load_prompt_with_elements)
+    monkeypatch.setattr(handler, "ensure_context", lambda: MagicMock(tz_info=None))
+    monkeypatch.setattr(handler.service_utils, "is_cua_task", AsyncMock(return_value=False))
+    monkeypatch.setattr(
+        handler.LLMAPIHandlerFactory,
+        "get_override_llm_api_handler",
+        lambda llm_key, default: fake_handler_call,
+    )
+    monkeypatch.setattr(handler.extraction_cache, "compute_cache_key", lambda **_: None)
+
+    scraped_page = _make_scraped_page()
+
+    task = MagicMock()
+    task.navigation_goal = None
+    task.navigation_payload = None
+    task.extracted_information = None
+    task.data_extraction_goal = "Extract documents"
+    task.extracted_information_schema = extracted_information_schema
+    task.error_code_mapping = None
+    task.llm_key = None
+    task.workflow_run_id = None
+    task.task_id = "tsk_test"
+    task.include_extracted_text = True
+
+    asyncio.run(handler.extract_information_for_navigation_goal(task=task, step=MagicMock(), scraped_page=scraped_page))
+
+    return captured
+
+
+def test_handler_caps_huge_extraction_schema(monkeypatch) -> None:
+    huge_schema = {
+        "type": "object",
+        "properties": {f"field_{i}": {"type": "string", "description": "lorem ipsum " * 40} for i in range(1000)},
+    }
+
+    captured = _capture_handler_schema(monkeypatch, huge_schema)
+
+    schema_passed = captured["extracted_information_schema"]
+    assert isinstance(schema_passed, dict)
+    assert schema_passed.get("_skyvern_schema_truncated") is True
+
+
+def test_handler_passes_small_schema_unchanged(monkeypatch) -> None:
+    small_schema = {"type": "object", "properties": {"title": {"type": "string"}}}
+
+    captured = _capture_handler_schema(monkeypatch, small_schema)
+
+    assert captured["extracted_information_schema"] == small_schema

tests/unit/test_extract_information_text_optout.py

@@ -0,0 +1,260 @@
+"""Tests for the include_extracted_text opt-out chain (SKY-8920 Phase A)."""
+
+from __future__ import annotations
+
+
+def test_task_base_has_include_extracted_text_field_with_default_true() -> None:
+    from skyvern.forge.sdk.schemas.tasks import TaskBase
+
+    assert "include_extracted_text" in TaskBase.model_fields
+    field = TaskBase.model_fields["include_extracted_text"]
+    assert field.default is True
+
+
+def test_task_base_accepts_include_extracted_text_false() -> None:
+    from skyvern.forge.sdk.schemas.tasks import TaskBase
+
+    task = TaskBase(url="https://example.test", include_extracted_text=False)
+    assert task.include_extracted_text is False
+
+
+def test_task_base_defaults_include_extracted_text_true() -> None:
+    from skyvern.forge.sdk.schemas.tasks import TaskBase
+
+    task = TaskBase(url="https://example.test")
+    assert task.include_extracted_text is True
+
+
+def test_base_task_block_has_include_extracted_text_field_default_true() -> None:
+    from skyvern.forge.sdk.workflow.models.block import BaseTaskBlock
+
+    assert "include_extracted_text" in BaseTaskBlock.model_fields
+    assert BaseTaskBlock.model_fields["include_extracted_text"].default is True
+
+
+def test_extraction_block_overrides_include_extracted_text_to_false() -> None:
+    from skyvern.forge.sdk.workflow.models.block import ExtractionBlock
+
+    assert "include_extracted_text" in ExtractionBlock.model_fields
+    assert ExtractionBlock.model_fields["include_extracted_text"].default is False
+
+
+def _make_scraped_page_refreshed(extracted_text: str):
+    from unittest.mock import MagicMock
+
+    refreshed = MagicMock()
+    refreshed.extracted_text = extracted_text
+    refreshed.url = "https://example.test"
+    refreshed.screenshots = []
+    refreshed.build_element_tree = MagicMock(return_value="<a href='/d.pdf'>Doc</a>")
+    refreshed.support_economy_elements_tree = MagicMock(return_value=False)
+    return refreshed
+
+
+def _make_task_for_extract_information(include_extracted_text: bool):
+    from unittest.mock import MagicMock
+
+    task = MagicMock()
+    task.navigation_goal = None
+    task.navigation_payload = None
+    task.extracted_information = None
+    task.data_extraction_goal = "Extract documents"
+    task.extracted_information_schema = {"type": "object"}
+    task.error_code_mapping = None
+    task.llm_key = None
+    task.workflow_run_id = None
+    task.task_id = "tsk_test"
+    task.include_extracted_text = include_extracted_text
+    return task
+
+
+def _capture_extract_information_kwargs(monkeypatch, include_extracted_text: bool):
+    """Run the handler with monkeypatches that capture what's passed to load_prompt_with_elements."""
+    import asyncio
+    from unittest.mock import AsyncMock, MagicMock
+
+    from skyvern.webeye.actions import handler
+
+    captured: dict = {}
+
+    def fake_load_prompt_with_elements(**kwargs):
+        captured.update(kwargs)
+        return "rendered-prompt"
+
+    async def fake_handler_call(**kwargs):
+        captured["prompt"] = kwargs.get("prompt")
+        return {}
+
+    # The handler calls compute_cache_key (may raise), LOG, LLMAPIHandlerFactory,
+    # service_utils.is_cua_task. Monkey-patch just enough to reach load_prompt_with_elements
+    # and the handler call.
+    monkeypatch.setattr(handler, "load_prompt_with_elements", fake_load_prompt_with_elements)
+    monkeypatch.setattr(handler, "ensure_context", lambda: MagicMock(tz_info=None))
+    monkeypatch.setattr(handler.service_utils, "is_cua_task", AsyncMock(return_value=False))
+    monkeypatch.setattr(
+        handler.LLMAPIHandlerFactory,
+        "get_override_llm_api_handler",
+        lambda llm_key, default: fake_handler_call,
+    )
+    # Short-circuit the extraction_cache so we always fall through to the LLM path.
+    monkeypatch.setattr(handler.extraction_cache, "compute_cache_key", lambda **_: None)
+
+    refreshed = _make_scraped_page_refreshed("PROHIBITED_TEXT_MARKER")
+    scraped_page = MagicMock()
+    scraped_page.refresh = AsyncMock(return_value=refreshed)
+    scraped_page.screenshots = []
+
+    task = _make_task_for_extract_information(include_extracted_text=include_extracted_text)
+
+    asyncio.run(handler.extract_information_for_navigation_goal(task=task, step=MagicMock(), scraped_page=scraped_page))
+
+    return captured
+
+
+def test_handler_omits_extracted_text_when_task_flag_is_false(monkeypatch) -> None:
+    captured = _capture_extract_information_kwargs(monkeypatch, include_extracted_text=False)
+    assert captured["extracted_text"] is None
+
+
+def test_handler_passes_extracted_text_when_task_flag_is_true(monkeypatch) -> None:
+    captured = _capture_extract_information_kwargs(monkeypatch, include_extracted_text=True)
+    assert captured["extracted_text"] == "PROHIBITED_TEXT_MARKER"
+
+
+def _render_extract_information(**kwargs) -> str:
+    from skyvern.forge.prompts import prompt_engine
+
+    base_kwargs = {
+        "data_extraction_goal": "Extract documents",
+        "extracted_information_schema": {"type": "object"},
+        "current_url": "https://example.test",
+        "elements": "<a>link</a>",
+        "extracted_text": None,
+        "error_code_mapping_str": None,
+        "navigation_payload": None,
+        "previous_extracted_information": None,
+        "local_datetime": "2026-04-14T12:00:00",
+    }
+    base_kwargs.update(kwargs)
+    return prompt_engine.load_prompt("extract-information", **base_kwargs)
+
+
+def test_extract_information_template_omits_text_line_when_extracted_text_is_none() -> None:
+    rendered = _render_extract_information(extracted_text=None)
+    assert "Text extracted from the webpage" not in rendered
+
+
+def test_extract_information_template_includes_text_line_when_extracted_text_is_set() -> None:
+    rendered = _render_extract_information(extracted_text="RENDERED_MARKER")
+    assert "RENDERED_MARKER" in rendered
+    assert "Text extracted from the webpage: RENDERED_MARKER" in rendered
+
+
+def _capture_ai_extract_kwargs(monkeypatch, include_extracted_text: bool):
+    """Run RealSkyvernPageAi.ai_extract with monkeypatches that capture the kwargs passed
+    to load_prompt_with_elements."""
+    import asyncio
+    from unittest.mock import MagicMock
+
+    from skyvern.core.script_generations import real_skyvern_page_ai as module
+
+    captured: dict = {}
+
+    def fake_load_prompt_with_elements(**kwargs):
+        captured.update(kwargs)
+        return "rendered-prompt"
+
+    scraped_page = MagicMock()
+    scraped_page.url = "https://example.test"
+    scraped_page.extracted_text = "PROHIBITED_MARKER"
+    scraped_page.screenshots = []
+    scraped_page.build_element_tree = MagicMock(return_value="<a>link</a>")
+    scraped_page.support_economy_elements_tree = MagicMock(return_value=False)
+
+    page = module.RealSkyvernPageAi.__new__(module.RealSkyvernPageAi)
+    page.scraped_page = scraped_page
+    page.current_label = None
+
+    async def fake_refresh(*_args, **_kwargs):
+        return None
+
+    async def fake_handler(*, prompt, step, screenshots, prompt_name, force_dict):
+        return {}
+
+    monkeypatch.setattr(module, "load_prompt_with_elements", fake_load_prompt_with_elements)
+    monkeypatch.setattr(module.app, "EXTRACTION_LLM_API_HANDLER", fake_handler)
+    monkeypatch.setattr(module.extraction_cache, "compute_cache_key", lambda **_: None)
+    monkeypatch.setattr(page, "_refresh_scraped_page", fake_refresh)
+    monkeypatch.setattr(module.skyvern_context, "current", lambda: None)
+
+    asyncio.run(
+        page.ai_extract(
+            prompt="Extract documents",
+            schema={"type": "object"},
+            include_extracted_text=include_extracted_text,
+        )
+    )
+
+    return captured
+
+
+def test_ai_extract_omits_extracted_text_when_flag_is_false(monkeypatch) -> None:
+    captured = _capture_ai_extract_kwargs(monkeypatch, include_extracted_text=False)
+    assert captured["extracted_text"] is None
+
+
+def test_ai_extract_passes_extracted_text_when_flag_is_true(monkeypatch) -> None:
+    captured = _capture_ai_extract_kwargs(monkeypatch, include_extracted_text=True)
+    assert captured["extracted_text"] == "PROHIBITED_MARKER"
+
+
+def _capture_ai_extract_kwargs_with_schema(monkeypatch, schema):
+    import asyncio
+    from unittest.mock import MagicMock
+
+    from skyvern.core.script_generations import real_skyvern_page_ai as module
+
+    captured: dict = {}
+
+    def fake_load_prompt_with_elements(**kwargs):
+        captured.update(kwargs)
+        return "rendered-prompt"
+
+    scraped_page = MagicMock()
+    scraped_page.url = "https://example.test"
+    scraped_page.extracted_text = "TXT"
+    scraped_page.screenshots = []
+    scraped_page.build_element_tree = MagicMock(return_value="<a>link</a>")
+    scraped_page.support_economy_elements_tree = MagicMock(return_value=False)
+
+    page = module.RealSkyvernPageAi.__new__(module.RealSkyvernPageAi)
+    page.scraped_page = scraped_page
+    page.current_label = None
+
+    async def fake_refresh(*_args, **_kwargs):
+        return None
+
+    async def fake_handler(*, prompt, step, screenshots, prompt_name, force_dict):
+        return {}
+
+    monkeypatch.setattr(module, "load_prompt_with_elements", fake_load_prompt_with_elements)
+    monkeypatch.setattr(module.app, "EXTRACTION_LLM_API_HANDLER", fake_handler)
+    monkeypatch.setattr(module.extraction_cache, "compute_cache_key", lambda **_: None)
+    monkeypatch.setattr(page, "_refresh_scraped_page", fake_refresh)
+    monkeypatch.setattr(module.skyvern_context, "current", lambda: None)
+
+    asyncio.run(page.ai_extract(prompt="Extract documents", schema=schema, include_extracted_text=True))
+    return captured
+
+
+def test_ai_extract_caps_huge_schema(monkeypatch) -> None:
+    big_props = {f"field_{i}": {"type": "string", "description": "x" * 200} for i in range(500)}
+    huge_schema = {"type": "object", "properties": big_props}
+    captured = _capture_ai_extract_kwargs_with_schema(monkeypatch, huge_schema)
+    assert captured["extracted_information_schema"].get("_skyvern_schema_truncated") is True
+
+
+def test_ai_extract_passes_small_schema_unchanged(monkeypatch) -> None:
+    small_schema = {"type": "object", "properties": {"x": {"type": "string"}}}
+    captured = _capture_ai_extract_kwargs_with_schema(monkeypatch, small_schema)
+    assert captured["extracted_information_schema"] == small_schema

tests/unit/test_load_prompt_with_elements_ceiling.py

@@ -0,0 +1,109 @@
+"""Tests for the post-render 180k token ceiling in load_prompt_with_elements (SKY-8920 Phase C + E)."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+
+def test_prompt_hard_ceiling_is_below_gpt5_mini_cap() -> None:
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS
+
+    assert PROMPT_HARD_CEILING_TOKENS == 180_000
+    assert PROMPT_HARD_CEILING_TOKENS < 272_000
+
+
+def test_ceiling_fallback_keys_by_template_has_known_mappings() -> None:
+    from skyvern.utils.prompt_engine import CEILING_FALLBACK_KEYS_BY_TEMPLATE
+
+    assert CEILING_FALLBACK_KEYS_BY_TEMPLATE["extract-information"] == [
+        "previous_extracted_information",
+        "extracted_information_schema",
+        "extracted_text",
+    ]
+    assert CEILING_FALLBACK_KEYS_BY_TEMPLATE["extract-action"] == [
+        "action_history",
+        "navigation_payload_str",
+    ]
+    assert CEILING_FALLBACK_KEYS_BY_TEMPLATE["data-extraction-summary"] == [
+        "data_extraction_schema",
+    ]
+
+
+def _make_element_tree_builder() -> MagicMock:
+    builder = MagicMock()
+    builder.build_element_tree = MagicMock(return_value="<a>link</a>")
+    builder.support_economy_elements_tree = MagicMock(return_value=False)
+    return builder
+
+
+def test_load_prompt_with_elements_drops_previous_info_when_over_ceiling() -> None:
+    from skyvern.forge.prompts import prompt_engine as engine_module
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS, load_prompt_with_elements
+    from skyvern.utils.token_counter import count_tokens
+
+    # List of distinct English-ish words well over the 180k token ceiling.
+    oversized_prev = [{"iter": i, "marker": f"UNIQUE_BLOCK_{i}_" + ("lorem ipsum " * 200)} for i in range(3000)]
+
+    rendered = load_prompt_with_elements(
+        element_tree_builder=_make_element_tree_builder(),
+        prompt_engine=engine_module,
+        template_name="extract-information",
+        data_extraction_goal="Extract documents",
+        extracted_information_schema={"type": "object"},
+        current_url="https://example.test",
+        extracted_text=None,
+        error_code_mapping_str=None,
+        navigation_payload=None,
+        local_datetime="2026-04-14T12:00:00",
+        previous_extracted_information=oversized_prev,
+    )
+
+    assert count_tokens(rendered) <= PROMPT_HARD_CEILING_TOKENS
+    assert "UNIQUE_BLOCK_0_" not in rendered
+
+
+def test_enforce_prompt_ceiling_drops_fallback_keys_without_elements() -> None:
+    from skyvern.forge.prompts import prompt_engine as engine_module
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS, enforce_prompt_ceiling
+    from skyvern.utils.token_counter import count_tokens
+
+    giant_schema = {"type": "object", "_blob": "lorem " * 300_000}
+    kwargs = {
+        "data_extraction_goal": "Extract",
+        "data_extraction_schema": giant_schema,
+        "current_url": "https://example.test",
+        "local_datetime": "2026-04-14T12:00:00",
+    }
+    rendered = engine_module.load_prompt("data-extraction-summary", **kwargs)
+    assert count_tokens(rendered) > PROMPT_HARD_CEILING_TOKENS
+
+    rendered = enforce_prompt_ceiling(
+        rendered,
+        prompt_engine=engine_module,
+        template_name="data-extraction-summary",
+        kwargs=kwargs,
+    )
+    assert count_tokens(rendered) <= PROMPT_HARD_CEILING_TOKENS
+
+
+def test_load_prompt_with_elements_respects_ceiling_for_small_prompts() -> None:
+    from skyvern.forge.prompts import prompt_engine as engine_module
+    from skyvern.utils.prompt_engine import PROMPT_HARD_CEILING_TOKENS, load_prompt_with_elements
+    from skyvern.utils.token_counter import count_tokens
+
+    rendered = load_prompt_with_elements(
+        element_tree_builder=_make_element_tree_builder(),
+        prompt_engine=engine_module,
+        template_name="extract-information",
+        data_extraction_goal="Extract documents",
+        extracted_information_schema={"type": "object"},
+        current_url="https://example.test",
+        extracted_text=None,
+        error_code_mapping_str=None,
+        navigation_payload=None,
+        local_datetime="2026-04-14T12:00:00",
+        previous_extracted_information="small blob",
+    )
+
+    assert "small blob" in rendered
+    assert count_tokens(rendered) <= PROMPT_HARD_CEILING_TOKENS

tests/unit/test_prompt_truncation.py

@@ -0,0 +1,136 @@
+"""Tests for prompt truncation helpers (SKY-8920 Phase B + D)."""
+
+from __future__ import annotations
+
+
+def test_truncate_none_returns_none() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+
+    assert truncate_previous_extracted_information(None, max_tokens=1000) is None
+
+
+def test_truncate_short_string_returns_unchanged() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+
+    value = "small tail"
+    result = truncate_previous_extracted_information(value, max_tokens=1000)
+    assert result == value
+
+
+def test_truncate_long_string_keeps_tail() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+
+    value = "HEAD" + ("x" * 500_000) + "TAIL"
+    result = truncate_previous_extracted_information(value, max_tokens=100)
+    assert isinstance(result, str)
+    assert result.endswith("TAIL")
+    assert "HEAD" not in result
+
+
+def test_truncate_long_string_respects_exact_token_cap() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+    from skyvern.utils.token_counter import count_tokens
+
+    value = ("lorem ipsum dolor sit amet " * 20_000) + "UNIQUE_TAIL_MARKER"
+    for cap in (50, 500, 5_000):
+        result = truncate_previous_extracted_information(value, max_tokens=cap)
+        assert isinstance(result, str)
+        assert count_tokens(result) <= cap, f"cap={cap} overshot: {count_tokens(result)}"
+
+
+def test_truncate_long_list_keeps_recent_entries() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+
+    value = [{"i": i, "pad": "x" * 1000} for i in range(500)]
+    result = truncate_previous_extracted_information(value, max_tokens=500)
+    assert isinstance(result, list)
+    assert result[-1]["i"] == 499
+    assert len(result) < len(value)
+
+
+def test_truncate_dict_preserves_top_level_keys_and_caps_values() -> None:
+    import json
+
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+    from skyvern.utils.token_counter import count_tokens
+
+    value = {"a": "x" * 50_000, "b": "y" * 50_000}
+    result = truncate_previous_extracted_information(value, max_tokens=200)
+    assert isinstance(result, dict)
+    assert set(result.keys()) == {"a", "b"}
+    assert count_tokens(json.dumps(result)) <= 400  # small slack for JSON wrapping
+
+
+def test_truncate_dict_preserves_value_types_when_under_per_key_budget() -> None:
+    from skyvern.utils.prompt_truncation import truncate_previous_extracted_information
+
+    value = {
+        "small_dict": {"nested": "data", "count": 42},
+        "small_list": [1, 2, 3],
+        "small_str": "hello",
+    }
+    result = truncate_previous_extracted_information(value, max_tokens=10_000)
+    # Each item is well under the per_key budget; original types should survive,
+    # not be coerced to JSON-serialized strings.
+    assert result == value
+    assert isinstance(result["small_dict"], dict)
+    assert isinstance(result["small_list"], list)
+    assert isinstance(result["small_str"], str)
+
+
+def test_truncate_respects_default_budget() -> None:
+    from skyvern.utils.prompt_truncation import PREVIOUS_EXTRACTED_INFO_MAX_TOKENS
+
+    assert PREVIOUS_EXTRACTED_INFO_MAX_TOKENS == 20_000
+
+
+def test_truncate_extraction_schema_none_returns_none() -> None:
+    from skyvern.utils.prompt_truncation import truncate_extraction_schema
+
+    assert truncate_extraction_schema(None, max_tokens=1000) is None
+
+
+def test_truncate_extraction_schema_short_passes_through() -> None:
+    from skyvern.utils.prompt_truncation import truncate_extraction_schema
+
+    schema = {"type": "object", "properties": {"name": {"type": "string"}}}
+    result = truncate_extraction_schema(schema, max_tokens=1000)
+    assert result == schema
+
+
+def test_truncate_extraction_schema_large_returns_summary_placeholder() -> None:
+    import json
+
+    from skyvern.utils.prompt_truncation import truncate_extraction_schema
+    from skyvern.utils.token_counter import count_tokens
+
+    big_props = {f"field_{i}": {"type": "string", "description": "x" * 200} for i in range(500)}
+    schema = {"type": "object", "properties": big_props}
+    original_tokens = count_tokens(json.dumps(schema))
+    assert original_tokens > 10_000
+
+    result = truncate_extraction_schema(schema, max_tokens=2_000)
+    result_tokens = count_tokens(json.dumps(result))
+
+    assert result_tokens <= 2_200
+    assert result["type"] == "object"
+    assert result.get("_skyvern_schema_truncated") is True
+
+
+def test_truncate_extraction_schema_default_budget() -> None:
+    from skyvern.utils.prompt_truncation import EXTRACTION_SCHEMA_MAX_TOKENS
+
+    assert EXTRACTION_SCHEMA_MAX_TOKENS == 10_000
+
+
+def test_truncate_extraction_schema_preserves_array_top_level() -> None:
+    import json
+
+    from skyvern.utils.prompt_truncation import truncate_extraction_schema
+    from skyvern.utils.token_counter import count_tokens
+
+    items = [{"f": f"val_{i}_" + ("lorem ipsum " * 40)} for i in range(1000)]
+    schema = {"type": "array", "items": {"type": "object", "properties": {"f": {"type": "string"}}}, "_items": items}
+    result = truncate_extraction_schema(schema, max_tokens=2_000)
+    assert count_tokens(json.dumps(result)) <= 2_200
+    assert result["type"] == "array"