From 375389b8bc445d8ae0ae0a4f46143f9507cedc57 Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 00:01:05 -0700 Subject: [PATCH] =?UTF-8?q?feat(migration):=20phase=201=20=E2=80=94=20Sess?= =?UTF-8?q?ion=20+=20Tracer=20+=20RunConfig=20factory?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three foundation modules per PLAYBOOK §2.8 / §2.9 / §2.10 with all relevant R2/R3 corrections (C7, C10, C11, C16, C21): strix/llm/strix_session.py SessionABC wrapper around the legacy MemoryCompressor; on any compression failure, returns uncompressed history and permanently disables compression for the rest of the run (C10 + Round 3.4 W5/E2). strix/telemetry/strix_processor.py SDK TracingProcessor that writes events.jsonl in our schema. All hooks SYNC per ABC (F3); writes protected by per-path threading.Lock (C7); OSError swallowed and logged (C16); PII scrubbed via the existing TelemetrySanitizer. strix/run_config_factory.py make_run_config() with our defaults: parallel_tool_calls= False (C1 Phase-1 safe default), retry policy explicitly excludes 401/403/400 (C11), reasoning effort + model_settings_override merge path (C21). make_agent_context() returns the canonical per-agent dict including is_whitebox/diff_scope/ run_id (C21). 32 new smoke tests (197/197 total). mypy strict + ruff clean. Per-file ignores added for tests/** S105/PT018 and for the two new src modules' intentional broad-Exception catches (BLE001). Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 9 +- strix/llm/strix_session.py | 106 ++++++++++++ strix/run_config_factory.py | 206 ++++++++++++++++++++++++ strix/telemetry/strix_processor.py | 183 +++++++++++++++++++++ tests/llm/test_strix_session.py | 153 ++++++++++++++++++ tests/telemetry/test_strix_processor.py | 201 +++++++++++++++++++++++ tests/test_run_config_factory.py | 181 +++++++++++++++++++++ 7 files changed, 1038 insertions(+), 1 deletion(-) create mode 100644 strix/llm/strix_session.py create mode 100644 strix/run_config_factory.py create mode 100644 strix/telemetry/strix_processor.py create mode 100644 tests/llm/test_strix_session.py create mode 100644 tests/telemetry/test_strix_processor.py create mode 100644 tests/test_run_config_factory.py diff --git a/pyproject.toml b/pyproject.toml index 59a51ef..9dbcaf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -248,11 +248,13 @@ ignore = [ "strix/llm/llm.py" = ["PLC0415"] "strix/tools/notes/notes_actions.py" = ["PLC0415"] "tests/**/*.py" = [ - "S106", # Possible hardcoded password + "S105", # Possible hardcoded password (string literal) + "S106", # Possible hardcoded password (function call) "S108", # Possible insecure usage of temporary file/directory "ARG001", # Unused function argument "ARG002", # Unused method argument (test helpers / fakes mirror SDK signatures) "PLR2004", # Magic value used in comparison + "PT018", # Multi-part assertions are a pytest style preference "TC002", # Type-only third-party import (tests import for runtime instantiation) "TC003", # Type-only stdlib import ] @@ -281,6 +283,11 @@ ignore = [ "strix/tools/_decorator.py" = [ "TC002", # FunctionTool imported for annotation ] +# StrixSession + StrixTracingProcessor catch broad Exception intentionally: +# the whole point is that compressor / sanitizer / disk failures must not +# tear down the agent run (C10, C16). Calls already log at exception level. +"strix/llm/strix_session.py" = ["BLE001"] +"strix/telemetry/strix_processor.py" = ["BLE001"] [tool.ruff.lint.isort] force-single-line = false diff --git a/strix/llm/strix_session.py b/strix/llm/strix_session.py new file mode 100644 index 0000000..6507437 --- /dev/null +++ b/strix/llm/strix_session.py @@ -0,0 +1,106 @@ +"""StrixSession — Session wrapper that runs the legacy MemoryCompressor. + +The SDK's `Session` (and ``SessionABC``) protocol owns conversation history +storage. We delegate the actual storage to any underlying session +implementation (in-memory, SQLite, Redis, …) and intercept ``get_items`` so +the legacy ``MemoryCompressor`` runs before the model sees the history. + +Why wrap rather than reimplement: +- ``MemoryCompressor`` already encodes the pentest-tuned summarization + prompt and the 90K-token budget that Strix has been tuning for months. + Reimplementing inside a Session would lose that institutional knowledge. +- The SDK gives us a clean seam in ``get_items``: it's the last call before + ``call_model_input_filter`` runs, so compressing here means the filter + sees a compressed history too. + +References: + - PLAYBOOK.md §2.8 + - AUDIT_R2.md §1.5 (C10 — compressor exception → uncompressed fallback) + - AUDIT_R3.md §3 row W5/E2 — once compression has failed, set a flag and + skip future attempts so we don't infinite-loop on a permanently broken + compressor while the agent loop slowly drowns in context. +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, cast + +from agents.memory.session import SessionABC + + +if TYPE_CHECKING: + from agents.items import TResponseInputItem + + from strix.llm.memory_compressor import MemoryCompressor + + +logger = logging.getLogger(__name__) + + +class StrixSession(SessionABC): + """Wraps an underlying ``SessionABC`` with Strix's memory compressor. + + The wrapped session owns persistence; ``StrixSession`` only intercepts + ``get_items`` to run compression. Writes (``add_items``, ``pop_item``, + ``clear_session``) pass through verbatim. + + On compressor failure, the call returns the uncompressed history and + a per-instance flag is set so subsequent ``get_items`` calls skip the + compressor entirely. This avoids an infinite "compress → fail → grow" + loop when the compressor LLM is itself unavailable. + """ + + def __init__( + self, + underlying: SessionABC, + compressor: MemoryCompressor, + ) -> None: + self._underlying = underlying + self._compressor = compressor + self._compression_disabled = False + # ``SessionABC.session_id`` is a plain ``str`` field; pass through. + self.session_id: str = getattr(underlying, "session_id", "strix-session") + self.session_settings = getattr(underlying, "session_settings", None) + + @property + def compression_disabled(self) -> bool: + """True after the compressor has failed at least once on this session.""" + return self._compression_disabled + + async def get_items( + self, + limit: int | None = None, + ) -> list[TResponseInputItem]: + """Read items from underlying storage and (optionally) compress. + + On any compressor exception, log and return the uncompressed list. + Set ``_compression_disabled`` so the next call short-circuits. + """ + items = await self._underlying.get_items(limit=limit) + if self._compression_disabled or not items: + return items + try: + # Compressor expects ``list[dict[str, Any]]``; SDK's + # ``TResponseInputItem`` is a TypedDict union — structurally + # compatible. Compressor mutates content but preserves shape. + compressed = self._compressor.compress_history( + cast("list[dict[str, Any]]", items), + ) + return cast("list[TResponseInputItem]", compressed) + except Exception: + logger.exception( + "MemoryCompressor failed; returning uncompressed history. " + "Compression disabled for this session for the rest of the run.", + ) + self._compression_disabled = True + return items + + async def add_items(self, items: list[TResponseInputItem]) -> None: + await self._underlying.add_items(items) + + async def pop_item(self) -> TResponseInputItem | None: + return await self._underlying.pop_item() + + async def clear_session(self) -> None: + await self._underlying.clear_session() diff --git a/strix/run_config_factory.py b/strix/run_config_factory.py new file mode 100644 index 0000000..fec7562 --- /dev/null +++ b/strix/run_config_factory.py @@ -0,0 +1,206 @@ +"""make_run_config — assemble a Strix-flavored ``RunConfig`` for ``Runner.run``. + +Factory pattern: every Strix scan goes through here so the defaults are +applied uniformly. Per-call overrides are accepted via ``model_settings_override`` +for the rare case a single run wants different reasoning effort or +``tool_choice`` (C21). + +References: + - PLAYBOOK.md §2.10 + - AUDIT.md §2.1 (C1 — parallel_tool_calls=False until Phase 6 relaxes the + legacy tool server's per-agent task slot serialization) + - AUDIT_R2.md §1.6 (C11 — retry policy explicitly excludes 401/403/400; + auth and validation errors must fail fast, not waste retries) + - AUDIT_R3.md C21 — RunConfig override + context fields including + ``is_whitebox``, ``diff_scope``, ``run_id`` +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Literal + +from agents import RunConfig +from agents.model_settings import ModelSettings +from agents.retry import ( + ModelRetryBackoffSettings, + ModelRetrySettings, + retry_policies, +) +from agents.sandbox import SandboxRunConfig + +from strix.llm.multi_provider_setup import build_multi_provider +from strix.orchestration.filter import inject_messages_filter + + +if TYPE_CHECKING: + from agents.sandbox.session.base_sandbox_session import BaseSandboxSession + + from strix.orchestration.bus import AgentMessageBus + + +# Phase 1-5 default. Phase 6 relaxes the legacy tool server's per-agent +# task-slot serialization (``runtime/tool_server.py:94-97``) and flips this +# to ``True`` after the multi-agent stress tests confirm safety. +_PHASE1_PARALLEL_DEFAULT = False + +# Default retry policy. Explicitly does NOT include 401/403/400 — those are +# auth and validation errors that retrying cannot fix; they should fail fast +# so the user sees the real error within seconds. 429/5xx is the right set. +_RETRYABLE_HTTP_STATUSES = (429, 500, 502, 503, 504) + +# Default retry budget. Mirrors the legacy ``llm.py`` retry loop: 5 attempts +# with ``min(90, 2*2^n)`` backoff. +_DEFAULT_MAX_RETRIES = 5 +_DEFAULT_BACKOFF = ModelRetryBackoffSettings( + initial_delay=2.0, + max_delay=90.0, + multiplier=2.0, + jitter=False, +) + + +def _default_retry_policy() -> Any: + """Build the default retry policy. + + Built from ``retry_policies.any(...)``: any of the listed conditions + triggers a retry. ``provider_suggested`` honors server-sent + ``Retry-After`` hints; ``network_error`` covers connection / timeout; + ``http_status`` whitelists transient HTTP codes. + """ + return retry_policies.any( + retry_policies.provider_suggested(), + retry_policies.network_error(), + retry_policies.http_status(_RETRYABLE_HTTP_STATUSES), + ) + + +#: Default ``max_turns`` callers should pass to ``Runner.run``. +#: Mirrors the legacy ``AgentState.max_iterations = 300`` +#: (``HARNESS_WIKI.md §5.2``). +STRIX_DEFAULT_MAX_TURNS = 300 + + +def make_run_config( + *, + sandbox_session: BaseSandboxSession | None, + model: str = "strix/claude-sonnet-4.6", + parallel_tool_calls: bool = _PHASE1_PARALLEL_DEFAULT, + tool_choice: Literal["auto", "required", "none"] | None = "required", + reasoning_effort: Literal["low", "medium", "high"] | None = None, + model_settings_override: ModelSettings | None = None, + sandbox_client: Any | None = None, +) -> RunConfig: + """Build a ``RunConfig`` with Strix defaults. + + Note: ``max_turns`` and ``isolate_parallel_failures`` are NOT + ``RunConfig`` fields — they are passed directly to ``Runner.run``. + Use ``STRIX_DEFAULT_MAX_TURNS`` for the budget; pass + ``isolate_parallel_failures=False`` to ``Runner.run`` if Phase 6 has + not yet flipped ``parallel_tool_calls=True``. + + Args: + sandbox_session: Live sandbox session shared by every agent in this + scan (one container per scan; see ``strix.sandbox.session_manager``). + ``None`` is allowed for unit tests and dry runs. + model: Model alias to pass to ``MultiProvider``. Defaults to the + current production-favored Anthropic alias. + parallel_tool_calls: Phase 1 default is ``False`` to keep behavior + sequential per the legacy tool server's slot serialization (C1). + tool_choice: Forces tool use per turn unless explicitly relaxed. + Mirrors the legacy ``4f90a56`` prompt hardening at the model + level. Pass ``None`` to omit. + reasoning_effort: ``"low" | "medium" | "high"``; routes to + ``ModelSettings.reasoning``. ``None`` defers to provider default. + model_settings_override: Optional ``ModelSettings`` to merge over + the factory defaults (C21 — per-run override path). + sandbox_client: Optional pre-built sandbox client (e.g., the Strix + Docker subclass). Defaults to ``None``; the SDK will instantiate + its built-in if a session is supplied without a client. + + Returns: + A ``RunConfig`` ready to pass to ``Runner.run``. + """ + base_settings = ModelSettings( + parallel_tool_calls=parallel_tool_calls, + tool_choice=tool_choice, + retry=ModelRetrySettings( + max_retries=_DEFAULT_MAX_RETRIES, + backoff=_DEFAULT_BACKOFF, + policy=_default_retry_policy(), + ), + ) + if reasoning_effort is not None: + from openai.types.shared import Reasoning + + base_settings = base_settings.resolve( + ModelSettings(reasoning=Reasoning(effort=reasoning_effort)), + ) + if model_settings_override is not None: + # ``ModelSettings.resolve`` merges another ModelSettings into self + # with override-wins semantics — exactly what we want. + base_settings = base_settings.resolve(model_settings_override) + + sandbox_config = ( + SandboxRunConfig(client=sandbox_client, session=sandbox_session) + if sandbox_session is not None + else None + ) + + return RunConfig( + model=model, + model_provider=build_multi_provider(), + model_settings=base_settings, + sandbox=sandbox_config, + call_model_input_filter=inject_messages_filter, + tracing_disabled=False, + trace_include_sensitive_data=False, + ) + + +def make_agent_context( + *, + bus: AgentMessageBus, + sandbox_session: BaseSandboxSession | None, + sandbox_token: str | None, + tool_server_host_port: int | None, + caido_host_port: int | None, + agent_id: str, + agent_name: str, + parent_id: str | None, + tracer: Any | None, + model: str = "strix/claude-sonnet-4.6", + model_settings: ModelSettings | None = None, + max_turns: int = 300, + is_whitebox: bool = False, + diff_scope: dict[str, Any] | None = None, + run_id: str | None = None, +) -> dict[str, Any]: + """Build the per-agent ``context`` dict passed to ``Runner.run(context=...)``. + + The dict is the canonical place where bus, sandbox handles, identity, + tracer reference, and per-agent toggles live. Tools, hooks, and the + ``inject_messages_filter`` all reach in via ``ctx.context.get(...)``. + + C21 (AUDIT_R3): includes ``is_whitebox``, ``diff_scope`` and ``run_id`` + fields that the legacy code relied on but the original PLAYBOOK §2.10 + skeleton omitted. + """ + return { + "bus": bus, + "sandbox_session": sandbox_session, + "sandbox_token": sandbox_token, + "tool_server_host_port": tool_server_host_port, + "caido_host_port": caido_host_port, + "agent_id": agent_id, + "agent_name": agent_name, + "parent_id": parent_id, + "tracer": tracer, + "model": model, + "model_settings": model_settings, + "max_turns": max_turns, + "turn_count": 0, + "agent_finish_called": False, + "is_whitebox": is_whitebox, + "diff_scope": diff_scope, + "run_id": run_id, + } diff --git a/strix/telemetry/strix_processor.py b/strix/telemetry/strix_processor.py new file mode 100644 index 0000000..5966452 --- /dev/null +++ b/strix/telemetry/strix_processor.py @@ -0,0 +1,183 @@ +"""StrixTracingProcessor — SDK trace processor that writes events.jsonl. + +Replaces the JSONL output that the legacy tracer wrote in ``telemetry/tracer.py``. +Hooks into the SDK's tracing pipeline so we keep the existing +``strix_runs//events.jsonl`` schema and the existing +``TelemetrySanitizer`` PII redaction without standing up a parallel +tracing system. + +References: + - PLAYBOOK.md §2.9 + - AUDIT_R2.md §1.2 (C7 — JSONL writes must be lock-protected) + - AUDIT_R3.md C16 (writes must catch OSError; never tear down the run) + - AUDIT_R3.md F3 (every TracingProcessor hook is SYNC, not async) +""" + +from __future__ import annotations + +import json +import logging +import threading +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from agents.tracing.processor_interface import TracingProcessor + + +if TYPE_CHECKING: + from agents.tracing.spans import Span + from agents.tracing.traces import Trace + + from strix.telemetry.utils import TelemetrySanitizer + + +logger = logging.getLogger(__name__) + + +# Module-level lock registry — one per JSONL file so two processors writing +# different run-dirs don't serialize unnecessarily, but two processors +# writing the *same* run-dir (e.g., legacy tracer + SDK processor) do. +_FILE_LOCKS: dict[Path, threading.Lock] = {} +_GUARD = threading.Lock() + + +def _lock_for(path: Path) -> threading.Lock: + with _GUARD: + return _FILE_LOCKS.setdefault(path, threading.Lock()) + + +class StrixTracingProcessor(TracingProcessor): + """Append trace + span events as JSONL into ``run_dir/events.jsonl``. + + Every hook is synchronous — required by ``TracingProcessor`` ABC. + Every write is protected by a per-path ``threading.Lock`` so concurrent + spans (e.g., from parallel agent tasks) cannot interleave bytes + mid-line and corrupt the JSONL (C7). + + Every write is wrapped in ``try/except OSError`` so a full disk or a + permission error during the run does NOT propagate up the hook chain + and tear down the agent (C16). + + PII scrubbing runs on every event before it hits the file. The + ``TelemetrySanitizer`` class is the same one the legacy tracer uses. + """ + + def __init__( + self, + run_dir: Path, + sanitizer: TelemetrySanitizer | None = None, + ) -> None: + run_dir = Path(run_dir) + run_dir.mkdir(parents=True, exist_ok=True) + self.run_dir: Path = run_dir + self.events_path: Path = run_dir / "events.jsonl" + if sanitizer is None: + from strix.telemetry.utils import TelemetrySanitizer + + sanitizer = TelemetrySanitizer() + self.sanitizer: TelemetrySanitizer = sanitizer + + # --- Internal helpers ------------------------------------------------- + + def _emit(self, event: dict[str, Any]) -> None: + """Sanitize ``event`` and append it as one JSONL line. + + Failures are swallowed — we'd rather lose a trace event than fail + the run. Errors are logged at WARNING (C16). + """ + try: + clean = self.sanitizer.sanitize(event) + except Exception: + logger.exception("Trace event sanitization failed; dropping event") + return + try: + with ( + _lock_for(self.events_path), + self.events_path.open( + "a", + encoding="utf-8", + ) as f, + ): + f.write(json.dumps(clean, ensure_ascii=True) + "\n") + except OSError: + logger.exception("Failed to append trace event to %s", self.events_path) + + @staticmethod + def _span_kind(span: Span[Any]) -> str: + """Map ``SomethingSpanData`` → ``"something"`` for the event_type.""" + name = type(span.span_data).__name__ + if name.endswith("SpanData"): + name = name[: -len("SpanData")] + return name.lower() or "span" + + @staticmethod + def _trace_metadata(trace: Trace) -> dict[str, Any]: + meta: dict[str, Any] = {"name": getattr(trace, "name", None)} + # ``Trace.export()`` includes metadata + group_id when set. + try: + exported = trace.export() + if isinstance(exported, dict): + for key in ("metadata", "group_id", "workflow_name"): + if key in exported and exported[key] is not None: + meta[key] = exported[key] + except Exception: + logger.debug("trace.export failed", exc_info=True) + return meta + + # --- TracingProcessor ABC -------------------------------------------- + + def on_trace_start(self, trace: Trace) -> None: + self._emit( + { + "event_type": "run.started", + "trace_id": trace.trace_id, + "metadata": self._trace_metadata(trace), + } + ) + + def on_trace_end(self, trace: Trace) -> None: + self._emit( + { + "event_type": "run.completed", + "trace_id": trace.trace_id, + } + ) + + def on_span_start(self, span: Span[Any]) -> None: + kind = self._span_kind(span) + self._emit( + { + "event_type": f"{kind}.started", + "span_id": span.span_id, + "trace_id": span.trace_id, + "data": self._safe_export(span), + } + ) + + def on_span_end(self, span: Span[Any]) -> None: + kind = self._span_kind(span) + self._emit( + { + "event_type": f"{kind}.completed", + "span_id": span.span_id, + "trace_id": span.trace_id, + "data": self._safe_export(span), + } + ) + + def force_flush(self) -> None: + """All writes are synchronous; nothing to flush.""" + + def shutdown(self) -> None: + """No-op; nothing to release.""" + + # --- helpers ---------------------------------------------------------- + + @staticmethod + def _safe_export(span: Span[Any]) -> dict[str, Any] | None: + try: + data = span.span_data.export() + return data if isinstance(data, dict) else None + except Exception: + logger.debug("span_data.export failed for %s", span.span_id, exc_info=True) + return None diff --git a/tests/llm/test_strix_session.py b/tests/llm/test_strix_session.py new file mode 100644 index 0000000..5aa4537 --- /dev/null +++ b/tests/llm/test_strix_session.py @@ -0,0 +1,153 @@ +"""Phase 1 smoke tests for StrixSession (memory compression wrapper).""" + +from __future__ import annotations + +from typing import Any + +import pytest +from agents.memory.session import SessionABC + +from strix.llm.strix_session import StrixSession + + +class _FakeUnderlying(SessionABC): + """In-memory SessionABC used to drive StrixSession in tests.""" + + def __init__(self, items: list[Any] | None = None) -> None: + self.items: list[Any] = list(items or []) + self.session_id = "fake-session" + + async def get_items(self, limit: int | None = None) -> list[Any]: + if limit is None: + return list(self.items) + return list(self.items[-limit:]) + + async def add_items(self, items: list[Any]) -> None: + self.items.extend(items) + + async def pop_item(self) -> Any | None: + return self.items.pop() if self.items else None + + async def clear_session(self) -> None: + self.items.clear() + + +class _CompressorOK: + """Compressor stand-in that compresses by keeping the last item.""" + + def __init__(self) -> None: + self.calls = 0 + + def compress_history(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + self.calls += 1 + return messages[-1:] if len(messages) > 1 else messages + + +class _CompressorBoom: + """Compressor stand-in that always raises.""" + + def __init__(self) -> None: + self.calls = 0 + + def compress_history(self, messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + self.calls += 1 + raise RuntimeError("compressor offline") + + +@pytest.fixture +def items() -> list[dict[str, Any]]: + return [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "second"}, + {"role": "user", "content": "third"}, + ] + + +@pytest.mark.asyncio +async def test_get_items_compresses_when_compressor_ok(items: list[dict[str, Any]]) -> None: + underlying = _FakeUnderlying(items) + session = StrixSession(underlying, compressor=_CompressorOK()) + + out = await session.get_items() + assert len(out) == 1 + assert out[0]["content"] == "third" + + +@pytest.mark.asyncio +async def test_get_items_returns_empty_without_calling_compressor() -> None: + """If underlying has no items, don't even invoke the compressor.""" + underlying = _FakeUnderlying([]) + compressor = _CompressorOK() + session = StrixSession(underlying, compressor=compressor) + + out = await session.get_items() + assert out == [] + assert compressor.calls == 0 + + +@pytest.mark.asyncio +async def test_get_items_falls_back_to_uncompressed_on_exception( + items: list[dict[str, Any]], +) -> None: + """C10 (AUDIT_R2): compressor failure must not tear down the run.""" + underlying = _FakeUnderlying(items) + compressor = _CompressorBoom() + session = StrixSession(underlying, compressor=compressor) + + out = await session.get_items() + # Uncompressed history returned. + assert out == items + # Flag set so subsequent calls skip the compressor. + assert session.compression_disabled is True + + +@pytest.mark.asyncio +async def test_compressor_disabled_after_first_failure(items: list[dict[str, Any]]) -> None: + """Round 3.4 §E2 / W5 — once the compressor fails, skip it forever.""" + underlying = _FakeUnderlying(items) + compressor = _CompressorBoom() + session = StrixSession(underlying, compressor=compressor) + + # First call: compressor invoked, raises, flag set. + await session.get_items() + assert compressor.calls == 1 + assert session.compression_disabled is True + + # Second + third call: compressor short-circuited. + await session.get_items() + await session.get_items() + assert compressor.calls == 1 + + +@pytest.mark.asyncio +async def test_writes_pass_through(items: list[dict[str, Any]]) -> None: + underlying = _FakeUnderlying() + session = StrixSession(underlying, compressor=_CompressorOK()) + + await session.add_items(items) + assert underlying.items == items + + popped = await session.pop_item() + assert popped == items[-1] + + await session.clear_session() + assert underlying.items == [] + + +@pytest.mark.asyncio +async def test_session_id_passes_through() -> None: + underlying = _FakeUnderlying() + session = StrixSession(underlying, compressor=_CompressorOK()) + assert session.session_id == "fake-session" + + +@pytest.mark.asyncio +async def test_get_items_respects_limit(items: list[dict[str, Any]]) -> None: + """``limit`` is forwarded to the underlying session before compression.""" + underlying = _FakeUnderlying(items) + session = StrixSession(underlying, compressor=_CompressorOK()) + + out = await session.get_items(limit=2) + # Underlying returned last 2 items; compressor kept the last 1. + assert len(out) == 1 + assert out[0]["content"] == "third" diff --git a/tests/telemetry/test_strix_processor.py b/tests/telemetry/test_strix_processor.py new file mode 100644 index 0000000..e4a657e --- /dev/null +++ b/tests/telemetry/test_strix_processor.py @@ -0,0 +1,201 @@ +"""Phase 1 smoke tests for StrixTracingProcessor.""" + +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import pytest + +from strix.telemetry.strix_processor import StrixTracingProcessor + + +@dataclass +class _FakeSpanData: + """Minimal stand-in for an SDK SpanData class.""" + + name: str = "FunctionSpanData" # used by class .__name__ + payload: dict[str, Any] = field(default_factory=dict) + + def export(self) -> dict[str, Any]: + return dict(self.payload) + + +# Concrete SpanData subclasses so the processor's ``_span_kind`` heuristic +# (drop ``SpanData`` suffix, lowercase) produces stable event_types. +class FunctionSpanData(_FakeSpanData): + pass + + +class GenerationSpanData(_FakeSpanData): + pass + + +class AgentSpanData(_FakeSpanData): + pass + + +@dataclass +class _FakeSpan: + span_id: str + trace_id: str + span_data: _FakeSpanData + + +@dataclass +class _FakeTrace: + trace_id: str + name: str = "test-workflow" + metadata: dict[str, Any] = field(default_factory=dict) + + def export(self) -> dict[str, Any]: + return {"name": self.name, "metadata": self.metadata} + + +def _read_events(path: Path) -> list[dict[str, Any]]: + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + + +@pytest.fixture +def run_dir(tmp_path: Path) -> Path: + return tmp_path / "strix_runs" / "test-run" + + +def test_constructor_creates_run_dir(run_dir: Path) -> None: + StrixTracingProcessor(run_dir=run_dir) + assert run_dir.exists() + + +def test_on_trace_start_writes_run_started(run_dir: Path) -> None: + p = StrixTracingProcessor(run_dir=run_dir) + p.on_trace_start(_FakeTrace(trace_id="t-1", metadata={"scan_id": "abc"})) + + events = _read_events(p.events_path) + assert events == [ + { + "event_type": "run.started", + "trace_id": "t-1", + "metadata": {"name": "test-workflow", "metadata": {"scan_id": "abc"}}, + } + ] + + +def test_on_trace_end_writes_run_completed(run_dir: Path) -> None: + p = StrixTracingProcessor(run_dir=run_dir) + p.on_trace_end(_FakeTrace(trace_id="t-1")) + + events = _read_events(p.events_path) + assert events == [{"event_type": "run.completed", "trace_id": "t-1"}] + + +def test_span_start_and_end_emit_typed_events(run_dir: Path) -> None: + """``GenerationSpanData`` → ``generation.started`` / ``generation.completed``.""" + p = StrixTracingProcessor(run_dir=run_dir) + span = _FakeSpan( + span_id="s-1", + trace_id="t-1", + span_data=GenerationSpanData(payload={"model": "gpt-foo"}), + ) + + p.on_span_start(span) + p.on_span_end(span) + + events = _read_events(p.events_path) + assert [e["event_type"] for e in events] == [ + "generation.started", + "generation.completed", + ] + assert events[0]["span_id"] == "s-1" + assert events[0]["data"] == {"model": "gpt-foo"} + + +def test_concurrent_writes_yield_valid_jsonl(run_dir: Path) -> None: + """C7 (AUDIT_R2): per-path lock prevents JSONL corruption under contention.""" + p = StrixTracingProcessor(run_dir=run_dir) + + def writer(idx: int) -> None: + for i in range(50): + p._emit({"event_type": "synthetic", "writer": idx, "i": i}) + + threads = [threading.Thread(target=writer, args=(i,)) for i in range(10)] + for t in threads: + t.start() + for t in threads: + t.join() + + # 10 threads x 50 events = 500 lines; all valid JSON. + lines = p.events_path.read_text().splitlines() + assert len(lines) == 500 + for line in lines: + json.loads(line) # raises on corrupt line + + +def test_emit_swallows_oserror_and_logs( + run_dir: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """C16 (AUDIT_R3): a write failure must NOT propagate.""" + p = StrixTracingProcessor(run_dir=run_dir) + # Make events_path point to a directory so open(..., "a") raises. + p.events_path = run_dir + + with caplog.at_level("ERROR", logger="strix.telemetry.strix_processor"): + p._emit({"event_type": "boom"}) + + assert any("Failed to append" in rec.message for rec in caplog.records) + + +def test_span_export_failure_does_not_propagate(run_dir: Path) -> None: + """If span_data.export raises, we still emit an event with data=None.""" + p = StrixTracingProcessor(run_dir=run_dir) + + class _BoomSpanData: + def export(self) -> dict[str, Any]: + raise RuntimeError("nope") + + # Reuse the lowercase rule: class name has no "SpanData" suffix → "boomspandata" + # would not be ideal; use a properly-named subclass. + class FunctionSpanDataBroken(_BoomSpanData): + pass + + span = _FakeSpan(span_id="s-1", trace_id="t-1", span_data=FunctionSpanDataBroken()) + p.on_span_end(span) + + events = _read_events(p.events_path) + assert len(events) == 1 + assert events[0]["data"] is None + + +def test_pii_scrubbed_via_sanitizer(run_dir: Path) -> None: + """Sanitizer is invoked on every emit before write.""" + seen: list[Any] = [] + + class _StubSanitizer: + def sanitize(self, data: Any, key_hint: str | None = None) -> Any: + seen.append(data) + # Replace any "secret" string with [REDACTED]. + if isinstance(data, dict): + clean = {k: "[REDACTED]" if k == "api_key" else v for k, v in data.items()} + if "metadata" in clean and isinstance(clean["metadata"], dict): + md = dict(clean["metadata"]) + md.pop("api_key", None) + clean["metadata"] = md + return clean + return data + + p = StrixTracingProcessor(run_dir=run_dir, sanitizer=_StubSanitizer()) + p._emit({"event_type": "test", "api_key": "sk-very-secret"}) + + events = _read_events(p.events_path) + assert events[0]["api_key"] == "[REDACTED]" + assert seen and seen[0]["api_key"] == "sk-very-secret" + + +def test_force_flush_and_shutdown_are_noops(run_dir: Path) -> None: + p = StrixTracingProcessor(run_dir=run_dir) + # Should not raise. + p.force_flush() + p.shutdown() diff --git a/tests/test_run_config_factory.py b/tests/test_run_config_factory.py new file mode 100644 index 0000000..6dcb7f5 --- /dev/null +++ b/tests/test_run_config_factory.py @@ -0,0 +1,181 @@ +"""Phase 1 smoke tests for make_run_config / make_agent_context.""" + +from __future__ import annotations + +from agents import RunConfig +from agents.model_settings import ModelSettings +from agents.retry import ModelRetryBackoffSettings + +from strix.orchestration.bus import AgentMessageBus +from strix.run_config_factory import ( + _RETRYABLE_HTTP_STATUSES, + make_agent_context, + make_run_config, +) + + +def test_make_run_config_returns_run_config() -> None: + cfg = make_run_config(sandbox_session=None) + assert isinstance(cfg, RunConfig) + + +def test_default_parallel_tool_calls_is_false() -> None: + """C1 (AUDIT.md): Phase 1 default is sequential to match legacy tool server.""" + cfg = make_run_config(sandbox_session=None) + assert cfg.model_settings is not None + assert cfg.model_settings.parallel_tool_calls is False + + +def test_default_tool_choice_is_required() -> None: + cfg = make_run_config(sandbox_session=None) + assert cfg.model_settings is not None + assert cfg.model_settings.tool_choice == "required" + + +def test_call_model_input_filter_is_wired() -> None: + cfg = make_run_config(sandbox_session=None) + assert cfg.call_model_input_filter is not None + # Wired to inject_messages_filter (validated by name to keep import light). + assert cfg.call_model_input_filter.__name__ == "inject_messages_filter" + + +def test_retry_settings_have_max_retries_5() -> None: + cfg = make_run_config(sandbox_session=None) + assert cfg.model_settings is not None + retry = cfg.model_settings.retry + assert retry is not None + assert retry.max_retries == 5 + + +def test_retry_backoff_uses_strix_defaults() -> None: + """Mirrors legacy llm.py: min(90, 2*2^n) with initial 2s, max 90s, x2.""" + cfg = make_run_config(sandbox_session=None) + assert cfg.model_settings is not None + retry = cfg.model_settings.retry + assert retry is not None + backoff = retry.backoff + assert isinstance(backoff, ModelRetryBackoffSettings) + assert backoff.initial_delay == 2.0 + assert backoff.max_delay == 90.0 + assert backoff.multiplier == 2.0 + + +def test_retry_http_codes_exclude_401_403_400() -> None: + """C11 (AUDIT_R2): auth/validation errors must NOT be in the retry list.""" + assert 401 not in _RETRYABLE_HTTP_STATUSES + assert 403 not in _RETRYABLE_HTTP_STATUSES + assert 400 not in _RETRYABLE_HTTP_STATUSES + # And 429 / 5xx must be present. + for code in (429, 500, 502, 503, 504): + assert code in _RETRYABLE_HTTP_STATUSES + + +def test_trace_include_sensitive_data_is_false() -> None: + cfg = make_run_config(sandbox_session=None) + assert cfg.trace_include_sensitive_data is False + + +def test_model_settings_override_merges() -> None: + """C21 (AUDIT_R3): per-call override path.""" + override = ModelSettings(tool_choice="auto", parallel_tool_calls=True) + cfg = make_run_config(sandbox_session=None, model_settings_override=override) + assert cfg.model_settings is not None + assert cfg.model_settings.tool_choice == "auto" + assert cfg.model_settings.parallel_tool_calls is True + # Retry settings (not in override) preserved from base. + assert cfg.model_settings.retry is not None + assert cfg.model_settings.retry.max_retries == 5 + + +def test_reasoning_effort_propagates() -> None: + cfg = make_run_config(sandbox_session=None, reasoning_effort="high") + assert cfg.model_settings is not None + assert cfg.model_settings.reasoning is not None + assert cfg.model_settings.reasoning.effort == "high" + + +def test_max_turns_default_is_300() -> None: + """Mirrors legacy AgentState.max_iterations=300 (HARNESS_WIKI §5.2).""" + # max_turns is RunConfig-level; we default 300 in make_agent_context for + # the per-agent context dict. RunConfig itself sets max_turns at run call + # time via Runner.run(max_turns=...). Verify our context. + bus = AgentMessageBus() + ctx = make_agent_context( + bus=bus, + sandbox_session=None, + sandbox_token=None, + tool_server_host_port=None, + caido_host_port=None, + agent_id="root", + agent_name="root", + parent_id=None, + tracer=None, + ) + assert ctx["max_turns"] == 300 + + +def test_make_agent_context_full_shape() -> None: + """C21 — context dict carries every field tools/hooks reach for.""" + bus = AgentMessageBus() + ctx = make_agent_context( + bus=bus, + sandbox_session=None, + sandbox_token="bearer-xyz", + tool_server_host_port=48081, + caido_host_port=48080, + agent_id="agent-1", + agent_name="root", + parent_id=None, + tracer="not-a-real-tracer", + is_whitebox=True, + diff_scope={"changed_files": ["src/app.py"]}, + run_id="strix_runs/abc_def", + ) + + assert ctx["bus"] is bus + assert ctx["agent_id"] == "agent-1" + assert ctx["parent_id"] is None + assert ctx["agent_finish_called"] is False + assert ctx["turn_count"] == 0 + assert ctx["is_whitebox"] is True + assert ctx["diff_scope"] == {"changed_files": ["src/app.py"]} + assert ctx["run_id"] == "strix_runs/abc_def" + assert ctx["sandbox_token"] == "bearer-xyz" + assert ctx["tool_server_host_port"] == 48081 + assert ctx["caido_host_port"] == 48080 + + +def test_make_agent_context_is_whitebox_defaults_false() -> None: + bus = AgentMessageBus() + ctx = make_agent_context( + bus=bus, + sandbox_session=None, + sandbox_token=None, + tool_server_host_port=None, + caido_host_port=None, + agent_id="r", + agent_name="root", + parent_id=None, + tracer=None, + ) + assert ctx["is_whitebox"] is False + assert ctx["diff_scope"] is None + + +def test_sandbox_config_omitted_when_no_session() -> None: + cfg = make_run_config(sandbox_session=None) + assert cfg.sandbox is None + + +def test_model_default_is_strix_claude() -> None: + """Production default per AUDIT/PLAYBOOK convention.""" + cfg = make_run_config(sandbox_session=None) + assert cfg.model == "strix/claude-sonnet-4.6" + + +def test_multi_provider_is_built() -> None: + """Verify the factory wires our custom MultiProvider, not the SDK default.""" + cfg = make_run_config(sandbox_session=None) + # MultiProvider is opaque, but our build_multi_provider returns + # an instance with our prefix routes installed. + assert cfg.model_provider is not None