feat: centralized logging with --verbose/--quiet/--log-file/--output json CLI flags

p2m/cli.py

@@ -17,6 +17,7 @@
 
 from p2m.core.io import load_json, load_jsonl, get_permissible_flag
 from p2m.core.judge import get_verdict_dimension, infer_judge_status, is_valid_event_flag
+from p2m.logging_config import configure_logging
 from p2m.stages import STAGE_NAMES
 
 ROOT = Path(__file__).resolve().parent.parent
@@ -490,8 +491,33 @@ def _subrisk_metric_map(rows: Iterable[dict[str, Any]], metric: str) -> dict[str
     ),
 )
 @click.version_option(version="0.1.0", prog_name="p2m")
-def cli():
+@click.option("-v", "--verbose", is_flag=True, help="Enable debug-level logging.")
+@click.option("-q", "--quiet", is_flag=True, help="Suppress info-level output; show only warnings and errors.")
+@click.option(
+    "--log-file",
+    type=click.Path(path_type=Path),
+    default=None,
+    help="Write all log output to a file (in addition to stderr).",
+)
+@click.option(
+    "--output",
+    "output_format",
+    type=click.Choice(["text", "json"], case_sensitive=False),
+    default="text",
+    show_default=True,
+    help="Log output format. Use 'json' for CI pipelines.",
+)
+@click.pass_context
+def cli(ctx: click.Context, verbose: bool, quiet: bool, log_file: Path | None, output_format: str):
     """Safety evaluation workflows for pipeline runs, artifacts, and post-hoc analysis."""
+    ctx.ensure_object(dict)
+    ctx.obj["logging_configured"] = True
+    configure_logging(
+        verbose=verbose,
+        quiet=quiet,
+        log_file=log_file,
+        json_output=(output_format == "json"),
+    )
 
 
 @cli.command(short_help="Run a pipeline from a YAML config")
@@ -514,12 +540,43 @@ def cli():
     show_envvar=True,
 )
 @click.option("--strict", is_flag=True, help="Fail on malformed JSONL inputs instead of skipping bad rows.")
+@click.option("-v", "--verbose", is_flag=True, help="Enable debug-level logging.")
+@click.option("-q", "--quiet", is_flag=True, help="Suppress info-level output; show only warnings and errors.")
+@click.option(
+    "--log-file",
+    type=click.Path(path_type=Path),
+    default=None,
+    help="Write all log output to a file (in addition to stderr).",
+)
+@click.option(
+    "--output",
+    "output_format",
+    type=click.Choice(["text", "json"], case_sensitive=False),
+    default="text",
+    show_default=True,
+    help="Log output format. Use 'json' for CI pipelines.",
+)
+@click.pass_context
 def run(
+    ctx: click.Context,
     config: Path,
     force_stage: tuple[str, ...],
     strict: bool,
+    verbose: bool,
+    quiet: bool,
+    log_file: Path | None,
+    output_format: str,
 ):
     """Run the evaluation pipeline."""
+    # Re-configure logging if flags were passed on the subcommand
+    # (e.g. `p2m run --verbose` instead of `p2m --verbose run`).
+    if verbose or quiet or log_file or output_format != "text":
+        configure_logging(
+            verbose=verbose,
+            quiet=quiet,
+            log_file=log_file,
+            json_output=(output_format == "json"),
+        )
     runner = _load_runner_module()
     rc = runner.run_pipeline(
         config=str(config),

p2m/config.py

@@ -4,12 +4,14 @@
 
 from datetime import datetime, timezone
 from pathlib import Path
-import sys
 from typing import Any
 
+import logging
 import re
 import yaml
 
+log = logging.getLogger(__name__)
+
 from p2m.core.config_model import (
     DEFAULT_AUDITOR_MAX_TURNS,
     DEFAULT_JUDGE_MAX_TOKENS,
@@ -455,7 +457,7 @@ def _parse_top_level_factors(raw: Any) -> list[dict[str, Any]] | None:
     if not isinstance(raw, list):
         raise ValueError("factors must be a list")
     if len(raw) > 10:
-        print("warning: factors defines more than 10 factors", file=sys.stderr)
+        log.warning("factors defines more than 10 factors")
 
     factors: list[dict[str, Any]] = []
     seen_names: set[str] = set()
@@ -491,7 +493,7 @@ def _parse_top_level_factors(raw: Any) -> list[dict[str, Any]] | None:
             if len(levels_raw) == 1:
                 raise ValueError("single-level factor adds no variation")
             if len(levels_raw) > 20:
-                print(f"warning: factor '{name}' defines more than 20 levels", file=sys.stderr)
+                log.warning(f"factor '{name}' defines more than 20 levels")
             levels = []
             seen_level_names: set[str] = set()
             for level_index, level_raw in enumerate(levels_raw, start=1):

p2m/core/async_utils.py

@@ -3,10 +3,15 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import inspect
+import logging
+import time
 from collections.abc import Awaitable, Callable, Sequence
 from typing import Any, TypeVar
 
+log = logging.getLogger(__name__)
+
 ItemT = TypeVar("ItemT")
 ResultT = TypeVar("ResultT")
 
@@ -46,3 +51,32 @@ async def _guard(item: ItemT) -> ResultT:
             return await worker(item)
 
     return await asyncio.gather(*(_guard(item) for item in items))
+
+
+@contextlib.asynccontextmanager
+async def log_heartbeat(message: str, *, interval: float = 15.0):
+    """Async context manager that logs a heartbeat while a long operation runs.
+
+    Usage::
+
+        async with log_heartbeat("Converting policy"):
+            result = await slow_llm_call(...)
+
+    Logs "{message} (still working, {elapsed}s elapsed)" every *interval*
+    seconds until the block exits.
+    """
+    start = time.monotonic()
+
+    async def _beat():
+        while True:
+            await asyncio.sleep(interval)
+            elapsed = time.monotonic() - start
+            log.info(f"{message} (still working, {elapsed:.0f}s elapsed)")
+
+    task = asyncio.create_task(_beat())
+    try:
+        yield
+    finally:
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task

p2m/core/model_client.py

@@ -247,6 +247,7 @@ def normalize_response(
             _get_value(choice, "finish_reason")
             or _get_value(raw_response, "stop_reason")
             or _get_value(_get_value(raw_response, "incomplete_details"), "reason")
+            or _get_value(raw_response, "status")
         ),
         status=_get_value(raw_response, "status"),
         incomplete_details=_get_value(raw_response, "incomplete_details"),
@@ -514,10 +515,17 @@ def _first_choice(raw_response: Any) -> Any:
 def _normalize_usage(raw_usage: Any) -> UsageStats | None:
     if raw_usage is None:
         return None
+    # Chat Completions API uses prompt_tokens/completion_tokens;
+    # Responses API uses input_tokens/output_tokens.
+    prompt = _coerce_int(_get_value(raw_usage, "prompt_tokens")) or _coerce_int(_get_value(raw_usage, "input_tokens"))
+    completion = _coerce_int(_get_value(raw_usage, "completion_tokens")) or _coerce_int(_get_value(raw_usage, "output_tokens"))
+    total = _coerce_int(_get_value(raw_usage, "total_tokens"))
+    if total is None and prompt is not None and completion is not None:
+        total = prompt + completion
     return UsageStats(
-        prompt_tokens=_coerce_int(_get_value(raw_usage, "prompt_tokens")),
-        completion_tokens=_coerce_int(_get_value(raw_usage, "completion_tokens")),
-        total_tokens=_coerce_int(_get_value(raw_usage, "total_tokens")),
+        prompt_tokens=prompt,
+        completion_tokens=completion,
+        total_tokens=total,
         raw=raw_usage,
     )
 
@@ -587,6 +595,20 @@ def _get_value(obj: Any, key: str) -> Any:
 
 # ── Public API ─────────────────────────────────────────────────
 
+
+def _log_response(label: str, model: str, response: "ModelResponse", elapsed: float, **extra: object) -> None:
+    """Log a compact one-line summary of an LLM call at DEBUG level."""
+    usage = response.usage
+    if usage and usage.prompt_tokens is not None:
+        tokens = f"{usage.prompt_tokens}+{usage.completion_tokens or 0} tokens"
+    else:
+        tokens = "? tokens"
+    parts = [f"model={model}"]
+    for key, value in extra.items():
+        parts.append(f"{key}={value}")
+    parts.extend([f"{elapsed:.1f}s", tokens, f"finish={response.finish_reason or '?'}"])
+    log.debug(f"{label}: {', '.join(parts)}")
+
 __all__ = [
     "build_llm_call_trace",
     "GenerateOptions",
@@ -790,6 +812,8 @@ async def generate(
     """Run a standard async text generation call."""
     resolved_options = options or GenerateOptions()
     litellm = _get_litellm_module()
+    api_mode = "responses" if resolved_options.web_search else "chat_completion"
+    t0 = time.monotonic()
 
     if resolved_options.web_search:
         payload = _build_responses_payload(model, messages, resolved_options)
@@ -817,11 +841,13 @@ async def _call() -> Any:
             )
 
     raw_response = await _with_retries(_call, model=model, label=resolved_options.call_label)
-    return normalize_response(
+    result = normalize_response(
         raw_response,
         api_mode="responses" if resolved_options.web_search else "chat_completion",
         request_payload=payload,
     )
+    _log_response("generate", model, result, time.monotonic() - t0, api_mode=api_mode)
+    return result
 
 
 async def generate_structured(
@@ -835,6 +861,8 @@ async def generate_structured(
     """Run a structured generation call constrained by a JSON schema."""
     resolved_options = options or GenerateOptions()
     litellm = _get_litellm_module()
+    api_mode = "responses" if resolved_options.web_search else "chat_completion"
+    t0 = time.monotonic()
 
     if resolved_options.web_search:
         payload = _build_responses_payload(model, messages, resolved_options)
@@ -872,11 +900,13 @@ async def _call() -> Any:
             )
 
     raw_response = await _with_retries(_call, model=model, label=resolved_options.call_label)
-    return normalize_response(
+    result = normalize_response(
         raw_response,
         api_mode="responses" if resolved_options.web_search else "chat_completion",
         request_payload=payload,
     )
+    _log_response("generate_structured", model, result, time.monotonic() - t0, api_mode=api_mode, schema=schema_name)
+    return result
 
 
 async def generate_with_tools(
@@ -888,6 +918,7 @@ async def generate_with_tools(
 ) -> ModelResponse:
     """Run a tool-capable chat completion."""
     resolved_options = options or GenerateOptions()
+    t0 = time.monotonic()
     payload = _build_chat_payload(model, messages, resolved_options)
     payload["tools"] = tools
     if resolved_options.tool_choice is not None:
@@ -902,8 +933,10 @@ async def _call() -> Any:
         )
 
     raw_response = await _with_retries(_call, model=model, label=resolved_options.call_label)
-    return normalize_response(
+    result = normalize_response(
         raw_response,
         api_mode="chat_completion",
         request_payload=payload,
     )
+    _log_response("generate_with_tools", model, result, time.monotonic() - t0, tools=len(tools))
+    return result

p2m/core/transcript.py

@@ -7,6 +7,8 @@
 import json
 import logging
 from dataclasses import dataclass, field
+
+log = logging.getLogger(__name__)
 from enum import Enum
 from pathlib import Path
 from typing import Any, Dict, List, Literal, Optional, Union
@@ -288,7 +290,7 @@ def _transcript_from_dict(data: Dict[str, Any]) -> "Transcript":
         try:
             llm_calls.append(LLMCallTrace(**call_data))
         except Exception:
-            logging.warning("Skipping malformed LLM call trace in transcript")
+            log.warning("Skipping malformed LLM call trace in transcript")
             continue
     return Transcript(
         metadata=_metadata_from_dict(data),
@@ -524,7 +526,7 @@ def load_jsonl(cls, path: Path) -> List["Transcript"]:
                 try:
                     transcripts.append(_transcript_from_dict(json.loads(line)))
                 except Exception:
-                    logging.warning("Skipping malformed JSONL line in %s", path)
+                    log.warning(f"Skipping malformed JSONL line in {path}")
         return transcripts