Skip to main content

PostHog loop

This playbook closes a product loop end to end on the local stack, and CI runs it on every change to the docs or the API spec: product events, a protocol declared first, a baseline written down before any model exists, the model judged against it, a band written to PostHog as a person property, and a holdout group that never receives it.

note

The events come from the quickstart's sample dataset, so every number on this page is a fixture-class result: it shows the machinery works, never how well it predicts on a real product.

On the local stack PostHog is a stand-in reached through the stack's control origin, and the destination writer runs one pass per call. Against a real deployment the same calls run unchanged against your PostHog project.

1. Land the product's events​

The quickstart's sample dataset (generated e-commerce orders), checksum-verified, uploaded, mapped with the quickstart's mapping and synced.

01-send-product-events.py
"""PostHog-loop playbook, step 1: land the product's events.

The events are the quickstart's sample dataset (generated e-commerce orders
and their repeat-purchase outcomes; checksums verified from its manifest). On
your own product, the PostHog or HTTP connector sends these; here the orders
file is uploaded, mapped with the quickstart's mapping and synced.
"""

import hashlib
import io
import json
import os
import sys
import zipfile
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("posthog-loop")

SAMPLE_URL = os.environ.get(
    "TESSERA_SAMPLE_URL", "https://docs.staging.tessera.zone/downloads/tessera-sample-v1.zip"
)
archive = zipfile.ZipFile(io.BytesIO(httpx.get(SAMPLE_URL, timeout=120.0).content))
manifest = json.loads(archive.read("manifest.json"))
for name, meta in manifest["files"].items():
    assert hashlib.sha256(archive.read(name)).hexdigest() == meta["sha256"], name
print("sample dataset verified")
sample_dir = state.path.parent / "posthog-loop-sample"
archive.extractall(sample_dir)

session = api.post(
    "/uploads", {"filename": "ecommerce_orders.csv", "format": "csv"}, key="posthog-loop-orders"
)
orders = (sample_dir / "ecommerce_orders.csv").read_bytes()
httpx.put(session["upload_url"], content=orders, timeout=120.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=150,
)
mapping_path = Path(__file__).resolve().parents[2] / "quickstart" / "mapping-orders.json"
api.put(f"/sources/{upload['source_id']}/mapping", json.loads(mapping_path.read_text()))
sync = api.post(
    f"/sources/{upload['source_id']}/syncs", {"mode": "full"}, key="posthog-loop-orders-sync"
)
sync = api.wait(
    f"/syncs/{sync['sync_id']}",
    lambda item: item.get("status") in ("succeeded", "failed"),
    what="the orders sync",
    polls=150,
)
print(f"orders sync: {sync['status']}, rows_rejected: {sync.get('rows_rejected')}")
state.update(
    sample_dir=str(sample_dir),
    start="2025-01-01T00:00:00Z",
    end="2025-07-01T00:00:00Z",
)

Download playbooks/posthog-loop/01-send-product-events.py — this exact file runs in CI on every change (hosts shown are for this environment).

2. Approve the contract​

The draft profiles what arrived; approving it signs off the event types and attributes.

02-approve-the-contract.py
"""PostHog-loop playbook, step 2: approve the data contract the events profiled."""

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("posthog-loop")

draft = api.post("/contracts", key="posthog-loop-contract")
seen = sorted(draft.get("event_types") or {})
print(f"draft contract profiles: {', '.join(seen)}")
approved = api.post(f"/contracts/{draft['version']}:approve")
print(f"contract: {approved['status']}")

Download playbooks/posthog-loop/02-approve-the-contract.py — this exact file runs in CI on every change (hosts shown are for this environment).

3. The label the loop acts on​

repeat_purchase_30d, the sample's label table of each user's outcome and when it became known, uploaded to a file_upload source.

03-define-the-label.py
"""PostHog-loop playbook, step 3: the label the loop acts on.

`repeat_purchase_30d` reads a label table: one row per user with the outcome
and the instant it became known (the sample dataset's
`labels_repeat_purchase.csv`), uploaded to a `file_upload` source. On your
product it is whatever records the outcome: billing, a warehouse table, a
webhook.
"""

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("posthog-loop")

source = api.post(
    "/sources",
    {"name": "repeat-purchase outcomes", "kind": "file_upload", "config": {}},
    key="posthog-loop-label-source",
)
label = api.post(
    "/labels",
    {
        "name": "repeat_purchase_30d",
        "entity_type": "user",
        "task": "binary",
        "source": {
            "kind": "label_table",
            "source_id": source["source_id"],
            "entity_id_column": "entity_id",
            "value_column": "value",
            "observed_at_column": "observed_at",
        },
        "maturation_days": 7,
    },
    key="posthog-loop-label",
)
print(f"label: {label['name']}")

table = (Path(state["sample_dir"]) / "labels_repeat_purchase.csv").read_bytes()
session = api.post(
    "/uploads",
    {"filename": "labels_repeat_purchase.csv", "format": "csv", "source_id": source["source_id"]},
    key="posthog-loop-label-upload",
)
httpx.put(session["upload_url"], content=table, timeout=120.0).raise_for_status()
api.post(f"/uploads/{session['upload_id']}:complete")
upload = api.get(f"/uploads/{session['upload_id']}")
sync = api.wait(
    f"/syncs/{upload['sync_id']}",
    lambda item: item.get("status") in ("succeeded", "failed"),
    what="the label table sync",
    polls=150,
)
print(f"label table sync: {sync['status']}")
stats = api.get(f"/labels/{label['label_id']}/stats")
print(f"label observations stored: {int(stats.get('observations') or 0) > 0}")
state.update(label_id=label["label_id"], label_source_id=source["source_id"])

Download playbooks/posthog-loop/03-define-the-label.py — this exact file runs in CI on every change (hosts shown are for this environment).

4. Declare the protocol, holdout salt included​

Before any data is built: the label, the metrics, the two internal quality safeguards, the margins, the bands PostHog will carry, and the salt that fixes which users are held out. Class fixture.

04-declare-the-protocol.py
"""PostHog-loop playbook, step 4: the protocol, before any data is built.

It names the label, the metrics, the two internal quality safeguards the model
must beat, the margins, the bands the PostHog property carries, and the holdout
salt: the held-out users are fixed by this salt now, before anything is scored.
Its class is `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("posthog-loop")

# Pre-registered with the protocol: this salt fixes who is held out.
HOLDOUT_SALT = "posthog-loop-holdout-v1"
protocol = api.post(
    "/protocols",
    {
        "name": "repeat purchase against the quality safeguards, with a holdout (fixture)",
        "provenance": "fixture",
        "labels": [state["label_id"]],
        "metrics": ["auc", "lift@10pct"],
        "arms": ["gbm_default", "gbm_tuned"],
        "margins": {"auc": 0.01, "lift@10pct": 0.0},
        "holdout_salt": HOLDOUT_SALT,
        "banding_policy": {
            "bands": [
                {"band": "high", "min_score": 0.6},
                {"band": "medium", "min_score": 0.3},
                {"band": "low", "min_score": 0.0},
            ],
            "list_size": 100,
            "alert_budget": 50,
        },
    },
    key="posthog-loop-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"],
    holdout_salt=HOLDOUT_SALT,
)

Download playbooks/posthog-loop/04-declare-the-protocol.py — this exact file runs in CI on every change (hosts shown are for this environment).

5. Baseline first​

The dataset is built after the declaration; before any model exists, the baseline evaluation reports what the protocol's internal quality safeguards reach on it. That bar is written down first.

05-baseline-first.py
"""PostHog-loop playbook, step 5: baseline first.

The dataset is built after the protocol, from each user's last 8 events
(the quickstart's CPU-sized window). Before any model exists, the baseline
evaluation fits the protocol's internal quality safeguards on it and reports
what they reach: the bar the model will have to clear, written down first.
"""

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("posthog-loop")

build = api.post(
    "/datasets",
    {
        "entity_type": "user",
        "labels": [state["label_id"]],
        "window": {
            "max_events_per_entity": 8,
            "time_range": {"start": state["start"], "end": state["end"]},
        },
        "split": {"method": "entity_hash", "train": 0.7, "val": 0.1, "test": 0.2, "seed": 7},
    },
    key="posthog-loop-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])
baseline = api.post(
    "/evaluations:baseline",
    {"dataset_id": dataset["dataset_id"], "protocol_id": state["protocol_id"], "seed": 0},
    key="posthog-loop-baseline",
)
detail = api.wait(
    f"/evaluations/{baseline['report_id']}",
    lambda item: (item.get("evaluation") or {}).get("status") in ("succeeded", "failed"),
    what="the baseline evaluation",
)
print(f"baseline evaluation: {detail['evaluation']['status']}")
verdict = (detail.get("report") or {}).get("verdict") or {}
print(f"baseline verdict: {verdict.get('overall')}")
# Reference levels of your own comparators are printed; the internal reference levels are only
# counted (the platform's internal models are never named).
internal = 0
for criterion in verdict.get("criteria") or []:
    if str(criterion.get("arm", "")).startswith("cmp_"):
        print(f"  bar {criterion['metric']} of {criterion['arm']}: {criterion.get('arm_value')}")
    else:
        internal += 1
print(f"internal reference levels recorded: {internal}")
state.update(dataset_id=dataset["dataset_id"], baseline_id=baseline["report_id"])

Download playbooks/posthog-loop/05-baseline-first.py — this exact file runs in CI on every change (hosts shown are for this environment).

6. Train, then judge​

The model trains on the CPU CI profile and is judged against the same protocol and the same bar. A fail or incomplete verdict is the result.

06-train-and-judge.py
"""PostHog-loop playbook, step 6: train, then judge the model against the protocol.

The run uses the quickstart's documented CPU CI profile. The evaluation judges
the model against the same protocol and the same bar as step 5; a `fail` or
`incomplete` verdict is the result, and it is published as it is.
"""

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("posthog-loop")

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"] = state["dataset_id"]
run = api.post("/runs", profile, key="posthog-loop-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']}")
evaluation = api.post(
    "/evaluations",
    {"model_id": run["model_id"], "protocol_id": state["protocol_id"], "seed": 0},
    key="posthog-loop-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')}")
state.update(model_id=run["model_id"], evaluation_id=evaluation["report_id"])

Download playbooks/posthog-loop/06-train-and-judge.py — this exact file runs in CI on every change (hosts shown are for this environment).

7. Activate and bind the protocol's bands​

The slot scores every user; the bands PostHog receives are the protocol's, fingerprint and all.

07-activate-and-bind-bands.py
"""PostHog-loop playbook, step 7: activate, score, bind the protocol's bands.

The bands written to PostHog are the protocol's bands, fingerprint and all.
"""

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("posthog-loop")

model_id = state["model_id"]
model = api.get(f"/models/{model_id}")
body = {}
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="posthog-loop-bundle")
slot = api.post(
    "/slots",
    {
        "environment": "test",
        "name": "growth",
        "entity_type": "user",
        "mode": "live",
        "scoring": "batch_only",
    },
    key="posthog-loop-slot",
)
activated = api.post(f"/slots/{slot['slot_id']}:activate", {"bundle_id": bundle["bundle_id"]})
print(f"slot growth active: {activated['active_bundle_id'] == bundle['bundle_id']}")
refresh = api.post("/slots/growth/refreshes", {"cutoff": state["end"]}, key="posthog-loop-refresh")
refresh = api.wait(
    f"/slots/growth/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')}")
binding = {
    "label": "repeat_purchase_30d",
    "policy": state["banding_policy"],
    "policy_fingerprint": state["banding_policy_fingerprint"],
}
bands = api.put("/slots/growth/bands", {"bindings": [binding]})
names = [band["band"] for band in bands["bindings"][0]["policy"]["bands"]]
print(f"bands bound: {', '.join(names)}")
state.update(slot_id=slot["slot_id"], scored=refresh.get("entities_scored"))

Download playbooks/posthog-loop/07-activate-and-bind-bands.py — this exact file runs in CI on every change (hosts shown are for this environment).

8. Hold users out​

The holdout experiment takes its salt from the protocol, so who is held out was fixed before anything was scored.

08-hold-out-users.py
"""PostHog-loop playbook, step 8: the holdout experiment.

The experiment takes its salt from the protocol (declared in step 4), so who
is held out was fixed before anything was scored. Held-out users are never
delivered under the experiment's purpose; they are the comparison group.
"""

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("posthog-loop")

experiment = api.post(
    "/slots/growth/experiments",
    {"holdout_fraction": 0.2, "purpose": "general", "protocol_id": state["protocol_id"]},
    key="posthog-loop-experiment",
)
print(f"experiment: {experiment['status']}, holdout_fraction: {experiment['holdout_fraction']}")
state.update(
    experiment_id=experiment["experiment_id"],
    holdout_fraction=experiment["holdout_fraction"],
)

Download playbooks/posthog-loop/08-hold-out-users.py — this exact file runs in CI on every change (hosts shown are for this environment).

9. Write the band to PostHog, without the held-out users​

A PostHog destination writes tessera_band and the score as person properties. The delivery receipts record how many users were withheld and under which experiment, and the check at the end recomputes the holdout from the protocol's salt and confirms no held-out user reached PostHog.

09-deliver-to-posthog.py
"""PostHog-loop playbook, step 9: write the band to PostHog, without the held-out users.

A PostHog destination writes `tessera_band` (and the score, slot and policy
fingerprint) as person properties. A backfill delivers every scored user
once; afterwards each band change is delivered as it happens. The destination
writer withholds every held-out user before it builds a request.

On the local stack PostHog is a twin read through the control origin, and the
writer runs one pass per control call, with its clock set eight days back: the
follow-up outcomes step 10 reads mature one label horizon (7 days) after the
delivery, and an outcome can only be stored once it has happened. The check at
the end recomputes the holdout from the protocol's salt with the documented
assignment rule:
sha256(salt || "{project_id}#{type}#{id}"), first 8 bytes as an unsigned
integer, divided by 2**64, below the holdout fraction.
"""

import hashlib
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("posthog-loop")

CONTROL = os.environ["TESSERA_CONTROL_BASE"]
POSTHOG_KEY = "phc_fixture_project_key"
httpx.post(f"{CONTROL}/fakes/posthog/keys", json={"api_key": POSTHOG_KEY}, timeout=30.0)
destination = api.post(
    "/destinations",
    {
        "vendor": "posthog",
        "slot_id": state["slot_id"],
        "config": {"host": "https://posthog.fake.tessera.test", "project_id": 4242},
        "credentials": {"api_key": POSTHOG_KEY},
    },
    key="posthog-loop-destination",
)
properties = ", ".join(sorted(destination["declared_property_names"]))
print(f"destination: {destination['status']}, properties: {properties}")
api.post(f"/destinations/{destination['destination_id']}:backfill", key="posthog-loop-backfill")
drained = httpx.post(
    f"{CONTROL}/ticks/destination-writer", json={"advance_seconds": -8 * 86_400}, timeout=600.0
).json()
print(f"writer delivered batches: {drained.get('batches', 0) > 0}")
receipts = api.get(f"/destinations/{destination['destination_id']}/deliveries").get("data") or []
withheld = sum(int(receipt.get("withheld_count") or 0) for receipt in receipts)
applied = {
    experiment
    for receipt in receipts
    for experiment in (receipt.get("suppression_manifest") or {}).get("experiments_applied") or []
}
print(f"deliveries succeeded: {all(receipt['status'] == 'succeeded' for receipt in receipts)}")
print(f"held-out users withheld: {withheld > 0}")
print(f"experiment applied: {state['experiment_id'] in applied}")

persons = httpx.get(f"{CONTROL}/fakes/posthog", timeout=30.0).json().get("persons") or {}
project_id = os.environ["TESSERA_PROJECT_ID"]


# The salt declared with the protocol (step 4) and the experiment's fraction
# (step 8), as stored: the check never restates them.
salt = str(state["holdout_salt"]).encode("utf-8")
numerator, denominator = float(state["holdout_fraction"]).as_integer_ratio()


def held_out(user: str) -> bool:
    key = f"{project_id}#user#{user}".encode()
    bucket = int.from_bytes(hashlib.sha256(salt + key).digest()[:8], "big")
    # bucket / 2**64 < fraction, in exact integer arithmetic
    return bucket * denominator < numerator * (1 << 64)


banded = sorted(user for user, properties in persons.items() if "tessera_band" in properties)
leaked = [user for user in banded if held_out(user)]
print(f"users with a tessera_band in PostHog: {len(banded) > 0}")
print(f"held-out users in PostHog: {len(leaked)}")
if leaked:
    sys.exit(f"held-out users reached PostHog: {leaked[:10]}")
state.update(destination_id=destination["destination_id"])

Download playbooks/posthog-loop/09-deliver-to-posthog.py — this exact file runs in CI on every change (hosts shown are for this environment).

10. Read the holdout​

Every delivery cycle recorded the users it delivered to and the users it withheld. The readout compares the two groups' outcome rates over a window, with a bootstrap 95% interval. An outcome counts when it matures one label horizon (7 days) after the delivery that anchored the user, so on the local stack step 9 ran the writer with its clock eight days back. This fixture has no users who could react, so the step uploads a generated follow-up outcome per user, observed now and drawn from one seeded coin that ignores the group. The readout shows the computation and finds no effect, as it should.

10-read-the-holdout.py
"""PostHog-loop playbook, step 10: the incrementality readout of the holdout.

Every delivery cycle under the experiment recorded which users it delivered
to (the treated group) and which it withheld (the holdout). The readout
compares the label's outcome rates of the two groups over a window: the
difference and a bootstrap 95% interval.

An outcome counts when it matures one label horizon (7 days) after the
delivery that anchored the user; on the local stack the delivery ran eight days
ago (step 9). This fixture has no real users who could react, so the step
uploads a generated follow-up outcome for every user, observed now and drawn
from one seeded coin that ignores the group: the readout shows the computation
and should find no effect, and the page says so.
"""

import csv
import hashlib
import io
import sys
from datetime import UTC, datetime, timedelta
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("posthog-loop")

deliveries = api.get(f"/destinations/{state['destination_id']}/deliveries").get("data") or []
first_delivery = min(
    datetime.fromisoformat(str(item["created_at"]).replace("Z", "+00:00")) for item in deliveries
)
window_start = first_delivery - timedelta(hours=1)
users = sorted(
    row["entity_id"]
    for row in csv.DictReader(
        (Path(state["sample_dir"]) / "labels_repeat_purchase.csv").read_text().splitlines()
    )
)
outcome_at = (datetime.now(UTC) - timedelta(minutes=1)).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
table = io.StringIO()
writer = csv.writer(table)
writer.writerow(["entity_id", "value", "observed_at"])
for user in sorted(set(users)):
    # One seeded coin per user, blind to the experiment: no planted effect.
    draw = int(hashlib.sha256(f"follow-up:{user}".encode()).hexdigest()[:8], 16) / 0xFFFFFFFF
    writer.writerow([user, "true" if draw < 0.3 else "false", outcome_at])
session = api.post(
    "/uploads",
    {
        "filename": "followup_outcomes.csv",
        "format": "csv",
        "source_id": state["label_source_id"],
    },
    key="posthog-loop-followup-upload",
)
httpx.put(
    session["upload_url"], content=table.getvalue().encode(), timeout=120.0
).raise_for_status()
api.post(f"/uploads/{session['upload_id']}:complete")
upload = api.get(f"/uploads/{session['upload_id']}")
sync = api.wait(
    f"/syncs/{upload['sync_id']}",
    lambda item: item.get("status") in ("succeeded", "failed"),
    what="the follow-up outcome sync",
    polls=150,
)
print(f"follow-up outcomes: {sync['status']}")
window_end = datetime.now(UTC)  # after the outcomes were observed
readout = api.get(
    f"/slots/growth/experiments/{state['experiment_id']}/readout",
    label_id=state["label_id"],
    start=window_start.strftime("%Y-%m-%dT%H:%M:%SZ"),
    end=window_end.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
)
treated, holdout = readout["treated"], readout["holdout"]
print(f"treated users with an outcome: {treated['n']}, rate {treated['rate']:.3f}")
print(f"held-out users with an outcome: {holdout['n']}, rate {holdout['rate']:.3f}")
print(f"difference: {readout['delta']:+.3f}, 95% interval {readout['ci95']}")
low, high = readout["ci95"]
print(f"readout computed: {treated['n'] >= 30 and holdout['n'] >= 30}")
print(f"no planted effect found (the interval covers zero): {low <= 0.0 <= high}")

Download playbooks/posthog-loop/10-read-the-holdout.py — this exact file runs in CI on every change (hosts shown are for this environment).

11. Publish the comparison​

The baseline bar and the model's verdict, in the published-results format: class fixture, a fingerprint on every number.

11-publish-the-results.py
"""PostHog-loop playbook, step 11: publish the comparison, baseline bar included.

The document carries the baseline evaluation (the bar, written down first)
and the model's verdict against the same protocol, each number with the
fingerprint of the report that holds it. Its class is the protocol's:
`fixture`.

The incrementality readout of step 10 is not in this document: it is a live
measurement over a window, not an archived report with a fingerprint, and on
this fixture its outcomes are generated.
"""

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("posthog-loop")

document = state.path.parent / "posthog-loop-published-results.json"
published = subprocess.run(
    [
        "tessera-doctest",
        "publish",
        "--protocol-id",
        state["protocol_id"],
        "--evaluation-id",
        state["evaluation_id"],
        "--baseline-id",
        state["baseline_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/posthog-loop/11-publish-the-results.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
# PostHog-loop playbook, step 12: recompute the published comparison 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/posthog-loop-published-results.json" \
  --repo-root "$TESSERA_REPO_ROOT"

Download playbooks/posthog-loop/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​

  • An incrementality result: the follow-up outcomes are generated and ignore the group, so the readout shows the measurement, not a lift.
  • How well the model predicts repeat purchases on a real product: the events come from the quickstart's sample dataset.