Merge pull request #6604 from Andrea-Bruno/feat/superfast-decision-gate

Verified original PR #6604 at 88b0b6f22f: exact-head Linux, macOS and Windows tests and doctests, required checks and resolved review threads. Contributor authorship and existing history are preserved.

Signed-off-by: CodeWhale Bot <[email protected]>
This commit is contained in:
Hunter Bown authored and GitHub committed 2026-09-30 12:33:30 -07:00
commit 06436c0543
21 files changed
+2688 -107

No files matched your search

+8
View File
@@ -59,6 +59,7 @@ quieter, and Fleet runs can be checked before they spend anything.
### Contributors
- **[@Andrea-Bruno](https://github.com/Andrea-Bruno)** — designed the Superfast Decision Gate and contributed its off-by-default shadow classifier ([#6604](https://github.com/Hmbown/Codewhale/pull/6604), [#6603](https://github.com/Hmbown/Codewhale/issues/6603)).
- **[@aiapienthusiast](https://github.com/aiapienthusiast)** — added Cheaper Inference to the bundled provider catalog ([#6761](https://github.com/Hmbown/Codewhale/pull/6761)).
- **[@gaord](https://github.com/gaord)** — let a client fork a thread at a named turn ([#6580](https://github.com/Hmbown/Codewhale/pull/6580)), let undo roll back files for the turn it is undoing ([#6483](https://github.com/Hmbown/Codewhale/pull/6483)), stopped resume and fork from duplicating threads and sessions ([#6406](https://github.com/Hmbown/Codewhale/pull/6406)), exposed user-defined provider routes to native clients ([#6404](https://github.com/Hmbown/Codewhale/pull/6404)), and kept a fork going when a turn lost its tool call ([#6664](https://github.com/Hmbown/Codewhale/pull/6664)).
- **[@Lstarsky0](https://github.com/Lstarsky0)** — moved the docs/work, legal, digest and FAQ pages onto the dictionary spine ([#6405](https://github.com/Hmbown/Codewhale/pull/6405), [#6417](https://github.com/Hmbown/Codewhale/pull/6417), [#6499](https://github.com/Hmbown/Codewhale/pull/6499), [#6574](https://github.com/Hmbown/Codewhale/pull/6574)), tightened the Chinese-branching ceiling to 18 ([#6403](https://github.com/Hmbown/Codewhale/pull/6403)), and made Fleet publish without a two-link window ([#6431](https://github.com/Hmbown/Codewhale/pull/6431)). Also moved the constitution page onto the dictionary spine and kept its install link in the selected locale ([#6733](https://github.com/Hmbown/Codewhale/pull/6733)).
@@ -72,6 +73,13 @@ quieter, and Fleet runs can be checked before they spend anything.
### Added
- An optional, off-by-default shadow Decision Gate classifies the first request
of a turn — tool need, answerability from context, intent — through the
existing System One client and only logs a typed recommendation; it never
changes routing or delays the model call. Configure it with
`SUPERFAST_ENABLED` and `SUPERFAST_PROVIDER` (docs/CONFIGURATION.md)
([#6603](https://github.com/Hmbown/Codewhale/issues/6603),
[#6604](https://github.com/Hmbown/Codewhale/pull/6604), thanks @Andrea-Bruno).
- `tool_call_after` hooks for shell tools receive `DEEPSEEK_TOOL_EXECUTION_RECEIPT`: the command that actually ran after admission, its working directory, how it ended, and bounded stdout/stderr previews, so a hook can record exactly what executed ([#6689](https://github.com/Hmbown/Codewhale/issues/6689), requested by [@wuisabel-gif](https://github.com/wuisabel-gif)).
- Runtime API: turns now record what they produced. Each item and turn
carries typed artifact references (path, kind, size, revision, and a
+8
View File
@@ -59,6 +59,7 @@ quieter, and Fleet runs can be checked before they spend anything.
### Contributors
- **[@Andrea-Bruno](https://github.com/Andrea-Bruno)** — designed the Superfast Decision Gate and contributed its off-by-default shadow classifier ([#6604](https://github.com/Hmbown/Codewhale/pull/6604), [#6603](https://github.com/Hmbown/Codewhale/issues/6603)).
- **[@aiapienthusiast](https://github.com/aiapienthusiast)** — added Cheaper Inference to the bundled provider catalog ([#6761](https://github.com/Hmbown/Codewhale/pull/6761)).
- **[@gaord](https://github.com/gaord)** — let a client fork a thread at a named turn ([#6580](https://github.com/Hmbown/Codewhale/pull/6580)), let undo roll back files for the turn it is undoing ([#6483](https://github.com/Hmbown/Codewhale/pull/6483)), stopped resume and fork from duplicating threads and sessions ([#6406](https://github.com/Hmbown/Codewhale/pull/6406)), exposed user-defined provider routes to native clients ([#6404](https://github.com/Hmbown/Codewhale/pull/6404)), and kept a fork going when a turn lost its tool call ([#6664](https://github.com/Hmbown/Codewhale/pull/6664)).
- **[@Lstarsky0](https://github.com/Lstarsky0)** — moved the docs/work, legal, digest and FAQ pages onto the dictionary spine ([#6405](https://github.com/Hmbown/Codewhale/pull/6405), [#6417](https://github.com/Hmbown/Codewhale/pull/6417), [#6499](https://github.com/Hmbown/Codewhale/pull/6499), [#6574](https://github.com/Hmbown/Codewhale/pull/6574)), tightened the Chinese-branching ceiling to 18 ([#6403](https://github.com/Hmbown/Codewhale/pull/6403)), and made Fleet publish without a two-link window ([#6431](https://github.com/Hmbown/Codewhale/pull/6431)). Also moved the constitution page onto the dictionary spine and kept its install link in the selected locale ([#6733](https://github.com/Hmbown/Codewhale/pull/6733)).
@@ -72,6 +73,13 @@ quieter, and Fleet runs can be checked before they spend anything.
### Added
- An optional, off-by-default shadow Decision Gate classifies the first request
of a turn — tool need, answerability from context, intent — through the
existing System One client and only logs a typed recommendation; it never
changes routing or delays the model call. Configure it with
`SUPERFAST_ENABLED` and `SUPERFAST_PROVIDER` (docs/CONFIGURATION.md)
([#6603](https://github.com/Hmbown/Codewhale/issues/6603),
[#6604](https://github.com/Hmbown/Codewhale/pull/6604), thanks @Andrea-Bruno).
- `tool_call_after` hooks for shell tools receive `DEEPSEEK_TOOL_EXECUTION_RECEIPT`: the command that actually ran after admission, its working directory, how it ended, and bounded stdout/stderr previews, so a hook can record exactly what executed ([#6689](https://github.com/Hmbown/Codewhale/issues/6689), requested by [@wuisabel-gif](https://github.com/wuisabel-gif)).
- Runtime API: turns now record what they produced. Each item and turn
carries typed artifact references (path, kind, size, revision, and a
+499 -29
View File
@@ -2,9 +2,9 @@
//!
//! `[auto.router] kind = "decision"` asks a non-generative decision model
//! (TypeSafe's Jev) one typed Choice per turn. The wire is a plain JSON
//! `POST {base}/systemone` — not chat completions — served on two routes:
//! `POST to the route’s decision endpoint` — not chat completions — served on two routes:
//!
//! * **OpenRouter** — `{openrouter base}/systemone` with the user's OpenRouter
//! * **OpenRouter** — `https://openrouter.ai/api/alpha/decisions` with the user's OpenRouter
//! key, built exactly like every other OpenRouter client.
//! * **TypeSafe direct** — `https://api.typesafe.ai/v1/systemone` with a
//! TypeSafe key.
@@ -16,12 +16,15 @@
//! * TypeSafe is **not** an [`ApiProvider`]: it serves no chat route, so it is
//! the decision router's own endpoint + key (`TYPESAFE_API_KEY`, the
//! `typesafe` secret-store slot, or `[providers.typesafe] api_key` /
//! `api_key_env`). Its spend is shown on the Auto receipt as the
//! provider-reported cost; it is not entered in session cost totals.
//! `api_key_env`). Its tokens enter the shared usage ledger under a frozen
//! `custom` / `typesafe` route with unknown billing; reported cost is retained
//! on decision receipts. Unknown pricing is never interpreted as free.
//! * The routing call makes one attempt (no retry): it is bounded by the
//! router timeout, and a retried decision would arrive after the turn has
//! already fallen back.
//! * Only the `choice` answer shape is parsed; `score` / `noul` are later.
//! * The router parses the `choice` answer shape; the shadow Decision Gate
//! (`crate::superfast`) also reads `noul`. The transport strictly validates
//! Choice, Noul and fractional Score responses before either policy uses them.
use std::collections::BTreeMap;
@@ -37,6 +40,9 @@ pub(crate) const TYPESAFE_DEFAULT_BASE_URL: &str = "https://api.typesafe.ai/v1";
pub(crate) const TYPESAFE_API_KEY_ENV: &str = "TYPESAFE_API_KEY";
/// Secret-store slot and `[providers.<name>]` table name for the TypeSafe key.
pub(crate) const TYPESAFE_KEY_NAME: &str = "typesafe";
/// Bound a non-generative decision response before allocating/decoding it.
const DECISION_RESPONSE_MAX_BYTES: usize = 256 * 1024;
const DECISION_REQUEST_MAX_BYTES: usize = 1024 * 1024;
/// Which endpoint serves a decision router.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -117,25 +123,76 @@ pub(crate) fn typesafe_api_key(config: &Config) -> Option<String> {
}
/// A decoded System One response. Unknown fields are ignored.
#[derive(Debug, Clone, Deserialize)]
#[derive(Debug, Clone)]
pub(crate) struct SystemOneResponse {
#[serde(default)]
pub(crate) id: Option<String>,
#[serde(default)]
pub(crate) model: Option<String>,
#[serde(default)]
pub(crate) answers: BTreeMap<String, SystemOneAnswer>,
#[serde(default)]
pub(crate) usage: Option<SystemOneUsage>,
/// Transport validation is separate from decoding so a rejected policy
/// answer still preserves the provider's usage/cost evidence.
pub(crate) answers_validated: Option<bool>,
}
/// One answer. Only the `choice` subset is interpreted.
#[derive(Debug, Clone, Deserialize)]
impl<'de> Deserialize<'de> for SystemOneResponse {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
#[derive(Deserialize)]
struct Envelope {
#[serde(default)]
id: Option<Box<RawValue>>,
#[serde(default)]
model: Option<Box<RawValue>>,
#[serde(default)]
answers: Option<Box<RawValue>>,
#[serde(default)]
usage: Option<Box<RawValue>>,
}
let wire = Envelope::deserialize(deserializer)?;
let text = |raw: Option<&RawValue>| {
raw.and_then(|raw| serde_json::from_str::<String>(raw.get()).ok())
};
let id = text(wire.id.as_deref());
let model = text(wire.model.as_deref());
let malformed_identity =
(wire.id.is_some() && id.is_none()) || (wire.model.is_some() && model.is_none());
// Invalid policy fields must fail validation without discarding the
// separately reported billing evidence. Empty/default answers never
// satisfy the required question types or probabilities.
let answers: BTreeMap<String, Box<RawValue>> = wire
.answers
.and_then(|raw| serde_json::from_str(raw.get()).ok())
.unwrap_or_default();
Ok(Self {
id,
model,
answers: answers
.into_iter()
.map(|(key, raw)| (key, serde_json::from_str(raw.get()).unwrap_or_default()))
.collect(),
usage: wire
.usage
.and_then(|raw| serde_json::from_str(raw.get()).ok()),
answers_validated: malformed_identity.then_some(false),
})
}
}
/// One answer. The `choice` and `noul` subsets are interpreted.
#[derive(Debug, Clone, Default, Deserialize)]
pub(crate) struct SystemOneAnswer {
#[serde(rename = "type", default)]
pub(crate) kind: String,
#[serde(default)]
pub(crate) choice: Option<String>,
/// A `noul` answer's probability; validated by its reader.
#[serde(default)]
pub(crate) noul: Option<f64>,
#[serde(default)]
pub(crate) score: Option<f64>,
#[serde(default)]
pub(crate) legend: BTreeMap<String, Value>,
#[serde(default)]
pub(crate) probabilities: BTreeMap<String, Option<f64>>,
#[serde(default)]
@@ -144,15 +201,35 @@ pub(crate) struct SystemOneAnswer {
/// Provider usage. OpenRouter adds `cost`; the token-count casing differs
/// between surfaces, so both spellings are accepted.
#[derive(Debug, Clone, Deserialize)]
#[derive(Debug, Clone)]
pub(crate) struct SystemOneUsage {
/// Kept verbatim so the receipt never re-renders a float.
#[serde(default)]
pub(crate) cost: Option<Box<RawValue>>,
#[serde(default, alias = "inputTokens")]
pub(crate) input_tokens: u32,
#[serde(default, alias = "outputTokens")]
pub(crate) output_tokens: u32,
pub(crate) complete: bool,
}
impl<'de> Deserialize<'de> for SystemOneUsage {
fn deserialize<D: serde::Deserializer<'de>>(
deserializer: D,
) -> std::result::Result<Self, D::Error> {
let mut wire = BTreeMap::<String, Box<RawValue>>::deserialize(deserializer)?;
let count = |snake: &str, camel: &str| match (wire.get(snake), wire.get(camel)) {
(Some(raw), None) | (None, Some(raw)) => serde_json::from_str::<u32>(raw.get()).ok(),
// Missing, malformed or ambiguous counts cannot authorize policy
// or a priced subtotal; keep an independently valid raw cost.
_ => None,
};
let input_tokens = count("input_tokens", "inputTokens");
let output_tokens = count("output_tokens", "outputTokens");
Ok(Self {
complete: input_tokens.is_some() && output_tokens.is_some(),
input_tokens: input_tokens.unwrap_or(0),
output_tokens: output_tokens.unwrap_or(0),
cost: wire.remove("cost"),
})
}
}
impl SystemOneUsage {
@@ -162,7 +239,7 @@ impl SystemOneUsage {
pub(crate) fn reported_cost(&self) -> Option<String> {
let raw = self.cost.as_ref()?.get().trim();
let value: f64 = raw.parse().ok()?;
(value.is_finite() && value >= 0.0).then(|| raw.to_string())
(raw.len() <= 128 && value.is_finite() && value >= 0.0).then(|| raw.to_string())
}
}
@@ -172,8 +249,8 @@ impl CodewhaleClient {
/// OpenRouter is the ordinary OpenRouter client (its key, base URL,
/// attribution headers). TypeSafe re-points a clone of the active route's
/// client — keeping its retry, TLS and redaction policy — at the TypeSafe
/// endpoint with the TypeSafe key only, and drops the active provider's
/// concurrency permit so a routing call never holds a chat slot.
/// endpoint with the TypeSafe key only. Its captured request budget and
/// remote-control ownership continue to apply to the one decision attempt.
pub(crate) fn for_decision_route(
config: &Config,
route: DecisionRouterRoute,
@@ -212,6 +289,15 @@ impl CodewhaleClient {
)?
.build()?;
client.base_url = base_url;
client.http1_client = client.http_client.clone();
// Repointing auth/URL must also replace the inherited chat
// route identity. No TypeSafe price/product is guessed.
client.api_provider = ApiProvider::Custom;
client.provider_identity = TYPESAFE_KEY_NAME.to_string();
client.openrouter_vendor = None;
client.billing_surface = None;
client.billing_mode = crate::cost_status::RouteBillingMode::Unknown;
client.route_limits = None;
// `Self::new` froze the redaction set from the chat provider's
// secrets; the TypeSafe key is none of them, so add it before
// any decision body is built from untrusted context.
@@ -219,15 +305,13 @@ impl CodewhaleClient {
push_model_bound_secret(&mut secrets, Some(&key));
client.model_bound_secret_values = Arc::new(secrets);
client.api_key = key;
client.request_concurrency = None;
client.remote_control_inference_participant = false;
Ok(client)
}
}
}
/// `POST {base}/systemone` once, isolated like the Auto chat classifier:
/// no global retry banners, no shared token bucket, no response cache.
/// `POST to the route’s decision endpoint` once, isolated like the Auto chat classifier:
/// no global retry banners or response cache; shared admission still applies.
///
/// Only a failure class leaves this function — provider error bodies can
/// echo the prompt and must never reach receipts.
@@ -240,26 +324,162 @@ impl CodewhaleClient {
body: &Value,
dispatched: &std::sync::atomic::AtomicBool,
) -> std::result::Result<SystemOneResponse, AutoRouterFailure> {
if serde_json::to_vec(body).map_or(true, |bytes| bytes.len() > DECISION_REQUEST_MAX_BYTES)
|| !valid_decision_request(body)
{
return Err(AutoRouterFailure::NotRunnable);
}
let mut isolated = self.clone();
isolated.isolated_request_state = true;
isolated.rate_limiter = Arc::new(AsyncMutex::new(TokenBucket::from_env()));
isolated.retry.max_retries = 0;
let _inference = isolated.acquire_remote_control_inference_permit().await;
let _permit = isolated.acquire_provider_request_permit().await;
let url = api_url(&isolated.base_url, "systemone");
let url = if isolated.api_provider == ApiProvider::Openrouter {
// The OpenRouter Decisions API is a sibling of /api/v1, so keep
// the configured origin/proxy prefix and replace only /v1.
let mut url = reqwest::Url::parse(&isolated.base_url)
.map_err(|_| AutoRouterFailure::NotRunnable)?;
let prefix = url.path().trim_end_matches('/').trim_end_matches("/v1");
url.set_path(&format!("{prefix}/alpha/decisions"));
url.to_string()
} else {
api_url(&isolated.base_url, "systemone")
};
dispatched.store(true, std::sync::atomic::Ordering::Release);
let response = isolated
.send_json_with_retry(&url, body)
.await
.map_err(|error| router_failure_from_error(&error))?;
let bytes = response
.bytes()
let text = bounded_provider_catalog_text(response, DECISION_RESPONSE_MAX_BYTES)
.await
.map_err(|_| AutoRouterFailure::Transport)?;
serde_json::from_slice(&bytes).map_err(|_| AutoRouterFailure::InvalidAnswer)
.map_err(|error| match error {
CatalogRefreshError::Network => AutoRouterFailure::Transport,
_ => AutoRouterFailure::InvalidAnswer,
})?;
let mut response: SystemOneResponse =
serde_json::from_str(&text).map_err(|_| AutoRouterFailure::InvalidAnswer)?;
response.answers_validated = Some(valid_decision_response(body, &response));
response.model = response.model.map(|model| {
self.redact_model_bound_text(&model)
.chars()
.take(128)
.collect()
});
Ok(response)
}
}
fn valid_decision_request(body: &Value) -> bool {
let Some(questions) = body.get("questions").and_then(Value::as_object) else {
return false;
};
let supported_value =
|value: &Value| matches!(value, Value::String(_) | Value::Object(_) | Value::Array(_));
let model = body.get("model").and_then(Value::as_str);
model.is_some_and(|m| !m.trim().is_empty() && m.len() <= 256)
&& body.get("state").is_some_and(|state| {
matches!(state, Value::String(_) | Value::Object(_) | Value::Array(_))
})
&& !questions.is_empty()
&& questions.len() <= 64
&& questions
.values()
.all(|q| match q.get("type").and_then(Value::as_str) {
Some("noul") => true,
Some("choice") => q
.get("criteria")
.and_then(Value::as_object)
.is_some_and(|c| {
!c.is_empty()
&& c.len() <= 64
&& c.iter().all(|(name, v)| {
!name.trim().is_empty()
&& name.len() <= 128
&& (v.is_null() || supported_value(v))
})
}),
Some("score") => q
.get("criteria")
.and_then(Value::as_array)
.is_some_and(|c| {
!c.is_empty() && c.len() <= 10 && c.iter().all(supported_value)
}),
_ => false,
})
}
fn valid_decision_response(body: &Value, response: &SystemOneResponse) -> bool {
let Some(questions) = body.get("questions").and_then(Value::as_object) else {
return false;
};
response.answers_validated != Some(false)
&& response
.model
.as_deref()
.is_some_and(|m| !m.trim().is_empty() && m.len() <= 256)
&& response.usage.as_ref().is_some_and(|usage| usage.complete)
&& response.answers.len() == questions.len()
&& questions.iter().all(|(name, question)| {
let Some(answer) = response.answers.get(name) else {
return false;
};
if Some(answer.kind.as_str()) != question.get("type").and_then(Value::as_str) {
return false;
}
match answer.kind.as_str() {
"noul" => answer
.noul
.is_some_and(|v| v.is_finite() && (0.0..=1.0).contains(&v)),
"choice" => {
let Some(criteria) = question.get("criteria").and_then(Value::as_object) else {
return false;
};
let options = criteria.keys().map(String::as_str).collect::<Vec<_>>();
crate::model_routing::validated_choice(Some(answer), &options).is_some()
}
"score" => {
let Some(criteria) = question.get("criteria").and_then(Value::as_array) else {
return false;
};
if criteria.is_empty() || criteria.len() > 10 {
return false;
}
let Some(score) = answer.score else {
return false;
};
if !score.is_finite()
|| !(0.0..=(criteria.len() - 1) as f64).contains(&score)
|| !answer
.confidence
.is_some_and(|v| v.is_finite() && (0.0..=1.0).contains(&v))
|| answer.legend.len() != criteria.len()
|| answer.probabilities.len() != criteria.len()
{
return false;
}
let mut sum = 0.0;
let mut expected_score = 0.0;
for (level, criterion) in criteria.iter().enumerate() {
let key = level.to_string();
let Some(Some(probability)) = answer.probabilities.get(&key) else {
return false;
};
if answer.legend.get(&key) != Some(criterion)
|| !probability.is_finite()
|| !(0.0..=1.0).contains(probability)
{
return false;
}
sum += probability;
expected_score += level as f64 * probability;
}
(sum - 1.0).abs() <= 0.02 && (score - expected_score).abs() <= 0.02
}
_ => false,
}
})
}
/// Collapse a client error into a non-secret failure class. The HTTP status
/// is kept where the error carries it; the body never is.
pub(crate) fn router_failure_from_error(error: &anyhow::Error) -> AutoRouterFailure {
@@ -289,3 +509,253 @@ pub(crate) fn router_failure_from_error(error: &anyhow::Error) -> AutoRouterFail
}
}
}
#[cfg(test)]
mod decisions_compatibility_tests {
use super::*;
fn request() -> Value {
json!({"model":"typesafe/jev-1.13", "state":{"ticket":"charged twice"}, "questions": {
"intent": {"type":"choice", "criteria":{"billing":"money", "other":"anything else"}},
"refund": {"type":"noul", "instructions":"Does it ask for a refund?"},
"urgency": {"type":"score", "criteria":["Can wait", "Needs attention this week", "Needs attention today"]}
}})
}
fn response() -> Value {
json!({"id":"fixture-decision", "model":"typesafe/jev-1.13-20260917", "usage":{"input_tokens":287,"output_tokens":20,"cost":0.000012054}, "answers": {
"intent":{"type":"choice","choice":"billing","confidence":0.9,"probabilities":{"billing":0.9,"other":0.1}},
"refund":{"type":"noul","noul":0.99},
"urgency":{"type":"score","score":1.7,"confidence":0.9,"legend":{"0":"Can wait","1":"Needs attention this week","2":"Needs attention today"},"probabilities":{"0":0.1,"1":0.1,"2":0.8}}
}})
}
#[test]
fn all_documented_primitives_validate_without_rescaling_score() {
let decoded: SystemOneResponse =
serde_json::from_value(response()).expect("documented fixture");
assert!(valid_decision_request(&request()));
assert!(valid_decision_response(&request(), &decoded));
assert_eq!(decoded.answers["urgency"].score, Some(1.7));
assert_eq!(
decoded.usage.expect("usage").reported_cost().as_deref(),
Some("0.000012054")
);
}
#[test]
fn wrong_types_unoffered_choices_scores_and_partial_shapes_are_rejected() {
let changes = [
("/answers/refund/noul", json!(1.01)),
("/answers/refund/type", json!("score")),
("/answers/intent/choice", json!("not-offered")),
("/answers/intent/choice", json!("other")),
("/answers/intent/confidence", json!(-0.1)),
("/answers/intent/probabilities", json!({"billing":0.9})),
(
"/answers/intent/probabilities",
json!({"billing":0.6,"other":0.6}),
),
("/answers/urgency/score", json!(2.01)),
("/answers/urgency/score", json!(0.5)),
("/answers/urgency/legend/1", json!("different rubric")),
("/answers/urgency/probabilities/1", Value::Null),
("/answers/urgency/confidence", json!(1.1)),
("/model", Value::Null),
("/usage", Value::Null),
("/usage", json!({"input_tokens": 287, "cost":0.000012054})),
("/answers", json!({})),
];
for (pointer, value) in changes {
let mut body = response();
*body.pointer_mut(pointer).expect("fixture field") = value;
let decoded: SystemOneResponse =
serde_json::from_value(body).expect("structural decode");
assert!(
!valid_decision_response(&request(), &decoded),
"accepted {pointer}"
);
}
}
#[test]
fn non_finite_in_memory_answers_and_unbounded_request_shapes_are_rejected() {
let mut decoded: SystemOneResponse = serde_json::from_value(response()).expect("fixture");
decoded.answers.get_mut("refund").expect("noul").noul = Some(f64::NAN);
assert!(!valid_decision_response(&request(), &decoded));
let mut decoded: SystemOneResponse = serde_json::from_value(response()).expect("fixture");
decoded.answers.get_mut("urgency").expect("score").score = Some(f64::INFINITY);
assert!(!valid_decision_response(&request(), &decoded));
for shape in [
json!({}),
json!({"model":"jev", "state":"hello", "questions":{}}),
json!({"model":"jev", "state":"hello", "questions":{"q":{"type":"score","criteria":vec!["level";11]}}}),
] {
assert!(!valid_decision_request(&shape));
}
}
#[tokio::test]
async fn decision_transport_preserves_typesafe_auth_and_shared_admission_before_dispatch() {
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(header("authorization", "Bearer decision-fixture-key"))
.respond_with(ResponseTemplate::new(200).set_body_json(response()))
.expect(1)
.mount(&server)
.await;
let _lock = crate::test_support::lock_test_env();
let home = tempfile::tempdir().expect("home");
let _home = crate::test_support::EnvVarGuard::set("CODEWHALE_HOME", home.path());
let _key =
crate::test_support::EnvVarGuard::set(TYPESAFE_API_KEY_ENV, "decision-fixture-key");
let config = Config {
provider: Some("deepseek".to_string()),
providers: Some(crate::config::ProvidersConfig {
deepseek: crate::config::ProviderConfig {
api_key: Some("chat-fixture-key".into()),
max_concurrency: Some(1),
..Default::default()
},
..Default::default()
}),
..Default::default()
};
let base = format!("{}/v1", server.uri());
let ticket = crate::test_support::env_scope_ticket();
let client = tokio::task::spawn_blocking(move || {
let _membership = crate::test_support::join_env_scope(ticket);
CodewhaleClient::for_decision_route(&config, DecisionRouterRoute::Typesafe, Some(&base))
})
.await
.expect("worker")
.expect("client");
assert_eq!(client.provider_request_concurrency_limit(), Some(1));
assert!(
client.remote_control_inference_participant,
"the cloned budget must remain an attached inference participant"
);
let clone = client.clone();
let held = client
.acquire_provider_request_permit()
.await
.expect("shared permit");
let dispatched = std::sync::atomic::AtomicBool::new(false);
assert!(
tokio::time::timeout(
Duration::from_millis(30),
clone.system_one_decide(&request(), &dispatched)
)
.await
.is_err()
);
assert!(
!dispatched.load(std::sync::atomic::Ordering::Acquire),
"a queued request has no dispatched-spend marker"
);
assert!(
server
.received_requests()
.await
.expect("requests")
.is_empty()
);
drop(held);
let answer = client
.system_one_decide(&request(), &dispatched)
.await
.expect("one admitted request");
assert!(dispatched.load(std::sync::atomic::Ordering::Acquire));
assert_eq!(answer.answers_validated, Some(true));
assert_eq!(
client
.effective_route_envelope("jev-latest", chrono::Utc::now())
.provider_identity,
"typesafe"
);
}
#[test]
fn decision_partial_usage_retains_raw_cost_but_cannot_authorize_policy() {
let mut body = response();
body["usage"] = json!({"inputTokens": 287, "cost": 0.000012054});
let decoded: SystemOneResponse = serde_json::from_value(body).expect("partial receipt");
assert!(!valid_decision_response(&request(), &decoded));
let usage = decoded.usage.expect("reported usage");
assert!(!usage.complete);
assert_eq!(usage.input_tokens, 287);
assert_eq!(usage.reported_cost().as_deref(), Some("0.000012054"));
let route = crate::cost_status::decision_receipt_fixture("partial").route;
let mut body = response();
body["usage"] = json!({"inputTokens": 287, "cost": 0.000012054});
let response: SystemOneResponse = serde_json::from_value(body).expect("partial response");
let batch = crate::model_routing::decision_usage_batch(&route, &response);
assert!(
batch.records.is_empty(),
"a partial token count cannot produce a complete priced subtotal"
);
assert_eq!(batch.dropped_records, 1);
assert_eq!(batch.drop_records.len(), 1);
let mut invalid = request();
invalid["questions"]["urgency"]["criteria"] = json!([null]);
assert!(!valid_decision_request(&invalid));
}
#[test]
fn malformed_policy_fields_retain_independent_raw_cost() {
for (pointer, value) in [
("/answers/intent/choice", json!(17)),
("/answers/intent/confidence", json!("high")),
("/answers/refund", Value::Null),
("/answers", json!([])),
("/model", json!({"invalid":"model"})),
("/id", json!([])),
] {
let mut body = response();
*body.pointer_mut(pointer).expect("fixture field") = value;
let decoded: SystemOneResponse =
serde_json::from_value(body).expect("billing envelope");
assert!(
!valid_decision_response(&request(), &decoded),
"accepted {pointer}"
);
let usage = decoded
.usage
.expect("independent usage survives rejected policy");
assert!(usage.complete);
assert_eq!(usage.input_tokens, 287);
assert_eq!(usage.output_tokens, 20);
assert_eq!(usage.reported_cost().as_deref(), Some("0.000012054"));
}
}
#[test]
fn malformed_or_overflowed_counters_retain_cost_without_priced_usage() {
for counters in [
json!({"input_tokens": u64::MAX, "output_tokens":20}),
json!({"input_tokens": -1, "output_tokens":20}),
json!({"input_tokens": "287", "output_tokens":20}),
json!({"input_tokens": 287, "inputTokens":287, "output_tokens":20}),
json!({"input_tokens": 287, "output_tokens":1.5}),
] {
let mut body = response();
body["usage"] = counters;
body["usage"]["cost"] = json!(0.000012054);
let decoded: SystemOneResponse =
serde_json::from_value(body).expect("billing envelope");
assert!(!valid_decision_response(&request(), &decoded));
let usage = decoded.usage.as_ref().expect("independent cost");
assert!(!usage.complete);
assert_eq!(usage.reported_cost().as_deref(), Some("0.000012054"));
let route = crate::cost_status::decision_receipt_fixture("malformed-count").route;
let batch = crate::model_routing::decision_usage_batch(&route, &decoded);
assert!(
batch.records.is_empty(),
"invalid counters cannot fabricate priced usage"
);
assert_eq!(batch.dropped_records, 1);
assert_eq!(batch.drop_records.len(), 1);
}
}
}
+8
View File
@@ -5562,6 +5562,14 @@ impl Engine {
);
turn.add_routed_usage_dropped_records(residual_dropped_records);
let initial_cost_scope = crate::cost_status::scope_token();
crate::cost_status::report_runtime_usage_batch(
initial_cost_scope,
initial_usage_owner.as_deref(),
&crate::cost_status::RuntimeUsageBatch {
decisions: initial_routed_usage.decisions.clone(),
..Default::default()
},
);
for record in &initial_routed_usage.records {
crate::cost_status::report_effective_route_for_runtime(
initial_cost_scope,
+1
View File
@@ -8520,6 +8520,7 @@ async fn initial_routed_usage_is_total_only_emitted_once_and_keeps_parent_route_
);
missing_usage_route.billing_mode = crate::cost_status::RouteBillingMode::Metered;
**initial_routed_usage = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![crate::cost_status::RuntimeUsageRecord {
source_id: "auto-router:engine-fixture".to_string(),
usage: crate::cost_status::EffectiveRouteUsage {
+16
View File
@@ -1813,6 +1813,22 @@ impl Engine {
// first call) so we can resend it on a transparent retry below
// when the wire dies before any content was streamed (#103).
let stream_request = request;
// Superfast Decision Gate (shadow mode, off by default). When
// SUPERFAST_ENABLED is set, this spawns a detached task that asks
// a small System One decision model about the user's turn and only
// logs the recommendation. It never changes routing, never skips
// the model call below, and never waits on the decision call.
// Fired only on the first model request of the turn, where the raw
// user message decides intent. See `crate::superfast`.
if turn.step == 0 {
// Detached on purpose: dropping the handle does not cancel it.
drop(crate::superfast::spawn_shadow_gate(
&self.api_config,
&stream_request.messages,
self.config.compaction.runtime_cost_owner.as_deref(),
&self.cancel_token,
));
}
let _ = self
.tx_event
.send(Event::ToolRequestSnapshot {
+356 -5
View File
@@ -488,6 +488,7 @@ pub fn attach_child_usage_metadata(
pub const MAX_CHILD_USAGE_RECORDS: usize = 64;
const CHILD_USAGE_RECORDS_KEY: &str = "child_usage_records";
const CHILD_DECISION_RECEIPTS_KEY: &str = "child_decision_receipts";
const CHILD_USAGE_DROP_RECORDS_KEY: &str = "child_usage_drop_records";
const CHILD_USAGE_DROPPED_RECORDS_KEY: &str = "child_usage_dropped_records";
@@ -506,6 +507,18 @@ pub fn attach_child_usage_batch_metadata(
let Some(object) = metadata.as_object_mut() else {
return;
};
object.insert(
CHILD_DECISION_RECEIPTS_KEY.into(),
serde_json::json!(
batch
.decisions
.iter()
.filter(|receipt| receipt.is_bounded())
.take(MAX_CHILD_USAGE_RECORDS)
.map(RuntimeDecisionReceipt::sanitized)
.collect::<Vec<_>>()
),
);
let retained_records = batch
.records
.iter()
@@ -545,10 +558,23 @@ pub fn attach_child_usage_batch_metadata(
serde_json::json!(retained_drops),
);
let usage_overflow = batch.records.len().saturating_sub(MAX_CHILD_USAGE_RECORDS);
let decision_overflow = batch
.decisions
.len()
.saturating_sub(MAX_CHILD_USAGE_RECORDS)
.saturating_add(
batch
.decisions
.iter()
.take(MAX_CHILD_USAGE_RECORDS)
.filter(|r| !r.is_bounded())
.count(),
);
let dropped_records = batch
.dropped_records
.max(u64::try_from(batch.drop_records.len()).unwrap_or(u64::MAX))
.saturating_add(u64::try_from(usage_overflow).unwrap_or(u64::MAX));
.saturating_add(u64::try_from(usage_overflow).unwrap_or(u64::MAX))
.saturating_add(u64::try_from(decision_overflow).unwrap_or(u64::MAX));
if dropped_records > 0 {
object.insert(
CHILD_USAGE_DROPPED_RECORDS_KEY.into(),
@@ -581,6 +607,7 @@ pub fn child_usage_records_from_metadata(
.unwrap_or(0);
let Some(values) = value.as_array() else {
return Some(RuntimeUsageBatch {
decisions: Vec::new(),
records: Vec::new(),
drop_records: Vec::new(),
dropped_records: declared_dropped.saturating_add(1),
@@ -589,6 +616,7 @@ pub fn child_usage_records_from_metadata(
let overflow = values.len().saturating_sub(MAX_CHILD_USAGE_RECORDS);
let mut batch = RuntimeUsageBatch {
decisions: Vec::new(),
records: Vec::with_capacity(values.len().min(MAX_CHILD_USAGE_RECORDS)),
drop_records: Vec::with_capacity(drop_values.len().min(MAX_CHILD_USAGE_RECORDS)),
dropped_records: declared_dropped
@@ -636,6 +664,27 @@ pub fn child_usage_records_from_metadata(
// Every declared drop slot already contributes to dropped_records,
// including malformed entries; do not count the same gap twice.
}
if let Some(decisions) = metadata.get(CHILD_DECISION_RECEIPTS_KEY) {
if let Some(values) = decisions.as_array() {
for value in values.iter().take(MAX_CHILD_USAGE_RECORDS) {
let bounded = serde_json::to_vec(value).is_ok_and(|bytes| bytes.len() <= 8 * 1024);
let parsed = bounded
.then(|| serde_json::from_value::<RuntimeDecisionReceipt>(value.clone()).ok())
.flatten();
if let Some(receipt) = parsed.filter(RuntimeDecisionReceipt::is_bounded) {
batch.decisions.push(receipt.sanitized());
} else {
batch.dropped_records = batch.dropped_records.saturating_add(1);
}
}
batch.dropped_records = batch.dropped_records.saturating_add(
u64::try_from(values.len().saturating_sub(MAX_CHILD_USAGE_RECORDS))
.unwrap_or(u64::MAX),
);
} else {
batch.dropped_records = batch.dropped_records.saturating_add(1);
}
}
Some(batch)
}
@@ -794,7 +843,7 @@ fn with_pending_state_mut<R>(f: impl FnOnce(&mut ScopedPendingBackgroundCost) ->
.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
.lock()
.unwrap_or_else(|e| e.into_inner());
f(by_thread.entry(std::thread::current().id()).or_default())
f(by_thread.entry(test_cost_scope_id()).or_default())
}
}
@@ -807,6 +856,7 @@ const MAX_RUNTIME_USAGE_RECORDS_PER_OWNER: usize = 64;
#[derive(Default)]
struct OwnerRuntimeUsageJournal {
records: VecDeque<RuntimeUsageRecord>,
decisions: VecDeque<RuntimeDecisionReceipt>,
drop_records: VecDeque<RuntimeUsageDropRecord>,
dropped_records: u64,
dropped_source_fingerprints: HashSet<String>,
@@ -820,6 +870,8 @@ type RuntimeUsageJournal = HashMap<String, OwnerRuntimeUsageJournal>;
/// closed instead of silently presenting a partial cost as complete.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct RuntimeUsageBatch {
/// Bounded decision evidence, owned and retired with the same usage ledger.
pub decisions: Vec<RuntimeDecisionReceipt>,
pub records: Vec<RuntimeUsageRecord>,
/// Exact provider-success calls whose usage payload was absent. The
/// bounded records retain route billing truth; `dropped_records` remains
@@ -846,16 +898,138 @@ pub struct RuntimeUsageDropRecord {
pub route: EffectiveRouteEnvelope,
}
/// Provider decision evidence attached to the existing origin-turn ledger.
/// It is diagnostic evidence, never an instruction to change the model route.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct RuntimeDecisionReceipt {
pub source_id: String,
pub route: EffectiveRouteEnvelope,
pub usage: Option<Usage>,
#[serde(default)]
pub usage_complete: bool,
pub shadow: bool,
pub valid_answers: bool,
pub evidence: crate::model_routing::AutoRouteDecisionEvidence,
}
impl RuntimeDecisionReceipt {
pub(crate) fn is_bounded(&self) -> bool {
!self.source_id.trim().is_empty()
&& self.source_id.len() <= 128
&& self.evidence.choice.len() <= 128
&& self.evidence.probabilities_bp.len() <= 64
&& self
.evidence
.probabilities_bp
.iter()
.all(|(key, value)| key.len() <= 128 && *value <= 10_000)
&& self.evidence.confidence_bp <= 10_000
&& self.evidence.min_confidence_bp <= 10_000
&& self
.evidence
.thinking
.as_ref()
.is_none_or(|v| v.len() <= 128)
&& self
.evidence
.response_model
.as_ref()
.is_none_or(|v| v.len() <= 128)
&& self
.evidence
.provider_reported_cost_usd
.as_ref()
.is_none_or(|v| {
v.len() <= 128
&& v.parse::<f64>()
.is_ok_and(|cost| cost.is_finite() && cost >= 0.0)
})
&& serde_json::to_vec(self).is_ok_and(|bytes| bytes.len() <= 8 * 1024)
}
pub(crate) fn sanitized(&self) -> Self {
let mut receipt = self.clone();
receipt.source_id = usage_source_fingerprint(&receipt.source_id);
receipt.route = receipt.route.sanitized_for_persistence();
receipt.evidence.choice = sanitize_persisted_route_label(&receipt.evidence.choice);
receipt.evidence.thinking = receipt
.evidence
.thinking
.as_deref()
.map(sanitize_persisted_route_label);
receipt.evidence.response_model = receipt
.evidence
.response_model
.as_deref()
.map(sanitize_persisted_route_label);
receipt.evidence.probabilities_bp = receipt
.evidence
.probabilities_bp
.into_iter()
.map(|(key, value)| (sanitize_persisted_route_label(&key), value))
.collect();
receipt
}
pub(crate) fn diagnostic_receipt(&self) -> String {
// Reuse cost_status's persisted diagnostic receipt surface. No prompt,
// response id, URL, auth header or unoffered choice enters this string.
format!(
"decision:{}",
serde_json::to_string(&self.sanitized()).unwrap_or_default()
)
}
}
pub(crate) type RuntimeDecisionSink = Arc<dyn Fn(RuntimeDecisionReceipt) -> bool + Send + Sync>;
pub(crate) type RuntimeUsageSink = Arc<dyn Fn(RuntimeUsageRecord) -> bool + Send + Sync>;
pub(crate) type RuntimeUsageDropSink = Arc<dyn Fn(RuntimeUsageDropRecord) -> bool + Send + Sync>;
struct RuntimeUsageSinkEntry {
sink: RuntimeUsageSink,
dropped_sink: Option<RuntimeUsageDropSink>,
decision_sink: Option<RuntimeDecisionSink>,
leases: usize,
terminal: bool,
}
#[cfg(test)]
pub(crate) fn decision_receipt_fixture(source_id: &str) -> RuntimeDecisionReceipt {
let mut route = EffectiveRouteEnvelope::capture(
None,
crate::config::ApiProvider::Custom,
"typesafe",
"jev-latest",
Some("https://api.typesafe.ai/v1"),
Utc::now(),
);
route.billing_mode = RouteBillingMode::Unknown;
RuntimeDecisionReceipt {
source_id: source_id.to_string(),
route,
usage: Some(Usage {
input_tokens: 9,
output_tokens: 4,
..Default::default()
}),
usage_complete: true,
shadow: true,
valid_answers: false,
evidence: crate::model_routing::AutoRouteDecisionEvidence {
choice: "invalid".to_string(),
probabilities_bp: Default::default(),
confidence_bp: 0,
min_confidence_bp: 5_000,
cost_saving_kept_fast: false,
thinking: None,
provider_reported_cost_usd: Some("0.000012054".to_string()),
latency_ms: 12,
response_model: Some("jev-latest".to_string()),
},
}
}
/// Keeps an owner sink alive while a detached child can still report usage.
/// The runtime turn may already be terminal; the last child release retires
/// the sink only after its final provider response has been durably appended.
@@ -908,7 +1082,7 @@ fn with_runtime_usage_sinks<R>(
.get_or_init(|| Mutex::new(HashMap::new()))
.lock()
.unwrap_or_else(|error| error.into_inner());
f(by_thread.entry(std::thread::current().id()).or_default())
f(by_thread.entry(test_cost_scope_id()).or_default())
}
}
@@ -928,7 +1102,7 @@ fn with_existing_runtime_usage_sinks<R>(
{
let by_thread = TEST_RUNTIME_USAGE_SINKS.get()?;
let mut by_thread = by_thread.lock().unwrap_or_else(|error| error.into_inner());
let sinks = by_thread.get_mut(&std::thread::current().id())?;
let sinks = by_thread.get_mut(&test_cost_scope_id())?;
Some(f(sinks))
}
}
@@ -948,7 +1122,7 @@ fn with_runtime_usage_journal_mut<R>(f: impl FnOnce(&mut RuntimeUsageJournal) ->
.get_or_init(|| Mutex::new(std::collections::HashMap::new()))
.lock()
.unwrap_or_else(|error| error.into_inner());
f(by_thread.entry(std::thread::current().id()).or_default())
f(by_thread.entry(test_cost_scope_id()).or_default())
}
}
@@ -1079,6 +1253,7 @@ pub(crate) fn register_runtime_usage_sink_with_drop(
RuntimeUsageSinkEntry {
sink,
dropped_sink,
decision_sink: None,
leases: 0,
terminal: false,
},
@@ -1086,6 +1261,74 @@ pub(crate) fn register_runtime_usage_sink_with_drop(
});
}
/// Extend the existing owner entry; this is not another sink registry.
pub(crate) fn register_runtime_decision_sink(owner: &str, sink: RuntimeDecisionSink) {
with_runtime_usage_sinks(|sinks| {
if let Some(entry) = sinks.get_mut(owner.trim()) {
entry.decision_sink = Some(sink);
}
});
}
fn record_runtime_decision(owner: &str, record: &RuntimeDecisionReceipt) {
if !record.is_bounded() {
record_runtime_usage_drop(owner, &record.source_id, &record.route);
return;
}
let record = record.sanitized();
let sink = with_runtime_usage_sinks(|sinks| {
sinks
.get(owner)
.and_then(|entry| entry.decision_sink.clone())
});
if sink.is_some_and(|sink| sink(record.clone())) {
return;
}
with_runtime_usage_journal_mut(|journal| {
let entry = journal.entry(owner.to_string()).or_default();
if entry
.decisions
.iter()
.any(|old| old.source_id == record.source_id)
{
return;
}
if entry.decisions.len() == MAX_RUNTIME_USAGE_RECORDS_PER_OWNER {
entry.decisions.pop_front();
entry.dropped_records = entry.dropped_records.saturating_add(1);
}
entry.decisions.push_back(record);
});
}
fn report_interactive_decision(scope: CostScopeToken, receipt: &RuntimeDecisionReceipt) {
if !receipt.is_bounded() {
return;
}
with_pending_state_mut(|state| {
if state.generation == scope.0 {
if state
.pending
.route_receipts
.iter()
.filter(|v| v.starts_with("decision:"))
.count()
< MAX_RUNTIME_USAGE_RECORDS_PER_OWNER
{
state
.pending
.route_receipts
.insert(receipt.diagnostic_receipt());
} else {
state
.pending
.route_receipts
.insert("decision:diagnostic_receipt_bound_reached".to_string());
}
}
});
}
/// Redacted durable identity shared by runtime-turn, worker, and interactive
/// session accounting. Raw response ids never need to be persisted merely to
/// make replay idempotent.
@@ -1151,6 +1394,9 @@ fn register_persistent_interactive_runtime_usage_sink_at(
let drop_session_id = usage_session_id.clone();
let drop_turn_id = usage_turn_id.clone();
let usage_sessions_dir = sessions_dir.clone();
let decision_sessions_dir = sessions_dir.clone();
let decision_session_id = session_id.to_string();
let decision_turn_id = turn_id.to_string();
register_runtime_usage_sink_with_drop(
owner,
Arc::new(move |record| {
@@ -1183,6 +1429,28 @@ fn register_persistent_interactive_runtime_usage_sink_at(
.unwrap_or(false)
})),
);
register_runtime_decision_sink(
owner,
Arc::new(move |receipt| {
let Ok(manager) =
crate::session_manager::SessionManager::new(decision_sessions_dir.clone())
else {
return false;
};
let accepted = manager
.persist_late_decision_receipt(&decision_session_id, &decision_turn_id, &receipt)
.unwrap_or(false);
if accepted {
// The append may have been handled by a deletion tombstone. Only
// a live origin may enter the foreground diagnostic projection.
let _ = manager.with_live_session_origin(&decision_session_id, || {
report_interactive_decision(scope, &receipt);
true
});
}
accepted
}),
);
}
#[cfg(test)]
@@ -1288,6 +1556,7 @@ pub fn take_runtime_usage(owner: &str) -> RuntimeUsageBatch {
journal
.remove(owner)
.map_or_else(RuntimeUsageBatch::default, |entry| RuntimeUsageBatch {
decisions: entry.decisions.into_iter().collect(),
records: entry.records.into_iter().collect(),
drop_records: entry.drop_records.into_iter().collect(),
dropped_records: entry.dropped_records,
@@ -1295,6 +1564,33 @@ pub fn take_runtime_usage(owner: &str) -> RuntimeUsageBatch {
})
}
#[cfg(test)]
thread_local! {
static TEST_COST_ORIGIN: std::cell::Cell<Option<std::thread::ThreadId>> = const { std::cell::Cell::new(None) };
}
#[cfg(test)]
pub(crate) fn test_cost_scope_id() -> std::thread::ThreadId {
TEST_COST_ORIGIN
.with(|scope| scope.get())
.unwrap_or_else(|| std::thread::current().id())
}
#[cfg(test)]
pub(crate) struct TestCostScopeBinding(Option<std::thread::ThreadId>);
#[cfg(test)]
pub(crate) fn bind_test_cost_scope(origin: std::thread::ThreadId) -> TestCostScopeBinding {
TestCostScopeBinding(TEST_COST_ORIGIN.with(|scope| scope.replace(Some(origin))))
}
#[cfg(test)]
impl Drop for TestCostScopeBinding {
fn drop(&mut self) {
TEST_COST_ORIGIN.with(|scope| scope.set(self.0));
}
}
/// Capture the current session/run generation before starting a background
/// provider request. The same token must be supplied when its usage returns.
#[must_use]
@@ -1845,6 +2141,13 @@ pub(crate) fn report_runtime_usage_batch(
runtime_owner: Option<&str>,
batch: &RuntimeUsageBatch,
) {
for receipt in &batch.decisions {
if let Some(owner) = runtime_owner {
record_runtime_decision(owner, receipt);
} else {
report_interactive_decision(scope, receipt);
}
}
for record in &batch.records {
report_effective_route_for_runtime(
scope,
@@ -3189,6 +3492,7 @@ mod tests {
attach_child_usage_batch_metadata(
&mut metadata,
&RuntimeUsageBatch {
decisions: Vec::new(),
records,
drop_records: Vec::new(),
dropped_records: 0,
@@ -3209,6 +3513,53 @@ mod tests {
assert_eq!(malformed.dropped_records, 1);
}
#[test]
fn decision_receipts_metadata_round_trip_is_bounded_and_legacy_compatible() {
let receipt = decision_receipt_fixture("raw-child-decision-id");
let mut metadata = serde_json::json!({});
attach_child_usage_batch_metadata(
&mut metadata,
&RuntimeUsageBatch {
decisions: vec![receipt.clone()],
..Default::default()
},
);
assert!(!metadata.to_string().contains("raw-child-decision-id"));
let decoded = child_usage_records_from_metadata(&metadata).expect("batch");
assert_eq!(decoded.decisions, vec![receipt.sanitized()]);
assert_eq!(decoded.dropped_records, 0);
metadata[CHILD_DECISION_RECEIPTS_KEY][0]["evidence"]["response_model"] =
serde_json::json!("x".repeat(129));
let invalid = child_usage_records_from_metadata(&metadata).expect("batch");
assert!(invalid.decisions.is_empty());
assert_eq!(invalid.dropped_records, 1);
metadata
.as_object_mut()
.expect("metadata")
.remove(CHILD_DECISION_RECEIPTS_KEY);
assert!(
child_usage_records_from_metadata(&metadata)
.expect("old batch")
.decisions
.is_empty()
);
let mut oversized = receipt;
oversized.evidence.response_model = Some("x".repeat(129));
attach_child_usage_batch_metadata(
&mut metadata,
&RuntimeUsageBatch {
decisions: vec![oversized],
..Default::default()
},
);
let invalid = child_usage_records_from_metadata(&metadata).expect("bounded batch");
assert!(invalid.decisions.is_empty());
assert_eq!(
invalid.dropped_records, 1,
"discarded evidence must leave an explicit coverage gap"
);
}
fn deepseek() -> BackgroundRoute<'static> {
BackgroundRoute::new(ApiProvider::Deepseek, "deepseek-v4-flash")
.with_base_url(Some(crate::config::DEFAULT_DEEPSEEK_BASE_URL))
+1
View File
@@ -127,6 +127,7 @@ mod shell_dispatcher;
mod skills;
mod snapshot;
mod startup_trace;
mod superfast;
mod task_manager;
mod telemetry_notice;
#[cfg(test)]
+192 -63
View File
@@ -443,7 +443,7 @@ pub(crate) struct AutoRouteReceipt {
/// What a decision model answered and what it cost. Probabilities are basis
/// points (0..=10000) so the receipt stays `Eq` and never re-renders floats.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) struct AutoRouteDecisionEvidence {
pub struct AutoRouteDecisionEvidence {
/// The tier the decision model chose (`fast` | `strong`).
pub(crate) choice: String,
pub(crate) probabilities_bp: BTreeMap<String, u16>,
@@ -473,6 +473,7 @@ pub(crate) enum AutoRouterFailure {
/// Declared but unusable: missing key, unknown kind, or client setup.
NotRunnable,
Timeout,
Cancelled,
Http {
status: u16,
},
@@ -492,6 +493,7 @@ impl AutoRouterFailure {
"not runnable (check the router key and [auto.router])".to_string()
}
Self::Timeout => "timed out".to_string(),
Self::Cancelled => "cancelled".to_string(),
Self::Http { status } => format!("HTTP {status}"),
Self::QuotaExhausted => "provider quota or credits exhausted".to_string(),
Self::Rejected => "request rejected by the provider".to_string(),
@@ -1223,7 +1225,10 @@ fn auto_route_usage_has_reported_data(usage: &codewhale_models::Usage) -> bool {
/// Stable, persistence-safe identity for one classifier response. The raw
/// provider response id is hashed with the frozen dispatch route and instant;
/// neither it nor any custom route label crosses into telemetry/persistence.
fn auto_route_usage_source_id(route: &EffectiveRouteEnvelope, response_id: &str) -> String {
pub(crate) fn auto_route_usage_source_id(
route: &EffectiveRouteEnvelope,
response_id: &str,
) -> String {
use sha2::{Digest as _, Sha256};
let mut digest = Sha256::new();
@@ -1546,8 +1551,17 @@ async fn auto_route_decision_recommendation(
let route = inventory
.router_decision_route
.ok_or_else(|| anyhow::anyhow!("decision router has no route"))?;
let client =
CodewhaleClient::for_decision_route(config, route, inventory.router_base_url.as_deref())?;
// Resolve environment/keyring credentials off the async worker.
let decision_config = config.clone();
let base_url = inventory.router_base_url.clone();
#[cfg(test)]
let ticket = crate::test_support::env_scope_ticket();
let client = tokio::task::spawn_blocking(move || {
#[cfg(test)]
let _membership = crate::test_support::join_env_scope(ticket);
CodewhaleClient::for_decision_route(&decision_config, route, base_url.as_deref())
})
.await??;
let body = decision_request_body(
&client,
&inventory.router_model,
@@ -1556,11 +1570,10 @@ async fn auto_route_decision_recommendation(
session_mode,
selected_thinking_mode,
);
// OpenRouter spend enters the session through the ordinary routed-usage
// path. TypeSafe is not a chat provider, so its spend is shown on the
// receipt only (see `client::system_one`).
let request_route = (route == DecisionRouterRoute::Openrouter)
.then(|| client.effective_route_envelope(&inventory.router_model, chrono::Utc::now()));
// Both transports enter the existing usage ledger. TypeSafe carries its
// own Custom/Unknown identity; no chat price or zero-cost claim is inherited.
let request_route =
Some(client.effective_route_envelope(&inventory.router_model, chrono::Utc::now()));
let started = Instant::now();
let dispatched = std::sync::atomic::AtomicBool::new(false);
let outcome = tokio::time::timeout(
@@ -1580,8 +1593,10 @@ async fn auto_route_decision_recommendation(
_ => InventoryAutoRouteAttempt::failed(AutoRouterFailure::Timeout),
},
Ok(Err(failure)) => match request_route {
Some(route) => auto_route_attempt_with_dropped_response(route, failure),
None => InventoryAutoRouteAttempt::failed(failure),
Some(route) if dispatched.load(std::sync::atomic::Ordering::Acquire) => {
auto_route_attempt_with_dropped_response(route, failure)
}
_ => InventoryAutoRouteAttempt::failed(failure),
},
Ok(Ok(response)) => decision_attempt_from_response(
config.auto_cost_saving(),
@@ -1595,15 +1610,18 @@ async fn auto_route_decision_recommendation(
}
/// A validated `choice` answer.
struct ValidChoice {
choice: String,
probabilities_bp: BTreeMap<String, u16>,
confidence_bp: u16,
pub(crate) struct ValidChoice {
pub(crate) choice: String,
pub(crate) probabilities_bp: BTreeMap<String, u16>,
pub(crate) confidence_bp: u16,
}
/// Validate one `choice` answer against the offered options. Any violation
/// rejects the whole answer; nothing is repaired.
fn validated_choice(answer: Option<&SystemOneAnswer>, options: &[&str]) -> Option<ValidChoice> {
pub(crate) fn validated_choice(
answer: Option<&SystemOneAnswer>,
options: &[&str],
) -> Option<ValidChoice> {
let answer = answer?;
if answer.kind != "choice" {
return None;
@@ -1625,6 +1643,14 @@ fn validated_choice(answer: Option<&SystemOneAnswer>, options: &[&str]) -> Optio
if (sum - 1.0).abs() > 0.02 {
return None;
}
let selected = (*answer.probabilities.get(choice)?)?;
if answer
.probabilities
.values()
.any(|probability| probability.is_some_and(|v| v > selected))
{
return None;
}
let confidence = answer.confidence?;
if !confidence.is_finite() || !(0.0..=1.0).contains(&confidence) {
return None;
@@ -1636,6 +1662,42 @@ fn validated_choice(answer: Option<&SystemOneAnswer>, options: &[&str]) -> Optio
})
}
/// One decision response settled through the existing routed-usage ledger.
/// Malformed policy answers still carry provider usage; missing/all-zero usage
/// remains an explicit coverage gap. The source identity is route-bound.
pub(crate) fn decision_usage_batch(
request_route: &EffectiveRouteEnvelope,
response: &SystemOneResponse,
) -> crate::cost_status::RuntimeUsageBatch {
let usage = codewhale_models::Usage {
input_tokens: response.usage.as_ref().map_or(0, |u| u.input_tokens),
output_tokens: response.usage.as_ref().map_or(0, |u| u.output_tokens),
..Default::default()
};
let source_id =
auto_route_usage_source_id(request_route, response.id.as_deref().unwrap_or("systemone"));
let route = request_route.sanitized_for_persistence();
if response.usage.as_ref().is_some_and(|u| u.complete)
&& auto_route_usage_has_reported_data(&usage)
{
crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![RuntimeUsageRecord {
source_id,
usage: EffectiveRouteUsage { route, usage },
}],
..Default::default()
}
} else {
crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
drop_records: vec![RuntimeUsageDropRecord { source_id, route }],
dropped_records: 1,
..Default::default()
}
}
}
/// Turn a decoded System One response into a routing attempt. All policy is
/// here, in code: the decision model does no arithmetic.
fn decision_attempt_from_response(
@@ -1656,41 +1718,35 @@ fn decision_attempt_from_response(
fallback_reason: None,
};
if let Some(request_route) = request_route {
let usage = codewhale_models::Usage {
input_tokens: response
.usage
.as_ref()
.map_or(0, |usage| usage.input_tokens),
output_tokens: response
.usage
.as_ref()
.map_or(0, |usage| usage.output_tokens),
..Default::default()
};
let response_id = response.id.as_deref().unwrap_or("systemone");
if auto_route_usage_has_reported_data(&usage) {
attempt.routed_usage.push(RuntimeUsageRecord {
source_id: auto_route_usage_source_id(&request_route, response_id),
usage: EffectiveRouteUsage {
route: request_route.sanitized_for_persistence(),
usage,
},
});
} else {
let drop_route = request_route.sanitized_for_persistence();
attempt
.routed_usage_drop_records
.push(RuntimeUsageDropRecord {
source_id: auto_route_usage_source_id(
&drop_route,
&format!("missing-usage:{response_id}"),
),
route: drop_route,
});
attempt.routed_usage_dropped_records = 1;
}
let batch = decision_usage_batch(&request_route, response);
attempt.routed_usage = batch.records;
attempt.routed_usage_dropped_records = batch.dropped_records;
attempt.routed_usage_drop_records = batch.drop_records;
}
// Rejected answers still incurred spend. Keep bounded diagnostic evidence
// before any policy return; valid tier evidence below replaces this shell.
attempt.decision = Some(AutoRouteDecisionEvidence {
choice: "invalid".to_string(),
probabilities_bp: Default::default(),
confidence_bp: 0,
min_confidence_bp: inventory.router_min_confidence_bp,
cost_saving_kept_fast: false,
thinking: None,
provider_reported_cost_usd: response
.usage
.as_ref()
.and_then(|usage| usage.reported_cost()),
latency_ms,
response_model: response
.model
.as_deref()
.map(|model| model.chars().take(128).collect()),
});
if response.answers_validated == Some(false) {
attempt.failure = Some(AutoRouterFailure::InvalidAnswer);
return attempt;
}
let Some(tier) = validated_choice(response.answers.get("tier"), &DECISION_TIER_OPTIONS) else {
attempt.failure = Some(AutoRouterFailure::InvalidAnswer);
return attempt;
@@ -1886,7 +1942,7 @@ fn append_router_text(out: &mut String, text: &str) {
out.push_str(text);
}
fn truncate_for_auto_router(text: &str, max_chars: usize) -> String {
pub(crate) fn truncate_for_auto_router(text: &str, max_chars: usize) -> String {
let mut chars = text.chars();
let truncated: String = chars.by_ref().take(max_chars).collect();
if chars.next().is_some() {
@@ -3280,13 +3336,13 @@ mod decision_router_tests {
let _env = hermetic_env();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(path("/api/alpha/decisions"))
.and(header("authorization", "Bearer or-test-key"))
.respond_with(ResponseTemplate::new(200).set_body_json(answer_body(0.82, 0.64)))
.expect(1)
.mount(&server)
.await;
let mut config = decision_config(&server.uri(), false, 2);
let mut config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
// A configured credential that a tool result echoed back.
let secret = "cw-router-secret-should-never-leave-process";
config
@@ -3363,11 +3419,11 @@ mod decision_router_tests {
let _env = hermetic_env();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(path("/api/alpha/decisions"))
.respond_with(ResponseTemplate::new(200).set_body_json(answer_body(0.82, 0.64)))
.mount(&server)
.await;
let config = decision_config(&server.uri(), false, 2);
let config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
let selection = route(&config, "Debug the release pipeline", "").await;
assert_eq!(selection.provider, ApiProvider::Deepseek);
@@ -3418,11 +3474,11 @@ mod decision_router_tests {
let _env = hermetic_env();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(path("/api/alpha/decisions"))
.respond_with(ResponseTemplate::new(200).set_body_json(answer_body(0.35, 0.3)))
.mount(&server)
.await;
let config = decision_config(&server.uri(), false, 2);
let config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
let selection = route(&config, "What does this function do?", "").await;
assert_eq!(selection.model, "deepseek-v4-pro", "declared default");
@@ -3449,7 +3505,7 @@ mod decision_router_tests {
let _env = hermetic_env();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(path("/api/alpha/decisions"))
.respond_with(
ResponseTemplate::new(status)
.set_body_string(format!(r#"{{"error":{{"message":"{MARKER}"}}}}"#)),
@@ -3457,7 +3513,7 @@ mod decision_router_tests {
.expect(1)
.mount(&server)
.await;
let config = decision_config(&server.uri(), false, 2);
let config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
let selection = route(&config, "Explain the diff", "").await;
assert_eq!(selection.model, "deepseek-v4-pro", "status {status}");
@@ -3485,7 +3541,7 @@ mod decision_router_tests {
let _env = hermetic_env();
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/systemone"))
.and(path("/api/alpha/decisions"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(answer_body(0.9, 0.8))
@@ -3493,7 +3549,7 @@ mod decision_router_tests {
)
.mount(&server)
.await;
let config = decision_config(&server.uri(), false, 1);
let config = decision_config(&format!("{}/api/v1", server.uri()), false, 1);
let selection = route(&config, "Explain the diff", "").await;
let receipt = receipt(&selection);
@@ -3517,7 +3573,7 @@ mod decision_router_tests {
.expect(0)
.mount(&server)
.await;
let mut config = decision_config(&server.uri(), false, 2);
let mut config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
// A single-tier active provider: nothing for the decision to choose.
config.provider = Some("openrouter".to_string());
config.default_text_model = Some("synthetic/single-tier-model".to_string());
@@ -3590,8 +3646,17 @@ mod decision_router_tests {
}
);
assert!(receipt.decision.is_some());
// TypeSafe is not a chat provider: spend stays on the receipt only.
assert!(selection.routed_usage.is_empty());
// A decision endpoint never inherits the active DeepSeek billing identity.
assert_eq!(selection.routed_usage.len(), 1);
let usage = &selection.routed_usage[0];
assert_eq!(usage.usage.route.provider, ApiProvider::Custom);
assert_eq!(usage.usage.route.provider_identity, "typesafe");
assert_eq!(
usage.usage.route.billing_mode,
crate::cost_status::RouteBillingMode::Unknown
);
assert_eq!(usage.usage.usage.input_tokens, 476);
assert_eq!(usage.usage.usage.output_tokens, 70);
assert!(selection.routed_usage_drop_records.is_empty());
}
@@ -3672,6 +3737,18 @@ mod decision_router_tests {
for body in [outside, bad_sum, missing, null_probability, bad_confidence] {
let attempt = policy(false, &parsed(body.clone()));
assert_eq!(attempt.recommendation, None, "{body}");
assert_eq!(
attempt
.decision
.as_ref()
.and_then(|r| r.provider_reported_cost_usd.as_deref()),
parsed(body.clone())
.usage
.as_ref()
.and_then(|u| u.reported_cost())
.as_deref(),
"rejected policy retains cost"
);
assert_eq!(
attempt.failure,
Some(AutoRouterFailure::InvalidAnswer),
@@ -3688,6 +3765,58 @@ mod decision_router_tests {
assert_eq!(recommendation.reasoning_effort, None);
}
#[tokio::test]
async fn malformed_decision_envelopes_preserve_cost_without_a_route_hop() {
let _env = hermetic_env();
for (pointer, value, incomplete_usage) in [
("/answers/tier/choice", serde_json::json!(17), false),
("/model", serde_json::json!({"invalid":"model"}), false),
("/usage/input_tokens", serde_json::json!(u64::MAX), true),
] {
let server = MockServer::start().await;
let mut body = answer_body(0.9, 0.9);
*body.pointer_mut(pointer).expect("fixture field") = value;
Mock::given(method("POST"))
.and(path("/api/alpha/decisions"))
.respond_with(ResponseTemplate::new(200).set_body_json(body))
.expect(1)
.mount(&server)
.await;
let config = decision_config(&format!("{}/api/v1", server.uri()), false, 2);
let fallback =
auto_route_without_router(&config, &ModelInventory::from_config(&config));
let selection = route(&config, "Explain a variable name", "").await;
assert_eq!(selection.source, AutoRouteSource::Heuristic);
assert_eq!(selection.provider, fallback.provider);
assert_eq!(
selection.model, fallback.model,
"malformed policy cannot change the route"
);
let receipt = receipt(&selection);
assert_eq!(
receipt.router_failure,
Some(AutoRouterFailure::InvalidAnswer)
);
assert_eq!(
receipt
.decision
.as_ref()
.expect("billing evidence")
.provider_reported_cost_usd
.as_deref(),
Some("0.000019992")
);
if incomplete_usage {
assert!(selection.routed_usage.is_empty());
assert_eq!(selection.routed_usage_dropped_records, 1);
assert_eq!(selection.routed_usage_drop_records.len(), 1);
} else {
assert_eq!(selection.routed_usage.len(), 1);
assert!(selection.routed_usage_drop_records.is_empty());
}
}
}
#[test]
fn camel_case_usage_and_verbatim_cost_parse() {
let response = parsed(serde_json::json!({
+4
View File
@@ -449,6 +449,7 @@ fn messages_from_thread_detail_batches_tool_results() {
saved_session_checkpoint: None,
};
let turn = TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: 2,
id: turn_id.clone(),
@@ -9382,6 +9383,7 @@ async fn session_save_merges_thread_cost_split_and_records_coverage() -> Result<
let store = runtime_threads.test_store();
let now = Utc::now();
let mut turn = TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: 2,
id: "turn_cost_merge".to_string(),
@@ -9580,6 +9582,7 @@ async fn session_save_persists_parent_cny_unpriced_reasons_without_double_count(
let store = runtime_threads.test_store();
let now = Utc::now();
let mut turn = TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: 2,
id: "turn_cny_reasons".to_string(),
@@ -10830,6 +10833,7 @@ fn seed_summary_search_transcript(
item_ids.push(item_id);
}
store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: 2,
id: turn_id.clone(),
+117 -1
View File
@@ -603,6 +603,20 @@ where
.collect::<Vec<_>>()
.serialize(serializer)
}
fn serialize_decision_receipts<S>(
values: &[crate::cost_status::RuntimeDecisionReceipt],
serializer: S,
) -> std::result::Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
values
.iter()
.map(crate::cost_status::RuntimeDecisionReceipt::sanitized)
.collect::<Vec<_>>()
.serialize(serializer)
}
const RUNTIME_RESTART_REASON: &str = "Interrupted by process restart";
const EMPTY_TURN_REASON: &str = "Turn completed without engine output";
const DYNAMIC_TOOL_RESULT_TIMEOUT: Duration = Duration::from_secs(300);
@@ -1457,6 +1471,14 @@ pub struct TurnRecord {
/// inline RLM/guardian calls.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub routed_usage: Vec<EffectiveRouteUsage>,
/// Decision evidence shares this durable origin-turn ledger, including
/// responses arriving after the parent terminal event.
#[serde(
default,
skip_serializing_if = "Vec::is_empty",
serialize_with = "serialize_decision_receipts"
)]
pub decision_receipts: Vec<crate::cost_status::RuntimeDecisionReceipt>,
/// Exact missing-usage provider responses paired with their frozen route.
/// Sources are persisted only as fingerprints, routes are sanitized, and
/// the shared source ledger bounds usage plus missing-usage records.
@@ -1523,6 +1545,11 @@ pub struct TurnRecord {
impl TurnRecord {
fn validate_output_token_limit(&self) -> Result<()> {
if self.decision_receipts.len() > MAX_ROUTED_USAGE_RECORDS_PER_TURN
|| self.decision_receipts.iter().any(|r| !r.is_bounded())
{
bail!("Turn decision evidence exceeds its bound");
}
if self.schema_version > MAX_SUPPORTED_RUNTIME_SCHEMA_VERSION {
bail!(
"Turn schema v{} is newer than supported v{}",
@@ -1669,6 +1696,32 @@ fn append_routed_usage_drop_record(turn: &mut TurnRecord, record: RuntimeUsageDr
true
}
fn append_decision_receipt(
turn: &mut TurnRecord,
receipt: &crate::cost_status::RuntimeDecisionReceipt,
) -> bool {
if !receipt.is_bounded() {
return false;
}
let receipt = receipt.sanitized();
if turn
.decision_receipts
.iter()
.any(|old| old.source_id == receipt.source_id)
{
return false;
}
if turn.decision_receipts.len() >= MAX_ROUTED_USAGE_RECORDS_PER_TURN {
if turn.routed_usage_dropped_records == 0 {
turn.routed_usage_dropped_records = 1;
return true;
}
return false;
}
turn.decision_receipts.push(receipt);
true
}
/// Bind pre-parent auxiliary calls to a reserved Runtime turn before the
/// engine accepts the parent operation. Exact records are persisted now so a
/// crash cannot erase them; the engine's terminal event remains the sole
@@ -1678,6 +1731,9 @@ fn append_initial_routed_usage_to_turn(
turn: &mut TurnRecord,
batch: &crate::cost_status::RuntimeUsageBatch,
) {
for receipt in &batch.decisions {
append_decision_receipt(turn, receipt);
}
for record in &batch.records {
append_routed_usage_record(turn, &record.source_id, record.usage.clone());
}
@@ -1694,7 +1750,10 @@ const UNACCEPTED_TURN_REASON: &str =
"Turn was not accepted; this record keeps the completed pre-turn provider call";
fn routed_usage_batch_is_empty(batch: &crate::cost_status::RuntimeUsageBatch) -> bool {
batch.records.is_empty() && batch.drop_records.is_empty() && batch.dropped_records == 0
batch.decisions.is_empty()
&& batch.records.is_empty()
&& batch.drop_records.is_empty()
&& batch.dropped_records == 0
}
/// Durable identity of the settlement record for one completed pre-turn
@@ -1721,6 +1780,12 @@ fn unaccepted_routed_usage_turn_id(
.map(|record| routed_usage_source_fingerprint(&record.source_id)),
)
.collect::<Vec<_>>();
identities.extend(
batch
.decisions
.iter()
.map(|record| routed_usage_source_fingerprint(&record.source_id)),
);
if identities.is_empty() {
return runtime_record_id("turn_unaccepted");
}
@@ -1758,6 +1823,7 @@ fn settle_unaccepted_routed_usage(
} else {
let now = Utc::now();
TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: turn_id.clone(),
@@ -9405,6 +9471,23 @@ impl RuntimeThreadManager {
&thread,
);
}
for receipt in &turn.decision_receipts {
if usage_timestamp_in_range(receipt.route.dispatched_at, since, until) {
if totals
.route_receipts
.iter()
.filter(|r| r.starts_with("decision:"))
.count()
< 64
{
totals.route_receipts.insert(receipt.diagnostic_receipt());
} else {
totals
.route_receipts
.insert("decision:diagnostic_receipt_bound_reached".to_string());
}
}
}
for child in &turn.routed_usage {
if usage_timestamp_in_range(child.route.dispatched_at, since, until) {
accumulate_runtime_child_usage_record(
@@ -11432,6 +11515,7 @@ impl RuntimeThreadManager {
// Only create a turn if there's content.
if !item_ids.is_empty() {
self.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: if turn_seed.image_content.is_empty() {
CURRENT_RUNTIME_SCHEMA_VERSION
@@ -11800,6 +11884,9 @@ impl RuntimeThreadManager {
let _turn_mutation = self.store.turn_mutation.lock();
match self.store.load_turn(turn_id) {
Ok(mut turn) => {
for receipt in &background_usage.decisions {
append_decision_receipt(&mut turn, receipt);
}
for record in background_usage.records.iter().cloned() {
append_routed_usage_record(&mut turn, &record.source_id, record.usage);
}
@@ -12478,6 +12565,7 @@ impl RuntimeThreadManager {
// malformed/stale parent catalog entry cannot erase real
// auxiliary spend on the error path.
let initial_routed_usage = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: std::mem::take(&mut selection.routed_usage),
drop_records: std::mem::take(&mut selection.routed_usage_drop_records),
dropped_records: std::mem::take(
@@ -12594,6 +12682,7 @@ impl RuntimeThreadManager {
.clone()
.unwrap_or_else(|| summarize_text(&prompt, SUMMARY_LIMIT));
let mut turn = TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens,
schema_version: if max_output_tokens.is_some() { OUTPUT_LIMIT_RUNTIME_SCHEMA_VERSION } else { CURRENT_RUNTIME_SCHEMA_VERSION },
id: turn_id.clone(),
@@ -13087,6 +13176,7 @@ impl RuntimeThreadManager {
thread.auto_approve,
);
let turn = TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: turn_id.clone(),
@@ -14309,6 +14399,21 @@ impl RuntimeThreadManager {
drop_store.save_turn(&turn).is_ok()
})),
);
let decision_store = self.store.clone();
let decision_turn_id = turn_id.to_string();
crate::cost_status::register_runtime_decision_sink(
turn_id,
Arc::new(move |receipt| {
let _turn_mutation = decision_store.turn_mutation.lock();
let Ok(mut turn) = decision_store.load_turn(&decision_turn_id) else {
return false;
};
if !append_decision_receipt(&mut turn, &receipt) {
return true;
}
decision_store.save_turn(&turn).is_ok()
}),
);
}
/// Persist an engine status line as a completed `status` item.
@@ -14813,6 +14918,14 @@ impl RuntimeThreadManager {
if let Some(batch) =
crate::cost_status::child_usage_records_from_metadata(metadata)
{
if !batch.decisions.is_empty() {
let _turn_mutation = self.store.turn_mutation.lock();
let mut turn = self.store.load_turn(&turn_id)?;
for receipt in &batch.decisions {
append_decision_receipt(&mut turn, receipt);
}
self.store.save_turn(&turn)?;
}
for record in batch.records {
self.append_routed_usage_to_turn(
&turn_id,
@@ -16191,6 +16304,9 @@ impl RuntimeThreadManager {
turn.duration_ms = turn.started_at.map(|start| duration_ms(start, ended_at));
turn.usage = turn_usage;
turn.effective_route_usage = turn_effective_route_usage;
for receipt in &background_usage.decisions {
append_decision_receipt(&mut turn, receipt);
}
for record in background_usage.records {
append_routed_usage_record(&mut turn, &record.source_id, record.usage);
}
+75
View File
@@ -2447,6 +2447,7 @@ fn sample_thread(thread_id: &str) -> ThreadRecord {
fn sample_turn(thread_id: &str, turn_id: &str, status: RuntimeTurnStatus) -> TurnRecord {
let now = Utc::now();
TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: turn_id.to_string(),
@@ -8464,6 +8465,7 @@ async fn initial_classifier_usage_is_persisted_before_terminal_and_merged_exactl
Utc::now(),
);
let classifier_batch = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![crate::cost_status::RuntimeUsageRecord {
source_id: "auto-router:runtime-fixture".to_string(),
usage: crate::cost_status::EffectiveRouteUsage {
@@ -8600,6 +8602,7 @@ fn classifier_settlement_batch(
);
route.billing_mode = crate::cost_status::RouteBillingMode::Subscription;
crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![crate::cost_status::RuntimeUsageRecord {
source_id: format!("auto-router:{source_prefix}-usage"),
usage: crate::cost_status::EffectiveRouteUsage {
@@ -16747,6 +16750,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
manager.store.save_item(&queued_item)?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: "turn_in_progress".to_string(),
@@ -16785,6 +16789,7 @@ fn opening_manager_recovers_stale_queued_and_in_progress_work() -> Result<()> {
workspace_snapshots: Vec::new(),
})?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: "turn_queued".to_string(),
@@ -16981,6 +16986,7 @@ fn seed_turns_with_user_messages(
ended_at: Some(created_at),
})?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: turn_id.clone(),
@@ -17800,6 +17806,7 @@ fn restart_rebuild_restores_tool_call_identity_from_persisted_items() -> Result<
manager.store.save_item(&user_item)?;
manager.store.save_item(&call_item)?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: "turn_5823".to_string(),
@@ -17905,6 +17912,7 @@ fn restart_rebuild_keeps_in_flight_tool_call_identity() -> Result<()> {
};
manager.store.save_item(&call_item)?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: "turn_5823_inflight".to_string(),
@@ -18005,6 +18013,7 @@ fn restart_rebuild_skips_steers_the_engine_never_delivered() -> Result<()> {
manager.store.save_item(&dropped)?;
manager.store.save_item(&pending)?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: turn_id.clone(),
@@ -18106,6 +18115,7 @@ fn restart_rebuild_skips_legacy_tool_items_without_identity() -> Result<()> {
manager.store.save_item(&user_item)?;
manager.store.save_item(&legacy_tool_item)?;
manager.store.save_turn(&TurnRecord {
decision_receipts: Vec::new(),
max_output_tokens: None,
schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
id: "turn_5823_legacy".to_string(),
@@ -21962,3 +21972,68 @@ mod execution_identity {
Ok(())
}
}
#[tokio::test]
async fn decision_receipt_lease_persists_terminal_turn_and_replays_exactly_once() -> Result<()> {
let _cost_scope = crate::cost_status::test_scope();
let directory = test_runtime_dir();
let manager = test_manager(directory.clone())?;
let thread = manager
.create_thread(CreateThreadRequest::default())
.await?;
let turn = sample_turn(
&thread.id,
"decision-terminal-origin",
RuntimeTurnStatus::Completed,
);
manager.store.save_turn(&turn)?;
manager.register_runtime_usage_sink(&turn.id);
let lease = crate::cost_status::acquire_runtime_usage_lease(&turn.id).expect("origin lease");
crate::cost_status::finish_runtime_usage_owner(&turn.id);
let receipt = crate::cost_status::decision_receipt_fixture("raw-runtime-decision-id");
let batch = crate::cost_status::RuntimeUsageBatch {
decisions: vec![receipt.clone()],
..Default::default()
};
for _ in 0..2 {
crate::cost_status::report_runtime_usage_batch(
crate::cost_status::scope_token(),
Some(&turn.id),
&batch,
);
}
drop(lease);
drop(manager);
let restarted = test_manager(directory)?;
let reloaded = restarted.store.load_turn(&turn.id)?;
assert_eq!(reloaded.status, RuntimeTurnStatus::Completed);
assert_eq!(reloaded.decision_receipts, vec![receipt.sanitized()]);
assert_eq!(
reloaded.decision_receipts[0]
.evidence
.provider_reported_cost_usd
.as_deref(),
Some("0.000012054")
);
let aggregate = restarted
.aggregate_usage(None, None, UsageGroupBy::Thread)
.await?;
assert!(
aggregate
.totals
.route_receipts
.iter()
.any(|r| r.contains("0.000012054")),
"existing cost diagnostics expose retained provider evidence"
);
assert!(
reloaded.routed_usage.is_empty(),
"diagnostic receipt alone must not mint a second token charge"
);
assert_eq!(
unaccepted_routed_usage_turn_id(&thread.id, &batch),
unaccepted_routed_usage_turn_id(&thread.id, &batch)
);
assert!(!serde_json::to_string(&reloaded)?.contains("raw-runtime-decision-id"));
Ok(())
}
+179 -4
View File
@@ -71,6 +71,8 @@ struct LateUsageRecord {
route: crate::cost_status::EffectiveRouteEnvelope,
#[serde(default, skip_serializing_if = "Option::is_none")]
usage: Option<codewhale_models::Usage>,
#[serde(default, skip_serializing_if = "Option::is_none")]
decision: Option<crate::cost_status::RuntimeDecisionReceipt>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -1792,6 +1794,7 @@ impl SessionManager {
|| ledger.records.iter().any(|record| {
!is_sha256_fingerprint(&record.source_fingerprint)
|| !is_sha256_fingerprint(&record.turn_fingerprint)
|| record.decision.as_ref().is_some_and(|r| !r.is_bounded())
})
{
return Err(io::Error::new(
@@ -1853,6 +1856,7 @@ impl SessionManager {
source_id: &str,
route: &crate::cost_status::EffectiveRouteEnvelope,
usage: Option<&codewhale_models::Usage>,
decision: Option<&crate::cost_status::RuntimeDecisionReceipt>,
) -> io::Result<bool> {
let (path, lock_path) = self.ensure_late_usage_paths(session_id)?;
let lock_file = open_private_lock_file(&lock_path)?;
@@ -1864,11 +1868,21 @@ impl SessionManager {
}
let mut ledger = Self::load_late_usage_unlocked(&path)?;
let source_fingerprint = crate::cost_status::usage_source_fingerprint(source_id);
if ledger
if let Some(record) = ledger
.records
.iter()
.any(|record| record.source_fingerprint == source_fingerprint)
.iter_mut()
.find(|record| record.source_fingerprint == source_fingerprint)
{
if let Some(decision) = decision {
if !decision.is_bounded() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"decision receipt exceeds its bound",
));
}
record.decision = Some(decision.sanitized());
Self::write_late_usage_ledger(&path, &ledger)?;
}
return Ok(true);
}
if ledger.records.len() == MAX_LATE_USAGE_RECORDS_PER_SESSION {
@@ -1883,11 +1897,48 @@ impl SessionManager {
turn_fingerprint: crate::cost_status::usage_source_fingerprint(turn_id),
route: route.sanitized_for_persistence(),
usage: usage.cloned(),
decision: decision.map(crate::cost_status::RuntimeDecisionReceipt::sanitized),
});
Self::write_late_usage_ledger(&path, &ledger)?;
Ok(true)
}
pub(crate) fn persist_late_decision_receipt(
&self,
session_id: &str,
turn_id: &str,
receipt: &crate::cost_status::RuntimeDecisionReceipt,
) -> io::Result<bool> {
if !receipt.is_bounded() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"decision receipt exceeds its bound",
));
}
self.persist_late_usage_record(
session_id,
turn_id,
&receipt.source_id,
&receipt.route,
receipt.usage.as_ref().filter(|_| receipt.usage_complete),
Some(receipt),
)
}
/// Read the bounded provider decision evidence from the existing origin ledger.
#[cfg(test)]
pub(crate) fn decision_receipts_for_session(
&self,
session_id: &str,
) -> io::Result<Vec<crate::cost_status::RuntimeDecisionReceipt>> {
Ok(self
.load_late_usage(session_id)?
.records
.into_iter()
.filter_map(|record| record.decision)
.collect())
}
pub(crate) fn persist_late_runtime_usage(
&self,
session_id: &str,
@@ -1900,6 +1951,7 @@ impl SessionManager {
&record.source_id,
&record.usage.route,
Some(&record.usage.usage),
None,
)
}
@@ -1909,7 +1961,14 @@ impl SessionManager {
turn_id: &str,
record: &crate::cost_status::RuntimeUsageDropRecord,
) -> io::Result<bool> {
self.persist_late_usage_record(session_id, turn_id, &record.source_id, &record.route, None)
self.persist_late_usage_record(
session_id,
turn_id,
&record.source_id,
&record.route,
None,
None,
)
}
fn with_session_read_lock<T>(
@@ -1974,6 +2033,14 @@ impl SessionManager {
}
};
for record in ledger.records {
if let Some(receipt) = &record.decision
&& receipt.is_bounded()
{
metadata
.cost
.route_receipts
.insert(receipt.diagnostic_receipt());
}
let source_fingerprint = record.source_fingerprint.clone();
let source_id = format!("late:{}", record.source_fingerprint);
let mut pending = if let Some(usage) = record.usage.as_ref() {
@@ -4357,6 +4424,114 @@ mod tests {
}
}
#[test]
fn decision_receipt_survives_restart_replay_and_session_deletion() {
let tmp = tempdir().expect("tempdir");
let manager = SessionManager::new(tmp.path().join("sessions")).expect("manager");
save_late_usage_test_session(&manager, "decision-origin");
let receipt = crate::cost_status::decision_receipt_fixture("raw-decision-response-id");
manager
.persist_late_decision_receipt("decision-origin", "origin-turn", &receipt)
.expect("decision append");
manager
.persist_late_runtime_usage(
"decision-origin",
"origin-turn",
&crate::cost_status::RuntimeUsageRecord {
source_id: receipt.source_id.clone(),
usage: crate::cost_status::EffectiveRouteUsage {
route: receipt.route.clone(),
usage: receipt.usage.clone().expect("usage"),
},
},
)
.expect("token append dedupes");
drop(manager);
let manager = SessionManager::new(tmp.path().join("sessions")).expect("restart");
let retained = manager
.decision_receipts_for_session("decision-origin")
.expect("receipts");
assert_eq!(retained, vec![receipt.sanitized()]);
let mut restored = manager
.load_session_snapshot("decision-origin")
.expect("restore");
assert_eq!(restored.metadata.total_tokens, 13);
assert_eq!(
restored.metadata.cost.unpriced_turns, 1,
"unknown TypeSafe billing remains explicit"
);
assert!(
restored
.metadata
.cost
.route_receipts
.iter()
.any(|r| r.contains("0.000012054"))
);
manager.apply_late_usage_to_metadata(&mut restored.metadata);
assert_eq!(
restored.metadata.total_tokens, 13,
"replay must not count tokens twice"
);
assert_eq!(restored.metadata.cost.unpriced_turns, 1);
manager.save_session(&restored).expect("save overlay");
assert_eq!(
manager
.load_session_snapshot("decision-origin")
.expect("resume again")
.metadata
.total_tokens,
13
);
let (ledger, _) = manager.late_usage_paths("decision-origin").expect("paths");
assert!(
!fs::read_to_string(&ledger)
.expect("ledger")
.contains("raw-decision-response-id")
);
save_late_usage_test_session(&manager, "decision-partial-origin");
let mut partial = receipt.clone();
partial.source_id = "partial-provider-response-id".into();
partial.usage_complete = false;
partial.usage.as_mut().expect("partial usage").output_tokens = 0;
manager
.persist_late_decision_receipt("decision-partial-origin", "partial-turn", &partial)
.expect("partial diagnostic append");
let partial_session = manager
.load_session_snapshot("decision-partial-origin")
.expect("partial replay");
assert_eq!(
partial_session.metadata.total_tokens, 0,
"partial provider counters remain diagnostic rather than an authoritative subtotal"
);
assert_eq!(partial_session.metadata.cost.unpriced_turns, 1);
assert!(
partial_session
.metadata
.cost
.route_receipts
.iter()
.any(|r| r.contains("0.000012054"))
);
manager.delete_session("decision-origin").expect("delete");
assert!(
manager
.persist_late_decision_receipt("decision-origin", "origin-turn", &receipt)
.expect("retired replay")
);
assert!(
!ledger.exists(),
"late evidence must never recreate a deleted origin"
);
let mut oversized = receipt;
oversized.evidence.response_model = Some("x".repeat(129));
assert!(
manager
.persist_late_decision_receipt("decision-origin", "origin-turn", &oversized)
.is_err()
);
}
#[test]
fn late_usage_reads_do_not_create_accounting_storage() {
let tmp = tempdir().expect("tempdir");
File diff suppressed because it is too large. Load diff
+3
View File
@@ -494,6 +494,7 @@ impl RlmTool {
&error.to_string(),
started.elapsed(),
&crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: usage.records,
drop_records: usage.drop_records,
dropped_records: usage.dropped_records,
@@ -504,6 +505,7 @@ impl RlmTool {
(
round,
crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: usage.records,
drop_records: usage.drop_records,
dropped_records: usage.dropped_records,
@@ -996,6 +998,7 @@ mod tests {
"kernel stdout closed",
Duration::from_millis(11),
&crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![record],
drop_records: vec![drop_record],
dropped_records: 1,
+2
View File
@@ -72,6 +72,7 @@ fn failed_engine_channel_settles_classifier_batch_once() {
chrono::Utc::now(),
);
let batch = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![crate::cost_status::RuntimeUsageRecord {
source_id: "auto-router:dispatch-usage".to_string(),
usage: crate::cost_status::EffectiveRouteUsage {
@@ -21927,6 +21928,7 @@ fn routed_missing_usage_batch_prices_exact_routes_and_only_residual_as_generic()
local_route.provider_identity = "local-computer".to_string();
local_route.billing_mode = crate::cost_status::RouteBillingMode::Local;
let batch = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: Vec::new(),
drop_records: vec![
crate::cost_status::RuntimeUsageDropRecord {
+1
View File
@@ -312,6 +312,7 @@ async fn test_preset(app: &mut App, config: &Config, preset: RouterPreset) {
cost_scope,
None,
&crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: selection.routed_usage.clone(),
drop_records: selection.routed_usage_drop_records.clone(),
dropped_records: selection.routed_usage_dropped_records,
+2
View File
@@ -186,6 +186,7 @@ pub(crate) async fn plan_turn_route(
let initial_routed_usage = auto_selection
.as_mut()
.map(|selection| crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: std::mem::take(&mut selection.routed_usage),
drop_records: std::mem::take(&mut selection.routed_usage_drop_records),
dropped_records: std::mem::take(&mut selection.routed_usage_dropped_records),
@@ -339,6 +340,7 @@ mod tests {
chrono::Utc::now(),
);
let batch = crate::cost_status::RuntimeUsageBatch {
decisions: Vec::new(),
records: vec![crate::cost_status::RuntimeUsageRecord {
source_id: "auto-router:plan-usage".to_string(),
usage: crate::cost_status::EffectiveRouteUsage {
+52 -5
View File
@@ -751,9 +751,9 @@ read from `TYPESAFE_API_KEY`, the `typesafe` secret-store entry, or
#### Decision routers (`kind = "decision"`)
A decision router asks a non-generative decision model one typed question per
turn — a Choice between the active provider's `fast` and `strong` tiers, plus a
thinking level — and gets calibrated probabilities back. No prose is parsed.
A decision router asks a non-generative decision model typed Choice questions
for the active provider's `fast` and `strong` tiers and thinking level, then
reads calibrated probabilities. No prose is parsed.
```toml
[auto.router]
@@ -771,8 +771,55 @@ min_confidence = 0.5 # default 0.5, clamped to 0..1
0.75, or the turn stays on the fast tier.
- An unknown `kind`, or a decision `provider` other than `openrouter` /
`typesafe`, leaves the router unconfigured and shown as failing.
- `thinking` is ignored for decision routers. OpenRouter spend is recorded like
any routed usage; TypeSafe-direct spend appears on the receipt only.
- `[auto.router] thinking` is ignored for decision routers. Both routes settle tokens through
the originating session's routed-usage ledger. TypeSafe is a named Custom route
with unknown billing; its price is never borrowed from the active chat provider.
Provider-reported cost remains verbatim on the decision receipt.
- OpenRouter uses `POST /api/alpha/decisions`; TypeSafe uses `POST /v1/systemone`.
The shared client validates Choice, Noul and Score against the offered questions
and bounds responses to 256 KiB. Malformed answers fail open with usage retained.
#### Shadow Decision Gate (experimental, off by default)
The Superfast Decision Gate asks a decision model three typed questions about
the latest user message — does it need a tool, can it be answered from the
conversation, and what is its intent — and logs a conservative recommendation.
It is shadow-only: it never changes routing, never skips or delays the model
call, and fails open on any error, timeout or malformed answer. It uses the same
System One client as the decision router above; there is no separate HTTP
client. It is configured from the environment and reads it when a turn starts:
```sh
SUPERFAST_ENABLED=1 # off unless set
SUPERFAST_PROVIDER=typesafe # or openrouter; required when enabled
SUPERFAST_BASE_URL=http://localhost:8000/v1 # optional TypeSafe-route base, e.g. self-hosted
SUPERFAST_MODEL=jev-latest # default jev-latest / ~typesafe/jev-latest
SUPERFAST_TIMEOUT_MS=150 # 1..=10000, default 150
```
- Enabling the gate never picks an endpoint by itself: without
`SUPERFAST_PROVIDER` nothing is sent and a warning is logged.
- The key comes from the same place the decision router reads it. The TypeSafe
route always authenticates, so a self-hosted server that ignores auth still
needs a placeholder `TYPESAFE_API_KEY`.
- Only the latest user message is sent, truncated to 4,000 characters and
redacted of configured secrets. The log (target `superfast`) carries the
route, failure class and latency, never prompt text.
- The gate retains the originating turn's accounting owner and cancellation.
Its tokens settle through the shared ledger; missing usage or a cancelled/timed
out request after dispatch creates an explicit coverage gap. Unknown TypeSafe
pricing is recorded as unpriced rather than free.
- Provider-reported cost survives rejected answers and late responses in bounded
receipts on the originating turn or session. These are diagnostic evidence;
unknown pricing never becomes an authoritative dollar total. Incomplete token
counters record a coverage gap while preserving the reported raw evidence.
The wire contracts are documented in [TypeSafe's OpenAPI schema](https://api.typesafe.ai/openapi.json)
and [OpenRouter's Decisions examples](https://openrouter.ai/blog/insights/what-is-jev/).
The Decision Gate concept and reference implementation are by Andrea Bruno,
released under CC BY 4.0:
[harness-superfast](https://github.com/Andrea-Bruno/harness-superfast).
Two `[auto]` keys shape routing (`AutoConfig` in `crates/tui/src/config.rs`):
+1
View File
@@ -42,6 +42,7 @@ notes, and relevant issue/PR comments.
**Merged or adapted contributions**
- **[Andrea-Bruno](https://github.com/Andrea-Bruno)** — designed the Superfast Decision Gate and contributed its off-by-default shadow classifier ([#6604](https://github.com/Hmbown/Codewhale/pull/6604), [#6603](https://github.com/Hmbown/Codewhale/issues/6603)).
- **[aiapienthusiast](https://github.com/aiapienthusiast)** — added Cheaper Inference to the bundled provider catalog ([#6761](https://github.com/Hmbown/Codewhale/pull/6761)).
- **[gaord](https://github.com/gaord)** — let a client fork a thread at a named turn ([#6580](https://github.com/Hmbown/Codewhale/pull/6580)), let undo roll back files for the turn it is undoing ([#6483](https://github.com/Hmbown/Codewhale/pull/6483)), stopped resume and fork from duplicating threads and sessions ([#6406](https://github.com/Hmbown/Codewhale/pull/6406)), exposed user-defined provider routes to native clients ([#6404](https://github.com/Hmbown/Codewhale/pull/6404)), and kept a fork going when a turn lost its tool call ([#6664](https://github.com/Hmbown/Codewhale/pull/6664)).
- **[Lstarsky0](https://github.com/Lstarsky0)** — moved the docs/work, legal, digest and FAQ pages onto the dictionary spine ([#6405](https://github.com/Hmbown/Codewhale/pull/6405), [#6417](https://github.com/Hmbown/Codewhale/pull/6417), [#6499](https://github.com/Hmbown/Codewhale/pull/6499), [#6574](https://github.com/Hmbown/Codewhale/pull/6574)), tightened the Chinese-branching ceiling to 18 ([#6403](https://github.com/Hmbown/Codewhale/pull/6403)), and made Fleet publish without a two-link window ([#6431](https://github.com/Hmbown/Codewhale/pull/6431)). Also moved the constitution page onto the dictionary spine and kept its install link in the selected locale ([#6733](https://github.com/Hmbown/Codewhale/pull/6733)).
+1
View File
@@ -20,6 +20,7 @@
/** Contributors whose PRs were merged or harvested into this release. */
export const RELEASE_CONTRIBUTORS: string[] = [
"@Andrea-Bruno",
"@aiapienthusiast",
"@gaord",
"@Lstarsky0",