feat: Add live tracing mode for complete audit trail

strix/agents/base_agent.py

@@ -22,6 +22,96 @@
 from .state import AgentState
 
 
+def _log_live_agent_created(
+    agent_id: str,
+    agent_name: str,
+    task: str,
+    parent_id: str | None,
+    agent_type: str | None,
+) -> None:
+    """Log agent creation to live tracer if enabled."""
+    try:
+        from strix.telemetry.live_tracer import get_live_tracer
+
+        tracer = get_live_tracer()
+        if tracer:
+            tracer.log_agent_created(
+                agent_id=agent_id,
+                agent_name=agent_name,
+                task=task,
+                parent_id=parent_id,
+                agent_type=agent_type,
+            )
+    except Exception:  # noqa: BLE001, S110
+        pass
+
+
+def _log_live_agent_completed(
+    agent_id: str,
+    status: str,
+    result: dict[str, Any] | None = None,
+    error_message: str | None = None,
+) -> None:
+    """Log agent completion to live tracer if enabled."""
+    try:
+        from strix.telemetry.live_tracer import get_live_tracer
+
+        tracer = get_live_tracer()
+        if tracer:
+            tracer.log_agent_completed(
+                agent_id=agent_id,
+                status=status,
+                result=result,
+                error_message=error_message,
+            )
+    except Exception:  # noqa: BLE001, S110
+        pass
+
+
+def _log_live_state_change(
+    agent_id: str,
+    field: str,
+    old_value: Any,
+    new_value: Any,
+) -> None:
+    """Log agent state change to live tracer if enabled."""
+    try:
+        from strix.telemetry.live_tracer import get_live_tracer
+
+        tracer = get_live_tracer()
+        if tracer:
+            tracer.log_agent_state_change(
+                agent_id=agent_id,
+                field=field,
+                old_value=old_value,
+                new_value=new_value,
+            )
+    except Exception:  # noqa: BLE001, S110
+        pass
+
+
+def _log_live_message(
+    agent_id: str | None,
+    role: str,
+    content: str,
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    """Log a chat message to live tracer if enabled."""
+    try:
+        from strix.telemetry.live_tracer import get_live_tracer
+
+        tracer = get_live_tracer()
+        if tracer:
+            tracer.log_message(
+                agent_id=agent_id,
+                role=role,
+                content=content if isinstance(content, str) else str(content),
+                metadata=metadata,
+            )
+    except Exception:  # noqa: BLE001, S110
+        pass
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -112,6 +202,15 @@ def __init__(self, config: dict[str, Any]):
                 )
                 tracer.update_tool_execution(execution_id=exec_id, status="completed", result={})
 
+        # Log to live tracer
+        _log_live_agent_created(
+            agent_id=self.state.agent_id,
+            agent_name=self.state.agent_name,
+            task=self.state.task,
+            parent_id=self.state.parent_id,
+            agent_type=self.__class__.__name__,
+        )
+
         self._add_to_agents_graph()
 
     def _add_to_agents_graph(self) -> None:
@@ -284,17 +383,38 @@ async def _enter_waiting_state(
         error_occurred: bool = False,
         was_cancelled: bool = False,
     ) -> None:
+        old_waiting = self.state.waiting_for_input
         self.state.enter_waiting_state()
 
+        # Determine status for logging
+        if task_completed:
+            status = "completed"
+        elif error_occurred:
+            status = "error"
+        elif was_cancelled:
+            status = "stopped"
+        else:
+            status = "stopped"
+
         if tracer:
-            if task_completed:
-                tracer.update_agent_status(self.state.agent_id, "completed")
-            elif error_occurred:
-                tracer.update_agent_status(self.state.agent_id, "error")
-            elif was_cancelled:
-                tracer.update_agent_status(self.state.agent_id, "stopped")
-            else:
-                tracer.update_agent_status(self.state.agent_id, "stopped")
+            tracer.update_agent_status(self.state.agent_id, status)
+
+        # Log state change to live tracer
+        _log_live_state_change(
+            agent_id=self.state.agent_id,
+            field="waiting_for_input",
+            old_value=old_waiting,
+            new_value=True,
+        )
+
+        # Log completion to live tracer if task completed or error
+        if task_completed or error_occurred:
+            _log_live_agent_completed(
+                agent_id=self.state.agent_id,
+                status=status,
+                result=self.state.final_result,
+                error_message=self.state.errors[-1] if self.state.errors else None,
+            )
 
         if task_completed:
             self.state.add_message(
@@ -344,6 +464,13 @@ async def _initialize_sandbox_and_state(self, task: str) -> None:
 
         self.state.add_message("user", task)
 
+        # Log user message to live tracer
+        _log_live_message(
+            agent_id=self.state.agent_id,
+            role="user",
+            content=task,
+        )
+
     async def _process_iteration(self, tracer: Optional["Tracer"]) -> bool:
         final_response = None
 
@@ -381,6 +508,14 @@ async def _process_iteration(self, tracer: Optional["Tracer"]) -> bool:
                 agent_id=self.state.agent_id,
             )
 
+        # Log message to live tracer
+        _log_live_message(
+            agent_id=self.state.agent_id,
+            role="assistant",
+            content=final_response.content or "",
+            metadata={"has_thinking": bool(thinking_blocks)},
+            )
+
         actions = (
             final_response.tool_invocations
             if hasattr(final_response, "tool_invocations") and final_response.tool_invocations
@@ -418,6 +553,14 @@ async def _execute_actions(self, actions: list[Any], tracer: Optional["Tracer"])
             self.state.set_completed({"success": True})
             if tracer:
                 tracer.update_agent_status(self.state.agent_id, "completed")
+
+            # Log agent completion to live tracer
+            _log_live_agent_completed(
+                agent_id=self.state.agent_id,
+                status="completed",
+                result={"success": True},
+            )
+
             if self.non_interactive and self.state.parent_id is None:
                 return True
             return True

strix/config/config.py

@@ -45,6 +45,11 @@ class Config:
     # Telemetry
     strix_telemetry = "1"
 
+    # Live Tracing
+    strix_trace = None  # Enable live tracing (set to "1" or "true")
+    strix_trace_output = None  # Custom trace output path (defaults to strix_runs/<run>/trace.jsonl)
+    strix_redact_secrets = None  # Redact secrets in trace output (set to "1" or "true")
+
     # Config file override (set via --config CLI arg)
     _config_file_override: Path | None = None
 

strix/interface/cli.py

@@ -107,12 +107,29 @@ def display_vulnerability(report: dict[str, Any]) -> None:
 
     def cleanup_on_exit() -> None:
         from strix.runtime import cleanup_runtime
+        from strix.telemetry.live_tracer import get_live_tracer, set_live_tracer
 
         tracer.cleanup()
+
+        # Clean up live tracer
+        live_tracer = get_live_tracer()
+        if live_tracer:
+            live_tracer.close()
+            set_live_tracer(None)
+
         cleanup_runtime()
 
     def signal_handler(_signum: int, _frame: Any) -> None:
+        from strix.telemetry.live_tracer import get_live_tracer, set_live_tracer
+
         tracer.cleanup()
+
+        # Clean up live tracer
+        live_tracer = get_live_tracer()
+        if live_tracer:
+            live_tracer.close()
+            set_live_tracer(None)
+
         sys.exit(1)
 
     atexit.register(cleanup_on_exit)

strix/interface/main.py

@@ -40,6 +40,7 @@
 )
 from strix.runtime.docker_runtime import HOST_GATEWAY_HOSTNAME  # noqa: E402
 from strix.telemetry import posthog  # noqa: E402
+from strix.telemetry.live_tracer import LiveTracer, set_live_tracer  # noqa: E402
 from strix.telemetry.tracer import get_global_tracer  # noqa: E402
 
 
@@ -363,13 +364,71 @@ def parse_arguments() -> argparse.Namespace:
         help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
     )
 
+    parser.add_argument(
+        "--trace",
+        action="store_true",
+        help=(
+            "Enable live tracing mode. Creates a complete JSONL audit trail of the run "
+            "including LLM requests/responses, tool calls, and agent events. "
+            "Output defaults to strix_runs/<run>/trace.jsonl. "
+            "Can also be enabled via STRIX_TRACE=1 environment variable."
+        ),
+    )
+
+    parser.add_argument(
+        "--trace-output",
+        type=str,
+        help=(
+            "Custom path for trace output file (requires --trace). "
+            "Defaults to strix_runs/<run>/trace.jsonl."
+        ),
+    )
+
+    parser.add_argument(
+        "--redact-secrets",
+        action="store_true",
+        help=(
+            "Redact sensitive information (API keys, tokens, passwords) in trace output. "
+            "Recommended when sharing traces externally. "
+            "Can also be enabled via STRIX_REDACT_SECRETS=1 environment variable."
+        ),
+    )
+
+    parser.add_argument(
+        "--trace-verbose",
+        action="store_true",
+        help=(
+            "Output human-readable trace to console (requires --non-interactive). "
+            "Shows tool calls, agent actions, and LLM activity in real-time."
+        ),
+    )
+
     args = parser.parse_args()
 
     if args.instruction and args.instruction_file:
         parser.error(
             "Cannot specify both --instruction and --instruction-file. Use one or the other."
         )
 
+    if args.trace_output and not args.trace:
+        parser.error("--trace-output requires --trace to be enabled.")
+
+    if args.trace_verbose and not args.non_interactive:
+        parser.error("--trace-verbose requires --non-interactive mode.")
+
+    # Check environment variables for tracing options
+    trace_env = Config.get("strix_trace")
+    if trace_env and trace_env.lower() in ("1", "true", "yes"):
+        args.trace = True
+
+    redact_env = Config.get("strix_redact_secrets")
+    if redact_env and redact_env.lower() in ("1", "true", "yes"):
+        args.redact_secrets = True
+
+    # Check environment variable for trace output path
+    if not args.trace_output:
+        args.trace_output = Config.get("strix_trace_output")
+
     if args.instruction_file:
         instruction_path = Path(args.instruction_file)
         try:
@@ -555,6 +614,25 @@ def main() -> None:
         has_instructions=bool(args.instruction),
     )
 
+    # Initialize live tracer if enabled
+    live_tracer: LiveTracer | None = None
+    trace_verbose = getattr(args, "trace_verbose", False)
+    if getattr(args, "trace", False) or trace_verbose:
+        live_tracer = LiveTracer(
+            output_path=getattr(args, "trace_output", None),
+            run_name=args.run_name,
+            redact_secrets=getattr(args, "redact_secrets", False),
+            verbose=trace_verbose,
+        )
+        set_live_tracer(live_tracer)
+
+        console = Console()
+        if not trace_verbose:
+            console.print(f"[dim]Live trace enabled:[/] {live_tracer.output_path}")
+            if getattr(args, "redact_secrets", False):
+                console.print("[dim]Secret redaction:[/] enabled")
+            console.print()
+
     exit_reason = "user_exit"
     try:
         if args.non_interactive:
@@ -572,6 +650,11 @@ def main() -> None:
         if tracer:
             posthog.end(tracer, exit_reason=exit_reason)
 
+        # Close live tracer
+        if live_tracer:
+            live_tracer.close()
+            set_live_tracer(None)
+
     results_path = Path("strix_runs") / args.run_name
     display_completion_message(args, results_path)
 

strix/interface/tui.py

@@ -743,12 +743,29 @@ def _build_agent_config(self, args: argparse.Namespace) -> dict[str, Any]:
     def _setup_cleanup_handlers(self) -> None:
         def cleanup_on_exit() -> None:
             from strix.runtime import cleanup_runtime
+            from strix.telemetry.live_tracer import get_live_tracer, set_live_tracer
 
             self.tracer.cleanup()
+
+            # Clean up live tracer
+            live_tracer = get_live_tracer()
+            if live_tracer:
+                live_tracer.close()
+                set_live_tracer(None)
+
             cleanup_runtime()
 
         def signal_handler(_signum: int, _frame: Any) -> None:
+            from strix.telemetry.live_tracer import get_live_tracer, set_live_tracer
+
             self.tracer.cleanup()
+
+            # Clean up live tracer
+            live_tracer = get_live_tracer()
+            if live_tracer:
+                live_tracer.close()
+                set_live_tracer(None)
+
             sys.exit(0)
 
         atexit.register(cleanup_on_exit)