Don't Let the AI Studio Developer Log Be Your Source of Truth: A Two-Layer Way to Observe the Interactions API

The two-layer design: separate the lens from the ledger

So I split observability cleanly into two layers.

The first layer is the AI Studio developer log. I use it for triage right after an incident, to grasp "what happened at which step" in the shortest path. I add nothing to it and use it as provided.

The second layer is a self-hosted, append-only log. This is the source of truth. The key is interaction_id. With that single identifier, I can always tie a run I saw on the dashboard back to the record I kept in my own log.

The split works because it lets you hold separate expectations of each. Ask the dashboard only to be fast, not to be complete or durable. Ask the source of truth only to return the same answer forever, not to be visually quick. Not mixing those expectations became the design guideline itself.

A minimal source-of-truth implementation

The source-of-truth side is easiest to hold as a thin wrapper around the model call. Before and after the call, append one structured record per call: the interaction_id, a fingerprint of the input, tokens used, elapsed time, and a summary of the outcome.

Here is a Python example. The model-call line shows the Interactions API pattern; confirm the exact endpoint name and parameters against the current changelog. The recording and reconciliation logic is self-contained application-layer code.

import hashlib
import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Optional
 
 
@dataclass
class InteractionRecord:
    interaction_id: str
    prev_interaction_id: Optional[str]
    prompt_fingerprint: str   # keep only a fingerprint in the ledger, not the input
    feature: str              # which surface made the call (cost attribution axis)
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: int
    status: str               # "ok" | "empty" | "error"
    created_at: float
 
 
def fingerprint(text: str) -> str:
    # minimization so no PII lands in the ledger; enough for reconciliation
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]
 
 
class ObservedClient:
    """Thinly wrap the Interactions API call and append one ledger record per call."""
 
    def __init__(self, client, sink):
        self.client = client      # the Gemini Interactions client
        self.sink = sink          # a sink exposing append_only(record: dict)
 
    def run(self, prompt: str, *, model: str, feature: str,
            prev_interaction_id: Optional[str] = None) -> dict:
        started = time.perf_counter()
        # prepare an id client-side so even an empty response leaves a trace
        local_id = prev_interaction_id or str(uuid.uuid4())
        status = "error"
        usage = {"input_tokens": 0, "output_tokens": 0}
        server_id = local_id
        try:
            resp = self.client.interactions.create(
                model=model,
                input=prompt,
                previous_interaction_id=prev_interaction_id,
            )
            server_id = getattr(resp, "id", local_id)
            usage = {
                "input_tokens": resp.usage.input_tokens,
                "output_tokens": resp.usage.output_tokens,
            }
            status = "ok" if resp.output_text else "empty"
            return resp
        finally:
            # try/finally guarantees at least one record even on exception
            rec = InteractionRecord(
                interaction_id=server_id,
                prev_interaction_id=prev_interaction_id,
                prompt_fingerprint=fingerprint(prompt),
                feature=feature,
                model=model,
                input_tokens=usage["input_tokens"],
                output_tokens=usage["output_tokens"],
                latency_ms=int((time.perf_counter() - started) * 1000),
                status=status,
                created_at=time.time(),
            )
            self.sink.append_only(json.dumps(asdict(rec)))

The point is try/finally. Whether the response comes back empty or the call blows up with an exception, exactly one trace lands in the ledger. The reason I could now pick up that "empty response from three weeks ago" as a single status="empty" line is that I moved to this structure. I keep only the fingerprint of the input, never the raw text. A fingerprint is enough for reconciliation, and there is no reason to warehouse PII for the long term.

Choosing a sink: what 100,000 records actually cost

The sink.append_only() my wrapper calls started life as a plain JSON Lines file append. It is the shortest thing that works, and append-only comes for free.

What bothered me was shipping it without knowing how it would feel once records piled up. So I ran 100,000 records through it locally and measured. Same record shape as above, with a realistic status mix (97% ok, 2% empty, 1% error).

Those numbers moved my hot tier to SQLite. Even with two indexes it lands smaller than uncompressed JSON Lines. At 3,000 calls a day, ninety days works out to roughly half a million rows and something near 90 MB. For a solo operator that is a size I can live with.

I measured write latency too. Inserting one row at a time into a WAL-mode SQLite database and committing each time: p50 of 0.31 ms, p95 of 1.29 ms, p99 of 1.49 ms. Against a model round trip measured in hundreds of milliseconds to seconds, writing to the ledger never shows up in response time. Had it been heavy I would have put an async queue in front of it. It was not.

import sqlite3
from dataclasses import asdict
 
DDL = """
CREATE TABLE IF NOT EXISTS interactions(
  interaction_id      TEXT PRIMARY KEY,
  prev_interaction_id TEXT,
  prompt_fingerprint  TEXT,
  feature             TEXT,          -- which product surface made the call
  model               TEXT,
  input_tokens        INTEGER,
  output_tokens       INTEGER,
  latency_ms          INTEGER,
  status              TEXT,
  created_at          REAL
);
CREATE INDEX IF NOT EXISTS idx_created ON interactions(created_at);
CREATE INDEX IF NOT EXISTS idx_status  ON interactions(status);
"""
 
 
class SqliteSink:
    """Hot-tier ledger. INSERT OR IGNORE keeps retries idempotent."""
 
    COLS = ("interaction_id", "prev_interaction_id", "prompt_fingerprint",
            "feature", "model", "input_tokens", "output_tokens",
            "latency_ms", "status", "created_at")
 
    def __init__(self, path: str):
        self.con = sqlite3.connect(path, check_same_thread=False)
        self.con.execute("PRAGMA journal_mode=WAL")
        self.con.executescript(DDL)
 
    def append_only(self, record) -> None:
        row = asdict(record) if not isinstance(record, dict) else record
        placeholders = ",".join(f":{c}" for c in self.COLS)
        self.con.execute(
            f"INSERT OR IGNORE INTO interactions VALUES({placeholders})",
            {c: row.get(c) for c in self.COLS},
        )
        self.con.commit()

INSERT OR IGNORE is there so a retry delivering the same interaction_id twice cannot duplicate the ledger. Append-only and idempotent pair well, and that one clause is what keeps reconciliation trustworthy.

The feature column earns its place because it lets the cost attribution below fall out of a single query. One extra string at call time buys back information you cannot reconstruct later. Cheap insurance, the way I see it.

Demoting to the cold tier: project, then let gzip sit on it

The smallest row in that table — the cold projection — is not the ledger compressed whole. It is the set of columns I know I will still want, pulled out of rows older than 90 days and rewritten. Id, fingerprint, feature, tokens, status, timestamp: with those six, audits, cost attribution, and reproduction work all still hold. prev_interaction_id and the model string, on the other hand, almost never got read on old rows.

import gzip
import json
import sqlite3
import time
 
COLD_COLS = ("interaction_id", "prompt_fingerprint", "feature",
             "input_tokens", "output_tokens", "status", "created_at")
HOT_DAYS = 90
 
 
def demote_to_cold(db_path: str, cold_path: str, now: float | None = None) -> int:
    """Project rows older than HOT_DAYS into gzip, then drop them from the hot tier."""
    now = now or time.time()
    cutoff = now - HOT_DAYS * 86400
    con = sqlite3.connect(db_path)
    rows = con.execute(
        f"SELECT {','.join(COLD_COLS)} FROM interactions WHERE created_at < ?",
        (cutoff,),
    ).fetchall()
    if not rows:
        con.close()
        return 0
    # append mode: never rewrite the existing cold file
    with gzip.open(cold_path, "at", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(dict(zip(COLD_COLS, r))) + "\n")
    con.execute("DELETE FROM interactions WHERE created_at < ?", (cutoff,))
    con.commit()
    con.execute("VACUUM")   # DELETE alone does not shrink the file
    con.close()
    return len(rows)

The VACUUM at the end is there for a reason. SQLite keeps the freed pages internally after a DELETE and does not hand the space back to the filesystem. I ran without it for a while and only later noticed that the hot tier I thought I was pruning had grown well past what I expected. A demotion is finished when the file size says it is finished.

Opening the gzip in append mode ("at") causes fewer accidents than rewriting the archive monthly. Concatenated gzip streams decompress correctly, so the read side needs no changes no matter how many times you append.

Catching gaps with reconciliation

Splitting into two layers raises the next question: is my ledger actually missing anything? Here the fast lens helps in the reverse direction. Join the runs visible on the dashboard against the interaction_ids in your own log, and surface any id that exists on only one side.

def reconcile(dashboard_ids: set[str], sink_ids: set[str]) -> dict:
    """Join interaction_ids from the lens and the ledger, and return the skew."""
    missing_in_sink = dashboard_ids - sink_ids        # a recording gap; top priority
    missing_in_dashboard = sink_ids - dashboard_ids   # e.g. provider retention expired
    return {
        "missing_in_sink": sorted(missing_in_sink),
        "missing_in_dashboard": sorted(missing_in_dashboard),
        "sink_coverage": (
            len(sink_ids & dashboard_ids) / len(dashboard_ids)
            if dashboard_ids else 1.0
        ),
    }

If missing_in_sink appears, it is a strong signal that some call path is not going through your wrapper. Closing that path is the work of keeping the ledger complete. Conversely, missing_in_dashboard is often just the dashboard's retention expiring, and it is actually evidence that your ledger is doing its job. Understanding this asymmetry keeps you from pointing your alerts in the wrong direction.

I run this reconciliation as a daily batch and only notify when sink_coverage drops. Rather than a human reading every log, someone acts only when skew appears. For an indie developer, this "stay quiet when things are quiet" property is what decides whether a practice is sustainable.

Close the bypass first: finding calls that skip the wrapper

That reconciliation has a soft spot worth naming. If there is no machine-readable way to export interaction_ids from the dashboard, dashboard_ids is something a person assembles by hand. I only pull the window I am actually investigating; I do not diff every call, every day.

So I added a way to attack the cause of missing_in_sink directly, without waiting for the batch: catch raw client calls that never went through the wrapper, at runtime.

import contextvars
import functools
import logging
 
log = logging.getLogger("interactions.guard")
_inside_wrapper = contextvars.ContextVar("inside_wrapper", default=False)
 
 
def install_bypass_guard(client, *, strict: bool = False) -> None:
    """Detect direct calls on the raw client. Calls from ObservedClient pass through."""
    raw = client.interactions.create
 
    @functools.wraps(raw)
    def guarded(*args, **kwargs):
        if not _inside_wrapper.get():
            msg = "Interactions call did not go through ObservedClient"
            if strict:
                raise RuntimeError(msg)
            log.warning("%s (model=%s)", msg, kwargs.get("model"))
        return raw(*args, **kwargs)
 
    client.interactions.create = guarded

Inside ObservedClient.run(), set token = _inside_wrapper.set(True) right after entering the try, and call _inside_wrapper.reset(token) in the finally. The flag lives in a ContextVar rather than a module global so concurrent calls cannot bleed into each other. My first version used a plain global, and the warning appeared and disappeared depending on how the coroutines interleaved — that cost me an afternoon.

I run strict=True in staging and warning-only in production. Bypasses almost always come from an admin script added later, or a small verification snippet someone wrote to check something once. Staging is as far as failing loudly gets you, if I am honest about it.

Since this went in, the only skew my daily reconciliation reports is the dashboard's own retention expiring. Preventing a gap turns out to help future-me more than finding one does.

The question only a ledger can answer: cost by feature

The clearest payoff from the split showed up the first time someone asked me to break down a bill.

You look at the month's API charges and want to know which surface spent them. The dashboard lets you inspect runs one at a time, but it gives you no way to bundle them by feature and total them up. With a ledger of your own, that is one query.

SELECT feature,
       model,
       COUNT(*)                              AS calls,
       SUM(input_tokens)                     AS in_tok,
       SUM(output_tokens)                    AS out_tok,
       ROUND(AVG(latency_ms))                AS avg_ms,
       ROUND(100.0 * SUM(status <> 'ok') / COUNT(*), 2) AS fail_pct
FROM interactions
WHERE created_at >= :period_start
GROUP BY feature, model
ORDER BY in_tok + out_tok DESC;

Across 100,000 rows this aggregation returned in 85.8 ms. Nothing you wait on when building a monthly report. Prices move, so I multiply the token counts by whatever the current rate card says rather than baking a unit price into the schema. Keeping the rate out of the data means a pricing change never retroactively corrupts an old rollup.

The breakdown told me I had been wrong about my own system. The feature with the most calls and the feature burning the most tokens were not the same one. The high-frequency surface was a short classification prompt fired over and over; the spend was concentrated in a much rarer long-document summarization path. Had I ordered my optimization work by call count, I would have shaved the wrong thing.

fail_pct sits in the same query because I want cost and failure side by side. A feature that consumes a lot of tokens and also fails often is expensive and unreliable at once, which is a different problem from either alone. Having those two columns adjacent took most of the guesswork out of prioritizing.

The calls that paid off in operation

As I kept the design running, a few concrete lines hardened. Here they are, by situation.

The one that paid off most was separating empty from error. At first I lumped both under "failure," but an empty response points at model-side suppression or filtering, while an exception points at my own implementation or the network. You cannot put things with different likely causes into the same bucket. Had I known the opening incident was empty, my very first investigative move would have been different.

Where to start

If you already run the Interactions API in production, the first step can be small. Wrap a single model call with the wrapper above and append just the interaction_id and status. Then install the bypass guard in warning mode and watch it for a week before you build any batch job — the paths it names are the ones that would have shown up as gaps months later. That much alone means that the next time someone asks about "that one call three weeks ago," you have a line you can trace.

The fast lens will keep getting more convenient. That is exactly why you keep the source of truth on your own side. I have come to see that small extra step as something that pays back later rather than today.

I am still tuning my reconciliation threshold as I operate this. If it gives anyone else who supports a service on their own a reason to revisit their design, I would be glad. Thank you for reading.