Skip to content
Merged
47 changes: 47 additions & 0 deletions kernel/relayflowd-core/src/entry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -391,10 +391,35 @@ pub struct StepCompletedPayload {
pub trajectory_tail: Option<Value>,
#[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<ReportedCost>,
pub completed_by: String,
pub next_attempt_at_ms: Option<i64>,
}

/// 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,
Expand Down Expand Up @@ -691,3 +716,25 @@ pub struct RunCompletedPayload {
pub failed_step_id: Option<String>,
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"]);
}
}
4 changes: 4 additions & 0 deletions kernel/relayflowd-core/src/machine.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@ pub struct AttemptResult {
pub end_pins: Option<Pins>,
pub effects: Vec<EffectRef>,
pub trajectory_tail: Option<Value>,
/// Display-only actual cost; journaled beside `budget`, never charged.
pub reported_cost: Option<crate::entry::ReportedCost>,
/// Execution failures bypass verification but still follow retry policy.
pub failure_reason: Option<CompletionReason>,
/// Why the attempt was rejected, in the vocabulary of whoever rejected it.
Expand All @@ -76,6 +78,7 @@ impl AttemptResult {
end_pins: None,
effects: Vec::new(),
trajectory_tail: None,
reported_cost: None,
failure_reason: None,
failure_detail: None,
}
Expand Down Expand Up @@ -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)];
Expand Down
1 change: 1 addition & 0 deletions kernel/relayflowd-core/src/machine/cancel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ pub(super) fn cancel_run_actions(state: &RunState, now_ms: i64) -> Vec<Action> {
budget: Budget::default(),
completed_by: "kernel".to_owned(),
next_attempt_at_ms: None,
reported_cost: None,
},
)));
}
Expand Down
1 change: 1 addition & 0 deletions kernel/relayflowd-core/src/machine/parallel_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down
1 change: 1 addition & 0 deletions kernel/relayflowd-core/src/machine/recovery.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
3 changes: 3 additions & 0 deletions kernel/relayflowd-core/src/machine/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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()
Expand Down
3 changes: 3 additions & 0 deletions kernel/relayflowd-core/src/memoization.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
1 change: 1 addition & 0 deletions kernel/relayflowd-core/src/state/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
3 changes: 3 additions & 0 deletions kernel/relayflowd-core/tests/memoization.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand All @@ -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());
}
Expand Down
17 changes: 17 additions & 0 deletions kernel/relayflowd/src/engine/remote.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ pub struct OutOfBandCompletion {
pub end_pins: Option<Pins>,
pub effects: Vec<EffectRef>,
pub trajectory_tail: Option<Value>,
/// Display-only actual cost. Journaled with the completion; never charged.
pub reported_cost: Option<relayflowd_core::ReportedCost>,
}

/// A human question a leased attempt parks on. The worker names the wait so
Expand Down Expand Up @@ -136,6 +138,20 @@ impl Engine<WallClock> {
{
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");
Comment thread
khaliqgant marked this conversation as resolved.
}
// `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())?;
Expand Down Expand Up @@ -238,6 +254,7 @@ impl Engine<WallClock> {
trajectory_tail: completion.trajectory_tail,
failure_reason,
failure_detail,
reported_cost: completion.reported_cost,
};
for action in completion_actions(
run_id,
Expand Down
2 changes: 2 additions & 0 deletions kernel/relayflowd/src/exec_det.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}
}

Expand Down Expand Up @@ -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,
}
}

Expand Down
10 changes: 9 additions & 1 deletion kernel/relayflowd/src/server.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)?;
Expand Down Expand Up @@ -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)?;
Expand Down
2 changes: 2 additions & 0 deletions kernel/relayflowd/src/server/wire.rs
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,8 @@ pub(super) struct StepCompleteParams {
pub effects: Vec<EffectRef>,
#[serde(default)]
pub trajectory_tail: Option<Value>,
#[serde(default)]
pub reported_cost: Option<relayflowd_core::ReportedCost>,
}

#[derive(Deserialize)]
Expand Down
Loading
Loading