Coverage for node / src / stigmem_node / cid.py: 96%

1"""Content-addressed fact IDs — spec §25 (CID v2). 

2 

3CID = "sha256:" + hex_lowercase(SHA-256(canonical_fact_body)) 

4 

5Canonicalization is **sorted-key compact ``json.dumps``**, NOT full RFC 8785 

6(JCS). Concretely the canonical body is serialized with 

7``json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)`` 

8and UTF-8 encoded. This gives deterministic key ordering and no insignificant 

9whitespace, but does NOT apply RFC 8785's number canonicalization or Unicode 

10normalization. Producers and verifiers MUST use this same serialization (see 

11``compute_cid``); the documented field set below is what makes it stable. 

12 

13**CID v2 (breaking change, 2026-06-06).** The canonical body is a JSON object 

14with exactly 8 fields in lexicographic key order: 

15 confidence, entity, interpret_as, relation, scope, source, value_type, value_v 

16 

17`interpret_as` is now bound (per ADR-003 / threat R-23): flipping a fact's 

18interpretation between `content` and `instruction` changes the CID, so it is 

19detected on the read path. **v1 CIDs (which omitted `interpret_as`) are NOT 

20accepted** — pre-v2 facts must be upgraded via the CID backfill migration; 

21until migrated they fail read-path verification (409 cid_mismatch). 

22 

23Security-relevant excluded fields (§25.2.1 rev 15): 

24 valid_until, derived_from, attestation_chain, source_trust, signature, reason 

25 (these require independent validation; CID coverage alone is not sufficient) 

26 

27fact_id and cid are also excluded (circular). 

28timestamp/created_at is excluded so the same assertion at different times shares one CID. 

29""" 

30 

31from __future__ import annotations 

32 

33import hashlib 

34import json 

35import re 

36from typing import Any 

37 

38_CID_PREFIX = "sha256:" 

39_CID_HEX_RE = re.compile(r"^sha256:[0-9a-f]{64}$") 

40 

41 

42def compute_cid( 

43 entity: str, 

44 relation: str, 

45 value_type: str, 

46 value_v: str, 

47 source: str, 

48 scope: str, 

49 confidence: float = 1.0, 

50 interpret_as: str = "content", 

51) -> str: 

52 """Return the CID v2 for a fact's canonical body (spec §25.2.1, §25.2.2). 

53 

54 `interpret_as` is part of the canonical body (CID v2); a flip between 

55 `content` and `instruction` produces a different CID. 

56 """ 

57 body: dict[str, Any] = { 

58 "confidence": confidence, 

59 "entity": entity, 

60 "interpret_as": interpret_as, 

61 "relation": relation, 

62 "scope": scope, 

63 "source": source, 

64 "value_type": value_type, 

65 "value_v": value_v, 

66 } 

67 canonical = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode( 

68 "utf-8" 

69 ) 

70 digest = hashlib.sha256(canonical).hexdigest() 

71 return f"{_CID_PREFIX}{digest}" 

72 

73 

74def compute_cid_from_row(row: Any) -> str: 

75 """Convenience wrapper: compute CID from a facts-table row.""" 

76 return compute_cid( 

77 entity=row["entity"], 

78 relation=row["relation"], 

79 value_type=row["value_type"], 

80 value_v=row["value_v"] or "", 

81 source=row["source"], 

82 scope=row["scope"], 

83 confidence=float(row["confidence"]), 

84 interpret_as=(_optional_row_value(row, "interpret_as") or "content"), 

85 ) 

86 

87 

88class CidMismatchError(ValueError): 

89 """Raised when a stored fact CID does not match its canonical body.""" 

90 

91 def __init__(self, *, fact_id: str, stored_cid: str, computed_cid: str) -> None: 

92 super().__init__(f"CID mismatch for fact {fact_id}") 

93 self.fact_id = fact_id 

94 self.stored_cid = stored_cid 

95 self.computed_cid = computed_cid 

96 

97 

98def _optional_row_value(row: Any, key: str) -> Any: 

99 try: 

100 keys = row.keys() 

101 except AttributeError: 

102 return row.get(key) if isinstance(row, dict) else None 

103 return row[key] if key in keys else None 

104 

105 

106def stored_cid_from_row(row: Any) -> str | None: 

107 """Return the stored/projected CID for a fact row, if one is present.""" 

108 projected = _optional_row_value(row, "projected_cid") 

109 if projected is not None: 

110 return str(projected) 

111 stored = _optional_row_value(row, "cid") 

112 return None if stored is None else str(stored) 

113 

114 

115def verify_cid_from_row(row: Any) -> None: 

116 """Verify a fact row's stored CID, preserving legacy NULL-CID rows.""" 

117 stored = stored_cid_from_row(row) 

118 if stored is None: 

119 return 

120 computed = compute_cid_from_row(row) 

121 if computed != stored: 

122 raise CidMismatchError( 

123 fact_id=str(row["id"]), 

124 stored_cid=stored, 

125 computed_cid=computed, 

126 ) 

127 

128 

129def is_valid_cid(s: str) -> bool: 

130 """Return True if *s* looks like a well-formed sha256 CID (spec §25.2).""" 

131 return bool(_CID_HEX_RE.match(s)) 

132 

133 

134def is_cid(s: str) -> bool: 

135 """Return True if *s* starts with the sha256: prefix (quick pre-filter).""" 

136 return s.startswith(_CID_PREFIX)