Coverage for src/lumen/cortex/dispatch.py: 100%

1""" 

2Running write-time hooks, without letting them take the server down. 

3 

4Three rules hold here, and they are the reason this module is separate from 

5the registry: 

6 

71. **After the commit.** A hook only ever sees data that is durable. It cannot 

8 change or veto the write that triggered it, and it cannot hold the writing 

9 transaction open while a model thinks. 

102. **Inside the request, with a budget.** The hook is awaited, so a Flag a 

11 plugin writes is there by the time the client reads the response — but only 

12 until its ``timeout_ms`` expires, after which it is cancelled. 

133. **A failing hook is a logged failure, not a failed write.** The triggering 

14 write already succeeded; nothing a plugin does afterwards can undo it. 

15 

16Recursion is bounded by ``LUMEN_CORTEX_MAX_HOOK_DEPTH`` (default 1): with the 

17default, writes made *by* a hook do not trigger hooks again. 

18""" 

19 

20from __future__ import annotations 

21 

22import asyncio 

23import time 

24from contextvars import ContextVar 

25from typing import Any 

26from uuid import UUID 

27 

28import structlog 

29 

30from ..api.deps import ResolvedTenantSettings 

31from ..config import get_settings 

32from ..core.write_events import WriteEvent 

33from ..observability import record_hook_duration, record_hook_failure 

34from .context import build_context 

35from .contract import EVENT_RESOURCE_DELETED, EVENT_RESOURCE_WRITTEN 

36from .registry import RegisteredHook, get_plugin_registry 

37 

38logger = structlog.get_logger(__name__) 

39 

40_depth: ContextVar[int] = ContextVar("lumen_cortex_hook_depth", default=0) 

41 

42 

43async def dispatch_write_events( 

44 events: list[WriteEvent], 

45 tenant_settings: ResolvedTenantSettings | None = None, 

46) -> None: 

47 """Run every hook matching ``events``. Never raises. 

48 

49 ``tenant_settings`` may be omitted; it is then resolved lazily, and only 

50 if a hook actually matches — so a server without plugins pays one boolean 

51 check per write. 

52 """ 

53 registry = get_plugin_registry() 

54 if not events or not registry.has_hooks: 

55 return 

56 

57 settings = get_settings() 

58 if not settings.cortex.enabled: 

59 return 

60 if _depth.get() >= settings.cortex.max_hook_depth: 

61 return 

62 

63 matched: list[tuple[WriteEvent, RegisteredHook]] = [] 

64 for event in events: 

65 name = EVENT_RESOURCE_DELETED if event.operation == "delete" else EVENT_RESOURCE_WRITTEN 

66 matched.extend((event, hook) for hook in registry.hooks_for(name, event.resource_type)) 

67 if not matched: 

68 return 

69 

70 resolved = tenant_settings or await _resolve_settings(events[0].tenant_id) 

71 

72 token = _depth.set(_depth.get() + 1) 

73 try: 

74 for event, hook in matched: 

75 await _run_hook(event, hook, resolved) 

76 finally: 

77 _depth.reset(token) 

78 

79 

80def _describe(exc: BaseException) -> str: 

81 """A message that is never empty. 

82 

83 Several exceptions a plugin can raise stringify to nothing at all — 

84 ``httpx.ReadTimeout`` is the one that costs the most debugging time, since 

85 ``error=""`` in the log says only that *something* went wrong. Fall back to 

86 the repr, and to the class name if even that is empty. 

87 """ 

88 return str(exc) or repr(exc) or type(exc).__name__ 

89 

90 

91async def _resolve_settings(tenant_id: UUID) -> ResolvedTenantSettings: 

92 from ..api.deps import resolve_tenant_settings 

93 

94 return await resolve_tenant_settings(tenant_id) 

95 

96 

97async def _run_hook( 

98 event: WriteEvent, 

99 hook: RegisteredHook, 

100 tenant_settings: ResolvedTenantSettings, 

101) -> None: 

102 """Invoke one hook under its timeout, swallowing whatever it does wrong.""" 

103 event_name = EVENT_RESOURCE_DELETED if event.operation == "delete" else EVENT_RESOURCE_WRITTEN 

104 ctx = build_context(event.tenant_id, tenant_settings, hook.plugin_name, hook.plugin_version) 

105 payload: dict[str, Any] | None = event.resource 

106 started = time.perf_counter() 

107 

108 try: 

109 await asyncio.wait_for( 

110 hook.callback(ctx, payload), 

111 timeout=hook.spec.timeout_ms / 1000, 

112 ) 

113 except TimeoutError: 

114 record_hook_failure(hook.plugin_name, event_name, "timeout") 

115 await logger.awarning( 

116 "cortex.hook_timeout", 

117 plugin=hook.plugin_name, 

118 hook_event=event_name, 

119 resource_type=event.resource_type, 

120 timeout_ms=hook.spec.timeout_ms, 

121 ) 

122 except Exception as exc: 

123 record_hook_failure(hook.plugin_name, event_name, "error") 

124 await logger.aerror( 

125 "cortex.hook_failed", 

126 plugin=hook.plugin_name, 

127 hook_event=event_name, 

128 resource_type=event.resource_type, 

129 error=_describe(exc), 

130 error_type=type(exc).__name__, 

131 ) 

132 else: 

133 await logger.adebug( 

134 "cortex.hook_ran", 

135 plugin=hook.plugin_name, 

136 hook_event=event_name, 

137 resource_type=event.resource_type, 

138 ) 

139 finally: 

140 record_hook_duration(hook.plugin_name, event_name, (time.perf_counter() - started) * 1000)