fix(context): accept analyzer vocabulary in causal edges (#1184)

get_causal_chain() matched only the canonical uppercase spellings
(CAUSED, INFLUENCED, PRECEDENT_FOR), while CausalChainAnalyzer's
vocabulary includes the present-tense forms (causes, influences,
leads_to, supports) — and the two differ in word form, not just case,
so case-insensitive matching alone would still miss them. An edge
recorded as "causes" produced an empty audit chain.

Storage normalizes both vocabularies onto the canonical types via
_CAUSAL_EDGE_ALIASES; traversal accepts the union (_CAUSAL_TRAVERSAL_TYPES).
add_causal_relationship() now accepts either spelling and stores the
canonical form.
This commit is contained in:
Aldrin Joseph
2026-08-22 16:40:21 +05:30
parent 483f53aaa6
commit 283b7ada0c
2 changed files with 73 additions and 5 deletions
+25 -5
View File
@@ -438,6 +438,23 @@ _ATTRS_MISSING = object()
#: entities and timestamps.
_CAUSAL_EDGE_TYPES = ("CAUSED", "INFLUENCED", "PRECEDENT_FOR")
# Causal edges circulate under two vocabularies: this module's canonical
# spellings above, and the present-tense spellings CausalChainAnalyzer also
# accepts ("causes", "influences", "leads_to", "supports"). The present-tense
# forms normalize onto the canonical types for storage; traversal accepts
# both vocabularies so an edge recorded either way is never invisible.
_CAUSAL_EDGE_ALIASES = {
"CAUSES": "CAUSED",
"CAUSED": "CAUSED",
"INFLUENCES": "INFLUENCED",
"INFLUENCED": "INFLUENCED",
"PRECEDES": "PRECEDENT_FOR",
"PRECEDENT_FOR": "PRECEDENT_FOR",
}
_CAUSAL_TRAVERSAL_TYPES = frozenset(_CAUSAL_EDGE_ALIASES) | {
"LEADS_TO", "LEAD_TO", "SUPPORTS", "SUPPORT",
}
class ContextGraph:
"""
@@ -2745,9 +2762,12 @@ class ContextGraph:
target_decision_id: Target decision ID
relationship_type: Type of relationship (CAUSED, INFLUENCED, PRECEDENT_FOR)
"""
valid_types = ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"]
if relationship_type not in valid_types:
raise ValueError(f"Relationship type must be one of: {valid_types}")
# Normalize so callers may use either vocabulary's spelling
# ("causes" from CausalChainAnalyzer, or "CAUSED" from this module's
# canonical constant); the stored form is always canonical.
relationship_type = _CAUSAL_EDGE_ALIASES.get(relationship_type.upper())
if relationship_type is None:
raise ValueError(f"Relationship type must be one of: {_CAUSAL_EDGE_TYPES}")
# Check if decisions exist - if not, skip adding relationship
if source_decision_id not in self.nodes or target_decision_id not in self.nodes:
@@ -2839,11 +2859,11 @@ class ContextGraph:
# Find connected decisions
for edge in self.edges:
if direction == "upstream":
if edge.target_id == current_id and edge.edge_type in ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"]:
if edge.target_id == current_id and edge.edge_type.upper() in _CAUSAL_TRAVERSAL_TYPES:
if edge.source_id not in visited and depth < max_depth:
queue.append((edge.source_id, depth + 1))
else: # downstream
if edge.source_id == current_id and edge.edge_type in ["CAUSED", "INFLUENCED", "PRECEDENT_FOR"]:
if edge.source_id == current_id and edge.edge_type.upper() in _CAUSAL_TRAVERSAL_TYPES:
if edge.target_id not in visited and depth < max_depth:
queue.append((edge.target_id, depth + 1))
@@ -323,3 +323,51 @@ def test_entity_based_inference_still_applies_without_explicit_edges():
hop["from"] == earlier and hop["to"] == later and hop["type"] == "influences"
for hop in hops
)
def test_get_causal_chain_accepts_lowercase_causal_edge_types():
"""Issue #1184: edges recorded with the analyzer's lowercase vocabulary
must be traversed by get_causal_chain().
CausalChainAnalyzer documents causal types as lowercase ("causes",
"influences", ...) while get_causal_chain() matched only the uppercase
spellings, so an edge recorded as "causes" produced an empty audit
chain silent and in the dangerous direction.
"""
graph = ContextGraph(advanced_analytics=True)
cause = graph.record_decision(
category="a", scenario="upstream", reasoning="r",
outcome="x", confidence=0.9,
)
effect = graph.record_decision(
category="b", scenario="downstream", reasoning="r",
outcome="y", confidence=0.9,
)
graph.add_edge(cause, effect, "causes")
chain = graph.get_causal_chain(effect, direction="upstream")
assert [decision.decision_id for decision in chain] == [cause]
def test_add_causal_relationship_accepts_any_case_and_stores_canonical():
"""Issue #1184: add_causal_relationship() should accept either spelling
and store the canonical uppercase vocabulary."""
graph = ContextGraph(advanced_analytics=True)
cause = graph.record_decision(
category="a", scenario="upstream", reasoning="r",
outcome="x", confidence=0.9,
)
effect = graph.record_decision(
category="b", scenario="downstream", reasoning="r",
outcome="y", confidence=0.9,
)
graph.add_causal_relationship(cause, effect, relationship_type="causes")
edges = [
edge for edge in graph.edges
if edge.source_id == cause and edge.target_id == effect
]
assert edges, "add_causal_relationship must store the edge"
assert edges[0].edge_type == "CAUSED"