fix: resolve all outstanding lint, type, and viewer extra issues

lyrashield_adapter/cli.py

@@ -5,9 +5,16 @@
 import os
 import sys
 from importlib.metadata import PackageNotFoundError, version
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 
+try:
+    from dotenv import load_dotenv
+except ImportError:
+    load_dotenv = None  # type: ignore[assignment]
+
+
 if TYPE_CHECKING:
     from collections.abc import MutableMapping
 
@@ -42,10 +49,18 @@
 }
 
 
+_STALE_EMPTY_ENV_VARS = ("LLM_API_KEY", "LLM_API_BASE", "LLM_API_VERSION")
+
+
 def prepare_environment(
     environ: MutableMapping[str, str] | None = None,
 ) -> MutableMapping[str, str]:
     env = environ if environ is not None else os.environ
+    # Remove stale empty generic LLM env vars that would shadow Azure-specific
+    # aliases via pydantic AliasChoices priority (first match wins, even if empty).
+    for name in _STALE_EMPTY_ENV_VARS:
+        if env.get(name, "").strip() == "":
+            env.pop(name, None)
     for product_name, upstream_name in ENV_ALIASES.items():
         if upstream_name not in env and product_name in env:
             env[upstream_name] = env[product_name]
@@ -86,6 +101,8 @@ def _run_upstream() -> None:
 
 
 def main() -> None:
+    if load_dotenv is not None:
+        load_dotenv(Path(__file__).resolve().parents[1] / ".env", override=True)
     prepare_environment()
     if sys.argv[1:] in (["--version"], ["-v"]):
         print(f"lyrashield {get_version()}")  # noqa: T201

strix/config/settings.py

@@ -3,9 +3,9 @@
 
 from __future__ import annotations
 
-from typing import Literal
+from typing import Any, Literal
 
-from pydantic import AliasChoices, Field
+from pydantic import AliasChoices, Field, field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -104,6 +104,11 @@ class LlmSettings(BaseSettings):
         validation_alias=_lyra("STRIX_MAX_INPUT_TOKENS"),
     )
 
+    @field_validator("api_base", "api_key", "api_version", mode="before")
+    @classmethod
+    def _empty_env_to_none(cls, value: Any) -> Any:
+        return None if value == "" else value
+
 
 class DedupeSettings(BaseSettings):
     model_config = _BASE_CONFIG

strix/core/inputs.py

@@ -20,30 +20,38 @@
 
 
 if TYPE_CHECKING:
+    from openai.types.responses.response_create_params import PromptCacheOptions
+
     from strix.config.settings import ReasoningEffort
 
 
 DEFAULT_MAX_TURNS = 500
 MAX_CHILD_INHERITED_HISTORY_BYTES = 24 * 1024
 
-# Opt-in/opt-out for explicit prompt-cache breakpoints. "auto" (default) enables
-# breakpoints for known GPT-5.6 deployments; "1"/"true" forces them on; "0"/"false"
-# forces them off. Provider support should be confirmed with a smoke scan.
-_PROMPT_CACHE_BREAKPOINT_ENV = "LYRASHIELD_PROMPT_CACHE_BREAKPOINTS"
+# Gate for using explicit prompt-cache *options* together with breakpoints.
+# Enabling this tells the provider to honor ``prompt_cache_breakpoint`` content parts
+# and cache the prefix before each breakpoint. This is opt-in until a smoke scan proves
+# the target deployment supports it.
+_PROMPT_CACHE_EXPLICIT_ENV = "LYRASHIELD_PROMPT_CACHE_EXPLICIT"
 
 
-def _prompt_cache_breakpoints_enabled(model_name: str | None) -> bool:
-    """Return whether the model/provider pair should emit explicit cache breakpoints.
+def _prompt_cache_explicit_enabled(model_name: str | None) -> bool:
+    """Return whether to use ``prompt_cache_options: {mode: 'explicit'}``.
 
-    Defaults to the known GPT-5.6 allowlist (OpenAI/Azure AI) and can be overridden
-    with ``LYRASHIELD_PROMPT_CACHE_BREAKPOINTS``.
+    This is off by default regardless of model; turn it on by setting
+    ``LYRASHIELD_PROMPT_CACHE_EXPLICIT=1`` for the deployment being used.
     """
-    env = os.environ.get(_PROMPT_CACHE_BREAKPOINT_ENV, "").strip().lower()
-    if env in ("1", "true", "yes"):
-        return True
+    env = os.environ.get(_PROMPT_CACHE_EXPLICIT_ENV, "").strip().lower()
     if env in ("0", "false", "no"):
         return False
-    return is_gpt56_model(model_name)
+    return env in ("1", "true", "yes") and is_gpt56_model(model_name)
+
+
+def prompt_cache_options_for_model(model_name: str | None) -> PromptCacheOptions | None:
+    """Return explicit prompt-cache options for a model, or None if disabled."""
+    if not _prompt_cache_explicit_enabled(model_name):
+        return None
+    return {"mode": "explicit", "ttl": "30m"}
 
 
 def _accepts_required_tool_choice(model_name: str | None) -> bool:
@@ -161,26 +169,27 @@ def build_root_initial_input(
     For models that support explicit prompt-cache breakpoints, split the stable
     target/scope prefix from the variable per-scan instructions and mark the
     boundary. This lets the provider cache the prefix across turns.
+
+    The explicit breakpoint path is gated by ``LYRASHIELD_PROMPT_CACHE_EXPLICIT``
+    because it must be paired with ``prompt_cache_options`` in ``ModelSettings``;
+    emitting breakpoints without that option disables caching.
     """
     parts, user_instructions = _build_root_task_parts(scan_config)
     stable = " ".join(parts).strip()
 
-    if not _prompt_cache_breakpoints_enabled(model_name) or not user_instructions:
-        # No breakpoint needed when there is no variable suffix to separate.
-        return build_root_task(scan_config)
-
-    variable = f"Special instructions: {user_instructions}"
-    if not stable:
+    if not _prompt_cache_explicit_enabled(model_name) or not stable:
+        # No breakpoint needed or explicit caching not enabled.
         return build_root_task(scan_config)
 
     content: list[dict[str, Any]] = [
         {
             "type": "input_text",
             "text": stable,
             "prompt_cache_breakpoint": {"mode": "explicit"},
-        },
-        {"type": "input_text", "text": variable},
+        }
     ]
+    if user_instructions:
+        content.append({"type": "input_text", "text": f"Special instructions: {user_instructions}"})
     return [{"role": "user", "content": content}]
 
 
@@ -222,6 +231,7 @@ def make_model_settings(
     request_timeout: float | None = None,
     max_output_tokens: int | None = None,
     prompt_cache_key: str | None = None,
+    prompt_cache_options: PromptCacheOptions | None = None,
 ) -> ModelSettings:
     extra_args: dict[str, Any] = request_timeout_extra_args(request_timeout) or {}
     if prompt_cache_key:
@@ -232,6 +242,7 @@ def make_model_settings(
         include_usage=True,
         max_tokens=max_output_tokens,
         extra_args=extra_args or None,
+        prompt_cache_options=prompt_cache_options,
     )
     if (
         reasoning_effort is not None

strix/core/runner.py

@@ -12,6 +12,7 @@
 from typing import TYPE_CHECKING, Any, cast
 
 from agents import RunConfig
+from agents.exceptions import ModelBehaviorError
 from agents.sandbox import SandboxRunConfig
 from openai import RateLimitError
 
@@ -34,10 +35,12 @@
 from strix.core.hooks import BudgetExceededError, ReportUsageHooks, set_active_hooks
 from strix.core.inputs import (
     DEFAULT_MAX_TURNS,
+    _prompt_cache_explicit_enabled,
     build_root_initial_input,
     build_root_task,
     build_scope_context,
     make_model_settings,
+    prompt_cache_options_for_model,
 )
 from strix.core.paths import run_dir_for, runtime_state_dir
 from strix.core.sessions import open_agent_session
@@ -290,6 +293,7 @@ async def run_strix_scan(
             request_timeout=settings.llm.timeout,
             max_output_tokens=max_output_tokens,
             prompt_cache_key=f"lyrashield:{scan_id}:coordinator",
+            prompt_cache_options=prompt_cache_options_for_model(resolved_model),
         )
         delegate_max_output_tokens = min(max_output_tokens, DELEGATE_OUTPUT_TOKEN_CEILING)
         delegate_model_settings = make_model_settings(
@@ -462,20 +466,68 @@ async def spawn_child_agent(**kwargs: Any) -> dict[str, Any]:
 
         root_status = await coordinator.get_status(root_id)
 
-        result = await run_agent_loop(
-            agent=root_agent,
-            initial_input=initial_input,
-            run_config=run_config,
-            context=context,
-            max_turns=max_turns,
-            coordinator=coordinator,
-            agent_id=root_id,
-            interactive=interactive,
-            session=root_session,
-            start_parked=bool(interactive and is_resume and root_status != "running"),
-            event_sink=event_sink,
-            hooks=hooks,
-        )
+        try:
+            result = await run_agent_loop(
+                agent=root_agent,
+                initial_input=initial_input,
+                run_config=run_config,
+                context=context,
+                max_turns=max_turns,
+                coordinator=coordinator,
+                agent_id=root_id,
+                interactive=interactive,
+                session=root_session,
+                start_parked=bool(interactive and is_resume and root_status != "running"),
+                event_sink=event_sink,
+                hooks=hooks,
+            )
+        except ModelBehaviorError as exc:
+            if "content_filter" not in str(exc) or not _prompt_cache_explicit_enabled(
+                resolved_model
+            ):
+                raise
+            logger.warning(
+                "Scan %s hit content_filter with explicit prompt caching; "
+                "falling back to implicit caching and retrying.",
+                scan_id,
+            )
+            initial_input = build_root_task(scan_config)
+            model_settings = make_model_settings(
+                settings.llm.reasoning_effort,
+                model_name=resolved_model,
+                force_required_tool_choice=settings.llm.force_required_tool_choice,
+                request_timeout=settings.llm.timeout,
+                max_output_tokens=max_output_tokens,
+                prompt_cache_key=f"lyrashield:{scan_id}:coordinator",
+                prompt_cache_options=None,
+            )
+            root_agent = build_strix_agent(
+                name="Strix",
+                skills=skills,
+                is_root=True,
+                scan_mode=scan_mode,
+                is_whitebox=is_whitebox,
+                interactive=interactive,
+                chat_completions_tools=chat_completions_tools,
+                system_prompt_context=root_context,
+                instructions_override=root_instructions,
+                model=resolved_model,
+                model_settings=model_settings,
+            )
+            result = await run_agent_loop(
+                agent=root_agent,
+                initial_input=initial_input,
+                run_config=run_config,
+                context=context,
+                max_turns=max_turns,
+                coordinator=coordinator,
+                agent_id=root_id,
+                interactive=interactive,
+                session=root_session,
+                start_parked=bool(interactive and is_resume and root_status != "running"),
+                event_sink=event_sink,
+                hooks=hooks,
+            )
         if not interactive and result is not None:
             final = getattr(result, "final_output", None)
             scan_completed = False

strix/interface/main.py

@@ -1093,6 +1093,14 @@ def pull_docker_image() -> None:
 
 
 def main() -> None:
+    # Auto-load the engine .env if present; explicit shell exports still win.
+    try:
+        from dotenv import load_dotenv
+
+        load_dotenv(Path(__file__).resolve().parents[2] / ".env", override=True)
+    except Exception:
+        logger.debug("Could not load .env file; continuing without it", exc_info=True)
+
     configure_dependency_logging()
 
     if sys.platform == "win32":

tests/test_fenced_code.py

@@ -2,6 +2,9 @@
 
 from __future__ import annotations
 
+import importlib.util
+
+import pytest
 from pygments.lexers import BashLexer, PythonLexer
 
 from strix.report.writer import (
@@ -10,7 +13,9 @@
     resolve_lexer,
     safe_fence,
 )
-from strix.viewer.report_pdf import _strip_code_fence
+
+
+_viewer_extra_available = importlib.util.find_spec("pypdf") is not None
 
 
 def test_parse_fenced_code_extracts_language_and_body() -> None:
@@ -43,11 +48,23 @@ def test_parse_fenced_code_fence_without_language() -> None:
     assert code == "plain"
 
 
+@pytest.mark.skipif(
+    not _viewer_extra_available,
+    reason="requires the optional 'viewer' extra",
+)
 def test_strip_code_fence_removes_fence() -> None:
+    from strix.viewer.report_pdf import _strip_code_fence  # noqa: PLC0415
+
     assert _strip_code_fence("```python\nx = 1\n```") == "x = 1"
 
 
+@pytest.mark.skipif(
+    not _viewer_extra_available,
+    reason="requires the optional 'viewer' extra",
+)
 def test_strip_code_fence_passes_through_non_string_and_unfenced() -> None:
+    from strix.viewer.report_pdf import _strip_code_fence  # noqa: PLC0415
+
     assert _strip_code_fence(None) is None
     assert _strip_code_fence("x = 1") == "x = 1"