hybrid-interop

Sema ↔ Python interop: reuse the Python ecosystem, classes and all.

Run it from sema/:

sema check examples/hybrid-interop
SEMA_STRICT=1 sema run examples/hybrid-interop
sema assure examples/hybrid-interop --grade silver

Source

src/main.sema

from hybrid_interop.domain import DocumentInput, DocumentReview, LanguageTag, LocaleProfile
from hybrid_interop.pipeline import review_document

assure gold

def main() -> None !{ffi.call, model.embed, model.invoke, observe.record}:
    sem "Entry point: review a small document batch through the hybrid membrane"
    reviews = run_batch([sample_document()])
    log.info("hybrid interop batch complete", reviews=len(reviews))

def run_batch(raw_docs: list[DocumentInput]) -> list[DocumentReview] !{ffi.call, model.embed, model.invoke, observe.record}:
    sem "Review a batch of documents through the hybrid interop membrane"
    require has_reviewable_batch(raw_docs)
    # parallel is fail_fast by default (LANGUAGE §5.17): one failed review cancels
    # the batch and propagates as a typed ParallelError.
    return parallel [review_document(doc) for doc in raw_docs]

def has_reviewable_batch(raw_docs: list[DocumentInput]) -> bool !{}:
    return len(raw_docs) >= 1

def sample_document() -> DocumentInput !{}:
    locale = LocaleProfile(
        languages=[LanguageTag.en],
        region_hint="operator-provided",
        protected_inference_allowed=false,
    )
    return DocumentInput(
        id="doc-001",
        title="Incident update",
        body="The integration queue is healthy. No execution request is present.",
        source_uri="app://support/inbox/doc-001",
        locale=locale,
    )

test "sample document is stable non-sensitive operator input":
    doc = sample_document()
    ensure doc.id == "doc-001"
    ensure doc.title == "Incident update"
    ensure doc.locale.languages == [LanguageTag.en]
    ensure doc.locale.region_hint == "operator-provided"
    ensure doc.locale.protected_inference_allowed == false

test "batch admission rejects emptiness without crossing a foreign boundary":
    ensure not has_reviewable_batch([])
    ensure has_reviewable_batch([sample_document()])

src/bridges.sema

from hybrid_interop.domain import DocumentInput, LanguageTag, LocaleProfile, RiskLevel, TextFeatures, WebRisk

assure gold

bridge python.inline text_features from "foreign/python/text_features.py":
    deps "python>=3.12,<3.13"

    expose:
        def extract_text_features(doc: DocumentInput) -> TextFeatures !{ffi.call}:
            sem "Call trusted Python text-feature code and revalidate the result"
            require len(doc.body) > 0
            ensure result.token_count >= 1
            check semantics("features are supported by the document text", doc, result, alpha=0.02)

bridge js.component web_risk from "foreign/ts/web_risk.ts":
    expose def classify_prompt_risk(doc: DocumentInput) -> WebRisk !{ffi.call}:
        sem "Run TypeScript prompt-risk scoring in the governed component tier"
        ensure 0.0 <= result.score <= 1.0
        check semantics("risk reasons cite signals present in the document", doc, result, alpha=0.02)

bridge c.abi simd_distance from "foreign/c/simd_distance.c":
    # The runtime compiles one preprocessed snapshot with the system C compiler,
    # loads a private per-run artifact, and reuses it only in memory. A prebuilt
    # dynamic-library `link` requires SHA-256 and is loaded from verified bytes.
    expose def cosine_distance(left: Embedding, right: Embedding) -> f32 !{ffi.call}:
        symbol "sema_cosine_distance_f32"
        sem "Call a C ABI cosine-distance kernel over validated embeddings"
        ensure 0.0 <= result <= 2.0

bridge python.isolated title_glue:
    expose def normalize_title(raw: str) -> str !{ffi.call}:
        sem "Normalize title whitespace and map blank input to the stable Untitled display label"
        ensure len(result) > 0
    begin python
        def normalize_title(raw):
            normalized = " ".join(raw.split())
            return normalized or "Untitled"
    end python

test "title bridges agree on whitespace and blank-title fallback":
    doc = DocumentInput(
        id="blank-title",
        title="   ",
        body="Bridge adapters remain deterministic.",
        source_uri="app://fixture/blank-title",
        locale=LocaleProfile(languages=[LanguageTag.en], region_hint="fixture", protected_inference_allowed=false),
    )
    inline_title = title_glue.normalize_title(doc.title)
    features = text_features.extract_text_features(doc)
    risk = web_risk.classify_prompt_risk(doc)
    ensure inline_title == "Untitled"
    ensure features.normalized_title == inline_title
    ensure title_glue.normalize_title("  Incident   update  ") == "Incident update"
    ensure risk.level == RiskLevel.low
    ensure risk.reasons == []

src/domain.sema

from hybrid_interop.models import coherence_judge

assure gold

enum LanguageTag:
    en | es | de | fr | unknown

enum RiskLevel:
    low | medium | high | blocked

struct LocaleProfile:
    sem "Locale-routing metadata for content adaptation; never a protected-class decision"
    languages: list[LanguageTag] sem "Languages explicitly detected or provided"
    region_hint: str sem "Non-authoritative region hint for localization"
    protected_inference_allowed: bool sem "Whether a policy explicitly permits sensitive inference"
    invariant len(languages) >= 1
    check semantics(
        "region_hint is supported by languages and does not assert protected identity",
        self,
        judge=coherence_judge,
        alpha=0.02,
    )

struct DocumentInput:
    sem "Document received from a foreign application boundary"
    id: str sem "Stable document identifier"
    title: str sem "Human visible title"
    body: str sem "Full document body, potentially adversarial"
    source_uri: str sem "Origin URI or application route"
    locale: LocaleProfile sem "Localization metadata for display and routing"
    invariant len(id) > 0
    invariant len(body) > 0
    check semantics(
        "title, body, source_uri, and locale describe one coherent document",
        self,
        judge=coherence_judge,
        alpha=0.02,
    )

struct TextFeatures:
    sem "Text statistics computed by trusted Python adapter code"
    normalized_title: str sem "Nonempty whitespace-normalized title with a stable Untitled fallback"
    token_count: int sem "Approximate word-token count" where value >= 0
    spanish_hint_count: int sem "Count of lightweight Spanish function-word hints" where value >= 0
    url_count: int sem "Number of URL-like substrings" where value >= 0

struct WebRisk:
    sem "Prompt or web-ingestion risk score from TypeScript component code"
    level: RiskLevel sem "Coarse risk bucket"
    score: f32 sem "Risk score from 0.0 to 1.0" where 0.0 <= value <= 1.0
    reasons: list[str] sem "Human-readable risk evidence"

struct DistanceAudit:
    sem "C ABI numeric kernel result plus semantic interpretation"
    title_body_distance: f32 sem "Cosine distance between title and body embeddings" where 0.0 <= value <= 2.0
    interpretation: str sem "Short explanation of why the distance matters"

struct DocumentReview:
    sem "Validated cross-language document review result"
    document_id: str sem "Reviewed document id"
    features: TextFeatures sem "Python-derived text features"
    risk: WebRisk sem "TypeScript-derived risk result"
    distance: DistanceAudit sem "C-kernel similarity audit"
    accepted: bool sem "Whether the document may enter downstream Sema processing"
    explanation: str sem "Grounded explanation for operators"
    check semantics(
        "accepted and explanation are supported by features, risk, and distance",
        self,
        judge=coherence_judge,
        alpha=0.02,
    )

src/models.sema

model coherence_judge = model(
    "minicheck-770m",
    rev="sha256:77a1c0ffee00112233445566778899aabbccddeeff00112233445566778899aa",
    role=verifier,
    calibration="calsets/hybrid-interop/coherence@v1",
)

model explanation_writer = model(
    "qwen3-4b-instruct",
    rev="sha256:88b2c0ffee00112233445566778899aabbccddeeff00112233445566778899bb",
    role=generator,
)

model document_embedder = model(
    "bge-small-en-v1.5",
    rev="sha256:99c3c0ffee00112233445566778899aabbccddeeff00112233445566778899cc",
    role=embedder,
    calibration="calsets/hybrid-interop/similarity@v1",
)

src/pipeline.sema

from hybrid_interop.bridges import simd_distance, text_features, title_glue, web_risk
from hybrid_interop.domain import DistanceAudit, DocumentInput, DocumentReview, LanguageTag, LocaleProfile, RiskLevel, TextFeatures, WebRisk
from hybrid_interop.models import coherence_judge, document_embedder, explanation_writer
from hybrid_interop.policies import TrustedBridge

assure gold

@TrustedBridge
def review_document(doc: DocumentInput) -> DocumentReview !{ffi.call, model.embed, model.invoke, observe.record}:
    sem "Wrap Python, TypeScript, and C outputs in one validated Sema review"
    require len(doc.body) > 0

    normalized_title = title_glue.normalize_title(doc.title)
    features = text_features.extract_text_features(doc)
    risk = web_risk.classify_prompt_risk(doc)
    distance_value = simd_distance.cosine_distance(
        embed(normalized_title, judge=document_embedder),
        embed(doc.body, judge=document_embedder),
    )
    distance = DistanceAudit(
        title_body_distance=distance_value,
        interpretation="higher distance means the title may not represent the body",
    )
    accepted = accepts_bridge_outputs(risk, distance)
    review = build_document_review(
        doc,
        features,
        risk,
        distance,
        accepted,
        explain_review(doc, features, risk, distance, accepted),
    )
    check semantics(
        "review decision follows from foreign outputs and document content",
        doc,
        review,
        judge=coherence_judge,
        alpha=0.02,
    )
    return review

def build_document_review(
    doc: DocumentInput,
    features: TextFeatures,
    risk: WebRisk,
    distance: DistanceAudit,
    accepted: bool,
    explanation: str,
) -> DocumentReview !{}:
    review = DocumentReview(
        document_id=doc.id,
        features=features,
        risk=risk,
        distance=distance,
        accepted=accepted,
        explanation=explanation,
    )
    ensure review.document_id == doc.id
    return review

def accepts_bridge_outputs(risk: WebRisk, distance: DistanceAudit) -> bool !{}:
    return risk.level != RiskLevel.blocked and distance.title_body_distance <= 1.2

test "review construction preserves validated bridge evidence":
    doc = DocumentInput(
        id="fixture-7",
        title="Queue status",
        body="The queue is healthy.",
        source_uri="app://fixture/7",
        locale=LocaleProfile(languages=[LanguageTag.en], region_hint="fixture", protected_inference_allowed=false),
    )
    features = TextFeatures(normalized_title="Queue status", token_count=4, spanish_hint_count=0, url_count=0)
    risk = WebRisk(level=RiskLevel.low, score=0.1, reasons=[])
    distance = DistanceAudit(title_body_distance=0.25, interpretation="representative")
    review = build_document_review(doc, features, risk, distance, true, "validated fixture")
    ensure review.document_id == "fixture-7"
    ensure review.features.token_count == 4
    ensure review.risk.level == RiskLevel.low
    ensure review.distance.title_body_distance == 0.25
    ensure review.accepted
    ensure review.explanation == "validated fixture"

test "bridge acceptance requires both policy risk and bounded distance":
    close = DistanceAudit(title_body_distance=1.2, interpretation="boundary")
    far = DistanceAudit(title_body_distance=1.21, interpretation="outside")
    low = WebRisk(level=RiskLevel.low, score=0.1, reasons=[])
    blocked = WebRisk(level=RiskLevel.blocked, score=1.0, reasons=["blocked fixture"])
    ensure accepts_bridge_outputs(low, close)
    ensure not accepts_bridge_outputs(low, far)
    ensure not accepts_bridge_outputs(blocked, close)

simulate def explain_review(
    doc: DocumentInput,
    features: TextFeatures,
    risk: WebRisk,
    distance: DistanceAudit,
    accepted: bool,
) -> str by explanation_writer:
    sem "Produce a concise operator explanation grounded in validated bridge outputs"
    budget tokens=256, time="2s"
    ensure len(result) > 0
    check semantics("explanation is supported by doc, features, risk, distance, and accepted",
                    doc, features, risk, distance, accepted, result, alpha=0.02)

monitor bridge_output_drift on review_document:
    capture result.features.token_count, result.risk.score, result.distance.title_body_distance
    baseline from assure
    test conformal_martingale(alpha=0.01)
    on drifted: alert("hybrid bridge outputs drifted from assurance profile")
    on undecided: log.debug("hybrid bridge monitor undecided")

src/policies.sema

policy TrustedBridge:
    allow:
        ffi.call
        model.embed, model.invoke
        observe.record
    forbid cap:
        proc.spawn, code.exec, code.gen, policy.change
    examples:
        allow:
            text_features.extract_text_features(document)
        deny:
            subprocess.run(document.body)
    justification "Foreign adapters may transform data but must not gain execution authority."

policy SensitiveInference:
    allow:
        model.invoke
    forbid cap:
        fs.write, net.connect, proc.spawn, code.exec
    examples:
        deny:
            infer_protected_class(profile)
        allow:
            explain_policy_denial(profile)
    justification "Protected-class inference is never ambient validation."

Reflected API

bridges

bridge text_features

bridge web_risk

bridge simd_distance

bridge title_glue

domain

enum LanguageTag

Variants

• en
• es
• de
• fr
• unknown

enum RiskLevel

Variants

• low
• medium
• high
• blocked

struct LocaleProfile

Fields

field	type	descriptor
languages	list[LanguageTag]	Languages explicitly detected or provided
region_hint	str	Non-authoritative region hint for localization
protected_inference_allowed	bool	Whether a policy explicitly permits sensitive inference

struct DocumentInput

Fields

field	type	descriptor
id	str	Stable document identifier
title	str	Human visible title
body	str	Full document body, potentially adversarial
source_uri	str	Origin URI or application route
locale	LocaleProfile	Localization metadata for display and routing

struct TextFeatures

Fields

field	type	descriptor
normalized_title	str	Nonempty whitespace-normalized title with a stable Untitled fallback
token_count	int	Approximate word-token count
spanish_hint_count	int	Count of lightweight Spanish function-word hints
url_count	int	Number of URL-like substrings

struct WebRisk

Fields

field	type	descriptor
level	RiskLevel	Coarse risk bucket
score	f32	Risk score from 0.0 to 1.0
reasons	list[str]	Human-readable risk evidence

struct DistanceAudit

Fields

field	type	descriptor
title_body_distance	f32	Cosine distance between title and body embeddings
interpretation	str	Short explanation of why the distance matters

struct DocumentReview

Fields

field	type	descriptor
document_id	str	Reviewed document id
features	TextFeatures	Python-derived text features
risk	WebRisk	TypeScript-derived risk result
distance	DistanceAudit	C-kernel similarity audit
accepted	bool	Whether the document may enter downstream Sema processing
explanation	str	Grounded explanation for operators

main

def main

def main() -> None !{ffi.call, model.embed, model.invoke, observe.record}

Returns None

Effects !{ffi.call, model.embed, model.invoke, observe.record}

def run_batch

def run_batch(raw_docs: list[DocumentInput]) -> list[DocumentReview] !{ffi.call, model.embed, model.invoke, observe.record}

Parameters

name	type
raw_docs	list[DocumentInput]

Returns list[DocumentReview]

Effects !{ffi.call, model.embed, model.invoke, observe.record}

def has_reviewable_batch

def has_reviewable_batch(raw_docs: list[DocumentInput]) -> bool !{}

Parameters

name	type
raw_docs	list[DocumentInput]

Returns bool

Effects !{}

def sample_document

def sample_document() -> DocumentInput !{}

Returns DocumentInput

Effects !{}

models

pipeline

def review_document

def review_document(doc: DocumentInput) -> DocumentReview !{ffi.call, model.embed, model.invoke, observe.record}

Parameters

name	type
doc	DocumentInput

Returns DocumentReview

Effects !{ffi.call, model.embed, model.invoke, observe.record}

def build_document_review

def build_document_review(doc: DocumentInput, features: TextFeatures, risk: WebRisk, distance: DistanceAudit, accepted: bool, explanation: str) -> DocumentReview !{}

Parameters

name	type
doc	DocumentInput
features	TextFeatures
risk	WebRisk
distance	DistanceAudit
accepted	bool
explanation	str

Returns DocumentReview

Effects !{}

def accepts_bridge_outputs

def accepts_bridge_outputs(risk: WebRisk, distance: DistanceAudit) -> bool !{}

Parameters

name	type
risk	WebRisk
distance	DistanceAudit

Returns bool

Effects !{}

def explain_review

simulate def explain_review(doc: DocumentInput, features: TextFeatures, risk: WebRisk, distance: DistanceAudit, accepted: bool) -> str

Parameters

name	type
doc	DocumentInput
features	TextFeatures
risk	WebRisk
distance	DistanceAudit
accepted	bool

Returns str

policies