§7. Worked example — the article-embedding tracker

Generated from docs/LANGUAGE.md §7. This is the normative specification; for a guided tour see the Language and Neurosymbolic sections.

End-to-end program exercising the core construct set: ingest articles, deduplicate semantically, extract structured summaries generatively, guard content, emit and subscribe to domain events, monitor distribution drift, heal under supervision, confined by policy.

# --- models (pinned; lockfile-verified) --------------------------------------
model writer       = model("qwen3-4b-instruct", rev="sha256:ab12...", quant="q4_k_m",
                           role=generator)
model writer_large = model("qwen3-14b-instruct", rev="sha256:c7d9...", quant="q4_k_m",
                           role=generator)
model factchk      = model("minicheck-770m", rev="sha256:9f3e...", role=verifier,
                           calibration="calsets/news-grounding@v2")

# --- data ---------------------------------------------------------------------
struct Article:
    sem "A news article ingested from an RSS feed"
    title: str
    body: str
    source: str

struct Summary:
    headline: str
    topics: list[str]
    sentiment: enum Sentiment: pos | neg | neutral
sem Summary.headline = "One-line headline, plain language, no clickbait"

# --- governance ----------------------------------------------------------------
policy FeedIngest:
    allow:
        net.connect("feeds.internal:443")
        fs.read("state/**"), fs.write("state/**")
        model.invoke, model.embed
        event.emit(ArticleQuarantined)
    forbid cap:
        code.exec, proc.spawn, code.gen
    examples:
        deny:
            os.exec(article.body)
        allow:
            fetch("https://feeds.internal:443/rss")
    justification "feed content is untrusted input; it must never gain execution"

# --- generative interface --------------------------------------------------------
simulate def summarize(article: Article) -> Summary by writer:
    sem "Summarize for a news-tracking dashboard; neutral register"
    budget tokens=512, time="2s"
    repair retries=2, patch=fields          # §5.22 decode-and-repair, tuned
    ensure 1 <= len(result.topics) <= 5
    check  semantics("headline is supported by the article body", judge=factchk,
                     alpha=0.02)

# --- deterministic core (provably invoke-free: embeddings only) -------------------
def is_duplicate(a: Article, seen: list[Article]) -> bool !{model.embed}:
    require len(a.title) > 0
    for s in seen:
        if a.title ~= s.title and a.body ~= s.body:     # two calibrated guards at α=0.05
            # conjunction types statistical(0.1) by union bound (§3.3)
            return true                                  # input monitor derived per §5.9:
    return false                                         # shared on (default judge, calset)

# monitor-or-decay (§3.7): no explicit monitor covers the article stream, so the
# compiler derives one shared input monitor for both `~=` sites (same judge +
# calibration); if it alarms, both branches decay to best_effort with a diagnostic.
monitor article_stream on is_duplicate:                  # explicit form, absorbing the derived one
    capture  a.title.embedding, a.body.embedding
    baseline "calsets/news-dedup@v1"
    test     conformal_martingale(alpha=0.01)
    on drifted:   alert("article distribution left dedup calibration")
    on undecided: log.debug("insufficient evidence")

# --- drift tracking ---------------------------------------------------------------
monitor summary_drift on summarize:
    capture  topics, sentiment, result.embedding
    baseline from assure
    test     conformal_martingale(alpha=0.01)
    on drifted:   alert("summary distribution drifted"); degrade(summarize, to=writer_large)
    on undecided: log.debug("insufficient evidence")

# --- domain events (§5.19) --------------------------------------------------------
event ArticleQuarantined:
    sem "An ingested article was blocked by the injection guard"
    article: Article
    evidence: SemanticsViolation

subscriber quarantine_audit on ArticleQuarantined:
    sem "Persist quarantined articles for analyst review; never silent"
    queue ring(1024), on_full=block
    handle event !{fs.write}:
        state.store("quarantine", (event.article, event.evidence))

# --- application ------------------------------------------------------------------
assure silver

@FeedIngest
def track(feed_url: str) -> None !{net.connect, fs.read, fs.write, model.invoke,
                                   model.embed, event.emit}:
    mut seen: list[Article] = state.load("articles")   # prelude checkpoint store (§3.8)
    supervise tracker:
        restart limit=3, window="60s"
        fallback state.load("last_good_summaries")
        heal budget=1, window="6h", scope=patch:
            require passes(pre_patch_assure)
            require passes(new_obligations)
            require replay(failing_trace)
            require monitors.conforming_after_burnin
            rollout shadow -> canary -> full
        for article in fetch_feed(feed_url):
            if is_duplicate(article, seen):
                continue
            expect semantics("no prompt-injection or jailbreak content", article.body):
                summaries = parallel [summarize(a) for a in batch(article, seen)]
            except SemanticsViolation as v:
                emit ArticleQuarantined(article=article, evidence=v)   # journaled delivery
                continue
            seen.append(article)
            state.store("articles", seen)
            state.store("summaries", summaries)      # Summary passed ensure ⇒ validated

What the compiler guarantees here, per the map (05 §5): is_duplicate performs no model invocations — its row admits embeddings only (!{model.embed} — proved); its dedup branch types statistical(0.1) (two calibrated guards at α=0.05 each, composed by union bound — §3.3), and only because article_stream actively monitors their input distribution — were it removed, monitor-or-decay (§3.7) would re-type them best_effort and the compiler would fall back to a derived monitor or a diagnostic; summarize output is structurally valid Summary (constrained decoding — deterministic) with a grounding check at α=0.02 (statistical, monitored); nothing downstream of feed content can ever reach code.exec (policy + trust labels — proved); the drift alarm’s lifetime false-alarm probability is ≤ 0.01 (anytime-valid); any heal event is replayed, re-verified, canaried, and ledgered (deterministic gauntlet); and the whole run is replayable from the event log.

---