SDS-027: Isomorphic Semantic Compiler Service

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@dataclass
class ContextVector:
    semantic_memory: List[Concept]       # Relevant ontology nodes
    temporal_state: Dict[str, Any]       # Current system state
    causal_history: List[Event]          # Recent provenance chain
    constraint_set: List[Invariant]      # Active logic gates

def inject_context(intention: str, spec: Spec, history: EventLog) -> EnrichedPrompt:
    """
    C(I, Spec) → I'

    No inference without context injection.
    """
    vector = ContextVector(
        semantic_memory=spec.resolve_concepts(intention),
        temporal_state=spec.current_state(),
        causal_history=history.recent(limit=10),
        constraint_set=spec.active_gates()
    )

    return EnrichedPrompt(
        raw_intention=intention,
        context=vector,
        schema_constraints=spec.output_schema
    )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class LogicGate(ABC):
    @abstractmethod
    def validate(self, prompt: EnrichedPrompt) -> GateResult:
        """Returns PASS, FAIL, or DEFER (needs human)."""
        pass

# Core Gates
class AgencyPreservationGate(LogicGate):
    """Human must approve high-stakes actions."""

    def validate(self, prompt: EnrichedPrompt) -> GateResult:
        if prompt.is_high_stakes and not prompt.has_human_signature:
            return GateResult.DEFER(reason="Requires human approval")
        return GateResult.PASS

class StructuralIntegrityGate(LogicGate):
    """Output must conform to schema."""

    def validate(self, prompt: EnrichedPrompt) -> GateResult:
        if not prompt.output_schema:
            return GateResult.FAIL(reason="No output schema defined")
        return GateResult.PASS

class ProvenanceGate(LogicGate):
    """Every output must cite its sources."""

    def validate(self, prompt: EnrichedPrompt) -> GateResult:
        if not prompt.context.causal_history:
            return GateResult.FAIL(reason="No provenance chain")
        return GateResult.PASS

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class Reifier:
    def reify(
        self,
        prompt: EnrichedPrompt,
        llm_output: str,
        substrate: ThermodynamicSubstrate
    ) -> Union[Artifact, ReificationError]:
        """
        R(G(C(I))) → A

        Transforms stochastic text into structured artifact.
        """
        # Check resource availability
        if not substrate.can_afford(prompt.estimated_cost):
            return ReificationError("THERMODYNAMIC_CONSTRAINT_VIOLATED")

        # Parse output against schema
        try:
            parsed = prompt.output_schema.parse(llm_output)
        except ValidationError as e:
            return ReificationError("SCHEMA_VIOLATION", details=str(e))

        # Consume substrate
        substrate.consume(prompt.actual_cost)

        # Create artifact with provenance
        return Artifact(
            content=parsed,
            schema=prompt.output_schema,
            provenance=Provenance(
                source_intention=prompt.raw_intention,
                context_used=prompt.context,
                gates_passed=[g.name for g in prompt.gates_passed],
                timestamp=now(),
                cost=prompt.actual_cost
            )
        )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class GatePipeline:
    gates: List[LogicGate]

    def validate(self, prompt: EnrichedPrompt) -> PipelineResult:
        for gate in self.gates:
            result = gate.validate(prompt)

            if result == GateResult.FAIL:
                return PipelineResult.REJECTED(gate.name, result.reason)

            if result == GateResult.DEFER:
                return PipelineResult.DEFERRED(gate.name, result.reason)

            prompt.gates_passed.append(gate)

        return PipelineResult.PASSED(prompt)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def project(
    artifact: Artifact,
    target_runtime: Runtime
) -> ProjectedArtifact:
    """
    P: A → A_runtime

    Structural invariants preserved across projection.
    """
    mapper = RUNTIME_MAPPERS[target_runtime]

    projected = mapper.transform(artifact)

    # Verify invariants survived
    assert artifact.invariants.issubset(projected.invariants), \
        "Isomorphism violation: invariants lost in projection"

    return projected

1
2
3
4
5
6
7
@dataclass
class ContextVector:
    semantic_version: int               # Version at query time (REQUIRED)
    semantic_memory: List[Concept]      # Relevant ontology nodes
    temporal_state: Dict[str, Any]      # Current system state
    causal_history: List[Event]         # Recent provenance chain
    constraint_set: List[Invariant]     # Active logic gates

1
2
3
4
5
6
7
8
9
10
11
12
class SemanticVersionGate(LogicGate):
    """Ensures compiler operates on current semantic model."""

    def validate(self, prompt: EnrichedPrompt) -> GateResult:
        current = semantic_core.get_version()

        if prompt.context.semantic_version < current:
            return GateResult.FAIL(
                reason=f"Stale semantic model v{prompt.context.semantic_version}, current is v{current}",
                remediation="Re-inject context with current version"
            )
        return GateResult.PASS

1
2
3
4
5
6
7
8
@event_handler("governance.semantic-model.updated")
def on_semantic_update(event: SemanticModelUpdatedEvent):
    # Invalidate any cached concepts
    semantic_cache.invalidate(event.data.changed_concepts)
    semantic_cache.invalidate_policies(event.data.changed_policies)

    # Log version transition
    logger.info(f"Semantic model updated: v{event.data.previous_version} → v{event.data.current_version}")