checkpoint

⚒️ Generated with [Fabro](https://fabro.sh)
This commit is contained in:
Fabro 2026-05-21 19:10:15 -04:00
parent 4a3b0b9330
commit d62ab8f185
6 changed files with 703 additions and 20 deletions

247
run.json

File diff suppressed because one or more lines are too long

View file

@ -0,0 +1,279 @@
diff --git a/lib/crates/fabro-model/src/billing.rs b/lib/crates/fabro-model/src/billing.rs
index 29cad2335..f06cad06a 100644
--- a/lib/crates/fabro-model/src/billing.rs
+++ b/lib/crates/fabro-model/src/billing.rs
@@ -358,6 +358,19 @@ impl BilledTokenCounts {
}
}
+ /// Returns the five disjoint per-call token buckets, dropping the derived
+ /// `total_tokens` sum and the optional `total_usd_micros` cost.
+ #[must_use]
+ pub fn token_counts(&self) -> TokenCounts {
+ TokenCounts {
+ input_tokens: self.input_tokens,
+ output_tokens: self.output_tokens,
+ reasoning_tokens: self.reasoning_tokens,
+ cache_read_tokens: self.cache_read_tokens,
+ cache_write_tokens: self.cache_write_tokens,
+ }
+ }
+
pub fn add_counts(&mut self, source: &Self) {
self.input_tokens += source.input_tokens;
self.output_tokens += source.output_tokens;
@@ -435,6 +448,27 @@ impl Catalog {
self.provider(&model_ref.provider)
.and_then(|provider| ModelBillingFacts::for_policy(provider.billing_policy, tokens))
}
+
+ /// Price a partial token sample for `model` using catalog pricing.
+ ///
+ /// Returns `None` when the provider has no billing policy, the model is
+ /// unknown, or the pricing algorithm cannot produce a result for the given
+ /// tokens. Used by read-side rollups so in-flight stages can show an
+ /// exact cost for the tokens consumed so far.
+ #[must_use]
+ pub fn price_tokens(&self, model: &ModelRef, tokens: &TokenCounts) -> Option<i64> {
+ let facts = self.billing_facts_for(model, tokens)?;
+ let input = ModelBillingInput {
+ usage: ModelUsage {
+ model: model.clone(),
+ tokens: tokens.clone(),
+ },
+ facts,
+ };
+ self.pricing_for(model)
+ .and_then(|pricing| pricing.bill(&input))
+ .map(|amount| amount.0)
+ }
}
fn costs_for_speed(
diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs
index b5258fa65..731c3f926 100644
--- a/lib/crates/fabro-server/src/server.rs
+++ b/lib/crates/fabro-server/src/server.rs
@@ -3238,7 +3238,7 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
.expect("aggregate_billing lock poisoned");
accumulate_billing_rollup(
&mut agg,
- &fabro_workflow::billing_rollup_from_projection(projection),
+ &fabro_workflow::billing_rollup_from_projection(projection, None),
);
}
}
@@ -3523,7 +3523,7 @@ async fn execute_run_subprocess(state: Arc<AppState>, run_id: RunId) {
.expect("aggregate_billing lock poisoned");
accumulate_billing_rollup(
&mut agg,
- &fabro_workflow::billing_rollup_from_projection(&final_state),
+ &fabro_workflow::billing_rollup_from_projection(&final_state, None),
);
}
diff --git a/lib/crates/fabro-server/src/server/handler/billing.rs b/lib/crates/fabro-server/src/server/handler/billing.rs
index 6eee9bbfb..cbaf7cc3e 100644
--- a/lib/crates/fabro-server/src/server/handler/billing.rs
+++ b/lib/crates/fabro-server/src/server/handler/billing.rs
@@ -79,7 +79,8 @@ async fn get_run_billing(
};
let projection = cached.projection;
- let rollup = fabro_workflow::billing_rollup_from_projection(&projection);
+ let catalog = state.catalog();
+ let rollup = fabro_workflow::billing_rollup_from_projection(&projection, Some(&catalog));
let by_model = rollup
.by_model
.iter()
diff --git a/lib/crates/fabro-workflow/src/billing_rollup.rs b/lib/crates/fabro-workflow/src/billing_rollup.rs
index a91b06061..89791caa7 100644
--- a/lib/crates/fabro-workflow/src/billing_rollup.rs
+++ b/lib/crates/fabro-workflow/src/billing_rollup.rs
@@ -1,6 +1,7 @@
use std::collections::HashMap;
-use fabro_types::{BilledTokenCounts, ModelRef, RunProjection};
+use fabro_model::Catalog;
+use fabro_types::{BilledTokenCounts, ModelRef, RunProjection, StageProjection};
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectionBillingStage {
@@ -34,7 +35,10 @@ impl ProjectionBillingRollup {
}
#[must_use]
-pub fn billing_rollup_from_projection(projection: &RunProjection) -> ProjectionBillingRollup {
+pub fn billing_rollup_from_projection(
+ projection: &RunProjection,
+ catalog: Option<&Catalog>,
+) -> ProjectionBillingRollup {
let mut stage_indices = HashMap::<String, usize>::new();
let mut stages = Vec::<ProjectionBillingStage>::new();
let mut by_model = HashMap::<ModelRef, ProjectionBillingByModel>::new();
@@ -46,7 +50,8 @@ pub fn billing_rollup_from_projection(projection: &RunProjection) -> ProjectionB
if is_boundary_stage(projection, stage_id.node_id()) {
continue;
}
- if stage.completion.is_none() && stage.duration_ms.is_none() && stage.usage.is_zero() {
+ let priced = stage_usage_with_cost(catalog, stage);
+ if stage.completion.is_none() && stage.duration_ms.is_none() && priced.is_zero() {
continue;
}
@@ -68,10 +73,10 @@ pub fn billing_rollup_from_projection(projection: &RunProjection) -> ProjectionB
runtime_ms = runtime_ms.saturating_add(duration_ms);
}
- if !stage.usage.is_zero() {
+ if !priced.is_zero() {
billed_visit_count += 1;
- row.billing.add_counts(&stage.usage);
- totals.add_counts(&stage.usage);
+ row.billing.add_counts(&priced);
+ totals.add_counts(&priced);
if let Some(model) = &stage.model {
row.model = Some(model.clone());
@@ -84,7 +89,7 @@ pub fn billing_rollup_from_projection(projection: &RunProjection) -> ProjectionB
billing: BilledTokenCounts::default(),
});
model_entry.stages += 1;
- model_entry.billing.add_counts(&stage.usage);
+ model_entry.billing.add_counts(&priced);
}
}
}
@@ -113,6 +118,16 @@ pub fn billing_rollup_from_projection(projection: &RunProjection) -> ProjectionB
}
}
+fn stage_usage_with_cost(catalog: Option<&Catalog>, stage: &StageProjection) -> BilledTokenCounts {
+ let mut usage = stage.usage.clone();
+ if usage.total_usd_micros.is_none() {
+ if let (Some(catalog), Some(model)) = (catalog, stage.model.as_ref()) {
+ usage.total_usd_micros = catalog.price_tokens(model, &usage.token_counts());
+ }
+ }
+ usage
+}
+
fn is_boundary_stage(projection: &RunProjection, node_id: &str) -> bool {
projection
.spec()
@@ -126,6 +141,7 @@ fn is_boundary_stage(projection: &RunProjection, node_id: &str) -> bool {
mod tests {
use std::collections::HashMap;
+ use fabro_model::{Catalog, ModelRef, ProviderId};
use fabro_types::{
AttrValue, BilledModelUsage, BilledTokenCounts, Graph, Node, RunProjection, RunSpec,
StageCompletion, StageOutcome, WorkflowSettings, first_event_seq, fixtures,
@@ -190,7 +206,7 @@ mod tests {
timestamp: chrono::Utc::now(),
});
- let rollup = billing_rollup_from_projection(&projection);
+ let rollup = billing_rollup_from_projection(&projection, None);
assert_eq!(rollup.stages.len(), 1);
assert_eq!(rollup.stages[0].node_id, "verify");
@@ -233,7 +249,7 @@ mod tests {
timestamp: chrono::Utc::now(),
});
- let rollup = billing_rollup_from_projection(&projection);
+ let rollup = billing_rollup_from_projection(&projection, None);
assert_eq!(rollup.stages.len(), 1);
assert_eq!(rollup.stages[0].node_id, "build");
@@ -266,12 +282,48 @@ mod tests {
timestamp: chrono::Utc::now(),
});
- let rollup = billing_rollup_from_projection(&projection);
+ let rollup = billing_rollup_from_projection(&projection, None);
assert_eq!(rollup.stages.len(), 0);
assert_eq!(rollup.runtime_ms, 0);
}
+ #[test]
+ fn rollup_prices_in_flight_stage_usage_using_catalog() {
+ let mut projection = test_projection();
+ let model = ModelRef {
+ provider: ProviderId::openai(),
+ model_id: "gpt-5.4".to_string(),
+ speed: None,
+ };
+ let stage = projection.stage_entry("agent", 1, first_event_seq(1));
+ stage.started_at = Some(chrono::Utc::now());
+ stage.usage = BilledTokenCounts {
+ input_tokens: 500_000,
+ output_tokens: 125_000,
+ total_tokens: 625_000,
+ ..BilledTokenCounts::default()
+ };
+ stage.model = Some(model.clone());
+
+ let priced = billing_rollup_from_projection(&projection, Some(Catalog::builtin()));
+ let unpriced = billing_rollup_from_projection(&projection, None);
+
+ assert_eq!(priced.stages.len(), 1);
+ assert_eq!(priced.stages[0].node_id, "agent");
+ let stage_cost = priced.stages[0].billing.total_usd_micros;
+ assert!(
+ stage_cost.is_some_and(|cost| cost > 0),
+ "expected priced stage cost, got {stage_cost:?}"
+ );
+ assert_eq!(priced.totals.total_usd_micros, stage_cost);
+ assert_eq!(priced.by_model.len(), 1);
+ assert_eq!(priced.by_model[0].billing.total_usd_micros, stage_cost);
+ assert_eq!(unpriced.stages.len(), 1);
+ assert_eq!(unpriced.stages[0].billing.total_usd_micros, None);
+ assert_eq!(unpriced.totals.total_usd_micros, None);
+ }
+
fn run_spec_with_boundary_nodes() -> RunSpec {
let mut graph = Graph::new("test");
graph.nodes.insert("start".to_string(), {
diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
index 5345bfb4a..a7f4619db 100644
--- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs
@@ -82,7 +82,7 @@ pub(crate) async fn build_conclusion_from_store(
.unwrap_or_default();
let projection_billing = projection
.as_ref()
- .map(billing_rollup_from_projection)
+ .map(|projection| billing_rollup_from_projection(projection, None))
.unwrap_or_default();
let checkpoint = projection
.as_ref()
@@ -438,7 +438,7 @@ async fn compute_final_patch(
}
pub(crate) fn billing_from_projection(projection: &RunProjection) -> Option<BilledTokenCounts> {
- billing_rollup_from_projection(projection).billing_if_present()
+ billing_rollup_from_projection(projection, None).billing_if_present()
}
pub(crate) fn build_terminal_event(
@@ -554,7 +554,7 @@ pub async fn finalize(executed: Executed, options: &FinalizeOptions) -> Result<C
.unwrap_or_default();
let projection_billing = projection
.as_ref()
- .map(billing_rollup_from_projection)
+ .map(|projection| billing_rollup_from_projection(projection, None))
.unwrap_or_default();
let checkpoint = projection
.as_ref()
@@ -979,7 +979,7 @@ mod tests {
});
let projection_order = stage_projection_order(&projection);
- let projection_billing = billing_rollup_from_projection(&projection);
+ let projection_billing = billing_rollup_from_projection(&projection, None);
let mut latest_outcome = Outcome::success();
latest_outcome.usage = Some(success_usage);
latest_outcome.duration_ms = Some(800);

View file

@ -0,0 +1,6 @@
{
"outcome": "succeeded",
"notes": "Stage completed: implement",
"failure_reason": null,
"timestamp": "2026-05-21T22:59:35.631657Z"
}

View file

@ -0,0 +1,156 @@
Goal: # Plan: Compute LLM cost on-read for in-flight stages
## Context
On the run billing page (`/runs/{id}/billing`), an active stage shows token
usage but no dollar cost — cost renders as `—` until the stage completes.
Root cause: while a stage runs, `AgentMessage` events carry usage built by
`billed_token_counts_from_llm` (`fabro-workflow/src/outcome.rs:43`), which
hard-codes `total_usd_micros: None`. Dollar cost is only computed by
`billed_model_usage_from_llm` (`outcome.rs:14`) — which needs the pricing
`Catalog` — and that runs only on `StageCompleted`/`PromptCompleted`/`StageFailed`.
So an in-flight stage's `StageProjection.usage.total_usd_micros` stays `None`.
Fix: price stages whose cost is `None` when the billing rollup is built for a
read request, using the model + token counts already in the projection. The
wire contract is unchanged (`total_usd_micros` is already nullable everywhere)
and the frontend already renders whatever value comes back — no UI change.
## Decisions
- **Price any stage with `total_usd_micros == None`**, not just in-flight ones.
Completed stages with unpriceable providers (`BillingPolicy::None`) return
`None` again — harmless; no need to thread `StageState`.
- **No "estimated" label.** Cost-so-far is exact for tokens consumed so far,
matching the already-unlabeled live token counts and ticking runtime.
- **Aggregate billing stays finalized-only.** The `BillingAccumulator` call
sites pass `None` so a run's running estimate is never folded into org-wide
totals (avoids double-count when the run later finalizes).
- Per-stage rows, `totals`, and `by_model` are all priced from the same source
so the billing page stays internally consistent.
## Changes
### 1. `lib/crates/fabro-model/src/billing.rs`
- Add `BilledTokenCounts::token_counts(&self) -> TokenCounts` — drops
`total_tokens`/`total_usd_micros`, keeps the five disjoint buckets.
- Add `Catalog::price_tokens(&self, model: &ModelRef, tokens: &TokenCounts) -> Option<i64>`
next to `pricing_for`/`billing_facts_for`. Body mirrors the cost lines of
`billed_model_usage_from_llm`: build `ModelBillingFacts` via
`billing_facts_for`, assemble `ModelBillingInput { ModelUsage { model, tokens }, facts }`,
then `pricing_for(model).and_then(|p| p.bill(&input)).map(|a| a.0)`. Returns
`None` when the provider has no billing policy.
### 2. `lib/crates/fabro-workflow/src/billing_rollup.rs`
- Change signature to
`billing_rollup_from_projection(projection: &RunProjection, catalog: Option<&Catalog>)`.
- Add a module-private helper `stage_usage_with_cost(catalog, stage) -> BilledTokenCounts`:
clone `stage.usage`; if `total_usd_micros.is_none()` and both `catalog` and
`stage.model` are present, set it via `catalog.price_tokens(model, &usage.token_counts())`.
- In the loop, compute `priced` once per stage and use it in place of
`&stage.usage` for the `is_zero` check, `row.billing.add_counts`,
`totals.add_counts`, and `model_entry.billing.add_counts`.
- Update the existing tests to pass `None`; add one new test: an in-flight
stage (no `completion`, non-zero `usage` with `total_usd_micros: None`, a
builtin `model`) yields `Some(..)` cost on the stage row and in `totals` when
called with `Some(Catalog::builtin())`.
### 3. Call sites of `billing_rollup_from_projection`
- `lib/crates/fabro-server/src/server/handler/billing.rs:82` — bind
`let catalog = state.catalog();` (returns `Arc<Catalog>`) and pass
`Some(&catalog)`.
- `lib/crates/fabro-server/src/server.rs` (2 aggregate-billing sites) — pass `None`.
- `lib/crates/fabro-workflow/src/pipeline/finalize.rs` (4 sites) — pass `None`
(stages already priced at completion; pricing would be a no-op anyway).
No changes to `fabro-api.yaml`, the generated clients, or `apps/fabro-web`.
## Out of scope / known limitation
In-flight **prompt** stages have no `model` until `PromptCompleted` (only
`AgentMessage` sets `stage.model` mid-run), so they still show `—` while
running. Acceptable: prompt stages are a single short LLM call. The bug report
concerns agent stages, where `model` is available.
## Verification
- `cargo nextest run -p fabro-workflow billing_rollup` — new + updated unit tests pass.
- `cargo nextest run -p fabro-server billing` — handler conformance still passes.
- `cargo build --workspace` and `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings`.
- Manual: `fabro server start` + `cd apps/fabro-web && bun run dev`, start a
workflow with an agent stage, open `/runs/{id}/billing` mid-run — the active
stage row and totals show a non-`—` dollar amount that grows with tokens.
## Completed stages
- **toolchain**: succeeded
- Script: `command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1`
- Output:
```
cargo 1.95.0 (f2d3ce0bd 2026-03-21)
```
- **preflight_compile**: succeeded
- Script: `cargo check -q --workspace 2>&1`
- Output: (empty)
- **preflight_lint**: succeeded
- Script: `cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1`
- Output: (empty)
- **implement**: succeeded
- Model: claude-opus-4-7, 92.9k tokens in / 16.0k out
- Files: /home/daytona/workspace/fabro/lib/crates/fabro-model/src/billing.rs, /home/daytona/workspace/fabro/lib/crates/fabro-server/src/server.rs, /home/daytona/workspace/fabro/lib/crates/fabro-server/src/server/handler/billing.rs, /home/daytona/workspace/fabro/lib/crates/fabro-workflow/src/billing_rollup.rs, /home/daytona/workspace/fabro/lib/crates/fabro-workflow/src/pipeline/finalize.rs
# Simplify: Code Review and Cleanup
Review changes vs. origin for reuse, quality, and efficiency. Fix any issues found.
## Phase 1: Identify Changes
Run git diff (or git diff HEAD if there are staged changes) to see what changed. If there are no git changes, review the most recently modified files that the user mentioned or that you edited earlier in this conversation.
## Phase 2: Launch Three Review Agents in Parallel
Use the Agent tool to launch all three agents concurrently in a single message. Pass each agent the full diff so it has the complete context.
### Agent 1: Code Reuse Review
For each change:
1. Search for existing utilities and helpers that could replace newly written code. Use Grep to find similar patterns elsewhere in the codebase — common locations are utility directories, shared modules, and files adjacent to the changed ones.
2. Flag any new function that duplicates existing functionality. Suggest the existing function to use instead.
3. Flag any inline logic that could use an existing utility — hand-rolled string manipulation, manual path handling, custom environment checks, ad-hoc type guards, and similar patterns are common candidates.
Note: This is a greenfield app, so focus on maximizing simplicity and don't worry about changing things to achieve it.
### Agent 2: Code Quality Review
Review the same changes for hacky patterns:
1. Redundant state: state that duplicates existing state, cached values that could be derived, observers/effects that could be direct calls
2. Parameter sprawl: adding new parameters to a function instead of generalizing or restructuring existing ones
3. Copy-paste with slight variation: near-duplicate code blocks that should be unified with a shared abstraction
4. Leaky abstractions: exposing internal details that should be encapsulated, or breaking existing abstraction boundaries
5. Stringly-typed code: using raw strings where constants, enums (string unions), or branded types already exist in the codebase
Note: This is a greenfield app, so be aggressive in optimizing quality.
### Agent 3: Efficiency Review
Review the same changes for efficiency:
1. Unnecessary work: redundant computations, repeated file reads, duplicate network/API calls, N+1 patterns
2. Missed concurrency: independent operations run sequentially when they could run in parallel
3. Hot-path bloat: new blocking work added to startup or per-request/per-render hot paths
4. Unnecessary existence checks: pre-checking file/resource existence before operating (TOCTOU anti-pattern) — operate directly and handle the error
5. Memory: unbounded data structures, missing cleanup, event listener leaks
6. Overly broad operations: reading entire files when only a portion is needed, loading all items when filtering for one
## Phase 3: Fix Issues
Wait for all three agents to complete. Aggregate their findings and fix each issue directly. If a finding is a false positive or not worth addressing, note it and move on — do not argue with the finding, just skip it.
When done, briefly summarize what was fixed (or confirm the code was already clean).

View file

@ -0,0 +1,5 @@
{
"mode": "agent",
"provider": "anthropic",
"model": "claude-opus-4-7"
}

View file

@ -0,0 +1,30 @@
All clean. Here's the summary:
## Summary
Three review agents (reuse, quality, efficiency) ran in parallel against the diff. Aggregated findings:
### Fixed
1. **Renamed `BilledTokenCounts::token_counts()` → `to_token_counts()`** (`fabro-model/src/billing.rs`)
- Quality finding: original name read like a field accessor and hid the fact that the conversion is lossy (drops `total_tokens` and `total_usd_micros`). The `to_` prefix is the established Rust convention for lossy conversions.
2. **Moved pricing logic into `BilledTokenCounts::ensure_priced(catalog, model)`** (`fabro-model/src/billing.rs` + `fabro-workflow/src/billing_rollup.rs`)
- Quality finding: the previous `stage_usage_with_cost` helper cloned `BilledTokenCounts` across a crate boundary and directly poked the private `total_usd_micros` field, leaking knowledge of how to price a `BilledTokenCounts` out of `fabro-model`. The new method encapsulates that logic next to `add_billed_usage` / `replace_with_billed_usage`, and the rollup-side helper is now a 3-line wrapper.
### Considered and skipped
- **Reuse #1 (refactor `billed_model_usage_from_llm` to use `price_tokens`)** — the "duplicated" pricing tail is 3 lines and the surrounding error-vs-option shapes don't compose cleanly. Refactoring adds more indirection than it removes.
- **`Option<&Catalog>` parameter sprawl** — 6 of 7 call sites pass `None`, but splitting into two functions doubles the public surface for marginal benefit; the single signature is fine.
- **`pricing_for` memoization across stages** — real but minor (a few `String` clones per stage on a polled endpoint). Worth doing if the path ever shows up in profiles; not justified at the cost of HashMap setup for this fix.
- **`Cow::Borrowed` for the usage clone** — `BilledTokenCounts` is ~64 bytes of POD; no heap allocation.
- **Bare `i64` return from `price_tokens`** — matches the established `total_usd_micros: Option<i64>` convention used throughout the codebase.
### Verified
- `cargo nextest run -p fabro-workflow billing_rollup` — 4/4 pass (including the new in-flight pricing test).
- `cargo nextest run -p fabro-model billed_token_counts` — 5/5 pass.
- `cargo nextest run -p fabro-server billing` — 7/7 pass.
- `cargo build --workspace` — clean.
- `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings` — clean.
- `cargo +nightly-2026-04-14 fmt --check --all` — clean.