diff --git a/kernel/relayflowd-core/src/entry.rs b/kernel/relayflowd-core/src/entry.rs index 0f974b80e..5d70676c9 100644 --- a/kernel/relayflowd-core/src/entry.rs +++ b/kernel/relayflowd-core/src/entry.rs @@ -391,10 +391,35 @@ pub struct StepCompletedPayload { pub trajectory_tail: Option, #[serde(default)] pub budget: Budget, + /// What the step actually cost, as reported by the CLI or estimated from + /// its full token usage. Display only: it is never folded into the run's + /// budget, so `budget` alone decides `maxDollars`. Absent when unknown. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reported_cost: Option, pub completed_by: String, pub next_attempt_at_ms: Option, } +/// A step attempt's actual cost, kept apart from the metered `Budget`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct ReportedCost { + /// Non-negative decimal string, like `Budget::dollars`. + pub dollars: String, + pub source: ReportedCostSource, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ReportedCostSource { + /// The agent CLI's own reported total (e.g. Claude `total_cost_usd`). + Cli, + /// Estimated from the step's full token usage at a frozen price. + Priced, + /// No model ran: an internal effect worker (provider helper, plugin, MCP) + /// completed the step. `dollars` is zero. + NoModel, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct VerificationRecord { pub gate: String, @@ -691,3 +716,25 @@ pub struct RunCompletedPayload { pub failed_step_id: Option, pub budget_total: Budget, } + +#[cfg(test)] +mod reported_cost_tests { + use super::*; + + #[test] + fn step_completed_without_reported_cost_round_trips_unchanged() { + // A journal written before `reported_cost` existed must still decode, + // and a payload without one must serialize exactly as it always did. + let legacy = serde_json::json!({ + "completionReason": "success", "disposition": "step_done", "output": null, + "verification": null, "end_pins": null, "effects": [], + "budget": {"tokens_in": 1, "tokens_out": 2, "dollars": "0.1"}, + "completed_by": "w", "next_attempt_at_ms": null + }); + let payload: StepCompletedPayload = serde_json::from_value(legacy.clone()).unwrap(); + assert_eq!(payload.reported_cost, None); + let reserialized = serde_json::to_value(&payload).unwrap(); + assert!(reserialized.get("reported_cost").is_none()); + assert_eq!(reserialized["budget"], legacy["budget"]); + } +} diff --git a/kernel/relayflowd-core/src/machine.rs b/kernel/relayflowd-core/src/machine.rs index fd1be7840..2abd93f80 100644 --- a/kernel/relayflowd-core/src/machine.rs +++ b/kernel/relayflowd-core/src/machine.rs @@ -57,6 +57,8 @@ pub struct AttemptResult { pub end_pins: Option, pub effects: Vec, pub trajectory_tail: Option, + /// Display-only actual cost; journaled beside `budget`, never charged. + pub reported_cost: Option, /// Execution failures bypass verification but still follow retry policy. pub failure_reason: Option, /// Why the attempt was rejected, in the vocabulary of whoever rejected it. @@ -76,6 +78,7 @@ impl AttemptResult { end_pins: None, effects: Vec::new(), trajectory_tail: None, + reported_cost: None, failure_reason: None, failure_detail: None, } @@ -421,6 +424,7 @@ pub fn completion_actions( budget: result.budget, completed_by: result.completed_by, next_attempt_at_ms, + reported_cost: result.reported_cost, }, ); let mut actions = vec![Action::Append(completed)]; diff --git a/kernel/relayflowd-core/src/machine/cancel.rs b/kernel/relayflowd-core/src/machine/cancel.rs index 728d540bd..47b261dc9 100644 --- a/kernel/relayflowd-core/src/machine/cancel.rs +++ b/kernel/relayflowd-core/src/machine/cancel.rs @@ -57,6 +57,7 @@ pub(super) fn cancel_run_actions(state: &RunState, now_ms: i64) -> Vec { budget: Budget::default(), completed_by: "kernel".to_owned(), next_attempt_at_ms: None, + reported_cost: None, }, ))); } diff --git a/kernel/relayflowd-core/src/machine/parallel_tests.rs b/kernel/relayflowd-core/src/machine/parallel_tests.rs index 755bdb3d1..ad5861b32 100644 --- a/kernel/relayflowd-core/src/machine/parallel_tests.rs +++ b/kernel/relayflowd-core/src/machine/parallel_tests.rs @@ -450,6 +450,7 @@ fn failed_run_drains_open_siblings_before_terminal_entry() { budget: Budget::default(), completed_by: "late-worker".to_owned(), next_attempt_at_ms: None, + reported_cost: None, }, )); assert!(matches!( diff --git a/kernel/relayflowd-core/src/machine/recovery.rs b/kernel/relayflowd-core/src/machine/recovery.rs index e59c236c4..b0da7e398 100644 --- a/kernel/relayflowd-core/src/machine/recovery.rs +++ b/kernel/relayflowd-core/src/machine/recovery.rs @@ -134,6 +134,7 @@ pub fn abandonment_actions( budget: Budget::default(), completed_by: "kernel".to_owned(), next_attempt_at_ms, + reported_cost: None, }, ))]; if manual { diff --git a/kernel/relayflowd-core/src/machine/tests.rs b/kernel/relayflowd-core/src/machine/tests.rs index 22d3ca2d0..6f160c9b3 100644 --- a/kernel/relayflowd-core/src/machine/tests.rs +++ b/kernel/relayflowd-core/src/machine/tests.rs @@ -364,6 +364,7 @@ fn all_backing_off_steps_return_timers() { trajectory_tail: None, failure_reason: Some(CompletionReason::WorkerError), failure_detail: Some("stub rejection".to_owned()), + reported_cost: None, }; let mut entries = Vec::new(); for step in &spec.steps { @@ -475,6 +476,7 @@ fn inspect_recovery_injects_the_dirty_pin_completion_reason_and_tail() { trajectory_tail: Some(json!(["edited file"])), failure_reason: Some(CompletionReason::WorkerError), failure_detail: Some("stub rejection".to_owned()), + reported_cost: None, }; let completed = completion_actions( "run", @@ -595,6 +597,7 @@ fn worker_reported_failure_without_detail_still_records_a_verification() { // The point of the case: the worker reported a failure and sent no // detail with it. failure_detail: None, + reported_cost: None, }; let entries: Vec<_> = completion_actions("run", &spec.steps[0], 1, 0, result, 1_000) .into_iter() diff --git a/kernel/relayflowd-core/src/memoization.rs b/kernel/relayflowd-core/src/memoization.rs index a9b174808..c568a7a73 100644 --- a/kernel/relayflowd-core/src/memoization.rs +++ b/kernel/relayflowd-core/src/memoization.rs @@ -121,6 +121,9 @@ pub fn next_actions_with_reuse( seq: prior.seq, }); payload.budget = Budget::default(); + // The model did not run for this completion: the + // source's actual cost belongs to the source run. + payload.reported_cost = None; payload.completed_by = "kernel:reuse".into(); result.push(Action::Append(JournalEntry::new( EntryType::StepCompleted, diff --git a/kernel/relayflowd-core/src/state/tests.rs b/kernel/relayflowd-core/src/state/tests.rs index 9d7fc53c1..e3ab6c65b 100644 --- a/kernel/relayflowd-core/src/state/tests.rs +++ b/kernel/relayflowd-core/src/state/tests.rs @@ -40,6 +40,7 @@ fn journal_replays_data_gate_verdict_without_rerunning_completed_code() { budget: Budget::default(), completed_by: "kernel".to_owned(), next_attempt_at_ms: None, + reported_cost: None, }, ); let state = RunState::fold("run", spec(), &[completion]).unwrap(); diff --git a/kernel/relayflowd-core/tests/memoization.rs b/kernel/relayflowd-core/tests/memoization.rs index b1cfda4e5..2903396ca 100644 --- a/kernel/relayflowd-core/tests/memoization.rs +++ b/kernel/relayflowd-core/tests/memoization.rs @@ -60,6 +60,7 @@ fn source(state: &RunState) -> relayflowd_core::JournalEntry { entry.payload["step_spec_hash"] = step_spec_hash(step).into(); entry.payload["input_hash"] = canonical_hash(&json!({})).into(); entry.payload["budget"] = json!({"tokens_in":10,"tokens_out":20,"dollars":"1.25"}); + entry.payload["reported_cost"] = json!({"dollars":"9.500000","source":"cli"}); entry } #[test] @@ -77,6 +78,8 @@ fn match_reuses_output_with_provenance_and_zero_cost_without_dispatch() { json!({"run_id":"old","step_id":"a","seq":7}) ); assert_eq!(entry.payload["budget"]["tokens_in"], 0); + // The source run paid for the model; the reusing run's completion did not. + assert!(entry.payload.get("reported_cost").is_none()); let folded = RunState::fold("new", state.spec, &[entry.clone()]).unwrap(); assert!(folded.all_steps_succeeded()); } diff --git a/kernel/relayflowd/src/engine/remote.rs b/kernel/relayflowd/src/engine/remote.rs index c0b8a83d8..995d652ce 100644 --- a/kernel/relayflowd/src/engine/remote.rs +++ b/kernel/relayflowd/src/engine/remote.rs @@ -28,6 +28,8 @@ pub struct OutOfBandCompletion { pub end_pins: Option, pub effects: Vec, pub trajectory_tail: Option, + /// Display-only actual cost. Journaled with the completion; never charged. + pub reported_cost: Option, } /// A human question a leased attempt parks on. The worker names the wait so @@ -136,6 +138,20 @@ impl Engine { { bail!("step completion usage.dollars_unmetered must not carry non-zero dollars"); } + if let Some(cost) = &completion.reported_cost + && (!relayflowd_core::memory::valid_decimal(&cost.dollars) + || relayflowd_core::journal_dollars(&cost.dollars).is_err()) + { + bail!("step completion reported_cost.dollars must be a non-negative decimal string"); + } + // `no_model` asserts nothing ran that could cost money; a non-zero + // amount contradicts it and would be dropped by every reader. + if let Some(cost) = &completion.reported_cost + && cost.source == relayflowd_core::ReportedCostSource::NoModel + && cost.dollars.bytes().any(|b| b.is_ascii_digit() && b != b'0') + { + bail!("step completion reported_cost.source no_model must carry zero dollars"); + } let mut journal = self.open_run(run_id)?; let spec = journal.run_spec().context("read run spec")?; let state = self.load_state(&journal, spec.clone())?; @@ -238,6 +254,7 @@ impl Engine { trajectory_tail: completion.trajectory_tail, failure_reason, failure_detail, + reported_cost: completion.reported_cost, }; for action in completion_actions( run_id, diff --git a/kernel/relayflowd/src/exec_det.rs b/kernel/relayflowd/src/exec_det.rs index e588e9298..8bfdd039f 100644 --- a/kernel/relayflowd/src/exec_det.rs +++ b/kernel/relayflowd/src/exec_det.rs @@ -136,6 +136,7 @@ pub(crate) fn execute_placed_with_input( failure_reason: timed_out.then_some(CompletionReason::Timeout), failure_detail: timed_out .then(|| format!("step exceeded its {} ms timeout", timeout.as_millis())), + reported_cost: None, } } @@ -177,6 +178,7 @@ fn worker_error(detail: &str) -> AttemptResult { trajectory_tail: None, failure_reason: Some(CompletionReason::WorkerError), failure_detail: Some(detail.to_owned()), + reported_cost: None, } } diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index b60f099f6..a0960c53a 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -163,7 +163,14 @@ fn handle_request( format!("relayflowd supports protocol {PROTOCOL_VERSION}"), )); } - Ok(json!({"protocol": PROTOCOL_VERSION, "server": "relayflowd"})) + // Additive capabilities within protocol 0. A client sends an + // optional request field only when the daemon lists it, so a newer + // SDK never trips an older daemon's `deny_unknown_fields`. + Ok(json!({ + "protocol": PROTOCOL_VERSION, + "server": "relayflowd", + "features": ["reported_cost"], + })) } "run.start" => { let params: RunStartParams = decode_params(request.params)?; @@ -432,6 +439,7 @@ fn handle_request( end_pins: params.end_pins, effects: params.effects, trajectory_tail: params.trajectory_tail, + reported_cost: params.reported_cost, }, ) .map_err(internal_error)?; diff --git a/kernel/relayflowd/src/server/wire.rs b/kernel/relayflowd/src/server/wire.rs index 49646b436..b999c5daf 100644 --- a/kernel/relayflowd/src/server/wire.rs +++ b/kernel/relayflowd/src/server/wire.rs @@ -112,6 +112,8 @@ pub(super) struct StepCompleteParams { pub effects: Vec, #[serde(default)] pub trajectory_tail: Option, + #[serde(default)] + pub reported_cost: Option, } #[derive(Deserialize)] diff --git a/kernel/relayflowd/tests/budget_gate.rs b/kernel/relayflowd/tests/budget_gate.rs index 8e3ce33a6..4ae6b6a24 100644 --- a/kernel/relayflowd/tests/budget_gate.rs +++ b/kernel/relayflowd/tests/budget_gate.rs @@ -3,7 +3,8 @@ use relayflowd::{ worker::{DispatchOutcome, JournalObserver, StepDispatch, StepDispatcher}, }; use relayflowd_core::{ - Budget, CompletionReason, EntryType, JournalEntry, RunCompletionReason, RunSpec, StepType, + Budget, CompletionReason, EntryType, JournalEntry, ReportedCost, ReportedCostSource, + RunCompletionReason, RunSpec, StepType, }; use serde_json::json; use std::sync::{Arc, Mutex}; @@ -58,6 +59,7 @@ fn crossing_completion_is_durable_and_next_step_is_refused() { end_pins: None, effects: vec![], trajectory_tail: None, + reported_cost: None, }, ) .unwrap(); @@ -197,6 +199,7 @@ fn complete_first( end_pins: None, effects: vec![], trajectory_tail: None, + reported_cost: None, }, ) } @@ -428,3 +431,139 @@ fn prior_spend_metering_flag_is_additive_and_fails_closed_for_older_kernels() { .is_err() ); } + +#[test] +fn reported_cost_is_journaled_but_never_charged_to_the_budget() { + let dir = tempfile::tempdir().unwrap(); + let worker = Arc::new(Worker::default()); + let engine = Engine::with_runtime(dir.path(), worker.clone(), worker.clone()); + // A tight dollar cap the reported cost alone would cross many times over. + let spec = RunSpec::parse(&json!({"budget":{"max_dollars":"0.01"},"steps":[ + {"id":"first","type":"llm","prompt":"work"}, + {"id":"second","type":"llm","prompt":"answer","depends_on":["first"]} + ]})) + .unwrap(); + let started = engine.start(spec, "test", None).unwrap(); + let d = worker.0.lock().unwrap()[0].clone(); + let outcome = engine + .complete_out_of_band( + &started.run_id, + "first", + OutOfBandCompletion { + human_intervention: false, + attempt: d.attempt, + idempotency_key: d.idempotency_key, + completion_reason: CompletionReason::Success, + output: json!("done"), + budget: Budget { + tokens_in: 100, + tokens_out: 10, + dollars: "0.001000".into(), + dollars_unmetered: false, + }, + completed_by: "mock".into(), + started_pins: None, + end_pins: None, + effects: vec![], + trajectory_tail: None, + reported_cost: Some(ReportedCost { + dollars: "7.169405".into(), + source: ReportedCostSource::Cli, + }), + }, + ) + .unwrap(); + // Admission is decided by metered dollars only: the next step runs. + assert_eq!(outcome.completion_reason, None); + assert!( + worker + .0 + .lock() + .unwrap() + .iter() + .any(|d| d.step_id == "second") + ); + let entries = Engine::new(dir.path()) + .journal_entries(&started.run_id, 1, 100) + .unwrap(); + let completed = entries + .iter() + .find(|e| e.entry_type == EntryType::StepCompleted) + .unwrap(); + assert_eq!( + completed.payload["reported_cost"], + json!({"dollars": "7.169405", "source": "cli"}) + ); + assert_eq!(completed.payload["budget"]["dollars"], "0.001000"); + assert_eq!(completed.payload["spend"]["dollars"], json!(0.001)); +} + +#[test] +fn a_malformed_reported_cost_is_refused() { + let dir = tempfile::tempdir().unwrap(); + let worker = Arc::new(Worker::default()); + let engine = Engine::with_runtime(dir.path(), worker.clone(), worker.clone()); + let spec = + RunSpec::parse(&json!({"steps":[{"id":"first","type":"llm","prompt":"a"}]})).unwrap(); + let started = engine.start(spec, "test", None).unwrap(); + let d = worker.0.lock().unwrap()[0].clone(); + let error = engine + .complete_out_of_band( + &started.run_id, + "first", + OutOfBandCompletion { + human_intervention: false, + attempt: d.attempt, + idempotency_key: d.idempotency_key, + completion_reason: CompletionReason::Success, + output: json!("a"), + budget: Budget::default(), + completed_by: "mock".into(), + started_pins: None, + end_pins: None, + effects: vec![], + trajectory_tail: None, + reported_cost: Some(ReportedCost { + dollars: "-1".into(), + source: ReportedCostSource::Cli, + }), + }, + ) + .unwrap_err(); + assert!(error.to_string().contains("reported_cost"), "{error}"); +} + +#[test] +fn a_no_model_cost_must_be_zero() { + let dir = tempfile::tempdir().unwrap(); + let worker = Arc::new(Worker::default()); + let engine = Engine::with_runtime(dir.path(), worker.clone(), worker.clone()); + let spec = + RunSpec::parse(&json!({"steps":[{"id":"first","type":"llm","prompt":"a"}]})).unwrap(); + let started = engine.start(spec, "test", None).unwrap(); + let d = worker.0.lock().unwrap()[0].clone(); + let completion = |dollars: &str| OutOfBandCompletion { + human_intervention: false, + attempt: d.attempt, + idempotency_key: d.idempotency_key.clone(), + completion_reason: CompletionReason::Success, + output: json!("a"), + budget: Budget::default(), + completed_by: "mock".into(), + started_pins: None, + end_pins: None, + effects: vec![], + trajectory_tail: None, + reported_cost: Some(ReportedCost { + dollars: dollars.into(), + source: ReportedCostSource::NoModel, + }), + }; + let error = engine + .complete_out_of_band(&started.run_id, "first", completion("5")) + .unwrap_err(); + assert!(error.to_string().contains("no_model"), "{error}"); + engine + .complete_out_of_band(&started.run_id, "first", completion("0.000000")) + .unwrap(); +} diff --git a/kernel/relayflowd/tests/parallel_driver.rs b/kernel/relayflowd/tests/parallel_driver.rs index bf8086830..17eb24442 100644 --- a/kernel/relayflowd/tests/parallel_driver.rs +++ b/kernel/relayflowd/tests/parallel_driver.rs @@ -177,6 +177,7 @@ fn complete(engine: &Engine, run_id: &str, dispatch: &StepDispatch) -> RunStatus end_pins: None, effects: Vec::new(), trajectory_tail: None, + reported_cost: None, }, ) .unwrap() diff --git a/packages/sdk/src/authored-helper-effect.ts b/packages/sdk/src/authored-helper-effect.ts index f878e0b14..f10a7b641 100644 --- a/packages/sdk/src/authored-helper-effect.ts +++ b/packages/sdk/src/authored-helper-effect.ts @@ -14,6 +14,7 @@ import { helperWriteback, HelperDeliveryError } from './helper-writeback.js'; import { checkSlackHelpers } from './slack-preflight.js'; import { atomicJson, readSlackReceipt, receiptPath, slackWriteback, type SlackCall } from './slack-writeback.js'; import { authoredChildAdmissionKey } from './authored-admission.js'; +import { NO_MODEL_COST } from './reported-cost.js'; type ProviderCall = HelperCall | SlackCall; @@ -135,11 +136,11 @@ async function completeHelperDispatch(client: JournalClient, dispatch: StepDispa await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt, dispatch.idempotency_key, 'worker_error', { output: { ...call, diagnostic: error.message }, - started_pins: dispatch.pins, end_pins: dispatch.pins, effects: [], + started_pins: dispatch.pins, end_pins: dispatch.pins, effects: [], reported_cost: NO_MODEL_COST, }); return; } await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt, - dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins, + dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins, reported_cost: NO_MODEL_COST, effects: [{ surface_path: `/${call.provider}`, idempotency_key: dispatch.idempotency_key }] }); } diff --git a/packages/sdk/src/authored-mcp.ts b/packages/sdk/src/authored-mcp.ts index 7baa613bd..5b8b36f0b 100644 --- a/packages/sdk/src/authored-mcp.ts +++ b/packages/sdk/src/authored-mcp.ts @@ -12,6 +12,7 @@ import type { AuthoredFlowJournalStep } from './authored-flow-executor.js'; import { readCompletedStepOutput } from './authored-step-output.js'; import { withWorkerLease } from './worker-lease.js'; import { authoredChildAdmissionKey } from './authored-admission.js'; +import { NO_MODEL_COST } from './reported-cost.js'; /** Own keys expose precisely the preflight inventory, including prototype-like names. */ export function buildMcpProxy( @@ -116,6 +117,7 @@ export async function runMcpEffect( output: { ...receipt, output: output ?? null, ...(diagnostic ? { diagnostic } : {}) }, started_pins: event.pins, end_pins: event.pins, ...(diagnostic ? { trajectory_tail: { ...receipt, diagnostic } } : {}), + reported_cost: NO_MODEL_COST, effects: confirmed ? [{ surface_path: surfacePath, idempotency_key: event.idempotency_key }] : [], }); } diff --git a/packages/sdk/src/authored-plugin-effect.ts b/packages/sdk/src/authored-plugin-effect.ts index c853b55f8..7a8587e56 100644 --- a/packages/sdk/src/authored-plugin-effect.ts +++ b/packages/sdk/src/authored-plugin-effect.ts @@ -11,6 +11,7 @@ import type { AuthoredFlowJournalStep } from './authored-flow-executor.js'; import { readCompletedStepOutput } from './authored-step-output.js'; import { withWorkerLease } from './worker-lease.js'; import { authoredChildAdmissionKey } from './authored-admission.js'; +import { NO_MODEL_COST } from './reported-cost.js'; export class PluginStepError extends AuthoredFlowExecutionError { constructor(readonly diagnostic: string, runId: string) { @@ -88,6 +89,7 @@ export async function runPluginEffect( output: { ...receipt, output: output ?? null, ...(diagnostic ? { diagnostic } : {}) }, started_pins: event.pins, end_pins: event.pins, ...(diagnostic ? { trajectory_tail: { ...receipt, diagnostic } } : {}), + reported_cost: NO_MODEL_COST, effects: confirmed ? [{ surface_path: surfacePath, idempotency_key: event.idempotency_key }] : [], }); } diff --git a/packages/sdk/src/authored-verdict.ts b/packages/sdk/src/authored-verdict.ts index 78d58d563..95b309b16 100644 --- a/packages/sdk/src/authored-verdict.ts +++ b/packages/sdk/src/authored-verdict.ts @@ -67,7 +67,8 @@ export function authoredVerdictOf(events: readonly JournalEvent[]): AuthoredVerd return Object.freeze({ reason, ...(detail === undefined ? {} : { detail }) }); } -function declaresAuthoredRoot(spawned: JournalEvent): boolean { +/** Whether a journal's spawned spec is a real authored root, not a step merely named `authored-root`. */ +export function declaresAuthoredRoot(spawned: JournalEvent): boolean { const spec = record(record(spawned.payload)?.['spec']); const steps = spec?.['steps']; if (!Array.isArray(steps)) return false; diff --git a/packages/sdk/src/decimal-dollars.ts b/packages/sdk/src/decimal-dollars.ts new file mode 100644 index 000000000..b6b80164a --- /dev/null +++ b/packages/sdk/src/decimal-dollars.ts @@ -0,0 +1,38 @@ +// Exact decimal-dollar arithmetic shared by the journal folds. + +/** + * Sum two decimal dollar strings exactly, at whatever scale they carry — the + * kernel's budget fold (`relayflowd-core/src/state/budget.rs`) aligns every + * fractional digit before adding, and it accepts arbitrary precision. Scaling + * to a fixed six fractional digits dropped a charge of `0.0000001` to zero, so + * `flows status` underreported spend the kernel had totalled exactly. + * + * A value that is not a decimal number is left out of the sum rather than + * counted as zero, and the caller is told by `malformed`. + */ +export function addDollars(left: string, right: string): string { + const parsed = [left, right].map((value) => /^\d+(?:\.\d+)?$/.test(value) ? value : null); + const usable = parsed.filter((value): value is string => value !== null); + if (usable.length === 0) return '0'; + if (usable.length === 1) return normalizeDollars(usable[0]!); + const scale = Math.max(...usable.map((value) => (value.split('.')[1] ?? '').length)); + const scaled = usable.map((value) => { + const [whole, fraction = ''] = value.split('.'); + return BigInt(`${whole}${fraction.padEnd(scale, '0')}`); + }); + return fromScaled(scaled[0]! + scaled[1]!, scale); +} + +/** `12.3400` -> `12.34`, `0.000` -> `0`; the journal's own spelling otherwise. */ +function normalizeDollars(value: string): string { + const [whole, fraction = ''] = value.split('.'); + return fromScaled(BigInt(`${whole}${fraction}`), fraction.length); +} + +function fromScaled(total: bigint, scale: number): string { + if (scale === 0) return String(total); + const digits = String(total).padStart(scale + 1, '0'); + const whole = digits.slice(0, digits.length - scale); + const fraction = digits.slice(digits.length - scale).replace(/0+$/, ''); + return fraction.length === 0 ? whole : `${whole}.${fraction}`; +} diff --git a/packages/sdk/src/effect-channel.ts b/packages/sdk/src/effect-channel.ts index 4425a8ee0..0998271ae 100644 --- a/packages/sdk/src/effect-channel.ts +++ b/packages/sdk/src/effect-channel.ts @@ -4,6 +4,7 @@ import type { JournalClient } from './journal-client.js'; import type { StepDispatchEvent } from './protocol.js'; import { SPEC_SCHEMA_VERSION } from './spec.js'; import { withWorkerLease } from './worker-lease.js'; +import { NO_MODEL_COST } from './reported-cost.js'; import { snapshotJsonValue } from './json-value.js'; /** Structural subset of Agent Relay's messages.dm; credentials stay in the client. */ @@ -93,5 +94,6 @@ export async function completeChannelPost( messageId, channel: call.channel, to: call.to, text: call.text }, started_pins: dispatch.pins, end_pins: dispatch.pins, effects: [{ surface_path: surfacePath, idempotency_key: dispatch.idempotency_key }], + reported_cost: NO_MODEL_COST, }); } diff --git a/packages/sdk/src/helper-instruction.ts b/packages/sdk/src/helper-instruction.ts new file mode 100644 index 000000000..a6d36c056 --- /dev/null +++ b/packages/sdk/src/helper-instruction.ts @@ -0,0 +1,2 @@ +/** The instruction prefix a lowered provider helper step carries (`yaml-helpers.ts`). */ +export const HELPER_INSTRUCTION_PREFIX = 'relayflows:helper:v1\n'; diff --git a/packages/sdk/src/journal-client.ts b/packages/sdk/src/journal-client.ts index e42f81f59..9fe998eed 100644 --- a/packages/sdk/src/journal-client.ts +++ b/packages/sdk/src/journal-client.ts @@ -13,7 +13,7 @@ import { EventEmitter } from 'node:events'; export { walkJournal, JournalReadError, type JournalEvent, type JournalReadFailure } from './journal-reader.js'; import { randomUUID } from 'node:crypto'; import { createConnection, type Socket } from 'node:net'; -import type { VerbContract, EventEmitParams, EventSubmitParams } from './protocol.js'; +import type { VerbContract, EventEmitParams, EventSubmitParams, ReportedCost } from './protocol.js'; import { PROTOCOL_VERSION, type CompletionReason, @@ -61,6 +61,8 @@ export class JournalProtocolError extends Error { } export class JournalClient extends EventEmitter { + /** Additive request capabilities the daemon advertised at `hello`; none until then. */ + private features: ReadonlySet = new Set(); private socket: Socket | null = null; /** Why the connection ended, so a later "not connected" names its cause rather than hiding it. */ private disconnectCause: Error | undefined; @@ -215,9 +217,11 @@ export class JournalClient extends EventEmitter { // --- Typed verb methods (gate 1 minimal set, kernel DESIGN.md §5) -------- - /** Handshake; version mismatch is a hard error. */ - hello(client: string): Promise { - return this.request('hello', { protocol: PROTOCOL_VERSION, client }); + /** Handshake; version mismatch is a hard error. Records the daemon's additive features. */ + async hello(client: string): Promise { + const result = await this.request('hello', { protocol: PROTOCOL_VERSION, client }); + this.features = new Set(Array.isArray(result.features) ? result.features.filter(f => typeof f === 'string') : []); + return result; } /** @@ -403,15 +407,20 @@ export class JournalClient extends EventEmitter { effects?: EffectRef[]; trajectory_tail?: unknown; human_intervention?: boolean; + reported_cost?: ReportedCost; } = {}, ): Promise { + // `reported_cost` is display-only, and a daemon that predates it refuses + // unknown completion fields: drop it rather than fail the completion. + const { reported_cost: reportedCost, ...rest } = extra; return this.request('step.complete', { run_id: runId, step_id: stepId, attempt, idempotency_key: idempotencyKey, completionReason, - ...extra, + ...rest, + ...(reportedCost !== undefined && this.features.has('reported_cost') ? { reported_cost: reportedCost } : {}), }, null).catch(error => { if (error instanceof JournalProtocolError) error.verb = 'step.complete'; throw error; diff --git a/packages/sdk/src/llm-worker.ts b/packages/sdk/src/llm-worker.ts index 3890aab05..6e02e3987 100644 --- a/packages/sdk/src/llm-worker.ts +++ b/packages/sdk/src/llm-worker.ts @@ -1,4 +1,5 @@ import { workerSpend } from './worker-spend.js'; +import { reportedCost } from './reported-cost.js'; import { DEFAULT_LOCAL_AGENT_CAPACITY } from './worker-slots.js'; import type { WorkerCliResult } from './worker-cli.js'; import { EventEmitter } from 'node:events'; @@ -62,6 +63,7 @@ export class LlmWorker extends EventEmitter { ? runAgentCli(spec.cli, workerInstruction(prompt, dispatch), dispatch.wake_context, effectiveModel, undefined, signal, 'llm') : Promise.resolve({ exit_code: null, stdout_tail: '', stderr_tail: 'llm step has no declared CLI' })); const { result, usage } = workerSpend(completed, effectiveModel); + const cost = reportedCost(completed, effectiveModel); let reason: CompletionReason = result.exit_code === 0 ? 'success' : 'worker_error'; let output: unknown = result.stdout_tail; let detail = result.stderr_tail; @@ -94,6 +96,7 @@ export class LlmWorker extends EventEmitter { dispatch.idempotency_key, reason, { output, ...(usage !== undefined ? { usage } : {}), + ...(cost === undefined ? {} : { reported_cost: cost }), ...(Object.keys(trajectoryTail).length === 0 ? {} : { trajectory_tail: trajectoryTail }), }); } diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index ebb33cd96..d4296647c 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -79,6 +79,11 @@ export interface HelloParams { export interface HelloResult { protocol: 0; server: string; + /** + * Additive request capabilities within protocol 0. Absent from daemons that + * predate it, which then accept none of the optional fields it gates. + */ + features?: string[]; } export interface StepSpend { @@ -97,6 +102,21 @@ export interface StepSpend { * `dollars_unmetered: true` and no dollar amount, so unknown cost is never * journaled as a measured $0. See `workerSpend` for the full contract. */ +/** + * A step attempt's actual cost (`reported-cost.ts`), journaled beside the + * metered `usage` as `reported_cost`. Display only: the kernel never charges it + * against `maxDollars`. + */ +export interface ReportedCost { + /** Non-negative decimal string, six places. */ + dollars: string; + /** + * `cli`: the CLI's own reported total. `priced`: estimated from full token + * usage. `no_model`: an internal effect worker completed the step; $0. + */ + source: 'cli' | 'priced' | 'no_model'; +} + export type StepUsage = | { tokens_in: number; tokens_out: number; dollars: string; dollars_unmetered?: never } | { tokens_in: number; tokens_out: number; dollars_unmetered: true; dollars?: never }; @@ -342,6 +362,8 @@ export interface StepCompleteParams { effects?: EffectRef[]; /** `inspect` evidence for the next attempt. Rejected over 16 KiB of JSON. */ trajectory_tail?: unknown; + /** Display-only actual cost; never charged to the budget. */ + reported_cost?: ReportedCost; } export type StepCompleteResult = RunOutcome; diff --git a/packages/sdk/src/reported-cost-total.ts b/packages/sdk/src/reported-cost-total.ts new file mode 100644 index 000000000..1080edb3b --- /dev/null +++ b/packages/sdk/src/reported-cost-total.ts @@ -0,0 +1,82 @@ +// The display-only actual-cost fold beside run-state's metered spend. +// Nothing here is enforcement: the kernel charges `budget`, never `reported_cost`. + +import { declaresAuthoredRoot } from './authored-verdict.js'; +import { addDollars } from './decimal-dollars.js'; +import { HELPER_INSTRUCTION_PREFIX } from './helper-instruction.js'; +import type { JournalEvent } from './journal-reader.js'; + +type Payload = Record; + +/** + * What steps actually cost, summed from each attempt's journaled + * `reported_cost` (the CLI's own total, or a full-usage estimate). Display + * only: the kernel charges `spend` against `maxDollars`, never this. `complete` + * is false when some completed model attempt reported no cost (e.g. Codex, or + * an older runtime), so `dollars` is then a lower bound. Segment rollover keeps + * earlier completions in the journal, so an epoch summary does not lose any. + */ +export interface ReportedCostTotal { + /** Decimal string. */ + dollars: string; + complete: boolean; + /** Where the summed figures came from; `mixed` when attempts differ. */ + source: 'cli' | 'priced' | 'mixed' | null; +} + +/** + * Steps known to run no model, read from the spawned spec. Deliberately + * narrow: a crash-recovered or cancelled model attempt journals a default zero + * budget although the model ran, so metering nothing proves nothing. A step + * not listed here that reports no cost stays unknown, and the total reads as a + * lower bound rather than a false complete figure. + * - a deterministic step runs a command; + * - a provider helper (`relayflows:helper:v1`) performs a relayfile write; + * - the authored root runs the flow body, whose model calls are child runs with + * journals of their own. Only a spec carrying the authored-root discriminator + * counts: an ordinary flow may name a step `authored-root`. + * A memoized reuse is free too, per completion (`reused_from`): the source run + * paid for the model. + */ +export function modelFreeSteps(spawned: JournalEvent): Set { + const payload = spawned.payload !== null && typeof spawned.payload === 'object' ? spawned.payload as Payload : {}; + const spec = payload['spec'] !== null && typeof payload['spec'] === 'object' ? payload['spec'] as Payload : {}; + const steps = Array.isArray(spec['steps']) ? spec['steps'] as unknown[] : []; + const authoredRoot = declaresAuthoredRoot(spawned); + const free = new Set(); + for (const entry of steps) { + const step = entry !== null && typeof entry === 'object' ? entry as Payload : {}; + if (typeof step['id'] !== 'string') continue; + const helper = step['type'] === 'agent' && typeof step['instruction'] === 'string' + && step['instruction'].startsWith(HELPER_INSTRUCTION_PREFIX); + if (step['type'] === 'deterministic' || helper || (authoredRoot && step['id'] === 'authored-root')) free.add(step['id']); + } + return free; +} + +export const NO_REPORTED_COST: ReportedCostTotal = { dollars: '0', complete: true, source: null }; + + +/** + * Add one completed attempt's journaled `reported_cost` to a total. A model + * attempt without one (older runtime, Codex, unpriced) leaves the sum unchanged + * and marks it incomplete rather than counting an unknown cost as zero. + */ +export function addReportedCost(total: ReportedCostTotal, completion: Payload, modelFree: boolean): ReportedCostTotal { + const charge = completion['reported_cost']; + const cost = charge !== null && typeof charge === 'object' && !Array.isArray(charge) ? charge as Payload : null; + const reused = completion['reused_from'] !== undefined && completion['reused_from'] !== null; + if (cost === null && (modelFree || reused)) return total; + const dollars = cost === null ? null : cost['dollars']; + const source = cost === null ? null : cost['source']; + // An internal effect worker ran no model: a known $0 that names no cost source. + if (source === 'no_model') return total; + if (typeof dollars !== 'string' || !/^\d+(?:\.\d+)?$/.test(dollars) || (source !== 'cli' && source !== 'priced')) { + return { ...total, complete: false }; + } + return { + dollars: addDollars(total.dollars, dollars), + complete: total.complete, + source: total.source === null || total.source === source ? source : 'mixed', + }; +} diff --git a/packages/sdk/src/reported-cost.ts b/packages/sdk/src/reported-cost.ts new file mode 100644 index 000000000..231330e2d --- /dev/null +++ b/packages/sdk/src/reported-cost.ts @@ -0,0 +1,51 @@ +import { MODEL_PRICING } from './model-pricing.js'; +import type { ReportedCost } from './protocol.js'; +import type { WorkerCliResult } from './worker-cli.js'; + +/** + * What a step attempt actually cost, for display. Never enforcement. + * + * The metered `usage` (`workerSpend`) is what the kernel charges against + * `maxDollars`, and it prices only plain input and output tokens. That + * undercounts cache-heavy agents badly. This is the real figure a reader wants + * to see, and it rides beside the metered charge as `reported_cost`: the kernel + * journals it and never folds it into the budget. Changing which charges count + * toward a cap is a separate decision, and this module takes no part in it. + * + * Sources, in order: + * - `cli`: the CLI's own reported total (Claude `total_cost_usd`); + * - `priced`: an estimate from the digest's full token usage at the frozen + * `MODEL_PRICING` rates, with cache writes at 1.25x and cache reads at 0.1x + * the input rate (Codex's input total already includes its cached tokens, so + * they are split out, not added). Only for a model that already has a price. + * Otherwise undefined: an unknown cost is left unknown, never shown as $0. + */ +export function reportedCost(result: WorkerCliResult, model: string | undefined): ReportedCost | undefined { + const digest = result.transcript?.result; + const cli = digest?.total_cost_usd; + if (typeof cli === 'number' && Number.isFinite(cli) && cli >= 0) { + return { dollars: microDollars(BigInt(Math.round(cli * 1_000_000))), source: 'cli' }; + } + const usage = digest?.usage; + const priceModel = digest?.model ?? model; + if (usage === undefined || priceModel === undefined || !Object.hasOwn(MODEL_PRICING, priceModel)) return undefined; + const counts = [usage.input, usage.output, usage.cache_read, usage.cache_creation].map(n => n ?? 0); + if (!counts.every(n => Number.isSafeInteger(n) && n >= 0)) return undefined; + const [reportedInput, output, cacheRead, cacheCreation] = counts.map(BigInt) as [bigint, bigint, bigint, bigint]; + // Claude reports cache tokens beside `input_tokens`; Codex's `input_tokens` + // already includes its `cached_input_tokens`. Charge each token once. + const input = digest?.provider === 'codex' + ? (reportedInput > cacheRead ? reportedInput - cacheRead : 0n) : reportedInput; + const price = MODEL_PRICING[priceModel]!; + const inRate = BigInt(price.input); + const micro = input * inRate + output * BigInt(price.output) + + (cacheCreation * inRate * 125n) / 100n + (cacheRead * inRate * 10n) / 100n; + return { dollars: microDollars(micro), source: 'priced' }; +} + +/** What an internal effect worker (provider helper, plugin, MCP) reports: no model ran. */ +export const NO_MODEL_COST: ReportedCost = Object.freeze({ dollars: '0.000000', source: 'no_model' }); + +function microDollars(micro: bigint): string { + return `${micro / 1_000_000n}.${String(micro % 1_000_000n).padStart(6, '0')}`; +} diff --git a/packages/sdk/src/run-state.ts b/packages/sdk/src/run-state.ts index a0dacce33..4e2b23653 100644 --- a/packages/sdk/src/run-state.ts +++ b/packages/sdk/src/run-state.ts @@ -10,8 +10,13 @@ // No I/O, no clock: the caller passes `now_ms`, so a test folds a hand-built // journal against a fixed instant and asserts exact elapsed and overdue values. +import { addDollars } from './decimal-dollars.js'; +import { addReportedCost, modelFreeSteps, NO_REPORTED_COST, type ReportedCostTotal } from './reported-cost-total.js'; import type { JournalEvent } from './journal-reader.js'; +export { addDollars } from './decimal-dollars.js'; +export type { ReportedCostTotal } from './reported-cost-total.js'; + export type RunStatus = 'running' | 'parked' | 'completed' | 'failed' | 'cancelling' | 'cancelled'; /** `StepState` variant names from state.rs, snake_case. */ @@ -25,6 +30,7 @@ export interface Spend { dollars_unmetered: boolean; } + export interface StepCounts { total: number; done: number; @@ -91,6 +97,10 @@ export interface StepView { completion_reason: string | null; /** Paths the worker journaled for the last attempt, when its output carried them. */ artifacts: { paths: string[]; journaled: boolean }; + /** Metered charges of this step's attempts: what counted toward the budget. */ + spend: Spend; + /** Actual cost of this step's attempts. Null until an attempt completes. */ + reported_cost: ReportedCostTotal | null; } export interface RunView { @@ -101,6 +111,8 @@ export interface RunView { spawned_at_ms: number; now_ms: number; spend: Spend; + /** Actual cost of this run's own attempts; see {@link ReportedCostTotal}. */ + reported_cost: ReportedCostTotal; counts: StepCounts; steps: StepView[]; } @@ -127,42 +139,6 @@ function text(value: unknown, fallback = ''): string { return typeof value === 'string' ? value : fallback; } -/** - * Sum two decimal dollar strings exactly, at whatever scale they carry — the - * kernel's budget fold (`relayflowd-core/src/state/budget.rs`) aligns every - * fractional digit before adding, and it accepts arbitrary precision. Scaling - * to a fixed six fractional digits dropped a charge of `0.0000001` to zero, so - * `flows status` underreported spend the kernel had totalled exactly. - * - * A value that is not a decimal number is left out of the sum rather than - * counted as zero, and the caller is told by `malformed`. - */ -export function addDollars(left: string, right: string): string { - const parsed = [left, right].map((value) => /^\d+(?:\.\d+)?$/.test(value) ? value : null); - const usable = parsed.filter((value): value is string => value !== null); - if (usable.length === 0) return '0'; - if (usable.length === 1) return normalizeDollars(usable[0]!); - const scale = Math.max(...usable.map((value) => (value.split('.')[1] ?? '').length)); - const scaled = usable.map((value) => { - const [whole, fraction = ''] = value.split('.'); - return BigInt(`${whole}${fraction.padEnd(scale, '0')}`); - }); - return fromScaled(scaled[0]! + scaled[1]!, scale); -} - -/** `12.3400` -> `12.34`, `0.000` -> `0`; the journal's own spelling otherwise. */ -function normalizeDollars(value: string): string { - const [whole, fraction = ''] = value.split('.'); - return fromScaled(BigInt(`${whole}${fraction}`), fraction.length); -} - -function fromScaled(total: bigint, scale: number): string { - if (scale === 0) return String(total); - const digits = String(total).padStart(scale + 1, '0'); - const whole = digits.slice(0, digits.length - scale); - const fraction = digits.slice(digits.length - scale).replace(/0+$/, ''); - return fraction.length === 0 ? whole : `${whole}.${fraction}`; -} /** * Reduce `trajectory_tail.transcript` (the digest flows#491 journals) to the @@ -209,6 +185,7 @@ function addSpend(total: Spend, charge: unknown): Spend { const ZERO_SPEND: Spend = { tokens_in: 0, tokens_out: 0, dollars: '0', dollars_unmetered: false }; + interface StepFold { view: StepView; depends_on: string[]; @@ -221,6 +198,7 @@ function freshStep(id: string, type: string, maxIterations: number | null, depen id, type, state: 'pending', attempt: 0, max_iterations: maxIterations, started_at_ms: null, elapsed_ms: null, lease: null, backoff_until_ms: null, wait: null, last_attempt: null, completion_reason: null, artifacts: { paths: [], journaled: false }, + spend: ZERO_SPEND, reported_cost: null, }, }; } @@ -268,6 +246,8 @@ export function foldRunState(events: readonly JournalEvent[], now_ms: number): R // An authored child run carries its parent's totals so ceilings hold across // the family (state.rs `prior_spend`); the view reports the same running total. let spend = addSpend(ZERO_SPEND, budget['prior_spend']); + let reportedCost = NO_REPORTED_COST; + const modelFree = modelFreeSteps(first); let completion: string | null = null; let cancelRequested = false; @@ -297,6 +277,9 @@ export function foldRunState(events: readonly JournalEvent[], now_ms: number): R const attempt = event.attempt ?? view.attempt; view.attempt = Math.max(view.attempt, attempt); spend = addSpend(spend, payload['budget']); + view.spend = addSpend(view.spend, payload['budget']); + reportedCost = addReportedCost(reportedCost, payload, modelFree.has(view.id)); + view.reported_cost = addReportedCost(view.reported_cost ?? NO_REPORTED_COST, payload, modelFree.has(view.id)); const verification = payload['verification'] !== null && typeof payload['verification'] === 'object' ? payload['verification'] as Payload : undefined; view.last_attempt = { @@ -362,8 +345,12 @@ export function foldRunState(events: readonly JournalEvent[], now_ms: number): R // A new epoch restarts every step (state.rs `apply_epoch`); the // summary then restores the ones it carries forward. for (const step of steps.values()) { - const { id, type, max_iterations } = step.view; + const { id, type, max_iterations, spend: stepSpend, reported_cost: stepCost } = step.view; step.view = freshStep(id, type, max_iterations, step.depends_on).view; + // Rollover keeps the earlier segments' completions in the journal, + // and this fold has already read them: what the step cost stays. + step.view.spend = stepSpend; + step.view.reported_cost = stepCost; } spend = addSpend(ZERO_SPEND, payload['budget_spent']); const done = payload['steps_done'] !== null && typeof payload['steps_done'] === 'object' ? payload['steps_done'] as Payload : {}; @@ -462,6 +449,7 @@ export function foldRunState(events: readonly JournalEvent[], now_ms: number): R spawned_at_ms: first.at_ms, now_ms, spend, + reported_cost: reportedCost, counts, steps: views, }; diff --git a/packages/sdk/src/worker.ts b/packages/sdk/src/worker.ts index b9991bd73..bd913d1b3 100644 --- a/packages/sdk/src/worker.ts +++ b/packages/sdk/src/worker.ts @@ -1,5 +1,6 @@ import { communicationInstruction } from './communication/spec.js'; import { workerSpend } from './worker-spend.js'; +import { reportedCost } from './reported-cost.js'; import type { WorkerCliResult } from './worker-cli.js'; import { EventEmitter } from 'node:events'; import type { JournalClient } from './journal-client.js'; @@ -145,6 +146,7 @@ export class AgentWorker extends EventEmitter { dataDir: this.options.dataDir, resultSchema: spec.verification?.json_schema }) : Promise.resolve({ exit_code: null, stdout_tail: '', stderr_tail: 'agent step has no declared CLI' })); const { result, usage } = workerSpend(completed, effectiveModel); + const cost = reportedCost(completed, effectiveModel); const completionReason = result.exit_code === 0 ? 'success' : 'worker_error'; // Output shape: if the CLI's stdout parses as JSON, promote THAT @@ -195,6 +197,7 @@ export class AgentWorker extends EventEmitter { ...(Object.keys(trajectoryTail).length === 0 ? {} : { trajectory_tail: trajectoryTail }), ...(humanIntervention ? { human_intervention: true } : {}), ...(usage !== undefined ? { usage } : {}), + ...(cost === undefined ? {} : { reported_cost: cost }), started_pins: dispatch.pins, end_pins: dispatch.pins, }, diff --git a/packages/sdk/src/yaml-helper-effect.ts b/packages/sdk/src/yaml-helper-effect.ts index e9ddcc1a4..00e23550a 100644 --- a/packages/sdk/src/yaml-helper-effect.ts +++ b/packages/sdk/src/yaml-helper-effect.ts @@ -8,6 +8,7 @@ import type { StepDispatchEvent } from './protocol.js'; import { atomicJson, readSlackReceipt, receiptPath, slackWriteback } from './slack-writeback.js'; import { invokeHelper, type HelperCall } from './yaml-helpers.js'; import { withWorkerLease } from './worker-lease.js'; +import { NO_MODEL_COST } from './reported-cost.js'; export function helperMount(provider: string, env = process.env): string | undefined { const root = [env.RELAYFILE_MOUNT_PATH, env.WORKSPACE_ROOT, env.WORKFORCE_SANDBOX_ROOT, @@ -73,6 +74,6 @@ export async function completeHelperDispatch( return { ...call, idempotencyKey: `${dispatch.run_id}:${dispatch.step_id}`, receipt }; }); await client.stepComplete(dispatch.run_id, dispatch.step_id, dispatch.attempt, - dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins, + dispatch.idempotency_key, 'success', { output, started_pins: dispatch.pins, end_pins: dispatch.pins, reported_cost: NO_MODEL_COST, effects: [{ surface_path: surfacePath, idempotency_key: dispatch.idempotency_key }] }); } diff --git a/packages/sdk/src/yaml-helpers.ts b/packages/sdk/src/yaml-helpers.ts index 0d9b6339a..d96c75b07 100644 --- a/packages/sdk/src/yaml-helpers.ts +++ b/packages/sdk/src/yaml-helpers.ts @@ -3,6 +3,7 @@ import { githubClient, linearClient, slackClient, type RelayTransport } from '@r import type { AgentStepSpec, KernelAgentStep, YamlHelperParams } from './spec.js'; import { canonicalize } from './canonical.js'; import { unknownKeyErrors } from './unknown-keys.js'; +import { HELPER_INSTRUCTION_PREFIX } from './helper-instruction.js'; const string = { type: 'string', minLength: 1 }; const text = { type: 'string' }; @@ -61,7 +62,7 @@ export type HelperCall = { [P in keyof YamlHelperParams]: { [V in keyof YamlHelp { type: 'effect'; provider: P; verb: V; params: YamlHelperParams[P][V] } }[keyof YamlHelperParams[P]] }[keyof YamlHelperParams]; -const prefix = 'relayflows:helper:v1\n'; +const prefix = HELPER_INSTRUCTION_PREFIX; const commonFields = ['id', 'dependsOn', 'maxIterations', 'verification', 'output'] as const; const isObject = (value: unknown): value is Record => value !== null && typeof value === 'object' && !Array.isArray(value); diff --git a/packages/sdk/tests/budget-unmetered-live.test.ts b/packages/sdk/tests/budget-unmetered-live.test.ts index 641c6c0a0..c13f55b2a 100644 --- a/packages/sdk/tests/budget-unmetered-live.test.ts +++ b/packages/sdk/tests/budget-unmetered-live.test.ts @@ -91,6 +91,44 @@ describe('unmetered budget spend through the live kernel', () => { expect(failures).toEqual([]); }); + it('journals an unpriced claude step\'s reported cost beside unchanged metered usage, and the verdict is unchanged', async () => { + const { fixture, client, failures } = await setup(); + const claude = join(fixture.root, 'claude'); + // A model with no frozen price: metered dollars stay unknown, exactly as before. + writeFileSync(join(fixture.root, 'flows.json'), JSON.stringify({ models: ['claude-unpriced-9'] })); + writeFileSync(claude, `#!/usr/bin/env node +if (process.argv[2] === 'auth') process.exit(0); +if (process.argv.includes('Reply with exactly RELAYFLOWS_MODEL_READY and nothing else.')) { + process.stdout.write('RELAYFLOWS_MODEL_READY\\n'); process.exit(0); +} +process.stdout.write(JSON.stringify({ type: 'result', result: JSON.stringify({ message: 'reported-ok' }), total_cost_usd: 12.5, + usage: { input_tokens: 40, output_tokens: 9, cache_read_input_tokens: 900000, cache_creation_input_tokens: 50000 } }) + '\\n'); +`); + chmodSync(claude, 0o755); + const runIds: string[] = []; + // A dollar cap the reported $12.50 would cross many times over. + const handle = flow('reported-cost', { budget: '$0.000001/run' }, async f => { + expect(await f.llm('Answer.', { output: schema, cli: claude, model: 'claude-unpriced-9' })).toEqual({ message: 'reported-ok' }); + await f.run('printf after-reported'); + f.done('success'); + }); + const start = client.runStart.bind(client); + client.runStart = (async (...args: Parameters) => { + const outcome = await start(...args); + runIds.push(outcome.run_id); + return outcome; + }) as JournalClient['runStart']; + const result = await executeAuthoredFlow(handle, client, undefined, { flowPath: fixture.flowPath }); + // Same verdict as any unmetered step: the run is not stopped on dollars. + expect(result.completionReason).toBe('success'); + const completed = (await entries(client, runIds[0]!)).find(e => e.entry_type === 'step.completed')!; + expect(completed.payload.reported_cost).toEqual({ dollars: '12.500000', source: 'cli' }); + // Metered usage is exactly what it was without reported_cost. + expect(completed.payload.budget).toEqual({ tokens_in: 40, tokens_out: 9, dollars: '0', dollars_unmetered: true }); + expect(completed.payload.spend).toMatchObject({ dollars: 0, dollars_unmetered: true }); + expect(failures).toEqual([]); + }); + it('still counts an unpriced step toward a token budget', async () => { const { fixture, client, failures } = await setup(); const handle = flow('unmetered-tokens', { budget: { tokens: 100, dollars: 1 } }, async f => { diff --git a/packages/sdk/tests/cli-status.test.ts b/packages/sdk/tests/cli-status.test.ts index 42443cf69..780a61b27 100644 --- a/packages/sdk/tests/cli-status.test.ts +++ b/packages/sdk/tests/cli-status.test.ts @@ -112,12 +112,15 @@ describe('flows status', () => { v: 1, run_id: RUN_ID, name: 'journal-close-repro', status: 'completed', completion_reason: 'success', spawned_at_ms: EVENTS[0]!.at_ms, now_ms: T_END + 60_000, this_step: null, partial: [], spend: { tokens_in: 0, tokens_out: 0, dollars: '0', dollars_unmetered: false }, + // This journal predates reported_cost: its agent step reported nothing, + // so the total is a lower bound rather than a measured $0. + reported_cost: { dollars: '0', complete: false, source: null }, counts: { total: 3, done: 3, running: 0, pending: 0, backoff: 0, waiting: 0, needs_human: 0 }, }); expect(view.steps).toHaveLength(3); expect(Object.keys(view.steps[0]).sort()).toEqual([ 'artifacts', 'attempt', 'backoff_until_ms', 'completion_reason', 'elapsed_ms', 'id', - 'last_attempt', 'lease', 'max_iterations', 'started_at_ms', 'state', 'tails', 'type', 'wait', + 'last_attempt', 'lease', 'max_iterations', 'reported_cost', 'spend', 'started_at_ms', 'state', 'tails', 'type', 'wait', ]); expect(view.steps[0].tails).toBeNull(); // A run that journals no authored-step index keeps its original shape. diff --git a/packages/sdk/tests/effect-channel.test.ts b/packages/sdk/tests/effect-channel.test.ts index 330bfd03e..122c01350 100644 --- a/packages/sdk/tests/effect-channel.test.ts +++ b/packages/sdk/tests/effect-channel.test.ts @@ -92,12 +92,14 @@ it.each(['none', 'record', 'confirm', 'complete'] as const)('journals one post a expect(dm).toHaveBeenCalledTimes(boundary === 'confirm' ? 2 : 1); if (boundary === 'confirm') expect(dm.mock.calls[0]).toEqual(dm.mock.calls[1]); const entries = (await active.client.journalRead(run.run_id, 1)).entries as Array<{ - entry_type: string; payload: { output?: { messageId: string }; completionReason?: string }; + entry_type: string; payload: { output?: { messageId: string }; completionReason?: string; reported_cost?: unknown }; }>; expect(entries.filter(e => e.entry_type === 'effect.confirmed')).toHaveLength(1); const completed = entries.filter(e => e.entry_type === 'step.completed' && e.payload.completionReason === 'success'); expect(completed).toHaveLength(1); expect(completed[0]!.payload.completionReason).toBe('success'); expect(completed[0]!.payload.output!.messageId).toBe([...delivered.keys()][0]); + // A channel post is a broker effect: no model ran, and the completion says so. + expect(completed[0]!.payload.reported_cost).toEqual({ dollars: '0.000000', source: 'no_model' }); expect(JSON.stringify(entries)).not.toContain('must-not-leak'); }, 15_000); diff --git a/packages/sdk/tests/journal-client-completion.test.ts b/packages/sdk/tests/journal-client-completion.test.ts index 286ca7c11..03e69f5ed 100644 --- a/packages/sdk/tests/journal-client-completion.test.ts +++ b/packages/sdk/tests/journal-client-completion.test.ts @@ -56,3 +56,33 @@ it.each(['rejection', 'disconnect', 'caller close'] as const)( } }, ); + +it.each([ + ['a daemon that predates the feature', undefined, false], + ['a daemon that advertises it', ['reported_cost'], true], +] as const)('sends reported_cost only to %s', async (_name, features, sent) => { + const path = sockPath(); + const received: Record[] = []; + const server = startLoopback(path, { + hello: ctx => sendResult(ctx, { protocol: 0, server: 'relayflowd', ...(features === undefined ? {} : { features }) }), + 'step.complete': (ctx, params) => { received.push(params); sendResult(ctx, { status: 'running' }); }, + }); + const client = new JournalClient(path); + try { + await client.connect(); + await client.hello('feature-gate'); + await client.stepComplete('run', 'agent', 1, 'key', 'success', { + usage: { tokens_in: 1, tokens_out: 1, dollars: '0.1' }, + reported_cost: { dollars: '7.169405', source: 'cli' }, + }); + // An older daemon refuses unknown completion fields; the display-only cost + // is dropped for it rather than failing the step. Metered usage is always sent. + expect(received[0]!['usage']).toEqual({ tokens_in: 1, tokens_out: 1, dollars: '0.1' }); + if (sent) expect(received[0]!['reported_cost']).toEqual({ dollars: '7.169405', source: 'cli' }); + else expect(received[0]).not.toHaveProperty('reported_cost'); + } finally { + client.close(); + await new Promise(resolve => server.close(() => resolve())); + rmSync(path, { force: true }); + } +}); diff --git a/packages/sdk/tests/live-kernel.test.ts b/packages/sdk/tests/live-kernel.test.ts index bb4dee569..6f5865e45 100644 --- a/packages/sdk/tests/live-kernel.test.ts +++ b/packages/sdk/tests/live-kernel.test.ts @@ -1487,7 +1487,7 @@ describe('JournalClient wire conformance against live relayflowd', () => { await startDaemon(dataDir); const client = await connectClient(dataDir); - expect(await client.hello('live-conformance')).toEqual({ protocol: 0, server: 'relayflowd' }); + expect(await client.hello('live-conformance')).toEqual({ protocol: 0, server: 'relayflowd', features: ['reported_cost'] }); const deterministic = await client.runStart(toKernelSpec(compileYaml(` version: '0.1.0' steps: diff --git a/packages/sdk/tests/reported-cost.test.ts b/packages/sdk/tests/reported-cost.test.ts new file mode 100644 index 000000000..a6ec8968d --- /dev/null +++ b/packages/sdk/tests/reported-cost.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from 'vitest'; +import { reportedCost } from '../src/reported-cost.js'; +import { workerSpend } from '../src/worker-spend.js'; +import type { WorkerCliResult } from '../src/worker-cli.js'; + +const base = { exit_code: 0, stdout_tail: '', stderr_tail: '' } as const; +const withDigest = (result: NonNullable['result'], tokens?: { in: number; out: number }): WorkerCliResult => ({ + ...base, ...(tokens === undefined ? {} : { tokens_input: tokens.in, tokens_output: tokens.out }), + transcript: { result }, +}); + +describe('reportedCost', () => { + it('prefers the CLI\'s own total, at six decimal places', () => { + expect(reportedCost(withDigest({ provider: 'claude', total_cost_usd: 7.1694050000000025 }), 'claude-opus-5')) + .toEqual({ dollars: '7.169405', source: 'cli' }); + }); + + it('estimates a priced model from full usage, including cache tokens the metered charge leaves out', () => { + const result = withDigest({ provider: 'claude', model: 'claude-opus-5', + usage: { input: 166, output: 38_861, cache_read: 2_000_000, cache_creation: 100_000 } }, { in: 166, out: 38_861 }); + // 166*$5/M + 38861*$25/M + 100000*$5/M*1.25 + 2000000*$5/M*0.1 + expect(reportedCost(result, 'claude-opus-5')).toEqual({ dollars: '2.597355', source: 'priced' }); + // Enforcement is untouched: the metered charge still prices input and output only. + expect(workerSpend(result, 'claude-opus-5').usage).toEqual({ tokens_in: 166, tokens_out: 38_861, dollars: '0.972355' }); + }); + + it('charges a Codex step\'s cached input once: its input total already includes it', () => { + const result = withDigest({ provider: 'codex', model: 'codex-medium', usage: { input: 1_000_000, output: 0, cache_read: 800_000 } }); + // 200000 fresh * $2/M + 800000 cached * $2/M * 0.1 + expect(reportedCost(result, 'codex-medium')).toEqual({ dollars: '0.560000', source: 'priced' }); + }); + + it('leaves an unknown cost unknown rather than reporting $0', () => { + expect(reportedCost({ ...base }, 'claude-opus-5')).toBeUndefined(); + expect(reportedCost(withDigest({ provider: 'codex', usage: { input: 10, output: 10 } }), 'gpt-unpriced')).toBeUndefined(); + expect(reportedCost(withDigest({ provider: 'claude', total_cost_usd: Number.NaN }), undefined)).toBeUndefined(); + }); +}); diff --git a/packages/sdk/tests/run-state.test.ts b/packages/sdk/tests/run-state.test.ts index e45d07dbd..4641e1a16 100644 --- a/packages/sdk/tests/run-state.test.ts +++ b/packages/sdk/tests/run-state.test.ts @@ -298,6 +298,113 @@ describe('foldRunState', () => { expect(fetch.elapsed_ms).toBe(10_000); }); + it('shows each step\'s metered spend and actual reported cost, and the run total, without charging it', () => { + const view = foldRunState(journal( + spawned, + completed('fetch', 1, T0 + 1000, {}), + completed('analyze', 1, T0 + 2000, { + completionReason: 'verification_failed', disposition: 'retry', next_attempt_at_ms: T0 + 3000, + budget: { tokens_in: 100, tokens_out: 10, dollars: '0.001000' }, + reported_cost: { dollars: '4.500000', source: 'cli' }, + }), + completed('analyze', 2, T0 + 4000, { + budget: { tokens_in: 100, tokens_out: 10, dollars: '0.001000' }, + reported_cost: { dollars: '2.669405', source: 'cli' }, + }), + completed('post', 1, T0 + 5000, { budget: { tokens_in: 5, tokens_out: 5, dollars_unmetered: true, dollars: '0' } }), + ), T0 + 6000); + const [fetch, analyze, post] = view.steps; + // A shell step costs nothing and does not make the total a lower bound. + expect(fetch!.reported_cost).toEqual({ dollars: '0', complete: true, source: null }); + expect(analyze!.spend).toEqual({ tokens_in: 200, tokens_out: 20, dollars: '0.002', dollars_unmetered: false }); + expect(analyze!.reported_cost).toEqual({ dollars: '7.169405', complete: true, source: 'cli' }); + // A model step with no reported cost (Codex, older runtime) is unknown, not $0. + expect(post!.reported_cost).toEqual({ dollars: '0', complete: false, source: null }); + expect(view.reported_cost).toEqual({ dollars: '7.169405', complete: false, source: 'cli' }); + // The budget total is the metered charges alone. + expect(view.spend.dollars).toBe('0.002'); + }); + + it('counts only steps known to run no model as free: shell steps and memoized reuse', () => { + const view = foldRunState(journal( + spawned, + completed('fetch', 1, T0 + 1000, {}), + // A memoized completion copies the source's output; the source run paid. + completed('analyze', 1, T0 + 2000, { + reused_from: { run_id: 'old', step_id: 'analyze', seq: 7 }, completed_by: 'kernel:reuse', + }), + ), T0 + 3000); + expect(view.steps.slice(0, 2).map((step) => step.reported_cost)).toEqual([ + { dollars: '0', complete: true, source: null }, + { dollars: '0', complete: true, source: null }, + ]); + expect(view.reported_cost).toEqual({ dollars: '0', complete: true, source: null }); + }); + + it('counts a provider helper step as free: it performs a relayfile write, not a model call', () => { + const helperSpec = { name: 'helpers', steps: [{ + id: 'post', type: 'agent', depends_on: [], max_iterations: 1, + instruction: 'relayflows:helper:v1\n{"type":"effect","provider":"slack","verb":"postMessage","params":{}}', + }] }; + const view = foldRunState(journal( + { entry_type: 'run.spawned', payload: { spec: helperSpec, spec_hash: 'h', parent_run_id: null, journal_version: 1, created_by: 't' } }, + completed('post', 1, T0 + 1000, {}), + ), T0 + 2000); + expect(view.reported_cost).toEqual({ dollars: '0', complete: true, source: null }); + }); + + it('counts an internal effect worker\'s no_model report as a known $0 that names no cost source', () => { + const view = foldRunState(journal( + spawned, + completed('fetch', 1, T0 + 1000, {}), + completed('analyze', 1, T0 + 2000, { reported_cost: { dollars: '1.500000', source: 'cli' } }), + completed('post', 1, T0 + 3000, { reported_cost: { dollars: '0.000000', source: 'no_model' } }), + ), T0 + 4000); + expect(view.steps[2]!.reported_cost).toEqual({ dollars: '0', complete: true, source: null }); + expect(view.reported_cost).toEqual({ dollars: '1.5', complete: true, source: 'cli' }); + }); + + it('does not take a zero budget as proof no model ran: a recovered agent attempt stays unknown', () => { + // Crash recovery and cancellation journal a default budget although the model ran. + const view = foldRunState(journal( + spawned, + completed('fetch', 1, T0 + 1000, {}), + completed('analyze', 1, T0 + 2000, { completionReason: 'crashed', disposition: 'retry', next_attempt_at_ms: T0 + 3000, completed_by: 'kernel' }), + ), T0 + 4000); + expect(view.steps[1]!.reported_cost).toEqual({ dollars: '0', complete: false, source: null }); + expect(view.reported_cost.complete).toBe(false); + }); + + it('counts a real authored root as free, but not an ordinary step merely named authored-root', () => { + const rootView = (instruction?: string) => foldRunState(journal( + { entry_type: 'run.spawned', payload: { spec: { name: 'authored', steps: [{ + id: 'authored-root', type: 'agent', depends_on: [], max_iterations: 1, ...(instruction === undefined ? {} : { instruction }), + }] }, spec_hash: 'h', parent_run_id: null, journal_version: 1, created_by: 't' } }, + completed('authored-root', 1, T0 + 1000, {}), + ), T0 + 2000); + // Its model calls are child runs with journals of their own. + expect(rootView(JSON.stringify({ kind: 'relayflows.authored-root.v1' })).reported_cost) + .toEqual({ dollars: '0', complete: true, source: null }); + // Without the discriminator it is an ordinary agent step: its cost is unknown. + expect(rootView().reported_cost).toEqual({ dollars: '0', complete: false, source: null }); + expect(rootView('Do the work.').reported_cost).toEqual({ dollars: '0', complete: false, source: null }); + }); + + it('keeps each step\'s cost across an epoch summary, since rollover keeps the earlier completions', () => { + const view = foldRunState(journal( + spawned, + completed('analyze', 1, T0 + 1000, { + budget: { tokens_in: 3, tokens_out: 4, dollars: '0.5' }, reported_cost: { dollars: '1.000000', source: 'priced' }, + }), + { entry_type: 'epoch.summary', payload: { epoch: 1, budget_spent: { tokens_in: 3, tokens_out: 4, dollars: '0.5' }, + steps_done: { analyze: { completionReason: 'success' } }, steps_open: {} } }, + ), T0 + 3000); + const analyze = view.steps.find((step) => step.id === 'analyze')!; + expect(analyze.reported_cost).toEqual({ dollars: '1', complete: true, source: 'priced' }); + expect(analyze.spend.dollars).toBe('0.5'); + expect(view.reported_cost).toEqual({ dollars: '1', complete: true, source: 'priced' }); + }); + it('adds decimal dollars at full precision, beyond six fractional digits', () => { // Microdollar scaling returned null past six digits and the nullish // fallback contributed zero, so `flows status` underreported spend.