mortal/strix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: nuke `events.jsonl` pipeline and the unused PII sanitizer

Browse Source 
The JSONL trace sink was never read — TUI consumes ``Tracer`` state
directly (chat_messages, agents, tool_executions, vulnerability_reports,
LLM stats), and SQLiteSession owns the conversation history. The whole
``StrixTracingProcessor`` → ``_emit_event`` → ``append_jsonl_record``
pipeline was producing files nothing opens.

Deleted:
- ``strix/telemetry/strix_processor.py`` (the SDK ``TracingProcessor``).
- ``strix/telemetry/utils.py`` — ``TelemetrySanitizer`` (no remaining
  callers), ``append_jsonl_record``, ``get_events_write_lock``,
  ``reset_events_write_locks``.
- ``strix/telemetry/flags.py`` — ``is_telemetry_enabled`` /
  ``is_posthog_enabled`` collapsed into a 4-line check inside
  ``posthog._is_enabled`` (its only caller).
- ``Tracer._emit_event`` and every event-emit call inside the tracer
  (``run.started``, ``run.configured``, ``run.completed``,
  ``finding.created``, ``finding.reviewed``, ``chat.message``).
- ``Tracer._enrich_actor`` (only used by ``_emit_event``).
- ``Tracer._sanitize_data`` + ``_sanitizer`` field (PII scrub only ran
  on JSONL events).
- ``Tracer.events_file_path`` property and the ``_events_file_path`` /
  ``_telemetry_enabled`` / ``_run_completed_emitted`` /
  ``_next_execution_id`` fields.
- ``Tracer._calculate_duration`` (one caller in posthog — inlined).
- ``add_trace_processor(StrixTracingProcessor(run_dir))`` from
  ``entry.py``.

The ``Tracer`` class is now ~275 LoC of pure runtime state for the TUI
+ vulnerability artifact writer (markdown / CSV / pentest report).
Conversation history goes to ``SQLiteSession``; SDK trace events are
not persisted.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 13:47:37 -07:00
parent df51eeedd0
commit 9b31e9fd29
6 changed files with 39 additions and 463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strix/entry.py
-14
View File
@@ -23,7 +23,6 @@ from typing import TYPE_CHECKING, Any, Literal
from agents import Runner
from agents.memory import SQLiteSession
from agents.tracing import add_trace_processor
from strix.agents.factory import build_strix_agent, make_child_factory
from strix.config.config import Config
@@ -36,7 +35,6 @@ from strix.run_config_factory import (
)
from strix.sandbox import session_manager
from strix.sandbox.healthcheck import wait_for_http_ready, wait_for_tcp_ready
from strix.telemetry.strix_processor import StrixTracingProcessor
if TYPE_CHECKING:
@@ -203,18 +201,6 @@ async def run_strix_scan(
bus = AgentMessageBus()
root_id = uuid.uuid4().hex[:8]
# Wire SDK tracing into the scan's run-directory ``events.jsonl``.
# ``add_trace_processor`` is idempotent at the provider level — if
# the user runs multiple scans in one process they each get their
# own processor, all writing to their respective run dirs.
if tracer is not None:
try:
run_dir = tracer.get_run_dir() if hasattr(tracer, "get_run_dir") else None
if run_dir is not None:
add_trace_processor(StrixTracingProcessor(run_dir))
except Exception:
logger.exception("Failed to register StrixTracingProcessor")
bundle = await session_manager.create_or_reuse(
scan_id,
image=image,
strix/telemetry/flags.py
-21
View File
@@ -1,21 +0,0 @@
from strix.config import Config
_DISABLED_VALUES = {"0", "false", "no", "off"}
def _is_enabled(raw_value: str | None, default: str = "1") -> bool:
value = (raw_value if raw_value is not None else default).strip().lower()
return value not in _DISABLED_VALUES
def is_telemetry_enabled() -> bool:
"""Master gate — controls JSONL event emission and posthog."""
return _is_enabled(Config.get("strix_telemetry"), default="1")
def is_posthog_enabled() -> bool:
explicit = Config.get("strix_posthog_telemetry")
if explicit is not None:
return _is_enabled(explicit)
return is_telemetry_enabled()
strix/telemetry/posthog.py
+20 -3
View File
@@ -6,7 +6,7 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any
from uuid import uuid4
from strix.telemetry.flags import is_posthog_enabled
from strix.config import Config
if TYPE_CHECKING:
@@ -15,11 +15,18 @@ if TYPE_CHECKING:
_POSTHOG_PUBLIC_API_KEY = "phc_7rO3XRuNT5sgSKAl6HDIrWdSGh1COzxw0vxVIAR6vVZ"
_POSTHOG_HOST = "https://us.i.posthog.com"
_DISABLED_VALUES = {"0", "false", "no", "off"}
_SESSION_ID = uuid4().hex[:16]
def _is_enabled() -> bool:
return is_posthog_enabled()
"""Master telemetry gate. ``STRIX_POSTHOG_TELEMETRY`` overrides ``STRIX_TELEMETRY``."""
explicit = Config.get("strix_posthog_telemetry")
if explicit is not None:
return explicit.strip().lower() not in _DISABLED_VALUES
fallback = Config.get("strix_telemetry") or "1"
return fallback.strip().lower() not in _DISABLED_VALUES
def _is_first_run() -> bool:
@@ -114,12 +121,22 @@ def end(tracer: "Tracer", exit_reason: str = "completed") -> None:
llm = tracer.get_total_llm_stats()
total = llm.get("total", {})
duration = 0.0
try:
from datetime import datetime
start = datetime.fromisoformat(tracer.start_time.replace("Z", "+00:00"))
end_iso = tracer.end_time or datetime.now(start.tzinfo).isoformat()
duration = (datetime.fromisoformat(end_iso.replace("Z", "+00:00")) - start).total_seconds()
except (ValueError, TypeError, AttributeError):
pass
_send(
"scan_ended",
{
**_base_props(),
"exit_reason": exit_reason,
"duration_seconds": round(tracer._calculate_duration()),
"duration_seconds": round(duration),
"vulnerabilities_total": len(tracer.vulnerability_reports),
**{f"vulnerabilities_{k}": v for k, v in vulnerabilities_counts.items()},
"agent_count": len(tracer.agents),
strix/telemetry/strix_processor.py
-169
View File
@@ -1,169 +0,0 @@
"""``StrixTracingProcessor`` — SDK trace processor that writes ``events.jsonl``.
Hooks into the SDK's tracing pipeline and writes events to
``strix_runs/<run-name>/events.jsonl``. PII scrubbing runs through
:class:`TelemetrySanitizer`.
"""
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 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 the ``TracingProcessor``
ABC. Each write is protected by a per-path ``threading.Lock`` so
concurrent spans can't interleave bytes mid-line. ``OSError`` is
swallowed so a full disk or permission error doesn't tear the run
down. PII scrubbing runs on every event before it hits the file.
"""
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.
"""
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
strix/telemetry/tracer.py
+19 -154
View File
@@ -6,8 +6,6 @@ from typing import Any, Optional
from uuid import uuid4
from strix.telemetry import posthog
from strix.telemetry.flags import is_telemetry_enabled
from strix.telemetry.utils import TelemetrySanitizer, append_jsonl_record
logger = logging.getLogger(__name__)
@@ -25,6 +23,16 @@ def set_global_tracer(tracer: "Tracer") -> None:
class Tracer:
"""Per-scan in-memory state the TUI renders + per-scan artifact writer.
Holds live state the TUI reads (chat messages, agent tree, tool
executions, vulnerability reports, LLM usage). Writes vulnerability
markdown + CSV + final pentest report to ``strix_runs/<scan>/``.
Conversation history goes to the SDK's ``SQLiteSession`` instead;
SDK trace events are not persisted here.
"""
def __init__(self, run_name: str | None = None):
self.run_name = run_name
self.run_id = run_name or f"run-{uuid4().hex[:8]}"
@@ -58,111 +66,18 @@ class Tracer:
"status": "running",
}
self._run_dir: Path | None = None
self._events_file_path: Path | None = None
self._next_execution_id = 1
self._next_message_id = 1
self._saved_vuln_ids: set[str] = set()
self._run_completed_emitted = False
self._telemetry_enabled = is_telemetry_enabled()
self._sanitizer = TelemetrySanitizer()
self.caido_url: str | None = None
self.vulnerability_found_callback: Callable[[dict[str, Any]], None] | None = None
self._emit_run_started_event()
@property
def events_file_path(self) -> Path:
if self._events_file_path is None:
self._events_file_path = self.get_run_dir() / "events.jsonl"
return self._events_file_path
def _sanitize_data(self, data: Any, key_hint: str | None = None) -> Any:
return self._sanitizer.sanitize(data, key_hint=key_hint)
def _append_event_record(self, record: dict[str, Any]) -> None:
try:
append_jsonl_record(self.events_file_path, record)
except OSError:
logger.exception("Failed to append JSONL event record")
def _enrich_actor(self, actor: dict[str, Any] | None) -> dict[str, Any] | None:
if not actor:
return None
enriched = dict(actor)
if "agent_name" in enriched:
return enriched
agent_id = enriched.get("agent_id")
if not isinstance(agent_id, str):
return enriched
agent_data = self.agents.get(agent_id, {})
agent_name = agent_data.get("name")
if isinstance(agent_name, str) and agent_name:
enriched["agent_name"] = agent_name
return enriched
def _emit_event(
self,
event_type: str,
actor: dict[str, Any] | None = None,
payload: Any | None = None,
status: str | None = None,
error: Any | None = None,
source: str = "strix.tracer",
include_run_metadata: bool = False,
) -> None:
if not self._telemetry_enabled:
return
enriched_actor = self._enrich_actor(actor)
sanitized_actor = self._sanitize_data(enriched_actor) if enriched_actor else None
sanitized_payload = self._sanitize_data(payload) if payload is not None else None
sanitized_error = self._sanitize_data(error) if error is not None else None
record = {
"timestamp": datetime.now(UTC).isoformat(),
"event_type": event_type,
"run_id": self.run_id,
"trace_id": uuid4().hex,
"span_id": uuid4().hex[:16],
"actor": sanitized_actor,
"payload": sanitized_payload,
"status": status,
"error": sanitized_error,
"source": source,
}
if include_run_metadata:
record["run_metadata"] = self._sanitize_data(self.run_metadata)
self._append_event_record(record)
def set_run_name(self, run_name: str) -> None:
self.run_name = run_name
self.run_id = run_name
self.run_metadata["run_name"] = run_name
self.run_metadata["run_id"] = run_name
self._run_dir = None
self._events_file_path = None
self._run_completed_emitted = False
self._emit_run_started_event()
def _emit_run_started_event(self) -> None:
if not self._telemetry_enabled:
return
self._emit_event(
"run.started",
payload={
"run_name": self.run_name,
"start_time": self.start_time,
"local_jsonl_path": str(self.events_file_path),
},
status="running",
include_run_metadata=True,
)
def get_run_dir(self) -> Path:
if self._run_dir is None:
@@ -235,12 +150,6 @@ class Tracer:
self.vulnerability_reports.append(report)
logger.info(f"Added vulnerability report: {report_id} - {title}")
posthog.finding(severity)
self._emit_event(
"finding.created",
payload={"report": report},
status=report["severity"],
source="strix.findings",
)
if self.vulnerability_found_callback:
self.vulnerability_found_callback(report)
@@ -285,15 +194,6 @@ class Tracer:
"""
logger.info("Updated scan final fields")
self._emit_event(
"finding.reviewed",
payload={
"scan_completed": True,
"vulnerability_count": len(self.vulnerability_reports),
},
status="completed",
source="strix.findings",
)
self.save_run_data(mark_complete=True)
posthog.end(self, exit_reason="finished_by_tool")
@@ -307,22 +207,15 @@ class Tracer:
message_id = self._next_message_id
self._next_message_id += 1
message_data = {
"message_id": message_id,
"content": content,
"role": role,
"agent_id": agent_id,
"timestamp": datetime.now(UTC).isoformat(),
"metadata": metadata or {},
}
self.chat_messages.append(message_data)
self._emit_event(
"chat.message",
actor={"agent_id": agent_id, "role": role},
payload={"message_id": message_id, "content": content, "metadata": metadata or {}},
status="logged",
source="strix.chat",
self.chat_messages.append(
{
"message_id": message_id,
"content": content,
"role": role,
"agent_id": agent_id,
"timestamp": datetime.now(UTC).isoformat(),
"metadata": metadata or {},
}
)
return message_id
@@ -335,12 +228,6 @@ class Tracer:
"max_iterations": config.get("max_iterations", 200),
}
)
self._emit_event(
"run.configured",
payload={"scan_config": config},
status="configured",
source="strix.run",
)
def save_run_data(self, mark_complete: bool = False) -> None:
try:
@@ -489,32 +376,10 @@ class Tracer:
logger.info("Updated vulnerability index: %s", vuln_csv_file)
logger.info("📊 Essential scan data saved to: %s", run_dir)
if mark_complete and not self._run_completed_emitted:
self._emit_event(
"run.completed",
payload={
"duration_seconds": self._calculate_duration(),
"vulnerability_count": len(self.vulnerability_reports),
},
status="completed",
source="strix.run",
include_run_metadata=True,
)
self._run_completed_emitted = True
except (OSError, RuntimeError):
logger.exception("Failed to save scan data")
def _calculate_duration(self) -> float:
try:
start = datetime.fromisoformat(self.start_time.replace("Z", "+00:00"))
if self.end_time:
end = datetime.fromisoformat(self.end_time.replace("Z", "+00:00"))
return (end - start).total_seconds()
except (ValueError, TypeError):
pass
return 0.0
def get_real_tool_count(self) -> int:
return sum(
1
strix/telemetry/utils.py
-102
View File
@@ -1,102 +0,0 @@
import json
import logging
import re
import threading
from pathlib import Path
from typing import Any
from scrubadub import Scrubber
from scrubadub.detectors import RegexDetector
from scrubadub.filth import Filth
logger = logging.getLogger(__name__)
_REDACTED = "[REDACTED]"
_SCREENSHOT_OMITTED = "[SCREENSHOT_OMITTED]"
_SCREENSHOT_KEY_PATTERN = re.compile(r"screenshot", re.IGNORECASE)
_SENSITIVE_KEY_PATTERN = re.compile(
r"(api[_-]?key|token|secret|password|authorization|cookie|session|credential|private[_-]?key)",
re.IGNORECASE,
)
_SENSITIVE_TOKEN_PATTERN = re.compile(
r"(?i)\b("
r"bearer\s+[a-z0-9._-]+|"
r"sk-[a-z0-9_-]{8,}|"
r"gh[pousr]_[a-z0-9_-]{12,}|"
r"xox[baprs]-[a-z0-9-]{12,}"
r")\b"
)
_SCRUBADUB_PLACEHOLDER_PATTERN = re.compile(r"\{\{[^}]+\}\}")
_EVENTS_FILE_LOCKS_LOCK = threading.Lock()
_EVENTS_FILE_LOCKS: dict[str, threading.Lock] = {}
class _SecretFilth(Filth): # type: ignore[misc]
type = "secret"
class _SecretTokenDetector(RegexDetector): # type: ignore[misc]
name = "strix_secret_token_detector"
filth_cls = _SecretFilth
regex = _SENSITIVE_TOKEN_PATTERN
class TelemetrySanitizer:
def __init__(self) -> None:
self._scrubber = Scrubber(detector_list=[_SecretTokenDetector])
def sanitize(self, data: Any, key_hint: str | None = None) -> Any: # noqa: PLR0911
if data is None:
return None
if isinstance(data, dict):
sanitized: dict[str, Any] = {}
for key, value in data.items():
key_str = str(key)
if _SCREENSHOT_KEY_PATTERN.search(key_str):
sanitized[key_str] = _SCREENSHOT_OMITTED
elif _SENSITIVE_KEY_PATTERN.search(key_str):
sanitized[key_str] = _REDACTED
else:
sanitized[key_str] = self.sanitize(value, key_hint=key_str)
return sanitized
if isinstance(data, list):
return [self.sanitize(item, key_hint=key_hint) for item in data]
if isinstance(data, tuple):
return [self.sanitize(item, key_hint=key_hint) for item in data]
if isinstance(data, str):
if key_hint and _SENSITIVE_KEY_PATTERN.search(key_hint):
return _REDACTED
cleaned = self._scrubber.clean(data)
return _SCRUBADUB_PLACEHOLDER_PATTERN.sub(_REDACTED, cleaned)
if isinstance(data, int | float | bool):
return data
return str(data)
def get_events_write_lock(output_path: Path) -> threading.Lock:
path_key = str(output_path.resolve(strict=False))
with _EVENTS_FILE_LOCKS_LOCK:
lock = _EVENTS_FILE_LOCKS.get(path_key)
if lock is None:
lock = threading.Lock()
_EVENTS_FILE_LOCKS[path_key] = lock
return lock
def reset_events_write_locks() -> None:
with _EVENTS_FILE_LOCKS_LOCK:
_EVENTS_FILE_LOCKS.clear()
def append_jsonl_record(output_path: Path, record: dict[str, Any]) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
with get_events_write_lock(output_path), output_path.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")