From 657eb6a2acc69ad9f6341358b1f05d1c2a345217 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:07:05 +0200
Subject: [PATCH 01/11] docs: design bounded async observability flush

---
 ...-10-12-async-observability-flush-design.md | 87 +++++++++++++++++++
 1 file changed, 87 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-10-12-async-observability-flush-design.md

diff --git a/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md b/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
new file mode 100644
index 00000000..71e58e04
--- /dev/null
+++ b/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
@@ -0,0 +1,87 @@
+# Asynchronous observability flush with bounded exit grace
+
+## Context
+
+`ObservabilityService.flush()` currently calls `MeterProvider.force_flush()` on
+the caller's thread. A missing or slow OTLP collector can therefore delay a
+modular input at every `StanzaObservabilityRecorder.flush()` or context-manager
+exit. OpenTelemetry's `MeterProvider` also registers a synchronous `atexit`
+shutdown by default, so moving only `force_flush()` to another thread would
+still leave process exit potentially blocked.
+
+## Goals and contract
+
+- Keep `ObservabilityService.flush(timeout_millis=30_000)` and
+  `StanzaObservabilityRecorder.flush()` callable without changing their return
+  types. `flush()` schedules export work and returns promptly; it no longer
+  confirms delivery before returning. The timeout remains the budget passed
+  to the background `force_flush()` call, not a wait on the caller's thread.
+- Preserve OTel's periodic export while the process is alive.
+- On normal interpreter exit, give pending work and a final OTel shutdown a
+  **single process-wide grace period of at most 5 seconds** in solnlib's exit
+  handler. If work is unfinished, allow the process to exit and lose the last
+  metrics rather than hold the caller indefinitely.
+- Keep telemetry failures auxiliary: log them at INFO and do not raise into
+  modular-input code.
+
+This grace period bounds only solnlib's own exit handler. Other application
+`atexit` handlers, non-daemon threads, and separately created OTel providers
+remain outside its control. Abrupt termination (`SIGKILL`, `os._exit()`, etc.)
+does not invoke `atexit` and cannot guarantee a final export.
+
+## Design
+
+Each successfully initialized `ObservabilityService` owns one daemon worker,
+started during service initialization. The worker waits for requests and is
+the only solnlib thread that calls that service's provider `force_flush()` or
+`shutdown()`. A condition-protected pending flag coalesces repeated flush
+requests, so callers do not create an unbounded queue or new thread per call.
+If a request arrives during an export, the worker performs one further flush
+after the current one. A later request may replace the pending timeout budget.
+
+Construct solnlib's `MeterProvider` with `shutdown_on_exit=False`; otherwise
+OTel's synchronous exit hook would bypass the grace period. Register one
+process-wide solnlib `atexit` handler. It signals shutdown to all service
+workers, then joins them against one monotonic 5-second deadline rather than
+waiting 5 seconds for each service. A worker that receives shutdown skips
+redundant queued flushes and calls `provider.shutdown()` once, which triggers
+the periodic reader's final collection. It passes the remaining grace period
+as the provider's shutdown timeout. If it is already inside a flush, it
+finishes that operation before shutdown; if that exceeds the grace period,
+the daemon worker can be abandoned at interpreter exit. The handler never
+starts a new thread, which is required on Python 3.12+.
+
+Service initialization that fails before a provider exists does not start a
+worker. The existing singleton cache continues to share one service and
+worker per modular-input type. Directly constructed services are also tracked
+for exit cleanup. Shutdown requests are idempotent; flush requests made after
+shutdown begins are ignored. Worker exceptions are caught and logged without
+terminating the modular input.
+
+Register an `after_in_child` fork callback. In the child it replaces inherited
+manager and service synchronization objects, discards parent worker references,
+and starts one fresh daemon worker for each inherited live service before any
+child `atexit` handler can run. The parent keeps its workers unchanged. The
+callback does not call a provider's exporter or perform network I/O.
+
+## Compatibility and trade-offs
+
+This is an intentional semantic change: a returned `flush()` no longer means
+that metrics reached an exporter. Existing callers requiring confirmed
+delivery need a separate synchronous API; this change does not silently
+promise that guarantee. Normal periodic export continues independently.
+The added daemon worker costs one thread per active service. A slow export can
+coalesce multiple explicit flush requests, and the final 5-second deadline
+can sacrifice data when the collector is unavailable. The exporter circuit
+breaker remains unchanged and does not itself make flush non-blocking.
+
+## Verification
+
+Add deterministic unit tests for prompt return under a blocked exporter,
+coalescing, follow-up requests, exception logging, provider absence, shutdown
+idempotency, and the single deadline across multiple services. Add subprocess
+tests for normal exit with successful final delivery and a blocked exporter
+with a shortened test grace period. Exercise the supported Python versions,
+including Python 3.13's prohibition on starting threads in `atexit`, and a
+fork smoke test where available. Run the observability unit suite and the
+repository's broader relevant tests.

From 20b45c153ac982c8b610888c3de073dab8f84458 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:37:31 +0200
Subject: [PATCH 02/11] docs: plan bounded async observability flush

---
 .../2026-10-12-async-observability-flush.md   | 476 ++++++++++++++++++
 1 file changed, 476 insertions(+)
 create mode 100644 docs/superpowers/plans/2026-10-12-async-observability-flush.md

diff --git a/docs/superpowers/plans/2026-10-12-async-observability-flush.md b/docs/superpowers/plans/2026-10-12-async-observability-flush.md
new file mode 100644
index 00000000..34c554a7
--- /dev/null
+++ b/docs/superpowers/plans/2026-10-12-async-observability-flush.md
@@ -0,0 +1,476 @@
+# Asynchronous Observability Flush Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Make solnlib's public metric `flush()` non-blocking while allowing one process-wide 5-second grace period for final metric export.
+
+**Architecture:** A private daemon worker per initialized observability service serializes explicit flush and final provider shutdown. A process-wide registry registers one `atexit` hook, signals all workers, and joins them against a single monotonic deadline; solnlib-created OTel providers disable their own synchronous exit hook.
+
+**Tech Stack:** Python 3.9–3.13, OpenTelemetry Python SDK >=1.39.1,<2, pytest, `threading`, `atexit`.
+
+**Spec:** `docs/superpowers/specs/2026-10-12-async-observability-flush-design.md`
+
+## Global Constraints
+
+- Preserve `ObservabilityService.flush(timeout_millis: float = 30_000) -> None` and `StanzaObservabilityRecorder.flush() -> None` signatures while documenting the changed asynchronous semantics.
+- Keep periodic OTel export; use `MeterProvider(shutdown_on_exit=False)` only for providers constructed by solnlib.
+- Give all solnlib workers one total 5-second grace period at normal exit, not 5 seconds each; no thread creation inside `atexit`.
+- Keep telemetry failures at INFO and never propagate them into modular-input code.
+- Treat pending final metrics as lossy after the grace period; unrelated `atexit` handlers and non-daemon threads are outside this guarantee.
+- Keep existing untracked files untouched. Edit only the files listed below and stage/commit only intended files.
+
+## File structure
+
+- Create `solnlib/_observability_flush.py`: private worker, registry, process-exit and fork lifecycle; independent of `observability.py` to avoid growing its already-large public module.
+- Create `tests/unit/test_observability_flush.py`: deterministic worker/registry/concurrency tests and subprocess exit/fork smoke tests.
+- Modify `solnlib/observability.py`: wire the private manager into `ObservabilityService`, disable OTel's own exit handler, and update public docstrings.
+- Modify `tests/unit/test_observability.py`: replace synchronous-flush assertions with asynchronous assertions and cover recorder compatibility.
+
+---
+
+### Task 1: Daemon worker and flush coalescing
+
+**Files:**
+- Create: `solnlib/_observability_flush.py`
+- Create: `tests/unit/test_observability_flush.py`
+
+**Interfaces:**
+- Produces: `_AsyncFlushController(provider, logger, format_error)`, `request_flush(timeout_millis: float) -> None`, `request_shutdown(deadline: float) -> None`, `join_until(deadline: float) -> None`, `restart_after_fork() -> None`.
+- Consumes: an OTel-like provider exposing `force_flush(timeout_millis=...)` and `shutdown(timeout_millis=...)`, a logger with `info`, and a safe exception formatter.
+
+- [ ] **Step 1: Write failing tests** in `tests/unit/test_observability_flush.py` for prompt return and coalescing. Use `threading.Event` gates so no network or sleeps are needed:
+
+```python
+import logging
+import threading
+import time
+from unittest.mock import MagicMock
+
+from solnlib._observability_flush import _AsyncFlushController
+
+
+def test_request_flush_returns_while_export_is_blocked():
+    entered = threading.Event()
+    release = threading.Event()
+    provider = MagicMock()
+
+    def blocked_flush(**kwargs):
+        entered.set()
+        assert release.wait(2)
+
+    provider.force_flush.side_effect = blocked_flush
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    try:
+        start = time.monotonic()
+        controller.request_flush(30_000)
+        assert time.monotonic() - start < 0.2
+        assert entered.wait(1)
+        provider.force_flush.assert_called_once_with(timeout_millis=30_000)
+    finally:
+        release.set()
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+
+
+def test_repeated_requests_during_export_coalesce_to_one_followup():
+    entered = threading.Event()
+    release = threading.Event()
+    second_done = threading.Event()
+    provider = MagicMock()
+
+    def flush(**kwargs):
+        if provider.force_flush.call_count == 1:
+            entered.set()
+            assert release.wait(2)
+        else:
+            second_done.set()
+
+    provider.force_flush.side_effect = flush
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    try:
+        controller.request_flush(30_000)
+        assert entered.wait(1)
+        controller.request_flush(10_000)
+        controller.request_flush(5_000)
+        release.set()
+        assert second_done.wait(1)
+        assert provider.force_flush.call_count == 2
+        assert provider.force_flush.call_args_list[-1].kwargs == {"timeout_millis": 5_000}
+    finally:
+        release.set()
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+```
+
+- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect import failure because the private module does not exist.
+- [ ] **Step 3: Implement the worker** in `solnlib/_observability_flush.py` with one pending timeout slot; never hold its condition across an OTel call:
+
+```python
+import threading
+import time
+from typing import Callable
+
+
+class _AsyncFlushController:
+    def __init__(self, provider, logger, format_error: Callable[[BaseException], str]):
+        self._provider = provider
+        self._logger = logger
+        self._format_error = format_error
+        self._condition = threading.Condition()
+        self._pending_timeout = None
+        self._shutdown_deadline = None
+        self._worker = threading.Thread(target=self._run, name="SolnlibMetricFlush", daemon=True)
+        self._worker.start()
+
+    def request_flush(self, timeout_millis: float) -> None:
+        with self._condition:
+            if self._shutdown_deadline is None:
+                self._pending_timeout = int(timeout_millis)
+                self._condition.notify()
+
+    def request_shutdown(self, deadline: float) -> None:
+        with self._condition:
+            if self._shutdown_deadline is None:
+                self._shutdown_deadline = deadline
+                self._pending_timeout = None
+                self._condition.notify()
+
+    def join_until(self, deadline: float) -> None:
+        self._worker.join(timeout=max(0.0, deadline - time.monotonic()))
+
+    def _run(self) -> None:
+        while True:
+            with self._condition:
+                while self._pending_timeout is None and self._shutdown_deadline is None:
+                    self._condition.wait()
+                deadline = self._shutdown_deadline
+                timeout = self._pending_timeout
+                self._pending_timeout = None
+            if deadline is not None:
+                try:
+                    remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
+                    self._provider.shutdown(timeout_millis=remaining_ms)
+                except Exception as error:
+                    self._logger.info("Failed to shut down metrics: %s", self._format_error(error))
+                return
+            try:
+                self._provider.force_flush(timeout_millis=timeout)
+            except Exception as error:
+                self._logger.info("Failed to flush metrics: %s", self._format_error(error))
+
+    def restart_after_fork(self) -> None:
+        self._condition = threading.Condition()
+        self._pending_timeout = None
+        self._shutdown_deadline = None
+        self._worker = threading.Thread(target=self._run, name="SolnlibMetricFlush", daemon=True)
+        self._worker.start()
+```
+
+- [ ] **Step 4: Add error/idempotency tests** below the first tests, then run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and expect PASS:
+
+```python
+def test_worker_logs_export_error_without_raising_to_caller():
+    logged = threading.Event()
+    provider = MagicMock()
+    provider.force_flush.side_effect = RuntimeError("boom")
+    logger = MagicMock(spec=logging.Logger)
+    logger.info.side_effect = lambda *args: logged.set()
+    controller = _AsyncFlushController(provider, logger, str)
+    try:
+        controller.request_flush(30_000)
+        assert logged.wait(1)
+        assert "boom" in logger.info.call_args.args[-1]
+    finally:
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+
+
+def test_shutdown_is_once_and_later_flush_is_ignored():
+    provider = MagicMock()
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    deadline = time.monotonic() + 1
+    controller.request_shutdown(deadline)
+    controller.request_shutdown(deadline)
+    controller.join_until(deadline)
+    controller.request_flush(30_000)
+    provider.shutdown.assert_called_once()
+    provider.force_flush.assert_not_called()
+```
+- [ ] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: add async metric flush worker"`.
+
+### Task 2: One process-wide exit deadline and fork reset
+
+**Files:**
+- Modify: `solnlib/_observability_flush.py`
+- Modify: `tests/unit/test_observability_flush.py`
+
+**Interfaces:**
+- Consumes: `_AsyncFlushController` from Task 1.
+- Produces: module-level `_register_controller(provider, logger, format_error) -> _AsyncFlushController` backed by `_FlushRegistry`; `_shutdown_all()` is the `atexit` entry point.
+
+- [ ] **Step 1: Write failing registry tests** in the same test file. Use a local registry instance so tests never shut down production's module-level registry:
+
+```python
+from solnlib import _observability_flush as flushmod
+
+
+def test_exit_budget_is_shared_by_two_controllers(monkeypatch):
+    monkeypatch.setattr(flushmod, "_EXIT_GRACE_SECONDS", 0.05)
+    release = threading.Event()
+    entered = [threading.Event(), threading.Event()]
+    controllers = []
+    registry = flushmod._FlushRegistry()
+    for gate in entered:
+        provider = MagicMock()
+        provider.shutdown.side_effect = lambda gate=gate, **kwargs: (gate.set(), release.wait(2))
+        controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+        registry._controllers.append(controller)
+        controllers.append(controller)
+    try:
+        start = time.monotonic()
+        registry.shutdown_all()
+        assert time.monotonic() - start < 0.2
+        assert all(gate.wait(1) for gate in entered)
+    finally:
+        release.set()
+        for controller in controllers:
+            controller.join_until(time.monotonic() + 1)
+
+
+def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
+    register = MagicMock()
+    monkeypatch.setattr(flushmod.atexit, "register", register)
+    if hasattr(flushmod.os, "register_at_fork"):
+        monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock())
+    registry = flushmod._FlushRegistry()
+    registry.add(MagicMock())
+    registry.add(MagicMock())
+    register.assert_called_once_with(flushmod._shutdown_all)
+```
+- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and confirm `_register_controller`/`_shutdown_all` are missing.
+- [ ] **Step 3: Implement the registry** in the private module. Use a global registry object and these exact operations; guard `os.register_at_fork` for platforms without fork:
+
+```python
+import atexit
+import os
+
+_EXIT_GRACE_SECONDS = 5.0
+
+
+class _FlushRegistry:
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._controllers = []
+        self._registered = False
+
+    def add(self, controller):
+        with self._lock:
+            if not self._registered:
+                atexit.register(_shutdown_all)
+                if hasattr(os, "register_at_fork"):
+                    os.register_at_fork(after_in_child=self._after_fork_child)
+                self._registered = True
+            self._controllers.append(controller)
+
+    def shutdown_all(self):
+        deadline = time.monotonic() + _EXIT_GRACE_SECONDS
+        with self._lock:
+            controllers = tuple(self._controllers)
+        for controller in controllers:
+            controller.request_shutdown(deadline)
+        for controller in controllers:
+            controller.join_until(deadline)
+
+    def _after_fork_child(self):
+        self._lock = threading.Lock()
+        for controller in self._controllers:
+            controller.restart_after_fork()
+
+
+_registry = _FlushRegistry()
+
+
+def _register_controller(provider, logger, format_error):
+    controller = _AsyncFlushController(provider, logger, format_error)
+    _registry.add(controller)
+    return controller
+
+
+def _shutdown_all():
+    _registry.shutdown_all()
+```
+
+- [ ] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`; verify the shared deadline and single registration assertions.
+- [ ] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: bound metric shutdown across services"`.
+
+### Task 3: Wire the public solnlib API and document the contract change
+
+**Files:**
+- Modify: `solnlib/observability.py:45-65,410-465,713-733,1015-1046`
+- Modify: `tests/unit/test_observability.py:603-615,947-974,1557-1570,1730-1760`
+
+**Interfaces:**
+- Consumes: `_register_controller` and `_AsyncFlushController.request_flush` from Tasks 1–2.
+- Produces: unchanged public method signatures, but asynchronous `flush()` semantics and `MeterProvider(..., shutdown_on_exit=False)`.
+
+- [ ] **Step 1: Replace the old synchronous test** with this integration test in `tests/unit/test_observability.py`. Task 1 separately proves that the real controller runs exports off-thread:
+
+```python
+def test_flush_schedules_without_calling_provider_on_caller(logger, monkeypatch):
+    controller = MagicMock()
+    provider_cls = MagicMock()
+    monkeypatch.setattr("solnlib.observability.MeterProvider", provider_cls)
+    monkeypatch.setattr(
+        "solnlib.observability.ObservabilityService._create_otlp_exporter",
+        lambda self: None,
+    )
+    monkeypatch.setattr(
+        "solnlib.observability._register_controller",
+        lambda provider, logger, formatter: controller,
+    )
+    svc = ObservabilityService("test-input", logger, ta_name="my_ta", ta_version="1.0.0")
+    svc.flush(timeout_millis=5000)
+    controller.request_flush.assert_called_once_with(5000)
+    provider_cls.return_value.force_flush.assert_not_called()
+    assert provider_cls.call_args.kwargs["shutdown_on_exit"] is False
+```
+
+  Keep the provider-missing no-op test. Replace the old exception tests with these scheduling-failure assertions; worker-side errors are covered in Task 1:
+
+```python
+def test_flush_logs_info_when_scheduling_fails(logger, monkeypatch):
+    svc = _make_service(logger, monkeypatch)
+    svc._flush_controller = MagicMock()
+    svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
+    svc.flush()
+    assert any("boom" in call.args for call in logger.info.call_args_list)
+    logger.warning.assert_not_called()
+
+
+def test_flush_scheduling_error_uses_safe_str(logger, monkeypatch):
+    monkeypatch.setattr("solnlib.observability._safe_exception_str", lambda error: "SAFE")
+    svc = _make_service(logger, monkeypatch)
+    svc._flush_controller = MagicMock()
+    svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
+    logger.reset_mock()
+    svc.flush()
+    assert any("SAFE" in call.args for call in logger.info.call_args_list)
+```
+- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py -q`; expect the new thread/constructor assertions to fail.
+- [ ] **Step 3: Wire the constructor and method:**
+
+```python
+from ._observability_flush import _register_controller
+
+# At the beginning of ObservabilityService.__init__:
+self._flush_controller = None
+
+# Replace the provider constructor:
+self._provider = MeterProvider(
+    resource=resource,
+    metric_readers=metric_readers,
+    shutdown_on_exit=False,
+)
+
+# After counters are created and before the success log:
+self._flush_controller = _register_controller(
+    self._provider, self._logger, _safe_exception_str
+)
+
+# Replace the body of ObservabilityService.flush after the provider guard:
+if self._flush_controller is not None:
+    self._flush_controller.request_flush(timeout_millis)
+```
+
+  Keep the existing `try/except` boundary around `flush()` so scheduling failures stay at INFO. Update `ObservabilityService.flush`, `StanzaObservabilityRecorder.flush`, `__exit__`, and module/class docstrings: returned flush means scheduled attempt, not completed attempt; `timeout_millis` is a worker budget; normal exit may wait up to 5 seconds total.
+- [ ] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q`. Ensure existing recorder `with` tests still call `flush()` once and propagate user exceptions.
+- [ ] **Step 5: Commit only intended files** with `git add -- solnlib/observability.py tests/unit/test_observability.py` and `git commit -m "feat: make solnlib observability flush nonblocking"`.
+
+### Task 4: Process-exit regression tests and cross-version verification
+
+**Files:**
+- Modify: `tests/unit/test_observability_flush.py`
+- Modify: `solnlib/_observability_flush.py` only if the tests expose an issue.
+
+**Interfaces:**
+- Consumes: final registry and public API from Tasks 1–3.
+- Produces: reproducible evidence that normal exit attempts a final export and an unavailable exporter cannot hold solnlib's exit hook beyond its grace period.
+
+- [ ] **Step 1: Add subprocess tests** to `tests/unit/test_observability_flush.py`, with these imports and tests. They use a fake provider and never contact a collector:
+
+```python
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _run_child(script):
+    repo_root = Path(__file__).resolve().parents[2]
+    env = dict(os.environ, PYTHONPATH=str(repo_root))
+    return subprocess.run(
+        [sys.executable, "-c", script], capture_output=True, text=True,
+        env=env, timeout=3,
+    )
+
+
+def test_normal_exit_attempts_final_shutdown_without_explicit_flush():
+    result = _run_child("""
+import logging
+from solnlib._observability_flush import _register_controller
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs): print('FINAL_SHUTDOWN', flush=True)
+_register_controller(Provider(), logging.getLogger('test'), str)
+""")
+    assert result.returncode == 0, result.stderr
+    assert "FINAL_SHUTDOWN" in result.stdout
+
+
+def test_blocked_final_export_uses_shortened_grace_period():
+    start = time.monotonic()
+    result = _run_child("""
+import logging
+import threading
+import solnlib._observability_flush as flushmod
+flushmod._EXIT_GRACE_SECONDS = 0.05
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs): threading.Event().wait(10)
+flushmod._register_controller(Provider(), logging.getLogger('test'), str)
+""")
+    assert result.returncode == 0, result.stderr
+    assert time.monotonic() - start < 1.5
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
+def test_child_gets_fresh_worker_after_fork():
+    result = _run_child("""
+import logging
+import os
+import sys
+from solnlib._observability_flush import _register_controller
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs): pass
+controller = _register_controller(Provider(), logging.getLogger('test'), str)
+parent_worker = controller._worker
+pid = os.fork()
+if pid == 0:
+    fresh = controller._worker is not parent_worker and controller._worker.is_alive()
+    print('CHILD_FRESH_WORKER=' + str(fresh), flush=True)
+    sys.exit(0 if fresh else 2)
+_, status = os.waitpid(pid, 0)
+print('PARENT_WORKER_ALIVE=' + str(parent_worker.is_alive()), flush=True)
+sys.exit(os.waitstatus_to_exitcode(status))
+""")
+    assert result.returncode == 0, result.stderr
+    assert "CHILD_FRESH_WORKER=True" in result.stdout
+    assert "PARENT_WORKER_ALIVE=True" in result.stdout
+```
+
+- [ ] **Step 2: Run exit/fork tests.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect all to pass; if the fork test fails, inspect whether the child's callback ran and repair `restart_after_fork()` before proceeding.
+- [ ] **Step 3: Verify both supported runtimes.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q` and `. .venv/bin/activate && pytest tests/unit -q`. Run the focused suite under Python 3.13 if dependencies are installed; otherwise run `python3.13 -m py_compile solnlib/observability.py solnlib/_observability_flush.py` and report that runtime tests on 3.13 remain CI-only. Check `git diff --check` and `git status --short`.
+- [ ] **Step 4: Commit only intended files** with `git add -- tests/unit/test_observability_flush.py solnlib/_observability_flush.py` and `git commit -m "test: cover bounded metric export at process exit"`.

From 53269141d13580754503d7dc9ecd6143f8641f18 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:41:56 +0200
Subject: [PATCH 03/11] feat: add async metric flush worker

---
 solnlib/_observability_flush.py        |  88 +++++++++++++++++++++
 tests/unit/test_observability_flush.py | 105 +++++++++++++++++++++++++
 2 files changed, 193 insertions(+)
 create mode 100644 solnlib/_observability_flush.py
 create mode 100644 tests/unit/test_observability_flush.py

diff --git a/solnlib/_observability_flush.py b/solnlib/_observability_flush.py
new file mode 100644
index 00000000..709d603a
--- /dev/null
+++ b/solnlib/_observability_flush.py
@@ -0,0 +1,88 @@
+#
+# Copyright 2026 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Private background worker for OpenTelemetry metric flush and shutdown."""
+
+import threading
+import time
+from typing import Callable
+
+
+class _AsyncFlushController:
+    """Serialize one provider's explicit flushes and final shutdown."""
+
+    def __init__(
+        self, provider, logger, format_error: Callable[[BaseException], str]
+    ) -> None:
+        self._provider = provider
+        self._logger = logger
+        self._format_error = format_error
+        self._condition = threading.Condition()
+        self._pending_timeout = None
+        self._shutdown_deadline = None
+        self._worker = threading.Thread(
+            target=self._run, name="SolnlibMetricFlush", daemon=True
+        )
+        self._worker.start()
+
+    def request_flush(self, timeout_millis: float) -> None:
+        with self._condition:
+            if self._shutdown_deadline is None:
+                self._pending_timeout = int(timeout_millis)
+                self._condition.notify()
+
+    def request_shutdown(self, deadline: float) -> None:
+        with self._condition:
+            if self._shutdown_deadline is None:
+                self._shutdown_deadline = deadline
+                self._pending_timeout = None
+                self._condition.notify()
+
+    def join_until(self, deadline: float) -> None:
+        self._worker.join(timeout=max(0.0, deadline - time.monotonic()))
+
+    def _run(self) -> None:
+        while True:
+            with self._condition:
+                while self._pending_timeout is None and self._shutdown_deadline is None:
+                    self._condition.wait()
+                deadline = self._shutdown_deadline
+                timeout = self._pending_timeout
+                self._pending_timeout = None
+            if deadline is not None:
+                try:
+                    remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
+                    self._provider.shutdown(timeout_millis=remaining_ms)
+                except Exception as error:
+                    self._logger.info(
+                        "Failed to shut down metrics: %s", self._format_error(error)
+                    )
+                return
+            try:
+                self._provider.force_flush(timeout_millis=timeout)
+            except Exception as error:
+                self._logger.info(
+                    "Failed to flush metrics: %s", self._format_error(error)
+                )
+
+    def restart_after_fork(self) -> None:
+        """Replace parent-thread state after fork in the child process."""
+        self._condition = threading.Condition()
+        self._pending_timeout = None
+        self._shutdown_deadline = None
+        self._worker = threading.Thread(
+            target=self._run, name="SolnlibMetricFlush", daemon=True
+        )
+        self._worker.start()
diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py
new file mode 100644
index 00000000..553474e2
--- /dev/null
+++ b/tests/unit/test_observability_flush.py
@@ -0,0 +1,105 @@
+#
+# Copyright 2026 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+import logging
+import threading
+import time
+from unittest.mock import MagicMock
+
+from solnlib._observability_flush import _AsyncFlushController
+
+
+def test_request_flush_returns_while_export_is_blocked():
+    entered = threading.Event()
+    release = threading.Event()
+    provider = MagicMock()
+
+    def blocked_flush(**kwargs):
+        entered.set()
+        assert release.wait(2)
+
+    provider.force_flush.side_effect = blocked_flush
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    try:
+        start = time.monotonic()
+        controller.request_flush(30_000)
+        assert time.monotonic() - start < 0.2
+        assert entered.wait(1)
+        provider.force_flush.assert_called_once_with(timeout_millis=30_000)
+    finally:
+        release.set()
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+
+
+def test_repeated_requests_during_export_coalesce_to_one_followup():
+    entered = threading.Event()
+    release = threading.Event()
+    second_done = threading.Event()
+    provider = MagicMock()
+
+    def flush(**kwargs):
+        if provider.force_flush.call_count == 1:
+            entered.set()
+            assert release.wait(2)
+        else:
+            second_done.set()
+
+    provider.force_flush.side_effect = flush
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    try:
+        controller.request_flush(30_000)
+        assert entered.wait(1)
+        controller.request_flush(10_000)
+        controller.request_flush(5_000)
+        release.set()
+        assert second_done.wait(1)
+        assert provider.force_flush.call_count == 2
+        assert provider.force_flush.call_args_list[-1].kwargs == {
+            "timeout_millis": 5_000
+        }
+    finally:
+        release.set()
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+
+
+def test_worker_logs_export_error_without_raising_to_caller():
+    logged = threading.Event()
+    provider = MagicMock()
+    provider.force_flush.side_effect = RuntimeError("boom")
+    logger = MagicMock(spec=logging.Logger)
+    logger.info.side_effect = lambda *args: logged.set()
+    controller = _AsyncFlushController(provider, logger, str)
+    try:
+        controller.request_flush(30_000)
+        assert logged.wait(1)
+        assert "boom" in logger.info.call_args.args[-1]
+    finally:
+        controller.request_shutdown(time.monotonic() + 1)
+        controller.join_until(time.monotonic() + 1)
+
+
+def test_shutdown_is_once_and_later_flush_is_ignored():
+    provider = MagicMock()
+    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    deadline = time.monotonic() + 1
+    controller.request_shutdown(deadline)
+    controller.request_shutdown(deadline)
+    controller.join_until(deadline)
+    controller.request_flush(30_000)
+    provider.shutdown.assert_called_once()
+    provider.force_flush.assert_not_called()

From 280fba96fe1f0f14df256b4840fc52a16283c581 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:43:49 +0200
Subject: [PATCH 04/11] feat: bound metric shutdown across services

---
 solnlib/_observability_flush.py        | 49 ++++++++++++++++++++++++++
 tests/unit/test_observability_flush.py | 40 +++++++++++++++++++++
 2 files changed, 89 insertions(+)

diff --git a/solnlib/_observability_flush.py b/solnlib/_observability_flush.py
index 709d603a..d0c8c44b 100644
--- a/solnlib/_observability_flush.py
+++ b/solnlib/_observability_flush.py
@@ -15,10 +15,14 @@
 #
 """Private background worker for OpenTelemetry metric flush and shutdown."""
 
+import atexit
+import os
 import threading
 import time
 from typing import Callable
 
+_EXIT_GRACE_SECONDS = 5.0
+
 
 class _AsyncFlushController:
     """Serialize one provider's explicit flushes and final shutdown."""
@@ -86,3 +90,48 @@ def restart_after_fork(self) -> None:
             target=self._run, name="SolnlibMetricFlush", daemon=True
         )
         self._worker.start()
+
+
+class _FlushRegistry:
+    """Coordinate all metric providers against one process-exit deadline."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._controllers = []
+        self._registered = False
+
+    def add(self, controller: _AsyncFlushController) -> None:
+        with self._lock:
+            if not self._registered:
+                atexit.register(_shutdown_all)
+                if hasattr(os, "register_at_fork"):
+                    os.register_at_fork(after_in_child=self._after_fork_child)
+                self._registered = True
+            self._controllers.append(controller)
+
+    def shutdown_all(self) -> None:
+        deadline = time.monotonic() + _EXIT_GRACE_SECONDS
+        with self._lock:
+            controllers = tuple(self._controllers)
+        for controller in controllers:
+            controller.request_shutdown(deadline)
+        for controller in controllers:
+            controller.join_until(deadline)
+
+    def _after_fork_child(self) -> None:
+        self._lock = threading.Lock()
+        for controller in self._controllers:
+            controller.restart_after_fork()
+
+
+_registry = _FlushRegistry()
+
+
+def _register_controller(provider, logger, format_error) -> _AsyncFlushController:
+    controller = _AsyncFlushController(provider, logger, format_error)
+    _registry.add(controller)
+    return controller
+
+
+def _shutdown_all() -> None:
+    _registry.shutdown_all()
diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py
index 553474e2..4b05a6b9 100644
--- a/tests/unit/test_observability_flush.py
+++ b/tests/unit/test_observability_flush.py
@@ -19,6 +19,7 @@
 import time
 from unittest.mock import MagicMock
 
+from solnlib import _observability_flush as flushmod
 from solnlib._observability_flush import _AsyncFlushController
 
 
@@ -103,3 +104,42 @@ def test_shutdown_is_once_and_later_flush_is_ignored():
     controller.request_flush(30_000)
     provider.shutdown.assert_called_once()
     provider.force_flush.assert_not_called()
+
+
+def test_exit_budget_is_shared_by_two_controllers(monkeypatch):
+    monkeypatch.setattr(flushmod, "_EXIT_GRACE_SECONDS", 0.05)
+    release = threading.Event()
+    entered = [threading.Event(), threading.Event()]
+    controllers = []
+    registry = flushmod._FlushRegistry()
+    for gate in entered:
+        provider = MagicMock()
+        provider.shutdown.side_effect = lambda gate=gate, **kwargs: (
+            gate.set(),
+            release.wait(2),
+        )
+        controller = _AsyncFlushController(
+            provider, MagicMock(spec=logging.Logger), str
+        )
+        registry._controllers.append(controller)
+        controllers.append(controller)
+    try:
+        start = time.monotonic()
+        registry.shutdown_all()
+        assert time.monotonic() - start < 0.2
+        assert all(gate.wait(1) for gate in entered)
+    finally:
+        release.set()
+        for controller in controllers:
+            controller.join_until(time.monotonic() + 1)
+
+
+def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
+    register = MagicMock()
+    monkeypatch.setattr(flushmod.atexit, "register", register)
+    if hasattr(flushmod.os, "register_at_fork"):
+        monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock())
+    registry = flushmod._FlushRegistry()
+    registry.add(MagicMock())
+    registry.add(MagicMock())
+    register.assert_called_once_with(flushmod._shutdown_all)

From c6897eb9b5e6274f38232e7e696850842fe56388 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:46:02 +0200
Subject: [PATCH 05/11] feat: make solnlib observability flush nonblocking

---
 solnlib/observability.py         | 56 ++++++++++++++++++++------------
 tests/unit/test_observability.py | 53 ++++++++++++++++++------------
 2 files changed, 68 insertions(+), 41 deletions(-)

diff --git a/solnlib/observability.py b/solnlib/observability.py
index 5c3eae7f..d809948c 100644
--- a/solnlib/observability.py
+++ b/solnlib/observability.py
@@ -31,8 +31,8 @@
 - :class:`StanzaObservabilityRecorder` — a stanza-scoped recorder that wraps
   ``ObservabilityService`` with a per-process singleton cache.  Bind it to a
   single stanza name and call :meth:`~StanzaObservabilityRecorder.record` after
-  each batch of ingested events.  Use as a context manager for automatic flush
-  on exit.
+  each batch of ingested events.  Use as a context manager to schedule a
+  background flush on exit.
 
 Typical usage::
 
@@ -52,6 +52,7 @@
 import threading
 import urllib.request
 from typing import Callable, ClassVar, Optional, Union
+from ._observability_flush import _register_controller
 from .splunkenv import get_conf_stanzas
 from opentelemetry.metrics import Instrument, Meter
 from opentelemetry.sdk.metrics import MeterProvider, Counter, Histogram
@@ -410,6 +411,7 @@ def __init__(
         self.event_bytes_counter: Optional[Counter] = None
         self._meter: Optional[Meter] = None
         self._provider: Optional[MeterProvider] = None
+        self._flush_controller = None
 
         for logger_name in _METRICS_SDK_LOGGERS:
             logging.getLogger(logger_name).addFilter(_downgrade_to_info_filter)
@@ -452,7 +454,9 @@ def __init__(
                 metric_readers.append(PeriodicExportingMetricReader(exporter))
 
             self._provider = MeterProvider(
-                resource=resource, metric_readers=metric_readers
+                resource=resource,
+                metric_readers=metric_readers,
+                shutdown_on_exit=False,
             )
             self._meter = self._provider.get_meter(ta_name, ta_version)
 
@@ -467,6 +471,10 @@ def __init__(
                 unit="By",
             )
 
+            self._flush_controller = _register_controller(
+                self._provider, self._logger, _safe_exception_str
+            )
+
             self._logger.info(
                 "ObservabilityService initialised: ta_name=%s ta_version=%s "
                 "modinput_type=%s",
@@ -711,26 +719,31 @@ def register_instrument(
         return callback(self._meter)
 
     def flush(self, timeout_millis: float = 30_000) -> None:
-        """Force-flush all metric readers.
+        """Schedule a background flush of all metric readers.
 
-        Blocks until all buffered data points have been handed off to their
-        exporters or *timeout_millis* elapses.  Call this before the modular
-        input process exits to avoid dropping the last batch of metrics.
+        Returns promptly without waiting for an export attempt to finish.
+        Repeated calls while an export is active are coalesced.  At normal
+        process exit, solnlib gives its metric workers a single five-second
+        grace period for final export; the last metrics can be lost if it
+        expires.
 
         Prefer using :class:`StanzaObservabilityRecorder` as a context manager
         rather than calling this method directly — it calls
         :meth:`StanzaObservabilityRecorder.flush` on exit automatically.
 
         Args:
-            timeout_millis: Maximum time to wait for exporters to drain, in
-                milliseconds.  Defaults to 30 seconds.
+            timeout_millis: Budget passed to the background force-flush call,
+                in milliseconds.  It is not a caller-side wait.  Defaults to
+                30 seconds.
         """
-        if self._provider is None:
+        if self._provider is None or self._flush_controller is None:
             return
         try:
-            self._provider.force_flush(timeout_millis=int(timeout_millis))
+            self._flush_controller.request_flush(timeout_millis)
         except Exception as error:
-            self._logger.info("Failed to flush metrics: %s", _safe_exception_str(error))
+            self._logger.info(
+                "Failed to schedule metrics flush: %s", _safe_exception_str(error)
+            )
 
 
 _INT64_MIN = -(2**63)
@@ -806,7 +819,8 @@ class StanzaObservabilityRecorder:
     **Best practices:**
 
     - Use as a context manager (``with`` statement) so that :meth:`flush` is
-      always called when the stanza collection loop exits, even on exceptions.
+      always scheduled when the stanza collection loop exits, even on
+      exceptions.  This does not wait for export completion.
     - Pass the same *modinput_type* string for all stanzas of the same input
       type.  The string should be lowercase, hyphenated, and stable across
       restarts (e.g. ``"event-hub"``).
@@ -1013,15 +1027,15 @@ def record(
             self._service.event_bytes_counter.add(byte_count, attributes=attrs)
 
     def flush(self) -> None:
-        """Force-flush all metric readers.
+        """Schedule a background flush of all metric readers.
 
-        Delegates to :meth:`ObservabilityService.flush` (which calls
-        ``MeterProvider.force_flush()`` internally).  Called automatically by
-        ``__exit__`` when the recorder is used as a context manager, so you
-        rarely need to call this directly.
+        Delegates to :meth:`ObservabilityService.flush`, which calls
+        ``MeterProvider.force_flush()`` on a daemon worker.  Called
+        automatically by ``__exit__`` when used as a context manager.  A
+        returned call means the attempt was scheduled, not completed.
 
         Call it explicitly only when you are not using the context manager and
-        need to guarantee delivery before the process exits::
+        want to request an export before the process exits::
 
             obs = StanzaObservabilityRecorder("my-input", logger, stanza_name)
             try:
@@ -1036,10 +1050,10 @@ def __enter__(self) -> "StanzaObservabilityRecorder":
         return self
 
     def __exit__(self, *_) -> bool:
-        """Flush all metric readers and allow exceptions to propagate.
+        """Schedule a flush and allow exceptions to propagate.
 
         Returns ``False`` so any exception raised inside the ``with`` block
-        is re-raised after flushing.
+        is re-raised after the flush has been scheduled.
         """
         self.flush()
         return False
diff --git a/tests/unit/test_observability.py b/tests/unit/test_observability.py
index 84c907f9..378a4f62 100644
--- a/tests/unit/test_observability.py
+++ b/tests/unit/test_observability.py
@@ -944,15 +944,29 @@ def test_create_otlp_exporter_uses_delta_temporality(
         assert temporality[Counter] == AggregationTemporality.DELTA
         assert temporality[Histogram] == AggregationTemporality.DELTA
 
-    def test_flush_calls_provider_force_flush(self, logger, monkeypatch):
-        # Arrange
-        svc = _make_service(logger, monkeypatch)
-        mock_provider = MagicMock()
-        svc._provider = mock_provider
-        # Act
+    def test_flush_schedules_without_calling_provider_on_caller(
+        self, logger, monkeypatch
+    ):
+        controller = MagicMock()
+        provider_cls = MagicMock()
+        monkeypatch.setattr("solnlib.observability.MeterProvider", provider_cls)
+        monkeypatch.setattr(
+            "solnlib.observability.ObservabilityService._create_otlp_exporter",
+            lambda self: None,
+        )
+        monkeypatch.setattr(
+            "solnlib.observability._register_controller",
+            lambda provider, logger, formatter: controller,
+        )
+        svc = ObservabilityService(
+            "test-input", logger, ta_name="my_ta", ta_version="1.0.0"
+        )
+
         svc.flush(timeout_millis=5_000)
-        # Assert
-        mock_provider.force_flush.assert_called_once_with(timeout_millis=5000)
+
+        controller.request_flush.assert_called_once_with(5_000)
+        provider_cls.return_value.force_flush.assert_not_called()
+        assert provider_cls.call_args.kwargs["shutdown_on_exit"] is False
 
     def test_flush_is_noop_when_provider_missing(self, logger, monkeypatch):
         # Arrange
@@ -961,16 +975,14 @@ def test_flush_is_noop_when_provider_missing(self, logger, monkeypatch):
         # Act / Assert — must not raise
         svc.flush()
 
-    def test_flush_logs_info_on_exception(self, logger, monkeypatch):
-        # Arrange
+    def test_flush_logs_info_when_scheduling_fails(self, logger, monkeypatch):
         svc = _make_service(logger, monkeypatch)
-        mock_provider = MagicMock()
-        mock_provider.force_flush.side_effect = RuntimeError("boom")
-        svc._provider = mock_provider
-        # Act
+        svc._flush_controller = MagicMock()
+        svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
+
         svc.flush()
-        # Assert
-        logger.info.assert_called()
+
+        assert any("boom" in call.args for call in logger.info.call_args_list)
         logger.warning.assert_not_called()
 
     def test_init_attaches_filter_to_metrics_sdk_loggers(self, logger, monkeypatch):
@@ -1554,16 +1566,17 @@ def test_create_otlp_exporter_exception_uses_safe_str_no_exc_info(
         assert "exc_info" not in kwargs
         logger.warning.assert_not_called()
 
-    def test_flush_exception_uses_safe_str(self, logger, monkeypatch):
+    def test_flush_scheduling_error_uses_safe_str(self, logger, monkeypatch):
         monkeypatch.setattr(
             "solnlib.observability._safe_exception_str", lambda error: "SAFE"
         )
         svc = _make_service(logger, monkeypatch)
-        mock_provider = MagicMock()
-        mock_provider.force_flush.side_effect = RuntimeError("boom")
-        svc._provider = mock_provider
+        svc._flush_controller = MagicMock()
+        svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
         logger.reset_mock()
+
         svc.flush()
+
         assert any("SAFE" in call.args for call in logger.info.call_args_list)
         logger.warning.assert_not_called()
 

From c099970156e7c815ece389df461e5e8ecec454b4 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:52:32 +0200
Subject: [PATCH 06/11] test: cover bounded metric export at process exit

---
 tests/unit/test_observability_flush.py | 117 +++++++++++++++++++++++++
 1 file changed, 117 insertions(+)

diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py
index 4b05a6b9..0d42ccbe 100644
--- a/tests/unit/test_observability_flush.py
+++ b/tests/unit/test_observability_flush.py
@@ -15,10 +15,16 @@
 #
 
 import logging
+import os
+import subprocess
+import sys
 import threading
 import time
+from pathlib import Path
 from unittest.mock import MagicMock
 
+import pytest
+
 from solnlib import _observability_flush as flushmod
 from solnlib._observability_flush import _AsyncFlushController
 
@@ -143,3 +149,114 @@ def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
     registry.add(MagicMock())
     registry.add(MagicMock())
     register.assert_called_once_with(flushmod._shutdown_all)
+
+
+def _run_child(script):
+    repo_root = Path(__file__).resolve().parents[2]
+    env = dict(os.environ, PYTHONPATH=str(repo_root))
+    return subprocess.run(
+        [sys.executable, "-c", script],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=3,
+    )
+
+
+def test_normal_exit_attempts_final_shutdown_without_explicit_flush():
+    result = _run_child(
+        """
+import logging
+from solnlib._observability_flush import _register_controller
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs): print('FINAL_SHUTDOWN', flush=True)
+_register_controller(Provider(), logging.getLogger('test'), str)
+"""
+    )
+    assert result.returncode == 0, result.stderr
+    assert "FINAL_SHUTDOWN" in result.stdout
+
+
+def test_blocked_final_export_uses_shortened_grace_period():
+    start = time.monotonic()
+    result = _run_child(
+        """
+import logging
+import threading
+import solnlib._observability_flush as flushmod
+flushmod._EXIT_GRACE_SECONDS = 0.05
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs):
+        print('SHUTDOWN_ENTERED', flush=True)
+        threading.Event().wait(10)
+flushmod._register_controller(Provider(), logging.getLogger('test'), str)
+"""
+    )
+    assert result.returncode == 0, result.stderr
+    assert "SHUTDOWN_ENTERED" in result.stdout
+    assert time.monotonic() - start < 1.5
+
+
+def test_real_otel_reader_does_not_restore_blocking_exit():
+    start = time.monotonic()
+    result = _run_child(
+        """
+import logging
+import threading
+from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
+import solnlib._observability_flush as flushmod
+from solnlib.observability import ObservabilityService
+
+flushmod._EXIT_GRACE_SECONDS = 0.5
+ObservabilityService._create_otlp_exporter = lambda self: None
+
+class SlowExporter(MetricExporter):
+    def export(self, metrics_data, timeout_millis=10_000, **kwargs):
+        print('OTEL_EXPORT_ENTERED', flush=True)
+        threading.Event().wait(10)
+        return MetricExportResult.SUCCESS
+    def shutdown(self, timeout_millis=30_000, **kwargs): pass
+    def force_flush(self, timeout_millis=10_000): return True
+
+service = ObservabilityService(
+    'test-input', logging.getLogger('test'), ta_name='test-ta',
+    ta_version='1.0.0', extra_exporters=[SlowExporter()],
+)
+service.event_count_counter.add(1)
+print('MAIN_DONE', flush=True)
+"""
+    )
+    assert result.returncode == 0, result.stderr
+    assert "MAIN_DONE" in result.stdout
+    assert "OTEL_EXPORT_ENTERED" in result.stdout
+    assert time.monotonic() - start < 2.0
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
+def test_child_gets_fresh_worker_after_fork():
+    result = _run_child(
+        """
+import logging
+import os
+import sys
+from solnlib._observability_flush import _register_controller
+class Provider:
+    def force_flush(self, **kwargs): pass
+    def shutdown(self, **kwargs): pass
+controller = _register_controller(Provider(), logging.getLogger('test'), str)
+parent_worker = controller._worker
+pid = os.fork()
+if pid == 0:
+    fresh = controller._worker is not parent_worker and controller._worker.is_alive()
+    print('CHILD_FRESH_WORKER=' + str(fresh), flush=True)
+    sys.exit(0 if fresh else 2)
+_, status = os.waitpid(pid, 0)
+print('PARENT_WORKER_ALIVE=' + str(parent_worker.is_alive()), flush=True)
+sys.exit(os.waitstatus_to_exitcode(status))
+"""
+    )
+    assert result.returncode == 0, result.stderr
+    assert "CHILD_FRESH_WORKER=True" in result.stdout
+    assert "PARENT_WORKER_ALIVE=True" in result.stdout

From 47d897dcbeeaaa48861750ed6b60630e755d4f67 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 17:55:41 +0200
Subject: [PATCH 07/11] docs: mark async flush implementation verified

---
 .../2026-10-12-async-observability-flush.md   | 38 +++++++++----------
 1 file changed, 19 insertions(+), 19 deletions(-)

diff --git a/docs/superpowers/plans/2026-10-12-async-observability-flush.md b/docs/superpowers/plans/2026-10-12-async-observability-flush.md
index 34c554a7..b89920dd 100644
--- a/docs/superpowers/plans/2026-10-12-async-observability-flush.md
+++ b/docs/superpowers/plans/2026-10-12-async-observability-flush.md
@@ -38,7 +38,7 @@
 - Produces: `_AsyncFlushController(provider, logger, format_error)`, `request_flush(timeout_millis: float) -> None`, `request_shutdown(deadline: float) -> None`, `join_until(deadline: float) -> None`, `restart_after_fork() -> None`.
 - Consumes: an OTel-like provider exposing `force_flush(timeout_millis=...)` and `shutdown(timeout_millis=...)`, a logger with `info`, and a safe exception formatter.
 
-- [ ] **Step 1: Write failing tests** in `tests/unit/test_observability_flush.py` for prompt return and coalescing. Use `threading.Event` gates so no network or sleeps are needed:
+- [x] **Step 1: Write failing tests** in `tests/unit/test_observability_flush.py` for prompt return and coalescing. Use `threading.Event` gates so no network or sleeps are needed:
 
 ```python
 import logging
@@ -102,8 +102,8 @@ def test_repeated_requests_during_export_coalesce_to_one_followup():
         controller.join_until(time.monotonic() + 1)
 ```
 
-- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect import failure because the private module does not exist.
-- [ ] **Step 3: Implement the worker** in `solnlib/_observability_flush.py` with one pending timeout slot; never hold its condition across an OTel call:
+- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect import failure because the private module does not exist.
+- [x] **Step 3: Implement the worker** in `solnlib/_observability_flush.py` with one pending timeout slot; never hold its condition across an OTel call:
 
 ```python
 import threading
@@ -166,7 +166,7 @@ class _AsyncFlushController:
         self._worker.start()
 ```
 
-- [ ] **Step 4: Add error/idempotency tests** below the first tests, then run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and expect PASS:
+- [x] **Step 4: Add error/idempotency tests** below the first tests, then run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and expect PASS:
 
 ```python
 def test_worker_logs_export_error_without_raising_to_caller():
@@ -196,7 +196,7 @@ def test_shutdown_is_once_and_later_flush_is_ignored():
     provider.shutdown.assert_called_once()
     provider.force_flush.assert_not_called()
 ```
-- [ ] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: add async metric flush worker"`.
+- [x] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: add async metric flush worker"`.
 
 ### Task 2: One process-wide exit deadline and fork reset
 
@@ -208,7 +208,7 @@ def test_shutdown_is_once_and_later_flush_is_ignored():
 - Consumes: `_AsyncFlushController` from Task 1.
 - Produces: module-level `_register_controller(provider, logger, format_error) -> _AsyncFlushController` backed by `_FlushRegistry`; `_shutdown_all()` is the `atexit` entry point.
 
-- [ ] **Step 1: Write failing registry tests** in the same test file. Use a local registry instance so tests never shut down production's module-level registry:
+- [x] **Step 1: Write failing registry tests** in the same test file. Use a local registry instance so tests never shut down production's module-level registry:
 
 ```python
 from solnlib import _observability_flush as flushmod
@@ -247,8 +247,8 @@ def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
     registry.add(MagicMock())
     register.assert_called_once_with(flushmod._shutdown_all)
 ```
-- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and confirm `_register_controller`/`_shutdown_all` are missing.
-- [ ] **Step 3: Implement the registry** in the private module. Use a global registry object and these exact operations; guard `os.register_at_fork` for platforms without fork:
+- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and confirm `_register_controller`/`_shutdown_all` are missing.
+- [x] **Step 3: Implement the registry** in the private module. Use a global registry object and these exact operations; guard `os.register_at_fork` for platforms without fork:
 
 ```python
 import atexit
@@ -300,8 +300,8 @@ def _shutdown_all():
     _registry.shutdown_all()
 ```
 
-- [ ] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`; verify the shared deadline and single registration assertions.
-- [ ] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: bound metric shutdown across services"`.
+- [x] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`; verify the shared deadline and single registration assertions.
+- [x] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: bound metric shutdown across services"`.
 
 ### Task 3: Wire the public solnlib API and document the contract change
 
@@ -313,7 +313,7 @@ def _shutdown_all():
 - Consumes: `_register_controller` and `_AsyncFlushController.request_flush` from Tasks 1–2.
 - Produces: unchanged public method signatures, but asynchronous `flush()` semantics and `MeterProvider(..., shutdown_on_exit=False)`.
 
-- [ ] **Step 1: Replace the old synchronous test** with this integration test in `tests/unit/test_observability.py`. Task 1 separately proves that the real controller runs exports off-thread:
+- [x] **Step 1: Replace the old synchronous test** with this integration test in `tests/unit/test_observability.py`. Task 1 separately proves that the real controller runs exports off-thread:
 
 ```python
 def test_flush_schedules_without_calling_provider_on_caller(logger, monkeypatch):
@@ -356,8 +356,8 @@ def test_flush_scheduling_error_uses_safe_str(logger, monkeypatch):
     svc.flush()
     assert any("SAFE" in call.args for call in logger.info.call_args_list)
 ```
-- [ ] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py -q`; expect the new thread/constructor assertions to fail.
-- [ ] **Step 3: Wire the constructor and method:**
+- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py -q`; expect the new thread/constructor assertions to fail.
+- [x] **Step 3: Wire the constructor and method:**
 
 ```python
 from ._observability_flush import _register_controller
@@ -383,8 +383,8 @@ if self._flush_controller is not None:
 ```
 
   Keep the existing `try/except` boundary around `flush()` so scheduling failures stay at INFO. Update `ObservabilityService.flush`, `StanzaObservabilityRecorder.flush`, `__exit__`, and module/class docstrings: returned flush means scheduled attempt, not completed attempt; `timeout_millis` is a worker budget; normal exit may wait up to 5 seconds total.
-- [ ] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q`. Ensure existing recorder `with` tests still call `flush()` once and propagate user exceptions.
-- [ ] **Step 5: Commit only intended files** with `git add -- solnlib/observability.py tests/unit/test_observability.py` and `git commit -m "feat: make solnlib observability flush nonblocking"`.
+- [x] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q`. Ensure existing recorder `with` tests still call `flush()` once and propagate user exceptions.
+- [x] **Step 5: Commit only intended files** with `git add -- solnlib/observability.py tests/unit/test_observability.py` and `git commit -m "feat: make solnlib observability flush nonblocking"`.
 
 ### Task 4: Process-exit regression tests and cross-version verification
 
@@ -396,7 +396,7 @@ if self._flush_controller is not None:
 - Consumes: final registry and public API from Tasks 1–3.
 - Produces: reproducible evidence that normal exit attempts a final export and an unavailable exporter cannot hold solnlib's exit hook beyond its grace period.
 
-- [ ] **Step 1: Add subprocess tests** to `tests/unit/test_observability_flush.py`, with these imports and tests. They use a fake provider and never contact a collector:
+- [x] **Step 1: Add subprocess tests** to `tests/unit/test_observability_flush.py`, with these imports and tests. They use a fake provider and never contact a collector:
 
 ```python
 import os
@@ -471,6 +471,6 @@ sys.exit(os.waitstatus_to_exitcode(status))
     assert "PARENT_WORKER_ALIVE=True" in result.stdout
 ```
 
-- [ ] **Step 2: Run exit/fork tests.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect all to pass; if the fork test fails, inspect whether the child's callback ran and repair `restart_after_fork()` before proceeding.
-- [ ] **Step 3: Verify both supported runtimes.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q` and `. .venv/bin/activate && pytest tests/unit -q`. Run the focused suite under Python 3.13 if dependencies are installed; otherwise run `python3.13 -m py_compile solnlib/observability.py solnlib/_observability_flush.py` and report that runtime tests on 3.13 remain CI-only. Check `git diff --check` and `git status --short`.
-- [ ] **Step 4: Commit only intended files** with `git add -- tests/unit/test_observability_flush.py solnlib/_observability_flush.py` and `git commit -m "test: cover bounded metric export at process exit"`.
+- [x] **Step 2: Run exit/fork tests.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect all to pass; if the fork test fails, inspect whether the child's callback ran and repair `restart_after_fork()` before proceeding.
+- [x] **Step 3: Verify both supported runtimes.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q` and `. .venv/bin/activate && pytest tests/unit -q`. Run the focused suite under Python 3.13 if dependencies are installed; otherwise run `python3.13 -m py_compile solnlib/observability.py solnlib/_observability_flush.py` and report that runtime tests on 3.13 remain CI-only. Check `git diff --check` and `git status --short`.
+- [x] **Step 4: Commit only intended files** with `git add -- tests/unit/test_observability_flush.py solnlib/_observability_flush.py` and `git commit -m "test: cover bounded metric export at process exit"`.

From 4555232d53b999195916fbc93431d37d2f1bb4bc Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Mon, 12 Oct 2026 18:58:45 +0200
Subject: [PATCH 08/11] fix: drain pending metric flush before shutdown

---
 ...-10-12-async-observability-flush-design.md | 16 +++---
 solnlib/_observability_flush.py               | 13 ++++-
 tests/unit/test_observability_flush.py        | 56 +++++++++++++++++++
 3 files changed, 77 insertions(+), 8 deletions(-)

diff --git a/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md b/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
index 71e58e04..55d090ca 100644
--- a/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
+++ b/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
@@ -43,13 +43,15 @@ Construct solnlib's `MeterProvider` with `shutdown_on_exit=False`; otherwise
 OTel's synchronous exit hook would bypass the grace period. Register one
 process-wide solnlib `atexit` handler. It signals shutdown to all service
 workers, then joins them against one monotonic 5-second deadline rather than
-waiting 5 seconds for each service. A worker that receives shutdown skips
-redundant queued flushes and calls `provider.shutdown()` once, which triggers
-the periodic reader's final collection. It passes the remaining grace period
-as the provider's shutdown timeout. If it is already inside a flush, it
-finishes that operation before shutdown; if that exceeds the grace period,
-the daemon worker can be abandoned at interpreter exit. The handler never
-starts a new thread, which is required on Python 3.12+.
+waiting 5 seconds for each service. A worker that receives shutdown performs
+one pending explicit flush, if present, within the remaining grace period,
+then calls `provider.shutdown()` once with the remaining budget. This matters
+when the periodic reader has an infinite interval: its shutdown does not
+perform a final collection. If the worker is already inside a flush, it
+finishes that operation before handling the pending request and shutdown; if
+the grace period expires, the daemon worker can be abandoned at interpreter
+exit. The handler never starts a new thread, which is required on Python
+3.12+.
 
 Service initialization that fails before a provider exists does not start a
 worker. The existing singleton cache continues to share one service and
diff --git a/solnlib/_observability_flush.py b/solnlib/_observability_flush.py
index d0c8c44b..e401ec01 100644
--- a/solnlib/_observability_flush.py
+++ b/solnlib/_observability_flush.py
@@ -51,7 +51,6 @@ def request_shutdown(self, deadline: float) -> None:
         with self._condition:
             if self._shutdown_deadline is None:
                 self._shutdown_deadline = deadline
-                self._pending_timeout = None
                 self._condition.notify()
 
     def join_until(self, deadline: float) -> None:
@@ -66,6 +65,18 @@ def _run(self) -> None:
                 timeout = self._pending_timeout
                 self._pending_timeout = None
             if deadline is not None:
+                if timeout is not None:
+                    # A reader with an infinite interval does not collect on shutdown.
+                    try:
+                        remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
+                        if remaining_ms:
+                            self._provider.force_flush(
+                                timeout_millis=min(timeout, remaining_ms)
+                            )
+                    except Exception as error:
+                        self._logger.info(
+                            "Failed to flush metrics: %s", self._format_error(error)
+                        )
                 try:
                     remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
                     self._provider.shutdown(timeout_millis=remaining_ms)
diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py
index 0d42ccbe..5e75e209 100644
--- a/tests/unit/test_observability_flush.py
+++ b/tests/unit/test_observability_flush.py
@@ -234,6 +234,62 @@ def force_flush(self, timeout_millis=10_000): return True
     assert time.monotonic() - start < 2.0
 
 
+def test_shutdown_exports_pending_flush_without_periodic_reader():
+    result = _run_child(
+        """
+import logging
+import os
+import threading
+import time
+from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
+from solnlib.observability import ObservabilityService
+
+os.environ['OTEL_METRIC_EXPORT_INTERVAL'] = 'inf'
+ObservabilityService._create_otlp_exporter = lambda self: None
+entered = threading.Event()
+release = threading.Event()
+exported_values = []
+
+class RecordingExporter(MetricExporter):
+    def export(self, metrics_data, timeout_millis=10_000, **kwargs):
+        value = sum(
+            point.value
+            for resource in metrics_data.resource_metrics
+            for scope in resource.scope_metrics
+            for metric in scope.metrics
+            if metric.name == 'splunk.addon.events'
+            for point in metric.data.data_points
+        )
+        exported_values.append(value)
+        if len(exported_values) == 1:
+            entered.set()
+            assert release.wait(1)
+        return MetricExportResult.SUCCESS
+
+    def shutdown(self, timeout_millis=30_000, **kwargs): pass
+    def force_flush(self, timeout_millis=10_000): return True
+
+service = ObservabilityService(
+    'test-input', logging.getLogger('test'), ta_name='test-ta',
+    ta_version='1.0.0', extra_exporters=[RecordingExporter()],
+)
+service.event_count_counter.add(1)
+service.flush()
+assert entered.wait(1)
+service.event_count_counter.add(2)
+service.flush()
+deadline = time.monotonic() + 1
+service._flush_controller.request_shutdown(deadline)
+release.set()
+service._flush_controller.join_until(deadline)
+print('EXPORTED_VALUES=' + repr(exported_values), flush=True)
+assert exported_values == [1, 3]
+"""
+    )
+    assert result.returncode == 0, result.stderr + result.stdout
+    assert "EXPORTED_VALUES=[1, 3]" in result.stdout
+
+
 @pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
 def test_child_gets_fresh_worker_after_fork():
     result = _run_child(

From 1d7bdc48b673b634c991a72d3c61ef197405709e Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Tue, 20 Oct 2026 17:27:07 +0200
Subject: [PATCH 09/11] docs: remove spec files

---
 .../2026-10-12-async-observability-flush.md   | 476 ------------------
 ...-10-12-async-observability-flush-design.md |  89 ----
 2 files changed, 565 deletions(-)
 delete mode 100644 docs/superpowers/plans/2026-10-12-async-observability-flush.md
 delete mode 100644 docs/superpowers/specs/2026-10-12-async-observability-flush-design.md

diff --git a/docs/superpowers/plans/2026-10-12-async-observability-flush.md b/docs/superpowers/plans/2026-10-12-async-observability-flush.md
deleted file mode 100644
index b89920dd..00000000
--- a/docs/superpowers/plans/2026-10-12-async-observability-flush.md
+++ /dev/null
@@ -1,476 +0,0 @@
-# Asynchronous Observability Flush Implementation Plan
-
-> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
-
-**Goal:** Make solnlib's public metric `flush()` non-blocking while allowing one process-wide 5-second grace period for final metric export.
-
-**Architecture:** A private daemon worker per initialized observability service serializes explicit flush and final provider shutdown. A process-wide registry registers one `atexit` hook, signals all workers, and joins them against a single monotonic deadline; solnlib-created OTel providers disable their own synchronous exit hook.
-
-**Tech Stack:** Python 3.9–3.13, OpenTelemetry Python SDK >=1.39.1,<2, pytest, `threading`, `atexit`.
-
-**Spec:** `docs/superpowers/specs/2026-10-12-async-observability-flush-design.md`
-
-## Global Constraints
-
-- Preserve `ObservabilityService.flush(timeout_millis: float = 30_000) -> None` and `StanzaObservabilityRecorder.flush() -> None` signatures while documenting the changed asynchronous semantics.
-- Keep periodic OTel export; use `MeterProvider(shutdown_on_exit=False)` only for providers constructed by solnlib.
-- Give all solnlib workers one total 5-second grace period at normal exit, not 5 seconds each; no thread creation inside `atexit`.
-- Keep telemetry failures at INFO and never propagate them into modular-input code.
-- Treat pending final metrics as lossy after the grace period; unrelated `atexit` handlers and non-daemon threads are outside this guarantee.
-- Keep existing untracked files untouched. Edit only the files listed below and stage/commit only intended files.
-
-## File structure
-
-- Create `solnlib/_observability_flush.py`: private worker, registry, process-exit and fork lifecycle; independent of `observability.py` to avoid growing its already-large public module.
-- Create `tests/unit/test_observability_flush.py`: deterministic worker/registry/concurrency tests and subprocess exit/fork smoke tests.
-- Modify `solnlib/observability.py`: wire the private manager into `ObservabilityService`, disable OTel's own exit handler, and update public docstrings.
-- Modify `tests/unit/test_observability.py`: replace synchronous-flush assertions with asynchronous assertions and cover recorder compatibility.
-
----
-
-### Task 1: Daemon worker and flush coalescing
-
-**Files:**
-- Create: `solnlib/_observability_flush.py`
-- Create: `tests/unit/test_observability_flush.py`
-
-**Interfaces:**
-- Produces: `_AsyncFlushController(provider, logger, format_error)`, `request_flush(timeout_millis: float) -> None`, `request_shutdown(deadline: float) -> None`, `join_until(deadline: float) -> None`, `restart_after_fork() -> None`.
-- Consumes: an OTel-like provider exposing `force_flush(timeout_millis=...)` and `shutdown(timeout_millis=...)`, a logger with `info`, and a safe exception formatter.
-
-- [x] **Step 1: Write failing tests** in `tests/unit/test_observability_flush.py` for prompt return and coalescing. Use `threading.Event` gates so no network or sleeps are needed:
-
-```python
-import logging
-import threading
-import time
-from unittest.mock import MagicMock
-
-from solnlib._observability_flush import _AsyncFlushController
-
-
-def test_request_flush_returns_while_export_is_blocked():
-    entered = threading.Event()
-    release = threading.Event()
-    provider = MagicMock()
-
-    def blocked_flush(**kwargs):
-        entered.set()
-        assert release.wait(2)
-
-    provider.force_flush.side_effect = blocked_flush
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
-    try:
-        start = time.monotonic()
-        controller.request_flush(30_000)
-        assert time.monotonic() - start < 0.2
-        assert entered.wait(1)
-        provider.force_flush.assert_called_once_with(timeout_millis=30_000)
-    finally:
-        release.set()
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
-
-
-def test_repeated_requests_during_export_coalesce_to_one_followup():
-    entered = threading.Event()
-    release = threading.Event()
-    second_done = threading.Event()
-    provider = MagicMock()
-
-    def flush(**kwargs):
-        if provider.force_flush.call_count == 1:
-            entered.set()
-            assert release.wait(2)
-        else:
-            second_done.set()
-
-    provider.force_flush.side_effect = flush
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
-    try:
-        controller.request_flush(30_000)
-        assert entered.wait(1)
-        controller.request_flush(10_000)
-        controller.request_flush(5_000)
-        release.set()
-        assert second_done.wait(1)
-        assert provider.force_flush.call_count == 2
-        assert provider.force_flush.call_args_list[-1].kwargs == {"timeout_millis": 5_000}
-    finally:
-        release.set()
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
-```
-
-- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect import failure because the private module does not exist.
-- [x] **Step 3: Implement the worker** in `solnlib/_observability_flush.py` with one pending timeout slot; never hold its condition across an OTel call:
-
-```python
-import threading
-import time
-from typing import Callable
-
-
-class _AsyncFlushController:
-    def __init__(self, provider, logger, format_error: Callable[[BaseException], str]):
-        self._provider = provider
-        self._logger = logger
-        self._format_error = format_error
-        self._condition = threading.Condition()
-        self._pending_timeout = None
-        self._shutdown_deadline = None
-        self._worker = threading.Thread(target=self._run, name="SolnlibMetricFlush", daemon=True)
-        self._worker.start()
-
-    def request_flush(self, timeout_millis: float) -> None:
-        with self._condition:
-            if self._shutdown_deadline is None:
-                self._pending_timeout = int(timeout_millis)
-                self._condition.notify()
-
-    def request_shutdown(self, deadline: float) -> None:
-        with self._condition:
-            if self._shutdown_deadline is None:
-                self._shutdown_deadline = deadline
-                self._pending_timeout = None
-                self._condition.notify()
-
-    def join_until(self, deadline: float) -> None:
-        self._worker.join(timeout=max(0.0, deadline - time.monotonic()))
-
-    def _run(self) -> None:
-        while True:
-            with self._condition:
-                while self._pending_timeout is None and self._shutdown_deadline is None:
-                    self._condition.wait()
-                deadline = self._shutdown_deadline
-                timeout = self._pending_timeout
-                self._pending_timeout = None
-            if deadline is not None:
-                try:
-                    remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
-                    self._provider.shutdown(timeout_millis=remaining_ms)
-                except Exception as error:
-                    self._logger.info("Failed to shut down metrics: %s", self._format_error(error))
-                return
-            try:
-                self._provider.force_flush(timeout_millis=timeout)
-            except Exception as error:
-                self._logger.info("Failed to flush metrics: %s", self._format_error(error))
-
-    def restart_after_fork(self) -> None:
-        self._condition = threading.Condition()
-        self._pending_timeout = None
-        self._shutdown_deadline = None
-        self._worker = threading.Thread(target=self._run, name="SolnlibMetricFlush", daemon=True)
-        self._worker.start()
-```
-
-- [x] **Step 4: Add error/idempotency tests** below the first tests, then run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and expect PASS:
-
-```python
-def test_worker_logs_export_error_without_raising_to_caller():
-    logged = threading.Event()
-    provider = MagicMock()
-    provider.force_flush.side_effect = RuntimeError("boom")
-    logger = MagicMock(spec=logging.Logger)
-    logger.info.side_effect = lambda *args: logged.set()
-    controller = _AsyncFlushController(provider, logger, str)
-    try:
-        controller.request_flush(30_000)
-        assert logged.wait(1)
-        assert "boom" in logger.info.call_args.args[-1]
-    finally:
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
-
-
-def test_shutdown_is_once_and_later_flush_is_ignored():
-    provider = MagicMock()
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
-    deadline = time.monotonic() + 1
-    controller.request_shutdown(deadline)
-    controller.request_shutdown(deadline)
-    controller.join_until(deadline)
-    controller.request_flush(30_000)
-    provider.shutdown.assert_called_once()
-    provider.force_flush.assert_not_called()
-```
-- [x] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: add async metric flush worker"`.
-
-### Task 2: One process-wide exit deadline and fork reset
-
-**Files:**
-- Modify: `solnlib/_observability_flush.py`
-- Modify: `tests/unit/test_observability_flush.py`
-
-**Interfaces:**
-- Consumes: `_AsyncFlushController` from Task 1.
-- Produces: module-level `_register_controller(provider, logger, format_error) -> _AsyncFlushController` backed by `_FlushRegistry`; `_shutdown_all()` is the `atexit` entry point.
-
-- [x] **Step 1: Write failing registry tests** in the same test file. Use a local registry instance so tests never shut down production's module-level registry:
-
-```python
-from solnlib import _observability_flush as flushmod
-
-
-def test_exit_budget_is_shared_by_two_controllers(monkeypatch):
-    monkeypatch.setattr(flushmod, "_EXIT_GRACE_SECONDS", 0.05)
-    release = threading.Event()
-    entered = [threading.Event(), threading.Event()]
-    controllers = []
-    registry = flushmod._FlushRegistry()
-    for gate in entered:
-        provider = MagicMock()
-        provider.shutdown.side_effect = lambda gate=gate, **kwargs: (gate.set(), release.wait(2))
-        controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
-        registry._controllers.append(controller)
-        controllers.append(controller)
-    try:
-        start = time.monotonic()
-        registry.shutdown_all()
-        assert time.monotonic() - start < 0.2
-        assert all(gate.wait(1) for gate in entered)
-    finally:
-        release.set()
-        for controller in controllers:
-            controller.join_until(time.monotonic() + 1)
-
-
-def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
-    register = MagicMock()
-    monkeypatch.setattr(flushmod.atexit, "register", register)
-    if hasattr(flushmod.os, "register_at_fork"):
-        monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock())
-    registry = flushmod._FlushRegistry()
-    registry.add(MagicMock())
-    registry.add(MagicMock())
-    register.assert_called_once_with(flushmod._shutdown_all)
-```
-- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q` and confirm `_register_controller`/`_shutdown_all` are missing.
-- [x] **Step 3: Implement the registry** in the private module. Use a global registry object and these exact operations; guard `os.register_at_fork` for platforms without fork:
-
-```python
-import atexit
-import os
-
-_EXIT_GRACE_SECONDS = 5.0
-
-
-class _FlushRegistry:
-    def __init__(self):
-        self._lock = threading.Lock()
-        self._controllers = []
-        self._registered = False
-
-    def add(self, controller):
-        with self._lock:
-            if not self._registered:
-                atexit.register(_shutdown_all)
-                if hasattr(os, "register_at_fork"):
-                    os.register_at_fork(after_in_child=self._after_fork_child)
-                self._registered = True
-            self._controllers.append(controller)
-
-    def shutdown_all(self):
-        deadline = time.monotonic() + _EXIT_GRACE_SECONDS
-        with self._lock:
-            controllers = tuple(self._controllers)
-        for controller in controllers:
-            controller.request_shutdown(deadline)
-        for controller in controllers:
-            controller.join_until(deadline)
-
-    def _after_fork_child(self):
-        self._lock = threading.Lock()
-        for controller in self._controllers:
-            controller.restart_after_fork()
-
-
-_registry = _FlushRegistry()
-
-
-def _register_controller(provider, logger, format_error):
-    controller = _AsyncFlushController(provider, logger, format_error)
-    _registry.add(controller)
-    return controller
-
-
-def _shutdown_all():
-    _registry.shutdown_all()
-```
-
-- [x] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`; verify the shared deadline and single registration assertions.
-- [x] **Step 5: Commit only these two files** with `git add -- solnlib/_observability_flush.py tests/unit/test_observability_flush.py` and `git commit -m "feat: bound metric shutdown across services"`.
-
-### Task 3: Wire the public solnlib API and document the contract change
-
-**Files:**
-- Modify: `solnlib/observability.py:45-65,410-465,713-733,1015-1046`
-- Modify: `tests/unit/test_observability.py:603-615,947-974,1557-1570,1730-1760`
-
-**Interfaces:**
-- Consumes: `_register_controller` and `_AsyncFlushController.request_flush` from Tasks 1–2.
-- Produces: unchanged public method signatures, but asynchronous `flush()` semantics and `MeterProvider(..., shutdown_on_exit=False)`.
-
-- [x] **Step 1: Replace the old synchronous test** with this integration test in `tests/unit/test_observability.py`. Task 1 separately proves that the real controller runs exports off-thread:
-
-```python
-def test_flush_schedules_without_calling_provider_on_caller(logger, monkeypatch):
-    controller = MagicMock()
-    provider_cls = MagicMock()
-    monkeypatch.setattr("solnlib.observability.MeterProvider", provider_cls)
-    monkeypatch.setattr(
-        "solnlib.observability.ObservabilityService._create_otlp_exporter",
-        lambda self: None,
-    )
-    monkeypatch.setattr(
-        "solnlib.observability._register_controller",
-        lambda provider, logger, formatter: controller,
-    )
-    svc = ObservabilityService("test-input", logger, ta_name="my_ta", ta_version="1.0.0")
-    svc.flush(timeout_millis=5000)
-    controller.request_flush.assert_called_once_with(5000)
-    provider_cls.return_value.force_flush.assert_not_called()
-    assert provider_cls.call_args.kwargs["shutdown_on_exit"] is False
-```
-
-  Keep the provider-missing no-op test. Replace the old exception tests with these scheduling-failure assertions; worker-side errors are covered in Task 1:
-
-```python
-def test_flush_logs_info_when_scheduling_fails(logger, monkeypatch):
-    svc = _make_service(logger, monkeypatch)
-    svc._flush_controller = MagicMock()
-    svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
-    svc.flush()
-    assert any("boom" in call.args for call in logger.info.call_args_list)
-    logger.warning.assert_not_called()
-
-
-def test_flush_scheduling_error_uses_safe_str(logger, monkeypatch):
-    monkeypatch.setattr("solnlib.observability._safe_exception_str", lambda error: "SAFE")
-    svc = _make_service(logger, monkeypatch)
-    svc._flush_controller = MagicMock()
-    svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
-    logger.reset_mock()
-    svc.flush()
-    assert any("SAFE" in call.args for call in logger.info.call_args_list)
-```
-- [x] **Step 2: Verify red.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py -q`; expect the new thread/constructor assertions to fail.
-- [x] **Step 3: Wire the constructor and method:**
-
-```python
-from ._observability_flush import _register_controller
-
-# At the beginning of ObservabilityService.__init__:
-self._flush_controller = None
-
-# Replace the provider constructor:
-self._provider = MeterProvider(
-    resource=resource,
-    metric_readers=metric_readers,
-    shutdown_on_exit=False,
-)
-
-# After counters are created and before the success log:
-self._flush_controller = _register_controller(
-    self._provider, self._logger, _safe_exception_str
-)
-
-# Replace the body of ObservabilityService.flush after the provider guard:
-if self._flush_controller is not None:
-    self._flush_controller.request_flush(timeout_millis)
-```
-
-  Keep the existing `try/except` boundary around `flush()` so scheduling failures stay at INFO. Update `ObservabilityService.flush`, `StanzaObservabilityRecorder.flush`, `__exit__`, and module/class docstrings: returned flush means scheduled attempt, not completed attempt; `timeout_millis` is a worker budget; normal exit may wait up to 5 seconds total.
-- [x] **Step 4: Verify green** with `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q`. Ensure existing recorder `with` tests still call `flush()` once and propagate user exceptions.
-- [x] **Step 5: Commit only intended files** with `git add -- solnlib/observability.py tests/unit/test_observability.py` and `git commit -m "feat: make solnlib observability flush nonblocking"`.
-
-### Task 4: Process-exit regression tests and cross-version verification
-
-**Files:**
-- Modify: `tests/unit/test_observability_flush.py`
-- Modify: `solnlib/_observability_flush.py` only if the tests expose an issue.
-
-**Interfaces:**
-- Consumes: final registry and public API from Tasks 1–3.
-- Produces: reproducible evidence that normal exit attempts a final export and an unavailable exporter cannot hold solnlib's exit hook beyond its grace period.
-
-- [x] **Step 1: Add subprocess tests** to `tests/unit/test_observability_flush.py`, with these imports and tests. They use a fake provider and never contact a collector:
-
-```python
-import os
-import subprocess
-import sys
-from pathlib import Path
-
-import pytest
-
-
-def _run_child(script):
-    repo_root = Path(__file__).resolve().parents[2]
-    env = dict(os.environ, PYTHONPATH=str(repo_root))
-    return subprocess.run(
-        [sys.executable, "-c", script], capture_output=True, text=True,
-        env=env, timeout=3,
-    )
-
-
-def test_normal_exit_attempts_final_shutdown_without_explicit_flush():
-    result = _run_child("""
-import logging
-from solnlib._observability_flush import _register_controller
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs): print('FINAL_SHUTDOWN', flush=True)
-_register_controller(Provider(), logging.getLogger('test'), str)
-""")
-    assert result.returncode == 0, result.stderr
-    assert "FINAL_SHUTDOWN" in result.stdout
-
-
-def test_blocked_final_export_uses_shortened_grace_period():
-    start = time.monotonic()
-    result = _run_child("""
-import logging
-import threading
-import solnlib._observability_flush as flushmod
-flushmod._EXIT_GRACE_SECONDS = 0.05
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs): threading.Event().wait(10)
-flushmod._register_controller(Provider(), logging.getLogger('test'), str)
-""")
-    assert result.returncode == 0, result.stderr
-    assert time.monotonic() - start < 1.5
-
-
-@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
-def test_child_gets_fresh_worker_after_fork():
-    result = _run_child("""
-import logging
-import os
-import sys
-from solnlib._observability_flush import _register_controller
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs): pass
-controller = _register_controller(Provider(), logging.getLogger('test'), str)
-parent_worker = controller._worker
-pid = os.fork()
-if pid == 0:
-    fresh = controller._worker is not parent_worker and controller._worker.is_alive()
-    print('CHILD_FRESH_WORKER=' + str(fresh), flush=True)
-    sys.exit(0 if fresh else 2)
-_, status = os.waitpid(pid, 0)
-print('PARENT_WORKER_ALIVE=' + str(parent_worker.is_alive()), flush=True)
-sys.exit(os.waitstatus_to_exitcode(status))
-""")
-    assert result.returncode == 0, result.stderr
-    assert "CHILD_FRESH_WORKER=True" in result.stdout
-    assert "PARENT_WORKER_ALIVE=True" in result.stdout
-```
-
-- [x] **Step 2: Run exit/fork tests.** Run `. .venv/bin/activate && pytest tests/unit/test_observability_flush.py -q`. Expect all to pass; if the fork test fails, inspect whether the child's callback ran and repair `restart_after_fork()` before proceeding.
-- [x] **Step 3: Verify both supported runtimes.** Run `. .venv/bin/activate && pytest tests/unit/test_observability.py tests/unit/test_observability_flush.py -q` and `. .venv/bin/activate && pytest tests/unit -q`. Run the focused suite under Python 3.13 if dependencies are installed; otherwise run `python3.13 -m py_compile solnlib/observability.py solnlib/_observability_flush.py` and report that runtime tests on 3.13 remain CI-only. Check `git diff --check` and `git status --short`.
-- [x] **Step 4: Commit only intended files** with `git add -- tests/unit/test_observability_flush.py solnlib/_observability_flush.py` and `git commit -m "test: cover bounded metric export at process exit"`.
diff --git a/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md b/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
deleted file mode 100644
index 55d090ca..00000000
--- a/docs/superpowers/specs/2026-10-12-async-observability-flush-design.md
+++ /dev/null
@@ -1,89 +0,0 @@
-# Asynchronous observability flush with bounded exit grace
-
-## Context
-
-`ObservabilityService.flush()` currently calls `MeterProvider.force_flush()` on
-the caller's thread. A missing or slow OTLP collector can therefore delay a
-modular input at every `StanzaObservabilityRecorder.flush()` or context-manager
-exit. OpenTelemetry's `MeterProvider` also registers a synchronous `atexit`
-shutdown by default, so moving only `force_flush()` to another thread would
-still leave process exit potentially blocked.
-
-## Goals and contract
-
-- Keep `ObservabilityService.flush(timeout_millis=30_000)` and
-  `StanzaObservabilityRecorder.flush()` callable without changing their return
-  types. `flush()` schedules export work and returns promptly; it no longer
-  confirms delivery before returning. The timeout remains the budget passed
-  to the background `force_flush()` call, not a wait on the caller's thread.
-- Preserve OTel's periodic export while the process is alive.
-- On normal interpreter exit, give pending work and a final OTel shutdown a
-  **single process-wide grace period of at most 5 seconds** in solnlib's exit
-  handler. If work is unfinished, allow the process to exit and lose the last
-  metrics rather than hold the caller indefinitely.
-- Keep telemetry failures auxiliary: log them at INFO and do not raise into
-  modular-input code.
-
-This grace period bounds only solnlib's own exit handler. Other application
-`atexit` handlers, non-daemon threads, and separately created OTel providers
-remain outside its control. Abrupt termination (`SIGKILL`, `os._exit()`, etc.)
-does not invoke `atexit` and cannot guarantee a final export.
-
-## Design
-
-Each successfully initialized `ObservabilityService` owns one daemon worker,
-started during service initialization. The worker waits for requests and is
-the only solnlib thread that calls that service's provider `force_flush()` or
-`shutdown()`. A condition-protected pending flag coalesces repeated flush
-requests, so callers do not create an unbounded queue or new thread per call.
-If a request arrives during an export, the worker performs one further flush
-after the current one. A later request may replace the pending timeout budget.
-
-Construct solnlib's `MeterProvider` with `shutdown_on_exit=False`; otherwise
-OTel's synchronous exit hook would bypass the grace period. Register one
-process-wide solnlib `atexit` handler. It signals shutdown to all service
-workers, then joins them against one monotonic 5-second deadline rather than
-waiting 5 seconds for each service. A worker that receives shutdown performs
-one pending explicit flush, if present, within the remaining grace period,
-then calls `provider.shutdown()` once with the remaining budget. This matters
-when the periodic reader has an infinite interval: its shutdown does not
-perform a final collection. If the worker is already inside a flush, it
-finishes that operation before handling the pending request and shutdown; if
-the grace period expires, the daemon worker can be abandoned at interpreter
-exit. The handler never starts a new thread, which is required on Python
-3.12+.
-
-Service initialization that fails before a provider exists does not start a
-worker. The existing singleton cache continues to share one service and
-worker per modular-input type. Directly constructed services are also tracked
-for exit cleanup. Shutdown requests are idempotent; flush requests made after
-shutdown begins are ignored. Worker exceptions are caught and logged without
-terminating the modular input.
-
-Register an `after_in_child` fork callback. In the child it replaces inherited
-manager and service synchronization objects, discards parent worker references,
-and starts one fresh daemon worker for each inherited live service before any
-child `atexit` handler can run. The parent keeps its workers unchanged. The
-callback does not call a provider's exporter or perform network I/O.
-
-## Compatibility and trade-offs
-
-This is an intentional semantic change: a returned `flush()` no longer means
-that metrics reached an exporter. Existing callers requiring confirmed
-delivery need a separate synchronous API; this change does not silently
-promise that guarantee. Normal periodic export continues independently.
-The added daemon worker costs one thread per active service. A slow export can
-coalesce multiple explicit flush requests, and the final 5-second deadline
-can sacrifice data when the collector is unavailable. The exporter circuit
-breaker remains unchanged and does not itself make flush non-blocking.
-
-## Verification
-
-Add deterministic unit tests for prompt return under a blocked exporter,
-coalescing, follow-up requests, exception logging, provider absence, shutdown
-idempotency, and the single deadline across multiple services. Add subprocess
-tests for normal exit with successful final delivery and a blocked exporter
-with a shortened test grace period. Exercise the supported Python versions,
-including Python 3.13's prohibition on starting threads in `atexit`, and a
-fork smoke test where available. Run the observability unit suite and the
-repository's broader relevant tests.

From 9b3a2a59460a3aa11b4033b879746b7d47078c69 Mon Sep 17 00:00:00 2001
From: Wojciech Tobis <wtobis@splunk.com>
Date: Wed, 21 Oct 2026 01:07:00 +0200
Subject: [PATCH 10/11] fix: harden async observability lifecycle

---
 solnlib/_observability_flush.py               |  89 ++--
 solnlib/observability.py                      |  53 ++-
 .../helpers/observability_flush_process.py    |  86 ++++
 tests/unit/test_observability.py              |  90 +++-
 tests/unit/test_observability_flush.py        | 428 ++++++++++--------
 tests/unit/test_observability_lifecycle.py    |  72 +++
 6 files changed, 561 insertions(+), 257 deletions(-)
 create mode 100644 tests/unit/helpers/observability_flush_process.py
 create mode 100644 tests/unit/test_observability_lifecycle.py

diff --git a/solnlib/_observability_flush.py b/solnlib/_observability_flush.py
index e401ec01..92d3dec1 100644
--- a/solnlib/_observability_flush.py
+++ b/solnlib/_observability_flush.py
@@ -19,11 +19,15 @@
 import os
 import threading
 import time
-from typing import Callable
+from typing import Callable, Optional
 
 _EXIT_GRACE_SECONDS = 5.0
 
 
+def _remaining_ms(deadline: float) -> int:
+    return max(0, int((deadline - time.monotonic()) * 1000))
+
+
 class _AsyncFlushController:
     """Serialize one provider's explicit flushes and final shutdown."""
 
@@ -58,39 +62,40 @@ def join_until(self, deadline: float) -> None:
 
     def _run(self) -> None:
         while True:
-            with self._condition:
-                while self._pending_timeout is None and self._shutdown_deadline is None:
-                    self._condition.wait()
-                deadline = self._shutdown_deadline
-                timeout = self._pending_timeout
-                self._pending_timeout = None
-            if deadline is not None:
-                if timeout is not None:
-                    # A reader with an infinite interval does not collect on shutdown.
-                    try:
-                        remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
-                        if remaining_ms:
-                            self._provider.force_flush(
-                                timeout_millis=min(timeout, remaining_ms)
-                            )
-                    except Exception as error:
-                        self._logger.info(
-                            "Failed to flush metrics: %s", self._format_error(error)
-                        )
-                try:
-                    remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
-                    self._provider.shutdown(timeout_millis=remaining_ms)
-                except Exception as error:
-                    self._logger.info(
-                        "Failed to shut down metrics: %s", self._format_error(error)
-                    )
-                return
-            try:
-                self._provider.force_flush(timeout_millis=timeout)
-            except Exception as error:
-                self._logger.info(
-                    "Failed to flush metrics: %s", self._format_error(error)
-                )
+            flush_timeout, shutdown_deadline = self._wait_for_request()
+            if shutdown_deadline is None:
+                if flush_timeout is not None:
+                    self._flush(flush_timeout)
+                continue
+            self._shut_down(flush_timeout, shutdown_deadline)
+            return
+
+    def _has_request(self) -> bool:
+        return self._pending_timeout is not None or self._shutdown_deadline is not None
+
+    def _wait_for_request(self) -> tuple[Optional[int], Optional[float]]:
+        with self._condition:
+            self._condition.wait_for(self._has_request)
+            flush_timeout, self._pending_timeout = self._pending_timeout, None
+            return flush_timeout, self._shutdown_deadline
+
+    def _shut_down(self, flush_timeout: Optional[int], deadline: float) -> None:
+        # Readers with an infinite export interval do not collect on shutdown.
+        remaining = _remaining_ms(deadline)
+        if flush_timeout is not None and remaining:
+            self._flush(min(flush_timeout, remaining))
+        try:
+            self._provider.shutdown(timeout_millis=_remaining_ms(deadline))
+        except Exception as error:
+            self._logger.info(
+                "Failed to shut down metrics: %s", self._format_error(error)
+            )
+
+    def _flush(self, timeout_ms: int) -> None:
+        try:
+            self._provider.force_flush(timeout_millis=timeout_ms)
+        except Exception as error:
+            self._logger.info("Failed to flush metrics: %s", self._format_error(error))
 
     def restart_after_fork(self) -> None:
         """Replace parent-thread state after fork in the child process."""
@@ -110,19 +115,27 @@ def __init__(self) -> None:
         self._lock = threading.Lock()
         self._controllers = []
         self._registered = False
+        self._closed = False
 
-    def add(self, controller: _AsyncFlushController) -> None:
+    def register(self, provider, logger, format_error) -> _AsyncFlushController:
         with self._lock:
+            if self._closed:
+                raise RuntimeError("Metrics shutdown has started")
             if not self._registered:
                 atexit.register(_shutdown_all)
                 if hasattr(os, "register_at_fork"):
                     os.register_at_fork(after_in_child=self._after_fork_child)
                 self._registered = True
+            controller = _AsyncFlushController(provider, logger, format_error)
             self._controllers.append(controller)
+            return controller
 
     def shutdown_all(self) -> None:
         deadline = time.monotonic() + _EXIT_GRACE_SECONDS
         with self._lock:
+            if self._closed:
+                return
+            self._closed = True
             controllers = tuple(self._controllers)
         for controller in controllers:
             controller.request_shutdown(deadline)
@@ -131,6 +144,8 @@ def shutdown_all(self) -> None:
 
     def _after_fork_child(self) -> None:
         self._lock = threading.Lock()
+        if self._closed:
+            return
         for controller in self._controllers:
             controller.restart_after_fork()
 
@@ -139,9 +154,7 @@ def _after_fork_child(self) -> None:
 
 
 def _register_controller(provider, logger, format_error) -> _AsyncFlushController:
-    controller = _AsyncFlushController(provider, logger, format_error)
-    _registry.add(controller)
-    return controller
+    return _registry.register(provider, logger, format_error)
 
 
 def _shutdown_all() -> None:
diff --git a/solnlib/observability.py b/solnlib/observability.py
index d809948c..a8249e1f 100644
--- a/solnlib/observability.py
+++ b/solnlib/observability.py
@@ -416,6 +416,7 @@ def __init__(
         for logger_name in _METRICS_SDK_LOGGERS:
             logging.getLogger(logger_name).addFilter(_downgrade_to_info_filter)
 
+        provider = None
         try:
             if type(modinput_type) is not str or not _is_safe_identifier_str(
                 modinput_type
@@ -453,40 +454,54 @@ def __init__(
             for exporter in extra_exporters or []:
                 metric_readers.append(PeriodicExportingMetricReader(exporter))
 
-            self._provider = MeterProvider(
+            provider = MeterProvider(
                 resource=resource,
                 metric_readers=metric_readers,
                 shutdown_on_exit=False,
             )
-            self._meter = self._provider.get_meter(ta_name, ta_version)
+            meter = provider.get_meter(ta_name, ta_version)
 
-            self.event_count_counter = self._meter.create_counter(
+            event_count_counter = meter.create_counter(
                 name="splunk.addon.events",
                 description="Number of events ingested by the add-on modular input",
                 unit="1",
             )
-            self.event_bytes_counter = self._meter.create_counter(
+            event_bytes_counter = meter.create_counter(
                 name="splunk.addon.events.bytes",
                 description="Volume of data ingested by the add-on modular input",
                 unit="By",
             )
 
-            self._flush_controller = _register_controller(
-                self._provider, self._logger, _safe_exception_str
-            )
-
-            self._logger.info(
-                "ObservabilityService initialised: ta_name=%s ta_version=%s "
-                "modinput_type=%s",
-                ta_name,
-                ta_version,
-                modinput_type,
+            controller = _register_controller(
+                provider, self._logger, _safe_exception_str
             )
         except Exception as error:
+            if provider is not None:
+                try:
+                    provider.shutdown()
+                except Exception as cleanup_error:
+                    self._logger.info(
+                        "Failed to clean up metrics after init failure: %s",
+                        _safe_exception_str(cleanup_error),
+                    )
             self._logger.info(
                 "Failed to initialise ObservabilityService: %s",
                 _safe_exception_str(error),
             )
+            return
+
+        self._provider = provider
+        self._meter = meter
+        self.event_count_counter = event_count_counter
+        self.event_bytes_counter = event_bytes_counter
+        self._flush_controller = controller
+        self._logger.info(
+            "ObservabilityService initialised: ta_name=%s ta_version=%s "
+            "modinput_type=%s",
+            ta_name,
+            ta_version,
+            modinput_type,
+        )
 
     def _read_ta_info(self) -> tuple[Optional[str], Optional[str]]:
         """Read the add-on name and version from ``app.conf``.
@@ -665,6 +680,7 @@ def _create_otlp_exporter(self) -> Optional[MetricExporter]:
             exporter = OTLPMetricExporter(
                 endpoint=endpoint,
                 credentials=credentials,
+                timeout=2.0,
                 preferred_temporality={
                     Counter: AggregationTemporality.DELTA,
                     Histogram: AggregationTemporality.DELTA,
@@ -732,9 +748,12 @@ def flush(self, timeout_millis: float = 30_000) -> None:
         :meth:`StanzaObservabilityRecorder.flush` on exit automatically.
 
         Args:
-            timeout_millis: Budget passed to the background force-flush call,
-                in milliseconds.  It is not a caller-side wait.  Defaults to
-                30 seconds.
+            timeout_millis: Budget forwarded to the background
+                ``MeterProvider.force_flush()`` call, in milliseconds.
+                The budget starts when that call begins; waiting for the
+                worker is not included. Readers and exporters may exceed
+                this budget. It is not a hard limit on export duration.
+                Defaults to 30 seconds.
         """
         if self._provider is None or self._flush_controller is None:
             return
diff --git a/tests/unit/helpers/observability_flush_process.py b/tests/unit/helpers/observability_flush_process.py
new file mode 100644
index 00000000..d92af18f
--- /dev/null
+++ b/tests/unit/helpers/observability_flush_process.py
@@ -0,0 +1,86 @@
+#
+# Copyright 2026 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Executable scenarios for process-level observability regression tests."""
+
+import logging
+import os
+import sys
+import threading
+
+from solnlib import _observability_flush as flushmod
+
+
+def blocked_otel_exit():
+    from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
+
+    from solnlib.observability import ObservabilityService
+
+    class SlowExporter(MetricExporter):
+        def export(self, metrics_data, timeout_millis=10_000, **kwargs):
+            print("OTEL_EXPORT_ENTERED", flush=True)
+            threading.Event().wait(10)
+            return MetricExportResult.SUCCESS
+
+        def shutdown(self, timeout_millis=30_000, **kwargs):
+            pass
+
+        def force_flush(self, timeout_millis=10_000):
+            return True
+
+    flushmod._EXIT_GRACE_SECONDS = 0.5
+    ObservabilityService._create_otlp_exporter = lambda self: None
+    service = ObservabilityService(
+        "test-input",
+        logging.getLogger("test"),
+        ta_name="test-ta",
+        ta_version="1.0.0",
+        extra_exporters=[SlowExporter()],
+    )
+    service.event_count_counter.add(1)
+    print("MAIN_DONE", flush=True)
+    return 0
+
+
+def fork_worker():
+    class Provider:
+        def force_flush(self, **kwargs):
+            pass
+
+        def shutdown(self, **kwargs):
+            pass
+
+    controller = flushmod._register_controller(
+        Provider(), logging.getLogger("test"), str
+    )
+    parent_worker = controller._worker
+    pid = os.fork()
+    if pid == 0:
+        fresh = (
+            controller._worker is not parent_worker and controller._worker.is_alive()
+        )
+        print("CHILD_FRESH_WORKER=" + str(fresh), flush=True)
+        return 0 if fresh else 2
+    _, status = os.waitpid(pid, 0)
+    print("PARENT_WORKER_ALIVE=" + str(parent_worker.is_alive()), flush=True)
+    return os.waitstatus_to_exitcode(status)
+
+
+if __name__ == "__main__":
+    scenarios = {
+        "blocked-otel-exit": blocked_otel_exit,
+        "fork-worker": fork_worker,
+    }
+    sys.exit(scenarios[sys.argv[1]]())
diff --git a/tests/unit/test_observability.py b/tests/unit/test_observability.py
index 378a4f62..fd53a367 100644
--- a/tests/unit/test_observability.py
+++ b/tests/unit/test_observability.py
@@ -633,6 +633,84 @@ def test_meter_is_available(self, logger, monkeypatch):
         # Assert
         assert svc._meter is not None
 
+    @pytest.mark.parametrize("failure_stage", ["counter", "registration"])
+    @pytest.mark.parametrize("cleanup_fails", [False, True])
+    def test_failed_initialisation_leaves_service_disabled(
+        self, logger, monkeypatch, failure_stage, cleanup_fails
+    ):
+        # Arrange
+        provider = MagicMock()
+        callback = MagicMock()
+        monkeypatch.setattr(
+            "solnlib.observability.MeterProvider", MagicMock(return_value=provider)
+        )
+        register = MagicMock()
+        monkeypatch.setattr("solnlib.observability._register_controller", register)
+        error = RuntimeError("initialisation failure")
+        if failure_stage == "counter":
+            provider.get_meter.return_value.create_counter.side_effect = [
+                MagicMock(),
+                error,
+            ]
+        else:
+            register.side_effect = error
+        if cleanup_fails:
+            provider.shutdown.side_effect = RuntimeError("cleanup failure")
+
+        # Act
+        svc = _make_service(logger, monkeypatch)
+        instrument = svc.register_instrument(callback)
+        flush_result = svc.flush()
+
+        # Assert
+        for attribute in (
+            "_provider",
+            "_meter",
+            "_flush_controller",
+            "event_count_counter",
+            "event_bytes_counter",
+        ):
+            assert getattr(svc, attribute) is None
+        provider.shutdown.assert_called_once_with()
+        assert instrument is None
+        assert flush_result is None
+        callback.assert_not_called()
+        register.return_value.request_flush.assert_not_called()
+        assert any(
+            "initialisation failure" in call.args for call in logger.info.call_args_list
+        )
+
+    def test_closed_registry_does_not_raise_from_public_service(
+        self, logger, monkeypatch
+    ):
+        # Arrange
+        from solnlib import _observability_flush as flushmod
+
+        registry = flushmod._FlushRegistry()
+        registry.shutdown_all()
+        monkeypatch.setattr(flushmod, "_registry", registry)
+        callback = MagicMock()
+
+        # Act
+        svc = _make_service(logger, monkeypatch)
+        instrument = svc.register_instrument(callback)
+        flush_result = svc.flush()
+
+        # Assert
+        assert svc._provider is None
+        assert svc._meter is None
+        assert svc._flush_controller is None
+        assert svc.event_count_counter is None
+        assert svc.event_bytes_counter is None
+        assert instrument is None
+        assert flush_result is None
+        callback.assert_not_called()
+        assert registry._controllers == []
+        assert any(
+            "Metrics shutdown has started" in call.args
+            for call in logger.info.call_args_list
+        )
+
     def test_extra_exporter_is_added(self, logger, monkeypatch):
         # Arrange
         extra = MagicMock()
@@ -905,7 +983,7 @@ def test_create_otlp_exporter_wrapper_forwards_shutdown(
         result.shutdown(timeout_millis=1234)
         mock_exporter.shutdown.assert_called_once_with(timeout_millis=1234)
 
-    def test_create_otlp_exporter_uses_delta_temporality(
+    def test_create_otlp_exporter_uses_delta_temporality_and_bounded_timeout(
         self, logger, monkeypatch, tmp_path
     ):
         # Arrange
@@ -940,6 +1018,7 @@ def test_create_otlp_exporter_uses_delta_temporality(
         ObservabilityService._create_otlp_exporter(svc)
         # Assert — preferred_temporality must set DELTA for Counter and Histogram
         _, kwargs = mock_exporter_cls.call_args
+        assert kwargs["timeout"] == 2.0
         temporality = kwargs["preferred_temporality"]
         assert temporality[Counter] == AggregationTemporality.DELTA
         assert temporality[Histogram] == AggregationTemporality.DELTA
@@ -947,6 +1026,7 @@ def test_create_otlp_exporter_uses_delta_temporality(
     def test_flush_schedules_without_calling_provider_on_caller(
         self, logger, monkeypatch
     ):
+        # Arrange
         controller = MagicMock()
         provider_cls = MagicMock()
         monkeypatch.setattr("solnlib.observability.MeterProvider", provider_cls)
@@ -962,8 +1042,10 @@ def test_flush_schedules_without_calling_provider_on_caller(
             "test-input", logger, ta_name="my_ta", ta_version="1.0.0"
         )
 
+        # Act
         svc.flush(timeout_millis=5_000)
 
+        # Assert
         controller.request_flush.assert_called_once_with(5_000)
         provider_cls.return_value.force_flush.assert_not_called()
         assert provider_cls.call_args.kwargs["shutdown_on_exit"] is False
@@ -976,12 +1058,15 @@ def test_flush_is_noop_when_provider_missing(self, logger, monkeypatch):
         svc.flush()
 
     def test_flush_logs_info_when_scheduling_fails(self, logger, monkeypatch):
+        # Arrange
         svc = _make_service(logger, monkeypatch)
         svc._flush_controller = MagicMock()
         svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
 
+        # Act
         svc.flush()
 
+        # Assert
         assert any("boom" in call.args for call in logger.info.call_args_list)
         logger.warning.assert_not_called()
 
@@ -1567,6 +1652,7 @@ def test_create_otlp_exporter_exception_uses_safe_str_no_exc_info(
         logger.warning.assert_not_called()
 
     def test_flush_scheduling_error_uses_safe_str(self, logger, monkeypatch):
+        # Arrange
         monkeypatch.setattr(
             "solnlib.observability._safe_exception_str", lambda error: "SAFE"
         )
@@ -1575,8 +1661,10 @@ def test_flush_scheduling_error_uses_safe_str(self, logger, monkeypatch):
         svc._flush_controller.request_flush.side_effect = RuntimeError("boom")
         logger.reset_mock()
 
+        # Act
         svc.flush()
 
+        # Assert
         assert any("SAFE" in call.args for call in logger.info.call_args_list)
         logger.warning.assert_not_called()
 
diff --git a/tests/unit/test_observability_flush.py b/tests/unit/test_observability_flush.py
index 5e75e209..5b490395 100644
--- a/tests/unit/test_observability_flush.py
+++ b/tests/unit/test_observability_flush.py
@@ -15,44 +15,86 @@
 #
 
 import logging
-import os
-import subprocess
-import sys
 import threading
 import time
-from pathlib import Path
 from unittest.mock import MagicMock
 
 import pytest
+from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
 
 from solnlib import _observability_flush as flushmod
 from solnlib._observability_flush import _AsyncFlushController
+from solnlib.observability import ObservabilityService
+
+
+@pytest.fixture
+def controller_factory():
+    controllers = []
+
+    def create(provider, logger=None):
+        controller = _AsyncFlushController(
+            provider, logger or MagicMock(spec=logging.Logger), str
+        )
+        controllers.append(controller)
+        return controller
+
+    yield create
+
+    deadline = time.monotonic() + 1
+    for controller in controllers:
+        controller.request_shutdown(deadline)
+    for controller in controllers:
+        controller.join_until(deadline)
+
+
+@pytest.fixture
+def registry_and_callbacks(monkeypatch):
+    registry = flushmod._FlushRegistry()
+    callbacks = []
+
+    def register_callback(callback):
+        callbacks.append(callback)
+        return callback
+
+    monkeypatch.setattr(flushmod, "_registry", registry)
+    monkeypatch.setattr(flushmod.atexit, "register", register_callback)
+    if hasattr(flushmod.os, "register_at_fork"):
+        monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock())
+
+    yield registry, callbacks
+
+    registry.shutdown_all()
 
 
-def test_request_flush_returns_while_export_is_blocked():
+def test_request_flush_returns_while_export_is_blocked(controller_factory):
+    # Arrange
     entered = threading.Event()
     release = threading.Event()
     provider = MagicMock()
 
     def blocked_flush(**kwargs):
         entered.set()
-        assert release.wait(2)
+        release.wait()
 
     provider.force_flush.side_effect = blocked_flush
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    controller = controller_factory(provider)
     try:
+        # Act
         start = time.monotonic()
         controller.request_flush(30_000)
-        assert time.monotonic() - start < 0.2
-        assert entered.wait(1)
+        elapsed = time.monotonic() - start
+        export_started = entered.wait(1)
+
+        # Assert
+        assert elapsed < 0.2
+        assert export_started
         provider.force_flush.assert_called_once_with(timeout_millis=30_000)
     finally:
         release.set()
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
 
 
-def test_repeated_requests_during_export_coalesce_to_one_followup():
+def test_repeated_requests_during_export_coalesce_to_one_followup(controller_factory):
+    # Arrange
     entered = threading.Event()
     release = threading.Event()
     second_done = threading.Event()
@@ -61,78 +103,90 @@ def test_repeated_requests_during_export_coalesce_to_one_followup():
     def flush(**kwargs):
         if provider.force_flush.call_count == 1:
             entered.set()
-            assert release.wait(2)
+            release.wait()
         else:
             second_done.set()
 
     provider.force_flush.side_effect = flush
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    controller = controller_factory(provider)
     try:
+        # Act
         controller.request_flush(30_000)
-        assert entered.wait(1)
+        first_started = entered.wait(1)
         controller.request_flush(10_000)
         controller.request_flush(5_000)
         release.set()
-        assert second_done.wait(1)
+        followup_finished = second_done.wait(1)
+
+        # Assert
+        assert first_started
+        assert followup_finished
         assert provider.force_flush.call_count == 2
         assert provider.force_flush.call_args_list[-1].kwargs == {
             "timeout_millis": 5_000
         }
     finally:
         release.set()
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
 
 
-def test_worker_logs_export_error_without_raising_to_caller():
+def test_worker_logs_export_error_without_raising_to_caller(controller_factory):
+    # Arrange
     logged = threading.Event()
     provider = MagicMock()
     provider.force_flush.side_effect = RuntimeError("boom")
     logger = MagicMock(spec=logging.Logger)
     logger.info.side_effect = lambda *args: logged.set()
-    controller = _AsyncFlushController(provider, logger, str)
-    try:
-        controller.request_flush(30_000)
-        assert logged.wait(1)
-        assert "boom" in logger.info.call_args.args[-1]
-    finally:
-        controller.request_shutdown(time.monotonic() + 1)
-        controller.join_until(time.monotonic() + 1)
+    controller = controller_factory(provider, logger)
+
+    # Act
+    controller.request_flush(30_000)
+    error_logged = logged.wait(1)
+
+    # Assert
+    assert error_logged
+    assert "boom" in logger.info.call_args.args[-1]
 
 
-def test_shutdown_is_once_and_later_flush_is_ignored():
+def test_shutdown_is_once_and_later_flush_is_ignored(controller_factory):
+    # Arrange
     provider = MagicMock()
-    controller = _AsyncFlushController(provider, MagicMock(spec=logging.Logger), str)
+    controller = controller_factory(provider)
     deadline = time.monotonic() + 1
+
+    # Act
     controller.request_shutdown(deadline)
     controller.request_shutdown(deadline)
     controller.join_until(deadline)
     controller.request_flush(30_000)
+
+    # Assert
     provider.shutdown.assert_called_once()
     provider.force_flush.assert_not_called()
+    assert not controller._worker.is_alive()
 
 
-def test_exit_budget_is_shared_by_two_controllers(monkeypatch):
+def test_exit_budget_is_shared_by_two_controllers(monkeypatch, registry_and_callbacks):
+    # Arrange
     monkeypatch.setattr(flushmod, "_EXIT_GRACE_SECONDS", 0.05)
+    registry, _ = registry_and_callbacks
     release = threading.Event()
     entered = [threading.Event(), threading.Event()]
     controllers = []
-    registry = flushmod._FlushRegistry()
     for gate in entered:
         provider = MagicMock()
         provider.shutdown.side_effect = lambda gate=gate, **kwargs: (
             gate.set(),
-            release.wait(2),
+            release.wait(),
         )
-        controller = _AsyncFlushController(
-            provider, MagicMock(spec=logging.Logger), str
-        )
-        registry._controllers.append(controller)
-        controllers.append(controller)
+        controllers.append(registry.register(provider, MagicMock(), str))
     try:
+        # Act
         start = time.monotonic()
         registry.shutdown_all()
-        assert time.monotonic() - start < 0.2
+        elapsed = time.monotonic() - start
+
+        # Assert
+        assert elapsed < 0.2
         assert all(gate.wait(1) for gate in entered)
     finally:
         release.set()
@@ -140,179 +194,151 @@ def test_exit_budget_is_shared_by_two_controllers(monkeypatch):
             controller.join_until(time.monotonic() + 1)
 
 
-def test_one_atexit_registration_for_multiple_controllers(monkeypatch):
-    register = MagicMock()
-    monkeypatch.setattr(flushmod.atexit, "register", register)
-    if hasattr(flushmod.os, "register_at_fork"):
-        monkeypatch.setattr(flushmod.os, "register_at_fork", MagicMock())
-    registry = flushmod._FlushRegistry()
-    registry.add(MagicMock())
-    registry.add(MagicMock())
-    register.assert_called_once_with(flushmod._shutdown_all)
-
-
-def _run_child(script):
-    repo_root = Path(__file__).resolve().parents[2]
-    env = dict(os.environ, PYTHONPATH=str(repo_root))
-    return subprocess.run(
-        [sys.executable, "-c", script],
-        capture_output=True,
-        text=True,
-        env=env,
-        timeout=3,
-    )
+def test_one_atexit_registration_for_multiple_controllers(registry_and_callbacks):
+    # Arrange
+    registry, callbacks = registry_and_callbacks
 
+    # Act
+    registry.register(MagicMock(), MagicMock(), str)
+    registry.register(MagicMock(), MagicMock(), str)
 
-def test_normal_exit_attempts_final_shutdown_without_explicit_flush():
-    result = _run_child(
-        """
-import logging
-from solnlib._observability_flush import _register_controller
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs): print('FINAL_SHUTDOWN', flush=True)
-_register_controller(Provider(), logging.getLogger('test'), str)
-"""
-    )
-    assert result.returncode == 0, result.stderr
-    assert "FINAL_SHUTDOWN" in result.stdout
+    # Assert
+    assert callbacks == [flushmod._shutdown_all]
 
 
-def test_blocked_final_export_uses_shortened_grace_period():
-    start = time.monotonic()
-    result = _run_child(
-        """
-import logging
-import threading
-import solnlib._observability_flush as flushmod
-flushmod._EXIT_GRACE_SECONDS = 0.05
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs):
-        print('SHUTDOWN_ENTERED', flush=True)
-        threading.Event().wait(10)
-flushmod._register_controller(Provider(), logging.getLogger('test'), str)
-"""
-    )
-    assert result.returncode == 0, result.stderr
-    assert "SHUTDOWN_ENTERED" in result.stdout
-    assert time.monotonic() - start < 1.5
+def test_closed_registry_does_not_start_a_worker(monkeypatch, registry_and_callbacks):
+    # Arrange
+    registry, _ = registry_and_callbacks
+    registry.shutdown_all()
+    controller_cls = MagicMock()
+    monkeypatch.setattr(flushmod, "_AsyncFlushController", controller_cls)
 
+    # Act
+    with pytest.raises(RuntimeError, match="shutdown has started"):
+        flushmod._register_controller(MagicMock(), MagicMock(), str)
 
-def test_real_otel_reader_does_not_restore_blocking_exit():
-    start = time.monotonic()
-    result = _run_child(
-        """
-import logging
-import threading
-from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
-import solnlib._observability_flush as flushmod
-from solnlib.observability import ObservabilityService
+    # Assert
+    controller_cls.assert_not_called()
+    assert registry._controllers == []
 
-flushmod._EXIT_GRACE_SECONDS = 0.5
-ObservabilityService._create_otlp_exporter = lambda self: None
-
-class SlowExporter(MetricExporter):
-    def export(self, metrics_data, timeout_millis=10_000, **kwargs):
-        print('OTEL_EXPORT_ENTERED', flush=True)
-        threading.Event().wait(10)
-        return MetricExportResult.SUCCESS
-    def shutdown(self, timeout_millis=30_000, **kwargs): pass
-    def force_flush(self, timeout_millis=10_000): return True
-
-service = ObservabilityService(
-    'test-input', logging.getLogger('test'), ta_name='test-ta',
-    ta_version='1.0.0', extra_exporters=[SlowExporter()],
-)
-service.event_count_counter.add(1)
-print('MAIN_DONE', flush=True)
-"""
-    )
-    assert result.returncode == 0, result.stderr
-    assert "MAIN_DONE" in result.stdout
-    assert "OTEL_EXPORT_ENTERED" in result.stdout
-    assert time.monotonic() - start < 2.0
 
+def test_closed_registry_does_not_restart_workers_after_fork(registry_and_callbacks):
+    # Arrange
+    registry, _ = registry_and_callbacks
+    controller = registry.register(MagicMock(), MagicMock(), str)
+    registry.shutdown_all()
+    original_worker = controller._worker
+    try:
+        # Act
+        registry._after_fork_child()
 
-def test_shutdown_exports_pending_flush_without_periodic_reader():
-    result = _run_child(
-        """
-import logging
-import os
-import threading
-import time
-from opentelemetry.sdk.metrics.export import MetricExporter, MetricExportResult
-from solnlib.observability import ObservabilityService
+        # Assert
+        assert controller._worker is original_worker
+        assert not controller._worker.is_alive()
+    finally:
+        deadline = time.monotonic() + 1
+        controller.request_shutdown(deadline)
+        controller.join_until(deadline)
+
+
+def test_exit_handler_cannot_register_a_worker_after_registry_shutdown(
+    registry_and_callbacks,
+):
+    # Arrange
+    registry, callbacks = registry_and_callbacks
+    registration_errors = []
+
+    def late_handler():
+        try:
+            flushmod._register_controller(MagicMock(), MagicMock(), str)
+        except RuntimeError as error:
+            registration_errors.append(error)
+
+    callbacks.append(late_handler)
+    controller = flushmod._register_controller(MagicMock(), MagicMock(), str)
+
+    # Act
+    for callback in reversed(callbacks):
+        callback()
+
+    # Assert
+    assert len(registration_errors) == 1
+    assert registry._controllers == [controller]
+    assert not controller._worker.is_alive()
+
+
+def test_registered_exit_callback_attempts_shutdown_without_explicit_flush(
+    registry_and_callbacks,
+):
+    # Arrange
+    _, callbacks = registry_and_callbacks
+    provider = MagicMock()
+    controller = flushmod._register_controller(provider, MagicMock(), str)
 
-os.environ['OTEL_METRIC_EXPORT_INTERVAL'] = 'inf'
-ObservabilityService._create_otlp_exporter = lambda self: None
-entered = threading.Event()
-release = threading.Event()
-exported_values = []
-
-class RecordingExporter(MetricExporter):
-    def export(self, metrics_data, timeout_millis=10_000, **kwargs):
-        value = sum(
-            point.value
-            for resource in metrics_data.resource_metrics
-            for scope in resource.scope_metrics
-            for metric in scope.metrics
-            if metric.name == 'splunk.addon.events'
-            for point in metric.data.data_points
-        )
-        exported_values.append(value)
-        if len(exported_values) == 1:
-            entered.set()
-            assert release.wait(1)
-        return MetricExportResult.SUCCESS
-
-    def shutdown(self, timeout_millis=30_000, **kwargs): pass
-    def force_flush(self, timeout_millis=10_000): return True
-
-service = ObservabilityService(
-    'test-input', logging.getLogger('test'), ta_name='test-ta',
-    ta_version='1.0.0', extra_exporters=[RecordingExporter()],
-)
-service.event_count_counter.add(1)
-service.flush()
-assert entered.wait(1)
-service.event_count_counter.add(2)
-service.flush()
-deadline = time.monotonic() + 1
-service._flush_controller.request_shutdown(deadline)
-release.set()
-service._flush_controller.join_until(deadline)
-print('EXPORTED_VALUES=' + repr(exported_values), flush=True)
-assert exported_values == [1, 3]
-"""
-    )
-    assert result.returncode == 0, result.stderr + result.stdout
-    assert "EXPORTED_VALUES=[1, 3]" in result.stdout
+    # Act
+    callbacks[0]()
 
+    # Assert
+    provider.shutdown.assert_called_once()
+    assert not controller._worker.is_alive()
 
-@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
-def test_child_gets_fresh_worker_after_fork():
-    result = _run_child(
-        """
-import logging
-import os
-import sys
-from solnlib._observability_flush import _register_controller
-class Provider:
-    def force_flush(self, **kwargs): pass
-    def shutdown(self, **kwargs): pass
-controller = _register_controller(Provider(), logging.getLogger('test'), str)
-parent_worker = controller._worker
-pid = os.fork()
-if pid == 0:
-    fresh = controller._worker is not parent_worker and controller._worker.is_alive()
-    print('CHILD_FRESH_WORKER=' + str(fresh), flush=True)
-    sys.exit(0 if fresh else 2)
-_, status = os.waitpid(pid, 0)
-print('PARENT_WORKER_ALIVE=' + str(parent_worker.is_alive()), flush=True)
-sys.exit(os.waitstatus_to_exitcode(status))
-"""
+
+def test_shutdown_exports_pending_flush_without_periodic_reader(
+    monkeypatch, registry_and_callbacks
+):
+    # Arrange
+    monkeypatch.setenv("OTEL_METRIC_EXPORT_INTERVAL", "inf")
+    monkeypatch.setattr(
+        ObservabilityService, "_create_otlp_exporter", lambda self: None
     )
-    assert result.returncode == 0, result.stderr
-    assert "CHILD_FRESH_WORKER=True" in result.stdout
-    assert "PARENT_WORKER_ALIVE=True" in result.stdout
+    entered = threading.Event()
+    release = threading.Event()
+    exported_values = []
+
+    class RecordingExporter(MetricExporter):
+        def export(self, metrics_data, timeout_millis=10_000, **kwargs):
+            value = sum(
+                point.value
+                for resource in metrics_data.resource_metrics
+                for scope in resource.scope_metrics
+                for metric in scope.metrics
+                if metric.name == "splunk.addon.events"
+                for point in metric.data.data_points
+            )
+            exported_values.append(value)
+            if len(exported_values) == 1:
+                entered.set()
+                release.wait()
+            return MetricExportResult.SUCCESS
+
+        def shutdown(self, timeout_millis=30_000, **kwargs):
+            pass
+
+        def force_flush(self, timeout_millis=10_000):
+            return True
+
+    service = ObservabilityService(
+        "test-input",
+        MagicMock(spec=logging.Logger),
+        ta_name="test-ta",
+        ta_version="1.0.0",
+        extra_exporters=[RecordingExporter()],
+    )
+    try:
+        # Act
+        service.event_count_counter.add(1)
+        service.flush()
+        first_export_started = entered.wait(1)
+        service.event_count_counter.add(2)
+        service.flush()
+        deadline = time.monotonic() + 1
+        service._flush_controller.request_shutdown(deadline)
+        release.set()
+        service._flush_controller.join_until(deadline)
+
+        # Assert
+        assert first_export_started
+        assert exported_values == [1, 3]
+        assert not service._flush_controller._worker.is_alive()
+    finally:
+        release.set()
diff --git a/tests/unit/test_observability_lifecycle.py b/tests/unit/test_observability_lifecycle.py
new file mode 100644
index 00000000..02b08220
--- /dev/null
+++ b/tests/unit/test_observability_lifecycle.py
@@ -0,0 +1,72 @@
+#
+# Copyright 2026 Splunk Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""Process-level regressions that require interpreter shutdown or a real fork."""
+
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_SCENARIO_SCRIPT = (
+    Path(__file__).with_name("helpers") / "observability_flush_process.py"
+)
+
+
+def test_interpreter_exit_is_bounded_with_a_blocked_otel_export():
+    # Arrange
+    env = dict(os.environ, PYTHONPATH=str(_REPO_ROOT))
+    start = time.monotonic()
+
+    # Act
+    result = subprocess.run(
+        [sys.executable, str(_SCENARIO_SCRIPT), "blocked-otel-exit"],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=5,
+    )
+    elapsed = time.monotonic() - start
+
+    # Assert
+    assert result.returncode == 0, result.stderr
+    assert "MAIN_DONE" in result.stdout
+    assert "OTEL_EXPORT_ENTERED" in result.stdout
+    assert elapsed < 3
+
+
+@pytest.mark.skipif(not hasattr(os, "fork"), reason="fork is unavailable")
+def test_child_gets_fresh_worker_after_fork():
+    # Arrange
+    env = dict(os.environ, PYTHONPATH=str(_REPO_ROOT))
+
+    # Act
+    result = subprocess.run(
+        [sys.executable, str(_SCENARIO_SCRIPT), "fork-worker"],
+        capture_output=True,
+        text=True,
+        env=env,
+        timeout=5,
+    )
+
+    # Assert
+    assert result.returncode == 0, result.stderr
+    assert "CHILD_FRESH_WORKER=True" in result.stdout
+    assert "PARENT_WORKER_ALIVE=True" in result.stdout

From 68e2ad34b381c25c9c15aa81c0583a5791081e17 Mon Sep 17 00:00:00 2001
From: srv-rr-github-token
 <94607705+srv-rr-github-token@users.noreply.github.com>
Date: Wed, 21 Oct 2026 12:56:21 +0000
Subject: [PATCH 11/11] chore(release): 8.2.0-beta.1

# [8.2.0-beta.1](https://github.com/splunk/addonfactory-solutions-library-python/compare/v8.1.1...v8.2.0-beta.1) (2026-10-21)

### Bug Fixes

* ADDON-91243 change telemetry flush to async ([#467](https://github.com/splunk/addonfactory-solutions-library-python/issues/467)) ([3e538df](https://github.com/splunk/addonfactory-solutions-library-python/commit/3e538dfc173bc97c3d27a5a8caf469bd85841d8b))
* drain pending metric flush before shutdown ([4555232](https://github.com/splunk/addonfactory-solutions-library-python/commit/4555232d53b999195916fbc93431d37d2f1bb4bc))
* harden async observability lifecycle ([9b3a2a5](https://github.com/splunk/addonfactory-solutions-library-python/commit/9b3a2a59460a3aa11b4033b879746b7d47078c69))

### Features

* add async metric flush worker ([5326914](https://github.com/splunk/addonfactory-solutions-library-python/commit/53269141d13580754503d7dc9ecd6143f8641f18))
* bound metric shutdown across services ([280fba9](https://github.com/splunk/addonfactory-solutions-library-python/commit/280fba96fe1f0f14df256b4840fc52a16283c581))
* make solnlib observability flush nonblocking ([c6897eb](https://github.com/splunk/addonfactory-solutions-library-python/commit/c6897eb9b5e6274f38232e7e696850842fe56388))
---
 pyproject.toml      | 2 +-
 solnlib/__init__.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index cd772ac5..d14031e8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,7 +16,7 @@
 
 [tool.poetry]
 name = "solnlib"
-version = "8.1.1"
+version = "8.2.0-beta.1"
 description = "The Splunk Software Development Kit for Splunk Solutions"
 authors = ["Splunk <addonfactory@splunk.com>"]
 license = "Apache-2.0"
diff --git a/solnlib/__init__.py b/solnlib/__init__.py
index 0fe61d4e..e18103d3 100644
--- a/solnlib/__init__.py
+++ b/solnlib/__init__.py
@@ -55,4 +55,4 @@
     "utils",
 ]
 
-__version__ = "8.1.1"
+__version__ = "8.2.0-beta.1"