vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-07-09 16:09:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

SKY-11498: Route request-policy classifier through lite Copilot model setting (#7191)

Browse source
This commit is contained in:
Andrew Neilson 2026-07-07 21:20:33 -07:00 • committed by GitHub
parent 95d1e966fa
commit b7c1383f81
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 383 additions and 150 deletions
Download patch file Download diff file Expand all files Collapse all files

1
skyvern/config.py
View file

@ -350,6 +350,7 @@ class Settings(BaseSettings):
SCRIPT_GENERATION_LLM_KEY: str | None = None
SCRIPT_REVIEWER_LLM_KEY: str | None = None
ADAPTIVE_SCRIPT_GEN_LLM_KEY: str | None = None
WORKFLOW_COPILOT_LLM_KEY: str | None = None
WORKFLOW_COPILOT_AGENT_LLM_KEY: str | None = None
WORKFLOW_COPILOT_FAST_LLM_KEY: str | None = None
# COMMON

8
skyvern/forge/forge_app.py
View file

@ -89,6 +89,7 @@ class ForgeApp:
SCRIPT_GENERATION_LLM_API_HANDLER: LLMAPIHandler
SCRIPT_REVIEWER_LLM_API_HANDLER: LLMAPIHandler
ADAPTIVE_SCRIPT_GEN_LLM_API_HANDLER: LLMAPIHandler
WORKFLOW_COPILOT_LLM_API_HANDLER: LLMAPIHandler
WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER: LLMAPIHandler
WORKFLOW_COPILOT_FAST_LLM_API_HANDLER: LLMAPIHandler
WORKFLOW_CONTEXT_MANAGER: WorkflowContextManager
@ -258,10 +259,15 @@ def create_forge_app() -> ForgeApp:
if settings.ADAPTIVE_SCRIPT_GEN_LLM_KEY
else app.LLM_API_HANDLER
)
app.WORKFLOW_COPILOT_LLM_API_HANDLER = (
LLMAPIHandlerFactory.get_llm_api_handler(settings.WORKFLOW_COPILOT_LLM_KEY)
if settings.WORKFLOW_COPILOT_LLM_KEY
else app.LLM_API_HANDLER
)
app.WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER = (
LLMAPIHandlerFactory.get_llm_api_handler(settings.WORKFLOW_COPILOT_AGENT_LLM_KEY)
if settings.WORKFLOW_COPILOT_AGENT_LLM_KEY
else app.LLM_API_HANDLER
else app.WORKFLOW_COPILOT_LLM_API_HANDLER
)
app.WORKFLOW_COPILOT_FAST_LLM_API_HANDLER = (
LLMAPIHandlerFactory.get_llm_api_handler(settings.WORKFLOW_COPILOT_FAST_LLM_KEY)

27
skyvern/forge/sdk/copilot/agent.py
View file

@ -32,6 +32,7 @@ from pydantic import ValidationError
from skyvern.forge import app
from skyvern.forge.prompts import prompt_engine
from skyvern.forge.sdk.copilot import llm_config
from skyvern.forge.sdk.copilot.blocker_signal import (
CopilotToolBlockerSignal,
assert_clean_user_facing_text,
@ -257,7 +258,17 @@ def _copilot_turn_span(
yield span
def _resolve_request_policy_handler(llm_api_handler: Any) -> Any:
async def _resolve_request_policy_handler(
llm_api_handler: LLMAPIHandler | None, workflow_permanent_id: str | None, organization_id: str | None
) -> Any:
lite_handler = await llm_config.resolve_lite_copilot_handler(workflow_permanent_id, organization_id)
if lite_handler is not None:
return lite_handler
LOG.warning(
"copilot request policy lite handler unavailable, falling back to main handler",
workflow_permanent_id=workflow_permanent_id,
organization_id=organization_id,
)
return llm_api_handler
@ -269,7 +280,8 @@ class RequestPolicyGuardrailInputs:
chat_history_messages: list[WorkflowCopilotChatHistoryMessage]
global_llm_context: str
organization_id: str
handler: Any
request_policy_handler: Any
turn_intent_handler: LLMAPIHandler | None
previous_user_message: str | None = None
workflow_id: str | None = None
workflow_permanent_id: str | None = None
@ -3647,7 +3659,7 @@ def _build_copilot_input_guardrails(
chat_history=policy_inputs.chat_history_messages,
global_llm_context=policy_inputs.global_llm_context,
organization_id=policy_inputs.organization_id,
handler=policy_inputs.handler,
handler=policy_inputs.request_policy_handler,
active_criteria=_stored_active_completion_criteria(policy_inputs),
config=getattr(ctx, "copilot_config", None) if isinstance(ctx, CopilotContext) else None,
)
@ -3660,7 +3672,7 @@ def _build_copilot_input_guardrails(
chat_history=policy_inputs.chat_history_messages,
global_llm_context=policy_inputs.global_llm_context,
request_policy=policy,
handler=policy_inputs.handler,
handler=policy_inputs.turn_intent_handler,
)
_store_request_policy_on_context(
ctx,
@ -4128,7 +4140,12 @@ async def _run_copilot_turn_impl(
chat_history_messages=list(chat_history),
global_llm_context=safe_global_llm_context,
organization_id=organization_id,
handler=_resolve_request_policy_handler(llm_api_handler),
request_policy_handler=await _resolve_request_policy_handler(
llm_api_handler,
chat_request.workflow_permanent_id,
organization_id,
),
turn_intent_handler=llm_api_handler,
previous_user_message=previous_user_message,
workflow_id=chat_request.workflow_id,
workflow_permanent_id=chat_request.workflow_permanent_id,

2
skyvern/forge/sdk/copilot/enforcement.py
View file

@ -2385,7 +2385,7 @@ def _uncovered_output_reject_rescout_key(canonical_paths: set[str], structural_f
def _active_uncovered_output_reject_paths(ctx: AgentContext) -> set[str]:
canonical = {
_canonical_output_path(path)
for path in author_time_reject_missing_output_paths(getattr(ctx, "latest_recorded_build_test_outcome", None))
for path in author_time_reject_missing_output_paths(ctx.latest_recorded_build_test_outcome)
}
return canonical & uncovered_requested_output_paths(ctx) if canonical else set()

49
skyvern/forge/sdk/copilot/llm_config.py
View file

@ -11,6 +11,20 @@ LOG = structlog.get_logger()
WORKFLOW_COPILOT_PROMPT_TYPE = "workflow-copilot"
WORKFLOW_COPILOT_FAST_PROMPT_TYPE = "workflow-copilot-fast"
WORKFLOW_COPILOT_LITE_PROMPT_TYPE = "workflow-copilot-lite"
def get_workflow_copilot_handler() -> Any | None:
try:
handler = app.WORKFLOW_COPILOT_LLM_API_HANDLER
except (RuntimeError, AttributeError):
handler = None
if handler is not None:
return handler
try:
return app.LLM_API_HANDLER
except (RuntimeError, AttributeError):
return None
def get_main_copilot_handler() -> Any | None:
@ -20,10 +34,7 @@ def get_main_copilot_handler() -> Any | None:
handler = None
if handler is not None:
return handler
try:
return app.LLM_API_HANDLER
except (RuntimeError, AttributeError):
return None
return get_workflow_copilot_handler()
def get_fast_copilot_handler() -> Any | None:
@ -53,6 +64,22 @@ async def resolve_main_copilot_handler(workflow_permanent_id: str | None, organi
return get_main_copilot_handler()
async def resolve_workflow_copilot_handler(
workflow_permanent_id: str | None, organization_id: str | None
) -> Any | None:
if workflow_permanent_id and organization_id:
try:
posthog_handler = await get_llm_handler_for_prompt_type(
WORKFLOW_COPILOT_PROMPT_TYPE, workflow_permanent_id, organization_id
)
except Exception as exc:
LOG.warning("workflow copilot PostHog lookup failed, falling back", error=str(exc))
posthog_handler = None
if posthog_handler is not None:
return posthog_handler
return get_workflow_copilot_handler()
async def resolve_fast_copilot_handler(workflow_permanent_id: str | None, organization_id: str | None) -> Any | None:
if workflow_permanent_id and organization_id:
try:
@ -65,3 +92,17 @@ async def resolve_fast_copilot_handler(workflow_permanent_id: str | None, organi
if posthog_handler is not None:
return posthog_handler
return get_fast_copilot_handler()
async def resolve_lite_copilot_handler(workflow_permanent_id: str | None, organization_id: str | None) -> Any | None:
if workflow_permanent_id and organization_id:
try:
posthog_handler = await get_llm_handler_for_prompt_type(
WORKFLOW_COPILOT_LITE_PROMPT_TYPE, workflow_permanent_id, organization_id
)
except Exception as exc:
LOG.warning("lite copilot PostHog lookup failed, falling back", error=str(exc))
posthog_handler = None
if posthog_handler is not None:
return posthog_handler
return await resolve_workflow_copilot_handler(workflow_permanent_id, organization_id)

2
skyvern/forge/sdk/copilot/tools/workflow_update.py
View file

@ -1574,7 +1574,7 @@ def _recorded_outcome_convergence_reject(
return None
if candidate_signature == latest.authored_structure_signature:
return _ConvergenceReject(candidate_signature, "identical_authored_structure", False)
constraint = ctx.recorded_outcome_binding_constraint
constraint = getattr(ctx, "recorded_outcome_binding_constraint", None)
if not isinstance(constraint, RecordedOutcomeBindingConstraint):
return None
# An author-time reject re-keys `latest` without an executed run, so keep the binding

4
skyvern/forge/sdk/experimentation/llm_prompt_config.py
View file

@ -13,13 +13,13 @@ LOG = structlog.get_logger()
async def get_llm_config_by_prompt_type(distinct_id: str, organization_id: str | None = None) -> dict[str, str] | None:
"""Return PostHog-configured LLM mapping for each prompt type."""
llm_config_experiment = await app.EXPERIMENTATION_PROVIDER.get_value(
llm_config_experiment = await app.EXPERIMENTATION_PROVIDER.get_value_cached(
"LLM_CONFIG_BY_PROMPT_TYPE", distinct_id, properties={"organization_id": organization_id}
)
if llm_config_experiment in (False, "False") or not llm_config_experiment:
return None
payload = await app.EXPERIMENTATION_PROVIDER.get_payload(
payload = await app.EXPERIMENTATION_PROVIDER.get_payload_cached(
"LLM_CONFIG_BY_PROMPT_TYPE", distinct_id, properties={"organization_id": organization_id}
)
if not payload:

90
skyvern/forge/sdk/workflow/code_block_safety.py Normal file
View file

@ -0,0 +1,90 @@
"""Shared static safety checks for CodeBlock Python snippets."""
from __future__ import annotations
import ast
import textwrap
from collections.abc import Callable
from skyvern.forge.sdk.workflow.exceptions import InsecureCodeDetected
# Keep this policy aligned with codeblock/codeblock_safety.py; the runner image carries a local copy.
BLOCKED_ATTRS: frozenset[str] = frozenset(
{
"create_subprocess_exec",
"create_subprocess_shell",
"system",
"popen",
"Popen",
"exec",
"spawn",
"spawnl",
"spawnle",
"spawnlp",
"spawnlpe",
"check_call",
"check_output",
"execv",
"execve",
"execvp",
"execvpe",
"execl",
"execlp",
"execlpe",
"fork",
"open_connection",
"start_server",
"create_connection",
"create_server",
"f_globals",
"f_locals",
"f_builtins",
"f_code",
"co_code",
"co_consts",
"co_names",
"co_varnames",
"gi_frame",
"gi_code",
"cr_frame",
"cr_code",
"tb_frame",
"tb_next",
"mro",
"listdir",
"makedirs",
"rmdir",
"codecs",
"modules",
"builtins",
"stdout",
"stderr",
"stdin",
"getattr",
"setattr",
"delattr",
"globals",
"eval",
"vars",
"format",
"format_map",
}
)
def is_safe_code(
code: str,
*,
error_factory: Callable[[str], Exception] = InsecureCodeDetected,
) -> None:
"""Reject imports, private members, and known escape hatches."""
tree = ast.parse(textwrap.dedent(code))
for node in ast.walk(tree):
if hasattr(node, "attr") and str(node.attr).startswith("__"):
raise error_factory("Not allowed to access private methods or attributes")
if isinstance(node, ast.Name) and node.id.startswith("__"):
raise error_factory("Not allowed to access private methods or attributes")
if isinstance(node, ast.Import | ast.ImportFrom):
raise error_factory("Not allowed to import modules")
if hasattr(node, "attr") and node.attr in BLOCKED_ATTRS:
raise error_factory(f"Not allowed to access '{node.attr}'")

93
skyvern/forge/sdk/workflow/models/block.py
View file

@ -109,6 +109,8 @@ from skyvern.forge.sdk.settings_manager import SettingsManager
from skyvern.forge.sdk.trace import traced
from skyvern.forge.sdk.utils.pdf_parser import extract_pdf_file, render_pdf_pages_as_images, validate_pdf_file
from skyvern.forge.sdk.utils.sanitization import sanitize_postgres_text
from skyvern.forge.sdk.workflow.code_block_safety import BLOCKED_ATTRS as CODE_BLOCK_BLOCKED_ATTRS
from skyvern.forge.sdk.workflow.code_block_safety import is_safe_code as _shared_is_safe_code
from skyvern.forge.sdk.workflow.constants import OUTPUT_PARAMETER_MAX_VALUE_BYTES
from skyvern.forge.sdk.workflow.context_manager import BlockMetadata, WorkflowRunContext
from skyvern.forge.sdk.workflow.exceptions import (
@ -3647,98 +3649,11 @@ class CodeBlock(Block):
prompt: str | None = None
steps: list[CodeBlockStep] | None = None
# Dangerous attribute names that must never be accessed in user code.
# This blocks subprocess creation, OS access, and sandbox-escape primitives.
# NOTE: This is a blocklist-based sandbox, not real process-level isolation.
# It is inherently incomplete — a determined attacker may find bypasses.
# Long-term we should run user code in a proper sandbox. This blocklist is
# a defense-in-depth layer, not a security boundary.
# NOTE: Do not add names that collide with safe module methods (e.g. re.compile).
# Builtin functions like compile(), eval(), exec() are already blocked via __builtins__: {}.
BLOCKED_ATTRS: ClassVar[frozenset[str]] = frozenset(
{
# Subprocess / OS execution
"create_subprocess_exec",
"create_subprocess_shell",
"system",
"popen",
"Popen",
"exec",
"spawn",
"spawnl",
"spawnle",
"spawnlp",
"spawnlpe",
"check_call",
"check_output",
"execv",
"execve",
"execvp",
"execvpe",
"execl",
"execlp",
"execlpe",
"fork",
# Network primitives
"open_connection",
"start_server",
"create_connection",
"create_server",
# Frame / code object internals (classic RestrictedPython escape vectors)
"f_globals",
"f_locals",
"f_builtins",
"f_code",
"co_code",
"co_consts",
"co_names",
"co_varnames",
"gi_frame",
"gi_code",
"cr_frame",
"cr_code",
"tb_frame",
"tb_next",
# Class hierarchy escape
"mro",
# Filesystem operations (unambiguous — these only appear on os/pathlib, not user objects)
"listdir",
"makedirs",
"rmdir",
# Module traversal (json.codecs.sys.modules etc.)
"codecs",
"modules",
"builtins",
"stdout",
"stderr",
"stdin",
# Sandbox-escape helpers (builtin equivalents already blocked via __builtins__: {})
"getattr",
"setattr",
"delattr",
"globals",
"eval",
"vars",
"format",
"format_map",
}
)
BLOCKED_ATTRS: ClassVar[frozenset[str]] = CODE_BLOCK_BLOCKED_ATTRS
@staticmethod
def is_safe_code(code: str) -> None:
tree = ast.parse(code)
for node in ast.walk(tree):
# Block dunder attribute access (obj.__foo__)
if hasattr(node, "attr") and str(node.attr).startswith("__"):
raise InsecureCodeDetected("Not allowed to access private methods or attributes")
# Block bare dunder identifiers (__capture_locals, __builtins__, etc.)
if isinstance(node, ast.Name) and node.id.startswith("__"):
raise InsecureCodeDetected("Not allowed to access private methods or attributes")
if isinstance(node, ast.Import) or isinstance(node, ast.ImportFrom):
raise InsecureCodeDetected("Not allowed to import modules")
# Block dangerous method/attribute access on any object
if hasattr(node, "attr") and node.attr in CodeBlock.BLOCKED_ATTRS:
raise InsecureCodeDetected(f"Not allowed to access '{node.attr}'")
_shared_is_safe_code(code)
@staticmethod
def build_safe_vars() -> dict[str, Any]:

88
tests/unit/test_copilot_agent_helpers.py
View file

@ -10,6 +10,8 @@ from unittest.mock import AsyncMock, MagicMock
import pytest
import yaml
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from structlog.testing import capture_logs
from skyvern.config import settings
@ -1570,11 +1572,6 @@ class TestSupersededAgentIntentGates:
class TestRequestPolicyInputGuardrail:
@pytest.mark.asyncio
async def test_sdk_input_guardrail_computes_and_stores_request_policy(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
policy = RequestPolicy(
testing_intent="skip_test",
credential_input_kind="credential_name",
@ -1591,7 +1588,8 @@ class TestRequestPolicyInputGuardrail:
chat_history_messages=_history(("user", "build the login workflow")),
global_llm_context="",
organization_id="org-1",
handler=object(),
request_policy_handler=object(),
turn_intent_handler=object(),
previous_user_message="build the login workflow",
)
@ -1616,19 +1614,48 @@ class TestRequestPolicyInputGuardrail:
chat_history=policy_inputs.chat_history_messages,
global_llm_context="",
organization_id="org-1",
handler=policy_inputs.handler,
handler=policy_inputs.request_policy_handler,
active_criteria=None,
config=None,
)
@pytest.mark.asyncio
async def test_sdk_input_guardrail_routes_request_policy_to_fast_handler_and_turn_intent_to_main(
self, monkeypatch
) -> None:
request_policy_handler = object()
turn_intent_handler = object()
policy = RequestPolicy()
build_request_policy = AsyncMock(return_value=policy)
classify_turn_intent = AsyncMock(return_value=None)
monkeypatch.setattr(agent_module, "build_request_policy", build_request_policy)
monkeypatch.setattr(agent_module, "classify_turn_intent", classify_turn_intent)
policy_inputs = agent_module.RequestPolicyGuardrailInputs(
user_message="build and test it",
workflow_yaml="workflow: yaml",
chat_history_text="",
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
request_policy_handler=request_policy_handler,
turn_intent_handler=turn_intent_handler,
)
guardrails = agent_module._build_copilot_input_guardrails(
InputGuardrail,
GuardrailFunctionOutput,
policy_inputs=policy_inputs,
)
await guardrails[0].run(SimpleNamespace(), "input", RunContextWrapper(context=_ctx()))
assert build_request_policy.await_args is not None
assert build_request_policy.await_args.kwargs["handler"] is request_policy_handler
assert classify_turn_intent.await_args is not None
assert classify_turn_intent.await_args.kwargs["handler"] is turn_intent_handler
@pytest.mark.asyncio
async def test_sdk_input_guardrail_forwards_stored_active_criteria(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.completion_criteria_store import StoredCriteriaSet, StoredCriteriaSnapshot
from skyvern.forge.sdk.copilot.request_policy import CompletionCriterion, RequestPolicy
stored = StoredCriteriaSet(
set_id="wccs_1",
goal_epoch=1,
@ -1643,7 +1670,8 @@ class TestRequestPolicyInputGuardrail:
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
handler=object(),
request_policy_handler=object(),
turn_intent_handler=object(),
stored_completion_criteria=StoredCriteriaSnapshot(active=stored, next_epoch=2),
)
guardrails = agent_module._build_copilot_input_guardrails(
@ -1658,11 +1686,6 @@ class TestRequestPolicyInputGuardrail:
@pytest.mark.asyncio
async def test_sdk_input_guardrail_trips_after_computing_blocked_policy(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
policy = RequestPolicy(
credential_input_kind="raw_secret",
user_response_policy="ask_clarification",
@ -1684,7 +1707,8 @@ class TestRequestPolicyInputGuardrail:
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
handler=None,
request_policy_handler=None,
turn_intent_handler=None,
),
)
@ -1698,11 +1722,6 @@ class TestRequestPolicyInputGuardrail:
@pytest.mark.asyncio
async def test_request_policy_proceeds_on_workflow_behavior_question(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
policy = RequestPolicy(
credential_input_kind="none",
testing_intent="unspecified",
@ -1726,7 +1745,8 @@ class TestRequestPolicyInputGuardrail:
),
global_llm_context="",
organization_id="org-1",
handler=object(),
request_policy_handler=object(),
turn_intent_handler=object(),
),
)
@ -1741,11 +1761,6 @@ class TestRequestPolicyInputGuardrail:
@pytest.mark.asyncio
async def test_request_policy_proceeds_on_bare_keyvault_slotfill(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
policy = RequestPolicy(
credential_input_kind="none",
testing_intent="unspecified",
@ -1765,7 +1780,8 @@ class TestRequestPolicyInputGuardrail:
),
global_llm_context="",
organization_id="org-1",
handler=object(),
request_policy_handler=object(),
turn_intent_handler=object(),
),
)
@ -1779,11 +1795,6 @@ class TestRequestPolicyInputGuardrail:
@pytest.mark.asyncio
async def test_raw_secret_redacted_draft_policy_sanitizes_agent_input(self, monkeypatch) -> None:
from agents import GuardrailFunctionOutput, InputGuardrail
from agents.run_context import RunContextWrapper
from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
raw_message = (
"Convert this SDK snippet into a workflow:\n"
"client = DemoClient(api_key='sk-abcdefghijklmnopqrstuvwxyz1234567890')"
@ -1808,7 +1819,8 @@ class TestRequestPolicyInputGuardrail:
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
handler=None,
request_policy_handler=None,
turn_intent_handler=None,
),
)

3
tests/unit/test_copilot_completion_criteria_lifecycle.py
View file

@ -1064,7 +1064,8 @@ def _policy_inputs(snapshot: StoredCriteriaSnapshot | None) -> RequestPolicyGuar
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
handler=None,
request_policy_handler=None,
turn_intent_handler=None,
stored_completion_criteria=snapshot,
)

2
tests/unit/test_copilot_output_contract_actuation.py
View file

@ -521,6 +521,8 @@ def _advisory_ctx() -> SimpleNamespace:
output_contract_run_bound_required_path_by_signature={},
output_contract_bail_actuated_this_call=False,
author_time_gate_ablation_events=[],
latest_recorded_build_test_outcome=None,
recorded_build_test_outcome_history=[],
scouted_output_covered_paths=set(),
composition_page_evidence=None,
)

3
tests/unit/test_copilot_turn_intent.py
View file

@ -872,7 +872,8 @@ def test_store_request_policy_attaches_classified_turn_intent_to_context() -> No
chat_history_messages=[],
global_llm_context="",
organization_id="org-1",
handler=None,
request_policy_handler=None,
turn_intent_handler=None,
previous_user_message=None,
)

22
tests/unit/test_llm_prompt_config.py
View file

@ -8,6 +8,7 @@ Verified contracts:
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
@ -43,6 +44,27 @@ def _make_log_recorder() -> tuple[MagicMock, list[tuple[str, dict[str, Any]]]]:
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_prompt_type_config_uses_cached_posthog_methods(monkeypatch: pytest.MonkeyPatch) -> None:
provider = SimpleNamespace(
get_value_cached=AsyncMock(return_value="enabled"),
get_payload_cached=AsyncMock(return_value='{"workflow-copilot-lite": "OPENAI_GPT5_4_MINI"}'),
get_value=AsyncMock(side_effect=AssertionError("must use cached value lookup")),
get_payload=AsyncMock(side_effect=AssertionError("must use cached payload lookup")),
)
monkeypatch.setattr(module, "app", SimpleNamespace(EXPERIMENTATION_PROVIDER=provider))
config = await module.get_llm_config_by_prompt_type("wpid_1", "org_1")
assert config == {"workflow-copilot-lite": "OPENAI_GPT5_4_MINI"}
provider.get_value_cached.assert_awaited_once_with(
"LLM_CONFIG_BY_PROMPT_TYPE", "wpid_1", properties={"organization_id": "org_1"}
)
provider.get_payload_cached.assert_awaited_once_with(
"LLM_CONFIG_BY_PROMPT_TYPE", "wpid_1", properties={"organization_id": "org_1"}
)
@pytest.mark.asyncio
async def test_no_config_returns_none_without_warning(monkeypatch: pytest.MonkeyPatch) -> None:
"""Flag disabled → config is None → return None, no LOG.warning."""

139
tests/unit/test_workflow_copilot_llm_config.py
View file

@ -1,8 +1,9 @@
"""Tests for the workflow-copilot v2 LLM key wiring (SKY-10642).
Two optional settings give operators independent control over (a) the main
Copilot reasoning/guardrail/evidence lane and (b) the fast-consumer lane:
``WORKFLOW_COPILOT_AGENT_LLM_KEY`` and ``WORKFLOW_COPILOT_FAST_LLM_KEY``.
Optional settings give operators independent control over the main Copilot lane,
agent-specific lane, and fast-consumer lane:
``WORKFLOW_COPILOT_LLM_KEY``, ``WORKFLOW_COPILOT_AGENT_LLM_KEY``, and
``WORKFLOW_COPILOT_FAST_LLM_KEY``.
These tests cover the public contract: defaults, fallback chains, and
PostHog → env-specific → default resolution order.
"""
@ -27,10 +28,11 @@ from skyvern.forge.sdk.routes import workflow_copilot as workflow_copilot_route
def test_workflow_copilot_agent_llm_key_default_is_none() -> None:
assert Settings.model_fields["WORKFLOW_COPILOT_LLM_KEY"].default is None
assert Settings.model_fields["WORKFLOW_COPILOT_AGENT_LLM_KEY"].default is None
def test_workflow_copilot_narration_llm_key_default_is_none() -> None:
def test_workflow_copilot_fast_llm_key_default_is_none() -> None:
assert Settings.model_fields["WORKFLOW_COPILOT_FAST_LLM_KEY"].default is None
@ -131,6 +133,7 @@ async def test_resolve_main_copilot_handler_posthog_override_wins(monkeypatch: p
"app",
SimpleNamespace(
WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER=dedicated,
WORKFLOW_COPILOT_LLM_API_HANDLER=primary,
LLM_API_HANDLER=primary,
),
)
@ -161,6 +164,31 @@ async def test_resolve_main_copilot_handler_falls_back_to_dedicated(monkeypatch:
assert handler is dedicated
@pytest.mark.asyncio
async def test_resolve_main_copilot_handler_falls_back_to_workflow_copilot_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
workflow_copilot = object()
primary = object()
async def _posthog_lookup(*_args: object, **_kwargs: object) -> None:
return None
monkeypatch.setattr(copilot_llm_config, "get_llm_handler_for_prompt_type", _posthog_lookup)
monkeypatch.setattr(
copilot_llm_config,
"app",
SimpleNamespace(
WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER=None,
WORKFLOW_COPILOT_LLM_API_HANDLER=workflow_copilot,
LLM_API_HANDLER=primary,
),
)
handler = await copilot_llm_config.resolve_main_copilot_handler("wpid_1", "org_1")
assert handler is workflow_copilot
@pytest.mark.asyncio
@pytest.mark.parametrize(
"make_app",
@ -173,7 +201,8 @@ async def test_resolve_main_copilot_handler_falls_back_to_dedicated(monkeypatch:
pytest.param(
lambda primary: SimpleNamespace(
WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER=None,
LLM_API_HANDLER=primary,
WORKFLOW_COPILOT_LLM_API_HANDLER=primary,
LLM_API_HANDLER=object(),
),
id="dedicated_is_none",
),
@ -287,14 +316,110 @@ async def test_resolve_narrator_handler_skips_posthog_when_ids_missing(monkeypat
assert posthog_called is False
# ---------------------------------------------------------------------------
# lite Copilot handler fallback chain
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_resolve_lite_copilot_handler_posthog_override_wins(monkeypatch: pytest.MonkeyPatch) -> None:
posthog_handler = object()
main_handler = object()
async def _posthog_lookup(prompt_type: str, *_args: object, **_kwargs: object) -> object:
assert prompt_type == "workflow-copilot-lite"
return posthog_handler
async def _main_lookup(*_args: object, **_kwargs: object) -> object:
return main_handler
monkeypatch.setattr(copilot_llm_config, "get_llm_handler_for_prompt_type", _posthog_lookup)
monkeypatch.setattr(copilot_llm_config, "resolve_main_copilot_handler", _main_lookup)
handler = await copilot_llm_config.resolve_lite_copilot_handler("wpid_1", "org_1")
assert handler is posthog_handler
@pytest.mark.asyncio
async def test_resolve_lite_copilot_handler_falls_back_to_workflow_copilot(
monkeypatch: pytest.MonkeyPatch,
) -> None:
workflow_handler = object()
async def _posthog_lookup(*_args: object, **_kwargs: object) -> None:
return None
async def _workflow_lookup(workflow_permanent_id: str | None, organization_id: str | None) -> object:
assert workflow_permanent_id == "wpid_1"
assert organization_id == "org_1"
return workflow_handler
monkeypatch.setattr(copilot_llm_config, "get_llm_handler_for_prompt_type", _posthog_lookup)
monkeypatch.setattr(copilot_llm_config, "resolve_workflow_copilot_handler", _workflow_lookup)
handler = await copilot_llm_config.resolve_lite_copilot_handler("wpid_1", "org_1")
assert handler is workflow_handler
@pytest.mark.asyncio
async def test_resolve_lite_copilot_handler_fallback_ignores_agent_specific_key(
monkeypatch: pytest.MonkeyPatch,
) -> None:
workflow_copilot = object()
agent = object()
async def _posthog_lookup(*_args: object, **_kwargs: object) -> None:
return None
monkeypatch.setattr(copilot_llm_config, "get_llm_handler_for_prompt_type", _posthog_lookup)
monkeypatch.setattr(
copilot_llm_config,
"app",
SimpleNamespace(
WORKFLOW_COPILOT_LLM_API_HANDLER=workflow_copilot,
WORKFLOW_COPILOT_AGENT_LLM_API_HANDLER=agent,
LLM_API_HANDLER=object(),
),
)
handler = await copilot_llm_config.resolve_lite_copilot_handler("wpid_1", "org_1")
assert handler is workflow_copilot
# ---------------------------------------------------------------------------
# non-narration Copilot helpers use the main lane
# ---------------------------------------------------------------------------
def test_resolve_request_policy_handler_uses_main_copilot_handler() -> None:
@pytest.mark.asyncio
async def test_resolve_request_policy_handler_uses_lite_copilot_handler(monkeypatch: pytest.MonkeyPatch) -> None:
main_handler = object()
assert copilot_agent._resolve_request_policy_handler(main_handler) is main_handler
lite_handler = object()
async def _lite_lookup(workflow_permanent_id: str | None, organization_id: str | None) -> object:
assert workflow_permanent_id == "wpid_1"
assert organization_id == "org_1"
return lite_handler
monkeypatch.setattr(copilot_agent.llm_config, "resolve_lite_copilot_handler", _lite_lookup)
handler = await copilot_agent._resolve_request_policy_handler(main_handler, "wpid_1", "org_1")
assert handler is lite_handler
@pytest.mark.asyncio
async def test_resolve_request_policy_handler_falls_back_to_main_handler(monkeypatch: pytest.MonkeyPatch) -> None:
main_handler = object()
async def _lite_lookup(workflow_permanent_id: str | None, organization_id: str | None) -> object | None:
assert workflow_permanent_id == "wpid_1"
assert organization_id == "org_1"
return None
monkeypatch.setattr(copilot_agent.llm_config, "resolve_lite_copilot_handler", _lite_lookup)
handler = await copilot_agent._resolve_request_policy_handler(main_handler, "wpid_1", "org_1")
assert handler is main_handler
@pytest.mark.asyncio