fix(diagnostics): bound in-flight trim skip

Author: coreyt

airlock/callbacks/oom_diagnostics.py

@@ -342,6 +342,25 @@ def _signal_usr1(self, _signum: int, _frame: Any) -> None:
     def _signal_usr2(self, _signum: int, _frame: Any) -> None:
         self._signal_snapshot("signal_usr2", trim=True)
 
+    def _record_inflight_trim_skip(self, phase: str, in_flight: int) -> None:
+        """Record why a trim was skipped without perturbing the live workload.
+
+        A full signal snapshot enumerates the Python heap.  That work is
+        specifically unhelpful while requests are in flight, and can hold the
+        signal gate long enough to make a second operator signal look stuck.
+        Preserve the bounded audit event while deliberately omitting sampled
+        process details.
+        """
+        self._append(
+            {
+                "ts_monotonic_ns": time.monotonic_ns(),
+                "phase": f"{phase}_skipped_in_flight",
+                "sequence": None,
+                "in_flight": in_flight,
+                "trim": {"attempted": False, "reason": "requests_in_flight"},
+            }
+        )
+
     def _signal_snapshot(self, phase: str, *, trim: bool) -> None:
         if not _enabled() or not self._signal_lock.acquire(blocking=False):
             return
@@ -351,7 +370,7 @@ def worker() -> None:
                 with self._lock:
                     in_flight = self._in_flight
                 if trim and in_flight:
-                    self.snapshot(f"{phase}_skipped_in_flight")
+                    self._record_inflight_trim_skip(phase, in_flight)
                     return
                 self.snapshot(f"{phase}_before")
                 if trim:

dev/debugging/instrumentation/oom-high-water.md

@@ -50,10 +50,13 @@ Each record has a monotonic timestamp and a phase:
 - `periodic` — every `AIRLOCK_OOM_DIAGNOSTICS_EVERY` requests.
 - `signal_usr1_*` or `signal_usr2_*` — an operator snapshot.
 
-The payload includes cgroup current/peak/high/max/event counters, process RSS,
+Most payloads include cgroup current/peak/high/max/event counters, process RSS,
 `smaps_rollup` anonymous/huge-page values, PSI, glibc `mallinfo2`, thread/FD
 counts, optional tracemalloc totals, aggregate GC type counts at checkpoints,
-and LiteLLM/httpx client-pool counts.
+and LiteLLM/httpx client-pool counts. `signal_usr2_skipped_in_flight` is the
+deliberate exception: it contains only its timestamp, phase, in-flight count,
+and `trim` decline reason. It does not scan the heap or sample process state
+while live requests make trimming unsafe.
 
 Important: `in_flight` is decremented at `callback_complete`, not at
 `provider_response`. A growing value can therefore mean a stuck callback or

docs/operations.md

@@ -221,12 +221,15 @@ AIRLOCK_OOM_DIAGNOSTICS_MAX_RECORDS=6000
 ```
 
 It writes aggregate process, allocator, file-descriptor, thread, cgroup, and
-memory-pressure counters. It does not write request/response bodies, headers,
-model names, exception text, or client metadata. The directory defaults to
-`/tmp/airlock-oom-diagnostics`; output files are owner-only. Secure or
-pre-create a configured directory with the permissions your deployment needs.
-Disable the recorder after the investigation. The repository's engineering OOM
-instrumentation runbook contains the signal and high-water procedure.
+memory-pressure counters. A `SIGUSR2` trim declined because requests are active
+is intentionally recorded as a minimal audit event instead, so it does not
+perturb those requests with a heap scan. It does not write request/response
+bodies, headers, model names, exception text, or client metadata. The directory
+defaults to `/tmp/airlock-oom-diagnostics`; output files are owner-only. Secure
+or pre-create a configured directory with the permissions your deployment
+needs. Disable the recorder after the investigation. The repository's
+engineering OOM instrumentation runbook contains the signal and high-water
+procedure.
 
 ## Startup Modes
 

tests/test_oom_diagnostics.py

@@ -83,6 +83,44 @@ def test_diagnostics_are_bounded_and_trim_skips_inflight(monkeypatch, tmp_path):
     assert any(record["phase"] == "signal_usr2_skipped_in_flight" for record in records)
 
 
+def test_inflight_trim_signal_uses_a_minimal_skip_record(monkeypatch, tmp_path):
+    """An in-flight trim must not walk the heap before releasing its signal lock."""
+    monkeypatch.setenv("AIRLOCK_OOM_DIAGNOSTICS", "true")
+    monkeypatch.setenv("AIRLOCK_OOM_DIAGNOSTICS_DIR", str(tmp_path))
+
+    diagnostics = OOMDiagnostics()
+    diagnostics.pre_call({"metadata": {}})
+    snapshot_calls: list[tuple[tuple[object, ...], dict[str, object]]] = []
+    monkeypatch.setattr(
+        diagnostics,
+        "snapshot",
+        lambda *args, **kwargs: snapshot_calls.append((args, kwargs)),
+    )
+    diagnostics._signal_snapshot("signal_usr2", trim=True)
+    deadline = time.monotonic() + 2
+    while diagnostics._signal_lock.locked() and time.monotonic() < deadline:
+        time.sleep(0.01)
+
+    assert not diagnostics._signal_lock.locked()
+    assert snapshot_calls == []
+    records = [
+        json.loads(line)
+        for line in next(tmp_path.glob("*.jsonl")).read_text().splitlines()
+    ]
+    skipped = next(
+        record
+        for record in records
+        if record["phase"] == "signal_usr2_skipped_in_flight"
+    )
+    assert skipped == {
+        "ts_monotonic_ns": skipped["ts_monotonic_ns"],
+        "phase": "signal_usr2_skipped_in_flight",
+        "sequence": None,
+        "in_flight": 1,
+        "trim": {"attempted": False, "reason": "requests_in_flight"},
+    }
+
+
 def test_tracemalloc_can_be_disabled_for_a_low_perturbation_replay(
     monkeypatch, tmp_path
 ):