# §7. Worked example — the article-embedding tracker > Sema language specification — §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. ```sema # --- 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](./research/05-pl-theory-guarantees.md)): `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. ---