Class RuntimeLogService

app/runtime_logging.py:86–308

Persist structured JSONL flow events alongside standard logs.

class RuntimeLogService:
    """Persist structured JSONL flow events alongside standard logs."""

    def __init__(self, settings: Settings | None = None) -> None:
        self.settings = settings or get_settings()
        self._lock = threading.Lock()
        self.logger = logging.getLogger("glm_desk.runtime")
        configure_logging(self.settings)
        (self.settings.runtime_logs_dir / "accounts").mkdir(parents=True, exist_ok=True)

    def start_run(
        self,
        *,
        account_id: str,
        action: str,
        source: str = "",
        product_id: str = "",
        pay_type: str = "",
        details: dict[str, Any] | None = None,
    ) -> FlowRun:
        run = FlowRun(
            run_id=self._make_run_id(),
            account_id=account_id,
            action=action.strip() or "flow",
            source=source.strip(),
            product_id=product_id.strip(),
            pay_type=pay_type.strip(),
        )
        self.log_event(
            run,
            stage="run",
            status="started",
            message=f"{run.action} started",
            details=details,
        )
        return run

    def finish_run(
        self,
        run: FlowRun,
        *,
        status: str,
        message: str,
        details: dict[str, Any] | None = None,
        level: int | None = None,
    ) -> None:
        resolved_level = level if level is not None else (logging.ERROR if status == "failed" else logging.INFO)
        self.log_event(
            run,
            stage="run",
            status=status,
            message=message,
            details=details,
            level=resolved_level,
        )

    def log_event(
        self,