Security lake
This playbook runs one security chain end to end on the local stack, and CI runs it on every
change to the docs or the API spec: an OCSF lake, the ocsf domain pack, scheduled
syncs, contract approval, training, scores that carry their evidence, an alert rule with a
disposition, a write-back to Snowflake, and a detection-quality report that you can recompute.
note
The lake comes from a seeded generator, so every number on this page is a fixture-class
result: it shows the machinery works, never how well it detects on real data. The report
says so in its headline, and the docs lint keeps it from being published anywhere else.
The steps run against the local stack, where the customer bucket, Snowflake and the schedules' ticks are stand-ins reached through the stack's control origin. Against a real deployment the same calls run unchanged; the bucket, the warehouse and the schedules are real.
1. Land the lake
A generated OCSF lake: the seeded ocsf_lake generator, written as the three OCSF classes the ocsf pack reads, in a customer bucket. On your own lake, skip this step and point step 3 at the bucket your OCSF export already writes.
01-land-the-lake.py
"""Security-lake playbook, step 1: land an OCSF lake in the customer's bucket.
The lake comes from the seeded `ocsf_lake` generator of tessera-testing
(routine users, shift workers and a few account takeovers whose tell is the
order of their steps), written as the three OCSF classes the `ocsf` pack reads.
On your own lake skip this step: point step 3 at the bucket your OCSF export
already writes.
On the local stack the bucket is a fake customer bucket, written through the
stack's control origin. Against a real deployment, upload the same objects to
your bucket with your own tooling.
"""
import base64
import json
import os
from collections import Counter
from datetime import UTC, datetime
from pathlib import Path
import httpx
from tessera_testing.domainpacks.ocsf_lake import OCSF_LAKE_FILES, ocsf_lake_tables
from tessera_testing.generators import generate
from tessera_testing.scenarios import PresenceRates
STATE = Path(os.environ.get("TESSERA_STATE_DIR", "."))
CONTROL = os.environ["TESSERA_CONTROL_BASE"] # the local stack's control origin
BUCKET = "acme-security-lake"
START = datetime(2026, 7, 1, tzinfo=UTC)
DAYS = 14
scenario = generate(
"ocsf_lake",
entities=400,
seed=11,
start=START,
days=DAYS,
presence_rates=PresenceRates(missing=0.0, sentinel=0.0, corrupted=0.0),
)
tables = ocsf_lake_tables(scenario.events)
objects = [
{
"provider": "gcs",
"bucket": BUCKET,
"name": f"ocsf/{catalog.removeprefix('ocsf.')}/{OCSF_LAKE_FILES[catalog]}",
"content_b64": base64.b64encode(data).decode("ascii"),
}
for catalog, data in sorted(tables.items())
]
response = httpx.post(f"{CONTROL}/customer-objects", json={"objects": objects}, timeout=120.0)
response.raise_for_status()
counts = Counter(event.event_type for event in scenario.events)
archetypes = Counter(entity["archetype"] for entity in scenario.ground_truth["entities"].values())
print(f"objects written: {response.json()['written']}")
for event_type in sorted(counts):
print(f" {event_type}: {counts[event_type]} events")
print("users by archetype:", json.dumps(dict(sorted(archetypes.items()))))
# The ground truth stays with the fixture: step 9 uses it to disposition alerts
# the way an analyst who knew the answer would.
takeovers = sorted(
user
for user, entity in scenario.ground_truth["entities"].items()
if entity["archetype"] == "takeover"
)
(STATE / "security-lake.json").write_text(
json.dumps(
{
"bucket": BUCKET,
"window": {"start": START.isoformat(), "days": DAYS},
"takeovers": takeovers,
"users": len(scenario.ground_truth["entities"]),
}
)
)
Download playbooks/security-lake/01-land-the-lake.py — this exact file runs in CI on every change (hosts shown are for this environment).
2. Install the ocsf pack
The install pins the pack by fingerprint, creates its draft contract and derived labels (repeated_logon_failure_1d among them) and makes its OCSF catalogs available as source mappings.
02-install-the-ocsf-pack.py
"""Security-lake playbook, step 2: install the `ocsf` domain pack, release 1.
The install pins the pack by fingerprint (a release whose bytes differ from the
published fingerprint is refused with `pack_fingerprint_mismatch`), creates the
pack's draft contract and its derived labels, and makes its three OCSF catalogs
available as source mappings.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
install = api.post("/domain-packs/ocsf:install", {"version": 1}, key="security-lake-install")
print(f"pack: {install['pack']} v{install['version']} {install['status']}")
print(f"pack_fingerprint: {install['pack_fingerprint']}")
print(f"entity_types: {', '.join(install['entity_types'])}")
labels = {entry["name"]: entry["label_id"] for entry in install["labels"]}
for name in sorted(labels):
print(f"label: {name}")
state.update(install_id=install["install_id"], label_ids=labels)
Download playbooks/security-lake/02-install-the-ocsf-pack.py — this exact file runs in CI on every change (hosts shown are for this environment).
3. Connect the lake and schedule its syncs
One gcs source per OCSF class, each mapped through the pack's catalog and each on a daily schedule.
03-connect-and-schedule.py
"""Security-lake playbook, step 3: one `gcs` source per OCSF class, each mapped
through the pack's catalog, each on a daily schedule.
The catalog decides how every OCSF row becomes an event (its `type_uid` picks
the event type; the actor's uid is the user). The schedules keep the lake
flowing in production; step 4 runs the same syncs now, through the documented
manual trigger, so this page does not wait for 03:00 UTC.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
bucket = state["bucket"]
sources = {}
for catalog in ("ocsf.authentication", "ocsf.account_change", "ocsf.api_activity"):
folder = catalog.removeprefix("ocsf.")
source = api.post(
"/sources",
{
"name": f"ocsf {folder}",
"kind": "gcs",
"config": {"bucket": bucket, "prefix": f"ocsf/{folder}/", "format": "parquet"},
},
key=f"security-lake-source-{folder}",
)
source_id = source["source_id"]
api.put(f"/sources/{source_id}/mapping", {"pack_catalog": {"pack": "ocsf", "catalog": catalog}})
provenance = api.get(f"/sources/{source_id}/mapping/provenance")
schedule = api.put(
f"/sources/{source_id}/schedule",
{"schedule": {"kind": "daily", "at_utc": "03:00"}, "enabled": True},
)
sources[catalog] = source_id
pinned = provenance.get("pack_fingerprint") is not None
print(f"{catalog}: mapped through the pack catalog, pack pinned: {pinned}")
print(f"{catalog}: schedule daily at 03:00 UTC, enabled: {schedule['enabled']}")
state.update(sources=sources)
Download playbooks/security-lake/03-connect-and-schedule.py — this exact file runs in CI on every change (hosts shown are for this environment).
4. Sync now
The documented manual trigger runs the sync the schedule's tick would run. Every mapped row lands as an event; nothing is rejected.
04-sync-now.py
"""Security-lake playbook, step 4: run each source's sync now.
`POST /sources/{source_id}/syncs` is the manual trigger; the schedule's tick
starts the same sync. Every row the catalog maps lands as an event.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
for catalog, source_id in sorted(state["sources"].items()):
folder = catalog.removeprefix("ocsf.")
sync = api.post(
f"/sources/{source_id}/syncs", {"mode": "full"}, key=f"security-lake-sync-{folder}"
)
done = api.wait(
f"/syncs/{sync['sync_id']}",
lambda job: job.get("status") in ("succeeded", "failed"),
what=f"the {catalog} sync",
)
landed = done.get("events_normalized")
print(f"{catalog}: {done['status']}, rows_rejected: {done.get('rows_rejected')}")
print(f"{catalog}: events landed: {landed}")
if done["status"] != "succeeded" or done.get("rows_rejected") or not landed:
sys.exit(f"{catalog} did not land cleanly: {done}")
Download playbooks/security-lake/04-sync-now.py — this exact file runs in CI on every change (hosts shown are for this environment).
5. Approve the contract
A fresh draft profiles what landed; approving it is the sign-off on the event types and attributes.
05-approve-the-contract.py
"""Security-lake playbook, step 5: approve the data contract.
A fresh draft profiles what the syncs landed; approving it is the sign-off
that the event types and attributes are what the lake should carry.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
draft = api.post("/contracts", key="security-lake-contract")
version = draft["version"]
seen = sorted(draft.get("event_types") or {})
print(f"draft contract profiles: {', '.join(seen)}")
approved = api.post(f"/contracts/{version}:approve")
print(f"contract: {approved['status']}")
state.update(contract_version=version)
Download playbooks/security-lake/05-approve-the-contract.py — this exact file runs in CI on every change (hosts shown are for this environment).
6. Declare the protocol
Before any dataset exists: the label, the metrics, the tuned internal quality safeguard the model must beat, the margins and the review bands with their alert budget. Class fixture.
06-declare-the-protocol.py
"""Security-lake playbook, step 6: declare the detection-quality protocol.
The protocol fixes, before any dataset is built, what the report of step 11
judges: the label, the metrics, the internal quality safeguard the model must
beat, the margins, and the review bands with their alert budget. Its class is
`fixture`: this lake comes from a seeded generator, and the report can never claim more.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
label_id = state["label_ids"]["repeated_logon_failure_1d"]
protocol = api.post(
"/protocols",
{
"name": "security lake detection quality (fixture)",
"provenance": "fixture",
"labels": [label_id],
"metrics": ["alert_precision@budget", "auc", "precision@list"],
"arms": ["gbm_tuned"],
"margins": {"alert_precision@budget": 0.0, "auc": 0.0, "precision@list": 0.0},
"banding_policy": {
"bands": [
{"band": "review", "min_score": 0.5},
{"band": "watch", "min_score": 0.2},
{"band": "baseline", "min_score": 0.0},
],
"list_size": 40,
"alert_budget": 20,
},
},
key="security-lake-protocol",
)
print(f"provenance: {protocol['provenance']}")
print(f"declared_at: {protocol['declared_at']}")
state.update(
protocol_id=protocol["protocol_id"],
banding_policy=protocol["banding_policy"],
banding_policy_fingerprint=protocol["banding_policy_fingerprint"],
)
Download playbooks/security-lake/06-declare-the-protocol.py — this exact file runs in CI on every change (hosts shown are for this environment).
7. Build and train
The dataset is built after the declaration, then the model trains on the quickstart's CPU CI profile.
07-build-and-train.py
"""Security-lake playbook, step 7: build the dataset, then train.
The build comes after the protocol (step 6), so the report can judge it. It
keeps each user's last 64 events: a derived label is supervised only at
anchors inside the kept window, so the window must reach back to the takeovers.
The run uses the quickstart's documented CPU CI profile; production runs drop its
`overrides` and `trainer`.
"""
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
label_id = state["label_ids"]["repeated_logon_failure_1d"]
start = datetime.fromisoformat(state["window"]["start"])
end = start + timedelta(days=state["window"]["days"] + 1)
start_text, end_text = (moment.strftime("%Y-%m-%dT%H:%M:%SZ") for moment in (start, end))
build = api.post(
"/datasets",
{
"entity_type": "user",
"labels": [label_id],
"window": {
"max_events_per_entity": 64,
"time_range": {"start": start_text, "end": end_text},
},
"split": {"method": "entity_hash", "train": 0.7, "val": 0.1, "test": 0.2, "seed": 7},
},
key="security-lake-dataset",
)
dataset = api.wait(
f"/datasets/{build['dataset_id']}",
lambda ds: ds.get("status") in ("succeeded", "failed", "aborted_leakage"),
what="the dataset build",
)
print(f"dataset: {dataset['status']}")
if dataset["status"] != "succeeded":
sys.exit(json.dumps(dataset)[:3000])
profile_path = Path(__file__).resolve().parents[2] / "quickstart" / "ci-train-profile.json"
profile = json.loads(profile_path.read_text())
profile.pop("_comment", None)
profile["dataset_id"] = dataset["dataset_id"]
run = api.post("/runs", profile, key="security-lake-train")
run = api.wait(
f"/runs/{run['run_id']}",
lambda item: item.get("status") in ("succeeded", "failed", "cancelled"),
what="the training run",
)
print(f"training: {run['status']}")
if run["status"] != "succeeded":
sys.exit(json.dumps(run)[:3000])
model = api.wait(
f"/models/{run['model_id']}",
lambda item: item.get("status") in ("evaluated", "promoted", "archived", "failed"),
what="the model's evaluation",
)
print(f"model: {model['status']}")
state.update(dataset_id=dataset["dataset_id"], model_id=run["model_id"], cutoff=end_text)
Download playbooks/security-lake/07-build-and-train.py — this exact file runs in CI on every change (hosts shown are for this environment).
8. Evidence on, activate, bind the bands
A signals evaluation with evidence on, a bundle that serves it, a batch slot, and the protocol's bands bound to the label.
08-enable-evidence-and-activate.py
"""Security-lake playbook, step 8: evidence on, bundle, activate, bind the bands.
A signals evaluation with evidence on makes every score carry the events that
moved it. The bundle copies that configuration, the batch slot serves it, and
the protocol's review bands are bound to the label, so alert rules and exports
read the same bands the report judges.
"""
import json
import os
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
model_id = state["model_id"]
created = api.post(
f"/models/{model_id}/signals-evaluations",
{"evidence": {"enabled": True, "top_k": 8}, "seed": 0},
key="security-lake-signals",
)
signals = api.wait(
f"/models/{model_id}/signals-evaluations/{created['report_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the evidence evaluation",
)
print(f"evidence evaluation: {signals['status']}")
if signals["status"] != "succeeded":
sys.exit(json.dumps(signals)[:3000])
model = api.get(f"/models/{model_id}")
body = {"signals": {"enabled": True}}
promotion = model.get("promotion") or {}
if promotion.get("verdict") == "needs_acknowledgement":
# Explicit consent only: the operator names the advisory gates they accept
# and the reason; the bundle acknowledges those that are failing now. A
# failing gate the operator did not name is left for the API to refuse.
raw = os.environ.get("TESSERA_QUICKSTART_ACKNOWLEDGEMENT", "")
if raw:
consent = json.loads(raw)
gates = [gate for gate in consent["gates"] if gate in promotion.get("advisory", [])]
if gates:
body["acknowledge_gates"] = {"gates": gates, "reason": consent["reason"]}
bundle = api.post(f"/models/{model_id}/bundles", body, key="security-lake-bundle")
runtime_signals = (bundle.get("runtime") or {}).get("signals") or {}
print(f"bundle: evidence in the runtime: {bool(runtime_signals.get('evidence'))}")
slot = api.post(
"/slots",
{
"environment": "test",
"name": "security",
"entity_type": "user",
"mode": "live",
"scoring": "batch_only",
},
key="security-lake-slot",
)
activated = api.post(f"/slots/{slot['slot_id']}:activate", {"bundle_id": bundle["bundle_id"]})
print(f"slot security active: {activated['active_bundle_id'] == bundle['bundle_id']}")
binding = {
"label": "repeated_logon_failure_1d",
"policy": state["banding_policy"],
"policy_fingerprint": state["banding_policy_fingerprint"],
}
bands = api.put("/slots/security/bands", {"bindings": [binding]})
names = [band["band"] for band in bands["bindings"][0]["policy"]["bands"]]
print(f"bands bound: {', '.join(names)}")
state.update(bundle_id=bundle["bundle_id"])
Download playbooks/security-lake/08-enable-evidence-and-activate.py — this exact file runs in CI on every change (hosts shown are for this environment).
9. Alert, evidence, disposition
A rule opens an episode when a user's behavior is more surprising than 95% of their peers. The batch refresh scores every user; the alerted user's served state carries the evidence; the disposition feeds the rule's precision rollup.
09-alert-and-disposition.py
"""Security-lake playbook, step 9: an alert rule, a scored book, a disposition.
The rule opens an episode when a user's behavior is more surprising than 95%
of their peers (`behavior_surprise`, a calibrated percentile). A batch refresh
scores every user as of the end of the lake window; the alerted user's served
state carries the evidence that moved the score. The analyst's disposition
(here taken from the fixture's ground truth) feeds the rule's precision rollup.
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
rule = api.post(
"/slots/security/alert-rules",
{
"name": "behavior more surprising than 95% of peers",
"predicate": {"select": {"kind": "behavior_surprise"}, "op": "gte", "value": 0.95},
},
key="security-lake-rule",
)
print(f"alert rule: {rule['status']}")
refresh = api.post(
"/slots/security/refreshes", {"cutoff": state["cutoff"]}, key="security-lake-refresh"
)
refresh = api.wait(
f"/slots/security/refreshes/{refresh['refresh_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the batch refresh",
)
print(f"refresh: {refresh['status']}, users scored: {refresh.get('entities_scored')}")
episodes = api.get("/slots/security/alert-episodes", rule_id=rule["rule_id"]).get("data") or []
print(f"episodes opened: {len(episodes) > 0}")
if not episodes:
sample = [
api.get(f"/slots/security/entities/user/{user}/state") for user in state["takeovers"][:10]
]
seen = [
((row.get("signals") or {}).get("behavior_surprise"), row.get("scores")) for row in sample
]
sys.exit(f"no user crossed the surprise threshold; takeover users' state: {seen}")
takeovers = set(state["takeovers"])
episode = sorted(episodes, key=lambda item: item["entity_id"] not in takeovers)[0]
user = episode["entity_id"]
served = api.get(f"/slots/security/entities/user/{user}/state")
evidence = ((served.get("signals") or {}).get("evidence") or {}).get("transitions") or []
print(f"evidence on the alerted user: {len(evidence) > 0}")
verdict = "true_positive" if user in takeovers else "false_positive"
judged = api.post(
f"/slots/security/alert-episodes/{episode['episode_id']}:disposition",
{"disposition": verdict, "reason": "fixture ground truth"},
)
print(f"disposition recorded: {judged['disposition']['disposition'] == verdict}")
rollup = (
api.get("/slots/security/monitors/alert-precision", rule_id=rule["rule_id"]).get("data") or [{}]
)[0]
print(f"precision rollup dispositioned: {rollup.get('dispositioned')}")
state.update(rule_id=rule["rule_id"])
Download playbooks/security-lake/09-alert-and-disposition.py — this exact file runs in CI on every change (hosts shown are for this environment).
10. Write back to Snowflake
A key-pair Snowflake credential, a daily export schedule, and the same export run once now, in append mode, with scores and evidence.
10-snowflake-write-back.py
"""Security-lake playbook, step 10: write scores, bands and evidence back to Snowflake.
A Snowflake credential is a source with key-pair setup (Tessera mints the key;
you run the printed `ALTER USER` once). The export schedule is the production
shape, daily after the scoring refresh; this page then runs the same export
once, now. On the local stack Snowflake is a twin, read through the control
origin.
"""
import os
import sys
from pathlib import Path
import httpx
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
credential = api.post(
"/sources",
{
"name": "snowflake write-back",
"kind": "snowflake",
"config": {
"account": "acme-security",
"warehouse": "TESSERA_WH",
"database": "LAKE",
"schema": "TESSERA",
"table": "USER_SCORES",
},
"credentials": {"username": "TESSERA_WRITER"},
},
key="security-lake-snowflake",
)
print(f"snowflake key-pair setup issued: {bool((credential.get('setup') or {}).get('public_key'))}")
destination = {
"kind": "snowflake",
"credential_id": credential["source_id"],
"database": "LAKE",
"schema": "TESSERA",
"table": "USER_SCORES",
}
schedule = api.post(
"/slots/security/export-schedules",
{
"cadence": {"kind": "daily", "at_utc": "04:00"},
"export": {"destination": destination, "mode": "append", "refresh_first": True},
},
key="security-lake-export-schedule",
)
print(f"export schedule: daily at 04:00 UTC, status: {schedule.get('status')}")
export = api.post(
"/slots/security/warehouse-exports",
{"destination": destination, "mode": "append", "include": ["scores", "evidence"]},
key="security-lake-export",
)
export = api.wait(
f"/slots/security/warehouse-exports/{export['export_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the Snowflake export",
)
print(f"export: {export['status']}, mode: {export.get('mode')}")
tables = httpx.get(f"{os.environ['TESSERA_CONTROL_BASE']}/fakes/snowflake", timeout=30.0).json()
rows = {table["name"]: table["rows"] for table in tables.get("tables") or []}
landed = rows.get("LAKE.TESSERA.USER_SCORES")
exported = int(export.get("rows_exported") or 0)
print(f"rows exported: {exported > 0}")
print(f"snowflake LAKE.TESSERA.USER_SCORES rows match the export: {landed == exported}")
if exported <= 0 or landed != exported:
sys.exit(f"the write-back is empty or incomplete: exported {exported}, landed {landed}")
Download playbooks/security-lake/10-snowflake-write-back.py — this exact file runs in CI on every change (hosts shown are for this environment).
11. The detection-quality report
The evaluation judges the model against the protocol only, and tessera-doctest publish writes it in the published-results format: class fixture in the headline, a fingerprint on every number, an appendix with the calls.
11-detection-quality-report.py
"""Security-lake playbook, step 11: the detection-quality report, as a published result.
The evaluation judges the model against the protocol of step 6 only. The
report is written in the published-results format: its class is the
protocol's (`fixture`), every number cites the fingerprint of the report that
holds it, and its appendix lists the calls that reproduce it (step 12).
"""
import os
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("security-lake")
evaluation = api.post(
"/evaluations",
{"model_id": state["model_id"], "protocol_id": state["protocol_id"], "seed": 0},
key="security-lake-evaluation",
)
detail = api.wait(
f"/evaluations/{evaluation['report_id']}",
lambda item: (item.get("evaluation") or {}).get("status") in ("succeeded", "failed"),
what="the protocol evaluation",
)
print(f"evaluation: {detail['evaluation']['status']}")
verdict = (detail.get("report") or {}).get("verdict") or {}
print(f"verdict provenance: {verdict.get('provenance')}")
document = state.path.parent / "security-lake-published-results.json"
published = subprocess.run(
[
"tessera-doctest",
"publish",
"--protocol-id",
state["protocol_id"],
"--evaluation-id",
evaluation["report_id"],
"--out",
str(document),
"--repo-root",
os.environ["TESSERA_REPO_ROOT"],
],
capture_output=True,
text=True,
)
print(published.stdout)
if published.returncode != 0:
sys.exit(published.stderr)
state.update(published_results=str(document))
Download playbooks/security-lake/11-detection-quality-report.py — this exact file runs in CI on every change (hosts shown are for this environment).
12. Recompute it yourself
The appendix replayed against this project: every fingerprint re-hashes, every number comes back exactly.
12-recompute-this-yourself.sh
#!/usr/bin/env bash
# Security-lake playbook, step 12: recompute the report yourself.
#
# Replays the appendix of the document step 11 wrote against this project:
# every fingerprint must re-hash byte for byte and every number must come back
# exactly (Playbooks, "Recompute this yourself").
set -euo pipefail
: "${TESSERA_API_BASE:?e.g. https://api.staging.tessera.zone/v1}"
: "${TESSERA_API_KEY:?the tsk_ key of this project}"
: "${TESSERA_REPO_ROOT:?a checkout of the Tessera repository (for the schemas)}"
: "${TESSERA_STATE_DIR:?set by the runner; ids are handed between steps}"
tessera-doctest recompute \
--document "$TESSERA_STATE_DIR/security-lake-published-results.json" \
--repo-root "$TESSERA_REPO_ROOT"
Download playbooks/security-lake/12-recompute-this-yourself.sh — this exact file runs in CI on every change (hosts shown are for this environment).
What this does not show
- How well the model detects takeovers on a real lake: the lake comes from a seeded generator.
- The schedules firing on their own: the page runs each scheduled sync and export once, now, through the same trigger the tick uses.
- Detection-quality thresholds judged by injection: the protocol declares none, because no public operation judges them yet.