feat: silo plugin registry (tessera-lake S1)

README.md

@@ -263,6 +263,44 @@ GET    /api/trends                pass^k/mean series for the Dashboard
 
 Dev setup, house rules (contract regeneration, `scorer_version` policy, the ADR process), and how to add an org: **[CONTRIBUTING.md](CONTRIBUTING.md)**.
 
+## Extending: adding a silo type
+
+Silos are pluggable. A silo type bundles: an MCP server module (run via
+`python -m`, reads `TESSERA_OUT` for the compiled org), its tool names, a
+prompt blurb, a consulted-claims credit function, and (optionally) custom
+compile build/write hooks.
+
+```python
+from tessera.silos.registry import SiloType
+
+LAKE = SiloType(
+    name="lake",
+    server_module="tessera_lake.mcp.catalog_server",
+    tool_names=("search_datasets", "get_dataset_metadata", "query_series", "list_tags"),
+    prompt_blurb=" You can also query a data-lake catalog (search_datasets, "
+                 "get_dataset_metadata, query_series, list_tags).",
+    consulted=lake_consulted,   # (tool_name, args, result, manifest) -> set[claim_id]
+    build=lake_build,           # (claims) -> (payload, manifest_entries)   [optional]
+    write=lake_write,           # (payload, out_dir) -> None                [optional]
+)
+```
+
+Publish it from your package via the entry-point group:
+
+```toml
+[project.entry-points."tessera.silo_types"]
+lake = "your_pack.silo:LAKE"
+```
+
+Tessera discovers it lazily on first registry lookup. An entry point may also
+load to an iterable of `SiloType`s, or a zero-arg callable returning either —
+useful when one pack registers several silo types. A broken entry point is
+skipped with a logged warning rather than breaking the registry. Claims with
+`silo="lake"` then compile through your build/write hooks (or the default
+field/prose renderer if you don't define them), the eval task launches your
+MCP server alongside the built-ins, and scoring credits consulted claims
+through your `consulted` function.
+
 ## Status and roadmap
 
 - [x] **v0 (shipped mid-2026)** generator, MCP harness, one core task suite, the scorer, a runnable quickstart.

src/tessera/compiler.py

@@ -17,8 +17,10 @@
 import json
 import re
 from pathlib import Path
+from typing import Any
 
-from .models import Blueprint, RenderAs
+from .models import Blueprint, Claim, RenderAs
+from .silos.registry import registry as silo_registry
 
 
 def _slug(text: str) -> str:
@@ -47,6 +49,8 @@ def build_artifacts(blueprint: Blueprint) -> dict:
     "docs": [{"path": rel, "content": str}]}``. This is what powers the compile-PREVIEW
     endpoint (show the resulting org without materializing it). Raises ``ValueError`` on
     an intra-silo ``(subject, predicate)`` collision: contradictions must be cross-silo.
+    A ``"plugins"`` key (mapping silo name to opaque payload) is present only when a
+    claim's silo type defines a custom ``build`` hook.
     """
     seen: set[tuple[str, str, str]] = set()
     for claim in blueprint.claims:
@@ -61,8 +65,13 @@ def build_artifacts(blueprint: Blueprint) -> dict:
     silos: dict[str, dict[str, dict]] = {}  # silo -> subject -> {predicate: value}
     docs: list[dict[str, str]] = []
     manifest: dict[str, dict] = {}
+    custom_claims: dict[str, list[Claim]] = {}
 
     for claim in blueprint.claims:
+        st = silo_registry.get_optional(claim.silo)
+        if st is not None and st.build is not None:
+            custom_claims.setdefault(claim.silo, []).append(claim)
+            continue
         if claim.render.as_ is RenderAs.field:
             silos.setdefault(claim.silo, {}).setdefault(claim.subject, {})[
                 claim.predicate
@@ -86,7 +95,16 @@ def build_artifacts(blueprint: Blueprint) -> dict:
                 "locator": rel,
             }
 
-    return {"manifest": manifest, "silos": silos, "docs": docs}
+    plugins: dict[str, Any] = {}
+    for silo_name, claims in custom_claims.items():
+        payload, entries = silo_registry.get(silo_name).build(claims)
+        plugins[silo_name] = payload
+        manifest.update(entries)
+
+    artifacts = {"manifest": manifest, "silos": silos, "docs": docs}
+    if plugins:
+        artifacts["plugins"] = plugins
+    return artifacts
 
 
 def write_artifacts(artifacts: dict, out_dir: str | Path) -> dict[str, dict]:
@@ -99,6 +117,8 @@ def write_artifacts(artifacts: dict, out_dir: str | Path) -> dict[str, dict]:
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(doc["content"])
     _write_json(out / "manifest.json", artifacts["manifest"])
+    for silo_name, payload in artifacts.get("plugins", {}).items():
+        silo_registry.get(silo_name).write(payload, out)
     return artifacts["manifest"]
 
 

src/tessera/evals/scoring.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import re
 from typing import Any
 
@@ -14,6 +13,7 @@
 from tessera.evals.dataset import ProbeMeta
 from tessera.evals.judges import accuracy_judge as _default_accuracy_judge
 from tessera.evals.judges import refusal_judge as _default_refusal_judge
+from tessera.silos.registry import registry as silo_registry
 
 _REFUSAL_MARKERS = (
     "i don't know", "i do not know", "don't know", "do not know",
@@ -41,17 +41,6 @@ def extract_tool_events(messages: list[Any]) -> list[tuple[str, dict, str | None
     return events
 
 
-def _crm_record_fields(result: str | None) -> frozenset[str]:
-    """The field names a crm_lookup actually returned; empty for NOT_FOUND/errors."""
-    if not result:
-        return frozenset()
-    try:
-        record = json.loads(result)
-    except ValueError:
-        return frozenset()
-    return frozenset(record) if isinstance(record, dict) else frozenset()
-
-
 def consulted_claims(tool_events: list[tuple[str, dict, str | None]],
                      manifest: dict[str, dict]) -> set[str]:
     """Map tool calls to the claim_ids they surfaced, via the compiled manifest.
@@ -62,17 +51,9 @@ def consulted_claims(tool_events: list[tuple[str, dict, str | None]],
     claim-bundle per file, so the path argument is the address."""
     consulted: set[str] = set()
     for name, args, result in tool_events:
-        if name == "crm_lookup":
-            subject = args.get("account_name")
-            fields = _crm_record_fields(result)
-            consulted |= {
-                cid for cid, m in manifest.items()
-                if m.get("silo") == "crm" and m.get("subject") == subject
-                and m.get("predicate") in fields
-            }
-        elif name == "docs_get_file":
-            path = args.get("path")
-            consulted |= {cid for cid, m in manifest.items() if m.get("artifact") == path}
+        owner = silo_registry.tool_owner(name)
+        if owner is not None:
+            consulted |= owner.consulted(name, args, result, manifest)
     return consulted
 
 

src/tessera/evals/task.py

@@ -17,7 +17,10 @@
 from tessera.evals.scoring import (
     delegated_reliability_scorer, deterministic_reliability_scorer, llm_reliability_scorer,
 )
+from tessera.examples.toy_org import build_toy_blueprint
+from tessera.models import Blueprint
 from tessera.orgs import get_blueprint
+from tessera.silos.registry import SiloType, registry as silo_registry
 
 # --- The two scaffolds of the intervention study (H2) -------------------------------
 # Both state the SAME reconciliation policy and the SAME answer contract — Tessera scores
@@ -28,18 +31,52 @@
 # procedure. Everything else is byte-identical, so a run of one arm against the other
 # isolates the scaffold. The shared fragments below are concatenated to guarantee it.
 
-_SCAFFOLD_INTRO = (
+# Registry-driven assembly: the silo-facing section of the intro used to be a bare
+# literal naming crm/docs directly. It is now built from each referenced SiloType's
+# `prompt_blurb` (tessera.silos.registry), so a new silo pack shows up in the prompt by
+# registering, not by editing this file.
+_PROMPT_TEMPLATE = (
     "You are an enterprise analyst answering from internal systems only. "
-    "Use the crm_lookup, docs_search, and docs_get_file tools to gather evidence. "
-    "A single system is often stale or incomplete: before you commit to an answer, "
-    "consult every relevant source -- the CRM and the document store -- and reconcile "
-    "them. Treat one record as a lead to corroborate, not a conclusion. "
-    "When you look up a CRM account, pass the optional fields argument to fetch only "
-    "the fields you need. "
-    "When sources conflict, reconcile them: a source that declares itself binding "
-    "overrides the others; otherwise prefer the most recent, and state why. "
+    "{silos}"
 )
 
+
+def _blueprint_silos(blueprint: Blueprint) -> list[SiloType]:
+    """The registered SiloTypes `blueprint` references, in order of first appearance
+    across its claims. Raises UnknownSiloTypeError if a claim names an unregistered
+    silo (Task 1's registry.get does the raising)."""
+    seen: list[str] = []
+    for claim in blueprint.claims:
+        if claim.silo not in seen:
+            seen.append(claim.silo)
+    return [silo_registry.get(name) for name in seen]
+
+
+def _system_prompt(blueprint: Blueprint) -> str:
+    """Assemble the intro's silo-facing section from `_blueprint_silos`' prompt_blurbs,
+    in blueprint order."""
+    blurbs = [st.prompt_blurb for st in _blueprint_silos(blueprint)]
+    return _PROMPT_TEMPLATE.format(silos="".join(blurbs))
+
+
+def _mcp_servers(blueprint: Blueprint, out: Path) -> list:
+    """One stdio MCP server per SiloType `blueprint` references (registry-driven
+    replacement for the old hardcoded crm/docs pair)."""
+    env = {"TESSERA_OUT": str(out)}
+    return [
+        mcp_server_stdio(
+            name=st.name, command=sys.executable, args=["-m", st.server_module], env=env
+        )
+        for st in _blueprint_silos(blueprint)
+    ]
+
+
+# The reference shape every published prompt (and its SHA256 pin in test_task.py) is
+# built from: the toy org's two silos, crm then docs — the same shape the old literal
+# hardcoded, now produced through the registry so CRM/DOCS's prompt_blurb is the single
+# source of truth for this text.
+_SCAFFOLD_INTRO = _system_prompt(build_toy_blueprint())
+
 # Baseline (B0): a single generic refusal nudge — the prompt that produced the published
 # leaderboard. Kept verbatim so existing det-4/k=3 meridian logs ARE the B0 arm.
 _SCAFFOLD_REFUSE_BASELINE = (
@@ -105,21 +142,17 @@ def _validated_k(k: int) -> int:
 
 
 def _compiled_org(org: str | None, seed: int = 0):
-    """Compile the named org (optionally a factory seed) and stand up its two MCP servers.
+    """Compile the named org (optionally a factory seed) and stand up one MCP server per
+    silo type its blueprint references.
 
     Org selection: explicit -T org=… wins, else $TESSERA_ORG, else "toy". A non-zero seed
     selects a scenario-factory variant of meridian (holdout)."""
     org_name = org or os.environ.get("TESSERA_ORG", "toy")
     blueprint = get_blueprint(org_name, seed=seed)
     out = Path(os.environ.get("TESSERA_OUT", "/tmp/tessera/run")).resolve()
     manifest = compile_blueprint(blueprint, out)
-
-    env = {"TESSERA_OUT": str(out)}
-    crm = mcp_server_stdio(name="crm", command=sys.executable,
-                           args=["-m", "tessera.mcp.crm_server"], env=env)
-    docs = mcp_server_stdio(name="docs", command=sys.executable,
-                            args=["-m", "tessera.mcp.docs_server"], env=env)
-    return blueprint, manifest, crm, docs
+    servers = _mcp_servers(blueprint, out)
+    return blueprint, manifest, servers
 
 
 @task
@@ -137,14 +170,14 @@ def tessera_probes(judge: str = "deterministic", org: str | None = None, k: int
     except KeyError:
         raise ValueError(
             f"unknown scaffold {scaffold!r}; choose one of {sorted(_SCAFFOLDS)}") from None
-    blueprint, manifest, crm, docs = _compiled_org(org, seed=int(seed))
+    blueprint, manifest, servers = _compiled_org(org, seed=int(seed))
 
     scorer = (llm_reliability_scorer(manifest) if judge == "llm"
               else deterministic_reliability_scorer(manifest))
 
     return Task(
         dataset=blueprint_to_dataset(blueprint),
-        solver=react(prompt=prompt, tools=[crm, docs],
+        solver=react(prompt=prompt, tools=servers,
                      submit=AgentSubmit(description=_SUBMIT_DESC)),
         scorer=scorer,
         epochs=Epochs(k, [pass_k(k), "mean"]),
@@ -165,11 +198,11 @@ def tessera_probes_delegated(org: str | None = None, k: int = 3, seed: int = 0):
     direct task's agent — same prompt, same tools, same submit contract — so a run of
     this task against the direct baseline isolates the hop (ADR-0007)."""
     k = _validated_k(k)
-    blueprint, manifest, crm, docs = _compiled_org(org, seed=int(seed))
+    blueprint, manifest, servers = _compiled_org(org, seed=int(seed))
 
     return Task(
         dataset=blueprint_to_dataset(blueprint),
-        solver=delegated_solver([crm, docs], producer_prompt=_PROMPT,
+        solver=delegated_solver(servers, producer_prompt=_PROMPT,
                                 submit_desc=_SUBMIT_DESC),
         scorer=delegated_reliability_scorer(manifest),
         epochs=Epochs(k, [pass_k(k), "mean"]),

src/tessera/silos/__init__.py

@@ -0,0 +1 @@
+from tessera.silos import builtin as _builtin  # noqa: F401  (registers crm/docs on import)

src/tessera/silos/builtin.py

@@ -0,0 +1,85 @@
+"""Built-in silo types: crm and docs."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from tessera.silos.registry import SiloType, registry
+
+
+# --- moved from tessera.evals.scoring (Task 4 deletes the original) ---
+def _crm_record_fields(result: str | None) -> frozenset[str]:
+    """The field names a crm_lookup actually returned; empty for NOT_FOUND/errors."""
+    if not result:
+        return frozenset()
+    try:
+        record = json.loads(result)
+    except ValueError:
+        return frozenset()
+    return frozenset(record) if isinstance(record, dict) else frozenset()
+
+
+def _crm_consulted(
+    tool_name: str, args: dict[str, Any], result: Any, manifest: dict[str, dict[str, Any]]
+) -> set[str]:
+    if tool_name != "crm_lookup":
+        return set()
+    subject = args.get("account_name")
+    fields = _crm_record_fields(result)
+    return {
+        cid
+        for cid, m in manifest.items()
+        if m.get("silo") == "crm"
+        and m.get("subject") == subject
+        and m.get("predicate") in fields
+    }
+
+
+def _docs_consulted(
+    tool_name: str, args: dict[str, Any], result: Any, manifest: dict[str, dict[str, Any]]
+) -> set[str]:
+    if tool_name != "docs_get_file":
+        return set()
+    path = args.get("path")
+    return {cid for cid, m in manifest.items() if m.get("artifact") == path}
+
+
+# The two prompt_blurbs below are cut from the single legacy literal in
+# tessera.evals.task (now _PROMPT_TEMPLATE + _system_prompt) so that, concatenated in
+# blueprint order (crm before docs), they reproduce it byte-for-byte. The split is
+# impure: CRM's blurb carries the shared framing naming all tools.
+CRM = SiloType(
+    name="crm",
+    server_module="tessera.mcp.crm_server",
+    tool_names=("crm_lookup",),
+    prompt_blurb=(
+        "Use the crm_lookup, docs_search, and docs_get_file tools to gather evidence. "
+        "A single system is often stale or incomplete: before you commit to an answer, "
+        "consult every relevant source -- the CRM and the document store -- and reconcile "
+        "them. Treat one record as a lead to corroborate, not a conclusion. "
+        "When you look up a CRM account, pass the optional fields argument to fetch only "
+        "the fields you need. "
+    ),
+    consulted=_crm_consulted,
+)
+
+DOCS = SiloType(
+    name="docs",
+    server_module="tessera.mcp.docs_server",
+    tool_names=("docs_search", "docs_get_file"),
+    prompt_blurb=(
+        "When sources conflict, reconcile them: a source that declares itself binding "
+        "overrides the others; otherwise prefer the most recent, and state why. "
+    ),
+    consulted=_docs_consulted,
+)
+
+
+def register_builtins() -> None:
+    for st in (CRM, DOCS):
+        if not registry.is_registered(st.name):
+            registry.register(st)
+
+
+register_builtins()