Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pm-stack/aggregator/src/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ export interface Config {
/** The provider quota registry; the hub's second dated-read source
* (billing incidents) for the same fallback path. */
quotaRegistryFile: string;
/** The quota gatherer's latest-state file (#1151), written into the
* aggregator's own data dir by scripts/gather-quota.ts. */
quotaCurrentFile: string;
/** OpenChamber managed-opencode state file (laptop attention source). */
openchamberStateFile: string;
kaneo: {
Expand Down Expand Up @@ -67,6 +70,7 @@ export function loadConfig(env: Record<string, string | undefined> = process.env
env.PM_AGG_QUOTA_REGISTRY,
join(homedir(), "workspace/goals/keep-the-four-repos-healthy/hidden_files/provider-quota-registry.md"),
),
quotaCurrentFile: s(env.PM_AGG_QUOTA_CURRENT_FILE, join(process.cwd(), "quota-current.json")),
openchamberStateFile: s(
env.PM_AGG_OPENCHAMBER_STATE,
join(homedir(), ".config", "openchamber", "managed-opencode", "state.json"),
Expand Down
42 changes: 41 additions & 1 deletion pm-stack/aggregator/src/model.ts
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,10 @@ export type ProvenanceDetail = Record<keyof Provenance, ProvenanceFacet>;
// (FSL-1.1-MIT) is vendored — this is the data model only.
// ---------------------------------------------------------------------------

export type QuotaState = "healthy" | "near" | "walled";
/** "error"/"stale" are gatherer-overlay states (#1151): the gatherer
* could not read the pool, or its read is past the freshness bound.
* They are never rendered as healthy, and never as zero usage. */
export type QuotaState = "healthy" | "near" | "walled" | "error" | "stale";

/** One usage window a provider exposes (5-hour, weekly, monthly, ...). */
export interface QuotaWindow {
Expand Down Expand Up @@ -223,6 +226,40 @@ export interface QuotaPool {
burnTokensPerMin?: number;
/** Sessions currently burning this pool, with cumulative tokens. */
activeSessions?: { session: string; model: string; tokens: number }[];
/** Gatherer freshness for this pool (#1151), set when
* quota-current.json carried an entry for it. */
freshness?: PoolFreshness;
}

/** Per-pool freshness from the quota gatherer (#1151). */
export interface PoolFreshness {
measuredAt: string;
source: string;
state: "ok" | "stale" | "error";
/** Why the gatherer could not read the pool, when state is "error". */
error?: string;
}

/** One provider's freshness in the gatherer summary. */
export interface GathererProviderFreshness {
providerId: string;
serviceId: string;
credentialLabel?: string;
state: "ok" | "stale" | "error";
measuredAt: string;
source: string;
error?: string;
}

/** The gatherer's presence in the quota payload (#1151): tick time,
* health, and every provider it read — including providers with no
* pool on the page (OpenCode Go workspaces, Antigravity models). */
export interface GathererSummary {
generatedAt: string;
source: string;
tickAt: string | null;
ok: boolean;
providers: GathererProviderFreshness[];
}

/** Historical burn between resets, so headroom is projected, not observed. */
Expand All @@ -236,6 +273,9 @@ export interface BurnSegment {

export interface QuotaView {
pools: QuotaPool[];
/** Gatherer overlay provenance (#1151); absent when no
* quota-current.json was read. Additive for existing consumers. */
gatherer?: GathererSummary;
burn: BurnSegment[];
/** Guidance for choosing the next pool (burn order + reset rules). */
choices: {
Expand Down
165 changes: 165 additions & 0 deletions pm-stack/aggregator/src/quota-current.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
// The quota gatherer's quota-current.json, applied to the quota view
// (#1151, final visibility hop). The gatherer (scripts/gather-quota.ts)
// writes the latest measured state per provider into the aggregator's
// own data dir every 15 minutes; this module is the reader side: pure
// functions over the parsed file, so the precedence semantics are
// pinned by tests and the live source only does the file IO.
//
// Semantics (task #1151):
// - An entry whose state is "ok" and whose measuredAt is fresh
// (<= FRESH_MS old) takes precedence for the pool whose id equals
// the entry's serviceId: its percent windows replace the pool's
// dated-read windows, and the pool state is recomputed from them.
// - An entry that is an error, or "ok" but past the freshness bound,
// surfaces as the pool's state (error/stale, with the reason) —
// never as zeros, and the pool's older windows are never silently
// presented as current: the freshness stamp rides on the pool.
// - Pools with no entry keep their existing values untouched.
// - Several entries can share a serviceId (one per credential): a
// fresh ok entry wins over an error entry; otherwise the newest
// entry decides the surfaced state.
import { poolState } from "./derive.ts";
import { computeChoices } from "./quota-guidance.ts";
import type {
GathererSummary,
PoolFreshness,
QuotaCurrentEntry,
QuotaCurrentFile,
QuotaPool,
QuotaView,
QuotaWindow,
} from "./model.ts";

/** Freshness bound for gatherer precedence: two missed ticks. */
export const QUOTA_CURRENT_FRESH_MS = 20 * 60_000;

const isRecord = (v: unknown): v is Record<string, unknown> => typeof v === "object" && v !== null && !Array.isArray(v);

/** Structural check of a parsed quota-current.json document. */
export function parseQuotaCurrent(doc: unknown): QuotaCurrentFile | null {
if (!isRecord(doc)) return null;
if (doc.version !== 1 || typeof doc.generatedAt !== "string") return null;
if (!Array.isArray(doc.providers)) return null;
for (const p of doc.providers) {
if (!isRecord(p)) return null;
if (typeof p.providerId !== "string" || typeof p.serviceId !== "string") return null;
if (p.state !== "ok" && p.state !== "error" && p.state !== "unconfigured") return null;
if (typeof p.measuredAt !== "string" || typeof p.source !== "string") return null;
if (!Array.isArray(p.windows)) return null;
}
return doc as unknown as QuotaCurrentFile;
}

type FreshState = PoolFreshness["state"];

function entryState(entry: QuotaCurrentEntry, nowMs: number): FreshState {
if (entry.state === "ok") {
const t = Date.parse(entry.measuredAt);
return Number.isFinite(t) && nowMs - t <= QUOTA_CURRENT_FRESH_MS ? "ok" : "stale";
}
return "error";
}

function errorText(entry: QuotaCurrentEntry): string {
if (entry.error) return entry.error;
return entry.state === "unconfigured" ? "not configured on the gatherer host" : "gatherer could not read this pool";
}

/** Gatherer window names onto the pool card's naming. */
function windowName(name: string): string {
return name === "5h" ? "5-hour" : name;
}

function pctWindows(entry: QuotaCurrentEntry): QuotaWindow[] {
return entry.windows
.filter((w) => typeof w.pctOfLimit === "number" && Number.isFinite(w.pctOfLimit))
.map((w) => ({
name: windowName(w.name),
unit: "other" as const,
limit: 100,
used: w.pctOfLimit as number,
percent: w.pctOfLimit as number,
resetsAt: w.resetsAt ?? null,
}));
}

function applyEntry(pool: QuotaPool, entry: QuotaCurrentEntry, state: FreshState): boolean {
// Returns true when the pool's windows were replaced by a fresh read.
const freshness: PoolFreshness = {
measuredAt: entry.measuredAt,
source: entry.source,
state,
};
if (state === "error") freshness.error = errorText(entry);
pool.freshness = freshness;
if (state === "error") {
pool.state = "error";
return false;
}
if (state === "stale") {
pool.state = "stale";
return false;
}
const windows = pctWindows(entry);
if (windows.length === 0) return false; // balance-only read: freshness, no windows to show
pool.windows = windows;
pool.state = poolState(pool);
delete pool.wallLiftsAt;
return true;
}

function summarize(file: QuotaCurrentFile, nowMs: number): GathererSummary {
return {
generatedAt: file.generatedAt,
source: file.source,
tickAt: file.heartbeat?.tickAt ?? null,
ok: file.heartbeat?.ok ?? true,
providers: file.providers.map((e) => {
const s: GathererSummary["providers"][number] = {
providerId: e.providerId,
serviceId: e.serviceId,
state: entryState(e, nowMs),
measuredAt: e.measuredAt,
source: e.source,
};
if (e.credentialLabel) s.credentialLabel = e.credentialLabel;
if (s.state === "error") s.error = errorText(e);
return s;
}),
};
}

/**
* Overlay a parsed quota-current document onto a quota view. Pure:
* the input view is not mutated; pools are copied when changed.
*/
export function overlayQuotaCurrent(view: QuotaView, file: QuotaCurrentFile, now = new Date()): QuotaView {
const nowMs = now.getTime();
const byService = new Map<string, QuotaCurrentEntry[]>();
for (const e of file.providers) {
const list = byService.get(e.serviceId);
if (list) list.push(e);
else byService.set(e.serviceId, [e]);
}

let anyFreshWindows = false;
const pools = view.pools.map((orig) => {
const entries = byService.get(orig.id);
if (!entries || entries.length === 0) return orig;
const pool: QuotaPool = { ...orig, windows: [...orig.windows] };
const freshOk = entries
.filter((e) => entryState(e, nowMs) === "ok")
.sort((a, b) => Date.parse(b.measuredAt) - Date.parse(a.measuredAt));
if (freshOk.length > 0) {
if (applyEntry(pool, freshOk[0], "ok")) anyFreshWindows = true;
return pool;
}
const newest = [...entries].sort((a, b) => Date.parse(b.measuredAt) - Date.parse(a.measuredAt))[0];
applyEntry(pool, newest, entryState(newest, nowMs));
return pool;
});

const out: QuotaView = { ...view, pools, gatherer: summarize(file, nowMs) };
if (anyFreshWindows) out.choices = computeChoices(pools, nowMs);
return out;
}
5 changes: 4 additions & 1 deletion pm-stack/aggregator/src/render.ts
Original file line number Diff line number Diff line change
Expand Up @@ -758,7 +758,10 @@ export function renderQuota(
`<div class="muted">burning: ${esc(s.session)} (${esc(s.model)}) ${s.tokens.toLocaleString()} tok</div>`,
)
.join("");
return `<div class="card pool"><div><strong>${esc(p.provider)}</strong><br><span class="muted">${esc(p.plan)}</span><br><span class="badge ${esc(p.state)}">${esc(p.state)}${p.wallLiftsAt ? ` · lifts ${fmtDate(p.wallLiftsAt)}` : ""}</span>${p.burnTokensPerMin ? ` <span class="muted">${p.burnTokensPerMin} tok/min</span>` : ""}</div><div>${windows}${sessions}${p.note ? `<div class="muted">${esc(p.note)}</div>` : ""}</div></div>`;
const fresh = p.freshness
? `<div class="muted">gatherer · ${esc(p.freshness.state)} · measured ${fmtDate(p.freshness.measuredAt)} · ${esc(p.freshness.source)}${p.freshness.error ? ` · ${esc(p.freshness.error)}` : ""}</div>`
: "";
return `<div class="card pool"><div><strong>${esc(p.provider)}</strong><br><span class="muted">${esc(p.plan)}</span><br><span class="badge ${esc(p.state)}">${esc(p.state)}${p.wallLiftsAt ? ` · lifts ${fmtDate(p.wallLiftsAt)}` : ""}</span>${p.burnTokensPerMin ? ` <span class="muted">${p.burnTokensPerMin} tok/min</span>` : ""}</div><div>${windows}${sessions}${fresh}${p.note ? `<div class="muted">${esc(p.note)}</div>` : ""}</div></div>`;
})
.join("");
const burn = view.burn
Expand Down
25 changes: 22 additions & 3 deletions pm-stack/aggregator/src/sources/live.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import { buildAttentionFlags } from "../derive.ts";
import { summarizeDns } from "../dns.ts";
import { extractCharges, extractPaygLines, extractSaasServices, extractSubscriptions } from "../saas-extract.ts";
import { isProjectPoolItem, poolItemToGhItem } from "../projects.ts";
import { overlayQuotaCurrent, parseQuotaCurrent } from "../quota-current.ts";
import { SnapshotStore } from "../snapshots.ts";
import { fixtureSource } from "./fixtures.ts";
import { mapServerSessions, type MappedSessions } from "./opencode.ts";
Expand Down Expand Up @@ -716,13 +717,31 @@ export function liveSource(cfg: Config): Source {
// on the sandbox by scripts/collect-fleet-feed.ts. A stale feed
// keeps rendering, with its age in provenance; fixtures only when
// no feed exists at all.
// Stage 2 (#1151): the quota gatherer's quota-current.json —
// written into this aggregator's own data dir on a 15-minute
// tick — overlays the dated-read pools: fresh reads win, error /
// stale reads surface as the pool's state. Fail-soft: a missing
// or malformed file leaves the base view untouched.
const feed = await getFeed();
let view: QuotaView;
if (feed) {
provenance.quota = `live (fleet feed · ${feedAgeText(feed.generatedAt)} · collected ${feed.generatedAt})`;
return feed.quota;
view = feed.quota;
} else {
provenance.quota = "fixtures (no fleet feed file yet)";
view = await fix.quota();
}
provenance.quota = "fixtures (no fleet feed file yet)";
return fix.quota();
try {
const raw = await readFile(cfg.quotaCurrentFile, "utf8");
const file = parseQuotaCurrent(JSON.parse(raw));
if (file) {
provenance.quota += ` · gatherer ${file.generatedAt}`;
return overlayQuotaCurrent(view, file);
}
} catch {
/* no gatherer file (or unreadable): base view stands */
}
return view;
},
async usage(): Promise<UsageView> {
// Usage & billing hub (design §3): feed sections first (the
Expand Down
Loading
Loading