add cpu sampling feature

HISTORY.md

@@ -1,5 +1,14 @@
 # History
 
+## 1.2.0 (2026-08-24)
+
+* New feature: **CPU stack sampling** — a watchdog daemon thread samples the loop thread's stack during CPU-bound slices, attaching aggregated stacks with counts to `SlowTaskEvent.cpu_stack_samples`, so `cpu_blocking` events carry attribution like IO events do
+* On by default: `detect_slow_tasks()` starts it automatically (`cpu_sampling=False` to opt out); `start_cpu_sampling()` called beforehand customizes it
+* Derived defaults: arming delay follows half the slow-task threshold; idle wake-up follows `min(50ms, max(arm, interval))` with a 1ms sleep floor, so no configuration can busy-loop the watchdog
+* Samples the thread that published the slice — event loops running outside the main thread are attributed correctly
+* Fork-safe: forked children (e.g. gunicorn `--preload` workers) restart the watchdog via `os.register_at_fork`
+* aiocop's own frames are excluded from samples via a package-path prefix match
+
 ## 1.1.5 (2026-07-20)
 
 * Fix `time.sleep` being double-counted on Python 3.13+, which emits a native audit event for it (#11, #12)

README.md

@@ -20,6 +20,7 @@
 * **Works with asyncio and uvloop**: Compatible with both standard asyncio and uvloop event loops out of the box
 * **Blocking I/O Detection**: Automatically detects blocking I/O calls (file operations, network calls, subprocess, etc.) in your async code
 * **Stack Trace Capture**: Captures full stack traces to pinpoint exactly where blocking calls originate
+* **CPU Stack Sampling**: A lightweight watchdog samples the loop thread during CPU-bound slices, so `cpu_blocking` events carry stack attribution too — on by default, no profiler needed
 * **Severity Scoring**: Assigns severity scores to blocking events to help prioritize fixes
 * **Callback-based Events**: Register callbacks to handle slow task events however you need (logging, metrics, alerts)
 * **Dynamic Controls**: Enable/disable monitoring at runtime, useful for gradual rollout or debugging sessions
@@ -212,6 +213,35 @@ async def test_my_async_endpoint(client):
 
 We wrap only the async view (not the entire test) because test setup/teardown often has legitimate blocking code. See [Integrations](https://feverup.github.io/aiocop/integrations/) for complete examples.
 
+## CPU Stack Sampling
+
+Blocking I/O gets stack attribution from audit events, but a `cpu_blocking` slice is just Python executing — nothing auditable fires. CPU stack sampling closes that gap: a watchdog daemon thread samples the loop thread's stack while a monitored callback has been running longer than an arming delay, and attaches the aggregated samples to the resulting `SlowTaskEvent` as `cpu_stack_samples`.
+
+**On by default.** `detect_slow_tasks()` starts it automatically. The arming delay defaults to half the slow-task threshold (and follows it if the threshold changes), so any slice that goes on to violate has been under sampling since its midpoint.
+
+```python
+# Disable it:
+aiocop.detect_slow_tasks(threshold_ms=30, cpu_sampling=False)
+
+# Customize it — call BEFORE detect_slow_tasks() (the auto-start then steps aside):
+aiocop.start_cpu_sampling(interval_ms=5, arm_after_ms=10)
+aiocop.detect_slow_tasks(threshold_ms=30)
+```
+
+Reading the result in a callback:
+
+```python
+def on_slow_task(event: aiocop.SlowTaskEvent) -> None:
+    if event.reason == "cpu_blocking" and event.cpu_stack_samples:
+        top = event.cpu_stack_samples[0]
+        print(f"CPU-bound slice ({event.elapsed_ms:.1f}ms), hottest stack "
+              f"({top['count']} samples): {top['trace']}")
+```
+
+**Overhead**: the hot path adds two module-global stores per monitored callback (~0.1µs); the watchdog costs well under 1% of a core when idle and captures at most `max_samples_per_slice` (default 32) stacks per slice — and only for slices that are already frozen. Sampling works on any thread the loop runs on and survives `fork()` (gunicorn `--preload` workers restart the watchdog automatically).
+
+**Known limitation**: a single long-running C call that never releases the GIL starves the watchdog — few samples for a long slice is itself a signal that one C-level call dominated it.
+
 ## Context Providers
 
 Context providers allow you to capture external context (like tracing spans, request IDs, etc.) that will be passed to your callbacks. The context is captured **within the asyncio task's context**, ensuring proper propagation of contextvars.
@@ -306,6 +336,7 @@ class SlowTaskEvent:
     reason: str              # "io_blocking" or "cpu_blocking"
     blocking_events: list[BlockingEventInfo]  # List of detected events (empty for cpu_blocking)
     context: dict[str, Any]  # Custom context from context providers (default: {})
+    cpu_stack_samples: list[CpuStackSample]   # Aggregated loop-thread stack samples (default: [])
 ```
 
 ### BlockingEventInfo
@@ -320,6 +351,19 @@ class BlockingEventInfo(TypedDict):
     severity: int     # Weight of this event
 ```
 
+### CpuStackSample
+
+Aggregated stack sample captured during a CPU-bound slice (see [CPU Stack Sampling](#cpu-stack-sampling)):
+
+```python
+class CpuStackSample(TypedDict):
+    trace: str        # Stack trace ("frame <- frame <- ...")
+    entry_point: str  # First frame in the trace
+    count: int        # How many samples showed this exact stack
+```
+
+Samples are ordered by `count` descending — the first entry is where the slice most likely spent its CPU time.
+
 ## Severity Weights
 
 Events are classified by severity:
@@ -342,7 +386,9 @@ Severity levels are determined by aggregate score:
 
 - `patch_audit_functions()` - Patches stdlib functions to emit audit events
 - `start_blocking_io_detection(trace_depth=20)` - Registers the audit hook
-- `detect_slow_tasks(threshold_ms=30, on_slow_task=None)` - Patches the event loop
+- `detect_slow_tasks(threshold_ms=30, on_slow_task=None, cpu_sampling=True)` - Patches the event loop; starts CPU stack sampling unless disabled
+- `start_cpu_sampling(interval_ms=10, arm_after_ms=None, idle_interval_ms=None, max_samples_per_slice=32, trace_depth=20)` - Start (or pre-configure) CPU stack sampling
+- `is_cpu_sampling_started()` - Whether the sampling watchdog is running
 - `activate()` / `deactivate()` - Control monitoring at runtime
 
 ### Callback Management

aiocop/__init__.py

@@ -17,6 +17,10 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
         if event.exceeded_threshold:
             print(f"Slow task: {event.elapsed_ms}ms, severity: {event.severity_level}")
 
+    # Also starts CPU stack sampling (cpu_sampling=True by default), so
+    # cpu_blocking events carry stack attribution like IO events do. To
+    # customize sampling, call aiocop.start_cpu_sampling(...) BEFORE this;
+    # to disable it, pass cpu_sampling=False.
     aiocop.detect_slow_tasks(threshold_ms=30, on_slow_task=my_callback)
 
     # 4. Activate monitoring when ready (e.g., after startup)
@@ -71,6 +75,8 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
     - reason: str - Why the task was flagged ("io_blocking" or "cpu_blocking")
     - blocking_events: list[BlockingEventInfo] - Details of each blocking operation
     - context: dict[str, Any] - Custom context from registered context providers
+    - cpu_stack_samples: list[CpuStackSample] - Aggregated main-thread stack samples
+      captured during the slice (empty unless start_cpu_sampling() was called)
 
 Severity Levels:
     aiocop calculates severity based on the type and number of blocking operations:
@@ -108,6 +114,7 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
     unregister_context_provider,
     unregister_slow_task_callback,
 )
+from aiocop.core.cpu_sampler import is_cpu_sampling_started, start_cpu_sampling
 from aiocop.core.severity import calculate_io_severity_score, get_severity_level_from_score
 from aiocop.core.slow_tasks import SlowTaskCallback, detect_slow_tasks, get_slow_task_threshold_ms
 from aiocop.core.state import (
@@ -122,7 +129,7 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
     raise_on_violations_context as raise_on_violations,
 )
 from aiocop.exceptions import HighSeverityBlockingIoException
-from aiocop.types.events import BlockingEventInfo, RawBlockingEvent, SlowTaskEvent
+from aiocop.types.events import BlockingEventInfo, CpuStackSample, RawBlockingEvent, SlowTaskEvent
 from aiocop.types.severity import (
     THRESHOLD_HIGH,
     THRESHOLD_LOW,
@@ -139,6 +146,8 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
     "patch_audit_functions",
     "start_blocking_io_detection",
     "detect_slow_tasks",
+    "start_cpu_sampling",
+    "is_cpu_sampling_started",
     # Activation controls
     "activate",
     "deactivate",
@@ -167,6 +176,7 @@ def my_callback(event: aiocop.SlowTaskEvent) -> None:
     "get_severity_level_from_score",
     # Types
     "BlockingEventInfo",
+    "CpuStackSample",
     "RawBlockingEvent",
     "SlowTaskEvent",
     "SlowTaskCallback",

aiocop/core/blocking_io.py

@@ -220,26 +220,31 @@ def format_blocking_event(raw_event: RawBlockingEvent) -> BlockingEventInfo:
 
     event_str = f"{raw_event['event_name']}({formatted_args})"
 
-    formatted_frames = []
-    for filename, lineno, func_name in raw_event["raw_stack"]:
-        short_path = filename
-        if "/src/" in filename:
-            short_path = filename.split("/src/", 1)[1]
-        elif "/site-packages/" in filename:
-            short_path = filename.split("/site-packages/", 1)[1]
-        elif "/lib/python" in filename:
-            path_parts = filename.split("/")
-            if len(path_parts) > 0:
-                short_path = path_parts[-1]
-
-        formatted_frames.append(f"{short_path}:{lineno}:{func_name}")
+    formatted_frames = [format_stack_frame(frame) for frame in raw_event["raw_stack"]]
 
     trace_str = " <- ".join(formatted_frames)
     entry_point = formatted_frames[0] if len(formatted_frames) > 0 else "unknown"
 
     return {"event": event_str, "trace": trace_str, "entry_point": entry_point, "severity": raw_event["severity"]}
 
 
+def format_stack_frame(frame: tuple[str, int, str]) -> str:
+    """Format a raw (filename, lineno, func_name) frame with a shortened path."""
+    filename, lineno, func_name = frame
+
+    short_path = filename
+    if "/src/" in filename:
+        short_path = filename.split("/src/", 1)[1]
+    elif "/site-packages/" in filename:
+        short_path = filename.split("/site-packages/", 1)[1]
+    elif "/lib/python" in filename:
+        path_parts = filename.split("/")
+        if len(path_parts) > 0:
+            short_path = path_parts[-1]
+
+    return f"{short_path}:{lineno}:{func_name}"
+
+
 def get_blocking_events_dict() -> dict[str, int]:
     """Return a copy of the blocking events dictionary with their severity weights."""
     return BLOCKING_EVENTS_DICT.copy()