Coverage for node / src / stigmem_node / plugins / context.py: 80%

1"""Capability-gated plugin context.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import Any 

7 

8from .errors import CapabilityError 

9 

10 

11@dataclass(frozen=True, slots=True) 

12class CoreApis: 

13 """Handles core chooses to expose to plugins. 

14 

15 PR 4-INF.1 keeps this deliberately small and optional; later PRs can replace 

16 these generic handles with narrower typed facades without changing the 

17 registry contract. 

18 """ 

19 

20 facts_reader: Any = None 

21 facts_writer: Any = None 

22 recall_reader: Any = None 

23 recall_writer: Any = None 

24 audit_emitter: Any = None 

25 audit_reader: Any = None 

26 federation_reader: Any = None 

27 federation_writer: Any = None 

28 identity_reader: Any = None 

29 tenant_reader: Any = None 

30 tenant_writer: Any = None 

31 config_reader: Any = None 

32 network_outbound: Any = None 

33 

34 

35class PluginContext: 

36 """Capability-restricted context passed to plugin handlers.""" 

37 

38 __slots__ = ("_capabilities", "_core_apis", "plugin_name", "plugin_version") 

39 

40 def __init__( 

41 self, 

42 *, 

43 plugin_name: str, 

44 capabilities: frozenset[str], 

45 plugin_version: str = "0.0.0", 

46 core_apis: CoreApis | None = None, 

47 ) -> None: 

48 self.plugin_name = plugin_name 

49 self.plugin_version = plugin_version 

50 self._capabilities = capabilities 

51 self._core_apis = core_apis or CoreApis() 

52 

53 @property 

54 def capabilities(self) -> frozenset[str]: 

55 return self._capabilities 

56 

57 def _require(self, capability: str, accessor: str, value: Any) -> Any: 

58 if capability not in self._capabilities: 

59 raise CapabilityError( 

60 f"plugin {self.plugin_name!r} cannot call {accessor}: " 

61 f"capability {capability!r} not declared" 

62 ) 

63 return value 

64 

65 def get_facts_reader(self) -> Any: 

66 return self._require("facts.read", "get_facts_reader", self._core_apis.facts_reader) 

67 

68 def get_facts_writer(self) -> Any: 

69 return self._require("facts.write", "get_facts_writer", self._core_apis.facts_writer) 

70 

71 def get_recall_reader(self) -> Any: 

72 return self._require("recall.read", "get_recall_reader", self._core_apis.recall_reader) 

73 

74 def get_recall_writer(self) -> Any: 

75 return self._require("recall.write", "get_recall_writer", self._core_apis.recall_writer) 

76 

77 def get_audit_emitter(self) -> Any: 

78 return self._require("audit.emit", "get_audit_emitter", self._core_apis.audit_emitter) 

79 

80 def get_audit_reader(self) -> Any: 

81 return self._require("audit.read", "get_audit_reader", self._core_apis.audit_reader) 

82 

83 def get_federation_reader(self) -> Any: 

84 return self._require( 

85 "federation.read", "get_federation_reader", self._core_apis.federation_reader 

86 ) 

87 

88 def get_federation_writer(self) -> Any: 

89 return self._require( 

90 "federation.write", "get_federation_writer", self._core_apis.federation_writer 

91 ) 

92 

93 def get_identity_reader(self) -> Any: 

94 return self._require( 

95 "identity.read", "get_identity_reader", self._core_apis.identity_reader 

96 ) 

97 

98 def get_tenant_reader(self) -> Any: 

99 return self._require("tenant.read", "get_tenant_reader", self._core_apis.tenant_reader) 

100 

101 def get_tenant_writer(self) -> Any: 

102 return self._require("tenant.write", "get_tenant_writer", self._core_apis.tenant_writer) 

103 

104 def get_config_reader(self) -> Any: 

105 return self._require("config.read", "get_config_reader", self._core_apis.config_reader) 

106 

107 def get_network_outbound(self) -> Any: 

108 return self._require( 

109 "network.outbound", "get_network_outbound", self._core_apis.network_outbound 

110 )