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.

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.

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. That 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 a quiet kind of reassurance.

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.