feat(pipeline): F2+F3 action/object ontology — DB-backed normalization
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-consent (push) Successful in 36s
CI / test-python-voice (push) Successful in 33s
CI / test-bqas (push) Successful in 31s

Migrates ACTION_TYPES (26+8 types), _NEGATIVE_PATTERNS (22), _ACTION_SYNONYMS
(65), and _OBJECT_SYNONYMS (75) from hardcoded dicts to DB tables.

- SQL migration: 003_action_object_ontology.sql (3 tables)
- Migration scripts: f2_migrate_actions.py (34 types, 145 synonyms), f3_migrate_objects.py (75 objects)
- OntologyRegistry cache: 5min TTL, raises RuntimeError if empty (safe fallback to dicts)
- control_ontology.classify_action/get_phase delegate to DB with dict fallback
- control_dedup.normalize_action/normalize_object delegate to DB with dict fallback
- 25 new tests, 446 total pass, 0 regressions

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-05-03 23:47:53 +02:00
parent aab8eeb335
commit 652e3a65a3
7 changed files with 854 additions and 16 deletions
+22 -10
View File
@@ -223,31 +223,43 @@ _FRAMEWORK_PATTERNS: list[str] = [
def classify_action(text: str) -> str:
"""Classify an obligation action text into a canonical action_type."""
text_lower = text.lower().strip()
"""Classify an obligation action text into a canonical action_type.
# Check negative patterns first
Delegates to DB-backed OntologyRegistry (with 5min cache).
Falls back to hardcoded dicts if DB is unavailable.
"""
try:
from .ontology_registry import get_ontology_registry
return get_ontology_registry().classify_action(text)
except Exception:
pass
# Fallback: original logic
text_lower = text.lower().strip()
for pattern, action_type in _NEGATIVE_PATTERNS:
if pattern in text_lower:
return action_type
# Direct alias match
if text_lower in _ALIAS_TO_ACTION:
return _ALIAS_TO_ACTION[text_lower]
# Substring match (longest first)
best_match = ""
best_action = "implement" # default fallback
best_action = "implement"
for alias, action_type in sorted(_ALIAS_TO_ACTION.items(), key=lambda x: -len(x[0])):
if alias in text_lower and len(alias) > len(best_match):
best_match = alias
best_action = action_type
return best_action
def get_phase(action_type: str) -> str:
"""Get the control_phase for an action_type."""
"""Get the control_phase for an action_type.
Delegates to DB-backed OntologyRegistry with dict fallback.
"""
try:
from .ontology_registry import get_ontology_registry
return get_ontology_registry().get_phase(action_type)
except Exception:
pass
info = ACTION_TYPES.get(action_type, {})
return info.get("phase", "implementation")