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

1""" 

2Marking what a plugin wrote, and why. 

3 

4A resource produced inside the server by a model looks exactly like one a 

5clinician posted — unless it says so. Every write through the Cortex write port 

6is therefore tagged with the plugin that made it, and a write that carries an 

7:class:`AiAttribution` additionally records the model, the confidence and a 

8``Provenance`` resource pointing back at the target. 

9 

10The vocabulary here is deliberately generic: the core knows about *a plugin 

11wrote this with a model at this confidence*, and nothing about what the model 

12was reasoning over. 

13""" 

14 

15from __future__ import annotations 

16 

17from dataclasses import dataclass 

18from datetime import UTC, datetime 

19from typing import Any 

20 

21BASE = "https://lumen.health/fhir" 

22 

23#: ``meta.tag`` on everything written through the Cortex write port. 

24TAG_SYSTEM = f"{BASE}/CodeSystem/cortex" 

25TAG_CODE = "cortex-agent" 

26TAG_DISPLAY = "Written in-process by a Lumen Cortex plugin" 

27 

28EXT_PLUGIN = f"{BASE}/StructureDefinition/cortex-plugin" 

29EXT_CONFIDENCE = f"{BASE}/StructureDefinition/ai-confidence" 

30EXT_MODEL = f"{BASE}/StructureDefinition/ai-model" 

31 

32#: ``Provenance.agent.type`` for a piece of software acting on its own. 

33_AGENT_TYPE_SYSTEM = "http://terminology.hl7.org/CodeSystem/provenance-participant-type" 

34_ACTIVITY_SYSTEM = "http://terminology.hl7.org/CodeSystem/v3-DataOperation" 

35 

36 

37@dataclass(frozen=True) 

38class AiAttribution: 

39 """What a plugin claims about a resource its model produced. 

40 

41 Passing this to a write is what turns "a plugin wrote this" into "a model 

42 wrote this, here is which one and how sure it was". 

43 """ 

44 

45 model: str | None = None 

46 #: 0.0 to 1.0. Recorded as-is; the server does not interpret it. 

47 confidence: float | None = None 

48 #: Free text for ``Provenance.activity.text`` — what the model was doing. 

49 activity: str | None = None 

50 #: Set False to tag the resource but not emit a Provenance resource. 

51 provenance: bool = True 

52 #: References to input resources (e.g. Observation/123) that triggered this attribution. 

53 inputs: list[str] | None = None 

54 

55 def __post_init__(self) -> None: 

56 if self.confidence is not None and not 0.0 <= self.confidence <= 1.0: 

57 raise ValueError(f"confidence must be between 0 and 1, got {self.confidence}") 

58 

59 

60def apply_attribution( 

61 resource: dict[str, Any], 

62 plugin_name: str, 

63 ai: AiAttribution | None, 

64) -> dict[str, Any]: 

65 """Return ``resource`` with the Cortex tag and, if given, the AI extensions. 

66 

67 Idempotent: re-applying does not duplicate the tag or the extensions. 

68 """ 

69 meta: dict[str, Any] = dict(resource.get("meta") or {}) 

70 

71 tags: list[dict[str, Any]] = [dict(t) for t in meta.get("tag") or []] 

72 if not any(t.get("system") == TAG_SYSTEM and t.get("code") == TAG_CODE for t in tags): 

73 tags.append({"system": TAG_SYSTEM, "code": TAG_CODE, "display": TAG_DISPLAY}) 

74 meta["tag"] = tags 

75 

76 extensions: list[dict[str, Any]] = [ 

77 dict(e) 

78 for e in meta.get("extension") or [] 

79 if e.get("url") not in (EXT_PLUGIN, EXT_CONFIDENCE, EXT_MODEL) 

80 ] 

81 extensions.append({"url": EXT_PLUGIN, "valueString": plugin_name}) 

82 if ai is not None: 

83 if ai.model is not None: 

84 extensions.append({"url": EXT_MODEL, "valueString": ai.model}) 

85 if ai.confidence is not None: 

86 extensions.append({"url": EXT_CONFIDENCE, "valueDecimal": ai.confidence}) 

87 meta["extension"] = extensions 

88 

89 return {**resource, "meta": meta} 

90 

91 

92def build_provenance( 

93 *, 

94 target_reference: str, 

95 plugin_name: str, 

96 plugin_version: str, 

97 ai: AiAttribution | None, 

98 recorded: datetime | None = None, 

99) -> dict[str, Any]: 

100 """A ``Provenance`` recording that ``plugin_name`` produced ``target_reference``.""" 

101 when = recorded or datetime.now(UTC) 

102 agent_display = f"Lumen Cortex plugin {plugin_name} {plugin_version}".strip() 

103 

104 provenance: dict[str, Any] = { 

105 "resourceType": "Provenance", 

106 "target": [{"reference": target_reference}], 

107 "recorded": when.isoformat(), 

108 "activity": { 

109 "coding": [{"system": _ACTIVITY_SYSTEM, "code": "CREATE", "display": "create"}], 

110 }, 

111 "agent": [ 

112 { 

113 "type": { 

114 "coding": [ 

115 {"system": _AGENT_TYPE_SYSTEM, "code": "author", "display": "Author"} 

116 ] 

117 }, 

118 "who": {"display": agent_display}, 

119 } 

120 ], 

121 } 

122 

123 if ai is not None: 

124 if ai.activity: 

125 provenance["activity"]["text"] = ai.activity 

126 if ai.inputs: 

127 provenance["entity"] = [ 

128 {"role": "source", "what": {"reference": ref}} for ref in ai.inputs 

129 ] 

130 

131 extensions: list[dict[str, Any]] = [{"url": EXT_PLUGIN, "valueString": plugin_name}] 

132 if ai is not None: 

133 if ai.model is not None: 

134 extensions.append({"url": EXT_MODEL, "valueString": ai.model}) 

135 if ai.confidence is not None: 

136 extensions.append({"url": EXT_CONFIDENCE, "valueDecimal": ai.confidence}) 

137 provenance["extension"] = extensions 

138 

139 return provenance