Deposit flight
This playbook follows a bank's deposit book from landing to a published lead-time result, and CI runs it on every change to the docs or the API spec. Its point is observation time: an account that goes quiet before it moves its balance out is caught by its silence, and the model's median risk for the quiet accounts is higher at the last of four re-scores than at the first, while no new event arrives.
note
The deposit book is generated by formula in step 1, so every number on
this page is a fixture-class result: it shows the machinery works, never how early it would
catch flight in a real bank.
Observation time is enabled per model, on evidence you see before you decide. The playbook builds the dataset
with and without it, trains both, pins the ablation that compares them at the same points,
and only then serves with it. The same suite runs with TESSERA_PLAYBOOK_OBSERVATION_TIME=off
in CI and must fail at step 9: without observation time nothing rises under silence.
1. Land the deposit book
1,200 generated accounts over 160 days, as one typed Parquet file: uploaded, mapped and synced. Steady accounts keep depositing and spending, with a small transfer now and then. Leaving accounts behave exactly like them, on the same cadence and with the same history length, then stop all activity and, about six weeks later, move most of their balance out. Nothing before the silence tells them apart.
01-land-the-deposit-book.py
"""Deposit-flight playbook, step 1: land the bank's deposit book.
The book is formula-generated with sha256 jitter (no RNG, no clock), so
every run lands the same book: 1,200 accounts over 160 days. Steady accounts
keep their deposits, card purchases and logins coming, with a small transfer
now and then. Leaving accounts behave exactly like them, on the same cadence
and with the same history length, then stop all activity and, about six
weeks later, move most of their balance out in one transfer. Nothing in a
leaving account's events before it goes quiet tells it apart: the tell is
the silence.
The book arrives the way a core-banking export does: one typed Parquet file,
uploaded, mapped and synced. Typed columns keep numbers numbers, which the
value-aware label of step 3 compares.
"""
import hashlib
import io
import json
import sys
from collections import Counter
from datetime import UTC, datetime, timedelta
from pathlib import Path
import httpx
import pyarrow as pa
import pyarrow.parquet as pq
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("deposit-flight")
START = datetime(2026, 3, 2, tzinfo=UTC)
DAYS = 160
# The datasets of step 5 read the book up to day 120: every leaving account has
# gone quiet by then and none has moved its balance yet.
WINDOW_DAYS = 120
# The book is fixed, but the as-of anchors the observation-time build draws
# (step 5) are seeded by that build's own fingerprint, which carries its freshly
# minted label and snapshot ids: every run trains the with-token model on a
# different draw (by design; the flag-off twin trains identically every run).
# On 400 accounts, with re-scores from day 100, some draws trained a model whose
# risk barely moved with silence in that range: a flat curve at step 9 in about
# 1 run in 10, and a G2 calibration error up to 0.040 against the 0.05 bar at
# step 7. 1,200 accounts and re-scores from day 93 (below) keep every draw far
# from both edges; the measurements are in the change that set these values.
ACCOUNTS = 1200
CYCLES = 8
# Step 9 re-scores weekly from day 93, the day after the last leaving account
# went quiet (its last cycle ends on day 81-92): the first re-score sees the
# leaving accounts in an ordinary pause, the last after three more weeks of
# silence, and none has moved its balance yet (the earliest flight is day 121).
RESCORE_FROM_DAY = 93
RESCORES = 4
CHANNELS = ("branch", "direct_deposit", "mobile", "wire")
def unit(*parts: object) -> float:
"""A deterministic draw in [0, 1) from the parts."""
digest = hashlib.sha256("|".join(map(str, parts)).encode("utf-8")).digest()
return int.from_bytes(digest[:8], "big") / 2**64
def account_events(index: int) -> tuple[str, bool, list[dict]]:
"""One account: eight deposit cycles inside the datasets' window, then its fate.
Every account has the same cycle (a deposit, then a card purchase or a
login) on gaps of 4-8 days, so every account carries exactly 16 events in
the window and the same history length. A steady account's last cycle in
the window ends 6-18 days before it closes (a pause) and it carries on
afterwards, now and then with a small transfer. A leaving account's last
cycle ends on day 81-92; 40-44 days later it moves 85-100% of its balance
out.
"""
account = f"acct-{index:04d}"
leaving = index % 2 == 1
gaps = [4.0 + 4.0 * unit(account, cycle, "gap") for cycle in range(CYCLES)]
if leaving:
last = 81.0 + 11.0 * unit(account, "stops")
else:
last = WINDOW_DAYS - 6.0 - 12.0 * unit(account, "last")
first = last - 0.6 * gaps[-1] - sum(gaps[:-1])
events: list[dict] = []
def add(day: float, event_type: str, **attributes: object) -> None:
events.append(
{
"event_id": f"evt-{account}-{len(events):04d}",
"account_id": account,
"event_type": event_type,
"occurred_at": (START + timedelta(days=day)).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
**attributes,
}
)
def cycle(day: float, gap: float, key: object) -> None:
add(
day,
"deposit.posted",
amount_minor=int(20_000 + 380_000 * unit(account, key, "amount")),
currency="USD",
channel=CHANNELS[int(4 * unit(account, key, "channel"))],
)
if unit(account, key, "kind") < 0.6:
add(
day + 0.6 * gap,
"card.purchase",
amount_minor=int(300 + 24_700 * unit(account, key, "card")),
currency="USD",
)
else:
add(day + 0.6 * gap, "app.login", is_new_device=unit(account, key, "device") < 0.02)
day = first
for number, gap in enumerate(gaps):
cycle(day, gap, number)
day += gap
if leaving:
flight = last + 40.0 + 4.0 * unit(account, "flight")
share = 0.85 + 0.15 * unit(account, "flight-share")
add(
flight,
"transfer.out",
amount_minor=int(3_000_000 * share),
currency="USD",
balance_share=round(share, 4),
is_external=True,
)
return account, leaving, events
day, number = WINDOW_DAYS + 1.0 + 3.0 * unit(account, "after"), CYCLES
while day < DAYS - 1.0:
gap = 4.0 + 4.0 * unit(account, number, "gap")
cycle(day, gap, number)
if unit(account, number, "move") < 0.1:
share = 0.02 + 0.18 * unit(account, number, "share") # a small move is not flight
add(
day + 0.8 * gap,
"transfer.out",
amount_minor=int(2_000_000 * share),
currency="USD",
balance_share=round(share, 4),
is_external=False,
)
day += gap
number += 1
return account, leaving, events
COLUMNS: dict[str, pa.DataType] = {
"amount_minor": pa.int64(),
"currency": pa.string(),
"channel": pa.string(),
"is_new_device": pa.bool_(),
"balance_share": pa.float64(),
"is_external": pa.bool_(),
}
leavers: list[str] = []
book: list[dict] = []
for index in range(ACCOUNTS):
account, leaving, events = account_events(index)
book.extend(events)
if leaving:
leavers.append(account)
rows = [
{
name: event.get(name)
for name in ("event_id", "account_id", "event_type", "occurred_at", *COLUMNS)
}
for event in book
]
schema = pa.schema(
[
("event_id", pa.string()),
("account_id", pa.string()),
("event_type", pa.string()),
("occurred_at", pa.string()),
*COLUMNS.items(),
]
)
buffer = io.BytesIO()
pq.write_table(pa.Table.from_pylist(rows, schema=schema), buffer)
session = api.post(
"/uploads",
{"filename": "deposit_book.parquet", "format": "parquet"},
key="deposit-flight-upload",
)
httpx.put(session["upload_url"], content=buffer.getvalue(), timeout=300.0).raise_for_status()
api.post(f"/uploads/{session['upload_id']}:complete")
upload = api.get(f"/uploads/{session['upload_id']}")
api.wait(
f"/syncs/{upload['sync_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the landing sync",
polls=300,
)
mapping = {
"version": 1,
"entity": {"type": {"constant": "account"}, "id_column": "account_id"},
"event_type": {"from_column": "event_type"},
"timestamp": {"column": "occurred_at"},
"event_id_column": "event_id",
"attributes": [{"column": name} for name in COLUMNS],
}
api.put(f"/sources/{upload['source_id']}/mapping", mapping)
sync = api.post(
f"/sources/{upload['source_id']}/syncs", {"mode": "full"}, key="deposit-flight-sync"
)
sync = api.wait(
f"/syncs/{sync['sync_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the book sync",
polls=300,
)
print(f"book sync: {sync['status']}, rows_rejected: {sync.get('rows_rejected')}")
print(f"events landed: {sync.get('events_normalized') == len(rows)}")
if sync["status"] != "succeeded" or sync.get("rows_rejected"):
sys.exit(json.dumps(sync)[:3000])
counts = Counter(event["event_type"] for event in book)
for event_type in sorted(counts):
print(f" {event_type}: {counts[event_type]}")
print(f"accounts: {ACCOUNTS}, quiet leavers: {len(leavers) > 0}")
end = START + timedelta(days=DAYS)
state.update(
start=START.strftime("%Y-%m-%dT%H:%M:%SZ"),
window_end=(START + timedelta(days=WINDOW_DAYS)).strftime("%Y-%m-%dT%H:%M:%SZ"),
# Step 9 re-scores weekly from day 93: every leaving account is quiet, none has left.
rescore_from=(START + timedelta(days=RESCORE_FROM_DAY)).strftime("%Y-%m-%dT%H:%M:%SZ"),
rescores=RESCORES,
end=end.strftime("%Y-%m-%dT%H:%M:%SZ"),
leavers=leavers,
accounts=ACCOUNTS,
)
Download playbooks/deposit-flight/01-land-the-deposit-book.py — this exact file runs in CI on every change (hosts shown are for this environment).
2. Approve the contract
02-approve-the-contract.py
"""Deposit-flight playbook, step 2: approve the data contract."""
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("deposit-flight")
draft = api.post("/contracts", key="deposit-flight-contract")
print(f"draft contract profiles: {', '.join(sorted(draft.get('event_types') or {}))}")
approved = api.post(f"/contracts/{draft['version']}:approve")
print(f"contract: {approved['status']}")
Download playbooks/deposit-flight/02-approve-the-contract.py — this exact file runs in CI on every change (hosts shown are for this environment).
3. A value-aware label
deposit_flight_45d is true when, within 45 days of an account's event, a single
transfer.out moves at least 80% of the balance out. A small transfer is not flight.
03-define-the-value-aware-label.py
"""Deposit-flight playbook, step 3: a value-aware outcome label.
`deposit_flight_45d` is true when, within 45 days of an account's event, a
single `transfer.out` moves at least 80% of the balance out
(`balance_share >= 0.8`). The value matters: a transfer of a small share of
the balance is not flight. It is a derived pattern label, so every account
event is a supervision point and no outcome table is needed.
"""
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("deposit-flight")
label = api.post(
"/pattern-labels",
{
"name": "deposit_flight_45d",
"entity_type": "account",
"source": {
"kind": "derived",
"rule": {
"template": "custom",
"outcome": "presence",
"horizon_days": 45,
"predicate": {
"kind": "pattern",
"pattern": {
"window_agg": {
"agg": "count",
"match": {
"event_type_pattern": "transfer.out",
"attribute_filters": [
{"attr_key": "balance_share", "op": "gte", "value": 0.8}
],
},
"over_days": 45,
"op": "gte",
"value": 1,
}
},
},
},
},
},
key="deposit-flight-label",
)
print(f"label: {label['name']}, task: {label['task']}")
state.update(label_id=label["label_id"])
Download playbooks/deposit-flight/03-define-the-value-aware-label.py — this exact file runs in CI on every change (hosts shown are for this environment).
4. Declare the protocol
Before any build: the label, the metrics, the tuned internal quality safeguard, the lead-time metrics and
margins, and the review bands with their alert budget. Class fixture.
04-declare-the-protocol.py
"""Deposit-flight playbook, step 4: the protocol, before any build.
The lead-time comparison is fixed first: the label, the metrics, the tuned
internal quality safeguard, the lead-time metrics and margins, and the bank's
review bands with their alert budget. Class `fixture`.
"""
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("deposit-flight")
protocol = api.post(
"/protocols",
{
"name": "deposit flight lead time against the quality safeguard (fixture)",
"provenance": "fixture",
"labels": [state["label_id"]],
"metrics": ["auc", "capture@10pct"],
"arms": ["gbm_tuned"],
"margins": {"auc": 0.0, "capture@10pct": 0.0},
"lead_time_metrics": ["median_lead_days", "share_alerted@14d"],
"lead_time_margins": {"median_lead_days": 0.0, "share_alerted@14d": 0.0},
"banding_policy": {
"bands": [
{"band": "retention_call", "min_score": 0.5},
{"band": "relationship_review", "min_score": 0.2},
{"band": "no_action", "min_score": 0.0},
],
"list_size": 40,
"alert_budget": 20,
},
},
key="deposit-flight-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/deposit-flight/04-declare-the-protocol.py — this exact file runs in CI on every change (hosts shown are for this environment).
5. Build the dataset and its flag-off twin
Both builds read the book up to day 120, when every leaving account has gone quiet and none has left yet. With observation time on, the build adds as-of anchors where an account has been silent, each carrying the elapsed silence. The twin is the same request without it, from the same snapshot.
05-build-the-twin-datasets.py
"""Deposit-flight playbook, step 5: the observation-time build and its flag-off twin.
With observation time on, the build adds as-of anchors: points between and
after an account's events where the account has been silent, each carrying
the elapsed silence. The twin is the identical request without it; the
ablation of step 7 compares the two.
"""
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("deposit-flight")
# The M16 switch this playbook proves: "on" (the default) builds, trains and
# serves with observation time; "off" runs the same chain without it, and the
# quiet-book step then fails, because nothing rises under silence.
FLAG_ON = os.environ.get("TESSERA_PLAYBOOK_OBSERVATION_TIME", "on") != "off"
request = {
"entity_type": "account",
"labels": [state["label_id"]],
"window": {
"max_events_per_entity": 64,
"time_range": {"start": state["start"], "end": state["window_end"]},
},
"split": {"method": "entity_hash", "train": 0.55, "val": 0.15, "test": 0.3, "seed": 7},
}
# Twins read the identical event snapshot: the flag-off build pins one, and
# the observation-time build reuses it.
sides = ["off", "on"] if FLAG_ON else ["off"]
datasets = {}
snapshot_id = None
for side in sides:
body = dict(request)
if side == "on":
body["observation_time"] = {"enabled": True, "between_offsets": 2, "tail_offsets": 4}
body["snapshot_id"] = snapshot_id
build = api.post("/datasets", body, key=f"deposit-flight-dataset-{side}")
built = api.wait(
f"/datasets/{build['dataset_id']}",
lambda ds: ds.get("status") in ("succeeded", "failed", "aborted_leakage"),
what=f"the {side} build",
)
print(f"dataset {side}: {built['status']}")
if built["status"] != "succeeded":
sys.exit(json.dumps(built)[:3000])
datasets[side] = built["dataset_id"]
snapshot_id = built["snapshot_id"]
stats = api.get(f"/labels/{state['label_id']}/stats")
print(f"label positive rate known: {stats.get('positive_rate') is not None}")
print("builds: succeeded")
state.update(datasets=datasets)
Download playbooks/deposit-flight/05-build-the-twin-datasets.py — this exact file runs in CI on every change (hosts shown are for this environment).
6. Train both
The CPU CI profile (forty epochs) and the last-event readout for both; only the with-token run consumes the silence token.
06-train-the-twin-models.py
"""Deposit-flight playbook, step 6: train the with-token model and its flag-off twin.
Both runs use the quickstart's CPU CI profile (forty epochs) and the
last-event readout; the with-token run also consumes the elapsed-silence
token. They are otherwise identical, so the ablation compares exactly one
thing.
"""
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("deposit-flight")
# The M16 switch this playbook proves: "on" (the default) builds, trains and
# serves with observation time; "off" runs the same chain without it, and the
# quiet-book step then fails, because nothing rises under silence.
FLAG_ON = os.environ.get("TESSERA_PLAYBOOK_OBSERVATION_TIME", "on") != "off"
profile_path = Path(__file__).resolve().parents[2] / "quickstart" / "ci-train-profile.json"
profile = json.loads(profile_path.read_text())
profile.pop("_comment", None)
models = {}
for side, dataset_id in sorted(state["datasets"].items()):
body = json.loads(json.dumps(profile))
body["dataset_id"] = dataset_id
# Forty epochs instead of the quickstart's ten: on this small book the twins
# need them to settle, so the ablation compares the token, not training noise.
body["trainer"]["epochs"] = 40
body["overrides"] = {**body["overrides"], "readout": "last_event"}
if side == "on":
body["overrides"]["observation_time"] = True
run = api.post("/runs", body, key=f"deposit-flight-train-{side}")
run = api.wait(
f"/runs/{run['run_id']}",
lambda item: item.get("status") in ("succeeded", "failed", "cancelled"),
what=f"the {side} training run",
)
print(f"training {side}: {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=f"the {side} evaluation",
)
report = api.get(f"/models/{run['model_id']}/eval-report")
failing = sorted(g for g, block in report["gates"].items() if block["status"] != "pass")
print(f"model {side}: {model['status']}, failing gates: {failing or 'none'}")
models[side] = run["model_id"]
print("trainings: succeeded")
state.update(models=models)
Download playbooks/deposit-flight/06-train-the-twin-models.py — this exact file runs in CI on every change (hosts shown are for this environment).
7. Pin the ablation evidence
POST /v1/models/{model_id}/observation-ablations grades the with-token model, its twin
and the platform's internal quality safeguards at the identical points and records the
evidence in the model's evaluation of record. It recommends enablement only when the model passes G1-G6 and does not
regress against its twin. Evidence that does not recommend enabling is information: the bundle
reports it as a finding that you acknowledge or not.
07-pin-the-ablation-evidence.py
"""Deposit-flight playbook, step 7: the observation-time ablation, pinned.
Observation time is enabled per model on evidence you see first: the model with the
silence token, its flag-off twin and the platform's internal quality safeguards,
all graded at the identical points. The evaluation records that evidence in the model's report
of record; it recommends enablement only when the model passes G1-G6 and does
not regress against its twin. Evidence that does not recommend enabling is
information, not a block: the bundle in step 8 reports it as the finding
`observation_time_not_recommended`, which the operator acknowledges or not.
"""
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("deposit-flight")
# The M16 switch this playbook proves: "on" (the default) builds, trains and
# serves with observation time; "off" runs the same chain without it, and the
# quiet-book step then fails, because nothing rises under silence.
FLAG_ON = os.environ.get("TESSERA_PLAYBOOK_OBSERVATION_TIME", "on") != "off"
if not FLAG_ON:
print("observation time is off: no ablation to run")
print("ablation step: done")
sys.exit(0)
models = state["models"]
created = api.post(
f"/models/{models['on']}/observation-ablations",
{"without_token_model_id": models["off"]},
key="deposit-flight-ablation",
)
ablation = api.wait(
f"/models/{models['on']}/observation-ablations/{created['report_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what="the observation-time ablation",
)
print(f"ablation: {ablation['status']}")
if ablation["status"] != "succeeded":
sys.exit(json.dumps(ablation)[:3000])
print(f"gates passed: {ablation.get('gates_passed')}")
print(f"recommend enable: {ablation.get('recommend_enable')}")
report = api.get(f"/models/{models['on']}/eval-report")
evidence = report.get("observation_time_ablation") or {}
for row in evidence.get("labels") or []:
print(
f"{row['metric']} with the token {row['with_token']:.4f}, "
f"without {row['without_token']:.4f} (delta {row['delta']:+.4f})"
)
for name, arm in sorted((evidence.get("arms") or {}).items()):
print(f"arm {name}: {arm.get('status')}")
if not ablation.get("recommend_enable"):
# Since WP-M26 G1 also compares the model with the safeguard arms, and this
# small CPU fixture can trail the tuned safeguard; the bundle reports it.
gates = {name: gate.get("status") for name, gate in (report.get("gates") or {}).items()}
print(f"gates: {json.dumps(gates, sort_keys=True)}")
print("the evidence does not recommend enabling: the bundle reports it as a finding")
print("ablation step: done")
Download playbooks/deposit-flight/07-pin-the-ablation-evidence.py — this exact file runs in CI on every change (hosts shown are for this environment).
8. Serve with observation time
The bundle enables observation time (the registry re-checks the pinned evidence) and a batch
slot records as_of and the quiet state on every write, with 7- and 14-day thresholds. The
bundle readiness read names exactly what needs an acknowledgement now (a failing advisory gate,
or the finding observation_time_not_recommended); the step acknowledges exactly that, under
your consent, and stops if anything needed was not consented to.
08-serve-with-observation-time.py
"""Deposit-flight playbook, step 8: serve with observation time on.
The bundle enables observation time (the registry re-checks the pinned
evidence), a batch slot serves it, and the slot records `as_of` and the
quiet state on every write, with 7- and 14-day silence thresholds.
The bundle readiness read (a dry run of the bundle request) names exactly
what needs an acknowledgement now: the failing advisory gates and the bundle
findings, each finding with the fingerprint of its facts. The step acknowledges
exactly that, under the operator's consent, and stops if anything needed was
not consented to. Without the readiness read, the bundle refusal lists the same.
"""
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("deposit-flight")
# The M16 switch this playbook proves: "on" (the default) builds, trains and
# serves with observation time; "off" runs the same chain without it, and the
# quiet-book step then fails, because nothing rises under silence.
FLAG_ON = os.environ.get("TESSERA_PLAYBOOK_OBSERVATION_TIME", "on") != "off"
model_id = state["models"]["on" if FLAG_ON else "off"]
body: dict = {"observation_time": {"enabled": True}} if FLAG_ON else {}
bundles = f"/models/{model_id}/bundles"
def needed() -> tuple[list[str], list[dict[str, str]], dict | None]:
"""Exactly the gates and findings a bundle request needs acknowledged now
(and the bundle, when the fallback request needed nothing and created it).
The readiness read (a dry run of the bundle request) names them; on a
deployment without it, the bundle refusal lists the same facts. Either way
nothing is assumed: a fixed list would be refused when a finding is absent.
"""
status, readiness = api.attempt(
"GET",
f"/models/{model_id}/bundle-readiness",
params={"observation_time": "true" if FLAG_ON else "false"},
)
if status == 200:
if readiness.get("verdict") == "blocked":
sys.exit(f"the bundle request is refused: {json.dumps(readiness)[:3000]}")
return (
list(readiness.get("acknowledge_gates") or []),
[
{"finding": entry["finding"], "fingerprint": entry["fingerprint"]}
for entry in readiness.get("acknowledge_findings") or []
],
None,
)
if status != 404:
sys.exit(f"GET bundle-readiness: HTTP {status}: {json.dumps(readiness)[:3000]}")
status, answer = api.attempt("POST", bundles, body, key="deposit-flight-bundle-review")
if status < 400:
return [], [], answer
details = (answer.get("error") or {}).get("details") or []
if not any(detail.get("reason") == "acknowledgement_required" for detail in details):
sys.exit(f"POST {bundles}: HTTP {status}: {json.dumps(answer)[:3000]}")
return (
[
detail["gate"]
for detail in details
if detail.get("type") == "promotion_gate" and detail.get("severity") == "advisory"
],
[
{"finding": detail["finding"], "fingerprint": detail["fingerprint"]}
for detail in details
if detail.get("type") == "bundle_finding"
],
None,
)
gates, findings, created = needed()
# Explicit consent only: the operator names the advisory gates and the bundle
# findings they accept, and the reason. What is needed must all be consented
# to; then exactly that is acknowledged, whether or not G1 happened to fail.
gate_consent = json.loads(os.environ.get("TESSERA_QUICKSTART_ACKNOWLEDGEMENT") or "{}")
finding_consent = json.loads(os.environ.get("TESSERA_PLAYBOOK_FINDINGS_ACKNOWLEDGEMENT") or "{}")
unconsented = [gate for gate in gates if gate not in gate_consent.get("gates", [])] + [
entry["finding"]
for entry in findings
if entry["finding"] not in finding_consent.get("findings", [])
]
if unconsented:
sys.exit(f"needs an acknowledgement the operator did not give: {', '.join(unconsented)}")
if gates:
body["acknowledge_gates"] = {"gates": gates, "reason": gate_consent["reason"]}
if findings:
body["acknowledge_findings"] = {"findings": findings, "reason": finding_consent["reason"]}
bundle = created if created is not None else api.post(bundles, body, key="deposit-flight-bundle")
slot = api.post(
"/slots",
{
"environment": "test",
"name": "deposits",
"entity_type": "account",
"mode": "live",
"scoring": "batch_only",
},
key="deposit-flight-slot",
)
activated = api.post(f"/slots/{slot['slot_id']}:activate", {"bundle_id": bundle["bundle_id"]})
print(f"slot deposits active: {activated['active_bundle_id'] == bundle['bundle_id']}")
if FLAG_ON:
config = api.put(
"/slots/deposits/observation-time",
{"enabled": True, "silence_thresholds": ["7d", "14d"], "maturation_lag": "0d"},
)
print(f"observation time on the slot: {config['enabled']}")
if not config["enabled"]:
sys.exit("the slot refused observation time")
state.update(serving_model=model_id, bundle_id=bundle["bundle_id"])
Download playbooks/deposit-flight/08-serve-with-observation-time.py — this exact file runs in CI on every change (hosts shown are for this environment).
9. Quiet accounts rise
Four re-scores as of days 93, 100, 107 and 114, a week apart: the leaving accounts are quiet and have not left, so only the observation instant moves. Their recorded silence grows at every re-score, and their median risk is higher at the last re-score than at the first.
09-quiet-accounts-rise.py
"""Deposit-flight playbook, step 9: quiet accounts score higher at the last re-score.
Four batch re-scores as of days 93, 100, 107 and 114 of the book, a week apart:
the leaving accounts have gone quiet and have not left yet, so they keep
their event histories and only the observation instant moves. With
observation time on, each re-score records `as_of` and the quiet state, and
the quiet accounts' median risk is higher at the last re-score than at the
first: the silence alone moved it. With it off, every
quiet account scores the same at every re-score, and this step fails: the
playbook proves the capability, not the prose.
"""
import sys
from datetime import datetime, timedelta
from itertools import pairwise
from pathlib import Path
from statistics import median
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # playbooks/playbook_api.py
from playbook_api import Api, State
api, state = Api(), State("deposit-flight")
first = datetime.fromisoformat(state["rescore_from"].replace("Z", "+00:00"))
count = int(state["rescores"])
cutoffs = [first + timedelta(days=7 * index) for index in range(count)]
label = "deposit_flight_45d"
for index, cutoff in enumerate(cutoffs):
refresh = api.post(
"/slots/deposits/refreshes",
{"cutoff": cutoff.strftime("%Y-%m-%dT%H:%M:%SZ")},
key=f"deposit-flight-refresh-{index}",
)
refresh = api.wait(
f"/slots/deposits/refreshes/{refresh['refresh_id']}",
lambda item: item.get("status") in ("succeeded", "failed"),
what=f"re-score {index + 1}",
)
print(f"re-score {index + 1}: {refresh['status']}")
leavers = state["leavers"]
scores, silences, observed = [], [], 0
for account in leavers:
served = api.get(f"/slots/deposits/entities/account/{account}/state")
value = (served.get("scores") or {}).get(label)
if isinstance(value, dict):
value = value.get("p", value.get("score"))
if isinstance(value, int | float):
scores.append(float(value))
observation = api.get(f"/slots/deposits/entities/account/{account}/observation")
if observation.get("as_of") and observation.get("quiet_state"):
observed += 1
silences.append(int(observation["quiet_state"]["silence_seconds"]))
state.update(
**{
f"quiet_{index}": {
"median_score": median(scores) if scores else None,
"median_silence": median(silences) if silences else None,
"observed": observed,
}
}
)
curve = [state[f"quiet_{index}"] for index in range(count)]
print("quiet accounts observed at their as-of: " + str(all(c["observed"] > 0 for c in curve)))
if not all(c["observed"] > 0 for c in curve):
sys.exit("no quiet account carries as_of and a quiet state: observation time is off")
silence = [c["median_silence"] for c in curve]
risk = [c["median_score"] for c in curve]
grows = all(earlier < later for earlier, later in pairwise(silence))
print(f"silence grows between re-scores: {grows}")
print(f"quiet-account median risk higher at the last re-score than the first: {risk[0] < risk[-1]}")
print("risk curve: " + ", ".join(f"{value:.4f}" for value in risk))
if not (grows and risk[0] < risk[-1]):
sys.exit(f"the quiet accounts' median risk was not higher at the last re-score: {risk}")
Download playbooks/deposit-flight/09-quiet-accounts-rise.py — this exact file runs in CI on every change (hosts shown are for this environment).
10. The cutoff-ladder backtest
How early each arm would have alerted, at the protocol's alert budget.
10-cutoff-ladder-backtest.py
"""Deposit-flight playbook, step 10: the cutoff-ladder backtest under the protocol.
The backtest replays the test accounts at a ladder of cutoffs and asks how
early each arm would have alerted, at the protocol's alert budget: the
with-token model counts every alert (observation-time regime), the tuned
internal quality safeguard only alerts that its event prefixes support.
"""
import json
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("deposit-flight")
model_id = state["serving_model"]
created = api.post(
"/backtests",
{"protocol_id": state["protocol_id"], "model_id": model_id, "seed": 0},
key="deposit-flight-backtest",
)
detail = api.wait(
f"/backtests/{created['backtest_id']}",
lambda item: (item.get("backtest") or {}).get("status") in ("succeeded", "failed"),
what="the backtest",
)
print(f"backtest: {detail['backtest']['status']}")
if detail["backtest"]["status"] != "succeeded":
sys.exit(json.dumps(detail)[:3000])
report = detail["report"]
print(f"cutoffs: {len(report['cutoffs'])}, alert budget: {report['alert_budget']}")
block = report["per_label"][state["label_id"]]
if block.get("status") == "completed":
# The model and your own comparators; the internal reference arms are never printed.
for arm, row in block["arms"].items():
if arm == "model" or arm.startswith("cmp_"):
print(f" {arm}: regime {row.get('regime')}, alerted {row.get('n_alerted')}")
print(f"lead-time verdict: {(report.get('verdict') or {}).get('overall')}")
state.update(backtest_id=created["backtest_id"])
Download playbooks/deposit-flight/10-cutoff-ladder-backtest.py — this exact file runs in CI on every change (hosts shown are for this environment).
11. The model-risk dossier
The sr11-7.v2 dossier assembles the model's documentation from its pinned evidence; a
section without evidence says so.
11-export-the-sr11-7-dossier.py
"""Deposit-flight playbook, step 11: the model-risk dossier (`sr11-7.v2`).
The dossier assembles the model's documentation from its pinned evidence;
a section whose evidence is missing says so, it is never filled in.
"""
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("deposit-flight")
dossier = api.post(
"/dossiers",
{"template_id": "sr11-7.v2", "subject": {"model_id": state["serving_model"]}},
key="deposit-flight-dossier",
)
present = sorted(s["section_id"] for s in dossier["sections"] if s["evidence"] == "present")
missing = sorted(s["section_id"] for s in dossier["sections"] if s["evidence"] != "present")
template = dossier["template"]
template_id = template["template_id"] if isinstance(template, dict) else template
print(f"dossier template: {template_id}")
print(f"sections with evidence: {', '.join(present)}")
print(f"sections without evidence: {', '.join(missing) or 'none'}")
download = api.get(f"/dossiers/{dossier['dossier_id']}:download", format="json")
print(f"download: {download['media_type']}, bytes: {download['byte_length'] > 0}")
Download playbooks/deposit-flight/11-export-the-sr11-7-dossier.py — this exact file runs in CI on every change (hosts shown are for this environment).
12. Publish the result
The verdict and the backtest's lead-time numbers in the published-results format: class
fixture, a fingerprint on every number.
12-publish-the-lead-time-result.py
"""Deposit-flight playbook, step 12: publish the comparison and the lead time.
The evaluation judges the served model against the protocol; the document
carries that verdict and the backtest's lead-time numbers, each with the
fingerprint of the report that holds it. Class `fixture`.
"""
import json
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("deposit-flight")
evaluation = api.post(
"/evaluations",
{"model_id": state["serving_model"], "protocol_id": state["protocol_id"], "seed": 0},
key="deposit-flight-evaluation",
)
judged = api.wait(
f"/evaluations/{evaluation['report_id']}",
lambda item: (item.get("evaluation") or {}).get("status") in ("succeeded", "failed"),
what="the protocol evaluation",
)
status = (judged.get("evaluation") or {}).get("status")
print(f"evaluation: {status}")
if status != "succeeded":
sys.exit(json.dumps(judged)[:3000])
document = state.path.parent / "deposit-flight-published-results.json"
published = subprocess.run(
[
"tessera-doctest",
"publish",
"--protocol-id",
state["protocol_id"],
"--evaluation-id",
evaluation["report_id"],
"--backtest-id",
state["backtest_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)
Download playbooks/deposit-flight/12-publish-the-lead-time-result.py — this exact file runs in CI on every change (hosts shown are for this environment).
13. Recompute it yourself
13-recompute-this-yourself.sh
#!/usr/bin/env bash
# Deposit-flight playbook, step 13: recompute the published result 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/deposit-flight-published-results.json" \
--repo-root "$TESSERA_REPO_ROOT"
Download playbooks/deposit-flight/13-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 early flight is caught in a real deposit book: the book is formula-generated, and its quiet leavers were made to go silent first.
- A separation the model found on its own: leaving accounts differ from steady ones only by their silence, by construction, so the separation the silence token learns was planted in the fixture.
- Count invariance: every account has exactly 16 events in the window, so G3 (count invariance) passes through its single-count rule and this playbook does not test it.
- That the published verdict passes: on this small fixture the protocol's verdict is what the data gives, and the page publishes it whatever it is.