feat(beacon): hourly R2 compaction, per-strategy savings, and a stack that reports (#2853)
Three beacon changes bundled because they are one story: the corpus got
too
slow to query, and then too coarse to answer the question it was
collected for.
## 1. Hourly compaction (`deploy/beacon`)
The beacon writes one ~1 KB object per heartbeat — **64,987 on
2026-08-06** and
climbing. A full analysis `pull` was ~100k HTTPS round trips for 95 MB:
minutes
of pure per-object latency. Listing the bucket alone took 88 seconds.
Moving the query server-side does not help — R2 SQL reads only Iceberg
tables,
and a pile of tiny files is the pathological case for every query
engine.
Compaction is the fix, and it is Iceberg's own answer to the same
problem.
An hourly cron collapses each **complete** hour of `sessions/` into one
`rollup/dt=…/hh=…/data.ndjson`, keeping the highest-`seq` heartbeat per
`(install, session)`.
| measured on `dt=2026-08-06/hh=14` | before | after |
|---|---|---|
| objects | 3,938 | **1** |
| rows | 3,938 | **1,061** |
| analysis `pull` | minutes | **seconds** |
Hourly rather than daily because every R2 binding call is a subrequest:
a day
is ~65k, an hour is ~4k. Newest-first, so a backlog drains from the
present
backwards and live data never starves behind it; a failing hour is
logged and
skipped rather than blocking every older hour behind it. **Raw objects
are
never deleted**, so any rollup is rebuildable by deleting it.
Backfill runs to the **oldest surviving raw day**, not a fixed window. A
fixed
lookback strands everything older than it the moment analysis stops
reading
`sessions/`: the raw objects are still there, but nothing would ever
compact
them, so they disappear from every report. `oldestRawDay()` finds that
floor in
one delimited LIST, and the rollup listing starts from it — so the work
is
bounded by retention rather than by total history.
Three failure modes the tests pin down, because each one is silent:
- A **failed `get`** is transient, so the hour throws and writes
nothing. A
rollup is built once and trusted forever, so a short read would quietly
become the permanent record.
- A **corrupt record** loses only itself. This Worker wrote that content
with
`JSON.stringify`; it will never become valid, so blocking on it strands
the
hour instead of the record.
- An **empty hour** writes an `empty` marker. Without one the hour stays
"missing" and is re-listed on every run forever.
`test-rollup.mjs` asserts the Worker's dedup picks exactly the same rows
as the
analysis-side `QUALIFY`. If those two ever disagree the reports go
quietly
wrong rather than loudly broken, which is why that check exists. Its
stub
paginates at 3 keys so the list cursor loop — load-bearing at the real
~4,000
objects/hour — runs in every case.
## 2. Per-strategy savings (`compression.by_strategy`)
`compression.transforms` counts *invocations*, which cannot distinguish
a
compressor that saved 60% from one that ran constantly and saved
nothing. The
fleet's top transform by count contributes an unknown share of
`tokens.saved`.
It is worse than that in practice. Transform labels are slugged with
`split(":", 1)[0]`, so every `router:<strategy>:<detail>` label
collapses into a
single `router` bucket. On 2026-08-08 that bucket held **19.1 M of the
day's
transform counts, across 8,528 of 9,616 sessions** — the compressors
that do
most of the work are indistinguishable from each other, by name as well
as by
yield:
| transform | n | sessions |
|---|---|---|
| `router` | 19,108,118 | 8,528 |
| `anthropic` | 688,124 | 4,734 |
| `output_shaper` | 548,017 | 1,120 |
No question about which strategy is earning its keep can be answered
from that,
which is what this field is for.
The measurement already existed. `PrometheusMetrics.record_compression`
is the
configured `CompressionObserver` and already accumulates
`tokens_saved_by_strategy` on the hot path — the numbers just never left
the
process. This forwards from that one chokepoint rather than adding a
second
observer and a second measurement pass. The paths that have **no**
observer
configured (MCP server, LangGraph, Strands hooks, the transform
pipeline) get
`BeaconCompressionObserver` passed directly.
Compression runs on the executor thread *before* that request's outcome
reaches
`record()`, so events are **staged** into module state and drained by
the next
outcome. Staging is what makes two things true at once:
- The first turn of a session still reports its numbers — otherwise
every
session's opening turn, and any session short enough to be one turn,
would
report nothing.
- A compression event **never opens a session**. An abandoned request
would
otherwise emit a phantom `turns=0` row with all-zero tokens, inflating
fleet
session and install counts.
Staging takes a dedicated mutex the request path never touches, so the
fan-out
stays off the aggregator's lock and `record_compression` keeps its
"synchronous + lock-free" contract.
```json
"by_strategy": [
{"strategy": "code_aware", "n": 1, "tokens_in": 800, "tokens_out": 800},
{"strategy": "smart_crusher", "n": 2, "tokens_in": 1500, "tokens_out": 700}
]
```
A **list of records, sorted by strategy** — not an object keyed by
strategy.
Keyed shapes change type as keys accumulate: DuckDB infers a STRUCT
under ~24
keys and a MAP over it, so the analysis query breaks on the day the
fleet
picks up a 25th strategy. Sorted so heartbeats are byte-comparable.
**These do not sum to `tokens.saved`,** and the field comment says so:
strategies compose (the router routes, a strategy runs inside it) so the
same
text is measured more than once. A row means "of what this strategy was
handed,
it removed this much" — a per-strategy yield, not a share of the total.
A
strategy that saved nothing still appears; dropping it would make every
strategy look effective.
## 3. `headroom.stack`
`resource_attributes()` was called with no arguments at its one call
site, so
`headroom.stack` was absent from **all 24,040 sessions** in the corpus
while
`detect_stack` sat unused — dead code on both ends of a wire nobody
connected.
The fleet was unsegmentable by agent, which is the question the corpus
is asked
most often.
Environment detection alone is not enough. It answers `wrap_claude` only
under
`headroom wrap`; every install that points an agent at a persistent
proxy — the
common deployment — reports `proxy`, which segments nothing. The
per-request
`X-Headroom-Stack` slugs are the only signal that names the harness
there, so
`record_stack()` stages them the same way and feeds `detect_stack`'s
`by_stack`
branch:
```
9x wrap_claude, 1x wrap_cursor -> wrap_claude (dominant harness wins)
5x wrap_claude, 5x wrap_cursor -> mixed
no per-request signal -> proxy (environment fallback)
junk slug -> dropped before staging
```
## Privacy
The strategy string is slugged through the same `_safe_slug` as skip
reasons
and capped at `MAX_STRATEGIES`, because the observer protocol takes a
free
string and an extension could otherwise invent keys per request. Stack
slugs
are normalized and capped the same way.
**Deliberately not collected:** the tool names in
`smart_crush:<count>:<names>`.
Those are user-defined MCP identifiers and can name internal tooling
(`acme_deploy_prod`). They stay stripped by the existing `split(":",
1)[0]`,
and this PR does not widen it. No new key was needed in `worker.js` —
`by_strategy` nests under the already-allowlisted `compression`, and
`headroom.stack` was already in `ALLOWED_RESOURCE`.
Nothing here deletes or rewrites existing data: the Worker only ever
writes,
`sessions/` is never pruned by it, and readers merge old and new shapes
with
`union_by_name`, so pre-change heartbeats keep reading with the new
fields null.
## Verification
- `python -m headroom.telemetry.session` — self-check covers staging,
the
no-phantom-session case, drain exhaustiveness, a 0%-yield strategy
staying
visible, the cardinality cap, slug safety, and dominant/mixed/junk stack
resolution
- `node test-rollup.mjs /tmp/hr` — 3,938 real corpus objects → 1,061
sessions
in 1 object, plus the pagination, partial-read, corrupt-record,
empty-hour
and `oldestRawDay` cases
- 51 passing in `test_compression_observability`,
`test_prometheus_obs_counters`,
`test_telemetry_context`, `test_compression_strategy_outcomes`
- Consumer side exists and is checked: `beacon.sh by_strategy` in
headroom-beacon-stats reads the field end to end (sessions, installs,
invocations, tokens in/out, yield %), verified against a synthetic
parquet for
the cases that matter — aggregation across installs, a 0%-yield strategy
staying visible, pre-field sessions dropping out rather than erroring.
It is
guarded on the column, so it prints an instruction instead of a binder
error
until a release carrying this PR reaches the fleet.
- Deployed and running against the live corpus on the `5 * * * *`
trigger
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Tejas Chopra
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"name": "headroom-beacon",
|
||||
"private": true,
|
||||
"type": "module"
|
||||
}
|
||||
@@ -0,0 +1,214 @@
|
||||
/**
|
||||
* Self-check for scheduled()'s hourly compaction. node test-rollup.mjs [dir]
|
||||
*
|
||||
* The one thing that must never drift: rollupHour() and the QUALIFY in
|
||||
* headroom-beacon-stats/beacon.sh have to agree on which heartbeat wins. If
|
||||
* they disagree the reports get quietly wrong rather than loudly broken, so
|
||||
* this asserts the JS picks exactly the max-seq row per (install, session).
|
||||
*
|
||||
* Point it at a directory of real beacon objects to check against the corpus:
|
||||
* aws s3 sync s3://headroom-telemetry/sessions/dt=.../hh=.../ /tmp/hr/ ...
|
||||
* node test-rollup.mjs /tmp/hr
|
||||
* With no argument it runs on a small fixture and needs no network.
|
||||
*/
|
||||
import { readdirSync, readFileSync } from 'node:fs';
|
||||
import assert from 'node:assert/strict';
|
||||
import { oldestRawDay, rollupHour } from './worker.js';
|
||||
|
||||
// R2 returns at most 1000 keys per list page, so on a real hour (~4,000
|
||||
// objects) the cursor loop in rollupHour is load-bearing. The stub paginates at
|
||||
// a deliberately tiny size so that loop is exercised by every case below: with
|
||||
// a single-page stub, a regression that dropped the cursor would still print
|
||||
// "ok" while silently rolling up only the first page of every hour.
|
||||
const PAGE = 3;
|
||||
|
||||
/** The slice of the R2 binding rollupHour uses, backed by a plain object. */
|
||||
function stubBucket(files, { failKeys = new Set() } = {}) {
|
||||
const written = {};
|
||||
const reads = [];
|
||||
return {
|
||||
written,
|
||||
reads,
|
||||
list: async ({ prefix, cursor, delimiter }) => {
|
||||
const keys = Object.keys(files)
|
||||
.filter((k) => k.startsWith(prefix))
|
||||
.sort();
|
||||
if (delimiter) {
|
||||
const seen = new Set();
|
||||
for (const k of keys) {
|
||||
const cut = k.indexOf(delimiter, prefix.length);
|
||||
if (cut >= 0) seen.add(k.slice(0, cut + 1));
|
||||
}
|
||||
return { objects: [], delimitedPrefixes: [...seen], truncated: false };
|
||||
}
|
||||
const start = cursor ? keys.indexOf(cursor) : 0;
|
||||
const page = keys.slice(start, start + PAGE);
|
||||
const next = start + PAGE;
|
||||
return {
|
||||
objects: page.map((key) => ({ key })),
|
||||
truncated: next < keys.length,
|
||||
cursor: next < keys.length ? keys[next] : undefined,
|
||||
};
|
||||
},
|
||||
get: async (key) => {
|
||||
reads.push(key);
|
||||
if (failKeys.has(key)) throw new Error(`simulated R2 failure: ${key}`);
|
||||
if (!(key in files)) return null;
|
||||
return { text: async () => files[key] };
|
||||
},
|
||||
put: async (key, body) => {
|
||||
written[key] = body;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const beacon = (install, id, seq) =>
|
||||
JSON.stringify({ resource: { 'headroom.install_id': install }, session: { id, seq } });
|
||||
|
||||
const PART = 'dt=2026-08-06/hh=14';
|
||||
|
||||
/** Run rollupHour against a stub bucket and decode whatever it wrote. */
|
||||
async function run(files, opts = {}) {
|
||||
const CORPUS = stubBucket(files, opts);
|
||||
const spend = { read: 0 };
|
||||
let threw = null;
|
||||
let out = null;
|
||||
try {
|
||||
out = await rollupHour({ CORPUS }, PART, spend);
|
||||
} catch (err) {
|
||||
threw = err;
|
||||
}
|
||||
const body = CORPUS.written[`rollup/${PART}/data.ndjson`];
|
||||
return {
|
||||
threw,
|
||||
spend,
|
||||
wrote: out ? out.wrote : 0,
|
||||
empty: `rollup/${PART}/empty` in CORPUS.written,
|
||||
keys: Object.keys(CORPUS.written),
|
||||
rows: body ? body.split('\n').map((l) => JSON.parse(l)) : [],
|
||||
};
|
||||
}
|
||||
|
||||
// 1. Highest seq wins, out-of-order input, one row per (install, session).
|
||||
// More objects than PAGE, so the list cursor loop runs.
|
||||
{
|
||||
const files = {
|
||||
[`sessions/${PART}/a.json`]: [beacon('i1', 's1', 3), beacon('i1', 's2', 1)].join('\n'),
|
||||
[`sessions/${PART}/b.json`]: beacon('i1', 's1', 9),
|
||||
[`sessions/${PART}/c.json`]: beacon('i1', 's1', 7),
|
||||
// Same session id under a different install must not collapse together.
|
||||
[`sessions/${PART}/d.json`]: beacon('i2', 's1', 2),
|
||||
[`sessions/${PART}/e.json`]: beacon('i1', 's1', 5),
|
||||
};
|
||||
const { rows, spend, threw } = await run(files);
|
||||
assert.equal(threw, null);
|
||||
// 5 objects at PAGE=3 is two pages: proves the cursor loop, which is
|
||||
// load-bearing at the real ~4,000 objects/hour.
|
||||
assert.ok(Object.keys(files).length > PAGE, 'fixture must span pages');
|
||||
assert.equal(spend.read, 5, 'reads every object across every page');
|
||||
assert.equal(rows.length, 3, 'one row per (install, session)');
|
||||
const seq = Object.fromEntries(
|
||||
rows.map((r) => [`${r.resource['headroom.install_id']} ${r.session.id}`, r.session.seq])
|
||||
);
|
||||
assert.deepEqual(seq, { 'i1 s1': 9, 'i1 s2': 1, 'i2 s1': 2 });
|
||||
}
|
||||
|
||||
// 2. An unparseable record loses only itself. Content this Worker wrote with
|
||||
// JSON.stringify never becomes valid later, so blocking the hour on it would
|
||||
// strand the hour rather than one record.
|
||||
{
|
||||
const files = {
|
||||
[`sessions/${PART}/a.json`]: '{ this is not json',
|
||||
[`sessions/${PART}/b.json`]: `\n${beacon('i1', 's1', 4)}\n`,
|
||||
};
|
||||
const { rows, threw } = await run(files);
|
||||
assert.equal(threw, null, 'corrupt content does not abandon the hour');
|
||||
assert.deepEqual(rows.map((r) => r.session.seq), [4], 'survives a corrupt object');
|
||||
}
|
||||
|
||||
// 3. A failed get is transient, so the hour must NOT be written — a rollup is
|
||||
// built once and then trusted forever, so a short read would silently become
|
||||
// the permanent record.
|
||||
{
|
||||
const files = {
|
||||
[`sessions/${PART}/a.json`]: beacon('i1', 's1', 1),
|
||||
[`sessions/${PART}/b.json`]: beacon('i1', 's2', 1),
|
||||
};
|
||||
const { threw, keys } = await run(files, {
|
||||
failKeys: new Set([`sessions/${PART}/b.json`]),
|
||||
});
|
||||
assert.ok(threw, 'a failed get throws so the hour is retried');
|
||||
assert.deepEqual(keys, [], 'nothing written on a partial read');
|
||||
}
|
||||
|
||||
// 4. Spend is reported even when the hour throws. Charging a flat guess instead
|
||||
// lets a run that failed late overshoot the subrequest ceiling.
|
||||
{
|
||||
const files = Object.fromEntries(
|
||||
Array.from({ length: 7 }, (_, i) => [`sessions/${PART}/o${i}.json`, beacon('i1', `s${i}`, 1)])
|
||||
);
|
||||
const { threw, spend } = await run(files, {
|
||||
failKeys: new Set([`sessions/${PART}/o6.json`]),
|
||||
});
|
||||
assert.ok(threw);
|
||||
assert.equal(spend.read, 7, 'caller sees real spend, not a guess');
|
||||
}
|
||||
|
||||
// 5. An empty hour writes a marker, not a zero-byte NDJSON. Without it the hour
|
||||
// stays "missing" and is re-listed on every run forever.
|
||||
{
|
||||
const { rows, empty, keys } = await run({});
|
||||
assert.deepEqual(rows, []);
|
||||
assert.ok(empty, 'empty hour leaves a marker');
|
||||
assert.ok(
|
||||
keys.every((k) => !k.endsWith('.ndjson')),
|
||||
'no zero-byte ndjson for readers to special-case'
|
||||
);
|
||||
}
|
||||
|
||||
// 6. oldestRawDay floors the backfill. A fixed lookback window silently strands
|
||||
// every hour older than it once analysis stopped reading sessions/.
|
||||
{
|
||||
const CORPUS = stubBucket({
|
||||
'sessions/dt=2026-08-03/hh=01/a.json': beacon('i1', 's1', 1),
|
||||
'sessions/dt=2026-08-06/hh=14/b.json': beacon('i1', 's2', 1),
|
||||
'sessions/dt=2026-08-07/hh=00/c.json': beacon('i1', 's3', 1),
|
||||
});
|
||||
assert.equal(await oldestRawDay({ CORPUS }), '2026-08-03');
|
||||
assert.equal(await oldestRawDay({ CORPUS: stubBucket({}) }), null, 'empty bucket -> null');
|
||||
}
|
||||
|
||||
// 7. Against real objects, if a directory was given: same answer as the QUALIFY
|
||||
// in beacon.sh, which is `count(DISTINCT install||session)` rows, each
|
||||
// carrying that pair's max seq.
|
||||
const dir = process.argv[2];
|
||||
if (dir) {
|
||||
const files = {};
|
||||
for (const f of readdirSync(dir).filter((f) => f.endsWith('.json'))) {
|
||||
files[`sessions/${PART}/${f}`] = readFileSync(`${dir}/${f}`, 'utf8');
|
||||
}
|
||||
const { rows, spend, threw } = await run(files);
|
||||
assert.equal(threw, null);
|
||||
|
||||
const expected = new Map();
|
||||
for (const text of Object.values(files)) {
|
||||
for (const line of text.split('\n')) {
|
||||
if (!line.trim()) continue;
|
||||
const r = JSON.parse(line);
|
||||
const k = `${r.resource?.['headroom.install_id']} ${r.session?.id}`;
|
||||
expected.set(k, Math.max(expected.get(k) ?? -1, r.session?.seq ?? 0));
|
||||
}
|
||||
}
|
||||
assert.equal(spend.read, Object.keys(files).length);
|
||||
assert.equal(rows.length, expected.size, 'row count matches DISTINCT sessions');
|
||||
for (const r of rows) {
|
||||
const k = `${r.resource['headroom.install_id']} ${r.session.id}`;
|
||||
assert.equal(r.session.seq, expected.get(k), `max seq for ${k}`);
|
||||
}
|
||||
console.log(
|
||||
`real corpus: ${spend.read} objects -> ${rows.length} sessions in 1 object` +
|
||||
` (${Math.ceil(spend.read / PAGE)} list pages)`
|
||||
);
|
||||
}
|
||||
|
||||
console.log('ok');
|
||||
+183
-7
@@ -121,7 +121,184 @@ function extract(payload) {
|
||||
return records;
|
||||
}
|
||||
|
||||
// ----------------------------------------------------------------- rollup --
|
||||
//
|
||||
// The corpus is one object per heartbeat, ~1KB each — 65k on 2026-08-06 and
|
||||
// climbing. DuckDB reads them correctly, but a full `pull` is ~100k HTTPS round
|
||||
// trips for 95MB: minutes of pure per-object latency, no real bytes or compute.
|
||||
// Listing the bucket alone took 88 seconds.
|
||||
//
|
||||
// This job collapses each COMPLETE hour into one object under rollup/, keeping
|
||||
// only the highest-seq heartbeat per (install, session). One measured hour
|
||||
// (dt=2026-08-06/hh=14): 3,938 objects and 3,938 rows in, 1 object and 1,061
|
||||
// rows out. Analysis reads rollup/**, never sessions/**. Raw is left exactly as
|
||||
// written, so any rollup can be rebuilt by deleting it.
|
||||
//
|
||||
// Hourly rather than daily because every R2 binding call is a subrequest: a day
|
||||
// is ~65k of them against a 10k-per-invocation ceiling, an hour is ~4k.
|
||||
|
||||
const READ_BUDGET = 60000; // objects per run; see [limits] in wrangler.toml
|
||||
// A get costs ~45ms of round trip and almost no CPU, so this is what decides
|
||||
// whether a run finishes: at 20 an hour took ~3 minutes, against a 15-minute
|
||||
// wall clock for a cron invocation. Raise it if an hour ever stops fitting.
|
||||
const FANOUT = 100; // concurrent R2 gets
|
||||
|
||||
const partition = (d) =>
|
||||
`dt=${d.toISOString().slice(0, 10)}/hh=${d.toISOString().slice(11, 13)}`;
|
||||
|
||||
/**
|
||||
* One hour of heartbeats -> one deduped NDJSON object.
|
||||
*
|
||||
* Returns `{ read, wrote }`. Spend is reported through the mutable `spend`
|
||||
* accumulator so the caller still knows it even when this throws: the budget
|
||||
* has to track real spend, and a flat guess lets a run that failed late
|
||||
* overshoot the subrequest ceiling and get killed inside an hour that would
|
||||
* otherwise have succeeded.
|
||||
*
|
||||
* Writes nothing unless the whole hour read cleanly. A rollup is built once and
|
||||
* then treated as done forever, so a partial read would silently become the
|
||||
* permanent record — better to write nothing and let the next run retry.
|
||||
*/
|
||||
export async function rollupHour(env, part, spend = { read: 0 }) {
|
||||
const best = new Map();
|
||||
let failed = 0; // transient: retry the hour
|
||||
let corrupt = 0; // permanent: record and move on
|
||||
let cursor;
|
||||
do {
|
||||
const page = await env.CORPUS.list({ prefix: `sessions/${part}/`, cursor });
|
||||
for (let i = 0; i < page.objects.length; i += FANOUT) {
|
||||
// allSettled, not all: one transient R2 error among the ~4,000 gets in a
|
||||
// real hour would otherwise reject the batch and discard the whole hour.
|
||||
const settled = await Promise.allSettled(
|
||||
page.objects
|
||||
.slice(i, i + FANOUT)
|
||||
.map((o) => env.CORPUS.get(o.key).then((r) => (r ? r.text() : null)))
|
||||
);
|
||||
for (const outcome of settled) {
|
||||
spend.read++;
|
||||
// A miss counts as a failure too. The key came from a LIST, so the
|
||||
// object existed; treating it as empty would quietly shrink the rollup.
|
||||
if (outcome.status !== 'fulfilled' || outcome.value === null) {
|
||||
failed++;
|
||||
continue;
|
||||
}
|
||||
for (const line of outcome.value.split('\n')) {
|
||||
if (!line) continue;
|
||||
let rec;
|
||||
try {
|
||||
rec = JSON.parse(line);
|
||||
} catch {
|
||||
// Counted and logged, but NOT a reason to abandon the hour. A
|
||||
// failed get is transient and worth retrying; content this Worker
|
||||
// itself wrote with JSON.stringify does not become valid later, so
|
||||
// blocking on it would strand the hour until its raw objects
|
||||
// expire and then lose the whole hour instead of one record.
|
||||
corrupt++;
|
||||
continue;
|
||||
}
|
||||
// A session heartbeats every 5 minutes carrying CUMULATIVE totals, so
|
||||
// the highest seq IS the whole session and every earlier row is a
|
||||
// strict subset. Sessions straddle hours, so readers still dedupe
|
||||
// across rollups on this same key — this only shrinks each hour.
|
||||
const id = `${rec.resource?.['headroom.install_id']} ${rec.session?.id}`;
|
||||
const prev = best.get(id);
|
||||
if (!prev || (rec.session?.seq ?? 0) > (prev.session?.seq ?? 0)) {
|
||||
best.set(id, rec);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
cursor = page.truncated ? page.cursor : undefined;
|
||||
} while (cursor);
|
||||
|
||||
if (failed) {
|
||||
throw new Error(`${part}: ${failed} of ${spend.read} objects unreadable`);
|
||||
}
|
||||
if (corrupt) {
|
||||
console.error(`rollup ${part}: skipped ${corrupt} unparseable record(s)`);
|
||||
}
|
||||
|
||||
// A genuinely empty hour gets a marker rather than a zero-byte NDJSON that
|
||||
// every reader would have to special-case. Without it the hour stays
|
||||
// "missing" and is re-listed on every run for the life of the bucket.
|
||||
if (best.size === 0) {
|
||||
await env.CORPUS.put(`rollup/${part}/empty`, '');
|
||||
return { read: spend.read, wrote: 0 };
|
||||
}
|
||||
await env.CORPUS.put(
|
||||
`rollup/${part}/data.ndjson`,
|
||||
[...best.values()].map((r) => JSON.stringify(r)).join('\n'),
|
||||
{ httpMetadata: { contentType: 'application/x-ndjson' } }
|
||||
);
|
||||
return { read: spend.read, wrote: best.size };
|
||||
}
|
||||
|
||||
/** Oldest `dt=` day still under sessions/, or null. One delimited LIST. */
|
||||
export async function oldestRawDay(env) {
|
||||
const page = await env.CORPUS.list({ prefix: 'sessions/', delimiter: '/' });
|
||||
const days = (page.delimitedPrefixes || [])
|
||||
.map((p) => p.slice('sessions/dt='.length).replace(/\/$/, ''))
|
||||
.filter((d) => /^\d{4}-\d{2}-\d{2}$/.test(d))
|
||||
.sort();
|
||||
return days.length ? days[0] : null;
|
||||
}
|
||||
|
||||
export default {
|
||||
/** Hourly cron. Builds every complete hour back to the oldest raw data. */
|
||||
async scheduled(event, env) {
|
||||
// Backfill reaches all the way to the oldest surviving raw day, NOT a fixed
|
||||
// window. A fixed window silently strands everything older than it the
|
||||
// moment analysis stopped reading sessions/ — the raw objects are still
|
||||
// there, but nothing would ever compact them, so they vanish from every
|
||||
// report. Bounding by real data instead means the floor rises only when a
|
||||
// lifecycle rule actually expires the raw objects.
|
||||
const oldest = await oldestRawDay(env);
|
||||
if (!oldest) return;
|
||||
const floorMs = Date.parse(`${oldest}T00:00:00Z`);
|
||||
if (Number.isNaN(floorMs)) return;
|
||||
|
||||
// Only list from the floor forward. Rollups older than the oldest raw day
|
||||
// can never be rebuilt, so enumerating them answers nothing — this is what
|
||||
// keeps the listing bounded by retention rather than by total history.
|
||||
const done = new Set();
|
||||
let cursor;
|
||||
do {
|
||||
const page = await env.CORPUS.list({
|
||||
prefix: 'rollup/',
|
||||
startAfter: `rollup/dt=${oldest}`,
|
||||
cursor,
|
||||
});
|
||||
for (const o of page.objects) {
|
||||
// Tolerates both `<part>/data.ndjson` and the `<part>/empty` marker.
|
||||
const rel = o.key.slice('rollup/'.length);
|
||||
const cut = rel.lastIndexOf('/');
|
||||
if (cut > 0) done.add(rel.slice(0, cut));
|
||||
}
|
||||
cursor = page.truncated ? page.cursor : undefined;
|
||||
} while (cursor);
|
||||
|
||||
// Newest first, so a backlog drains from the present backwards and the
|
||||
// freshest hour is never the one starved by the budget. Starts one hour
|
||||
// back: the current hour is still being written to.
|
||||
let budget = READ_BUDGET;
|
||||
for (let t = event.scheduledTime - 3600_000; t >= floorMs && budget > 0; t -= 3600_000) {
|
||||
const part = partition(new Date(t));
|
||||
if (done.has(part)) continue;
|
||||
// Shared with rollupHour so a throw still reports what it spent.
|
||||
const spend = { read: 0 };
|
||||
try {
|
||||
await rollupHour(env, part, spend);
|
||||
} catch (err) {
|
||||
// Newest-first means an hour that always throws — one grown past the
|
||||
// subrequest ceiling, say — would otherwise block every older hour
|
||||
// behind it forever. Skip it and keep draining; it has no marker, so
|
||||
// the next run retries it.
|
||||
console.error(`rollup ${part} failed after ${spend.read} objects: ${err}`);
|
||||
}
|
||||
budget -= spend.read;
|
||||
}
|
||||
},
|
||||
|
||||
async fetch(request, env, ctx) {
|
||||
if (request.method !== 'POST') {
|
||||
return new Response('beacon: POST OTLP logs to /v1/logs', { status: 405 });
|
||||
@@ -145,14 +322,13 @@ export default {
|
||||
}
|
||||
if (records.length === 0) return new Response(null, { status: 204 });
|
||||
|
||||
const now = new Date();
|
||||
const day = now.toISOString().slice(0, 10);
|
||||
const hour = now.toISOString().slice(11, 13);
|
||||
// Hive-style partitioning so DuckDB can prune by date without a catalog.
|
||||
// ponytail: one object per request. At beacon volume that is a few hundred
|
||||
// thousand objects a month, which globs fine. Add a daily compaction job
|
||||
// when the file count starts to slow queries, not before.
|
||||
const key = `sessions/dt=${day}/hh=${hour}/${crypto.randomUUID()}.json`;
|
||||
// Shares partition() with the rollup: the cron lists `sessions/<part>/`, so
|
||||
// two independent spellings of this scheme would mean the writer and the
|
||||
// compactor could drift apart and silently match zero objects.
|
||||
// ponytail: one object per request. Compacted hourly into rollup/ by
|
||||
// scheduled() above — analysis reads that, never this.
|
||||
const key = `sessions/${partition(new Date())}/${crypto.randomUUID()}.json`;
|
||||
const ndjson = records.map((r) => JSON.stringify(r)).join('\n');
|
||||
|
||||
// Respond immediately; durability work continues after the response.
|
||||
|
||||
@@ -32,6 +32,24 @@ bucket_name = "headroom-telemetry"
|
||||
# npx wrangler secret put METRICS_OTLP_AUTH
|
||||
# Absent = R2 only, which is the right place to start.
|
||||
|
||||
# Hourly compaction of sessions/ into rollup/ — see scheduled() in worker.js.
|
||||
# At :05 so the hour being rolled up is definitely closed. A >=1h interval also
|
||||
# buys the 15-minute CPU limit instead of 30s, which the backfill run needs.
|
||||
[triggers]
|
||||
crons = ["5 * * * *"]
|
||||
|
||||
# Every R2 binding call is a subrequest, and one hour is already ~4k objects.
|
||||
# The paid default of 10k would cap a run at two hours and stall the backfill
|
||||
# behind live traffic forever. This only raises a ceiling; a normal run spends
|
||||
# ~4k. READ_BUDGET in worker.js is what actually bounds the work.
|
||||
#
|
||||
# Workers Paid only — on the Free plan this key is rejected outright ("CPU
|
||||
# limits are not supported for the Free plan"), and the cron could not run
|
||||
# anyway: Free gives a scheduled handler 10ms of CPU, and parsing an hour of
|
||||
# heartbeats is tens of ms.
|
||||
[limits]
|
||||
subrequests = 100000
|
||||
|
||||
[observability]
|
||||
enabled = true
|
||||
|
||||
|
||||
@@ -49,6 +49,7 @@ except ImportError:
|
||||
|
||||
from headroom.ccr.tool_injection import CCR_TOOL_NAME
|
||||
from headroom.config import is_tool_excluded
|
||||
from headroom.telemetry.session import BeaconCompressionObserver
|
||||
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -136,7 +137,11 @@ class _CrusherSingleton:
|
||||
config = SmartCrusherConfig(
|
||||
min_tokens_to_crush=self._min_tokens,
|
||||
)
|
||||
self._crusher = SmartCrusher(config=config)
|
||||
# observer: no proxy here, so nothing else reports these
|
||||
# compressions to the beacon. See BeaconCompressionObserver.
|
||||
self._crusher = SmartCrusher(
|
||||
config=config, observer=BeaconCompressionObserver()
|
||||
)
|
||||
return self._crusher
|
||||
|
||||
|
||||
|
||||
@@ -56,6 +56,7 @@ from typing import Any
|
||||
|
||||
from headroom.config import HeadroomConfig, SmartCrusherConfig
|
||||
from headroom.providers.openai import OpenAIProvider
|
||||
from headroom.telemetry.session import BeaconCompressionObserver
|
||||
from headroom.transforms.smart_crusher import SmartCrusher
|
||||
|
||||
|
||||
@@ -263,7 +264,18 @@ class HeadroomMCPCompressor:
|
||||
min_tokens_to_crush=profile.min_tokens_to_compress,
|
||||
max_items_after_crush=profile.max_items,
|
||||
)
|
||||
crusher = SmartCrusher(config=smart_config, with_compaction=False) # type: ignore[arg-type]
|
||||
# observer: MCP runs outside the proxy, so PrometheusMetrics (the
|
||||
# proxy's observer, which forwards to the beacon) never sees these
|
||||
# compressions. Without one, an MCP install reports real tokens.saved
|
||||
# with an empty compression.by_strategy.
|
||||
crusher = SmartCrusher(
|
||||
# headroom.config.SmartCrusherConfig vs the transform's own
|
||||
# same-named dataclass; the ignore has to sit on the argument line
|
||||
# because that is where mypy reports a multi-line call's arg-type.
|
||||
config=smart_config, # type: ignore[arg-type]
|
||||
with_compaction=False,
|
||||
observer=BeaconCompressionObserver(),
|
||||
)
|
||||
|
||||
# Build messages for SmartCrusher (it expects conversation format)
|
||||
messages = [
|
||||
|
||||
@@ -50,6 +50,7 @@ except ImportError:
|
||||
from headroom import HeadroomConfig
|
||||
from headroom.ccr.tool_injection import CCR_TOOL_NAME
|
||||
from headroom.config import is_tool_excluded
|
||||
from headroom.telemetry.session import BeaconCompressionObserver
|
||||
from headroom.transforms.smart_crusher import SmartCrusher, SmartCrusherConfig
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -173,7 +174,11 @@ class HeadroomHookProvider(HookProvider): # type: ignore[misc]
|
||||
crusher_config = SmartCrusherConfig(
|
||||
min_tokens_to_crush=self.min_tokens_to_compress
|
||||
)
|
||||
self._crusher = SmartCrusher(config=crusher_config)
|
||||
# observer: no proxy here, so nothing else reports these
|
||||
# compressions to the beacon. See BeaconCompressionObserver.
|
||||
self._crusher = SmartCrusher(
|
||||
config=crusher_config, observer=BeaconCompressionObserver()
|
||||
)
|
||||
logger.debug(
|
||||
"SmartCrusher initialized with min_tokens=%d", self.min_tokens_to_compress
|
||||
)
|
||||
|
||||
@@ -417,14 +417,14 @@ class PrometheusMetrics:
|
||||
return total_input_tokens, total_input_cost_usd
|
||||
|
||||
try:
|
||||
# totals() rather than stats(): identical numbers, without the
|
||||
# 31-day cost-record walk that stats()["budget_basis"] performs and
|
||||
# this caller throws away. See CostTracker.totals.
|
||||
tracked_input_tokens, tracked_input_cost_usd = self.cost_tracker.totals()
|
||||
cost_stats = self.cost_tracker.stats()
|
||||
except Exception:
|
||||
logger.debug("Failed to read cost tracker totals for savings history", exc_info=True)
|
||||
return total_input_tokens, total_input_cost_usd
|
||||
|
||||
tracked_input_tokens = cost_stats.get("total_input_tokens")
|
||||
tracked_input_cost_usd = cost_stats.get("total_input_cost_usd")
|
||||
|
||||
if tracked_input_tokens is not None:
|
||||
try:
|
||||
total_input_tokens = self._savings_tracker_input_tokens_offset + max(
|
||||
@@ -467,6 +467,14 @@ class PrometheusMetrics:
|
||||
self.requests_by_stack[slug] += 1
|
||||
self.savings_tracker.record_lifetime_stack(slug)
|
||||
|
||||
# Same fan-out as record_compression. This header is the only signal
|
||||
# that names the harness when an agent is pointed at a persistent proxy
|
||||
# rather than launched by `headroom wrap`, and the beacon cannot import
|
||||
# headroom.proxy to read requests_by_stack itself.
|
||||
from headroom.telemetry.session import record_stack as _beacon_stack
|
||||
|
||||
_beacon_stack(slug)
|
||||
|
||||
def record_compression(
|
||||
self,
|
||||
strategy: str,
|
||||
@@ -497,6 +505,22 @@ class PrometheusMetrics:
|
||||
if saved > 0:
|
||||
self.tokens_saved_by_strategy[strategy] += saved
|
||||
|
||||
# Fan out to the beacon. This object is the configured
|
||||
# CompressionObserver for the proxy's pipelines, so it is where those
|
||||
# events already arrive with both token counts — a second observer here
|
||||
# would mean a second measurement pass for numbers in hand. (The paths
|
||||
# that have no observer at all pass telemetry's
|
||||
# BeaconCompressionObserver directly instead.)
|
||||
#
|
||||
# The beacon is ON by default, so this does not short-circuit in
|
||||
# practice and must stay off the aggregator's lock: it stages into a
|
||||
# dedicated mutex that the request path never takes, which is what
|
||||
# keeps this method's "synchronous + lock-free" contract honest with
|
||||
# respect to everything else in the process.
|
||||
from headroom.telemetry.session import record_compression as _beacon_compression
|
||||
|
||||
_beacon_compression(strategy, original_tokens, compressed_tokens)
|
||||
|
||||
def record_extension_savings(self, key: str, saved: int) -> None:
|
||||
"""Accumulate tokens saved by a proxy extension, keyed by ``key``.
|
||||
|
||||
|
||||
@@ -94,6 +94,39 @@ _SLUG_RE = re.compile(r"^[a-z][a-z0-9_]{0,31}$")
|
||||
# vocabulary. Values are slug-validated before they are counted.
|
||||
_REASON_TAGS = ("passthrough_reason", "image_skip_reason", "memory_skip_reason")
|
||||
|
||||
# Cardinality cap on `by_strategy`. The real vocabulary is CompressionStrategy
|
||||
# plus a couple of literals — under a dozen — but `record_compression` takes a
|
||||
# free string, so an extension or a future caller could invent keys per request.
|
||||
# Matches the same guard on `requests_by_stack` (MAX_DISTINCT_STACKS).
|
||||
MAX_STRATEGIES = 32
|
||||
|
||||
# Compression events arrive on the compression executor thread, mid-request,
|
||||
# before that request's outcome ever reaches `SessionAggregator.record`. They
|
||||
# are staged here rather than written straight into the live session, which
|
||||
# keeps three things true at once:
|
||||
#
|
||||
# * The executor thread never takes the aggregator's lock, so compression
|
||||
# cannot serialise against the request path. The beacon is on by default,
|
||||
# and ContentRouter observes once per routing decision — once per content
|
||||
# section per request — so that contention would be real.
|
||||
# * A compression event cannot CREATE a session. Sessions are started only by
|
||||
# an outcome, which preserves the invariant that every emitted session has
|
||||
# turns >= 1; otherwise a request abandoned between compression and its
|
||||
# outcome (Claude Code users interrupt streaming routinely) would emit a
|
||||
# phantom all-zero row that inflates fleet session and install counts.
|
||||
# * The first turn's numbers still survive, because the outcome that follows
|
||||
# milliseconds later drains this into the session it opens.
|
||||
#
|
||||
# A request that dies before its outcome leaves its events staged, and they are
|
||||
# attributed to the next session instead. That is a rounding error against
|
||||
# inventing a session that never happened.
|
||||
_staged_lock = threading.Lock()
|
||||
_staged_strategies: dict[str, list[int]] = {}
|
||||
# Per-request stack slugs, for `detect_stack`'s by_stack branch. Same staging
|
||||
# and the same reason: the proxy sees the X-Headroom-Stack header per request,
|
||||
# and this is the only place the beacon can learn it without importing proxy.
|
||||
_staged_stacks: dict[str, int] = {}
|
||||
|
||||
|
||||
def _pct(numerator: float, denominator: float) -> float:
|
||||
"""Percentage to 2dp, or 0.0 when undefined.
|
||||
@@ -218,6 +251,28 @@ def resource_attributes(
|
||||
}
|
||||
if install_mode:
|
||||
attrs["headroom.install_mode"] = install_mode
|
||||
# Detect when the caller did not supply one. Every caller so far supplies
|
||||
# nothing, so `headroom.stack` was absent from the entire corpus while the
|
||||
# detector sat unused — which made the fleet unsegmentable by agent, the
|
||||
# question the corpus is most often asked ("what does this look like under
|
||||
# Claude Code?").
|
||||
#
|
||||
# The env vars detect_stack checks first are only set by `headroom wrap`.
|
||||
# The common deployment points an agent at a persistent proxy through
|
||||
# ANTHROPIC_BASE_URL and sets neither, so environment-only detection would
|
||||
# answer the literal "proxy" for almost the whole fleet — a populated,
|
||||
# authoritative-looking column that cannot answer the question it exists
|
||||
# for. The slugs staged by `record_stack` are that fleet's only real
|
||||
# signal, so they are fed to detect_stack's by_stack branch.
|
||||
if stack is None:
|
||||
try:
|
||||
from headroom.telemetry.context import detect_stack
|
||||
|
||||
with _staged_lock:
|
||||
by_stack = dict(_staged_stacks)
|
||||
stack = detect_stack({"requests": {"by_stack": by_stack}} if by_stack else None)
|
||||
except Exception: # a broken detector must not silence telemetry
|
||||
logger.debug("telemetry: stack detection failed", exc_info=True)
|
||||
if stack:
|
||||
attrs["headroom.stack"] = stack
|
||||
return attrs
|
||||
@@ -252,6 +307,10 @@ class _Session:
|
||||
overhead_ms: float = 0.0
|
||||
latency_ms: float = 0.0
|
||||
transforms: dict[str, int] = field(default_factory=dict)
|
||||
# strategy slug -> [events, tokens_in, tokens_out]. `transforms` says which
|
||||
# compressors ran; this says whether they were worth running. A list rather
|
||||
# than three parallel dicts so the three numbers cannot drift apart.
|
||||
strategies: dict[str, list[int]] = field(default_factory=dict)
|
||||
skips: dict[str, int] = field(default_factory=dict)
|
||||
sources: dict[str, int] = field(default_factory=dict)
|
||||
providers: set[str] = field(default_factory=set)
|
||||
@@ -363,6 +422,37 @@ class _Session:
|
||||
},
|
||||
"compression": {
|
||||
"transforms": dict(self.transforms),
|
||||
# Per-strategy effectiveness. `transforms` counts invocations,
|
||||
# which cannot tell a compressor that saved 60% from one that
|
||||
# ran constantly and saved nothing — the fleet's top transform
|
||||
# by count contributes an unknown share of `tokens.saved`.
|
||||
#
|
||||
# These do NOT sum to `tokens.saved`, and must not be presented
|
||||
# as if they do: strategies compose (the router routes, a
|
||||
# strategy runs inside it) so the same text is measured by more
|
||||
# than one, and tool-schema savings never appear here at all.
|
||||
# Read a row as "of what this strategy was handed, it removed
|
||||
# this much" — a per-strategy yield, not a share of the total.
|
||||
#
|
||||
# A LIST of uniform records, not a {strategy: {...}} object,
|
||||
# and that shape is deliberate. DuckDB infers a JSON object as
|
||||
# a STRUCT while its keys are few and consistent and as a MAP
|
||||
# once they are not, so an object keyed by strategy would
|
||||
# change COLUMN TYPE as the fleet adopts new compressors —
|
||||
# exactly the break that silently took out the `transforms`
|
||||
# report. A list of records has fixed field names, so the type
|
||||
# is the same on day one and after the 30th strategy ships, and
|
||||
# a new field inside a record is absorbed by union_by_name.
|
||||
# Sorted so a payload is byte-comparable between heartbeats.
|
||||
"by_strategy": [
|
||||
{
|
||||
"strategy": name,
|
||||
"n": counts[0],
|
||||
"tokens_in": counts[1],
|
||||
"tokens_out": counts[2],
|
||||
}
|
||||
for name, counts in sorted(self.strategies.items())
|
||||
],
|
||||
"overhead_ms_total": round(self.overhead_ms, 1),
|
||||
# Sum of per-request durations, NOT elapsed time: concurrent turns
|
||||
# make this exceed `session.duration_s`. Kept under the original
|
||||
@@ -505,6 +595,20 @@ def _fold(sess: _Session, outcome: Any, now: float, source: str = "proxy") -> No
|
||||
sess.last_seen = now
|
||||
sess.turns += 1
|
||||
sess.sources[source] = sess.sources.get(source, 0) + 1
|
||||
|
||||
# Compression ran on the executor thread before this outcome arrived; take
|
||||
# what it staged. Done here rather than in the observer so the executor
|
||||
# thread never touches the aggregator lock — see the note on _staged_lock.
|
||||
for name, staged in _drain_staged_strategies().items():
|
||||
counts = sess.strategies.get(name)
|
||||
if counts is None:
|
||||
if len(sess.strategies) >= MAX_STRATEGIES:
|
||||
continue
|
||||
counts = [0, 0, 0]
|
||||
sess.strategies[name] = counts
|
||||
counts[0] += staged[0]
|
||||
counts[1] += staged[1]
|
||||
counts[2] += staged[2]
|
||||
sess.original_tokens += int(get("original_tokens") or 0)
|
||||
sess.attempted_tokens += int(get("attempted_input_tokens") or 0)
|
||||
# Billed/volume figure, so prefer the provider's own count and fall back to
|
||||
@@ -776,6 +880,116 @@ def record_mcp_compression(
|
||||
)
|
||||
|
||||
|
||||
def record_compression(strategy: str, original_tokens: int, compressed_tokens: int) -> None:
|
||||
"""Beacon entry point for one compression event.
|
||||
|
||||
Signature-compatible with
|
||||
:class:`headroom.transforms.observability.CompressionObserver`, so the
|
||||
proxy's existing observer can forward here without a second measurement
|
||||
pass — the numbers are already computed on the hot path for Prometheus
|
||||
(``PrometheusMetrics.tokens_saved_by_strategy``); they just never left the
|
||||
process.
|
||||
|
||||
Same discipline as the rest of this module: off by default and cheap when
|
||||
off, never raises. This runs once per routing decision, so it must not do
|
||||
anything a request would notice.
|
||||
"""
|
||||
from headroom.telemetry.beacon import is_beacon_enabled
|
||||
|
||||
if not is_beacon_enabled():
|
||||
return
|
||||
# A slug, not the raw string. The real values are CompressionStrategy enum
|
||||
# tags, but the observer protocol takes a free string, and anything that is
|
||||
# not already a bounded lowercase identifier collapses to "other" rather
|
||||
# than reaching the wire.
|
||||
slug = _safe_slug(strategy)
|
||||
try:
|
||||
before = int(original_tokens or 0)
|
||||
after = int(compressed_tokens or 0)
|
||||
except (TypeError, ValueError):
|
||||
return
|
||||
if before <= 0:
|
||||
return
|
||||
# Clamped at the input: a compressor that emits more than it received is a
|
||||
# bug, and letting `out` exceed `in` would surface downstream as negative
|
||||
# savings rather than as the bug it is. Prometheus clamps the same way.
|
||||
after = min(max(after, 0), before)
|
||||
with _staged_lock:
|
||||
counts = _staged_strategies.get(slug)
|
||||
if counts is None:
|
||||
if len(_staged_strategies) >= MAX_STRATEGIES:
|
||||
return
|
||||
counts = [0, 0, 0]
|
||||
_staged_strategies[slug] = counts
|
||||
counts[0] += 1
|
||||
counts[1] += before
|
||||
counts[2] += after
|
||||
|
||||
|
||||
def record_stack(slug: str) -> None:
|
||||
"""Beacon entry point for one request's stack slug.
|
||||
|
||||
The harness identity lives in the ``X-Headroom-Stack`` header, which only
|
||||
the proxy sees, and per request rather than per process. Counting slugs
|
||||
here lets :func:`resource_attributes` answer ``detect_stack``'s by_stack
|
||||
branch without the telemetry package importing ``headroom.proxy``.
|
||||
|
||||
Without this the beacon can only read the two environment variables, so
|
||||
every install that points an agent at a persistent proxy — the common
|
||||
deployment for Claude Code, Cursor, Codex and the adapters — reports the
|
||||
literal ``"proxy"`` and the fleet is unsegmentable by agent.
|
||||
"""
|
||||
from headroom.telemetry.beacon import is_beacon_enabled
|
||||
|
||||
if not is_beacon_enabled():
|
||||
return
|
||||
# normalize_stack is the same chokepoint the proxy applies at ingress; an
|
||||
# unbounded header value must not reach the wire or grow this dict.
|
||||
from headroom.telemetry.context import normalize_stack
|
||||
|
||||
clean = normalize_stack(slug)
|
||||
if not clean:
|
||||
return
|
||||
with _staged_lock:
|
||||
if clean not in _staged_stacks and len(_staged_stacks) >= MAX_STRATEGIES:
|
||||
return
|
||||
_staged_stacks[clean] = _staged_stacks.get(clean, 0) + 1
|
||||
|
||||
|
||||
class BeaconCompressionObserver:
|
||||
"""A `CompressionObserver` that forwards to the beacon and nothing else.
|
||||
|
||||
The proxy's `PrometheusMetrics` is already an observer and forwards from
|
||||
there, so this is for the paths that never had one: the MCP servers, the
|
||||
bare transform pipeline, and the LangChain/Strands integrations. Those
|
||||
processes report `tokens.saved` either way, so without this they emit
|
||||
sessions with real token totals and an empty `by_strategy` — a silently
|
||||
biased subset that cannot be reconciled with the fleet totals.
|
||||
|
||||
Only `record_compression` is implemented. ContentRouter's two other
|
||||
observer hooks (`record_kompress_size_gate`, `record_router_route_counts`)
|
||||
are each individually guarded at the call site, and both feed `/stats`
|
||||
rather than the beacon.
|
||||
"""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def record_compression(
|
||||
self, strategy: str, original_tokens: int, compressed_tokens: int
|
||||
) -> None:
|
||||
record_compression(strategy, original_tokens, compressed_tokens)
|
||||
|
||||
|
||||
def _drain_staged_strategies() -> dict[str, list[int]]:
|
||||
"""Take everything staged since the last drain. Caller merges it."""
|
||||
with _staged_lock:
|
||||
if not _staged_strategies:
|
||||
return {}
|
||||
drained = {name: counts[:] for name, counts in _staged_strategies.items()}
|
||||
_staged_strategies.clear()
|
||||
return drained
|
||||
|
||||
|
||||
def record_outcome(outcome: Any) -> None:
|
||||
"""Beacon entry point, called from the proxy's outcome funnel.
|
||||
|
||||
@@ -874,6 +1088,97 @@ def demo() -> None:
|
||||
agg.flush_all()
|
||||
assert len(emitted) == 2, emitted
|
||||
assert emitted[1]["session"]["turns"] == 1
|
||||
|
||||
# --- per-strategy compression -----------------------------------------
|
||||
# Compression runs on the executor thread before its request's outcome
|
||||
# arrives, so events are staged and drained by the next outcome. That is
|
||||
# what keeps the first turn's numbers while letting only an outcome open a
|
||||
# session. `_staged_*` is module state, so clear it between cases.
|
||||
_staged_strategies.clear()
|
||||
_staged_stacks.clear()
|
||||
|
||||
strat: list[dict[str, Any]] = []
|
||||
sa = SessionAggregator(strat.append, idle_s=10.0)
|
||||
record_compression("smart_crusher", 1000, 400)
|
||||
record_compression("smart_crusher", 500, 300)
|
||||
record_compression("code_aware", 800, 800)
|
||||
assert sa._current is None, "a compression event must not open a session"
|
||||
sa.record(FakeOutcome(), now=2000.0)
|
||||
sa.flush_all()
|
||||
by = {row["strategy"]: row for row in strat[-1]["compression"]["by_strategy"]}
|
||||
assert by["smart_crusher"] == {
|
||||
"strategy": "smart_crusher",
|
||||
"n": 2,
|
||||
"tokens_in": 1500,
|
||||
"tokens_out": 700,
|
||||
}, by
|
||||
# A strategy that ran and saved nothing must still appear: "ran 800 tokens
|
||||
# through and removed none" is the finding, and dropping it would make
|
||||
# every strategy look effective.
|
||||
assert by["code_aware"]["tokens_in"] == by["code_aware"]["tokens_out"] == 800, by
|
||||
assert strat[-1]["session"]["turns"] == 1, "compression events are not turns"
|
||||
# A list of records, not an object keyed by strategy: the type must not
|
||||
# change as strategies are added. See the note in payload().
|
||||
assert isinstance(strat[-1]["compression"]["by_strategy"], list)
|
||||
assert [r["strategy"] for r in strat[-1]["compression"]["by_strategy"]] == sorted(
|
||||
r["strategy"] for r in strat[-1]["compression"]["by_strategy"]
|
||||
), "sorted so heartbeats are byte-comparable"
|
||||
|
||||
# Draining is exhaustive: a second session must not re-count the first
|
||||
# session's events.
|
||||
assert not _staged_strategies, "record() drains everything it staged"
|
||||
again: list[dict[str, Any]] = []
|
||||
sb = SessionAggregator(again.append, idle_s=10.0)
|
||||
sb.record(FakeOutcome(), now=3000.0)
|
||||
sb.flush_all()
|
||||
assert again[-1]["compression"]["by_strategy"] == [], again[-1]
|
||||
|
||||
# An abandoned request — compression ran, the outcome never arrived — must
|
||||
# not invent a session. Before staging, this emitted a phantom turns=0 row
|
||||
# with all-zero tokens that inflated fleet session and install counts.
|
||||
ghost: list[dict[str, Any]] = []
|
||||
sc = SessionAggregator(ghost.append, idle_s=10.0)
|
||||
record_compression("smart_crusher", 900, 100)
|
||||
sc.flush_all()
|
||||
assert ghost == [], "no outcome, no session"
|
||||
_staged_strategies.clear()
|
||||
|
||||
# Over the cardinality cap, extra strategies are dropped rather than
|
||||
# allowed to grow the payload without bound.
|
||||
cap: list[dict[str, Any]] = []
|
||||
cc = SessionAggregator(cap.append, idle_s=10.0)
|
||||
for i in range(MAX_STRATEGIES + 5):
|
||||
record_compression(f"s{i}", 100, 50)
|
||||
cc.record(FakeOutcome(), now=4000.0)
|
||||
cc.flush_all()
|
||||
assert len(cap[-1]["compression"]["by_strategy"]) == MAX_STRATEGIES, cap[-1]
|
||||
_staged_strategies.clear()
|
||||
|
||||
# Strategy names are slugged, never passed through: the observer protocol
|
||||
# takes a free string and this is the only chokepoint before the wire.
|
||||
assert _safe_slug("smart_crusher") == "smart_crusher"
|
||||
assert _safe_slug("../../etc/passwd") == "other"
|
||||
|
||||
# --- stack detection ---------------------------------------------------
|
||||
# Environment-only detection answers "proxy" for every install that points
|
||||
# an agent at a persistent proxy instead of using `headroom wrap` — i.e.
|
||||
# most of the fleet. The per-request slugs are the only real signal.
|
||||
_staged_stacks.clear()
|
||||
for _ in range(9):
|
||||
record_stack("wrap_claude")
|
||||
record_stack("wrap_cursor")
|
||||
assert resource_attributes()["headroom.stack"] == "wrap_claude", "dominant stack wins"
|
||||
_staged_stacks.clear()
|
||||
for _ in range(5):
|
||||
record_stack("wrap_claude")
|
||||
for _ in range(5):
|
||||
record_stack("wrap_cursor")
|
||||
assert resource_attributes()["headroom.stack"] == "mixed", "no dominant stack"
|
||||
_staged_stacks.clear()
|
||||
record_stack("../../etc/passwd")
|
||||
assert not _staged_stacks, "junk slugs never reach the wire"
|
||||
assert resource_attributes()["headroom.stack"] == "proxy", "falls back with no signal"
|
||||
|
||||
assert emitted[1]["session"]["id"] != emitted[0]["session"]["id"]
|
||||
assert emitted[1]["session"]["ended"] == "shutdown"
|
||||
|
||||
|
||||
@@ -163,7 +163,13 @@ class TransformPipeline:
|
||||
# - Logs -> LogCompressor
|
||||
# - Search results -> SearchCompressor
|
||||
# - HTML -> HTMLExtractor
|
||||
transforms.append(ContentRouter())
|
||||
# observer: the proxy passes PrometheusMetrics; this bare pipeline is
|
||||
# used by the library/adapter paths, which would otherwise report
|
||||
# tokens.saved with an empty by_strategy. Imported here rather than at
|
||||
# module scope — transforms sits below telemetry in the import graph.
|
||||
from headroom.telemetry.session import BeaconCompressionObserver
|
||||
|
||||
transforms.append(ContentRouter(observer=BeaconCompressionObserver()))
|
||||
logger.info("Pipeline using ContentRouter for intelligent content-aware compression")
|
||||
|
||||
return transforms
|
||||
|
||||
Reference in New Issue
Block a user