feat: stream CLI log transcripts and run status to Socket backend

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2.2.87
+
+- Added a streaming log channel between the CLI and the Socket backend. Each CLI invocation now reports a per-run status (`in_progress` / `success` / `failure` / `cancelled`) and uploads a transcript of its own log output, visible in the Socket admin views. The transcript is captured regardless of the local `--enable-debug` state; the existing terminal verbosity is unchanged. The feature is best-effort — registration or upload failures silently degrade and never block the scan. Opt out with `--disable-server-log-streaming`.
+
 ## 2.2.83
 
 - Fixed branch detection in detached-HEAD CI checkouts. When `git name-rev --name-only HEAD` returned an output with a suffix operator (e.g. `remotes/origin/master~1`, `master^0`), the `~N`/`^N` was previously passed through as the branch name and rejected by the Socket API as an invalid Git ref. The suffix is now stripped before the prefix split, producing the bare branch name.

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "socketsecurity"
-version = "2.2.86"
+version = "2.2.87"
 requires-python = ">= 3.11"
 license = {"file" = "LICENSE"}
 dependencies = [

socketsecurity/__init__.py

@@ -1,3 +1,3 @@
 __author__ = 'socket.dev'
-__version__ = '2.2.86'
+__version__ = '2.2.87'
 USER_AGENT = f'SocketPythonCLI/{__version__}'

socketsecurity/config.py

@@ -92,6 +92,7 @@ class CliConfig:
     ignore_commit_files: bool = False
     disable_blocking: bool = False
     disable_ignore: bool = False
+    disable_server_log_streaming: bool = False
     strict_blocking: bool = False
     integration_type: IntegrationType = "api"
     integration_org_slug: Optional[str] = None
@@ -207,6 +208,7 @@ def from_args(cls, args_list: Optional[List[str]] = None) -> 'CliConfig':
             'ignore_commit_files': args.ignore_commit_files,
             'disable_blocking': args.disable_blocking,
             'disable_ignore': args.disable_ignore,
+            'disable_server_log_streaming': args.disable_server_log_streaming,
             'strict_blocking': args.strict_blocking,
             'integration_type': args.integration,
             'pending_head': args.pending_head,
@@ -716,6 +718,18 @@ def create_argument_parser() -> argparse.ArgumentParser:
         action="store_true",
         help=argparse.SUPPRESS
     )
+    advanced_group.add_argument(
+        "--disable-server-log-streaming",
+        dest="disable_server_log_streaming",
+        action="store_true",
+        help="Disable streaming server-side log lines to the terminal during long-running CLI operations."
+    )
+    advanced_group.add_argument(
+        "--disable_server_log_streaming",
+        dest="disable_server_log_streaming",
+        action="store_true",
+        help=argparse.SUPPRESS
+    )
     advanced_group.add_argument(
         "--strict-blocking",
         dest="strict_blocking",

socketsecurity/core/cli_run.py

@@ -0,0 +1,63 @@
+"""Lifecycle helpers for a CLI run on the Socket backend.
+
+A "run" represents a single CLI invocation. `register_cli_run` opens it and
+returns a server-issued `run_id`; `finalize_cli_run` closes it on exit. The
+run_id keys the rows that `BatchedLogUploader` POSTs to
+`/python-cli-runs/<run_id>/logs` during the run so the dashboard can show
+what the user saw in their terminal.
+
+Both calls are best-effort: failures fall back to no-streaming and never
+prevent the scan from running.
+"""
+
+import json
+import logging
+from typing import Optional
+
+from .cli_client import CliClient
+from .exceptions import APIFailure
+
+log = logging.getLogger("socketcli")
+
+
+def register_cli_run(
+    client: CliClient,
+    client_version: str,
+) -> Optional[str]:
+    try:
+        resp = client.request(
+            path="python-cli-runs",
+            method="POST",
+            payload=json.dumps({"client_version": client_version}),
+        )
+    except APIFailure as e:
+        log.debug(f"cli-run register failed (streaming disabled): {e}")
+        return None
+
+    try:
+        body = resp.json()
+    except (ValueError, json.JSONDecodeError) as e:
+        log.debug(f"cli-run register: bad JSON body: {e}")
+        return None
+
+    run_id = body.get("run_id")
+    if not isinstance(run_id, str) or not run_id:
+        log.debug(f"cli-run register: missing run_id in response: {body!r}")
+        return None
+    return run_id
+
+
+def finalize_cli_run(
+    client: CliClient,
+    run_id: str,
+    status: str = "success",
+    report_run_id: Optional[str] = None,
+) -> None:
+    try:
+        client.request(
+            path=f"python-cli-runs/{run_id}/finalize",
+            method="POST",
+            payload=json.dumps({"status": status, "report_run_id": report_run_id}),
+        )
+    except Exception as e:
+        log.debug(f"cli-run finalize failed (swallowed): {e}")

socketsecurity/core/log_uploader.py

@@ -0,0 +1,122 @@
+"""Buffer the CLI's local log records and POST them in batches to
+/python-cli-runs/<run_id>/logs so the dashboard's view of a CLI run
+mirrors what the user sees in their terminal.
+
+Behavior:
+- daemon thread, 5s flush
+- swallow all network errors (debug log only)
+- skip empty buffers
+- drain on shutdown
+- at-most-once semantics (failed batches dropped, not retried)
+
+A thread-local recursion guard prevents the uploader's own request-error
+log lines (emitted by `cli_client.py`'s `socketdev` logger) from being
+re-enqueued during a flush.
+"""
+
+import json
+import logging
+import threading
+from datetime import datetime, timezone
+from typing import Optional
+
+from .cli_client import CliClient
+
+log = logging.getLogger(__name__)
+
+_FLUSH_GUARD = threading.local()
+
+_LEVEL_MAP = {
+    logging.DEBUG: "DEBUG",
+    logging.INFO: "INFO",
+    logging.WARNING: "WARN",
+    logging.ERROR: "ERROR",
+    logging.CRITICAL: "ERROR",
+}
+
+
+def _now_str() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+
+
+class BatchedLogUploader:
+    def __init__(
+        self,
+        client: CliClient,
+        run_id: str,
+        flush_interval: float = 5.0,
+    ):
+        self._client = client
+        self._run_id = run_id
+        self._flush_interval = flush_interval
+        self._buf: list = []
+        self._lock = threading.Lock()
+        self._stop = threading.Event()
+        self._thread: Optional[threading.Thread] = None
+
+    def add(self, entry: dict) -> None:
+        with self._lock:
+            self._buf.append(entry)
+
+    def start(self) -> None:
+        if self._thread is not None:
+            return
+        self._thread = threading.Thread(
+            target=self._run,
+            name=f"socket-log-uploader-{self._run_id[:8]}",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def stop(self, timeout: float = 2.0) -> None:
+        if self._thread is None:
+            self._flush()
+            return
+        self._stop.set()
+        self._thread.join(timeout=timeout)
+        self._thread = None
+        self._flush()
+
+    def _run(self) -> None:
+        while not self._stop.is_set():
+            self._flush()
+            self._stop.wait(self._flush_interval)
+
+    def _flush(self) -> None:
+        with self._lock:
+            if not self._buf:
+                return
+            batch = self._buf
+            self._buf = []
+
+        _FLUSH_GUARD.active = True
+        try:
+            self._client.request(
+                path=f"python-cli-runs/{self._run_id}/logs",
+                method="POST",
+                payload=json.dumps({"logs": batch}),
+            )
+        except Exception as e:
+            log.debug(f"log upload failed (swallowed, {len(batch)} entries dropped): {e}")
+        finally:
+            _FLUSH_GUARD.active = False
+
+
+class UploadingLogHandler(logging.Handler):
+    def __init__(self, uploader: BatchedLogUploader, context: str = "socket-python-cli"):
+        super().__init__()
+        self._uploader = uploader
+        self._context = context
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if getattr(_FLUSH_GUARD, "active", False):
+            return
+        try:
+            self._uploader.add({
+                "timestamp": _now_str(),
+                "level": _LEVEL_MAP.get(record.levelno, "INFO"),
+                "message": self.format(record),
+                "context": self._context,
+            })
+        except Exception:
+            self.handleError(record)

socketsecurity/core/streaming.py

@@ -0,0 +1,83 @@
+"""Wire the server log streaming pipeline for one CLI run.
+
+`setup_streaming` registers the run with the backend, attaches handlers that
+route the CLI's own log output through both the local terminal and a batched
+uploader, and forces the loggers into DEBUG so the upload captures everything
+regardless of local terminal verbosity.
+
+Returns a teardown callable to invoke on exit (typically via `atexit.register`).
+Returns None if registration failed; in that case nothing was wired up.
+"""
+
+import logging
+from typing import Callable, Optional
+
+from .cli_client import CliClient
+from .cli_run import finalize_cli_run, register_cli_run
+from .log_uploader import BatchedLogUploader, UploadingLogHandler
+
+_run_status: str = "success"
+_report_run_id: Optional[str] = None
+
+
+def set_run_status(status: str) -> None:
+    global _run_status
+    _run_status = status
+
+
+def set_report_run_id(report_run_id: Optional[str]) -> None:
+    global _report_run_id
+    _report_run_id = report_run_id
+
+
+def setup_streaming(
+    *,
+    client: CliClient,
+    cli_logger: logging.Logger,
+    sdk_logger: logging.Logger,
+    client_version: str,
+    enable_debug: bool,
+) -> Optional[Callable[[], None]]:
+    run_id = register_cli_run(
+        client,
+        client_version=client_version,
+    )
+    if not run_id:
+        cli_logger.debug("server log streaming disabled (register failed)")
+        return None
+
+    log_uploader = BatchedLogUploader(client, run_id)
+    log_uploader.start()
+    upload_handler = UploadingLogHandler(log_uploader, context="socket-python-cli")
+    upload_handler.setFormatter(logging.Formatter("%(message)s"))
+
+    terminal_handler = logging.StreamHandler()
+    terminal_handler.setLevel(logging.DEBUG if enable_debug else logging.INFO)
+    terminal_handler.setFormatter(logging.Formatter("%(asctime)s: %(message)s"))
+
+    saved_levels = (cli_logger.level, sdk_logger.level)
+    saved_propagate = (cli_logger.propagate, sdk_logger.propagate)
+    cli_logger.setLevel(logging.DEBUG)
+    sdk_logger.setLevel(logging.DEBUG)
+    cli_logger.propagate = False
+    sdk_logger.propagate = False
+    cli_logger.addHandler(terminal_handler)
+    sdk_logger.addHandler(terminal_handler)
+    cli_logger.addHandler(upload_handler)
+    sdk_logger.addHandler(upload_handler)
+
+    cli_logger.debug(f"server log streaming enabled (run_id={run_id})")
+
+    def teardown() -> None:
+        cli_logger.removeHandler(upload_handler)
+        sdk_logger.removeHandler(upload_handler)
+        log_uploader.stop()
+        finalize_cli_run(client, run_id, status=_run_status, report_run_id=_report_run_id)
+        cli_logger.removeHandler(terminal_handler)
+        sdk_logger.removeHandler(terminal_handler)
+        cli_logger.setLevel(saved_levels[0])
+        sdk_logger.setLevel(saved_levels[1])
+        cli_logger.propagate = saved_propagate[0]
+        sdk_logger.propagate = saved_propagate[1]
+
+    return teardown

socketsecurity/socketcli.py

@@ -1,3 +1,4 @@
+import atexit
 import json
 import os
 import sys
@@ -20,6 +21,7 @@
 from socketsecurity.core.messages import Messages
 from socketsecurity.core.scm_comments import Comments
 from socketsecurity.core.socket_config import SocketConfig
+from socketsecurity.core.streaming import set_report_run_id, set_run_status, setup_streaming
 from socketsecurity.output import OutputHandler
 
 socket_logger, log = initialize_logging()
@@ -30,13 +32,19 @@ def cli():
     try:
         main_code()
     except KeyboardInterrupt:
+        set_run_status("cancelled")
         log.info("Keyboard Interrupt detected, exiting")
         config = CliConfig.from_args()  # Get current config
         if not config.disable_blocking:
             sys.exit(2)
         else:
             sys.exit(0)
+    except SystemExit as e:
+        if e.code:
+            set_run_status("failure")
+        raise
     except Exception as error:
+        set_run_status("failure")
         log.error("Unexpected error when running the cli")
         log.error(error)
         traceback.print_exc()
@@ -89,6 +97,18 @@ def main_code():
     client = CliClient(socket_config)
     sdk.api.api_url = socket_config.api_url
     log.debug("loaded client")
+
+    if not config.disable_server_log_streaming:
+        teardown = setup_streaming(
+            client=client,
+            cli_logger=log,
+            sdk_logger=socket_logger,
+            client_version=config.version,
+            enable_debug=config.enable_debug,
+        )
+        if teardown:
+            atexit.register(teardown)
+
     core = Core(socket_config, sdk, config)
     log.debug("loaded core")
 
@@ -741,6 +761,9 @@ def _is_unprocessed(c):
             )
             output_handler.handle_output(diff)
 
+    if diff.id not in ("NO_DIFF_RAN", "NO_SCAN_RAN"):
+        set_report_run_id(diff.id)
+
     # Handle license generation
     if not should_skip_scan and diff.id != "NO_DIFF_RAN" and diff.id != "NO_SCAN_RAN" and config.generate_license:
         all_packages = {}