Build interactive BF16 BoL operating room

README.md

@@ -1,13 +1,48 @@
 ---
-title: 0perating Soon
-emoji: ⚡
+title: 0perating-soon
+emoji: 🧬
 colorFrom: green
-colorTo: green
+colorTo: yellow
 sdk: gradio
 sdk_version: 6.26.0
-python_version: '3.12'
+python_version: "3.12"
 app_file: app.py
-pinned: false
+short_description: Open a language model. Inspect, intervene, compare.
+startup_duration_timeout: 1h
+preload_from_hub:
+  - HuggingFaceTB/SmolLM2-1.7B-Instruct
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 0perating-soon
+
+An interactive operating room for language models, built from Rick Holmberg's BoL tools.
+
+Choose a Hugging Face checkpoint, inspect its representations and weights, and measure what happens when selected parts are changed. The default model is **HuggingFaceTB/SmolLM2-1.7B-Instruct**. Execution uses **BF16**, with **FP32 numerical summaries**. Each report records the exact model revision, settings, elapsed GPU-function time, and peak allocated VRAM.
+
+## Instruments
+
+- Weight fingerprint, semantic silhouette, semantic clusters, layer CKA, attention maps, and gradient sensitivity.
+- Layer and component ablations, zone cuts, layer keep patterns, and controlled noise sensitivity.
+- **Layer Duets:** measure the interaction between paired ablations.
+- **Butterfly Effect:** compare hidden representations at aligned downstream tokens after a prompt edit.
+- **Context Tug-of-War:** compare candidate continuation scores under competing contexts.
+
+Outputs include interactive charts, measurement tables, JSON, and a self-contained printable HTML report. This app exports reports, not modified model weights.
+
+## Model support
+
+Use complete, unquantized Safetensors checkpoints supported by Transformers' `AutoModelForCausalLM`. Remote model code and pickle weights are not executed. Private or gated models require your own read token and model access. Intervention tools detect supported decoder layer layouts and fail explicitly when a requested structure is unavailable. Some tools require attention weights or hidden states that not every architecture provides.
+
+The app uses ZeroGPU's default 48 GB allocation. A model-size ceiling has not been established: the selected experiment, sequence length, and architecture all affect memory. Check the measured memory in your report. The default checkpoint is preloaded; other checkpoints download before acquiring a GPU and load temporarily inside the GPU worker. Custom-model cold loads therefore take longer. GPU time remains subject to the visitor's Hugging Face quota.
+
+## Interpretation
+
+Measurements describe the chosen checkpoint, probes, precision, and interventions. Semantic separation and attention patterns do not establish comprehension. CKA similarity alone does not establish that layers are removable. Gradient norms measure local sensitivity. Noise sensitivity is not an actual INT4/INT8 quantization evaluation. Ablations zero weight matrices while preserving norms and biases; they do not physically remove modules. Changed weights are restored, and failed restoration aborts the experiment.
+
+## Data contributions
+
+Research retention is opt-in through a separate contribution button after a report is displayed. Consented data is stored in a **private** Hugging Face dataset for BoL model-analysis research. By default a contribution includes model metadata, numerical settings, and numerical measurements. Custom text, token labels, and outputs require a second explicit permission. Read tokens are never included in reports or contributions. Contributions identify the run rather than the visitor's Hugging Face account.
+
+Uncontributed report files expire after about an hour. Checkpoints are cached for serving. Hosting providers may retain their own operational logs according to their policies.
+
+Maintainer configuration: set `BOL_CONTRIBUTION_REPO` to a private dataset and `BOL_CONTRIBUTION_TOKEN` as a Space secret with write access. These credentials are used only for consented contributions, never for visitor-selected model access.

app.py

@@ -0,0 +1,211 @@
+"""0perating-soon: a browser operating room for language models."""
+import os
+import tempfile
+
+os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
+os.environ.setdefault("GRADIO_TEMP_DIR", os.path.join(tempfile.gettempdir(), "bol-gradio"))
+import spaces  # Initialize ZeroGPU before any CUDA-touching imports.
+import gradio as gr
+
+from bol_space.runtime import DEFAULT_MODEL, DEFAULT_TICKET, analyze_gpu, model_summary, prepare_checkpoint, validate_options
+from bol_space.presentation import create_report, render_result
+from bol_space.contributions import save_contribution
+
+TOOLS = [
+    ("Weight fingerprint", "fingerprint"),
+    ("Semantic silhouette", "silhouette"),
+    ("Semantic probe", "semantic"),
+    ("Layer similarity · CKA", "cka"),
+    ("Attention map", "attention"),
+    ("Gradient sensitivity", "gradient"),
+    ("Layer sweep", "layer_sweep"),
+    ("Component ablation", "component"),
+    ("Trisect · layer zones", "trisect"),
+    ("Wormhole · layer patterns", "wormhole"),
+    ("Noise sensitivity", "noise"),
+    ("Layer Duets", "duets"),
+    ("Butterfly Effect", "butterfly"),
+    ("Context Tug-of-War", "tugofwar"),
+]
+DESCRIPTIONS = {
+    "fingerprint": "Inspect the distribution and scale of weight matrices throughout the model. Statistics describe the checkpoint loaded in BF16.",
+    "silhouette": "Follow the gap between related and unrelated word representations through the layers. Use the built-in groups or enter one comma-separated group per line, with at least two words in each of two groups.",
+    "semantic": "Explore which word groups sit together in the model's hidden representations.",
+    "cka": "Compare how sampled layers represent the same probes. Similar representations can still have different roles.",
+    "attention": "See how individual heads distribute attention over your text. Use a sentence or paragraph so heads have something to attend to.",
+    "gradient": "Measure how sensitive your text's next-token loss is to each layer's weights.",
+    "layer_sweep": "Zero one sampled layer's weight matrices at a time and measure the change in loss. Norms and biases are preserved; weights are restored after each intervention.",
+    "component": "Zero a projection type across selected layers and compare the result with the original model. Choose All available to discover the components this architecture exposes.",
+    "trisect": "Compare the model's inner, outer, and half-depth zones by zeroing the other zones' weight matrices.",
+    "wormhole": "Try alternating layers, outer layers, and other keep patterns. This zeros matrices in the unwanted layers; it does not physically remove the layers.",
+    "noise": "Add controlled random noise to sampled layers and measure sensitivity. This is a noise experiment, not a prediction of INT4 or INT8 quantization quality.",
+    "duets": "Find layer partnerships: compare individual ablations with paired ablations. Positive interaction means the pair caused more loss damage than the sum of the individual effects. Up to six layers are paired.",
+    "butterfly": "Change a small piece of text and watch the representations diverge at matching downstream tokens. Keep a shared ending in both texts for alignment.",
+    "tugofwar": "Choose two access codes. The experiment states each code as a fact, then tests both orders and repeats each fact in turn. Your text is a shared preamble before these four scenarios. Scores cover each complete candidate continuation.",
+}
+CSS = """
+.gradio-container {max-width:1440px !important; background:#f7f5ef; color:#22343e;}
+body {background:#f7f5ef;}
+#masthead {border-bottom:1px solid #cbd3ca; padding:20px 2px 26px; margin-bottom:18px;}
+#masthead .eyebrow {font:600 12px monospace; color:#56786a; letter-spacing:.18em;}
+#masthead h1 {font-size:clamp(35px,5vw,64px); line-height:1.02; letter-spacing:-.055em; color:#173a3b; margin:12px 0;}
+#masthead p {font-size:17px; color:#5b6869; max-width:680px; margin:0;}
+#run-button {background:#176d62 !important; border-color:#176d62 !important; color:white !important; min-height:48px;}
+#instrument-note {border-left:3px solid #d69a61; padding-left:14px; margin:10px 0 18px;}
+.block {border-radius:10px !important;}
+#results-panel {min-height:550px;}
+.quiet-note {font-size:12px; color:#687477;}
+footer {display:none !important;}
+"""
+
+
+def begin_run():
+    """Invalidate the prior result and permission before preparing another run."""
+    return (None, None, None, False, False, "", None,
+            "**Preparing your experiment…**", None, [], "", None, None,
+            gr.update(interactive=False), gr.update(interactive=False),
+            "Preparing the selected checkpoint…")
+
+
+def prepare_request(repo_id, revision, access_token, tool, text, compare_text, prompt, words,
+                    max_tokens, max_new_tokens, sample_layers, layer_start, layer_end,
+                    component, noise_scale, seed, choice_a, choice_b):
+    """Validate and download before GPU allocation; clear consent for API callers too."""
+    # The browser clears immediately through begin_run. Yielding a reset here
+    # also invalidates old server state when /prepare is called directly by API.
+    yield None, None, None, False, False, "", gr.update(interactive=False), "Preparing the selected checkpoint…"
+    try:
+        options = validate_options(tool, {"text": text, "compare_text": compare_text, "prompt": prompt,
+                    "words": words, "max_tokens": max_tokens, "max_new_tokens": max_new_tokens,
+                    "sample_layers": sample_layers, "layer_start": layer_start, "layer_end": layer_end,
+                    "component": component, "noise_scale": noise_scale, "seed": seed,
+                    "choice_a": choice_a, "choice_b": choice_b})
+        ticket = prepare_checkpoint(repo_id, revision, access_token)
+        yield ({"ticket": ticket, "tool": tool, "options": options}, None, None,
+               False, False, "", gr.update(interactive=False), model_summary(ticket))
+    except Exception as exc:
+        raise gr.Error(str(exc)) from exc
+
+
+def finish_report(result):
+    """Render measured output and create private, temporary report downloads."""
+    if not result or not result.get("run", {}).get("id"):
+        raise gr.Error("Run an experiment before creating a report. This session has no completed measurements.")
+    summary, figure, frame, comparisons = render_result(result)
+    json_path, html_path = create_report(result)
+    return (result, summary, figure, frame, comparisons, json_path, html_path,
+            result, False, False, "", gr.update(interactive=True), gr.update(interactive=True))
+
+
+def failed_run():
+    """Keep contributions unavailable after any failed stage and allow a retry."""
+    return (None, None, None, False, False, "", gr.update(interactive=False),
+            gr.update(interactive=True), "**The experiment did not finish.** Adjust its settings and try again.")
+
+
+def contribute(result, consent, include_text):
+    """Retain a completed report only after explicit permission."""
+    try:
+        return save_contribution(result, consent, include_text), False, False
+    except Exception as exc:
+        raise gr.Error(str(exc)) from exc
+
+
+def describe_tool(tool):
+    semantic = tool in {"silhouette", "semantic", "cka"}
+    return (DESCRIPTIONS[tool], gr.update(visible=tool == "butterfly"),
+            gr.update(visible=tool == "tugofwar"), gr.update(visible=semantic),
+            gr.update(visible=not semantic and tool != "fingerprint",
+                      label="Context preamble" if tool == "tugofwar" else "Text to analyze",
+                      info="Added before each access-code scenario." if tool == "tugofwar" else None))
+
+
+theme = gr.themes.Base(primary_hue="teal", secondary_hue="orange", neutral_hue="stone",
+                       font=[gr.themes.GoogleFont("DM Sans"), "Arial", "sans-serif"])
+with gr.Blocks(title="0perating-soon", delete_cache=(900, 3600)) as demo:
+    request_state = gr.State(value=None, time_to_live=3600)
+    computed_state = gr.State(value=None, time_to_live=3600)
+    report_state = gr.State(value=None, time_to_live=3600)
+    gr.HTML("""<div id="masthead"><div class="eyebrow">BOL TOOLS / MODEL OPERATING ROOM</div>
+    <h1>0perating-soon<span style="color:#d59a61">.</span></h1>
+    <p>Open a language model. Inspect its structure. Find out what changes when you intervene.</p></div>""")
+    with gr.Row(equal_height=False):
+        with gr.Column(scale=4, min_width=310):
+            gr.Markdown("### 01 / Admit a model")
+            repo = gr.Textbox(value=DEFAULT_MODEL, label="Hugging Face checkpoint", placeholder="organization/model-name")
+            with gr.Accordion("Revision & private model access", open=False):
+                revision = gr.Textbox(value="main", label="Revision", info="A branch, tag, or commit. Reports record the resolved commit.")
+                access_token = gr.Textbox(value="", type="password", label="Your Hugging Face read token (optional)", info="For private or gated models you can access. Used to download the checkpoint; excluded from reports and contributions.")
+            model_status = gr.Markdown(model_summary(DEFAULT_TICKET))
+            gr.Markdown("### 02 / Choose an experiment")
+            tool = gr.Dropdown(choices=TOOLS, value="silhouette", label="Instrument", filterable=True)
+            note = gr.Markdown(DESCRIPTIONS["silhouette"], elem_id="instrument-note")
+            text = gr.Textbox(value="The capital of France is Paris. Water freezes at zero degrees Celsius. The cat sat beside the window.", label="Text to analyze", lines=4, visible=False, max_length=10000)
+            compare_text = gr.Textbox(value="The capital of France is Lyon. Water freezes at zero degrees Celsius. The cat sat beside the window.", label="Comparison text", lines=4, visible=False, max_length=10000)
+            with gr.Row(visible=False) as candidates:
+                choice_a = gr.Textbox(value=" MAPLE", label="Candidate A", max_length=300)
+                choice_b = gr.Textbox(value=" RAVEN", label="Candidate B", max_length=300)
+            words = gr.Textbox(value="", label="Word probes (optional)", info="One comma-separated group per line. Silhouette needs at least two groups with two words each. Leave empty for the built-in groups.", lines=3, visible=True, max_length=10000)
+            with gr.Accordion("Experiment settings", open=False):
+                max_tokens = gr.Slider(32, 256, value=128, step=16, label="Input token limit")
+                sample_layers = gr.Slider(2, 16, value=8, step=1, label="Sampled layers", info="Duets uses at most six. Zone and pattern experiments use the full selected range.")
+                with gr.Row():
+                    layer_start = gr.Number(value=0, precision=0, minimum=0, label="First layer")
+                    layer_end = gr.Number(value=-1, precision=0, minimum=-1, label="Last layer (-1 = final)")
+                component = gr.Textbox(value="All available", label="Component", info="Run All available first; then paste a component name from its report.")
+                noise_scale = gr.Slider(0, 0.5, value=0.01, step=0.005, label="Noise strength")
+                seed = gr.Number(value=42, precision=0, minimum=0, label="Random seed")
+                prompt = gr.Textbox(value="The capital of France is", label="Generation probe", max_length=10000)
+                max_new_tokens = gr.Slider(1, 40, value=16, step=1, label="Generated tokens")
+            run_button = gr.Button("Run experiment →", variant="primary", elem_id="run-button")
+            gr.Markdown("BF16 checkpoints · FP32 statistics · on-demand ZeroGPU\n\nThe first run of a new model includes its download. GPU time and memory are measured in every report.", elem_classes=["quiet-note"])
+
+        with gr.Column(scale=7, min_width=480, elem_id="results-panel"):
+            gr.Markdown("### 03 / Read the evidence")
+            summary = gr.Markdown("**Your model, under the instruments.**\n\nStart with Semantic silhouette for a quick look through its layers, or choose Layer Duets to test how layers work together.")
+            chart = gr.Plot(label="Measured structure")
+            with gr.Tabs():
+                with gr.Tab("Measurements"):
+                    table = gr.Dataframe(interactive=False, label="Results", wrap=True)
+                with gr.Tab("Before / after"):
+                    comparisons = gr.Markdown("Text comparisons appear here for experiments that generate or score continuations.")
+                with gr.Tab("Raw report"):
+                    result_json = gr.JSON(label="Complete measurement record")
+            gr.Markdown("### 04 / Take your report")
+            with gr.Row():
+                json_report = gr.File(label="Download JSON", interactive=False)
+                html_report = gr.File(label="Download printable HTML", interactive=False)
+            gr.Markdown("The HTML report opens offline and can be printed or saved as a PDF from your browser.", elem_classes=["quiet-note"])
+            with gr.Accordion("Contribute this run to BoL", open=False):
+                gr.Markdown("Optional: help compare model structure across checkpoints. Contributions are stored **privately for BoL research**. Your report is available whether you contribute or not.")
+                consent = gr.Checkbox(value=False, label="Allow BoL to keep this run's checkpoint ID and revision, settings, and numerical measurements (including private checkpoint IDs).")
+                include_text = gr.Checkbox(value=False, label="Also include my custom text, token labels, and generated outputs.")
+                gr.Markdown("Without the second permission, text fields and token labels are excluded. Uncontributed reports are temporary and expire after about one hour.", elem_classes=["quiet-note"])
+                save = gr.Button("Contribute this report", variant="secondary", interactive=False)
+                contribution_status = gr.Markdown()
+
+    controls = [repo, revision, access_token, tool, text, compare_text, prompt, words, max_tokens,
+                max_new_tokens, sample_layers, layer_start, layer_end, component, noise_scale, seed, choice_a, choice_b]
+    tool.change(describe_tool, inputs=tool, outputs=[note, compare_text, candidates, words, text], queue=False, api_visibility="private")
+    started = run_button.click(begin_run, inputs=None,
+                                outputs=[request_state, computed_state, report_state, consent, include_text, contribution_status,
+                                         result_json, summary, chart, table, comparisons, json_report, html_report, save, run_button, model_status],
+                                queue=False, api_visibility="private")
+    prepared = started.success(prepare_request, inputs=controls,
+                                outputs=[request_state, computed_state, report_state, consent, include_text, contribution_status, save, model_status],
+                                api_name="prepare", concurrency_id="model_experiment", concurrency_limit=1)
+    computed = prepared.success(analyze_gpu, inputs=request_state, outputs=computed_state,
+                                 api_name="analyze", concurrency_id="model_experiment", concurrency_limit=1)
+    reported = computed.success(finish_report, inputs=computed_state,
+                     outputs=[result_json, summary, chart, table, comparisons, json_report, html_report, report_state, consent, include_text, contribution_status, save, run_button],
+                     api_name="report", concurrency_id="model_experiment", concurrency_limit=1)
+    for stage in (prepared, computed, reported):
+        stage.failure(failed_run, inputs=None,
+                      outputs=[request_state, computed_state, report_state, consent, include_text, contribution_status, save, run_button, summary],
+                      queue=False, api_visibility="private")
+    save.click(contribute, inputs=[report_state, consent, include_text], outputs=[contribution_status, consent, include_text],
+               api_name="contribute", concurrency_limit=1)
+    demo.queue(max_size=24, default_concurrency_limit=1)
+
+if __name__ == "__main__":
+    demo.launch(theme=theme, css=CSS, show_error=True)

bol_space/__init__.py

@@ -0,0 +1 @@
+"""Interactive BoL diagnostics for Hugging Face Spaces."""

bol_space/checkpoints.py

@@ -0,0 +1,55 @@
+"""Shared checkpoint loading for the Space and standalone BoL commands.
+
+Importing this module does not load a checkpoint. The caller initializes spaces
+before importing it when running in ZeroGPU.
+"""
+from __future__ import annotations
+
+import torch
+from huggingface_hub import HfApi, snapshot_download
+from huggingface_hub.utils import validate_repo_id
+from transformers import AutoConfig, AutoModelForCausalLM, AutoTokenizer
+
+DEFAULT_MODEL = "HuggingFaceTB/SmolLM2-1.7B-Instruct"
+
+
+def prepare_checkpoint(repo_id: str, revision: str = "main", access_token: str = "") -> dict:
+    """Resolve access and download only data files before entering the GPU queue."""
+    repo_id, revision = repo_id.strip(), revision.strip() or "main"
+    validate_repo_id(repo_id)
+    token = access_token.strip() or False
+    try:
+        info = HfApi(token=token).model_info(repo_id, revision=revision, files_metadata=True)
+    except Exception as exc:
+        raise ValueError("Cannot access that checkpoint. Check the model ID/revision and, for gated or private models, your read token and model access.") from exc
+    filenames = [item.rfilename for item in info.siblings if "/" not in item.rfilename]
+    weights = [name for name in filenames if name.endswith(".safetensors") and not name.startswith(("optimizer", "adapter_"))]
+    if not weights:
+        raise ValueError("Choose a complete, unquantized Hugging Face checkpoint with Safetensors weights. GGUF, adapter-only, and pickle checkpoints are not supported.")
+    # Reject incompatible formats before downloading their weights.
+    config = AutoConfig.from_pretrained(repo_id, revision=info.sha, token=token, trust_remote_code=False)
+    if getattr(config, "quantization_config", None):
+        raise ValueError("This checkpoint is quantized. Use its original BF16/FP16 Hugging Face checkpoint for weight interventions and gradients.")
+    data_files = [name for name in filenames if name.endswith((".json", ".model", ".tiktoken")) or name in {"merges.txt", "vocab.txt"}]
+    # Pin all files to the resolved revision, including tokenizer and config.
+    path = snapshot_download(repo_id, revision=info.sha, token=token, allow_patterns=weights + data_files)
+    count = getattr(info.safetensors, "total", None) if info.safetensors else None
+    return {"repo_id": repo_id, "revision": info.sha, "path": path, "architecture": config.model_type,
+            "parameters": count, "dtype": "bfloat16", "private": bool(info.private)}
+
+
+def load_checkpoint(ticket: dict):
+    """Use native floating-point modules so inspection and surgery remain available."""
+    tokenizer = AutoTokenizer.from_pretrained(ticket["path"], trust_remote_code=False, local_files_only=True)
+    if tokenizer.pad_token_id is None:
+        if tokenizer.eos_token_id is None:
+            raise ValueError("This tokenizer needs a padding or EOS token for batched analysis.")
+        tokenizer.pad_token = tokenizer.eos_token
+    model = AutoModelForCausalLM.from_pretrained(
+        ticket["path"], dtype=torch.bfloat16, trust_remote_code=False,
+        use_safetensors=True, local_files_only=True, attn_implementation="eager",
+    ).eval()
+    device = "cuda" if torch.cuda.is_available() else "cpu"
+    return model.to(device), tokenizer
+
+

bol_space/contributions.py

@@ -0,0 +1,30 @@
+"""Explicitly consented research contributions, kept in a private Hub dataset."""
+from datetime import datetime, timezone
+import json
+import os
+import uuid
+
+from huggingface_hub import HfApi
+from bol_space.presentation import contribution_payload
+
+
+def save_contribution(result: dict | None, consent: bool, include_text: bool) -> str:
+    """Save a completed run only after the user explicitly consents."""
+    if consent is not True:
+        raise ValueError("Check the contribution permission box first. Your report remains available without contributing.")
+    if not result or not result.get("run", {}).get("id"):
+        raise ValueError("Run an experiment before contributing a report.")
+    repo = os.environ.get("BOL_CONTRIBUTION_REPO", "")
+    token = os.environ.get("BOL_CONTRIBUTION_TOKEN", "")
+    if not repo or not token:
+        raise ValueError("Contributions are not configured yet. You can still download your report.")
+    api = HfApi(token=token)
+    if not api.dataset_info(repo).private:
+        raise ValueError("Contribution storage is not private. No report was saved.")
+    payload = contribution_payload(result, include_text=include_text is True)
+    payload["consent"] = {"version": "1", "measurements": True, "include_text": include_text is True,
+                          "timestamp": datetime.now(timezone.utc).isoformat(), "purpose": "Private BoL model-analysis research"}
+    content = json.dumps(payload, allow_nan=False, ensure_ascii=False).encode("utf-8")
+    contribution_id = uuid.uuid4().hex
+    api.upload_file(path_or_fileobj=content, path_in_repo=f"runs/{contribution_id}.json", repo_id=repo, repo_type="dataset", commit_message="Add a consented BoL research contribution")
+    return f"Saved privately for BoL research. Contribution ID: {contribution_id}. " + ("Your custom text and outputs were included." if include_text else "Custom text, token labels, and generated outputs were excluded.")

bol_space/inspection.py

@@ -0,0 +1,578 @@
+"""Request-local BoL measurements on an already loaded causal language model.
+
+No checkpoint loading, persistence, generation, or global model state lives here.
+The caller owns device placement, BF16 loading, request serialization and eval mode.
+Statistics use FP32; casting BF16 values does not recover an FP32 checkpoint.
+"""
+from __future__ import annotations
+
+from difflib import SequenceMatcher
+import math
+import re
+
+import torch
+import torch.nn.functional as F
+
+
+DEFAULT_TEXT = (
+    "The quick brown fox crossed the river and stopped beside an old oak tree. "
+    "The water reflected the evening sky. A small bird watched from a branch."
+)
+DEFAULT_GROUPS = (
+    ("canines", ("dog", "wolf", "puppy")),
+    ("felines", ("cat", "kitten", "tiger")),
+    ("vehicles", ("car", "truck", "vehicle")),
+    ("food", ("banana", "apple", "fruit")),
+    ("emotions", ("happy", "joy", "sadness")),
+    ("tech", ("code", "python", "programming")),
+    ("abstract", ("silence", "peace", "war")),
+)
+RELATED_PAIRS = (
+    ("dog", "wolf"), ("cat", "lion"), ("river", "ocean"),
+    ("happy", "joyful"), ("cold", "frozen"), ("king", "queen"),
+    ("python", "code"), ("car", "vehicle"), ("run", "sprint"),
+)
+UNRELATED_PAIRS = (
+    ("dog", "democracy"), ("cat", "algebra"), ("river", "justice"),
+    ("happy", "concrete"), ("cold", "philosophy"), ("king", "equation"),
+    ("python", "banana"), ("car", "silence"), ("run", "orange"),
+)
+
+
+def _number(value):
+    value = float(value)
+    return value if math.isfinite(value) else None
+
+
+def _device(model):
+    return model.get_input_embeddings().weight.device
+
+
+def _decoder_layers(model):
+    for path in ("model.layers", "transformer.h", "gpt_neox.layers", "model.decoder.layers",
+                 "transformer.blocks", "model.transformer.h"):
+        module = model
+        for name in path.split("."):
+            module = getattr(module, name, None)
+            if module is None:
+                break
+        if isinstance(module, torch.nn.ModuleList) and len(module):
+            return module
+    raise ValueError(
+        "This tool needs a recognized decoder layer list. Supported layouts include "
+        "Llama/SmolLM, GPT-2, GPT-NeoX and OPT. This model's layout is not supported yet."
+    )
+
+
+def _selection(model, options):
+    layers = _decoder_layers(model)
+    start = int(options.get("layer_start") or 0)
+    end_value = options.get("layer_end")
+    end = len(layers) - 1 if end_value is None or int(end_value) < 0 else int(end_value)
+    if start < 0 or start >= len(layers) or end < start or end >= len(layers):
+        raise ValueError(f"Choose decoder layers between 0 and {len(layers) - 1} (inclusive).")
+    count = min(16, max(1, int(options.get("sample_layers", 8))), end - start + 1)
+    indices = [start] if count == 1 else [round(start + i * (end - start) / (count - 1))
+                                         for i in range(count)]
+    return layers, indices
+
+
+def _limit(options):
+    return min(256, max(8, int(options.get("max_tokens", 128))))
+
+
+def _inputs(model, tokenizer, text, limit):
+    if not str(text).strip():
+        raise ValueError("Enter some text to inspect.")
+    encoded = tokenizer(str(text), return_tensors="pt", truncation=True, max_length=limit)
+    result = {k: v.to(_device(model)) for k, v in encoded.items()
+              if k in ("input_ids", "attention_mask")}
+    if "attention_mask" not in result:
+        result["attention_mask"] = torch.ones_like(result["input_ids"])
+    if not result["input_ids"].numel():
+        raise ValueError("The tokenizer returned no tokens for this input.")
+    return result
+
+
+def _forward(model, inputs, **kwargs):
+    return model(**inputs, use_cache=False, return_dict=True, **kwargs)
+
+
+def _states(output, selected, count):
+    hidden = getattr(output, "hidden_states", None)
+    if hidden is None or len(hidden) != count + 1:
+        raise ValueError("This model did not return one hidden state per decoder layer plus embeddings.")
+    return {i: hidden[i + 1] for i in selected}
+
+
+def _plot(kind, title, x_label, y_label, **data):
+    return {"kind": kind, "title": title, "x_label": x_label, "y_label": y_label, **data}
+
+
+def _result(title, summary, rows, plot, details=None, warnings=None, text_outputs=None):
+    return {"title": title, "summary": summary, "rows": rows, "plot": plot,
+            "details": details or {}, "warnings": warnings or [], "text_outputs": text_outputs or []}
+
+
+def _line(title, rows, metrics, x="layer"):
+    return _plot("line", title, "Decoder layer (zero based)", "Measurement", series=[
+        {"name": name, "x": [row[x] for row in rows], "y": [row[metric] for row in rows]}
+        for name, metric in metrics
+    ])
+
+
+def _fingerprint(model, tokenizer, options):
+    layers, selected = _selection(model, options)
+    rows = []
+    layer_rows = []
+    seed = int(options.get("seed", 42))
+    generator = torch.Generator(device="cpu").manual_seed(seed)
+    with torch.no_grad():
+        for index in selected:
+            total_squared = 0.
+            elements = 0
+            for name, parameter in layers[index].named_parameters():
+                if parameter.ndim != 2:
+                    continue
+                values = parameter.detach().float()
+                norm = values.norm().item()
+                # Bounded power iteration: an estimate, never an exact SVD claim.
+                vector = torch.randn(values.shape[1], generator=generator).to(values.device)
+                vector = F.normalize(vector, dim=0)
+                for _ in range(8):
+                    left = F.normalize(values @ vector, dim=0)
+                    vector = F.normalize(values.T @ left, dim=0)
+                spectral = (values @ vector).norm().item()
+                rows.append({
+                    "layer": index, "tensor": name, "shape": list(parameter.shape),
+                    "mean": _number(values.mean()), "std": _number(values.std(unbiased=False)),
+                    "abs_mean": _number(values.abs().mean()), "minimum": _number(values.min()),
+                    "maximum": _number(values.max()), "frobenius": _number(norm),
+                    "sparsity": _number((values.abs() < 1e-6).float().mean()),
+                    "spectral_estimate": _number(spectral),
+                })
+                total_squared += norm * norm
+                elements += values.numel()
+                del values, vector, left
+            if elements:
+                layer_rows.append({"layer": index, "weight_rms": math.sqrt(total_squared / elements)})
+    if not rows:
+        raise ValueError("No dense 2D weight matrices were found in the selected decoder layers.")
+    return _result(
+        "Weight fingerprint", f"Measured {len(rows)} matrices across {len(selected)} selected decoder layers.",
+        rows, _line("Weight scale through the model", layer_rows, [("Weight RMS", "weight_rms")]),
+        {"selected_layers": selected, "layer_rms": layer_rows, "spectral_power_iterations": 8,
+         "sparsity_threshold": 1e-6, "checkpoint_values": "Loaded model values, converted to FP32 for statistics"},
+        ["Spectral norms use eight power iterations and are estimates. Condition numbers are not estimated.",
+         "A BF16 fingerprint describes rounded BF16 weights, including when the source checkpoint was FP32."],
+    )
+
+
+def _groups(options):
+    raw = str(options.get("words") or "").strip()
+    if not raw:
+        return [(name, list(words)) for name, words in DEFAULT_GROUPS], False
+    lines = [line.strip() for line in raw.splitlines() if line.strip()]
+    groups = []
+    if len(lines) >= 2 and all("," in line for line in lines):
+        for index, line in enumerate(lines):
+            groups.append((f"Group {index + 1}", [w.strip() for w in line.split(",") if w.strip()]))
+    else:
+        groups = [("Custom words", [w.strip() for w in re.split(r"[,\n]", raw) if w.strip()])]
+    words = [word for _, group in groups for word in group]
+    if len(words) < 3 or len(words) > 32:
+        raise ValueError("Use 3–32 words or short phrases. For groups, put comma-separated words on separate lines.")
+    if len(set(words)) != len(words):
+        raise ValueError("Each probe word or phrase must appear only once.")
+    if any(len(word) > 160 for word in words):
+        raise ValueError("Keep each probe word or phrase below 160 characters.")
+    return groups, True
+
+
+def _word_vectors(model, tokenizer, options, probe_words=None):
+    layers, selected = _selection(model, options)
+    if probe_words is None:
+        groups, custom = _groups(options)
+        words = [word for _, group in groups for word in group]
+    else:
+        groups, custom, words = [], False, list(probe_words)
+    vectors = {index: [] for index in selected}
+    with torch.no_grad():
+        for word in words:
+            inputs = _inputs(model, tokenizer, word, min(32, _limit(options)))
+            out = _forward(model, inputs, output_hidden_states=True)
+            last = int(inputs["attention_mask"][0].nonzero()[-1])
+            for index, state in _states(out, selected, len(layers)).items():
+                vectors[index].append(state[0, last].detach().float().cpu())
+            del out
+    return {index: torch.stack(items) for index, items in vectors.items()}, groups, words, custom
+
+
+def _paired_silhouette(model, tokenizer, options):
+    """Preserve the original BoL silhouette's explicitly specified word pairs."""
+    words = list(dict.fromkeys(word for pair in RELATED_PAIRS + UNRELATED_PAIRS for word in pair))
+    vectors, _, words, _ = _word_vectors(model, tokenizer, options, probe_words=words)
+    positions = {word: index for index, word in enumerate(words)}
+    rows = []
+    for layer, states in vectors.items():
+        normalized = F.normalize(states, dim=-1)
+        means = []
+        for pairs in (RELATED_PAIRS, UNRELATED_PAIRS):
+            similarities = [(normalized[positions[a]] * normalized[positions[b]]).sum().item()
+                            for a, b in pairs]
+            means.append(sum(similarities) / len(similarities))
+        rows.append({"layer": layer, "within_cosine": means[0], "cross_cosine": means[1],
+                     "separation": means[0] - means[1]})
+    return _result(
+        "Semantic silhouette",
+        f"Compared the original nine related and nine unrelated word pairs at {len(vectors)} selected layers.",
+        rows, _line("Related versus unrelated word-pair geometry", rows,
+                    [("Related", "within_cosine"), ("Unrelated", "cross_cosine"), ("Gap", "separation")]),
+        {"selected_layers": list(vectors), "custom_probes": False, "words": words,
+         "probe_contract": "Original BoL silhouette: mean cosine of nine specified related pairs minus mean cosine of nine specified unrelated pairs",
+         "related_pairs": [list(pair) for pair in RELATED_PAIRS],
+         "unrelated_pairs": [list(pair) for pair in UNRELATED_PAIRS],
+         "pooling": "Last non-padding token; each word processed in isolation",
+         "row_metrics": {"within_cosine": "Mean specified related-pair cosine", "cross_cosine": "Mean specified unrelated-pair cosine"}},
+        ["This preserves the original word-pair measurement with BF16 model execution and FP32 statistics.",
+         "Cosine separation is a descriptive gap, not a standard silhouette coefficient, comprehension test or causal importance score.",
+         "Probe pairs share words and are not independent. No significance test is reported.",
+         "The last returned hidden state may include the model's final normalization."],
+    )
+
+
+def _semantic(model, tokenizer, options, silhouette=False):
+    if silhouette and not str(options.get("words") or "").strip():
+        return _paired_silhouette(model, tokenizer, options)
+    vectors, groups, words, custom = _word_vectors(model, tokenizer, options)
+    membership = [name for name, group in groups for _ in group]
+    comparable = len(groups) >= 2 and all(len(group) >= 2 for _, group in groups)
+    if silhouette and not comparable:
+        raise ValueError("Silhouette needs at least two groups of two words. Put each comma-separated group on its own line.")
+    matrices = {index: F.normalize(values, dim=-1) @ F.normalize(values, dim=-1).T
+                for index, values in vectors.items()}
+    rows = []
+    for index, similarity in matrices.items():
+        if comparable:
+            within, cross = [], []
+            for a in range(len(words)):
+                for b in range(a + 1, len(words)):
+                    (within if membership[a] == membership[b] else cross).append(float(similarity[a, b]))
+            w, c = sum(within) / len(within), sum(cross) / len(cross)
+            rows.append({"layer": index, "within_cosine": w, "cross_cosine": c, "separation": w - c})
+    display = max(matrices)
+    if not comparable:
+        rows = [{"layer": display, "word_a": words[a], "word_b": words[b],
+                 "cosine": float(matrices[display][a, b])}
+                for a in range(len(words)) for b in range(a + 1, len(words))]
+    plot = (_line("Related versus cross-group geometry", rows,
+                  [("Within group", "within_cosine"), ("Across groups", "cross_cosine"), ("Gap", "separation")])
+            if silhouette else _plot("heatmap", f"Word cosine similarity · layer {display}", "Word", "Word",
+                                     x=words, y=words, z=matrices[display].tolist()))
+    return _result(
+        "Semantic silhouette" if silhouette else "Semantic probe",
+        f"Compared final-token representations of {len(words)} isolated probes at {len(vectors)} layers.",
+        rows, plot,
+        {"selected_layers": list(vectors), "groups": dict(groups), "display_layer": display,
+         "custom_probes": custom, "pooling": "Last non-padding token; each phrase processed in isolation",
+         "probe_contract": "Mean within-group cosine minus mean cross-group cosine" if comparable else "Ungrouped pairwise cosine similarities",
+         "similarity_matrix": matrices[display].tolist()},
+        ["Cosine separation is a descriptive gap, not a standard silhouette coefficient, comprehension test or causal importance score.",
+         "Probe pairs share words and are not independent. No significance test is reported.",
+         "The last returned hidden state may include the model's final normalization."],
+    )
+
+
+def _linear_cka(x, y):
+    """Biased linear CKA on aligned samples; None for degenerate representations."""
+    if x.shape[0] != y.shape[0] or x.shape[0] < 3:
+        raise ValueError("CKA needs at least three aligned samples.")
+    x, y = x.float(), y.float()
+    x, y = x - x.mean(0, keepdim=True), y - y.mean(0, keepdim=True)
+    # Remove arbitrary activation scale before Gram products. An absolute
+    # denominator cutoff would incorrectly call small but varying states null.
+    scale_x, scale_y = x.abs().max(), y.abs().max()
+    if scale_x.item() == 0 or scale_y.item() == 0:
+        return None
+    x, y = x / scale_x, y / scale_y
+    k, l = x @ x.T, y @ y.T
+    denom = k.norm() * l.norm()
+    if not torch.isfinite(denom) or denom.item() == 0:
+        return None
+    return min(1., max(0., float((k * l).sum() / denom)))
+
+
+def _cka(model, tokenizer, options):
+    vectors, groups, words, custom = _word_vectors(model, tokenizer, options)
+    indices = list(vectors)
+    matrix = [[_linear_cka(vectors[a], vectors[b]) for b in indices] for a in indices]
+    return _result(
+        "Layer CKA", f"Compared representation geometry across {len(indices)} layers using {len(words)} aligned probes.",
+        [{"layer_a": a, "layer_b": b, "cka": matrix[i][j]}
+         for i, a in enumerate(indices) for j, b in enumerate(indices)],
+        _plot("heatmap", "Linear centered kernel alignment", "Decoder layer", "Decoder layer",
+              x=indices, y=indices, z=matrix),
+        {"words": words, "selected_layers": indices, "custom_probes": custom,
+         "estimator": "Biased linear CKA of centered sample Gram matrices", "sample_count": len(words)},
+        ["CKA depends on this small probe set; it does not show whether a layer is dispensable.",
+         "Null entries mean a representation has no measurable centered variation.",
+         "All selected layer representations are extracted from the same forward pass for each probe."],
+    )
+
+
+def _attention(model, tokenizer, options):
+    layers, selected = _selection(model, options)
+    inputs = _inputs(model, tokenizer, options.get("text") or DEFAULT_TEXT, _limit(options))
+    length = inputs["input_ids"].shape[1]
+    if length < 3:
+        raise ValueError("Attention needs a phrase or sentence with at least three tokens.")
+    with torch.no_grad():
+        out = _forward(model, inputs, output_attentions=True)
+        attentions = getattr(out, "attentions", None)
+        if attentions is None or len(attentions) != len(layers):
+            raise ValueError("This model did not expose layer attention weights. Use an eager-attention transformer checkpoint.")
+        rows, heatmap, last_maps = [], [], {}
+        for index in selected:
+            if attentions[index] is None:
+                raise ValueError("This layer does not expose attention weights; hybrid/SSM attention is not supported here.")
+            p = attentions[index][0].detach().float()
+            if p.shape[-2:] != (length, length):
+                raise ValueError("This model returned a nonstandard attention layout.")
+            p = p / p.sum(-1, keepdim=True).clamp_min(1e-12)
+            entropies = -(p * p.clamp_min(1e-12).log()).sum(-1)
+            valid = inputs["attention_mask"][0].bool()
+            # Each query has its own count of causal, non-padding key positions.
+            available = valid.long().cumsum(0).float()
+            normalizers = available.clamp_min(2).log()
+            use_queries = valid & (available > 1)
+            normalized = entropies[:, use_queries] / normalizers[use_queries]
+            last = int(valid.nonzero()[-1])
+            patterns = p[:, last, :]
+            cosines = F.normalize(patterns, dim=-1) @ F.normalize(patterns, dim=-1).T
+            offdiag = ~torch.eye(p.shape[0], dtype=torch.bool, device=p.device)
+            diversity = float((1 - cosines[offdiag]).mean()) if p.shape[0] > 1 else 0.
+            head_values = []
+            for head in range(p.shape[0]):
+                value = float(normalized[head].mean())
+                rows.append({"layer": index, "head": head, "entropy": float(entropies[head, use_queries].mean()),
+                             "normalized_entropy": value, "head_diversity": diversity,
+                             "last_token_entropy": float(entropies[head, last]),
+                             "last_token_normalized_entropy": float(entropies[head, last] / normalizers[last])})
+                head_values.append(value)
+            heatmap.append(head_values)
+            last_maps[str(index)] = patterns.cpu().tolist()
+    return _result(
+        "Attention map", f"Measured {len(selected)} layers over {length} tokens, with a separate distribution for every head.",
+        rows, _plot("heatmap", "Normalized causal attention entropy", "Head", "Decoder layer",
+                    x=list(range(len(heatmap[0]))), y=selected, z=heatmap),
+        {"selected_layers": selected, "token_count": length,
+         "tokens": tokenizer.convert_ids_to_tokens(inputs["input_ids"][0].tolist()),
+         "last_token_attention": last_maps,
+         "normalization": "Entropy / log(causal non-padding key count), averaged over queries with >1 key"},
+        ["Low entropy means concentrated attention; it does not establish understanding or head importance.",
+         "Head diversity is the mean pairwise cosine distance of last-token attention distributions.",
+         "For sliding-window or other restricted masks, the denominator is the full causal history, so entropy is relative to that history."],
+    )
+
+
+def _gradient(model, tokenizer, options):
+    layers, selected = _selection(model, options)
+    inputs = _inputs(model, tokenizer, options.get("text") or DEFAULT_TEXT, _limit(options))
+    if int(inputs["attention_mask"].sum()) < 2:
+        raise ValueError("Gradient sensitivity requires at least two tokens.")
+    labels = inputs["input_ids"].clone()
+    labels[inputs["attention_mask"] == 0] = -100
+    parameters = list(model.parameters())
+    flags = [p.requires_grad for p in parameters]
+    chosen = [(index, name, p) for index in selected for name, p in layers[index].named_parameters()
+              if p.ndim == 2 and p.is_floating_point()]
+    if not chosen:
+        raise ValueError("No dense floating-point matrix parameters were found in these layers.")
+    chosen_ids = {id(p) for _, _, p in chosen}
+    try:
+        for p in parameters:
+            p.requires_grad_(id(p) in chosen_ids)
+        with torch.inference_mode(False), torch.enable_grad():
+            # Rebuild tensors in case an outer caller created them in inference mode.
+            grad_inputs = {k: value.clone() for k, value in inputs.items()}
+            outputs = _forward(model, grad_inputs, labels=labels.clone())
+            loss = outputs.loss
+            if loss is None or not torch.isfinite(loss):
+                raise ValueError("The model did not produce a finite causal language-model loss.")
+            gradients = torch.autograd.grad(loss, [p for _, _, p in chosen], allow_unused=True)
+            loss_value = float(loss.detach())
+            aggregates = {index: [0., 0., 0., 0] for index in selected}
+            components = []
+            for (index, name, p), grad in zip(chosen, gradients):
+                if grad is None:
+                    continue
+                g = grad.detach().float()
+                grad_sq, param_sq = float(g.square().sum()), float(p.detach().float().square().sum())
+                state = aggregates[index]
+                state[0] += grad_sq
+                state[1] += param_sq
+                state[2] += float(g.abs().sum())
+                state[3] += g.numel()
+                components.append({"layer": index, "tensor": name, "gradient_norm": math.sqrt(grad_sq)})
+            del gradients, outputs, loss
+        rows = []
+        for index, (g2, p2, abs_sum, count) in aggregates.items():
+            if count:
+                rows.append({"layer": index, "gradient_norm": math.sqrt(g2), "gradient_mean": abs_sum / count,
+                             "parameter_norm": math.sqrt(p2),
+                             "gradient_to_param_ratio": math.sqrt(g2) / max(math.sqrt(p2), 1e-12)})
+        if not rows:
+            raise ValueError("The selected matrices were not connected to this model's returned loss.")
+        return _result(
+            "Gradient sensitivity", f"Measured local parameter sensitivity at a text loss of {loss_value:.4f} nats per token.",
+            rows, _line("Gradient relative to weight scale", rows, [("Gradient / parameter norm", "gradient_to_param_ratio")]),
+            {"loss": loss_value, "selected_layers": selected, "components": components,
+             "objective": "Mean next-token cross entropy; padding excluded by attention_mask"},
+            ["Gradients describe this text and parameterization. They are not causal ablation scores or Fisher information.",
+             "Only selected layers' 2D parameters are differentiated; BF16 backward values are summarized in FP32."],
+        )
+    finally:
+        for p, flag in zip(parameters, flags):
+            p.requires_grad_(flag)
+
+
+def _butterfly(model, tokenizer, options):
+    layers, selected = _selection(model, options)
+    a = str(options.get("text") or DEFAULT_TEXT)
+    b = str(options.get("compare_text") or a.replace("fox", "wolf", 1))
+    inputs_a = _inputs(model, tokenizer, a, _limit(options))
+    inputs_b = _inputs(model, tokenizer, b, _limit(options))
+    ids_a, ids_b = inputs_a["input_ids"][0].tolist(), inputs_b["input_ids"][0].tolist()
+    prefix = 0
+    while prefix < min(len(ids_a), len(ids_b)) and ids_a[prefix] == ids_b[prefix]:
+        prefix += 1
+    identical = ids_a == ids_b
+    pairs = [(block.a + offset, block.b + offset)
+             for block in SequenceMatcher(a=ids_a, b=ids_b, autojunk=False).get_matching_blocks()
+             for offset in range(block.size)
+             if identical or (block.a + offset >= prefix and block.b + offset >= prefix)]
+    if not pairs:
+        raise ValueError("The tokenized prompts have no matching tokens after the edit. Keep a shared suffix after the changed word or phrase.")
+    positions_a, positions_b = zip(*pairs)
+    with torch.no_grad():
+        out_a = _forward(model, inputs_a, output_hidden_states=True)
+        saved = {index: state[0, list(positions_a)].detach().float().cpu()
+                 for index, state in _states(out_a, selected, len(layers)).items()}
+        logits_a = out_a.logits[0, -1].detach().float().cpu()
+        del out_a
+        out_b = _forward(model, inputs_b, output_hidden_states=True)
+        rows = []
+        for index, state in _states(out_b, selected, len(layers)).items():
+            va, vb = saved[index], state[0, list(positions_b)].detach().float().cpu()
+            distance = (1 - F.cosine_similarity(va, vb, dim=-1)).clamp_min(0)
+            relative = (vb - va).norm(dim=-1) / va.norm(dim=-1).clamp_min(1e-12)
+            rows.append({"layer": index, "cosine_distance": float(distance.mean()), "relative_l2": float(relative.mean())})
+        logits_b = out_b.logits[0, -1].detach().float().cpu()
+    pa, pb = logits_a.softmax(-1), logits_b.softmax(-1)
+    mixture = (pa + pb) / 2
+    js = float(((pa * (pa.clamp_min(1e-30).log() - mixture.clamp_min(1e-30).log())).sum()
+                + (pb * (pb.clamp_min(1e-30).log() - mixture.clamp_min(1e-30).log())).sum()) / 2)
+    top = []
+    for token_id in torch.topk(mixture, min(8, mixture.numel())).indices.tolist():
+        top.append({"token": tokenizer.decode([token_id]), "probability_a": float(pa[token_id]),
+                    "probability_b": float(pb[token_id])})
+    return _result(
+        "Butterfly effect", f"Tracked {len(pairs)} matching token positions through {len(selected)} layers after the prompt edit.",
+        rows, _line("How the prompt edit changes downstream states", rows,
+                    [("Cosine distance", "cosine_distance"), ("Relative L2 change", "relative_l2")]),
+        {"selected_layers": selected, "aligned_token_count": len(pairs),
+         "alignment": [{"position_a": i, "position_b": j, "token_a": ids_a[i], "token_b": ids_b[j]}
+                       for i, j in pairs], "next_token_js_divergence_nats": max(0., js),
+         "next_token_probabilities": top},
+        (["The tokenized inputs are identical after truncation; this is a zero-change control."] if identical else []) +
+        ["Matched token IDs after the first difference are compared. Insertions may also change absolute positions.",
+         "Distances describe representations, not the model's thoughts. Final hidden states may include normalization."],
+        [{"label": "Prompt A", "text": a}, {"label": "Prompt B", "text": b}],
+    )
+
+
+def _candidate_score(model, prompt_ids, continuation):
+    if not prompt_ids:
+        raise ValueError("The tokenizer returned no prompt tokens.")
+    all_ids = torch.tensor([prompt_ids + continuation], dtype=torch.long, device=_device(model))
+    with torch.no_grad():
+        out = _forward(model, {"input_ids": all_ids, "attention_mask": torch.ones_like(all_ids)})
+        start = len(prompt_ids) - 1
+        logits = out.logits[0, start:start + len(continuation)].float()
+        targets = torch.tensor(continuation, dtype=torch.long, device=logits.device)
+        scores = logits.log_softmax(-1).gather(-1, targets[:, None]).squeeze(-1)
+    return float(scores.mean()), len(continuation)
+
+
+def _tugofwar(model, tokenizer, options):
+    answer_a = str(options.get("choice_a") or "MAPLE").strip()
+    answer_b = str(options.get("choice_b") or "RAVEN").strip()
+    if answer_a == answer_b or not answer_a or not answer_b:
+        raise ValueError("Choose two different nonempty candidate answers.")
+    a, b = " " + answer_a, " " + answer_b
+    # Explicit token boundary keeps every candidate token scored. Both candidates
+    # receive exactly the same prompt IDs, even when their token counts differ.
+    continuation_a = tokenizer(a, add_special_tokens=False)["input_ids"]
+    continuation_b = tokenizer(b, add_special_tokens=False)["input_ids"]
+    if any(not ids or len(ids) > 32 or len(ids) >= _limit(options)
+           for ids in (continuation_a, continuation_b)):
+        raise ValueError("Choose nonempty candidate continuations of at most 32 tokens, shorter than the input budget.")
+    prompt_budget = _limit(options) - max(len(continuation_a), len(continuation_b))
+    fact_a, fact_b = f"The access code is {a.strip()}.", f"The access code is {b.strip()}."
+    variants = (("a_then_b", [fact_a, fact_b]), ("b_then_a", [fact_b, fact_a]),
+                ("repeat_a", [fact_a, fact_a, fact_b]), ("repeat_b", [fact_a, fact_b, fact_b]))
+    preamble = str(options.get("text") or "").strip()
+    prepared = []
+    for name, facts in variants:
+        core = "\n".join(facts) + "\nThe access code is"
+        core_ids = tokenizer(core, add_special_tokens=True)["input_ids"]
+        if len(core_ids) > prompt_budget:
+            raise ValueError(
+                "The complete conflicting facts and answer candidates do not fit the token budget. "
+                "Increase input tokens or shorten the two candidate answers."
+            )
+        prompt = ((preamble + "\n\n") if preamble else "") + core
+        prompt_ids = tokenizer(prompt, add_special_tokens=True)["input_ids"]
+        prepared.append((name, core, core_ids, prompt, prompt_ids))
+    # Keep the background identical across variants: omit it from every variant
+    # if the largest fact block cannot fit alongside it.
+    omitted_preamble = any(len(item[4]) > prompt_budget for item in prepared)
+    rows, prompts, contexts = [], [], []
+    for name, core, core_ids, prompt, prompt_ids in prepared:
+        if omitted_preamble:
+            # Preserve the actual intervention: losing a repeated fact would
+            # make a 'repeat' label describe an experiment that never ran.
+            prompt, prompt_ids = core, core_ids
+        score_a, count_a = _candidate_score(model, prompt_ids, continuation_a)
+        score_b, count_b = _candidate_score(model, prompt_ids, continuation_b)
+        rows.append({"variant": name, "a_mean_logprob": score_a, "b_mean_logprob": score_b,
+                     "a_minus_b": score_a - score_b, "candidate_a_tokens": count_a, "candidate_b_tokens": count_b})
+        prompts.append({"label": name, "text": prompt})
+        contexts.append({"variant": name, "prompt_token_ids": prompt_ids,
+                         "background_was_omitted": omitted_preamble})
+    return _result(
+        "Context tug-of-war", "Compared two complete candidate continuations under reversed and repeated conflicting facts.",
+        rows, _plot("bar", "Which continuation is preferred?", "Context arrangement", "Mean log-probability A minus B",
+                    series=[{"name": "A − B (positive favors A)", "x": [r["variant"] for r in rows],
+                             "y": [r["a_minus_b"] for r in rows]}]),
+        {"candidate_a": a, "candidate_b": b, "scoring": "Mean token log-probability of the full candidate continuation",
+         "token_boundary": "Candidate tokenized separately without special tokens and appended to prompt IDs",
+         "answer_whitespace": "Candidate answers are trimmed and prefixed with one space after 'is'",
+         "contexts": contexts,
+         "truncation": "Omit optional background if necessary; preserve complete fact block and identical prompt IDs for both candidates"},
+        ["Scores are length-normalized log probabilities, not calibrated answer probabilities.",
+         "This probe tests order and repetition sensitivity for these statements; it is not a general instruction-following benchmark."] +
+        (["The optional background was omitted from all variants to fit the token budget. Every experimental fact was preserved."]
+         if any(context["background_was_omitted"] for context in contexts) else []),
+        prompts,
+    )
+
+
+def run_inspection(model, tokenizer, tool: str, options: dict) -> dict:
+    """Run one bounded tool without loading models or retaining request data."""
+    handlers = {"fingerprint": _fingerprint, "silhouette": lambda m, t, o: _semantic(m, t, o, True),
+                "semantic": _semantic, "cka": _cka, "attention": _attention, "gradient": _gradient,
+                "butterfly": _butterfly, "tugofwar": _tugofwar}
+    if tool not in handlers:
+        raise ValueError(f"Unknown inspection tool: {tool}")
+    return handlers[tool](model, tokenizer, dict(options or {}))

bol_space/presentation.py

@@ -0,0 +1,391 @@
+"""Browser presentation, offline reports, and consent-scoped research exports.
+
+This module never uploads data. ``contribution_payload`` is deliberately separate
+from the user's complete, local report so a consent choice controls retention.
+"""
+
+from __future__ import annotations
+
+import html
+import json
+import math
+import numbers
+import os
+import re
+import shutil
+import tempfile
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+import plotly.graph_objects as go
+import plotly.io as pio
+
+
+MEASUREMENT_NOTE = (
+    "These measurements describe this checkpoint, precision, input, and selected "
+    "settings. A change in this experiment is not a general measure of model "
+    "quality or proof that a layer has a particular function. BF16 execution and "
+    "sampling can affect small differences; compare runs using matching settings."
+)
+
+_PALETTE = ["#167c80", "#d88759", "#25394c", "#7d73a4", "#8c9e6c", "#b85e72"]
+_RESULT_KEYS = {
+    "title", "summary", "rows", "plot", "details", "warnings", "text_outputs",
+    "model", "run",
+}
+_MODEL_KEYS = {"repo_id", "revision", "architecture", "dtype", "parameters"}
+_RUN_KEYS = {"id", "created_at", "tool", "seconds", "peak_vram_gb", "gpu", "options"}
+_TOOL_IDS = {
+    "fingerprint", "silhouette", "semantic", "cka", "attention", "gradient",
+    "butterfly", "tugofwar", "layer_sweep", "component", "trisect", "wormhole",
+    "noise", "duets",
+}
+_NUMERIC_OPTIONS = {
+    "max_tokens", "max_new_tokens", "sample_layers", "layer_start", "layer_end",
+    "noise_scale", "seed",
+}
+# Every retained metric name is controlled by the application, never by a token,
+# prompt, candidate answer, custom label, or user-supplied dictionary key.
+_NUMERIC_ROW_KEYS = {
+    "layer", "head", "layer_a", "layer_b", "mean", "std", "abs_mean",
+    "minimum", "maximum", "frobenius", "sparsity", "spectral_estimate",
+    "within_cosine", "cross_cosine", "separation", "cosine", "cka", "entropy",
+    "normalized_entropy", "head_diversity", "last_token_entropy",
+    "last_token_normalized_entropy", "gradient_norm", "gradient_mean",
+    "parameter_norm", "gradient_to_param_ratio", "cosine_distance", "relative_l2",
+    "a_mean_logprob", "b_mean_logprob", "a_minus_b", "loss", "damage",
+    "modified_tensors", "single_damage_a", "single_damage_b", "pair_damage",
+    "interaction", "noise_scale", "changed_fraction", "relative_write_norm",
+    "kept_layers", "zeroed_layers", "shape", "weight_rms",
+}
+_SECRET_KEYS = {
+    "api_key", "hf_token", "access_token", "auth_token", "password", "secret",
+    "authorization", "credentials", "tokenizer_token", "hub_token",
+}
+
+
+def _json_value(value: Any) -> Any:
+    """Copy JSON-compatible values and reject non-finite measurements explicitly."""
+    if value is None or isinstance(value, (str, bool)):
+        return value
+    if isinstance(value, numbers.Integral):
+        return int(value)
+    if isinstance(value, numbers.Real):
+        number = float(value)
+        if not math.isfinite(number):
+            raise ValueError("Reports require finite measurements; found NaN or infinity.")
+        return number
+    if isinstance(value, dict):
+        if not all(isinstance(key, str) for key in value):
+            raise ValueError("Report object keys must be strings.")
+        return {key: _json_value(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_value(item) for item in value]
+    raise TypeError(f"Report values must be JSON-compatible, received {type(value).__name__}.")
+
+
+def _without_secrets(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            key: _without_secrets(item)
+            for key, item in value.items()
+            if key.casefold() not in _SECRET_KEYS
+        }
+    if isinstance(value, list):
+        return [_without_secrets(item) for item in value]
+    return value
+
+
+def _result_copy(result: dict) -> dict:
+    if not isinstance(result, dict):
+        raise TypeError("A report result must be a dictionary.")
+    copied = _json_value({key: value for key, value in result.items() if key in _RESULT_KEYS})
+    for key, allowed in (("model", _MODEL_KEYS), ("run", _RUN_KEYS)):
+        metadata = copied.get(key, {})
+        if not isinstance(metadata, dict):
+            raise ValueError(f"Report {key} metadata must be a dictionary.")
+        copied[key] = {name: value for name, value in metadata.items() if name in allowed}
+    return _without_secrets(copied)
+
+
+def _text(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, (dict, list)):
+        return json.dumps(value, ensure_ascii=False, allow_nan=False)
+    return str(value)
+
+
+def _escaped(value: Any) -> str:
+    return html.escape(_text(value), quote=True)
+
+
+def _markdown(value: Any) -> str:
+    return re.sub(r"([\\`*_{}\[\]()#!|])", r"\\\1", _escaped(value))
+
+
+def _plot_values(values: Any) -> list:
+    # Plotly supports a subset of HTML in labels. Treat all model/user text as
+    # literal text, including closing script tags in an offline report.
+    if not isinstance(values, list):
+        return []
+    return [_escaped(value) if isinstance(value, str) else value for value in values]
+
+
+def _figure(result: dict) -> go.Figure:
+    spec = result.get("plot") or {}
+    if not isinstance(spec, dict):
+        spec = {}
+    kind = spec.get("kind", "line")
+    figure = go.Figure()
+    if kind == "heatmap" and spec.get("z"):
+        figure.add_trace(go.Heatmap(
+            x=_plot_values(spec.get("x", [])),
+            y=_plot_values(spec.get("y", [])),
+            z=spec["z"],
+            colorscale=[[0, "#f0e8dc"], [0.5, "#79bbb4"], [1, "#174957"]],
+            colorbar={"thickness": 14, "outlinewidth": 0},
+            hoverongaps=False,
+        ))
+    else:
+        series = spec.get("series") or []
+        for index, item in enumerate(series):
+            if not isinstance(item, dict):
+                continue
+            common = {
+                "x": _plot_values(item.get("x", [])),
+                "y": _plot_values(item.get("y", [])),
+                "name": _escaped(item.get("name", f"Series {index + 1}")),
+            }
+            color = _PALETTE[index % len(_PALETTE)]
+            if kind == "bar":
+                figure.add_trace(go.Bar(**common, marker_color=color))
+            else:
+                figure.add_trace(go.Scatter(
+                    **common, mode="lines+markers", line={"color": color, "width": 2},
+                    marker={"size": 6}, connectgaps=False,
+                ))
+    if not figure.data:
+        figure.add_annotation(
+            text="Measurements are available in the table below.", showarrow=False,
+            x=0.5, y=0.5, xref="paper", yref="paper",
+        )
+    figure.update_layout(
+        title={"text": _escaped(spec.get("title", "Measurements")), "x": 0.025},
+        xaxis_title=_escaped(spec.get("x_label", "")),
+        yaxis_title=_escaped(spec.get("y_label", "")),
+        template="plotly_white", paper_bgcolor="#fbf8f1", plot_bgcolor="#fbf8f1",
+        font={"family": "Arial, sans-serif", "color": "#25394c", "size": 13},
+        margin={"l": 65, "r": 30, "t": 75, "b": 65}, height=500,
+        legend={"orientation": "h", "y": -0.2, "x": 0},
+        hovermode="closest", barmode="group",
+    )
+    figure.update_xaxes(gridcolor="#e5e0d6", zerolinecolor="#c4c1b7")
+    figure.update_yaxes(gridcolor="#e5e0d6", zerolinecolor="#c4c1b7")
+    return figure
+
+
+def render_result(result: dict) -> tuple[str, go.Figure, pd.DataFrame, str]:
+    """Return Gradio-ready summary, chart, measurement table, and text comparison."""
+    data = _result_copy(result)
+    summary = f"### {_markdown(data.get('title', 'Experiment report'))}\n\n"
+    summary += _markdown(data.get("summary", ""))
+    model = data.get("model", {})
+    if model.get("repo_id"):
+        summary += f"\n\n**Model:** {_markdown(model['repo_id'])}"
+        if model.get("dtype"):
+            summary += f" · {_markdown(model['dtype'])}"
+    warnings = data.get("warnings") or []
+    if warnings:
+        summary += "\n\n" + "\n\n".join(f"**Note:** {_markdown(item)}" for item in warnings)
+    text_blocks = []
+    for item in data.get("text_outputs") or []:
+        if isinstance(item, dict):
+            text_blocks.append(
+                f"#### {_markdown(item.get('label', 'Output'))}\n\n"
+                f"<pre>{_escaped(item.get('text', ''))}</pre>"
+            )
+    return summary, _figure(data), pd.DataFrame(data.get("rows") or []), "\n\n".join(text_blocks)
+
+
+_STYLE = """
+:root{color-scheme:light;--ink:#25394c;--muted:#66737b;--paper:#fbf8f1;--line:#dedbd1;--teal:#167c80}
+*{box-sizing:border-box}body{margin:0;background:#eae8df;color:var(--ink);font:15px/1.65 Arial,sans-serif}
+main{max-width:1180px;margin:32px auto;padding:48px;background:var(--paper);border-top:6px solid var(--teal)}
+header{border-bottom:1px solid var(--line);padding-bottom:24px;margin-bottom:24px}.brand{color:var(--teal);font:700 13px/1.5 monospace;letter-spacing:.13em}
+h1{font-size:38px;letter-spacing:-.04em;line-height:1.15;margin:16px 0}h2{font-size:22px;margin-top:34px}h3{font-size:16px}
+.subtitle,.muted{color:var(--muted)}.facts{display:flex;flex-wrap:wrap;gap:12px;margin:20px 0}.fact{border:1px solid var(--line);border-radius:8px;padding:10px 16px;min-width:130px}
+.fact span{display:block;color:var(--muted);font-size:11px;text-transform:uppercase;letter-spacing:.07em}.fact strong{display:block;font-size:16px;overflow-wrap:anywhere}
+.summary{font-size:18px;white-space:pre-wrap}.note{background:#eaf1ed;border-left:3px solid var(--teal);padding:14px 18px;margin:18px 0}.warning{border-left-color:#d88759;background:#f6ecdf}
+.table-wrap{overflow:auto}table{border-collapse:collapse;width:100%;font-size:13px}th,td{text-align:left;padding:10px 12px;border-bottom:1px solid var(--line);vertical-align:top}
+th{font-weight:700;background:#eeede5}tbody tr:nth-child(even){background:#f5f3ec}pre{white-space:pre-wrap;overflow-wrap:anywhere;padding:18px;background:#f0eee6;border:1px solid var(--line);border-radius:8px;font:13px/1.6 monospace}
+.outputs{display:grid;grid-template-columns:repeat(auto-fit,minmax(280px,1fr));gap:20px}.output{min-width:0}.output pre{height:calc(100% - 48px)}
+details{margin:22px 0}summary{cursor:pointer;font-weight:700}footer{margin-top:38px;border-top:1px solid var(--line);padding-top:16px;font-size:12px;color:var(--muted)}
+.print{border:1px solid var(--teal);border-radius:6px;color:var(--teal);background:transparent;padding:9px 14px;cursor:pointer;float:right}
+@media(max-width:700px){main{margin:0;padding:24px}h1{font-size:29px}.print{float:none;margin-bottom:14px}}
+@media print{@page{margin:14mm}body{background:white;font-size:11px}main{margin:0;padding:0;max-width:none;background:white;border-top-width:3px}.print{display:none}.modebar{display:none!important}h1{font-size:28px}h2,h3{break-after:avoid}pre,.fact,.note,.output{break-inside:avoid}.table-wrap{overflow:visible}td,th{padding:6px}footer{margin-top:20px}.js-plotly-plot{break-inside:avoid}}
+"""
+
+
+def create_report(result: dict) -> tuple[str, str]:
+    """Write a complete JSON result and self-contained printable HTML to unique files."""
+    data = _result_copy(result)
+    serialized = json.dumps(data, ensure_ascii=False, indent=2, allow_nan=False)
+    _, figure, table, _ = render_result(data)
+    chart = pio.to_html(
+        figure, include_plotlyjs=True, include_mathjax=False, full_html=False,
+        div_id=f"measurements-{uuid.uuid4().hex}", default_width="100%",
+        config={"responsive": True, "displaylogo": False, "toImageButtonOptions": {"format": "svg"}},
+    )
+    model, run = data.get("model", {}), data.get("run", {})
+    facts = [("Checkpoint", model.get("repo_id", "Unspecified")), ("Precision", model.get("dtype", "Unspecified"))]
+    if isinstance(run.get("seconds"), (int, float)):
+        facts.append(("Run time", f"{run['seconds']:.2f} s"))
+    if isinstance(run.get("peak_vram_gb"), (int, float)):
+        facts.append(("Peak GPU memory", f"{run['peak_vram_gb']:.2f} GB"))
+    fact_html = "".join(f'<div class="fact"><span>{_escaped(label)}</span><strong>{_escaped(value)}</strong></div>' for label, value in facts)
+    warnings = "".join(f'<div class="note warning">{_escaped(item)}</div>' for item in data.get("warnings") or [])
+    table_html = table.to_html(index=False, escape=True, border=0, na_rep="—") if not table.empty else '<p class="muted">No tabular measurements for this run.</p>'
+    outputs = "".join(
+        f'<article class="output"><h3>{_escaped(item.get("label", "Output"))}</h3><pre>{_escaped(item.get("text", ""))}</pre></article>'
+        for item in data.get("text_outputs") or [] if isinstance(item, dict)
+    )
+    output_section = f'<section><h2>Text comparisons</h2><div class="outputs">{outputs}</div></section>' if outputs else ""
+    metadata = {"model": model, "run": run, "details": data.get("details", {})}
+    metadata_html = _escaped(json.dumps(metadata, ensure_ascii=False, indent=2, allow_nan=False))
+    title = _escaped(data.get("title", "Experiment report"))
+    document = f'''<!doctype html>
+<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
+<title>{title} · 0perating-soon</title><style>{_STYLE}</style></head>
+<body><main><header><button class="print" onclick="window.print()">Print / save PDF</button>
+<div class="brand">0PERATING-SOON / BoL</div><h1>{title}</h1>
+<p class="subtitle">Model internals, opened for inspection.</p><div class="facts">{fact_html}</div></header>
+<p class="summary">{_escaped(data.get("summary", ""))}</p>{warnings}
+<section aria-label="Interactive measurements">{chart}</section>
+<section><h2>Measurements</h2><div class="table-wrap">{table_html}</div></section>{output_section}
+<section><h2>Reading this result</h2><div class="note">{_escaped(MEASUREMENT_NOTE)}</div></section>
+<details open><summary>Model, settings, and experiment details</summary><pre>{metadata_html}</pre></details>
+<footer>This report is self-contained and works offline. Downloading a report does not contribute its data; research contributions require a separate consent choice.</footer>
+</main></body></html>'''
+    # Files inside Gradio's cache are served in place and registered for its
+    # expiry sweep. Writing elsewhere would leave an untracked original behind.
+    cache_root = Path(os.environ.get("GRADIO_TEMP_DIR", os.path.join(tempfile.gettempdir(), "bol-gradio")))
+    cache_root.mkdir(parents=True, exist_ok=True)
+    directory = Path(tempfile.mkdtemp(prefix="bol-report-", dir=cache_root))
+    json_path, html_path = directory / "experiment.json", directory / "experiment.html"
+    try:
+        json_path.write_text(serialized, encoding="utf-8")
+        html_path.write_text(document, encoding="utf-8")
+    except Exception:
+        shutil.rmtree(directory, ignore_errors=True)
+        raise
+    return str(json_path), str(html_path)
+
+
+def _numeric(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _numeric_value(value: Any) -> bool:
+    return _numeric(value) or (
+        isinstance(value, list) and all(_numeric_value(item) for item in value)
+    )
+
+
+def _numeric_axis(values: Any) -> list:
+    if not isinstance(values, list):
+        return []
+    # Categorical axes may contain tokens or candidate text. Preserve their
+    # positional relationship to numeric measurements, without their labels.
+    if all(_numeric(item) or item is None for item in values):
+        return values
+    return list(range(len(values)))
+
+
+def _numeric_plot(spec: Any) -> dict:
+    if not isinstance(spec, dict):
+        return {}
+    kind = spec.get("kind")
+    if kind not in {"line", "bar", "heatmap"}:
+        return {}
+    output: dict = {"kind": kind}
+    if kind == "heatmap":
+        matrix = spec.get("z", [])
+        if not isinstance(matrix, list) or not all(
+            isinstance(row, list) and all(_numeric(value) or value is None for value in row)
+            for row in matrix
+        ):
+            return {}
+        output.update(x=_numeric_axis(spec.get("x")), y=_numeric_axis(spec.get("y")), z=matrix)
+    else:
+        output["series"] = []
+        for index, item in enumerate(spec.get("series") or []):
+            if not isinstance(item, dict):
+                continue
+            values = item.get("y", [])
+            if not isinstance(values, list) or not all(_numeric(value) or value is None for value in values):
+                continue
+            output["series"].append({
+                "name": f"series_{index + 1}", "x": _numeric_axis(item.get("x")), "y": values,
+            })
+    return output
+
+
+def contribution_payload(result: dict, include_text: bool) -> dict:
+    """Build the exact contribution, using no custom text without text consent.
+
+    Text-free exports intentionally omit summaries, details, tensor/component
+    labels, token IDs, candidate answers, arbitrary options, and plot labels.
+    Their row order and generic plot series IDs preserve numeric comparisons.
+    Model identifiers are retained as the provenance the user agrees to share.
+    """
+    if not isinstance(include_text, bool):
+        raise TypeError("include_text must be an explicit boolean consent choice.")
+    result = _result_copy(result)
+    if include_text:
+        data = result
+    else:
+        model, run = result.get("model", {}), result.get("run", {})
+        safe_model = {
+            key: model[key] for key in ("repo_id", "revision", "architecture", "dtype")
+            if isinstance(model.get(key), str)
+        }
+        if _numeric(model.get("parameters")):
+            safe_model["parameters"] = model["parameters"]
+        safe_run = {
+            key: run[key] for key in ("seconds", "peak_vram_gb")
+            if _numeric(run.get(key))
+        }
+        if run.get("tool") in _TOOL_IDS:
+            safe_run["tool"] = run["tool"]
+        run_id = run.get("id")
+        if isinstance(run_id, str) and re.fullmatch(r"[a-fA-F0-9-]{16,36}", run_id):
+            safe_run["id"] = run_id
+        created_at = run.get("created_at")
+        if isinstance(created_at, str):
+            try:
+                datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+            except ValueError:
+                pass
+            else:
+                safe_run["created_at"] = created_at
+        options = run.get("options") or {}
+        safe_run["options"] = {
+            key: value for key, value in options.items()
+            if key in _NUMERIC_OPTIONS and (_numeric_value(value) or isinstance(value, bool))
+        } if isinstance(options, dict) else {}
+        rows = []
+        for row in result.get("rows") or []:
+            if isinstance(row, dict):
+                rows.append({
+                    key: value for key, value in row.items()
+                    if key in _NUMERIC_ROW_KEYS and (_numeric_value(value) or value is None)
+                })
+        data = {"model": safe_model, "run": safe_run, "rows": rows, "plot": _numeric_plot(result.get("plot"))}
+    payload = {"schema_version": "1", "consent_version": "1", "includes_text": include_text, "data": data}
+    # Keep this promise local: upload code receives already-valid strict JSON.
+    json.dumps(payload, ensure_ascii=False, allow_nan=False)
+    return payload

bol_space/runtime.py

@@ -0,0 +1,116 @@
+"""Checkpoint preparation and short, isolated GPU experiments."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import gc
+import math
+import time
+import uuid
+
+import spaces  # Must precede torch and transformers.
+import torch
+from bol_space.checkpoints import DEFAULT_MODEL, prepare_checkpoint, load_checkpoint
+INSPECTION_TOOLS = {"fingerprint", "silhouette", "semantic", "cka", "attention", "gradient", "butterfly", "tugofwar"}
+SURGERY_TOOLS = {"layer_sweep", "component", "trisect", "wormhole", "noise", "duets"}
+ALL_TOOLS = INSPECTION_TOOLS | SURGERY_TOOLS
+
+
+def validate_options(tool: str, options: dict) -> dict:
+    if tool not in ALL_TOOLS:
+        raise ValueError("Choose an available experiment.")
+    result = dict(options)
+    limits = {"max_tokens": (32, 256), "max_new_tokens": (1, 40), "sample_layers": (2, 16), "layer_start": (0, 10000), "layer_end": (-1, 10000), "seed": (0, 2147483647)}
+    for key, (low, high) in limits.items():
+        value = result.get(key)
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or int(value) != value or not low <= value <= high:
+            raise ValueError(f"{key} must be an integer from {low} to {high}.")
+        result[key] = int(value)
+    scale = float(result.get("noise_scale", 0.01))
+    if not math.isfinite(scale) or not 0 <= scale <= 0.5:
+        raise ValueError("Noise strength must be between 0 and 0.5.")
+    result["noise_scale"] = scale
+    for key in ("text", "compare_text", "prompt", "words", "component", "choice_a", "choice_b"):
+        value = result.get(key, "")
+        if not isinstance(value, str) or len(value) > 10000:
+            raise ValueError(f"{key} must contain no more than 10,000 characters.")
+    if not result.get("text", "").strip():
+        raise ValueError("Enter some text to analyze.")
+    return result
+
+
+def model_summary(ticket: dict) -> str:
+    count = ticket.get("parameters")
+    size = f" · {count / 1e9:.2f}B parameters · approximately {count * 2 / 1e9:.2f} GB of BF16 weights" if count else ""
+    return f"**{ticket['repo_id']}** · {ticket['architecture']}{size}\n\nCheckpoint `{ticket['revision'][:12]}` is ready. Run an experiment to measure its actual GPU memory use."
+
+
+# ZeroGPU registers the default model's CUDA placement at startup. User-selected
+# checkpoints cannot be preloaded before they are known; their download is kept
+# outside the GPU call and their temporary model is released after each request.
+print("Preparing default SmolLM2 checkpoint...", flush=True)
+DEFAULT_TICKET = prepare_checkpoint(DEFAULT_MODEL)
+DEFAULT_INSTANCE, DEFAULT_TOKENIZER = load_checkpoint(DEFAULT_TICKET)
+print("Default checkpoint registered in BF16; application ready to start.", flush=True)
+
+
+def duration_for_request(request: dict | None) -> int:
+    if not request:
+        return 60
+    tool = request.get("tool")
+    duration = 90 if tool in {"fingerprint", "gradient", "duets", "component", "trisect", "wormhole", "noise", "layer_sweep"} else 60
+    if request.get("ticket", {}).get("revision") != DEFAULT_TICKET["revision"]:
+        duration += 60
+    return duration
+
+
+@spaces.GPU(duration=duration_for_request)
+def analyze_gpu(request: dict) -> dict:
+    """Run one BoL experiment in BF16 and return CPU measurements and resource use."""
+    from bol_space.inspection import run_inspection
+    from bol_space.surgery import run_surgery
+
+    if not request or "ticket" not in request:
+        raise ValueError("Prepare a checkpoint before starting an experiment.")
+    ticket, tool = request["ticket"], request["tool"]
+    options = validate_options(tool, request["options"])
+    started = time.perf_counter()
+    on_gpu = torch.cuda.is_available()
+    if on_gpu:
+        torch.cuda.reset_peak_memory_stats()
+    is_default = ticket["repo_id"] == DEFAULT_TICKET["repo_id"] and ticket["revision"] == DEFAULT_TICKET["revision"] and not getattr(DEFAULT_INSTANCE, "_bol_integrity_failed", False)
+    model = tokenizer = None
+    try:
+        if is_default:
+            model, tokenizer = DEFAULT_INSTANCE, DEFAULT_TOKENIZER
+        else:
+            model, tokenizer = load_checkpoint(ticket)
+        count = sum(p.numel() for p in model.parameters())
+        fn = run_inspection if tool in INSPECTION_TOOLS else run_surgery
+        with torch.random.fork_rng(devices=[torch.cuda.current_device()] if on_gpu else []):
+            torch.manual_seed(options["seed"])
+            result = fn(model, tokenizer, tool, options)
+        if on_gpu:
+            torch.cuda.synchronize()
+        result["model"] = {k: ticket[k] for k in ("repo_id", "revision", "architecture", "dtype")}
+        result["model"]["parameters"] = count
+        result["run"] = {"id": uuid.uuid4().hex, "created_at": datetime.now(timezone.utc).isoformat(), "tool": tool,
+                         "seconds": round(time.perf_counter() - started, 3), "peak_vram_gb": round(torch.cuda.max_memory_allocated() / 1e9, 3) if on_gpu else 0,
+                         "gpu": torch.cuda.get_device_name() if on_gpu else "CPU", "options": options}
+        result.setdefault("warnings", []).append("BF16 model execution; numerical summaries use FP32. Results describe these probes and this checkpoint.")
+        return result
+    except torch.cuda.OutOfMemoryError as exc:
+        raise ValueError("This experiment exceeded the available GPU memory. Try fewer sampled layers, shorter text, or a smaller checkpoint. No model-size ceiling has been established yet.") from exc
+    except RuntimeError as exc:
+        if is_default and "Weight restoration failed" in str(exc):
+            # A reused worker must never serve measurements from a failed restore.
+            # Future calls on this worker load an independent pristine checkpoint.
+            DEFAULT_INSTANCE._bol_integrity_failed = True
+        raise
+    finally:
+        if model is not None:
+            model.zero_grad(set_to_none=True)
+        if not is_default:
+            del model, tokenizer
+        gc.collect()
+        if on_gpu:
+            torch.cuda.empty_cache()

bol_space/surgery.py

@@ -0,0 +1,437 @@
+"""Bounded, reversible BoL experiments on an already loaded causal model.
+
+The caller owns model loading, request serialization, and disposal after errors.
+These experiments zero matrices rather than deleting decoder blocks: residual
+paths, normalization parameters, and biases remain in place.
+"""
+
+from contextlib import contextmanager
+from itertools import combinations
+import math
+
+import torch
+import torch.nn.functional as F
+from torch import nn
+
+
+def _layer_list(model):
+    paths = (
+        "model.layers", "transformer.h", "model.decoder.layers",
+        "gpt_neox.layers", "transformer.layers", "transformer.blocks",
+        "model.transformer.h", "model.transformer.blocks", "backbone.layers",
+        "decoder.layers", "layers",
+    )
+    for path in paths:
+        try:
+            layers = model.get_submodule(path)
+        except (AttributeError, KeyError):
+            continue
+        if isinstance(layers, nn.ModuleList) and len(layers):
+            return path, layers
+    raise ValueError(
+        "No supported decoder layer list found. This architecture cannot run "
+        "weight surgery yet; use a model with a standard decoder ModuleList."
+    )
+
+
+def _matrices(layers, indices):
+    """Group eligible parameters by layer, protecting normalization and biases."""
+    result = {}
+    for index in indices:
+        layer = layers[index]
+        modules = dict(layer.named_modules())
+        parameters = []
+        for name, parameter in layer.named_parameters():
+            if parameter.ndim < 2 or not parameter.is_floating_point():
+                continue
+            if name.rsplit(".", 1)[-1] == "bias":
+                continue
+            parts = name.split(".")[:-1]
+            ancestors = ["."] + [".".join(parts[:i]) for i in range(1, len(parts) + 1)]
+            if "norm" in type(layer).__name__.lower() or any(
+                "norm" in type(modules[prefix]).__name__.lower()
+                for prefix in ancestors if prefix in modules
+            ):
+                continue
+            parameters.append((name, parameter))
+        if not parameters:
+            raise ValueError(f"Layer {index} has no eligible floating-point weight matrices.")
+        result[index] = parameters
+    return result
+
+
+def _unique(parameters):
+    seen = set()
+    output = []
+    for name, parameter in parameters:
+        if id(parameter) not in seen:
+            output.append((name, parameter))
+            seen.add(id(parameter))
+    return output
+
+
+def _collect(by_layer, indices):
+    return _unique(
+        (f"{index}.{name}", parameter)
+        for index in indices for name, parameter in by_layer[index]
+    )
+
+
+@contextmanager
+def _temporary_edit(parameters, *, noise_scale=None, seed=42):
+    """Back up only edited tensors; restore even if editing/forward/generation fails."""
+    parameters = _unique(parameters)
+    if not parameters:
+        raise ValueError("The intervention selected no weight matrices.")
+    originals = []
+    info = {"modified_tensors": len(parameters), "integrity_verified": False}
+    generators = {}
+    changed, count, delta_squared, original_squared = 0, 0, 0.0, 0.0
+    try:
+        with torch.no_grad():
+            for name, parameter in parameters:
+                original = parameter.detach().clone()
+                # Record the backup before the first mutation, including partial failures.
+                originals.append((name, parameter, original))
+                if noise_scale is None:
+                    parameter.zero_()
+                else:
+                    device = str(parameter.device)
+                    if device not in generators:
+                        generators[device] = torch.Generator(device=parameter.device).manual_seed(seed)
+                    source = original.float()
+                    scale = source.abs().mean() * noise_scale
+                    noise = torch.randn(
+                        parameter.shape, device=parameter.device, dtype=torch.float32,
+                        generator=generators[device],
+                    )
+                    parameter.copy_(source + noise * scale)
+                    realized = parameter.detach().float() - source
+                    changed += int(torch.count_nonzero(realized).item())
+                    count += parameter.numel()
+                    delta_squared += float(realized.square().sum().item())
+                    original_squared += float(source.square().sum().item())
+                    del source, noise, realized
+            if noise_scale is not None:
+                info["changed_fraction"] = changed / count if count else 0.0
+                info["relative_write_norm"] = (
+                    math.sqrt(delta_squared / original_squared) if original_squared else 0.0
+                )
+        yield info
+    finally:
+        # Attempt every restoration before raising an integrity error.
+        failed = []
+        with torch.no_grad():
+            for name, parameter, original in originals:
+                try:
+                    parameter.copy_(original)
+                    if not torch.equal(parameter.detach(), original):
+                        failed.append(name)
+                except Exception:
+                    failed.append(name)
+        if failed:
+            raise RuntimeError(
+                "Weight restoration failed; discard this model instance. "
+                f"Affected tensors: {', '.join(failed[:4])}"
+            )
+        info["integrity_verified"] = True
+
+
+def _integer(options, key, default, lower, upper):
+    value = options.get(key, default)
+    try:
+        numeric = float(value)
+        integer = int(numeric)
+    except (TypeError, ValueError, OverflowError) as exc:
+        raise ValueError(f"{key} must be an integer from {lower} to {upper}.") from exc
+    if not math.isfinite(numeric) or integer != numeric or not lower <= integer <= upper:
+        raise ValueError(f"{key} must be an integer from {lower} to {upper}.")
+    return integer
+
+
+def _sample(indices, count):
+    if len(indices) <= count:
+        return list(indices)
+    if count == 1:
+        return [indices[len(indices) // 2]]
+    return [indices[round(i * (len(indices) - 1) / (count - 1))] for i in range(count)]
+
+
+def _prepare_batch(model, tokenizer, text, max_tokens):
+    texts = [line.strip() for line in text.splitlines() if line.strip()]
+    if not texts:
+        raise ValueError("Enter evaluation text containing at least two tokens.")
+    if len(texts) > 8:
+        raise ValueError("Use at most eight nonempty evaluation lines per experiment.")
+    batch = tokenizer(
+        texts, return_tensors="pt", padding=True,
+        truncation=True, max_length=max_tokens,
+    )
+    device = next(model.parameters()).device
+    batch = {key: value.to(device) for key, value in batch.items()
+             if key in ("input_ids", "attention_mask")}
+    if "attention_mask" not in batch:
+        raise ValueError("Tokenizer did not return an attention mask.")
+    targets = batch["input_ids"][:, 1:].clone()
+    valid = batch["attention_mask"][:, 1:].bool() & batch["attention_mask"][:, :-1].bool()
+    targets.masked_fill_(~valid, -100)
+    scored_tokens = int(valid.sum().item())
+    if not scored_tokens:
+        raise ValueError("Evaluation text must have at least two unpadded tokens in a line.")
+    return batch, targets, scored_tokens
+
+
+def _loss(model, batch, targets):
+    with torch.no_grad():
+        logits = model(**batch, use_cache=False).logits
+        # Explicit FP32 scoring, even when the model forward pass uses BF16.
+        shifted = logits[:, :-1, :].float().contiguous()
+        value = F.cross_entropy(shifted.view(-1, shifted.shape[-1]), targets.reshape(-1))
+    numeric = float(value.item())
+    if not math.isfinite(numeric):
+        raise FloatingPointError("The model produced a non-finite evaluation loss; the experiment was restored.")
+    return numeric
+
+
+def _generate(model, tokenizer, prompt, max_tokens, max_new_tokens):
+    device = next(model.parameters()).device
+    inputs = tokenizer(prompt, return_tensors="pt", truncation=True, max_length=max_tokens)
+    inputs = {key: value.to(device) for key, value in inputs.items()
+              if key in ("input_ids", "attention_mask")}
+    kwargs = {"max_new_tokens": max_new_tokens, "do_sample": False, "use_cache": True}
+    if tokenizer.pad_token_id is not None:
+        kwargs["pad_token_id"] = tokenizer.pad_token_id
+    with torch.no_grad():
+        output = model.generate(**inputs, **kwargs)
+    continuation = output[0, inputs["input_ids"].shape[1]:]
+    return tokenizer.decode(continuation, skip_special_tokens=True)
+
+
+def _zones(indices, tool):
+    n = len(indices)
+    q1, q2, q3 = n // 4, n // 2, n - n // 4
+    if tool == "trisect":
+        keeps = [
+            ("Inner core", indices[q1:q3]),
+            ("Outer shell", indices[:q1] + indices[q3:]),
+            ("Early half", indices[:q2]),
+            ("Late half", indices[q2:]),
+            ("Early inner quarter", indices[q1:q2]),
+            ("Late inner quarter", indices[q2:q3]),
+            ("Early outer + late inner", indices[:q1] + indices[q2:q3]),
+            ("Early inner + late outer", indices[q1:q2] + indices[q3:]),
+        ]
+    else:
+        keeps = [
+            ("Outer quarters", indices[:q1] + indices[q3:]),
+            ("Early outer + late inner", indices[:q1] + indices[q2:q3]),
+            ("Early inner + late outer", indices[q1:q2] + indices[q3:]),
+            ("Inner half", indices[q1:q3]),
+            ("Every even layer", [index for index in indices if index % 2 == 0]),
+            ("Every odd layer", [index for index in indices if index % 2 == 1]),
+            ("First + last", [indices[0], indices[-1]]),
+            ("First 3 + last 3", indices[:3] + indices[-3:]),
+            ("First 5 + last 5", indices[:5] + indices[-5:]),
+            ("Every third layer", indices[::3]),
+        ]
+    result, seen = [], set()
+    for label, keep in keeps:
+        keep = sorted(set(keep))
+        zero = tuple(index for index in indices if index not in keep)
+        # Small ranges produce duplicate/no-op patterns; do not call them findings.
+        if not keep or not zero or zero in seen:
+            continue
+        seen.add(zero)
+        result.append((label, keep, list(zero)))
+    return result
+
+
+def run_surgery(model, tokenizer, tool: str, options: dict) -> dict:
+    """Run one experiment and return a JSON-native report with restored weights.
+
+    Layer bounds are inclusive; -1 means the final layer. Sweep/noise sample at
+    most 16 layers; duets sample at most six. Component and zone cuts use every
+    layer inside the selected range. All scores are token-weighted causal NLL
+    on the supplied evaluation text, not a general capability benchmark.
+    """
+    supported = {"layer_sweep", "component", "trisect", "wormhole", "noise", "duets"}
+    if tool not in supported:
+        raise ValueError(f"Unknown surgery tool: {tool}")
+    path, layers = _layer_list(model)
+    total = len(layers)
+    try:
+        start = _integer(options, "layer_start", 0, 0, total - 1)
+        end = _integer(options, "layer_end", -1, -1, total - 1)
+        end = total - 1 if end == -1 else end
+        if end < start:
+            raise ValueError("end precedes start")
+    except ValueError as exc:
+        raise ValueError(f"Invalid layer range; use 0 through {total - 1}, with end >= start.") from exc
+    max_tokens = _integer(options, "max_tokens", 128, 2, 256)
+    max_new_tokens = _integer(options, "max_new_tokens", 16, 1, 40)
+    sample_count = _integer(options, "sample_layers", 8, 1, 16)
+    seed = _integer(options, "seed", 42, 0, 2**32 - 1)
+    text = str(options.get("text", ""))
+    prompt = str(options.get("prompt", "")).strip()
+    indices = list(range(start, end + 1))
+    selected = _sample(indices, min(sample_count, 6) if tool == "duets" else sample_count)
+    if tool in {"component", "trisect", "wormhole"}:
+        selected = indices
+    if tool == "duets" and len(selected) < 2:
+        raise ValueError("Layer Duets needs at least two selected layers.")
+    by_layer = _matrices(layers, selected)
+    warnings = [
+        "Results describe this evaluation text and this model precision; they do not establish general layer importance.",
+        "Matrix weights are edited in place; normalization, biases, residual routes, and block execution are retained.",
+    ]
+    if len(selected) < len(indices):
+        warnings.append(f"Sampled {len(selected)} of {len(indices)} layers in the requested range.")
+    selected_parameter_ids = {id(parameter) for values in by_layer.values() for _, parameter in values}
+    parameter_owners = {}
+    # Check the full layer list, including layers excluded by range/sampling.
+    # An alias outside the sample would otherwise receive an unreported cut.
+    for index, layer in enumerate(layers):
+        for parameter in layer.parameters():
+            if id(parameter) in selected_parameter_ids:
+                parameter_owners.setdefault(id(parameter), set()).add(index)
+    if any(len(owners) > 1 for owners in parameter_owners.values()):
+        raise ValueError("Selected layers share parameters; independent layer surgery is unsupported for this architecture.")
+
+    details = {
+        "layer_path": path, "total_layers": total, "selected_layers": selected,
+        "requested_range": [start, end], "sampled": len(selected) < len(indices),
+        "integrity_verified": False,
+        "scope": "2D-or-higher floating weight tensors in selected decoder layers; normalization and biases protected",
+        "loss_definition": "Mean next-token negative log likelihood over unpadded targets; computed in FP32",
+        "generation_format": "Raw prompt continuation, greedy decoding",
+    }
+    experiments = []
+    noise_scale = None
+    if tool in {"layer_sweep", "noise"}:
+        if tool == "noise":
+            try:
+                noise_scale = float(options.get("noise_scale", 0.01))
+            except (TypeError, ValueError) as exc:
+                raise ValueError("noise_scale must be a finite number between 0 and 1.") from exc
+            if not math.isfinite(noise_scale) or not 0 <= noise_scale <= 1:
+                raise ValueError("noise_scale must be a finite number between 0 and 1.")
+            details.update(noise_scale=noise_scale, seed=seed)
+            warnings.append("Gaussian weight noise is not quantization and cannot predict an INT4/INT8 bit budget.")
+        for index in selected:
+            experiments.append((f"Layer {index}", _collect(by_layer, [index]), {"layer": index}))
+    elif tool == "component":
+        components = {}
+        for index, entries in by_layer.items():
+            for name, parameter in entries:
+                component = name.rsplit(".", 1)[0] if name.endswith(".weight") else name
+                components.setdefault(component, []).append((f"{index}.{name}", parameter))
+        available = sorted(components)
+        requested = str(options.get("component", "All available")).strip()
+        if requested.casefold() in {"", "all", "all available"}:
+            chosen = available
+        elif requested in components:
+            chosen = [requested]
+        else:
+            chosen = [name for name in available if requested.casefold() in name.casefold()]
+        if not chosen:
+            raise ValueError(f"No matrix component matches {requested!r}. Available: {', '.join(available[:24])}")
+        if len(chosen) > 16:
+            warnings.append(f"Sampled 16 of {len(chosen)} matching component paths; refine the component field for a specific module.")
+            chosen = _sample(chosen, 16)
+        details["available_components"] = available[:128]
+        details["tested_components"] = chosen
+        for name in chosen:
+            experiments.append((name, _unique(components[name]), {"component": name}))
+    elif tool in {"trisect", "wormhole"}:
+        patterns = _zones(selected, tool)
+        requested = str(options.get("pattern", "")).strip()
+        if requested.casefold() not in {"", "all", "all patterns"}:
+            patterns = [entry for entry in patterns if requested.casefold() in entry[0].casefold()]
+        if not patterns:
+            raise ValueError("No nonempty cut pattern matches this range and pattern; choose a wider layer range or clear the pattern.")
+        for label, kept, zeroed in patterns:
+            experiments.append((label, _collect(by_layer, zeroed), {"kept_layers": kept, "zeroed_layers": zeroed}))
+        details["scope"] += "; layers outside the requested range stay intact; pattern names describe retained matrices"
+    else:
+        for index in selected:
+            experiments.append((f"Layer {index}", _collect(by_layer, [index]), {"layer": index}))
+        for a, b in combinations(selected, 2):
+            experiments.append((f"Layers {a} + {b}", _collect(by_layer, [a, b]), {"layer_a": a, "layer_b": b}))
+
+    modes = [(module, module.training) for module in model.modules()]
+    model.eval()
+    try:
+        batch, targets, scored_tokens = _prepare_batch(model, tokenizer, text, max_tokens)
+        baseline = _loss(model, batch, targets)
+        details.update(baseline_loss=baseline, scored_tokens=scored_tokens)
+        text_outputs = []
+        if prompt:
+            text_outputs.append({"label": "Original model", "text": _generate(model, tokenizer, prompt, max_tokens, max_new_tokens)})
+        measured = []
+        for label, parameters, fields in experiments:
+            with _temporary_edit(parameters, noise_scale=noise_scale, seed=seed) as edit:
+                loss = _loss(model, batch, targets)
+            row = {"name": label, **fields, "loss": loss, "damage": loss - baseline,
+                   "modified_tensors": edit["modified_tensors"]}
+            if noise_scale is not None:
+                row.update(noise_scale=noise_scale, changed_fraction=edit["changed_fraction"],
+                           relative_write_norm=edit["relative_write_norm"])
+            measured.append(row)
+        worst = max(range(len(measured)), key=lambda index: measured[index]["damage"])
+        if prompt:
+            label, parameters, _ = experiments[worst]
+            with _temporary_edit(parameters, noise_scale=noise_scale, seed=seed):
+                altered = _generate(model, tokenizer, prompt, max_tokens, max_new_tokens)
+            text_outputs.append({"label": f"Largest measured loss change: {label}", "text": altered})
+        details["integrity_verified"] = True
+        details["restoration_verification"] = "Exact tensor equality immediately after every intervention and generation comparison"
+        details["evaluation_forward_passes"] = 1 + len(experiments)
+        details["generation_comparisons"] = 2 if prompt else 0
+
+        titles = {
+            "layer_sweep": "Layer sweep", "component": "Component ablation",
+            "trisect": "Trisect: core and shell", "wormhole": "Wormhole patterns",
+            "noise": "Noise sensitivity", "duets": "Layer Duets",
+        }
+        if tool == "duets":
+            singles = {row["layer"]: row for row in measured if "layer" in row}
+            rows = []
+            matrix = [[None for _ in selected] for _ in selected]
+            for row in measured:
+                if "layer_a" not in row:
+                    continue
+                a, b = row["layer_a"], row["layer_b"]
+                da, db = singles[a]["damage"], singles[b]["damage"]
+                interaction = row["damage"] - da - db
+                rows.append({**row, "single_damage_a": da, "single_damage_b": db,
+                             "pair_damage": row["damage"], "interaction": interaction})
+                ai, bi = selected.index(a), selected.index(b)
+                matrix[ai][bi] = matrix[bi][ai] = interaction
+            details["singles"] = list(singles.values())
+            details["interaction_definition"] = "interaction = (pair loss - baseline) - (single A loss - baseline) - (single B loss - baseline)"
+            plot = {"kind": "heatmap", "title": "Excess pair damage (positive: more than additive)",
+                    "x_label": "Layer", "y_label": "Layer", "x": [f"L{i}" for i in selected],
+                    "y": [f"L{i}" for i in selected], "z": matrix}
+            summary = (
+                f"Tested {len(singles)} single cuts and {len(rows)} pairs. Positive interaction means "
+                "the pair raises loss more than the sum of its individual cuts; negative means less. "
+                "This measures non-additivity on the chosen text, not a proven functional partnership."
+            )
+        else:
+            rows = measured
+            x_values = [row["layer"] if "layer" in row else row["name"] for row in rows]
+            plot = {"kind": "line" if tool in {"layer_sweep", "noise"} else "bar",
+                    "title": "Loss change from original model", "x_label": "Layer" if tool in {"layer_sweep", "noise"} else "Intervention",
+                    "y_label": "Change in mean NLL", "series": [{"name": "Loss change", "x": x_values, "y": [row["damage"] for row in rows]}]}
+            summary = (
+                f"Original loss: {baseline:.4f}. Tested {len(rows)} interventions; "
+                f"largest measured loss change: {measured[worst]['damage']:+.4f} "
+                f"({measured[worst]['name']}). Every modified tensor was restored and checked."
+            )
+        if noise_scale is not None and any(row["changed_fraction"] == 0 for row in rows):
+            warnings.append("At least one noise edit changed no stored values; inspect changed_fraction before interpreting zero loss change.")
+        return {"title": titles[tool], "summary": summary, "rows": rows, "plot": plot,
+                "details": details, "warnings": warnings, "text_outputs": text_outputs}
+    finally:
+        for module, training in modes:
+            module.training = training

requirements.txt

@@ -0,0 +1,9 @@
+torch==2.11.0
+transformers==5.16.1
+accelerate>=1.12.0
+numpy>=2.0,<3
+scipy==1.18.1
+pandas==3.0.5
+plotly==7.0.0
+safetensors>=0.6.2
+sentencepiece>=0.2.0

tests/test_app_flow.py

@@ -0,0 +1,130 @@
+"""Exercise app callbacks and state bindings without downloading a checkpoint.
+
+The UI module starts its default model at import. Extracting its real callback
+definitions lets these regression tests run independently of that GPU startup;
+the deployed Gradio client smoke test covers the actual component integration.
+"""
+
+import ast
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+
+SOURCE = Path(__file__).resolve().parents[1] / "app.py"
+TREE = ast.parse(SOURCE.read_text())
+
+
+def binding(callback):
+    for node in ast.walk(TREE):
+        if (isinstance(node, ast.Call) and node.args
+                and isinstance(node.args[0], ast.Name) and node.args[0].id == callback):
+            return {keyword.arg: keyword.value for keyword in node.keywords}
+    raise AssertionError(f"No event is bound to {callback}")
+
+
+def names(node):
+    return [node.id] if isinstance(node, ast.Name) else [item.id for item in node.elts]
+
+
+@pytest.fixture
+def callbacks():
+    definitions = [node for node in TREE.body if isinstance(node, ast.FunctionDef)
+                   or (isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "DESCRIPTIONS"
+                                                           for target in node.targets))]
+    namespace = {
+        "gr": SimpleNamespace(update=lambda **kwargs: kwargs, Error=ValueError),
+        "validate_options": Mock(return_value={"seed": 42}),
+        "prepare_checkpoint": Mock(return_value={"repo_id": "test/model", "revision": "abc"}),
+        "model_summary": Mock(return_value="Ready"),
+        "render_result": Mock(return_value=("Summary", "Figure", "Table", "Comparison")),
+        "create_report": Mock(return_value=("experiment.json", "experiment.html")),
+        "save_contribution": Mock(return_value="Saved"),
+    }
+    exec(compile(ast.Module(body=definitions, type_ignores=[]), str(SOURCE), "exec"), namespace)
+    return namespace
+
+
+def prepare_arguments():
+    return ("test/model", "main", "VISITOR_TOKEN", "silhouette", "Main text", "Comparison",
+            "Generation", "", 128, 16, 8, 0, -1, "All available", 0.01, 42, " MAPLE", " RAVEN")
+
+
+def test_begin_run_clears_prior_state_permissions_and_visible_results(callbacks):
+    values = dict(zip(names(binding("begin_run")["outputs"]), callbacks["begin_run"](), strict=True))
+    for key in ("request_state", "computed_state", "report_state", "result_json", "chart", "json_report", "html_report"):
+        assert values[key] is None
+    assert values["table"] == []
+    assert values["comparisons"] == ""
+    assert values["consent"] is False
+    assert values["include_text"] is False
+    assert values["save"]["interactive"] is False
+    assert values["run_button"]["interactive"] is False
+    assert ast.literal_eval(binding("begin_run")["queue"]) is False
+
+
+def test_api_prepare_invalidates_prior_result_before_download_or_failure(callbacks):
+    callbacks["prepare_checkpoint"].side_effect = ValueError("No access")
+    generator = callbacks["prepare_request"](*prepare_arguments())
+    initial = dict(zip(names(binding("prepare_request")["outputs"]), next(generator), strict=True))
+    callbacks["prepare_checkpoint"].assert_not_called()
+    assert all(initial[key] is None for key in ("request_state", "computed_state", "report_state"))
+    assert initial["consent"] is False
+    assert initial["include_text"] is False
+    assert initial["save"]["interactive"] is False
+    with pytest.raises(ValueError, match="No access"):
+        next(generator)
+
+
+def test_prepare_keeps_18_controls_and_excludes_visitor_token_from_state(callbacks):
+    result = list(callbacks["prepare_request"](*prepare_arguments()))[-1]
+    values = dict(zip(names(binding("prepare_request")["outputs"]), result, strict=True))
+    assert len(names(binding("prepare_request")["inputs"])) == 1  # The named controls list.
+    controls = next(node.value for node in ast.walk(TREE) if isinstance(node, ast.Assign)
+                    and any(isinstance(target, ast.Name) and target.id == "controls" for target in node.targets))
+    assert len(controls.elts) == 18
+    assert "VISITOR_TOKEN" not in repr(values)
+    assert values["request_state"]["tool"] == "silhouette"
+    assert values["report_state"] is None
+
+
+def test_report_can_only_consume_server_computed_state():
+    assert names(binding("analyze_gpu")["inputs"]) == ["request_state"]
+    assert names(binding("analyze_gpu")["outputs"]) == ["computed_state"]
+    assert names(binding("finish_report")["inputs"]) == ["computed_state"]
+    assert "result_json" not in names(binding("finish_report")["inputs"])
+
+
+def test_report_reenables_contributions_only_for_completed_measurements(callbacks):
+    result = {"run": {"id": "completed-run"}, "title": "Measured"}
+    values = dict(zip(names(binding("finish_report")["outputs"]), callbacks["finish_report"](result), strict=True))
+    assert values["result_json"] == result
+    assert values["report_state"] == result
+    assert values["consent"] is False
+    assert values["include_text"] is False
+    assert values["save"]["interactive"] is True
+    assert values["run_button"]["interactive"] is True
+    with pytest.raises(ValueError, match="no completed measurements"):
+        callbacks["finish_report"](None)
+
+
+def test_failed_stage_clears_every_result_state_and_requires_fresh_consent(callbacks):
+    values = dict(zip(names(binding("failed_run")["outputs"]), callbacks["failed_run"](), strict=True))
+    assert all(values[key] is None for key in ("request_state", "computed_state", "report_state"))
+    assert values["consent"] is False
+    assert values["include_text"] is False
+    assert values["save"]["interactive"] is False
+    assert values["run_button"]["interactive"] is True
+
+
+def test_tool_controls_match_the_inputs_actually_used(callbacks):
+    silhouette = callbacks["describe_tool"]("silhouette")
+    assert silhouette[3]["visible"] is True  # grouped words
+    assert silhouette[4]["visible"] is False  # unused main text
+    tug = callbacks["describe_tool"]("tugofwar")
+    assert tug[1]["visible"] is False  # engine does not use comparison text
+    assert tug[2]["visible"] is True
+    assert tug[4]["label"] == "Context preamble"
+    assert callbacks["describe_tool"]("butterfly")[1]["visible"] is True

tests/test_contributions.py

@@ -0,0 +1,44 @@
+import json
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from bol_space.contributions import save_contribution
+
+
+def result():
+    return {"title": "Test", "summary": "PRIVATE_TEXT_CANARY", "model": {"repo_id": "owner/model", "revision": "abc", "dtype": "bfloat16", "architecture": "llama", "parameters": 10},
+            "run": {"id": "fixture", "tool": "gradient", "options": {"text": "PRIVATE_TEXT_CANARY", "max_tokens": 128}},
+            "rows": [{"layer": 0, "gradient_norm": 0.5}], "plot": {}, "text_outputs": [{"label": "prompt", "text": "PRIVATE_TEXT_CANARY"}]}
+
+
+def test_no_consent_never_connects_to_hub():
+    with patch("bol_space.contributions.HfApi") as api:
+        with pytest.raises(ValueError, match="permission"):
+            save_contribution(result(), False, True)
+        api.assert_not_called()
+
+
+def test_public_dataset_never_receives_contribution():
+    api = MagicMock()
+    api.dataset_info.return_value = SimpleNamespace(private=False)
+    with patch.dict("os.environ", {"BOL_CONTRIBUTION_REPO": "owner/private", "BOL_CONTRIBUTION_TOKEN": "unit-test-placeholder"}), patch("bol_space.contributions.HfApi", return_value=api):
+        with pytest.raises(ValueError, match="not private"):
+            save_contribution(result(), True, False)
+        api.upload_file.assert_not_called()
+
+
+def test_consented_measurements_exclude_text_and_record_scope():
+    api = MagicMock()
+    api.dataset_info.return_value = SimpleNamespace(private=True)
+    with patch.dict("os.environ", {"BOL_CONTRIBUTION_REPO": "owner/private", "BOL_CONTRIBUTION_TOKEN": "unit-test-placeholder"}), patch("bol_space.contributions.HfApi", return_value=api):
+        status = save_contribution(result(), True, False)
+        kwargs = api.upload_file.call_args.kwargs
+        assert kwargs["repo_type"] == "dataset"
+        assert b"PRIVATE_TEXT_CANARY" not in kwargs["path_or_fileobj"]
+        payload = json.loads(kwargs["path_or_fileobj"])
+        assert payload["consent"]["measurements"] is True
+        assert payload["consent"]["include_text"] is False
+        assert payload["data"]["rows"] == [{"layer": 0, "gradient_norm": 0.5}]
+        assert "Saved privately" in status

tests/test_inspection.py

@@ -0,0 +1,240 @@
+"""Numerical and state-restoration checks using a tiny local causal network."""
+import json
+import importlib.util
+import types
+import unittest
+
+import torch
+from torch import nn
+
+from bol_space.inspection import _linear_cka, run_inspection
+
+
+class CharacterTokenizer:
+    def __call__(self, text, return_tensors=None, truncation=False, max_length=None,
+                 add_special_tokens=True, **kwargs):
+        ids = ([1] if add_special_tokens else []) + [2 + ord(c) % 61 for c in text]
+        if truncation and max_length:
+            ids = ids[:max_length]
+        if return_tensors == "pt":
+            ids = torch.tensor([ids], dtype=torch.long)
+            return {"input_ids": ids, "attention_mask": torch.ones_like(ids)}
+        return {"input_ids": ids}
+
+    def decode(self, ids, **kwargs):
+        return " ".join(str(int(i)) for i in ids)
+
+    def convert_ids_to_tokens(self, ids):
+        return [str(i) for i in ids]
+
+
+class TinyCausalModel(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.embeddings = nn.Embedding(64, 8)
+        self.model = nn.Module()
+        self.model.layers = nn.ModuleList([nn.Linear(8, 8), nn.Linear(8, 8)])
+        self.lm_head = nn.Linear(8, 64)
+        self.config = types.SimpleNamespace(num_hidden_layers=2)
+        self.fail = False
+
+    def get_input_embeddings(self):
+        return self.embeddings
+
+    def forward(self, input_ids, attention_mask=None, output_hidden_states=False,
+                output_attentions=False, use_cache=False, labels=None, **kwargs):
+        if self.fail:
+            raise RuntimeError("injected failure")
+        h = self.embeddings(input_ids)
+        states = [h]
+        n = h.shape[1]
+        denom = torch.arange(1, n + 1, device=h.device)[None, :, None]
+        for layer in self.model.layers:
+            h = h + torch.tanh(layer(h.cumsum(1) / denom))
+            states.append(h)
+        logits = self.lm_head(h)
+        loss = None
+        if labels is not None:
+            loss = torch.nn.functional.cross_entropy(
+                logits[:, :-1].float().reshape(-1, 64), labels[:, 1:].reshape(-1),
+                ignore_index=-100,
+            )
+        attn = torch.ones(n, n, device=h.device).tril()
+        attn /= attn.sum(-1, keepdim=True)
+        return types.SimpleNamespace(
+            hidden_states=tuple(states) if output_hidden_states else None,
+            attentions=tuple(attn[None, None].repeat(1, 2, 1, 1) for _ in range(2))
+            if output_attentions else None,
+            logits=logits, loss=loss,
+        )
+
+
+class InspectionTests(unittest.TestCase):
+    def setUp(self):
+        with torch.random.fork_rng():
+            torch.manual_seed(7)
+            self.model = TinyCausalModel().eval()
+        self.tokenizer = CharacterTokenizer()
+
+    def test_cka_handles_geometry_and_degeneracy(self):
+        x = torch.tensor([[1., 0.], [0., 2.], [-2., 1.], [1., -1.]])
+        rotated = torch.stack([-x[:, 1], x[:, 0]], dim=1) * 4 + 12
+        self.assertAlmostEqual(_linear_cka(x, rotated), 1., places=5)
+        self.assertAlmostEqual(_linear_cka(x, x * 1e-8), 1., places=5)
+        self.assertIsNone(_linear_cka(torch.ones(4, 2), x))
+
+    def test_default_silhouette_uses_original_related_and_unrelated_pairs(self):
+        related = [("dog", "wolf"), ("cat", "lion"), ("river", "ocean"),
+                   ("happy", "joyful"), ("cold", "frozen"), ("king", "queen"),
+                   ("python", "code"), ("car", "vehicle"), ("run", "sprint")]
+        unrelated = [("dog", "democracy"), ("cat", "algebra"), ("river", "justice"),
+                     ("happy", "concrete"), ("cold", "philosophy"), ("king", "equation"),
+                     ("python", "banana"), ("car", "silence"), ("run", "orange")]
+        # Compute the expected measurement independently from direct model
+        # outputs, so substituting generic cluster means breaks this contract.
+        vectors = {}
+        with torch.no_grad():
+            for word in set(w for pair in related + unrelated for w in pair):
+                out = self.model(**self.tokenizer(word, return_tensors="pt"), output_hidden_states=True)
+                vectors[word] = [state[0, -1].float() for state in out.hidden_states[1:]]
+        result = run_inspection(self.model, self.tokenizer, "silhouette", {})
+        for row in result["rows"]:
+            layer = row["layer"]
+            means = [sum(float(torch.nn.functional.cosine_similarity(
+                vectors[a][layer], vectors[b][layer], dim=0)) for a, b in pairs) / 9
+                for pairs in (related, unrelated)]
+            self.assertAlmostEqual(row["within_cosine"], means[0], places=6)
+            self.assertAlmostEqual(row["cross_cosine"], means[1], places=6)
+            self.assertAlmostEqual(row["separation"], means[0] - means[1], places=6)
+        self.assertEqual(result["details"]["related_pairs"], [list(pair) for pair in related])
+        self.assertEqual(result["details"]["unrelated_pairs"], [list(pair) for pair in unrelated])
+
+    def test_custom_silhouette_still_uses_explicit_groups(self):
+        result = run_inspection(self.model, self.tokenizer, "silhouette", {"words": "dog, wolf\ncar, truck"})
+        self.assertTrue(result["details"]["custom_probes"])
+        self.assertEqual(result["details"]["groups"], {"Group 1": ["dog", "wolf"], "Group 2": ["car", "truck"]})
+
+    def test_attention_uniform_causal_rows_have_unit_normalized_entropy(self):
+        r = run_inspection(self.model, self.tokenizer, "attention", {"text": "abcdef"})
+        for row in r["rows"]:
+            self.assertAlmostEqual(row["normalized_entropy"], 1., places=5)
+            self.assertAlmostEqual(row["head_diversity"], 0., places=5)
+
+    def test_gradient_does_not_change_weights_flags_or_existing_grads(self):
+        params = list(self.model.parameters())
+        params[0].requires_grad_(False)
+        params[-1].grad = torch.ones_like(params[-1])
+        flags = [p.requires_grad for p in params]
+        weights = [p.detach().clone() for p in params]
+        original_grad = params[-1].grad
+        with torch.no_grad():
+            r = run_inspection(self.model, self.tokenizer, "gradient", {"text": "abcdef"})
+        self.assertTrue(any(row["gradient_norm"] > 0 for row in r["rows"]))
+        self.assertEqual(flags, [p.requires_grad for p in params])
+        self.assertIs(params[-1].grad, original_grad)
+        for old, p in zip(weights, params):
+            self.assertTrue(torch.equal(old, p))
+        self.assertTrue(all(p.grad is None for p in params[:-1]))
+
+    def test_gradient_restores_flags_on_failure(self):
+        params = list(self.model.parameters())
+        params[0].requires_grad_(False)
+        flags = [p.requires_grad for p in params]
+        self.model.fail = True
+        with self.assertRaisesRegex(RuntimeError, "injected"):
+            run_inspection(self.model, self.tokenizer, "gradient", {"text": "abcdef"})
+        self.assertEqual(flags, [p.requires_grad for p in params])
+
+    def test_butterfly_compares_matching_tokens_after_edit(self):
+        r = run_inspection(self.model, self.tokenizer, "butterfly", {
+            "text": "cat walks home", "compare_text": "dog walks home", "max_tokens": 40,
+        })
+        self.assertGreater(r["details"]["aligned_token_count"], 0)
+        self.assertTrue(any(row["cosine_distance"] > 0 for row in r["rows"]))
+        for pair in r["details"]["alignment"]:
+            self.assertEqual(pair["token_a"], pair["token_b"])
+            self.assertGreater(pair["position_a"], 1)
+
+    def test_identical_butterfly_inputs_measure_zero(self):
+        r = run_inspection(self.model, self.tokenizer, "butterfly", {
+            "text": "cat walks home", "compare_text": "cat walks home",
+        })
+        self.assertTrue(all(abs(row["cosine_distance"]) < 1e-6 for row in r["rows"]))
+
+    def test_tugofwar_scores_entire_multitoken_candidates(self):
+        r = run_inspection(self.model, self.tokenizer, "tugofwar", {
+            "choice_a": " MAPLE", "choice_b": " RAVEN", "max_tokens": 256,
+        })
+        self.assertEqual(len(r["rows"]), 4)
+        for row in r["rows"]:
+            self.assertEqual(row["candidate_a_tokens"], 6)
+            self.assertEqual(row["candidate_b_tokens"], 6)
+            self.assertLess(row["a_mean_logprob"], 0)
+
+    def test_tugofwar_candidates_share_context_when_lengths_differ(self):
+        class RecordingModel(TinyCausalModel):
+            def __init__(self):
+                super().__init__()
+                self.seen = []
+
+            def forward(self, input_ids, **kwargs):
+                self.seen.append(input_ids[0].tolist())
+                return super().forward(input_ids=input_ids, **kwargs)
+
+        model = RecordingModel().eval()
+        run_inspection(model, self.tokenizer, "tugofwar", {
+            "choice_a": "A", "choice_b": "BANANA", "text": "background " * 60, "max_tokens": 128,
+        })
+        for a, b in zip(model.seen[::2], model.seen[1::2]):
+            self.assertEqual(a[:-2], b[:-7])
+            self.assertLessEqual(len(a), 128)
+            self.assertLessEqual(len(b), 128)
+
+    def test_tugofwar_refuses_to_truncate_experimental_facts(self):
+        with self.assertRaisesRegex(ValueError, "complete conflicting facts"):
+            run_inspection(self.model, self.tokenizer, "tugofwar", {"max_tokens": 32})
+
+    def test_gradient_masks_padding_by_mask_and_preserves_real_eos_tokens(self):
+        class PaddedTokenizer(CharacterTokenizer):
+            pad_token_id = 1
+
+            def __call__(self, text, **kwargs):
+                return {"input_ids": torch.tensor([[1, 4, 1, 5, 1]]),
+                        "attention_mask": torch.tensor([[1, 1, 1, 1, 0]])}
+
+        tokenizer = PaddedTokenizer()
+        inputs = tokenizer("unused")
+        with torch.no_grad():
+            expected = float(self.model(**inputs, labels=torch.tensor([[1, 4, 1, 5, -100]])).loss)
+        result = run_inspection(self.model, tokenizer, "gradient", {"text": "unused"})
+        self.assertAlmostEqual(result["details"]["loss"], expected, places=6)
+
+    def test_tools_return_finite_serializable_results(self):
+        for tool in ("fingerprint", "silhouette", "semantic", "cka", "attention", "gradient"):
+            with self.subTest(tool=tool):
+                r = run_inspection(self.model, self.tokenizer, tool, {"text": "abcdef"})
+                self.assertTrue(r["rows"])
+                json.dumps(r, allow_nan=False)
+
+    @unittest.skipUnless(importlib.util.find_spec("transformers"), "Transformers is optional for local unit checks")
+    def test_real_bf16_llama_contract_without_download(self):
+        from transformers import LlamaConfig, LlamaForCausalLM
+        with torch.random.fork_rng():
+            torch.manual_seed(42)
+            model = LlamaForCausalLM(LlamaConfig(
+                vocab_size=64, hidden_size=16, intermediate_size=24,
+                num_hidden_layers=2, num_attention_heads=2, num_key_value_heads=1,
+                max_position_embeddings=256, attn_implementation="eager",
+            )).to(torch.bfloat16).eval()
+        for tool in ("fingerprint", "silhouette", "semantic", "cka", "attention", "gradient", "butterfly", "tugofwar"):
+            with self.subTest(tool=tool):
+                r = run_inspection(model, self.tokenizer, tool, {
+                    "text": "cat walks home", "compare_text": "dog walks home", "max_tokens": 256,
+                })
+                self.assertTrue(r["rows"])
+                json.dumps(r, allow_nan=False)
+                self.assertTrue(all(p.grad is None for p in model.parameters()))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_presentation.py

@@ -0,0 +1,168 @@
+"""Regression checks for report execution safety and consent boundaries."""
+
+import copy
+import json
+import math
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from bol_space.presentation import contribution_payload, create_report, render_result
+
+
+def example_result():
+    return {
+        "title": "Layer sweep", "summary": "Compare selected layer interventions.",
+        "rows": [{"layer": 1, "loss": 2.2, "damage": 0.15}],
+        "plot": {"kind": "line", "title": "Loss by layer", "x_label": "Layer", "y_label": "Loss",
+                 "series": [{"name": "Loss", "x": [1, 2], "y": [2.2, 2.3]}]},
+        "details": {"baseline_loss": 2.05}, "warnings": [],
+        "text_outputs": [{"label": "Prompt", "text": "The sky is"}],
+        "model": {"repo_id": "HuggingFaceTB/SmolLM2-1.7B-Instruct", "revision": "a" * 40,
+                  "architecture": "LlamaForCausalLM", "dtype": "bfloat16", "parameters": 1710000000},
+        "run": {"id": "0a234567-abcd-abcd-abcd-0123456789ab", "created_at": "2026-09-06T12:00:00+00:00",
+                "tool": "layer_sweep", "seconds": 2.7, "peak_vram_gb": 4.2,
+                "options": {"seed": 42, "max_tokens": 128, "sample_layers": 4}},
+    }
+
+
+class PresentationTests(unittest.TestCase):
+    def setUp(self):
+        self.directories = set()
+
+    def tearDown(self):
+        for directory in self.directories:
+            shutil.rmtree(directory)
+
+    def report(self, result):
+        paths = create_report(result)
+        self.directories.add(Path(paths[0]).parent)
+        return paths
+
+    def test_report_escapes_html_and_script_terminators(self):
+        result = example_result()
+        attack = '</script><script>window.HACKED="yes"</script><img src=x onerror=alert(1)>'
+        result["title"] = attack
+        result["summary"] = attack
+        result["rows"][0][attack] = attack
+        result["plot"]["series"][0]["name"] = attack
+        result["plot"]["series"][0]["x"] = [attack, attack]
+        result["plot"]["title"] = attack
+        result["details"][attack] = attack
+        result["text_outputs"] = [{"label": attack, "text": attack}]
+        json_path, html_path = self.report(result)
+        document = Path(html_path).read_text()
+        self.assertNotIn(attack, document)
+        self.assertNotIn('<script>window.HACKED=', document)
+        self.assertNotIn('<img src=x', document)
+        self.assertIn('&lt;/script&gt;', document)
+        self.assertNotRegex(document, r'<script\s+[^>]*src\s*=')
+        self.assertIn('Plotly.newPlot', document)
+        self.assertIn('@media print', document)
+        self.assertEqual(json.loads(Path(json_path).read_text())["title"], attack)
+        markdown, figure, frame, texts = render_result(result)
+        self.assertNotIn('<img src=x', markdown + texts)
+        self.assertNotIn('</script>', figure.to_json())
+        self.assertEqual(len(frame), 1)
+
+    def test_text_free_payload_has_no_custom_text_through_any_label(self):
+        result = example_result()
+        canary = "CUSTOM_CANARY_829740"
+        result["title"] = canary
+        result["summary"] = canary
+        result["details"] = {canary: [{"nested": canary}]}
+        result["warnings"] = [canary]
+        result["rows"][0].update({canary: 2, "component": canary, "name": canary, "loss": canary, "tokens": [5, 42]})
+        result["plot"].update({"title": canary, "x_label": canary, "y_label": canary})
+        result["plot"]["series"][0].update({"name": canary, "x": [canary, canary]})
+        result["text_outputs"] = [{"label": canary, "text": canary}]
+        result["run"]["options"].update({canary: 4, "prompt": canary, "component": canary, "layer_start": canary})
+        original = copy.deepcopy(result)
+        payload = contribution_payload(result, False)
+        encoded = json.dumps(payload, allow_nan=False)
+        self.assertNotIn(canary, encoded)
+        self.assertNotIn('"tokens"', encoded)
+        self.assertFalse(payload["includes_text"])
+        self.assertEqual(payload["consent_version"], "1")
+        self.assertEqual(payload["data"]["run"]["options"]["seed"], 42)
+        self.assertEqual(payload["data"]["rows"][0]["damage"], 0.15)
+        self.assertEqual(payload["data"]["plot"]["series"][0]["x"], [0, 1])
+        self.assertEqual(result, original)
+
+    def test_text_free_heatmap_keeps_measurements_without_labels(self):
+        result = example_result()
+        result["plot"] = {"kind": "heatmap", "x": ["SECRET_X"], "y": ["SECRET_Y"], "z": [[0.3]], "title": "SECRET_TITLE"}
+        plot = contribution_payload(result, False)["data"]["plot"]
+        self.assertEqual(plot, {"kind": "heatmap", "x": [0], "y": [0], "z": [[0.3]]})
+        self.assertEqual(len(render_result(result)[1].data), 1)
+
+    def test_text_consent_keeps_text_but_not_unknown_top_level_or_credentials(self):
+        result = example_result()
+        result["private_metadata"] = "UNEXPECTED_SECRET"
+        result["run"]["options"]["hf_token"] = "NEVER_EXPORT"
+        result["run"]["options"]["prompt"] = "An explicitly shared prompt"
+        result["model"]["authorization"] = "NEVER_EXPORT"
+        payload = contribution_payload(result, True)
+        self.assertTrue(payload["includes_text"])
+        self.assertEqual(payload["data"]["text_outputs"], result["text_outputs"])
+        self.assertIn("An explicitly shared prompt", json.dumps(payload))
+        self.assertNotIn("NEVER_EXPORT", json.dumps(payload))
+        self.assertNotIn("UNEXPECTED_SECRET", json.dumps(payload))
+
+    def test_unique_files_and_strict_json(self):
+        result = example_result()
+        first = self.report(result)
+        second = self.report(result)
+        self.assertTrue(set(first).isdisjoint(second))
+        for filename in (first[0], second[0]):
+            decoded = json.loads(Path(filename).read_text(), parse_constant=lambda value: self.fail(value))
+            self.assertEqual(decoded, result)
+
+    def test_original_reports_live_inside_the_configured_gradio_cache(self):
+        with tempfile.TemporaryDirectory() as cache_root:
+            with patch.dict(os.environ, {"GRADIO_TEMP_DIR": cache_root}):
+                paths = create_report(example_result())
+            for path in paths:
+                self.assertTrue(Path(path).is_relative_to(cache_root))
+                self.assertTrue(Path(path).is_file())
+
+    def test_partial_report_files_are_removed_when_writing_fails(self):
+        original_write = Path.write_text
+
+        def write_or_fail(path, *args, **kwargs):
+            if path.suffix == ".html":
+                raise OSError("Simulated full disk")
+            return original_write(path, *args, **kwargs)
+
+        with tempfile.TemporaryDirectory() as cache_root:
+            with patch.dict(os.environ, {"GRADIO_TEMP_DIR": cache_root}), patch.object(Path, "write_text", write_or_fail):
+                with self.assertRaisesRegex(OSError, "full disk"):
+                    create_report(example_result())
+            self.assertEqual(list(Path(cache_root).iterdir()), [])
+
+    def test_nonfinite_values_are_rejected(self):
+        for bad in (math.nan, math.inf, -math.inf):
+            for function in (render_result, create_report, lambda data: contribution_payload(data, False)):
+                with self.subTest(bad=bad, function=function):
+                    result = example_result()
+                    result["rows"][0]["loss"] = bad
+                    with self.assertRaisesRegex(ValueError, "finite"):
+                        function(result)
+
+    def test_missing_plot_and_rows_still_render(self):
+        summary, figure, frame, text = render_result({"title": "A result", "summary": "Complete"})
+        self.assertIn("A result", summary)
+        self.assertEqual(len(figure.layout.annotations), 1)
+        self.assertTrue(frame.empty)
+        self.assertEqual(text, "")
+
+    def test_consent_requires_actual_boolean(self):
+        with self.assertRaises(TypeError):
+            contribution_payload(example_result(), "false")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_surgery.py

@@ -0,0 +1,179 @@
+"""CPU contract tests; no checkpoint downloads are required."""
+
+import json
+import math
+from types import SimpleNamespace
+
+import pytest
+import torch
+from torch import nn
+
+from bol_space.surgery import run_surgery
+
+
+class TinyTokenizer:
+    # EOS and padding share an ID, as on many causal-model tokenizers.
+    pad_token_id = 0
+    eos_token_id = 0
+
+    def __call__(self, text, *, return_tensors, truncation, max_length, padding=False):
+        texts = [text] if isinstance(text, str) else text
+        lengths = [min(len(item.split()), max_length) for item in texts]
+        width = max(lengths)
+        return {
+            "input_ids": torch.zeros((len(texts), width), dtype=torch.long),
+            "attention_mask": torch.tensor(
+                [[1] * length + [0] * (width - length) for length in lengths]
+            ),
+        }
+
+    def decode(self, tokens, skip_special_tokens=True):
+        return "generated text"
+
+
+class TinyModel(nn.Module):
+    def __init__(self, count=4, fail_when_edited=False):
+        super().__init__()
+        self.model = nn.Module()
+        self.model.layers = nn.ModuleList()
+        self.fail_when_edited = fail_when_edited
+        for _ in range(count):
+            layer = nn.Module()
+            layer.proj = nn.Linear(2, 2)
+            layer.norm = nn.LayerNorm(2)
+            nn.init.ones_(layer.proj.weight)
+            self.model.layers.append(layer)
+        self.bfloat16()
+
+    def forward(self, input_ids, attention_mask, use_cache=False):
+        if self.fail_when_edited and any(
+            not bool(layer.proj.weight.count_nonzero()) for layer in self.model.layers
+        ):
+            raise RuntimeError("injected forward failure")
+        score = sum(layer.proj.weight.float().mean() for layer in self.model.layers)
+        logits = torch.stack([score, torch.zeros_like(score)])
+        return SimpleNamespace(logits=logits.expand(*input_ids.shape, 2))
+
+    def generate(self, **kwargs):
+        return kwargs["input_ids"]
+
+
+def options(**kwargs):
+    return {"text": "one two three\none two", "prompt": "", **kwargs}
+
+
+def snapshot(model):
+    return {name: tensor.detach().clone() for name, tensor in model.named_parameters()}
+
+
+def assert_restored(model, before):
+    for name, tensor in model.named_parameters():
+        assert torch.equal(tensor, before[name]), name
+
+
+def test_forward_error_restores_modified_weights_and_training_state():
+    model = TinyModel(fail_when_edited=True)
+    model.train()
+    before = snapshot(model)
+    with pytest.raises(RuntimeError, match="injected forward failure"):
+        run_surgery(model, TinyTokenizer(), "layer_sweep", options(sample_layers=2))
+    assert_restored(model, before)
+    assert model.training
+
+
+@pytest.mark.parametrize("tool", ["layer_sweep", "component", "trisect", "wormhole", "noise", "duets"])
+def test_every_tool_returns_finite_json_and_restores_weights(tool):
+    model = TinyModel()
+    before = snapshot(model)
+    result = run_surgery(model, TinyTokenizer(), tool, options(sample_layers=2))
+    assert result["rows"]
+    assert result["details"]["integrity_verified"] is True
+    assert_restored(model, before)
+    json.dumps(result, allow_nan=False)
+
+
+def test_eos_tokens_are_scored_even_when_eos_is_the_padding_token():
+    # Every real target token is EOS=0. Masking by token ID would erase them all.
+    result = run_surgery(TinyModel(), TinyTokenizer(), "layer_sweep", options(sample_layers=1))
+    assert result["details"]["baseline_loss"] == pytest.approx(math.log1p(math.exp(-4)), abs=1e-6)
+    assert result["details"]["scored_tokens"] == 3
+
+
+def test_duets_measure_excess_pair_damage_with_correct_sign():
+    result = run_surgery(
+        TinyModel(), TinyTokenizer(), "duets",
+        options(sample_layers=2, layer_start=0, layer_end=1),
+    )
+    # Logit for target class is 4 intact, 3 after one cut, 2 after both cuts.
+    baseline = math.log1p(math.exp(-4))
+    single = math.log1p(math.exp(-3)) - baseline
+    pair = math.log1p(math.exp(-2)) - baseline
+    row = result["rows"][0]
+    assert (row["layer_a"], row["layer_b"]) == (0, 1)
+    assert row["interaction"] == pytest.approx(pair - 2 * single, abs=1e-6)
+    assert row["interaction"] > 0
+    assert result["plot"]["z"][0][1] == result["plot"]["z"][1][0]
+
+
+def test_unknown_component_fails_without_changing_weights():
+    model = TinyModel()
+    before = snapshot(model)
+    with pytest.raises(ValueError, match="No matrix component"):
+        run_surgery(model, TinyTokenizer(), "component", options(component="does_not_exist"))
+    assert_restored(model, before)
+
+
+@pytest.mark.parametrize("start,end", [(-1, 2), (3, 1), (0, 4)])
+def test_invalid_layer_ranges_are_rejected(start, end):
+    with pytest.raises(ValueError, match="layer range"):
+        run_surgery(TinyModel(), TinyTokenizer(), "layer_sweep", options(layer_start=start, layer_end=end))
+
+
+def test_noise_is_reproducible_without_changing_global_rng():
+    model = TinyModel()
+    rng = torch.random.get_rng_state().clone()
+    first = run_surgery(model, TinyTokenizer(), "noise", options(sample_layers=2, noise_scale=0.1))
+    second = run_surgery(model, TinyTokenizer(), "noise", options(sample_layers=2, noise_scale=0.1))
+    assert first["rows"] == second["rows"]
+    assert torch.equal(rng, torch.random.get_rng_state())
+
+
+def test_models_without_a_recognized_layer_list_fail_explicitly():
+    with pytest.raises(ValueError, match="layer list"):
+        run_surgery(nn.Linear(2, 2), TinyTokenizer(), "layer_sweep", options())
+
+
+def test_shared_weights_in_an_unsampled_layer_are_rejected():
+    model = TinyModel()
+    model.model.layers[3].proj.weight = model.model.layers[0].proj.weight
+    with pytest.raises(ValueError, match="share parameters"):
+        run_surgery(model, TinyTokenizer(), "layer_sweep", options(layer_start=0, layer_end=0))
+
+
+def test_generation_error_after_a_cut_restores_weights():
+    class GenerationFailure(TinyModel):
+        def generate(self, **kwargs):
+            if any(not bool(layer.proj.weight.count_nonzero()) for layer in self.model.layers):
+                raise RuntimeError("injected generation failure")
+            return super().generate(**kwargs)
+
+    model = GenerationFailure()
+    before = snapshot(model)
+    with pytest.raises(RuntimeError, match="injected generation failure"):
+        run_surgery(model, TinyTokenizer(), "layer_sweep", options(sample_layers=2, prompt="one two"))
+    assert_restored(model, before)
+
+
+def test_nonfinite_intervention_loss_raises_and_restores():
+    class NonfiniteModel(TinyModel):
+        def forward(self, **kwargs):
+            output = super().forward(**kwargs)
+            if any(not bool(layer.proj.weight.count_nonzero()) for layer in self.model.layers):
+                output.logits = output.logits * float("nan")
+            return output
+
+    model = NonfiniteModel()
+    before = snapshot(model)
+    with pytest.raises(FloatingPointError, match="non-finite"):
+        run_surgery(model, TinyTokenizer(), "layer_sweep", options(sample_layers=2))
+    assert_restored(model, before)