fix: honor telemetry consent changes in running processes

TELEMETRY.md

@@ -59,6 +59,12 @@ Resolution order: `DO_NOT_TRACK` → `LOOPGAIN_TELEMETRY` → your saved choice
 environments (`CI`, `GITHUB_ACTIONS`, …) are auto-detected and declined
 silently — no notice, no data.
 
+Running processes recheck this choice on telemetry activity and before flushing
+a queued batch. An effective opt-out discards pending events and session counts;
+opting in again does not restore that discarded activity. No restart is required.
+This cannot recall a request that has already started. The resolution order above
+still applies, including an explicit `LOOPGAIN_TELEMETRY` environment setting.
+
 ### What is collected (only when you opt in)
 
 - A **random instance id** — a fresh `uuid4` generated and stored locally. It

loopgain/funnel.py

@@ -283,23 +283,28 @@ def _resolve_mode(self) -> str:
             return _DISABLED
         return _UNDECIDED
 
-    # ----- Lazy load / one-time setup -----
+    # ----- Consent refresh / lazy setup -----
 
     def _ensure_loaded(self) -> str:
-        """Resolve mode and prepare state exactly once per process.
+        """Recheck consent and lazily prepare state when the mode changes.
 
-        Returns the resolved mode. In ``disabled`` mode nothing is read,
-        written, or shown. In ``undecided`` mode we ensure a state file with
+        Returns the resolved mode. In ``disabled`` mode no state is written
+        and no notice is shown. In ``undecided`` mode we ensure a state file with
         an instance id exists and show the one-time notice. In ``enabled``
         mode we additionally bump the session counter and arm the flush
         machinery.
         """
-        if self._loaded:
-            return self._mode  # type: ignore[return-value]
         with self._lock:
-            if self._loaded:
-                return self._mode  # type: ignore[return-value]
             mode = self._resolve_mode()
+            if self._loaded and mode == self._mode:
+                return mode
+            if self._loaded:
+                # Consent can change in another process (the telemetry CLI).
+                # Do not retain activity from before a decline for later opt-in.
+                self._queue.clear()
+                self._outcomes.clear()
+                self._adapter = None
+                self._session_started = False
             self._mode = mode
 
             if mode == _DISABLED:
@@ -349,6 +354,8 @@ def _base_event(self, event: str) -> dict[str, Any]:
 
     def _enqueue(self, event: dict[str, Any]) -> None:
         with self._lock:
+            if self._ensure_loaded() != _ENABLED:
+                return
             self._queue.append(event)
         self._ensure_thread()
 
@@ -384,6 +391,8 @@ def flush_now(self) -> bool:
         otherwise (including when there was nothing to send).
         """
         with self._lock:
+            if self._ensure_loaded() != _ENABLED:
+                return False
             if not self._queue:
                 return False
             pending = self._queue
@@ -418,7 +427,7 @@ def _on_exit(self) -> None:
                 pass
 
     def _emit_session_summary(self) -> None:
-        if self._mode != _ENABLED or not self._session_started:
+        if self._ensure_loaded() != _ENABLED or not self._session_started:
             return
         event = self._base_event("session")
         event["session_seq"] = int(self._state.get("session_count", 0))
@@ -430,7 +439,7 @@ def _emit_session_summary(self) -> None:
     # ----- Public hooks (called from core / adapters) -----
     #
     # Each hook resolves consent through ``_ensure_loaded()`` rather than
-    # reading ``self._mode`` directly. Mode is resolved lazily and stays
+    # reading ``self._mode`` directly. Consent is rechecked on activity; mode stays
     # ``None`` until the first hook runs, so a bare comparison drops the record
     # whenever a hook happens to be the first one called. See GH #1.
 

tests/test_funnel.py

@@ -462,3 +462,91 @@ def test_cli_version_and_no_command(tmp_path, monkeypatch, capsys):
     assert __version__ in capsys.readouterr().out
     # No subcommand prints help and returns non-zero.
     assert cli.main([]) == 1
+
+
+@pytest.mark.parametrize("next_activity", ["flush", "observe", "outcome", "adapter", "exit"])
+def test_running_process_honors_external_opt_out(tmp_path, monkeypatch, next_activity):
+    """A separate CLI-like instance revokes consent; no queued/new data leaves."""
+    monkeypatch.delenv("LOOPGAIN_TELEMETRY", raising=False)
+    monkeypatch.delenv("DO_NOT_TRACK", raising=False)
+    cap = _Capture()
+    cli = _funnel(tmp_path, cap)
+    cli.set_consent(True)
+    running = _funnel(tmp_path, cap)
+    running.on_init()
+    running.note_adapter("before-decline")
+    running.note_outcome("TARGET_MET")
+    assert running._queue
+
+    cli.set_consent(False)
+    denied_state = (tmp_path / "funnel.json").read_bytes()
+    if next_activity == "observe":
+        running.on_first_observe()
+    elif next_activity == "outcome":
+        running.note_outcome("DIVERGING")
+    elif next_activity == "adapter":
+        running.note_adapter("after-decline")
+    elif next_activity == "exit":
+        running._on_exit()
+    else:
+        running.flush_now()
+    assert running.flush_now() is False
+    assert cap.batches == []
+    assert running._queue == []
+    assert running._outcomes == {}
+    assert running._adapter is None
+    assert (tmp_path / "funnel.json").read_bytes() == denied_state
+
+
+def test_runtime_environment_opt_out_discards_pending_batch(tmp_path, monkeypatch):
+    cap = _Capture()
+    running = _enabled(tmp_path, monkeypatch, cap)
+    running.on_init()
+    monkeypatch.setenv("DO_NOT_TRACK", "1")
+    assert running.flush_now() is False
+    running._on_exit()
+    assert cap.batches == []
+
+
+def test_reenable_never_sends_activity_discarded_on_decline(tmp_path, monkeypatch):
+    monkeypatch.delenv("LOOPGAIN_TELEMETRY", raising=False)
+    monkeypatch.delenv("DO_NOT_TRACK", raising=False)
+    cap = _Capture()
+    cli = _funnel(tmp_path, cap)
+    cli.set_consent(True)
+    running = _funnel(tmp_path, cap)
+    running.on_init()
+    running.note_adapter("old-adapter")
+    running.note_outcome("DIVERGING")
+    cli.set_consent(False)
+    running.flush_now()
+    cli.set_consent(True)
+    running.on_first_observe()
+    running._on_exit()
+    assert all(event["event"] != "first_init" for event in cap.events)
+    session = next(event for event in cap.events if event["event"] == "session")
+    assert session["adapter"] is None
+    assert "outcomes" not in session
+
+
+def test_empty_flush_discards_session_activity_on_external_opt_out(tmp_path, monkeypatch):
+    monkeypatch.delenv("LOOPGAIN_TELEMETRY", raising=False)
+    monkeypatch.delenv("DO_NOT_TRACK", raising=False)
+    cap = _Capture()
+    cli = _funnel(tmp_path, cap)
+    cli.set_consent(True)
+    running = _funnel(tmp_path, cap)
+    running.on_init()
+    assert running.flush_now() is True
+    running.note_adapter("before-decline")
+    running.note_outcome("DIVERGING")
+    assert running._queue == []
+
+    cli.set_consent(False)
+    assert running.flush_now() is False
+    cli.set_consent(True)
+    running._on_exit()
+
+    session = next(event for event in cap.events if event["event"] == "session")
+    assert session["adapter"] is None
+    assert "outcomes" not in session