Call the per-attempt provider log requests, not a ledger
"ledger" already names the daemon's store handle (runtime.ledger) and the work and schedule tables, so loop code read ledger.record(runtime.ledger(host), ...). The module is albedo/daemon/requests, the table provider_requests, the route GET /sessions/:id/requests, and robot-docs/provider-requests.md. Nothing outside this branch has stored rows yet, so no migration.
1233456711 unmodified lines1920212223232425262725 unmodified lines# cache TTL table
The prompt-cache TTL table is the prior the cache-warmth estimate starts from: how long each provider keeps a cached prefix, what a write costs, and whether a hit restarts the clock. It answers `lookup(extension, host, model)` in `src/albedo/harness/cache_ttl.gleam` until a session's request ledger has measured the real thing (robot-docs/usage-ledger.md); after that the ledger wins.The prompt-cache TTL table is the prior the cache-warmth estimate starts from: how long each provider keeps a cached prefix, what a write costs, and whether a hit restarts the clock. It answers `lookup(extension, host, model)` in `src/albedo/harness/cache_ttl.gleam` until a session's provider requests have measured the real thing (robot-docs/provider-requests.md); after that the measurements win.
The table is data, not code: `priv/cache-ttl.json` ships with the daemon, and nothing about a provider is hardcoded anywhere else.
## format11 unmodified lines- `source` (url), `checked` (date), `note` (text).
Unknown fields are ignored, so a newer file stays readable. A bad entry — missing `id`, `match`, `policy` or `evidence`, or a value that does not decode — is skipped with a logged reason and never fails the table.
**Values marked `folklore` or `unknown` are placeholders.** They are the shape of the answer, not the answer: phase 2 replaces them with what the session's own ledger rows measure.**Values marked `folklore` or `unknown` are placeholders.** They are the shape of the answer, not the answer: phase 2 replaces them with what the session's own request rows measure.
## layers
Three layers merge by id:25 unmodified lines27 unmodified lines2829303132323334353636373838394041426 unmodified lines27 unmodified lineseach ping costs about `read × cached` input; letting the cache go cold costs one rewrite of about `write × cached` — `write` the matching tier's write multiplier (default 1.0), `read` the entry's read multiplier (default 0.1). so at most `floor(write / read)` consecutive pings per idle stretch, then it stops and lets the cache go cold: ski rental, with the break-even at Claude's 5-minute tier at 1.25 / 0.1 = 12 pings ≈ 54 minutes. a new turn resets the budget.
warming also stops when:
- a ping comes back with no cached input tokens although the turn it repeats read or wrote cache — the TTL model was wrong, and the ledger keeps the evidence.- a ping comes back with no cached input tokens although the turn it repeats read or wrote cache — the TTL model was wrong, and its request row keeps the evidence.- the kernel was released, or the ping failed or was cancelled.- settings say `enabled: false`.
## the ledger kind## the request kind
every ping is a ledger row of kind `warm`, with the same prefix identity (`headHash`, `inputs`, `replaced`, `projectionHash`) and `cacheMarks` as the turn it repeats, and its own usage. its `cachedInputTokens` says whether the cache was still there, which makes every ping a free TTL measurement for phase 2. a ping never attaches a seq, never commits to the transcript, and never publishes stream events; the agents bus does not flap for it (robot-docs/usage-ledger.md).every ping is a provider request row of kind `warm`, with the same prefix identity (`headHash`, `inputs`, `replaced`, `projectionHash`) and `cacheMarks` as the turn it repeats, and its own usage. its `cachedInputTokens` says whether the cache was still there, which makes every ping a free TTL measurement for phase 2. a ping never attaches a seq, never commits to the transcript, and never publishes stream events; the agents bus does not flap for it (robot-docs/provider-requests.md).
## settings
`$ALBEDO_HOME/extensions.json` under `"warm"`, re-read each time a ping is scheduled:6 unmodified lines12345678910111213141516171819202122232425262728293031323334353637# Provider requests
Every provider call the daemon makes leaves one row in `provider_requests`, a durable account of what the transcript cannot say later: what the request cost, which account served it, how it ended, and what its prefix looked like. This is the recording half of cache- and quota-aware compaction; the analysis (a meter estimator and λ) is derived from these rows on demand, so nothing beyond them is stored.
Rows are written by the worker that streams the call, one per upstream attempt: a turn retried after a gateway hiccup leaves the failed attempt and the successful one as separate rows. Recording must never fail a turn — a failed row write is logged and swallowed, and the call proceeds.
## What a row holds
- `session`, `seq` — the conversation, and the transcript row the call produced: the first assistant row of the response that turn committed (for a tool-call turn, the row carrying the call itself, not the tool output after it). `seq` is null when the call produced no transcript row: a failed call, or a compaction summary, which writes strategy state instead of the transcript.- `kind` — `turn` for the model/tool loop, `summarizer` for the compaction summary call, `warm` for a cache-warming ping that repeats the last turn's request with a tiny output budget. A `warm` row never attaches a seq: pings commit nothing.- `profile`, `provider`, `account`, `model` — the saved profile the request went through, `protocol:endpoint` as it was reached, the non-secret label of the account that served (an account id, email, or short key hash; null when the provider has no account pool), and the model.- `startedMs`, `finishedMs` — when the attempt ran.- `outcome`, `status`, `error` — `ok`, or `error` with the HTTP status and a truncated error body, so a 429 is a quota reading.- token counts as the provider reported them: `inputTokens`, `cachedInputTokens`, `cacheCreationTokens`, `cacheWrite5mTokens`, `cacheWrite1hTokens`, `outputTokens`, `reasoningTokens`. Unreported counts are null, not zero.
Providers report these differently and each adapter fills what it can: Anthropic's `cache_creation.ephemeral_5m_input_tokens` / `ephemeral_1h_input_tokens` become the TTL-split writes; OpenAI's `completion_tokens_details.reasoning_tokens` and the Responses API's `output_tokens_details.reasoning_tokens` become reasoning; Antigravity's `thoughtsTokenCount` is reported as reasoning while output keeps folding thoughts in, as before.
## Prefix identity
What makes cache analysis possible without storing requests. Each row records:
- `headHash` — SHA-256 of the request head: instructions plus tool schemas. One head, one hash.- `inputs` — how many projected inputs the request carried.- `replaced` and `projectionHash` — the projection identity. The loop knows how many original inputs the projection replaced (`original length − shared suffix with the projection`); the row carries that count and a hash of the replacement part that stands in for them.- `strategy` — the active compaction strategy name, when one prepared the request.- `cacheMarks` — where the request asked the provider to end a cached prefix, and for how long, in prefix order: `{"through": "tools" | "system" | "input", "index"?, "ttlSeconds"}`, where `index` is the projected input the prefix runs through. Claude requests mark the last tool and the system prompt for an hour and the last input (unless it is a replayed assistant turn) for Anthropic's default five minutes. Usage splits cache writes by TTL but not reads; with the marks, a read's TTL follows from which marked prefix it covers. Providers that cache on their own (the OpenAI protocols) get `[]`.
Within one projection the request is append-only, so equal head hash plus equal projection identity means a warm prefix; a changed projection identity means a rewrite that invalidated the cache. A summarizer call carries its own head hash with `replaced` and `projectionHash` null: its history is exactly what it says.
## Local API
`GET /sessions/:id/requests` answers rows oldest first, read-only straight from the store:
- `?after=<id>` continues from the last row id of the previous page.- `?limit=<n>` bounds the page (1–500, default 100).
The response is `{session, rows, after}`; `after` is the id to continue from. Like every session route it needs the daemon token. The later `/quota` inspector reads the same rows; nothing is stored twice.12345678910111213141516171819202122232425262728293031323334353637# Usage ledger
Every provider call the daemon makes leaves one row in `request_ledger`, a durable account of what the transcript cannot say later: what the request cost, which account served it, how it ended, and what its prefix looked like. This is the recording half of cache- and quota-aware compaction; the analysis (a meter estimator and λ) is derived from these rows on demand, so nothing beyond them is stored.
Rows are written by the worker that streams the call, one per upstream attempt: a turn retried after a gateway hiccup leaves the failed attempt and the successful one as separate rows. Recording must never fail a turn — a ledger write error is logged and swallowed, and the call proceeds.
## What a row holds
- `session`, `seq` — the conversation, and the transcript row the call produced: the first assistant row of the response that turn committed (for a tool-call turn, the row carrying the call itself, not the tool output after it). `seq` is null when the call produced no transcript row: a failed call, or a compaction summary, which writes strategy state instead of the transcript.- `kind` — `turn` for the model/tool loop, `summarizer` for the compaction summary call, `warm` for a cache-warming ping that repeats the last turn's request with a tiny output budget. A `warm` row never attaches a seq: pings commit nothing.- `profile`, `provider`, `account`, `model` — the saved profile the request went through, `protocol:endpoint` as it was reached, the non-secret label of the account that served (an account id, email, or short key hash; null when the provider has no account pool), and the model.- `startedMs`, `finishedMs` — when the attempt ran.- `outcome`, `status`, `error` — `ok`, or `error` with the HTTP status and a truncated error body, so a 429 is a quota reading.- token counts as the provider reported them: `inputTokens`, `cachedInputTokens`, `cacheCreationTokens`, `cacheWrite5mTokens`, `cacheWrite1hTokens`, `outputTokens`, `reasoningTokens`. Unreported counts are null, not zero.
Providers report these differently and each adapter fills what it can: Anthropic's `cache_creation.ephemeral_5m_input_tokens` / `ephemeral_1h_input_tokens` become the TTL-split writes; OpenAI's `completion_tokens_details.reasoning_tokens` and the Responses API's `output_tokens_details.reasoning_tokens` become reasoning; Antigravity's `thoughtsTokenCount` is reported as reasoning while output keeps folding thoughts in, as before.
## Prefix identity
What makes cache analysis possible without storing requests. Each row records:
- `headHash` — SHA-256 of the request head: instructions plus tool schemas. One head, one hash.- `inputs` — how many projected inputs the request carried.- `replaced` and `projectionHash` — the projection identity. The loop knows how many original inputs the projection replaced (`original length − shared suffix with the projection`); the row carries that count and a hash of the replacement part that stands in for them.- `strategy` — the active compaction strategy name, when one prepared the request.- `cacheMarks` — where the request asked the provider to end a cached prefix, and for how long, in prefix order: `{"through": "tools" | "system" | "input", "index"?, "ttlSeconds"}`, where `index` is the projected input the prefix runs through. Claude requests mark the last tool and the system prompt for an hour and the last input (unless it is a replayed assistant turn) for Anthropic's default five minutes. Usage splits cache writes by TTL but not reads; with the marks, a read's TTL follows from which marked prefix it covers. Providers that cache on their own (the OpenAI protocols) get `[]`.
Within one projection the request is append-only, so equal head hash plus equal projection identity means a warm prefix; a changed projection identity means a rewrite that invalidated the cache. A summarizer call carries its own head hash with `replaced` and `projectionHash` null: its history is exactly what it says.
## Local API
`GET /sessions/:id/ledger` answers rows oldest first, read-only straight from the store:
- `?after=<id>` continues from the last row id of the previous page.- `?limit=<n>` bounds the page (1–500, default 100).
The response is `{session, rows, after}`; `after` is the id to continue from. Like every session route it needs the daemon token. The later `/quota` inspector reads the same rows; nothing is stored twice.123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182"""The request ledger: one durable row per provider call, recording what cacheand quota analysis cannot recover from the transcript.
Catches: usage counts that never reach the ledger (cache reads, TTL-splitwrites, reasoning), the account/provider identity of each call, the prefixidentity that distinguishes a warm append from a compaction rewrite, an HTTPfailure recorded as a quota reading, and the seq link to the transcript. Noneof this is observable from events or history alone."""import jsonimport unittest
from harness import Albedo, Provider, Reply, text
FIRST = {"input_tokens": 100, "output_tokens": 20, "input_tokens_details": {"cached_tokens": 40}, "output_tokens_details": {"reasoning_tokens": 7}}TOOL = {"input_tokens": 200, "output_tokens": 30, "input_tokens_details": {"cached_tokens": 80}, "output_tokens_details": {"reasoning_tokens": 11}}AFTER = {"input_tokens": 300, "output_tokens": 40, "input_tokens_details": {"cached_tokens": 120}, "output_tokens_details": {"reasoning_tokens": 3}}THIRD = {"input_tokens": 400, "output_tokens": 50, "input_tokens_details": {"cached_tokens": 160}, "output_tokens_details": {"reasoning_tokens": 5}}SUMMARY = {"input_tokens": 60, "output_tokens": 10}POST = {"input_tokens": 70, "output_tokens": 15, "input_tokens_details": {"cached_tokens": 30}, "output_tokens_details": {"reasoning_tokens": 2}}
def reply(request): inputs = request["input"] # The tool result arrives while the newest user message is still the one # that asked for the tool, so it is told apart by input shape. if inputs[-1].get("type") == "function_call_output": return text("tool done", usage=AFTER) user = next((str(item.get("content", "")) for item in reversed(inputs) if item.get("role") == "user"), "") if "<newly-evicted-history>" in user: return text("older conversation summary", usage=SUMMARY) if user == "use the tool": return Reply("python", "2 + 2", usage=TOOL) if user == "hit the limit": return Reply("error", "limit exceeded", status=429) if user == "after compaction": return text("post compaction reply", usage=POST) if user == "third": return text("third reply", usage=THIRD) return text("first reply", usage=FIRST)
class LedgerTest(unittest.TestCase): def rows(self, app, session, query=""): with app.api(f"/sessions/{session}/ledger{query}") as response: return json.load(response)
def tree(self, app, session): """Every transcript row through the tree route, oldest first.""" items, after = [], 0 while True: with app.api(f"/sessions/{session}/tree?after={after}&limit=100") \ as response: page = json.load(response) items.extend(page["items"]) if not page["hasMore"]: return items after = page["nextCursor"]
def test_rows_record_usage_identity_prefix_and_projection(self): provider = Provider(reply) try: with Albedo(provider, protocol="responses") as app: session = app.session() for prompt in ("first", "use the tool", "third", "hit the limit"): app.prompt(session, prompt).close() app.idle(session) with app.api(f"/sessions/{session}/commands", {"name": "/compact"}) as response: self.assertTrue(json.load(response)["result"]["started"]) app.idle(session) app.prompt(session, "after compaction").close() app.idle(session)
page = self.rows(app, session) rows = page["rows"] self.assertEqual([row["kind"] for row in rows], ["turn", "turn", "turn", "turn", "turn", "summarizer", "turn"]) # Every call carries its own reported counts, split as sent. self.assertEqual([(row["inputTokens"], row["outputTokens"], row["cachedInputTokens"], row["reasoningTokens"]) for row in rows], [(100, 20, 40, 7), (200, 30, 80, 11), (300, 40, 120, 3), (400, 50, 160, 5), (None, None, None, None), (60, 10, None, None), (70, 15, 30, 2)]) # The head (instructions and tools) never changed, so one hash. heads = {row["headHash"] for row in rows if row["kind"] == "turn"} self.assertEqual(len(heads), 1) self.assertNotIn(rows[5]["headHash"], heads)
for row in rows: self.assertIsNone(row["account"]) # The shared harness names the fixture profile per route. self.assertTrue(row["profile"].startswith("fixture")) self.assertEqual(row["model"], "fixture-model") self.assertTrue(row["provider"].startswith("responses:")) self.assertLessEqual(row["startedMs"], row["finishedMs"]) # An OpenAI-protocol provider caches on its own; albedo marks nothing. self.assertEqual(row["cacheMarks"], [])
for row in rows[:4] + [rows[6]]: self.assertEqual(row["outcome"], "ok") self.assertIsNone(row["status"]) self.assertIsNone(row["error"])
for row in rows[:4]: # Nothing replaced yet: the identity is a plain append. self.assertEqual(row["replaced"], 0) self.assertIsNone(row["projectionHash"])
# A 429 is a quota reading, with no usage and no transcript row. limited = rows[4] self.assertEqual(limited["outcome"], "error") self.assertEqual(limited["status"], 429) self.assertIn("limit exceeded", limited["error"]) self.assertIsNone(limited["seq"])
# Each turn row's seq is the transcript row its response # committed, exactly: the assistant row that turn produced, # found again through the tree route straight from the # transcript. tree = self.tree(app, session) assistant = {item["preview"]: item["id"] for item in tree if item["type"] == "assistant"} self.assertEqual(rows[0]["seq"], assistant["first reply"]) self.assertEqual(rows[2]["seq"], assistant["tool done"]) self.assertEqual(rows[3]["seq"], assistant["third reply"]) self.assertEqual(rows[6]["seq"], assistant["post compaction reply"]) # The tool-call turn's row points at its own assistant row, # the call it produced, not the tool output after it. calls = [item for item in tree if item["preview"] == "call python"] self.assertEqual(len(calls), 1) self.assertEqual(rows[1]["seq"], calls[0]["id"]) tool_output = tree[tree.index(calls[0]) + 1] self.assertEqual(tool_output["type"], "tool") self.assertNotEqual(tool_output["id"], rows[1]["seq"]) # Linked rows sit in transcript order. linked = [row["seq"] for row in rows if row["seq"] is not None] self.assertEqual(linked, sorted(linked))
# The compaction summary call is its own kind, outside the # session projection. summary = rows[5] self.assertIsNone(summary["seq"]) self.assertIsNone(summary["replaced"]) self.assertIsNone(summary["projectionHash"]) self.assertEqual(summary["inputs"], 1)
# The projection identity changes once compaction rewrites the # head the verbatim tail hangs from. after = rows[6] self.assertGreater(after["replaced"], 0) self.assertIsNotNone(after["projectionHash"]) self.assertTrue(after["strategy"])
# Paging: continue after the first row, bounded by limit. second = self.rows(app, session, f"?after={rows[0]['id']}&limit=2") self.assertEqual([row["id"] for row in second["rows"]], [row["id"] for row in rows[1:3]]) self.assertEqual(second["after"], rows[2]["id"]) finally: provider.close()
if __name__ == "__main__": unittest.main()123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182"""Provider requests: one durable row per attempt albedo sends, recording what cacheand quota analysis cannot recover from the transcript.
Catches: usage counts that never reach a request row (cache reads, TTL-splitwrites, reasoning), the account/provider identity of each call, the prefixidentity that distinguishes a warm append from a compaction rewrite, an HTTPfailure recorded as a quota reading, and the seq link to the transcript. Noneof this is observable from events or history alone."""import jsonimport unittest
from harness import Albedo, Provider, Reply, text
FIRST = {"input_tokens": 100, "output_tokens": 20, "input_tokens_details": {"cached_tokens": 40}, "output_tokens_details": {"reasoning_tokens": 7}}TOOL = {"input_tokens": 200, "output_tokens": 30, "input_tokens_details": {"cached_tokens": 80}, "output_tokens_details": {"reasoning_tokens": 11}}AFTER = {"input_tokens": 300, "output_tokens": 40, "input_tokens_details": {"cached_tokens": 120}, "output_tokens_details": {"reasoning_tokens": 3}}THIRD = {"input_tokens": 400, "output_tokens": 50, "input_tokens_details": {"cached_tokens": 160}, "output_tokens_details": {"reasoning_tokens": 5}}SUMMARY = {"input_tokens": 60, "output_tokens": 10}POST = {"input_tokens": 70, "output_tokens": 15, "input_tokens_details": {"cached_tokens": 30}, "output_tokens_details": {"reasoning_tokens": 2}}
def reply(request): inputs = request["input"] # The tool result arrives while the newest user message is still the one # that asked for the tool, so it is told apart by input shape. if inputs[-1].get("type") == "function_call_output": return text("tool done", usage=AFTER) user = next((str(item.get("content", "")) for item in reversed(inputs) if item.get("role") == "user"), "") if "<newly-evicted-history>" in user: return text("older conversation summary", usage=SUMMARY) if user == "use the tool": return Reply("python", "2 + 2", usage=TOOL) if user == "hit the limit": return Reply("error", "limit exceeded", status=429) if user == "after compaction": return text("post compaction reply", usage=POST) if user == "third": return text("third reply", usage=THIRD) return text("first reply", usage=FIRST)
class ProviderRequestsTest(unittest.TestCase): def rows(self, app, session, query=""): with app.api(f"/sessions/{session}/requests{query}") as response: return json.load(response)
def tree(self, app, session): """Every transcript row through the tree route, oldest first.""" items, after = [], 0 while True: with app.api(f"/sessions/{session}/tree?after={after}&limit=100") \ as response: page = json.load(response) items.extend(page["items"]) if not page["hasMore"]: return items after = page["nextCursor"]
def test_rows_record_usage_identity_prefix_and_projection(self): provider = Provider(reply) try: with Albedo(provider, protocol="responses") as app: session = app.session() for prompt in ("first", "use the tool", "third", "hit the limit"): app.prompt(session, prompt).close() app.idle(session) with app.api(f"/sessions/{session}/commands", {"name": "/compact"}) as response: self.assertTrue(json.load(response)["result"]["started"]) app.idle(session) app.prompt(session, "after compaction").close() app.idle(session)
page = self.rows(app, session) rows = page["rows"] self.assertEqual([row["kind"] for row in rows], ["turn", "turn", "turn", "turn", "turn", "summarizer", "turn"]) # Every call carries its own reported counts, split as sent. self.assertEqual([(row["inputTokens"], row["outputTokens"], row["cachedInputTokens"], row["reasoningTokens"]) for row in rows], [(100, 20, 40, 7), (200, 30, 80, 11), (300, 40, 120, 3), (400, 50, 160, 5), (None, None, None, None), (60, 10, None, None), (70, 15, 30, 2)]) # The head (instructions and tools) never changed, so one hash. heads = {row["headHash"] for row in rows if row["kind"] == "turn"} self.assertEqual(len(heads), 1) self.assertNotIn(rows[5]["headHash"], heads)
for row in rows: self.assertIsNone(row["account"]) # The shared harness names the fixture profile per route. self.assertTrue(row["profile"].startswith("fixture")) self.assertEqual(row["model"], "fixture-model") self.assertTrue(row["provider"].startswith("responses:")) self.assertLessEqual(row["startedMs"], row["finishedMs"]) # An OpenAI-protocol provider caches on its own; albedo marks nothing. self.assertEqual(row["cacheMarks"], [])
for row in rows[:4] + [rows[6]]: self.assertEqual(row["outcome"], "ok") self.assertIsNone(row["status"]) self.assertIsNone(row["error"])
for row in rows[:4]: # Nothing replaced yet: the identity is a plain append. self.assertEqual(row["replaced"], 0) self.assertIsNone(row["projectionHash"])
# A 429 is a quota reading, with no usage and no transcript row. limited = rows[4] self.assertEqual(limited["outcome"], "error") self.assertEqual(limited["status"], 429) self.assertIn("limit exceeded", limited["error"]) self.assertIsNone(limited["seq"])
# Each turn row's seq is the transcript row its response # committed, exactly: the assistant row that turn produced, # found again through the tree route straight from the # transcript. tree = self.tree(app, session) assistant = {item["preview"]: item["id"] for item in tree if item["type"] == "assistant"} self.assertEqual(rows[0]["seq"], assistant["first reply"]) self.assertEqual(rows[2]["seq"], assistant["tool done"]) self.assertEqual(rows[3]["seq"], assistant["third reply"]) self.assertEqual(rows[6]["seq"], assistant["post compaction reply"]) # The tool-call turn's row points at its own assistant row, # the call it produced, not the tool output after it. calls = [item for item in tree if item["preview"] == "call python"] self.assertEqual(len(calls), 1) self.assertEqual(rows[1]["seq"], calls[0]["id"]) tool_output = tree[tree.index(calls[0]) + 1] self.assertEqual(tool_output["type"], "tool") self.assertNotEqual(tool_output["id"], rows[1]["seq"]) # Linked rows sit in transcript order. linked = [row["seq"] for row in rows if row["seq"] is not None] self.assertEqual(linked, sorted(linked))
# The compaction summary call is its own kind, outside the # session projection. summary = rows[5] self.assertIsNone(summary["seq"]) self.assertIsNone(summary["replaced"]) self.assertIsNone(summary["projectionHash"]) self.assertEqual(summary["inputs"], 1)
# The projection identity changes once compaction rewrites the # head the verbatim tail hangs from. after = rows[6] self.assertGreater(after["replaced"], 0) self.assertIsNotNone(after["projectionHash"]) self.assertTrue(after["strategy"])
# Paging: continue after the first row, bounded by limit. second = self.rows(app, session, f"?after={rows[0]['id']}&limit=2") self.assertEqual([row["id"] for row in second["rows"]], [row["id"] for row in rows[1:3]]) self.assertEqual(second["after"], rows[2]["id"]) finally: provider.close()
if __name__ == "__main__": unittest.main()1 unmodified line2345667891011111213141588 unmodified lines10410510610710810810911011111232 unmodified lines14514614714814914915015115215364 unmodified lines1 unmodified lineits last request with a tiny output budget before the provider cache wouldexpire.
Catches: pings that rebuild the request instead of repeating the one the turnsent (which would compact or summarize), ping ledger rows that do not carrysent (which would compact or summarize), ping request rows that do not carrythe turn's prefix identity, pings that commit or publish anything, warmingthat never stops at its budget, warming that continues after the waking workfinished, and a submit that cannot get through while a ping holds the session.None of this is observable from the transcript alone; the provider and theledger are the witnesses.request rows are the witnesses."""import jsonimport threadingimport time88 unmodified lines return [request for request in session_requests if "max_output_tokens" in request]
def rows(self, session): with self.app.api(f"/sessions/{session}/ledger") as response: with self.app.api(f"/sessions/{session}/requests") as response: return json.load(response)["rows"]
def tree(self, session): items, after = [], 032 unmodified lines # A ping is the turn's request as sent, with only the output # budget lowered — the Responses protocol's minimum. self.assertEqual(ping, {**turn, "max_output_tokens": 16})
# The ledger rows carry the same prefix identity and cache marks as # The request rows carry the same prefix identity and cache marks as # the turn they repeat, and never a transcript row of their own. rows = self.rows(parent) self.assertEqual([row["kind"] for row in rows], ["turn", "warm", "warm"]) head = rows[0]64 unmodified lines12334567891059 unmodified lines70717273747475767778216 unmodified linesExpand all295296297298299299300301302303800 unmodified linesExpand allimport albedo/daemon/eventsimport albedo/daemon/imagesimport albedo/daemon/ledgerimport albedo/daemon/mailimport albedo/daemon/noteimport albedo/daemon/noticeimport albedo/daemon/requestsimport albedo/daemon/storeimport albedo/daemon/transcriptimport albedo/daemon/usageimport albedo/openai_api/types59 unmodified lines use _ <- result.try(store.exec( db, "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER,usage_cache_creation_tokens INTEGER,effort TEXT); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT,thought_ms INTEGER); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);" <> images.schema <> ledger.schema, <> requests.schema, )) use _ <- result.try( store.add_columns(db, "sessions", [ #("provider", "TEXT"),216 unmodified linesExpand all use _ <- result.try( list.try_each( [ "transcript", "request_ledger", "provider_requests", "schedules", "session_extensions", "rolling_compaction_state", "rolling_compaction_observation",800 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516//// One durable row per provider call.//////// The transcript keeps what a turn said; it cannot say what the request//// looked like when the provider saw it. Each row records what later cache//// and quota analysis cannot recover: which account served, how long the//// call took, what it cost in tokens (cached reads, cache writes split by//// TTL, reasoning), how it ended (an HTTP status turns a 429 into a quota//// reading), and a prefix identity — a hash of the request head plus the//// projection that stands in for replaced history — so two rows can be told//// apart as a warm append or a cache-busting rewrite without storing any//// request body. Requests are never stored, only hashed.
import albedo/daemon/storeimport albedo/harness/compactionimport albedo/openai_api/typesimport gleam/bit_arrayimport gleam/cryptoimport gleam/dynamic/decodeimport gleam/json.{type Json}import gleam/listimport gleam/option.{type Option, None, Some}import gleam/resultimport gleam/stringimport sqlight
pub const schema = "CREATE TABLE IF NOT EXISTS request_ledger(id INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),seq INTEGER,kind TEXT NOT NULL,profile TEXT NOT NULL,provider TEXT NOT NULL,account TEXT,model TEXT NOT NULL,started_ms INTEGER NOT NULL,finished_ms INTEGER NOT NULL,outcome TEXT NOT NULL,status INTEGER,error TEXT,input_tokens INTEGER,cached_input_tokens INTEGER,cache_creation_tokens INTEGER,cache_write_5m_tokens INTEGER,cache_write_1h_tokens INTEGER,output_tokens INTEGER,reasoning_tokens INTEGER,head_hash TEXT NOT NULL,inputs INTEGER NOT NULL,replaced INTEGER,projection_hash TEXT,strategy TEXT,cache_marks TEXT NOT NULL DEFAULT '[]'); CREATE INDEX IF NOT EXISTS request_ledger_session ON request_ledger(session,id);"
/// What the call was for. A turn is the model loop; a summarizer is the/// compaction summary call `loop.summarize` makes; a warm is a cache-warming/// ping that repeats the last turn's request with a tiny output budget.pub type Kind { Turn Summarizer Warm}
fn kind_name(kind: Kind) -> String { case kind { Turn -> "turn" Summarizer -> "summarizer" Warm -> "warm" }}
/// How a call ended: completed, or failed with the HTTP status when there was/// one and a bounded detail that keeps an error body readable, not whole.pub type Outcome { Completed Failed(status: Option(Int), detail: String)}
pub fn outcome(result: Result(a, types.Error)) -> Outcome { case result { Ok(_) -> Completed Error(types.HttpError(status, body)) -> Failed(Some(status), bounded(body)) Error(types.ProviderError(message)) -> Failed(None, bounded(message)) Error(error) -> Failed(None, string.inspect(error)) }}
/// Error bodies are truncated so one huge response cannot bloat the ledger.const detail_limit = 2000
fn bounded(text: String) -> String { case string.length(text) > detail_limit { True -> string.slice(text, 0, detail_limit) <> "…[truncated]" False -> text }}
/// What makes cache analysis possible without storing requests: the request/// head (instructions and tools) as a hash, how many projected inputs the/// request carries, and a projection identity that changes exactly when/// compaction changes what precedes the verbatim tail — the count of original/// inputs the projection replaced and a hash of the replacement part. Within/// one projection the request is append-only, so these tell a warm prefix/// from a rewrite.pub type Prefix { Prefix( head_hash: String, inputs: Int, replaced: Option(Int), projection_hash: Option(String), strategy: Option(String), )}
/// The prefix identity of a projected request: `projected` is the history the/// request carries, `original` the durable history it stands in for, and/// `strategy` the compaction strategy that prepared it, when one did.pub fn prefix( instructions: String, tools: List(types.Tool), projected: List(types.Input), original: List(types.Input), strategy: Option(String),) -> Prefix { let suffix = compaction.common_suffix(original, projected) let replaced = list.length(original) - suffix Prefix( head_hash: head_hash(instructions, tools), inputs: list.length(projected), replaced: Some(replaced), projection_hash: case replaced { 0 -> None _ -> Some(inputs_hash(list.take(projected, list.length(projected) - suffix))) }, strategy: strategy, )}
/// The prefix identity of a request outside the session projection, whose/// history is exactly what it says: nothing replaced anything.pub fn direct_prefix( instructions: String, tools: List(types.Tool), inputs: List(types.Input),) -> Prefix { Prefix( head_hash: head_hash(instructions, tools), inputs: list.length(inputs), replaced: None, projection_hash: None, strategy: None, )}
/// The hash of a request head: its instructions and the tools it offers.pub fn head_hash(instructions: String, tools: List(types.Tool)) -> String { json.object([ #("instructions", json.string(instructions)), #( "tools", json.array(tools, fn(tool) { json.object([ #("name", json.string(tool.name)), #("description", json.string(tool.description)), #("parameters", tool.parameters), #("strict", json.bool(tool.strict)), ]) }), ), ]) |> json.to_string |> hash}
/// The hash of a run of inputs, as the ledger identities them.pub fn inputs_hash(inputs: List(types.Input)) -> String { inputs |> list.map(input_identity) |> string.join("\n") |> hash}
fn input_identity(input: types.Input) -> String { case input { types.User(text) -> "user\n" <> text types.Assistant(text) -> "assistant\n" <> text types.UserImage(text, image) -> "user_image\n" <> image_identity(image) <> "\n" <> text types.ToolOutput(id, output, images) -> "tool_output\n" <> id <> "\n" <> output <> string.concat( list.map(images, fn(image) { "\n" <> image_identity(image) }), ) types.Replay(item) -> "replay\n" <> json.to_string(types.replay_json(item)) }}
/// An image by the content identity its payload already carries.fn image_identity(image: types.Image) -> String { case types.image_data(image) { types.InlineData(data) -> hash(data) types.StoredData(hash, _, _) -> hash }}
fn hash(text: String) -> String { text |> bit_array.from_string |> crypto.hash(crypto.Sha256, _) |> bit_array.base16_encode |> string.lowercase}
/// One provider call, as it is about to be recorded.pub type Call { Call( session: String, kind: Kind, /// The saved profile the request went through. profile: String, /// How the request reached the provider: protocol and endpoint. provider: String, /// Non-secret label of the account that served; None without a pool. account: Option(String), model: String, started_ms: Int, finished_ms: Int, outcome: Outcome, usage: Option(types.Usage), prefix: Prefix, /// Where the request asked the provider to cache, and for how long. cache_marks: List(types.CacheMark), )}
/// Writes one row and answers its id, so the caller can attach the transcript/// seq the call produced once it is committed.pub fn record(ledger: store.Store, call: Call) -> Result(Int, String) { store.query(ledger, fn(db) { use _ <- result.try(store.run(db, insert, values(call))) store.one( db, "SELECT last_insert_rowid()", [], decode.field(0, decode.int, decode.success), "request ledger row id", ) })}
// `seq` starts null and is attached after the call's transcript row commits.const insert = "INSERT INTO request_ledger(session,seq,kind,profile,provider,account,model,started_ms,finished_ms,outcome,status,error,input_tokens,cached_input_tokens,cache_creation_tokens,cache_write_5m_tokens,cache_write_1h_tokens,output_tokens,reasoning_tokens,head_hash,inputs,replaced,projection_hash,strategy,cache_marks) VALUES(?,NULL,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
fn values(call: Call) -> List(sqlight.Value) { let Call( session, kind, profile, provider, account, model, started_ms, finished_ms, outcome, usage, prefix, cache_marks, ) = call let Prefix(head_hash, inputs, replaced, projection_hash, strategy) = prefix let #(status, error) = case outcome { Completed -> #(None, None) Failed(status, detail) -> #(status, Some(bounded(detail))) } let #(input, output, cached, creation, write_5m, write_1h, reasoning) = case usage { Some(types.Usage( input, output, cached, creation, write_5m, write_1h, reasoning, )) -> #( Some(input), Some(output), cached, creation, write_5m, write_1h, reasoning, ) None -> #(None, None, None, None, None, None, None) } [ sqlight.text(session), sqlight.text(kind_name(kind)), sqlight.text(profile), sqlight.text(provider), sqlight.nullable(sqlight.text, account), sqlight.text(model), sqlight.int(started_ms), sqlight.int(finished_ms), sqlight.text(case outcome { Completed -> "ok" Failed(..) -> "error" }), sqlight.nullable(sqlight.int, status), sqlight.nullable(sqlight.text, error), sqlight.nullable(sqlight.int, input), sqlight.nullable(sqlight.int, cached), sqlight.nullable(sqlight.int, creation), sqlight.nullable(sqlight.int, write_5m), sqlight.nullable(sqlight.int, write_1h), sqlight.nullable(sqlight.int, output), sqlight.nullable(sqlight.int, reasoning), sqlight.text(head_hash), sqlight.int(inputs), sqlight.nullable(sqlight.int, replaced), sqlight.nullable(sqlight.text, projection_hash), sqlight.nullable(sqlight.text, strategy), sqlight.text(json.array(cache_marks, cache_mark_json) |> json.to_string), ]}
/// Attaches the transcript row a recorded call produced.pub fn attach( ledger: store.Store, session: String, row: Int, seq: Int,) -> Result(Nil, String) { store.write( ledger, "UPDATE request_ledger SET seq=? WHERE id=? AND session=?", [sqlight.int(seq), sqlight.int(row), sqlight.text(session)], )}
/// One stored row, as the read path answers it.pub type Row { Row( id: Int, session: String, seq: Option(Int), kind: String, profile: String, provider: String, account: Option(String), model: String, started_ms: Int, finished_ms: Int, outcome: String, status: Option(Int), error: Option(String), input_tokens: Option(Int), cached_input_tokens: Option(Int), cache_creation_tokens: Option(Int), cache_write_5m_tokens: Option(Int), cache_write_1h_tokens: Option(Int), output_tokens: Option(Int), reasoning_tokens: Option(Int), head_hash: String, inputs: Int, replaced: Option(Int), projection_hash: Option(String), strategy: Option(String), cache_marks: List(types.CacheMark), )}
/// Rows after `after`, oldest first, at most `limit`; the id to continue from.pub fn page( ledger: store.Store, session: String, after: Int, limit: Int,) -> Result(#(List(Row), Int), String) { store.query(ledger, fn(db) { use rows <- result.try(store.rows( db, "SELECT " <> columns <> " FROM request_ledger WHERE session=? AND id>? ORDER BY id LIMIT ?", [sqlight.text(session), sqlight.int(after), sqlight.int(limit)], row_decoder(), )) let next = rows |> list.last |> result.map(fn(row) { row.id }) |> result.unwrap(after) Ok(#(rows, next)) })}
// Named, in decoder order, so a column added later cannot shift a read.const columns = "id,session,seq,kind,profile,provider,account,model,started_ms,finished_ms,outcome,status,error,input_tokens,cached_input_tokens,cache_creation_tokens,cache_write_5m_tokens,cache_write_1h_tokens,output_tokens,reasoning_tokens,head_hash,inputs,replaced,projection_hash,strategy,cache_marks"
fn row_decoder() -> decode.Decoder(Row) { use id <- decode.field(0, decode.int) use session <- decode.field(1, decode.string) use seq <- decode.field(2, decode.optional(decode.int)) use kind <- decode.field(3, decode.string) use profile <- decode.field(4, decode.string) use provider <- decode.field(5, decode.string) use account <- decode.field(6, decode.optional(decode.string)) use model <- decode.field(7, decode.string) use started <- decode.field(8, decode.int) use finished <- decode.field(9, decode.int) use outcome <- decode.field(10, decode.string) use status <- decode.field(11, decode.optional(decode.int)) use error <- decode.field(12, decode.optional(decode.string)) use input <- decode.field(13, decode.optional(decode.int)) use cached <- decode.field(14, decode.optional(decode.int)) use creation <- decode.field(15, decode.optional(decode.int)) use write_5m <- decode.field(16, decode.optional(decode.int)) use write_1h <- decode.field(17, decode.optional(decode.int)) use output <- decode.field(18, decode.optional(decode.int)) use reasoning <- decode.field(19, decode.optional(decode.int)) use head_hash <- decode.field(20, decode.string) use inputs <- decode.field(21, decode.int) use replaced <- decode.field(22, decode.optional(decode.int)) use projection_hash <- decode.field(23, decode.optional(decode.string)) use strategy <- decode.field(24, decode.optional(decode.string)) use cache_marks <- decode.field(25, decode.then(decode.string, marks_decoder)) decode.success(Row( id, session, seq, kind, profile, provider, account, model, started, finished, outcome, status, error, input, cached, creation, write_5m, write_1h, output, reasoning, head_hash, inputs, replaced, projection_hash, strategy, cache_marks, ))}
/// One row as the read route serves it. Unreported counts are null, so a/// reader distinguishes unknown from zero.pub fn row_json(row: Row) -> Json { json.object([ #("id", json.int(row.id)), #("session", json.string(row.session)), #("seq", json.nullable(row.seq, json.int)), #("kind", json.string(row.kind)), #("profile", json.string(row.profile)), #("provider", json.string(row.provider)), #("account", json.nullable(row.account, json.string)), #("model", json.string(row.model)), #("startedMs", json.int(row.started_ms)), #("finishedMs", json.int(row.finished_ms)), #("outcome", json.string(row.outcome)), #("status", json.nullable(row.status, json.int)), #("error", json.nullable(row.error, json.string)), #("inputTokens", json.nullable(row.input_tokens, json.int)), #("cachedInputTokens", json.nullable(row.cached_input_tokens, json.int)), #("cacheCreationTokens", json.nullable(row.cache_creation_tokens, json.int)), #("cacheWrite5mTokens", json.nullable(row.cache_write_5m_tokens, json.int)), #("cacheWrite1hTokens", json.nullable(row.cache_write_1h_tokens, json.int)), #("outputTokens", json.nullable(row.output_tokens, json.int)), #("reasoningTokens", json.nullable(row.reasoning_tokens, json.int)), #("headHash", json.string(row.head_hash)), #("inputs", json.int(row.inputs)), #("replaced", json.nullable(row.replaced, json.int)), #("projectionHash", json.nullable(row.projection_hash, json.string)), #("strategy", json.nullable(row.strategy, json.string)), #("cacheMarks", json.array(row.cache_marks, cache_mark_json)), ])}
/// A mark as stored and served: `{"through":"tools"|"system"|"input",/// "index"?, "ttlSeconds"}`.fn cache_mark_json(mark: types.CacheMark) -> Json { let through = case mark.through { types.ToolsSpan -> [#("through", json.string("tools"))] types.SystemSpan -> [#("through", json.string("system"))] types.InputSpan(index) -> [ #("through", json.string("input")), #("index", json.int(index)), ] } json.object( list.append(through, [#("ttlSeconds", json.int(mark.ttl_seconds))]), )}
fn marks_decoder(stored: String) -> decode.Decoder(List(types.CacheMark)) { let mark = { use through <- decode.field("through", decode.string) use index <- decode.optional_field("index", 0, decode.int) use ttl_seconds <- decode.field("ttlSeconds", decode.int) case through { "tools" -> decode.success(types.ToolsSpan) "system" -> decode.success(types.SystemSpan) "input" -> decode.success(types.InputSpan(index)) _ -> decode.failure(types.SystemSpan, "a cache mark span") } |> decode.map(types.CacheMark(_, ttl_seconds)) } case json.parse(stored, decode.list(mark)) { Ok(marks) -> decode.success(marks) Error(_) -> decode.failure([], "stored cache marks") }}
/// The ms clock the ledger timestamps rows with.@external(erlang, "erlang", "system_time")fn system_time(unit: TimeUnit) -> Int
type TimeUnit { Millisecond}
pub fn now() -> Int { system_time(Millisecond)}
/// Rows a page asks for when it does not say.pub const page_rows = 100