# Copyright (c) 2026 PowerOn AG # All rights reserved. """Deterministic compiler: OntologyDescriptor -> sub-agent prompt block. Phase 2 replaces a feature's hand-written ``_AGENT_DOMAIN_HINTS`` text with a structured :class:`OntologyDescriptor`. This compiler renders the descriptor into a stable, terse Markdown-ish block that the sub-agent appends to its system prompt -- the same source of truth the :class:`QueryValidator` consults. The output is intentionally: * short (every token costs every call) * deterministic (no f-string ordering bugs, no Python dict iteration) * free of internal jargon ('canonicalQueryPattern' is rendered as 'CANONICAL PATTERN' for the LLM) """ from __future__ import annotations from typing import Iterable, List from modules.serviceCenter.services.serviceAgent.datamodelOntology import ( CanonicalQueryPattern, Constraint, ConstraintRule, Entity, OntologyDescriptor, Relation, ) def compileOntologyToPrompt(ontology: OntologyDescriptor) -> str: """Render *ontology* into a sub-agent prompt block. The output starts with a stable marker line (``DOMAIN ONTOLOGY (...)``) so downstream tooling can find/replace it deterministically. """ lines: List[str] = [] lines.append(f"DOMAIN ONTOLOGY ({ontology.featureCode}):") lines.append("") lines.extend(_renderEntities(ontology.entities)) relationLines = _renderRelations(ontology.relations) if relationLines: lines.append("") lines.extend(relationLines) constraintLines = _renderConstraints(ontology.constraints) if constraintLines: lines.append("") lines.extend(constraintLines) patternLines = _renderPatterns(ontology.canonicalPatterns) if patternLines: lines.append("") lines.extend(patternLines) return "\n".join(lines).rstrip() + "\n" def _renderEntities(entities: Iterable[Entity]) -> List[str]: out: List[str] = ["ENTITIES:"] for e in entities: head = f"- {e.name}" if e.parentEntity: head += f" (specializes {e.parentEntity})" if e.pythonClass: head += f" [table: {e.pythonClass}]" out.append(head) if e.description: out.append(f" {e.description}") for inv in e.invariants: out.append(f" * {inv.description}") return out def _renderRelations(relations: Iterable[Relation]) -> List[str]: rels = list(relations) if not rels: return [] out: List[str] = ["RELATIONS:"] for r in rels: line = f"- {r.fromEntity} -> {r.toEntity} ({r.cardinality.value}" if r.via: line += f" via {r.via}" line += ")" out.append(line) return out def _renderConstraints(constraints: Iterable[Constraint]) -> List[str]: cons = list(constraints) if not cons: return [] out: List[str] = ["CONSTRAINTS (validator-enforced):"] for c in cons: rule = _ruleLabel(c.rule) line = f"- {rule} on {c.appliesTo}: {c.message}" params = c.params or {} required = params.get("requiredFields") if isinstance(required, list) and required: line += f" (required filters: {', '.join(required)})" intents = params.get("intents") if isinstance(intents, list) and intents: line += f" (intents: {', '.join(intents)})" out.append(line) return out def _ruleLabel(rule: ConstraintRule) -> str: return rule.value.replace("_", " ").lower() def _renderPatterns(patterns: Iterable[CanonicalQueryPattern]) -> List[str]: pats = list(patterns) if not pats: return [] out: List[str] = ["CANONICAL QUERY PATTERNS (mimic these tool calls):"] for i, p in enumerate(pats, start=1): out.append(f"{i}) intent={p.intent}: {p.description}") out.append(f" call: {_renderPatternCall(p.pattern)}") extra = p.pattern.get("_postProcessing") if isinstance(p.pattern, dict) else None if isinstance(extra, str): out.append(f" note: {extra}") return out def _renderPatternCall(pattern: dict) -> str: """Render the pattern as a compact one-line tool call signature.""" tool = pattern.get("tool", "?") parts: List[str] = [] for key in ("tableName", "aggregate", "field", "groupBy", "orderBy"): if key in pattern and pattern[key] is not None and not str(key).startswith("_"): parts.append(f"{key}={pattern[key]!r}") if "fields" in pattern and pattern["fields"]: parts.append(f"fields={pattern['fields']}") if "filters" in pattern and pattern["filters"]: compact = ", ".join( f"{f.get('field')}{f.get('op','=')}{f.get('value')!r}" for f in pattern["filters"] if isinstance(f, dict) ) parts.append(f"filters=[{compact}]") return f"{tool}({', '.join(parts)})"