Merge branch 'claude/actor-context-hook'

Author: codeofaxel

kiln/src/kiln/events.py

@@ -36,6 +36,7 @@ async def on_print_done(event: Event) -> None:
 from __future__ import annotations
 
 import asyncio
+import contextvars
 import enum
 import logging
 import threading
@@ -49,6 +50,32 @@ async def on_print_done(event: Event) -> None:
 logger = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# Ambient actor context
+# ---------------------------------------------------------------------------
+#
+# Generic "who triggered this event" envelope.  An orchestrator (typically a
+# REST dispatcher, but anything that wraps a unit of work) ``set``s this
+# before invoking code that may publish events.  :class:`EventBus` attaches
+# its current value to every :class:`Event` whose ``actor`` field is not
+# already explicitly set, so downstream subscribers can attribute events to
+# their originating call.
+#
+# Deliberately opaque — Kiln itself has no schema for the dict.  Conventions
+# used by consumers (e.g. ``caller_tier``, ``caller_id``, ``request_id``) are
+# defined by the consumer, not the bus.
+#
+# Background-thread propagation: long-running publishers that spawn threads
+# (see :class:`kiln.plugins.monitoring_tools._PrintWatcher`) must capture
+# the current context via :func:`contextvars.copy_context` and run the
+# thread body inside it, otherwise the ContextVar's value is lost at the
+# thread boundary.  Synchronous publishers don't need to do anything —
+# the value is read at publish time on the same thread that set it.
+current_actor_context: contextvars.ContextVar[dict[str, Any] | None] = (
+    contextvars.ContextVar("kiln_actor_context", default=None)
+)
+
+
 class EventType(enum.Enum):
     """All event types emitted by the Kiln system."""
 
@@ -169,15 +196,25 @@ class Event:
     data: dict[str, Any] = field(default_factory=dict)
     timestamp: float = field(default_factory=time.time)
     source: str = ""  # e.g. "printer:voron-350" or "queue"
+    # Ambient actor context — who triggered this event (caller tier,
+    # tenant id, request correlation id, etc.).  Populated automatically
+    # by :class:`EventBus` from :data:`current_actor_context` at publish
+    # time when not explicitly set.  ``None`` means no actor metadata is
+    # attached; an empty dict means "actor explicitly empty" (e.g. a
+    # system-initiated event with no caller).
+    actor: dict[str, Any] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         """Return a JSON-serialisable dictionary."""
-        return {
+        out: dict[str, Any] = {
             "type": self.type.value,
             "data": self.data,
             "timestamp": self.timestamp,
             "source": self.source,
         }
+        if self.actor is not None:
+            out["actor"] = self.actor
+        return out
 
 
 # Type aliases for event handlers.
@@ -287,10 +324,24 @@ def _resolve_event(
         data: dict[str, Any] | None,
         source: str,
     ) -> Event:
-        """Normalise the flexible publish signature into an :class:`Event`."""
+        """Normalise the flexible publish signature into an :class:`Event`.
+
+        Attaches the ambient :data:`current_actor_context` to events that
+        don't carry an ``actor`` of their own.  Explicit
+        ``Event(actor=...)`` always wins over the ambient value.
+        """
         if isinstance(event_or_type, EventType):
-            return Event(type=event_or_type, data=data or {}, source=source)
-        return event_or_type
+            event = Event(type=event_or_type, data=data or {}, source=source)
+        else:
+            event = event_or_type
+        if event.actor is None:
+            ambient = current_actor_context.get()
+            if ambient is not None:
+                # Defensive copy — the ambient dict can be mutated by its
+                # owner after publish, and we want the event's actor to be
+                # a stable snapshot at publish time.
+                event.actor = dict(ambient)
+        return event
 
     def publish(
         self,

kiln/src/kiln/plugins/monitoring_tools.py

@@ -13,6 +13,7 @@
 
 from __future__ import annotations
 
+import contextvars
 import logging
 import os
 import secrets
@@ -126,10 +127,22 @@ def __init__(
     # -- public API --------------------------------------------------------
 
     def start(self) -> None:
-        """Start the background monitoring thread."""
+        """Start the background monitoring thread.
+
+        Captures the current Python context (incl. :data:`kiln.events.
+        current_actor_context`) via :func:`contextvars.copy_context` and
+        runs the watcher body inside it, so events published from the
+        background thread carry the caller's ambient actor metadata
+        (caller tier, tenant id, etc.) instead of falling off the
+        thread boundary.
+        """
         self._start_time = time.time()
+        # Snapshot the caller's context at start-time so the watcher
+        # thread sees the same ContextVar values the caller saw.  Python
+        # ContextVars do not auto-propagate across thread boundaries.
+        ctx = contextvars.copy_context()
         self._thread = threading.Thread(
-            target=self._run,
+            target=lambda: ctx.run(self._run),
             name=f"print-watcher-{self._watch_id}",
             daemon=True,
         )

kiln/tests/test_events.py

@@ -463,3 +463,171 @@ def publish_batch(event_type: EventType, count: int) -> None:
         started = [e for e in events if e.type == EventType.PRINT_STARTED]
         assert len(queued) == 50
         assert len(started) == 50
+
+
+# ---------------------------------------------------------------------------
+# Ambient actor context
+# ---------------------------------------------------------------------------
+
+
+class TestActorContext:
+    """Tests for ``current_actor_context`` and its propagation onto events.
+
+    The ContextVar is a generic "who triggered this event" envelope.  An
+    orchestrator (e.g. kiln-pro's REST dispatcher) sets it before invoking
+    code that publishes events; :class:`EventBus` attaches its current
+    value to every event whose ``actor`` is not already set.  This test
+    class pins the contract — bus-level behaviour only; cross-repo
+    integration is tested kiln-pro-side.
+    """
+
+    def test_actor_defaults_to_none_on_event(self):
+        """The dataclass default — un-stamped Events have actor=None."""
+        e = Event(type=EventType.JOB_QUEUED, data={"x": 1})
+        assert e.actor is None
+
+    def test_context_var_default_is_none(self):
+        from kiln.events import current_actor_context
+
+        assert current_actor_context.get() is None
+
+    def test_publish_without_context_leaves_actor_none(self):
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        # ContextVar default is None — sanity that no test before us leaked.
+        assert current_actor_context.get() is None
+        bus.publish(EventType.JOB_QUEUED, {"job_id": "j1"})
+        events = bus.recent_events()
+        assert len(events) == 1
+        assert events[0].actor is None
+
+    def test_publish_with_ambient_context_stamps_event(self):
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        token = current_actor_context.set({"caller_tier": "pro", "caller_id": "t-7"})
+        try:
+            bus.publish(EventType.JOB_QUEUED, {"job_id": "j2"})
+        finally:
+            current_actor_context.reset(token)
+
+        events = bus.recent_events()
+        assert len(events) == 1
+        assert events[0].actor == {"caller_tier": "pro", "caller_id": "t-7"}
+
+    def test_explicit_event_actor_wins_over_ambient(self):
+        """Pre-built Event with actor set is NOT overwritten by the
+        ambient ContextVar — explicit always beats ambient."""
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        explicit = Event(
+            type=EventType.JOB_QUEUED,
+            data={"job_id": "j3"},
+            actor={"caller_tier": "business"},
+        )
+        token = current_actor_context.set({"caller_tier": "free"})
+        try:
+            bus.publish(explicit)
+        finally:
+            current_actor_context.reset(token)
+
+        events = bus.recent_events()
+        assert len(events) == 1
+        assert events[0].actor == {"caller_tier": "business"}
+
+    def test_actor_snapshot_is_defensive_copy(self):
+        """Mutating the ambient dict AFTER publish must not mutate the
+        already-stamped event's actor."""
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        ambient = {"caller_tier": "pro"}
+        token = current_actor_context.set(ambient)
+        try:
+            bus.publish(EventType.JOB_QUEUED, {"job_id": "j4"})
+        finally:
+            current_actor_context.reset(token)
+
+        # Mutate the live dict — the event's actor must not reflect this.
+        ambient["caller_tier"] = "free"
+        ambient["leaked"] = True
+        events = bus.recent_events()
+        assert events[0].actor == {"caller_tier": "pro"}
+
+    def test_to_dict_includes_actor_when_set(self):
+        e = Event(
+            type=EventType.JOB_QUEUED,
+            data={"job_id": "j5"},
+            actor={"caller_tier": "pro"},
+        )
+        d = e.to_dict()
+        assert d["actor"] == {"caller_tier": "pro"}
+
+    def test_to_dict_omits_actor_when_none(self):
+        e = Event(type=EventType.JOB_QUEUED, data={"job_id": "j6"})
+        d = e.to_dict()
+        assert "actor" not in d
+
+    def test_context_propagates_to_background_thread_via_copy_context(self):
+        """The pattern :class:`_PrintWatcher.start` uses: capture
+        :func:`contextvars.copy_context` in the calling thread, run the
+        target inside ``ctx.run(...)`` from the background thread.  This
+        test pins that the value actually survives the thread boundary —
+        without ``copy_context``, the ContextVar would default to None
+        in the child thread and publish-time stamping would silently
+        drop the actor.
+        """
+        import contextvars
+
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        token = current_actor_context.set({"caller_tier": "pro", "caller_id": "t-9"})
+        try:
+            ctx = contextvars.copy_context()
+        finally:
+            current_actor_context.reset(token)
+
+        # ContextVar is back to default in the parent thread now.
+        assert current_actor_context.get() is None
+
+        def _publish_from_thread():
+            # Inside ctx.run, the ContextVar's value is the one snapshotted above.
+            bus.publish(EventType.JOB_QUEUED, {"job_id": "j7"})
+
+        t = threading.Thread(target=lambda: ctx.run(_publish_from_thread))
+        t.start()
+        t.join(timeout=2)
+
+        events = bus.recent_events()
+        assert len(events) == 1
+        assert events[0].actor == {"caller_tier": "pro", "caller_id": "t-9"}
+
+    def test_naked_thread_without_copy_context_loses_actor(self):
+        """Sanity check / regression guard: a thread that does NOT use
+        ``copy_context`` does NOT see the parent thread's ContextVar
+        value.  If this ever fails (e.g. Python changes ContextVar
+        semantics to auto-propagate), the ``copy_context`` ceremony in
+        _PrintWatcher.start becomes unnecessary and the comment there
+        can be simplified."""
+        from kiln.events import current_actor_context
+
+        bus = EventBus()
+        token = current_actor_context.set({"caller_tier": "pro"})
+        try:
+            def _publish_from_naked_thread():
+                # No ctx.run — ContextVar should be at its default here.
+                bus.publish(EventType.JOB_QUEUED, {"job_id": "j8"})
+
+            t = threading.Thread(target=_publish_from_naked_thread)
+            t.start()
+            t.join(timeout=2)
+        finally:
+            current_actor_context.reset(token)
+
+        events = bus.recent_events()
+        assert len(events) == 1
+        # Naked thread → ContextVar default (None) → event.actor stays None.
+        assert events[0].actor is None