feat(migration): phase 1 — Session + Tracer + RunConfig factory

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).

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()

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,
+    }

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/<run-name>/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