play-faults fix: a fault's detail is fixed text by type; one timeout owner; pin the malformed-body and default-timeout legs
InferenceFault.detail is a &'static str, so no inference port (local, connector, SDK) can put an endpoint, request or reply text into a fault. The connector and SDK details that were built from error text are now fixed per kind or reason. LocalBinding loses its public response_timeout field; the short timeout the PlayTable canary needs comes from open_local_port_with_timeout, which exists only under the ghostlight test-support feature (a dungeon dev-dependency feature), so a release has one timeout, RESPONSE_TIMEOUT. Canary tests on the malformed-body path (port and PlayTable), on the SDK and connector ports, and on the default timeout at both resolving sites. Not yet built or tested at this commit. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
5a876bf615
commit
29efabe4d8
8 changed files with 377 additions and 120 deletions
|
|
@ -42,5 +42,6 @@ uuid.workspace = true
|
||||||
zeroize.workspace = true
|
zeroize.workspace = true
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
|
ghostlight = { path = "../ghostlight", features = ["test-support"] }
|
||||||
tempfile = "3"
|
tempfile = "3"
|
||||||
tower = { version = "0.5", features = ["util"] }
|
tower = { version = "0.5", features = ["util"] }
|
||||||
|
|
|
||||||
|
|
@ -1758,11 +1758,12 @@ impl PlayTable {
|
||||||
async fn close_with_fault(&self, turn: &mut PlayTurn, detail: String) -> Result<(), PlayError> {
|
async fn close_with_fault(&self, turn: &mut PlayTurn, detail: String) -> Result<(), PlayError> {
|
||||||
turn.state = PlayTurnState::Closed;
|
turn.state = PlayTurnState::Closed;
|
||||||
// R2: the one place a fault closes a turn, so the one place it is
|
// R2: the one place a fault closes a turn, so the one place it is
|
||||||
// logged, whole. The local inference port builds its detail from the
|
// logged, whole. An `InferenceFault`'s detail is a `&'static str`, so
|
||||||
// fault kind and never from the transport error's text (tests pin
|
// no inference port (local, connector or SDK) can have put an
|
||||||
// that for each local failure class). The connector and SDK ports
|
// endpoint, request or reply text into it; the type guarantees that,
|
||||||
// build details from their own error text, which this line does not
|
// not this line. The details of the other failures that reach here
|
||||||
// scrub.
|
// (kernel, request build, snapshot) are their callers' own error
|
||||||
|
// text, and nothing here constrains them.
|
||||||
tracing::warn!(turn_id = %turn.turn_id, detail = %detail, "play turn closed on a fault");
|
tracing::warn!(turn_id = %turn.turn_id, detail = %detail, "play turn closed on a fault");
|
||||||
turn.fault = Some(detail);
|
turn.fault = Some(detail);
|
||||||
// PA.f173: a fault can close a turn that was `AwaitingPlayer`, and
|
// PA.f173: a fault can close a turn that was `AwaitingPlayer`, and
|
||||||
|
|
@ -3303,7 +3304,7 @@ pub(crate) mod tests {
|
||||||
/// exercise `world.play`'s own staleness comparison against a question
|
/// exercise `world.play`'s own staleness comparison against a question
|
||||||
/// that actually opened, rather than one hand-built.
|
/// that actually opened, rather than one hand-built.
|
||||||
pub(crate) struct ScriptedPort {
|
pub(crate) struct ScriptedPort {
|
||||||
queue: StdMutex<VecDeque<Result<InferenceOutput, String>>>,
|
queue: StdMutex<VecDeque<Result<InferenceOutput, &'static str>>>,
|
||||||
seen: StdMutex<Vec<InferenceRequest>>,
|
seen: StdMutex<Vec<InferenceRequest>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -8226,18 +8227,15 @@ pub(crate) mod tests {
|
||||||
"gpt-5.6-sol".into(),
|
"gpt-5.6-sol".into(),
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let inference = ghostlight::open_inference(
|
let binding = ghostlight::LocalBinding {
|
||||||
None,
|
endpoint,
|
||||||
None,
|
model_prefix: "local/".into(),
|
||||||
Some(ghostlight::LocalBinding {
|
caller_runtime_id: "play-test-runtime".into(),
|
||||||
endpoint,
|
};
|
||||||
model_prefix: "local/".into(),
|
let inference = match timeout {
|
||||||
caller_runtime_id: "play-test-runtime".into(),
|
Some(timeout) => ghostlight::open_local_port_with_timeout(binding, timeout).unwrap(),
|
||||||
response_timeout: timeout,
|
None => ghostlight::open_inference(None, None, Some(binding), &["local/play"]).unwrap(),
|
||||||
}),
|
};
|
||||||
&["local/play"],
|
|
||||||
)
|
|
||||||
.unwrap();
|
|
||||||
let table = PlayTable::new(
|
let table = PlayTable::new(
|
||||||
fixture.world.clone(),
|
fixture.world.clone(),
|
||||||
personas,
|
personas,
|
||||||
|
|
@ -8262,6 +8260,7 @@ pub(crate) mod tests {
|
||||||
account: &str,
|
account: &str,
|
||||||
turn_number: u128,
|
turn_number: u128,
|
||||||
kind: &str,
|
kind: &str,
|
||||||
|
reply_canaries: &[&str],
|
||||||
) {
|
) {
|
||||||
let (logs, _guard) = capture_logs();
|
let (logs, _guard) = capture_logs();
|
||||||
let directory = tempfile::tempdir().unwrap();
|
let directory = tempfile::tempdir().unwrap();
|
||||||
|
|
@ -8277,7 +8276,10 @@ pub(crate) mod tests {
|
||||||
let log = logs.text();
|
let log = logs.text();
|
||||||
let debug = format!("{view:?} {:?}", table.store.lock().await.current().unwrap());
|
let debug = format!("{view:?} {:?}", table.store.lock().await.current().unwrap());
|
||||||
assert!(stored.contains(kind), "the error kind is kept: {stored}");
|
assert!(stored.contains(kind), "the error kind is kept: {stored}");
|
||||||
for canary in [endpoint.to_string(), endpoint.ip().to_string(), endpoint.port().to_string(), "chat/completions".to_owned()] {
|
let canaries = [endpoint.to_string(), endpoint.ip().to_string(), endpoint.port().to_string(), "chat/completions".to_owned()]
|
||||||
|
.into_iter()
|
||||||
|
.chain(reply_canaries.iter().map(|canary| (*canary).to_owned()));
|
||||||
|
for canary in canaries {
|
||||||
for (where_, text) in [("stored fault", &stored), ("warn line", &log), ("card row", &row), ("debug", &debug)] {
|
for (where_, text) in [("stored fault", &stored), ("warn line", &log), ("card row", &row), ("debug", &debug)] {
|
||||||
assert!(!text.contains(&canary), "{canary} leaked into the {where_}: {text}");
|
assert!(!text.contains(&canary), "{canary} leaked into the {where_}: {text}");
|
||||||
}
|
}
|
||||||
|
|
@ -8291,7 +8293,7 @@ pub(crate) mod tests {
|
||||||
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
listener.local_addr().unwrap()
|
listener.local_addr().unwrap()
|
||||||
};
|
};
|
||||||
assert_faulted_turn_names_no_endpoint(endpoint, None, "player-fault-canary", 431, "refused the connection").await;
|
assert_faulted_turn_names_no_endpoint(endpoint, None, "player-fault-canary", 431, "refused the connection", &[]).await;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The whole path with a timeout, the leg a Bonsai stall takes: the
|
/// The whole path with a timeout, the leg a Bonsai stall takes: the
|
||||||
|
|
@ -8314,6 +8316,71 @@ pub(crate) mod tests {
|
||||||
"player-fault-timeout",
|
"player-fault-timeout",
|
||||||
432,
|
432,
|
||||||
"timed out",
|
"timed out",
|
||||||
|
&[],
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Reads one HTTP request (headers and the declared body) off `stream`.
|
||||||
|
fn read_http_request(stream: &mut std::net::TcpStream) {
|
||||||
|
use std::io::Read;
|
||||||
|
let mut buf = Vec::new();
|
||||||
|
let mut chunk = [0u8; 8192];
|
||||||
|
loop {
|
||||||
|
let Ok(read) = stream.read(&mut chunk) else { break };
|
||||||
|
if read == 0 {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
buf.extend_from_slice(&chunk[..read]);
|
||||||
|
let text = String::from_utf8_lossy(&buf).to_string();
|
||||||
|
if let Some(split) = text.find("\r\n\r\n") {
|
||||||
|
let len = text[..split]
|
||||||
|
.to_ascii_lowercase()
|
||||||
|
.lines()
|
||||||
|
.find_map(|l| l.strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
|
||||||
|
.unwrap_or(0);
|
||||||
|
if buf.len() >= split + 4 + len {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Loopback server that answers every request with a `200` whose whole
|
||||||
|
/// body is `body`.
|
||||||
|
fn replying_with(body: &'static str) -> std::net::SocketAddr {
|
||||||
|
use std::io::Write;
|
||||||
|
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
|
let addr = listener.local_addr().unwrap();
|
||||||
|
std::thread::spawn(move || {
|
||||||
|
for stream in listener.incoming() {
|
||||||
|
let Ok(mut stream) = stream else { return };
|
||||||
|
read_http_request(&mut stream);
|
||||||
|
let _ = write!(
|
||||||
|
stream,
|
||||||
|
"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
|
||||||
|
body.len(),
|
||||||
|
body
|
||||||
|
);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
addr
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The malformed-reply leg through the whole path: a whole body that is
|
||||||
|
/// not the declared shape closes the turn on a fault, and serde's text
|
||||||
|
/// for it, which echoes the reply's own value, reaches the stored fault,
|
||||||
|
/// the warn line, the card row and Debug nowhere.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_malformed_local_reply_closes_the_turn_without_echoing_it() {
|
||||||
|
let endpoint = replying_with("\\"CANARY-reply-5d02c8\\"");
|
||||||
|
assert_faulted_turn_names_no_endpoint(
|
||||||
|
endpoint,
|
||||||
|
None,
|
||||||
|
"player-fault-malformed",
|
||||||
|
433,
|
||||||
|
"not the declared shape",
|
||||||
|
&["CANARY-reply-5d02c8"],
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
}
|
}
|
||||||
|
|
@ -8323,7 +8390,7 @@ pub(crate) mod tests {
|
||||||
/// answers each later one with an `end_turn` tool call. Returns its
|
/// answers each later one with an `end_turn` tool call. Returns its
|
||||||
/// address and the count of connections seen.
|
/// address and the count of connections seen.
|
||||||
fn closing_then_ending(drops: usize) -> (std::net::SocketAddr, Arc<std::sync::atomic::AtomicUsize>) {
|
fn closing_then_ending(drops: usize) -> (std::net::SocketAddr, Arc<std::sync::atomic::AtomicUsize>) {
|
||||||
use std::io::{Read, Write};
|
use std::io::Write;
|
||||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||||||
let addr = listener.local_addr().unwrap();
|
let addr = listener.local_addr().unwrap();
|
||||||
|
|
@ -8337,26 +8404,7 @@ pub(crate) mod tests {
|
||||||
drop(stream);
|
drop(stream);
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
let mut buf = Vec::new();
|
read_http_request(&mut stream);
|
||||||
let mut chunk = [0u8; 8192];
|
|
||||||
loop {
|
|
||||||
let Ok(read) = stream.read(&mut chunk) else { break };
|
|
||||||
if read == 0 {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
buf.extend_from_slice(&chunk[..read]);
|
|
||||||
let text = String::from_utf8_lossy(&buf).to_string();
|
|
||||||
if let Some(split) = text.find("\r\n\r\n") {
|
|
||||||
let len = text[..split]
|
|
||||||
.to_ascii_lowercase()
|
|
||||||
.lines()
|
|
||||||
.find_map(|l| l.strip_prefix("content-length:").map(|v| v.trim().parse::<usize>().unwrap()))
|
|
||||||
.unwrap_or(0);
|
|
||||||
if buf.len() >= split + 4 + len {
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let body = serde_json::json!({
|
let body = serde_json::json!({
|
||||||
"id": format!("resp-{n}"),
|
"id": format!("resp-{n}"),
|
||||||
"choices": [{"message": {"role": "assistant", "content": null,
|
"choices": [{"message": {"role": "assistant", "content": null,
|
||||||
|
|
|
||||||
|
|
@ -576,7 +576,6 @@ fn open_controller(
|
||||||
model_prefix: std::env::var("GHOSTLIGHT_LOCAL_MODEL_PREFIX")
|
model_prefix: std::env::var("GHOSTLIGHT_LOCAL_MODEL_PREFIX")
|
||||||
.unwrap_or_else(|_| DEFAULT_LOCAL_MODEL_PREFIX.into()),
|
.unwrap_or_else(|_| DEFAULT_LOCAL_MODEL_PREFIX.into()),
|
||||||
caller_runtime_id: runtime_id.to_owned(),
|
caller_runtime_id: runtime_id.to_owned(),
|
||||||
response_timeout: None,
|
|
||||||
}),
|
}),
|
||||||
Err(_) => None,
|
Err(_) => None,
|
||||||
};
|
};
|
||||||
|
|
|
||||||
|
|
@ -22,5 +22,10 @@ tracing.workspace = true
|
||||||
uuid.workspace = true
|
uuid.workspace = true
|
||||||
zeroize.workspace = true
|
zeroize.workspace = true
|
||||||
|
|
||||||
|
[features]
|
||||||
|
# Exposes `open_local_port_with_timeout` to a consumer's tests. Production
|
||||||
|
# builds never enable it.
|
||||||
|
test-support = []
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
tempfile = "3"
|
tempfile = "3"
|
||||||
|
|
|
||||||
|
|
@ -199,7 +199,12 @@ impl InferenceOutput {
|
||||||
pub struct InferenceFault {
|
pub struct InferenceFault {
|
||||||
disposition: InferenceFaultDisposition,
|
disposition: InferenceFaultDisposition,
|
||||||
class: InferenceFaultClass,
|
class: InferenceFaultClass,
|
||||||
detail: String,
|
/// Fixed text chosen by the port that built the fault. It is a
|
||||||
|
/// `&'static str`, so no constructor can be handed text that was
|
||||||
|
/// computed from a request, a reply, an endpoint or an error's own
|
||||||
|
/// `Display`: a fault names what kind of trouble it is and never carries
|
||||||
|
/// an input value. The type enforces this for every port.
|
||||||
|
detail: &'static str,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// What kind of trouble a fault is, as a typed value: the one place a reader
|
/// What kind of trouble a fault is, as a typed value: the one place a reader
|
||||||
|
|
@ -236,37 +241,37 @@ pub enum InferenceFaultDisposition {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl InferenceFault {
|
impl InferenceFault {
|
||||||
pub(super) fn new(detail: impl Into<String>) -> Self {
|
pub(super) fn new(detail: &'static str) -> Self {
|
||||||
Self {
|
Self {
|
||||||
disposition: InferenceFaultDisposition::RecoveryRequired,
|
disposition: InferenceFaultDisposition::RecoveryRequired,
|
||||||
class: InferenceFaultClass::Unclassified,
|
class: InferenceFaultClass::Unclassified,
|
||||||
detail: detail.into(),
|
detail: detail,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Transport trouble the owning flow may retry.
|
/// Transport trouble the owning flow may retry.
|
||||||
pub fn retryable(detail: impl Into<String>) -> Self {
|
pub fn retryable(detail: &'static str) -> Self {
|
||||||
Self {
|
Self {
|
||||||
disposition: InferenceFaultDisposition::Retryable,
|
disposition: InferenceFaultDisposition::Retryable,
|
||||||
class: InferenceFaultClass::Unclassified,
|
class: InferenceFaultClass::Unclassified,
|
||||||
detail: detail.into(),
|
detail: detail,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The one disposition that makes `ControllerError::requires_quarantine`
|
/// The one disposition that makes `ControllerError::requires_quarantine`
|
||||||
/// true for an `Inference` fault: the provider's answer cannot be trusted,
|
/// true for an `Inference` fault: the provider's answer cannot be trusted,
|
||||||
/// so the tick driver stops rather than commits.
|
/// so the tick driver stops rather than commits.
|
||||||
pub fn integrity_violation(detail: impl Into<String>) -> Self {
|
pub fn integrity_violation(detail: &'static str) -> Self {
|
||||||
Self {
|
Self {
|
||||||
disposition: InferenceFaultDisposition::IntegrityViolation,
|
disposition: InferenceFaultDisposition::IntegrityViolation,
|
||||||
class: InferenceFaultClass::Unclassified,
|
class: InferenceFaultClass::Unclassified,
|
||||||
detail: detail.into(),
|
detail: detail,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// The non-quarantining failure: this purpose cannot complete, but a
|
/// The non-quarantining failure: this purpose cannot complete, but a
|
||||||
/// sibling cell in the same tick is unaffected.
|
/// sibling cell in the same tick is unaffected.
|
||||||
pub fn recovery_required(detail: impl Into<String>) -> Self {
|
pub fn recovery_required(detail: &'static str) -> Self {
|
||||||
Self::new(detail)
|
Self::new(detail)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -457,7 +462,8 @@ impl PreparedInference {
|
||||||
request: InferenceRequest,
|
request: InferenceRequest,
|
||||||
) -> Result<PreparedInference, InferenceFault> {
|
) -> Result<PreparedInference, InferenceFault> {
|
||||||
let request_bytes =
|
let request_bytes =
|
||||||
serde_json::to_vec(&request).map_err(|error| InferenceFault::new(error.to_string()))?;
|
serde_json::to_vec(&request)
|
||||||
|
.map_err(|_| InferenceFault::new("an inference request could not be serialized"))?;
|
||||||
let purpose = request.purpose;
|
let purpose = request.purpose;
|
||||||
let invocation = CodexTransportInvocation::new(
|
let invocation = CodexTransportInvocation::new(
|
||||||
caller_runtime_id,
|
caller_runtime_id,
|
||||||
|
|
@ -465,7 +471,7 @@ impl PreparedInference {
|
||||||
Sha256::digest(request_bytes).into(),
|
Sha256::digest(request_bytes).into(),
|
||||||
request.provider,
|
request.provider,
|
||||||
)
|
)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("an inference invocation could not be built"))?;
|
||||||
Ok(PreparedInference {
|
Ok(PreparedInference {
|
||||||
purpose,
|
purpose,
|
||||||
invocation,
|
invocation,
|
||||||
|
|
@ -500,7 +506,7 @@ impl CodexConnectorInferencePort {
|
||||||
MAX_CONNECTOR_FRAME_BYTES,
|
MAX_CONNECTOR_FRAME_BYTES,
|
||||||
Some(RESPONSE_TIMEOUT),
|
Some(RESPONSE_TIMEOUT),
|
||||||
)
|
)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("the CodexConnector client could not be built"))?;
|
||||||
Ok(Self {
|
Ok(Self {
|
||||||
client,
|
client,
|
||||||
caller_runtime_id,
|
caller_runtime_id,
|
||||||
|
|
@ -513,10 +519,11 @@ impl CodexConnectorInferencePort {
|
||||||
caller_runtime_id: impl Into<String>,
|
caller_runtime_id: impl Into<String>,
|
||||||
) -> Result<Self, InferenceFault> {
|
) -> Result<Self, InferenceFault> {
|
||||||
let bytes = Zeroizing::new(
|
let bytes = Zeroizing::new(
|
||||||
std::fs::read(path.as_ref()).map_err(|error| InferenceFault::new(error.to_string()))?,
|
std::fs::read(path.as_ref())
|
||||||
|
.map_err(|_| InferenceFault::new("the CodexConnector key file could not be read"))?,
|
||||||
);
|
);
|
||||||
let raw = std::str::from_utf8(bytes.as_slice())
|
let raw = std::str::from_utf8(bytes.as_slice())
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("the CodexConnector key file is not UTF-8"))?;
|
||||||
let key = raw.trim_end_matches(['\r', '\n']);
|
let key = raw.trim_end_matches(['\r', '\n']);
|
||||||
if key.is_empty() || key.len() != raw.trim().len() {
|
if key.is_empty() || key.len() != raw.trim().len() {
|
||||||
return Err(InferenceFault::new(
|
return Err(InferenceFault::new(
|
||||||
|
|
@ -537,42 +544,51 @@ impl CodexConnectorInferencePort {
|
||||||
.execute(&request.invocation)
|
.execute(&request.invocation)
|
||||||
.map_err(|error| match error {
|
.map_err(|error| match error {
|
||||||
CodexConnectorClientError::Connection(_) => {
|
CodexConnectorClientError::Connection(_) => {
|
||||||
InferenceFault::retryable(error.to_string())
|
InferenceFault::retryable("the CodexConnector connection failed")
|
||||||
}
|
}
|
||||||
CodexConnectorClientError::InvalidConfig
|
CodexConnectorClientError::InvalidConfig
|
||||||
| CodexConnectorClientError::FrameSize
|
| CodexConnectorClientError::FrameSize
|
||||||
| CodexConnectorClientError::Encoding
|
| CodexConnectorClientError::Encoding
|
||||||
| CodexConnectorClientError::Transport(_) => {
|
| CodexConnectorClientError::Transport(_) => {
|
||||||
InferenceFault::integrity_violation(error.to_string())
|
InferenceFault::integrity_violation("the CodexConnector transport failed")
|
||||||
}
|
}
|
||||||
})?;
|
})?;
|
||||||
let (events, receipt) = match result.disposition {
|
let (events, receipt) = match result.disposition {
|
||||||
CodexTransportDisposition::Refused(reason) => {
|
CodexTransportDisposition::Refused(reason) => {
|
||||||
let detail = format!("CodexConnector refused request: {reason:?}");
|
|
||||||
return Err(match reason {
|
return Err(match reason {
|
||||||
CodexRefusal::InFlight | CodexRefusal::Capacity => {
|
CodexRefusal::InFlight => {
|
||||||
InferenceFault::retryable(detail)
|
InferenceFault::retryable("CodexConnector refused request: in flight")
|
||||||
}
|
}
|
||||||
CodexRefusal::Expired | CodexRefusal::Indeterminate => {
|
CodexRefusal::Capacity => {
|
||||||
InferenceFault::new(detail)
|
InferenceFault::retryable("CodexConnector refused request: at capacity")
|
||||||
}
|
}
|
||||||
CodexRefusal::IdentitySubstitution
|
CodexRefusal::Expired => {
|
||||||
| CodexRefusal::ProviderDigestSubstitution
|
InferenceFault::new("CodexConnector refused request: expired")
|
||||||
| CodexRefusal::Policy
|
}
|
||||||
| CodexRefusal::ReplayConflict
|
CodexRefusal::Indeterminate => {
|
||||||
| CodexRefusal::Malformed => InferenceFault::integrity_violation(detail),
|
InferenceFault::new("CodexConnector refused request: indeterminate")
|
||||||
|
}
|
||||||
|
CodexRefusal::IdentitySubstitution => InferenceFault::integrity_violation(
|
||||||
|
"CodexConnector refused request: identity substitution",
|
||||||
|
),
|
||||||
|
CodexRefusal::ProviderDigestSubstitution => InferenceFault::integrity_violation(
|
||||||
|
"CodexConnector refused request: provider digest substitution",
|
||||||
|
),
|
||||||
|
CodexRefusal::Policy => {
|
||||||
|
InferenceFault::integrity_violation("CodexConnector refused request: policy")
|
||||||
|
}
|
||||||
|
CodexRefusal::ReplayConflict => InferenceFault::integrity_violation(
|
||||||
|
"CodexConnector refused request: replay conflict",
|
||||||
|
),
|
||||||
|
CodexRefusal::Malformed => InferenceFault::integrity_violation(
|
||||||
|
"CodexConnector refused request: malformed",
|
||||||
|
),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
CodexTransportDisposition::Transported { events, receipt } => (events, receipt),
|
CodexTransportDisposition::Transported { events, receipt } => (events, receipt),
|
||||||
};
|
};
|
||||||
if let CodexTransportOutcome::Failed {
|
if let CodexTransportOutcome::Failed { .. } = &receipt.outcome {
|
||||||
failure_kind,
|
return Err(InferenceFault::new("the Codex provider reported a failed outcome"));
|
||||||
message,
|
|
||||||
} = &receipt.outcome
|
|
||||||
{
|
|
||||||
return Err(InferenceFault::new(format!(
|
|
||||||
"Codex provider failed ({failure_kind}): {message}"
|
|
||||||
)));
|
|
||||||
}
|
}
|
||||||
let events = events
|
let events = events
|
||||||
.into_iter()
|
.into_iter()
|
||||||
|
|
@ -590,7 +606,7 @@ impl CodexConnectorInferencePort {
|
||||||
})
|
})
|
||||||
.collect();
|
.collect();
|
||||||
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("the CodexConnector receipt could not be encoded"))?;
|
||||||
Ok(InferenceOutput {
|
Ok(InferenceOutput {
|
||||||
events,
|
events,
|
||||||
receipt_digest: format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
receipt_digest: format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
||||||
|
|
@ -612,14 +628,14 @@ impl InferencePort for CodexConnectorInferencePort {
|
||||||
let port = self.clone();
|
let port = self.clone();
|
||||||
tokio::task::spawn_blocking(move || port.execute(request))
|
tokio::task::spawn_blocking(move || port.execute(request))
|
||||||
.await
|
.await
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?
|
.map_err(|_| InferenceFault::new("the CodexConnector worker task failed"))?
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(super) fn unix_ms() -> Result<u64, InferenceFault> {
|
pub(super) fn unix_ms() -> Result<u64, InferenceFault> {
|
||||||
let millis = SystemTime::now()
|
let millis = SystemTime::now()
|
||||||
.duration_since(UNIX_EPOCH)
|
.duration_since(UNIX_EPOCH)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?
|
.map_err(|_| InferenceFault::new("the system clock is before the Unix epoch"))?
|
||||||
.as_millis();
|
.as_millis();
|
||||||
u64::try_from(millis).map_err(|_| InferenceFault::new("system time exceeds u64 milliseconds"))
|
u64::try_from(millis).map_err(|_| InferenceFault::new("system time exceeds u64 milliseconds"))
|
||||||
}
|
}
|
||||||
|
|
@ -6833,6 +6849,34 @@ fn purpose_name(purpose: InferencePurpose) -> &'static str {
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
|
||||||
|
/// The connector port's share of the no-input-text rule: a key-file path
|
||||||
|
/// and a runtime identity each arrive as a canary, and the fault names
|
||||||
|
/// only what failed. (The legs that need a live connector, a refusal or a
|
||||||
|
/// failed provider outcome, are fixed text by the fault type.)
|
||||||
|
#[test]
|
||||||
|
fn a_connector_fault_carries_no_text_it_was_handed() {
|
||||||
|
const CANARY: &str = "CANARY-connector-2c4d7a";
|
||||||
|
let endpoint: SocketAddr = "127.0.0.1:9".parse().unwrap();
|
||||||
|
let directory = tempfile::tempdir().unwrap();
|
||||||
|
let not_utf8 = directory.path().join(CANARY);
|
||||||
|
std::fs::write(¬_utf8, [0xff_u8, 0xfe]).unwrap();
|
||||||
|
let faults = [
|
||||||
|
CodexConnectorInferencePort::from_secret_file(endpoint, directory.path().join("absent").join(CANARY), "runtime")
|
||||||
|
.err()
|
||||||
|
.expect("a missing key file opened"),
|
||||||
|
CodexConnectorInferencePort::from_secret_file(endpoint, ¬_utf8, "runtime")
|
||||||
|
.err()
|
||||||
|
.expect("a key file that is not UTF-8 opened"),
|
||||||
|
CodexConnectorInferencePort::new(endpoint, "key".into(), format!(" {CANARY} "))
|
||||||
|
.err()
|
||||||
|
.expect("a runtime identity with whitespace opened"),
|
||||||
|
];
|
||||||
|
for fault in faults {
|
||||||
|
let rendered = format!("{fault} | {fault:?}");
|
||||||
|
assert!(!rendered.contains(CANARY), "{CANARY} leaked into {rendered}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// The decision lanes' tools go to the same provider under the same
|
/// The decision lanes' tools go to the same provider under the same
|
||||||
/// strict rules as the patch catalog; the interpreter's fixed tools and a
|
/// strict rules as the patch catalog; the interpreter's fixed tools and a
|
||||||
/// granted catalog entry are checked offline here for the same reason.
|
/// granted catalog entry are checked offline here for the same reason.
|
||||||
|
|
@ -11445,7 +11489,6 @@ mod tests {
|
||||||
endpoint: local_endpoint,
|
endpoint: local_endpoint,
|
||||||
model_prefix: local_model_prefix,
|
model_prefix: local_model_prefix,
|
||||||
caller_runtime_id: runtime_id,
|
caller_runtime_id: runtime_id,
|
||||||
response_timeout: None,
|
|
||||||
}),
|
}),
|
||||||
&models.each(),
|
&models.each(),
|
||||||
)
|
)
|
||||||
|
|
|
||||||
|
|
@ -97,6 +97,8 @@ pub use elaboration::{
|
||||||
};
|
};
|
||||||
pub use lens::{Lens, LensWeights};
|
pub use lens::{Lens, LensWeights};
|
||||||
pub use local_inference::{DEFAULT_LOCAL_MODEL_PREFIX, LocalBinding};
|
pub use local_inference::{DEFAULT_LOCAL_MODEL_PREFIX, LocalBinding};
|
||||||
|
#[cfg(feature = "test-support")]
|
||||||
|
pub use local_inference::open_local_port_with_timeout;
|
||||||
pub(crate) use mailbox::ElaborationPort;
|
pub(crate) use mailbox::ElaborationPort;
|
||||||
pub use mailbox::{ConsumerPort, ControllerPort, MailboxError, PlayPort, SeedPort, WorldMailbox};
|
pub use mailbox::{ConsumerPort, ControllerPort, MailboxError, PlayPort, SeedPort, WorldMailbox};
|
||||||
pub use patch::{
|
pub use patch::{
|
||||||
|
|
|
||||||
|
|
@ -31,10 +31,6 @@ pub struct LocalBinding {
|
||||||
pub endpoint: SocketAddr,
|
pub endpoint: SocketAddr,
|
||||||
pub model_prefix: String,
|
pub model_prefix: String,
|
||||||
pub caller_runtime_id: String,
|
pub caller_runtime_id: String,
|
||||||
/// How long one request may take, reply body included. `None` is
|
|
||||||
/// `RESPONSE_TIMEOUT`; a consumer or test sets it to reach the timeout
|
|
||||||
/// path without waiting that out.
|
|
||||||
pub response_timeout: Option<std::time::Duration>,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const LOCAL_RECEIPT_SCHEMA: &str = "ghostlight.local_inference_receipt.v1";
|
const LOCAL_RECEIPT_SCHEMA: &str = "ghostlight.local_inference_receipt.v1";
|
||||||
|
|
@ -135,6 +131,10 @@ pub(super) struct LocalInferencePort {
|
||||||
endpoint: SocketAddr,
|
endpoint: SocketAddr,
|
||||||
prefix: String,
|
prefix: String,
|
||||||
caller_runtime_id: String,
|
caller_runtime_id: String,
|
||||||
|
/// Read only by the tests that pin which timeout a constructor resolves
|
||||||
|
/// to; the live timeout is the client's.
|
||||||
|
#[cfg(test)]
|
||||||
|
response_timeout: std::time::Duration,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl LocalInferencePort {
|
impl LocalInferencePort {
|
||||||
|
|
@ -154,8 +154,10 @@ impl LocalInferencePort {
|
||||||
Self::with_timeout(endpoint, prefix, caller_runtime_id, RESPONSE_TIMEOUT)
|
Self::with_timeout(endpoint, prefix, caller_runtime_id, RESPONSE_TIMEOUT)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// `new` with the response timeout named, so a test can reach the timeout
|
/// `new` with the response timeout named. `new` is the only production
|
||||||
/// path without waiting out `RESPONSE_TIMEOUT`.
|
/// caller and passes `RESPONSE_TIMEOUT`, so production has one timeout
|
||||||
|
/// owner; the other callers are tests that reach the timeout path without
|
||||||
|
/// waiting `RESPONSE_TIMEOUT` out.
|
||||||
fn with_timeout(
|
fn with_timeout(
|
||||||
endpoint: SocketAddr,
|
endpoint: SocketAddr,
|
||||||
prefix: impl Into<String>,
|
prefix: impl Into<String>,
|
||||||
|
|
@ -175,6 +177,8 @@ impl LocalInferencePort {
|
||||||
endpoint,
|
endpoint,
|
||||||
prefix: prefix.into(),
|
prefix: prefix.into(),
|
||||||
caller_runtime_id: caller_runtime_id.into(),
|
caller_runtime_id: caller_runtime_id.into(),
|
||||||
|
#[cfg(test)]
|
||||||
|
response_timeout: timeout,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -188,7 +192,7 @@ impl LocalInferencePort {
|
||||||
.map_err(send_fault)?;
|
.map_err(send_fault)?;
|
||||||
let status = response.status();
|
let status = response.status();
|
||||||
if !status.is_success() {
|
if !status.is_success() {
|
||||||
let detail = format!("the local inference endpoint returned {status}");
|
let detail = "the local inference endpoint returned a non-success status";
|
||||||
let class = InferenceFaultClass::Status(status.as_u16());
|
let class = InferenceFaultClass::Status(status.as_u16());
|
||||||
return Err(if status.as_u16() == 429 || status.as_u16() == 503 {
|
return Err(if status.as_u16() == 429 || status.as_u16() == 503 {
|
||||||
InferenceFault::retryable(detail)
|
InferenceFault::retryable(detail)
|
||||||
|
|
@ -268,11 +272,8 @@ fn lower_request(prepared: &PreparedInference, prefix: &str) -> Result<Value, In
|
||||||
}
|
}
|
||||||
let mut tools = Vec::with_capacity(request.tools.len());
|
let mut tools = Vec::with_capacity(request.tools.len());
|
||||||
for tool in &request.tools {
|
for tool in &request.tools {
|
||||||
let parameters: Value = serde_json::from_str(&tool.parameters_json).map_err(|error| {
|
let parameters: Value = serde_json::from_str(&tool.parameters_json).map_err(|_| {
|
||||||
InferenceFault::integrity_violation(format!(
|
InferenceFault::integrity_violation("a tool's parameters are not valid JSON")
|
||||||
"tool `{}` parameters are not valid JSON: {error}",
|
|
||||||
tool.name
|
|
||||||
))
|
|
||||||
})?;
|
})?;
|
||||||
tools.push(json!({
|
tools.push(json!({
|
||||||
"type": "function",
|
"type": "function",
|
||||||
|
|
@ -366,7 +367,7 @@ fn assemble_output(
|
||||||
.unwrap_or_default(),
|
.unwrap_or_default(),
|
||||||
};
|
};
|
||||||
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("the local inference receipt could not be encoded"))?;
|
||||||
Ok(InferenceOutput::new(
|
Ok(InferenceOutput::new(
|
||||||
events,
|
events,
|
||||||
format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
||||||
|
|
@ -405,15 +406,32 @@ impl InferencePort for LocalInferencePort {
|
||||||
/// caller of this function, not only `open_inference`.
|
/// caller of this function, not only `open_inference`.
|
||||||
pub(super) fn open_local_port(
|
pub(super) fn open_local_port(
|
||||||
binding: LocalBinding,
|
binding: LocalBinding,
|
||||||
|
) -> Result<Arc<dyn InferencePort>, ControllerOpenError> {
|
||||||
|
Ok(Arc::new(local_port(binding)?))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The same port with a named response timeout, for a consumer's tests that
|
||||||
|
/// reach the timeout path without waiting `RESPONSE_TIMEOUT` out. It exists
|
||||||
|
/// only under the `test-support` feature, which production builds never
|
||||||
|
/// enable, so `LocalBinding` carries no timeout and a release has exactly
|
||||||
|
/// one: `RESPONSE_TIMEOUT`.
|
||||||
|
#[cfg(feature = "test-support")]
|
||||||
|
pub fn open_local_port_with_timeout(
|
||||||
|
binding: LocalBinding,
|
||||||
|
timeout: std::time::Duration,
|
||||||
) -> Result<Arc<dyn InferencePort>, ControllerOpenError> {
|
) -> Result<Arc<dyn InferencePort>, ControllerOpenError> {
|
||||||
Ok(Arc::new(LocalInferencePort::with_timeout(
|
Ok(Arc::new(LocalInferencePort::with_timeout(
|
||||||
binding.endpoint,
|
binding.endpoint,
|
||||||
binding.model_prefix,
|
binding.model_prefix,
|
||||||
binding.caller_runtime_id,
|
binding.caller_runtime_id,
|
||||||
binding.response_timeout.unwrap_or(RESPONSE_TIMEOUT),
|
timeout,
|
||||||
)?))
|
)?))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn local_port(binding: LocalBinding) -> Result<LocalInferencePort, ControllerOpenError> {
|
||||||
|
LocalInferencePort::new(binding.endpoint, binding.model_prefix, binding.caller_runtime_id)
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
@ -805,7 +823,6 @@ mod tests {
|
||||||
endpoint: "93.184.216.34:443".parse().unwrap(),
|
endpoint: "93.184.216.34:443".parse().unwrap(),
|
||||||
model_prefix: DEFAULT_LOCAL_MODEL_PREFIX.into(),
|
model_prefix: DEFAULT_LOCAL_MODEL_PREFIX.into(),
|
||||||
caller_runtime_id: TEST_RUNTIME.into(),
|
caller_runtime_id: TEST_RUNTIME.into(),
|
||||||
response_timeout: None,
|
|
||||||
}),
|
}),
|
||||||
&[TEST_MODEL],
|
&[TEST_MODEL],
|
||||||
)
|
)
|
||||||
|
|
@ -912,7 +929,6 @@ mod tests {
|
||||||
endpoint: "127.0.0.1:1".parse().unwrap(),
|
endpoint: "127.0.0.1:1".parse().unwrap(),
|
||||||
model_prefix: "claude".into(),
|
model_prefix: "claude".into(),
|
||||||
caller_runtime_id: TEST_RUNTIME.into(),
|
caller_runtime_id: TEST_RUNTIME.into(),
|
||||||
response_timeout: None,
|
|
||||||
}),
|
}),
|
||||||
&["claude-opus-5"],
|
&["claude-opus-5"],
|
||||||
)
|
)
|
||||||
|
|
@ -1195,7 +1211,6 @@ mod tests {
|
||||||
endpoint: "127.0.0.1:1".parse().unwrap(),
|
endpoint: "127.0.0.1:1".parse().unwrap(),
|
||||||
model_prefix: String::new(),
|
model_prefix: String::new(),
|
||||||
caller_runtime_id: TEST_RUNTIME.into(),
|
caller_runtime_id: TEST_RUNTIME.into(),
|
||||||
response_timeout: None,
|
|
||||||
}),
|
}),
|
||||||
&[TEST_MODEL],
|
&[TEST_MODEL],
|
||||||
)
|
)
|
||||||
|
|
@ -1659,6 +1674,48 @@ mod tests {
|
||||||
assert_eq!(fault.disposition(), InferenceFaultDisposition::IntegrityViolation, "{fault:?}");
|
assert_eq!(fault.disposition(), InferenceFaultDisposition::IntegrityViolation, "{fault:?}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The malformed-body leg of the no-input-text rule: serde's `Display`
|
||||||
|
/// for a type mismatch echoes the offending value, so a detail built from
|
||||||
|
/// it carries the reply's own text. The canary is a value inside a whole
|
||||||
|
/// body that fails to deserialize.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_malformed_body_carries_neither_its_own_text_nor_the_endpoint() {
|
||||||
|
const CANARY: &str = "CANARY-reply-7f3e91";
|
||||||
|
let addr = ScriptedResponder::start(vec![(200, format!("\"{CANARY}\""))]).await.endpoint();
|
||||||
|
let fault = infer_once(&port(addr)).await.expect_err("a malformed body produced an output");
|
||||||
|
assert_eq!(fault.class(), InferenceFaultClass::BadReply, "{fault:?}");
|
||||||
|
let rendered = format!("{fault} | {fault:?}");
|
||||||
|
for canary in [CANARY.to_owned(), addr.to_string(), addr.ip().to_string()] {
|
||||||
|
assert!(!rendered.contains(&canary), "{canary} leaked into {rendered}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The production default is `RESPONSE_TIMEOUT` (a cold model load lives
|
||||||
|
/// inside one request) at the constructor and at the port a binding
|
||||||
|
/// opens: a shorter default turns every cold start into a Timeout fault
|
||||||
|
/// that is never retried.
|
||||||
|
#[test]
|
||||||
|
fn the_production_default_timeout_is_response_timeout_at_both_resolving_sites() {
|
||||||
|
assert_eq!(RESPONSE_TIMEOUT, std::time::Duration::from_secs(900));
|
||||||
|
let endpoint: SocketAddr = "127.0.0.1:9".parse().unwrap();
|
||||||
|
let constructed = {
|
||||||
|
let _guard = client_build_lock().lock().expect("the client-build lock is never poisoned");
|
||||||
|
LocalInferencePort::new(endpoint, DEFAULT_LOCAL_MODEL_PREFIX, TEST_RUNTIME)
|
||||||
|
.expect("a loopback endpoint opens")
|
||||||
|
};
|
||||||
|
assert_eq!(constructed.response_timeout, RESPONSE_TIMEOUT);
|
||||||
|
let opened = {
|
||||||
|
let _guard = client_build_lock().lock().expect("the client-build lock is never poisoned");
|
||||||
|
local_port(LocalBinding {
|
||||||
|
endpoint,
|
||||||
|
model_prefix: DEFAULT_LOCAL_MODEL_PREFIX.into(),
|
||||||
|
caller_runtime_id: TEST_RUNTIME.into(),
|
||||||
|
})
|
||||||
|
.expect("a loopback binding opens")
|
||||||
|
};
|
||||||
|
assert_eq!(opened.response_timeout, RESPONSE_TIMEOUT);
|
||||||
|
}
|
||||||
|
|
||||||
/// The timeout leg of the no-endpoint rule: reqwest's `Display` for a
|
/// The timeout leg of the no-endpoint rule: reqwest's `Display` for a
|
||||||
/// timeout ends with the request URL, so a detail built from it names the
|
/// timeout ends with the request URL, so a detail built from it names the
|
||||||
/// canary endpoint this test bound. Covers both a stalled send and a
|
/// canary endpoint this test bound. Covers both a stalled send and a
|
||||||
|
|
|
||||||
|
|
@ -146,8 +146,27 @@ pub(super) enum SidecarFaultReason {
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SidecarFaultReason {
|
impl SidecarFaultReason {
|
||||||
fn into_fault(self, detail: String) -> InferenceFault {
|
/// The sidecar's own `detail` text is never carried: the reason names the
|
||||||
let detail = format!("SDK sidecar reported {self:?}: {detail}");
|
/// failure and the fault's text is fixed per reason.
|
||||||
|
fn into_fault(self) -> InferenceFault {
|
||||||
|
let detail = match self {
|
||||||
|
Self::RateLimited => "SDK sidecar reported rate limited",
|
||||||
|
Self::Overloaded => "SDK sidecar reported overloaded",
|
||||||
|
Self::ServerError => "SDK sidecar reported a server error",
|
||||||
|
Self::ApiTimeout => "SDK sidecar reported an API timeout",
|
||||||
|
Self::AuthenticationFailed => "SDK sidecar reported authentication failed",
|
||||||
|
Self::OrgNotAllowed => "SDK sidecar reported organization not allowed",
|
||||||
|
Self::BillingError => "SDK sidecar reported a billing error",
|
||||||
|
Self::InvalidRequest => "SDK sidecar reported an invalid request",
|
||||||
|
Self::ModelNotFound => "SDK sidecar reported model not found",
|
||||||
|
Self::MaxOutputTokens => "SDK sidecar reported max output tokens",
|
||||||
|
Self::MaxBudgetUsd => "SDK sidecar reported max budget",
|
||||||
|
Self::ExecutionError => "SDK sidecar reported an execution error",
|
||||||
|
Self::Unknown => "SDK sidecar reported an unknown failure",
|
||||||
|
Self::ProtocolViolation => "SDK sidecar reported a protocol violation",
|
||||||
|
Self::ToolRegistrationFailed => "SDK sidecar reported tool registration failed",
|
||||||
|
Self::TurnCapRefused => "SDK sidecar reported the turn cap refused",
|
||||||
|
};
|
||||||
match self {
|
match self {
|
||||||
Self::RateLimited | Self::Overloaded | Self::ServerError | Self::ApiTimeout => {
|
Self::RateLimited | Self::Overloaded | Self::ServerError | Self::ApiTimeout => {
|
||||||
InferenceFault::retryable(detail)
|
InferenceFault::retryable(detail)
|
||||||
|
|
@ -468,7 +487,7 @@ impl SdkInferencePort {
|
||||||
}
|
}
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
let _ = self.link.restart().await;
|
let _ = self.link.restart().await;
|
||||||
return Err(InferenceFault::new(error.to_string()));
|
return Err(InferenceFault::new("the tool-result owner refused a tool call"));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -482,10 +501,10 @@ impl SdkInferencePort {
|
||||||
SidecarFrame::Fault {
|
SidecarFrame::Fault {
|
||||||
query_id: seen,
|
query_id: seen,
|
||||||
reason,
|
reason,
|
||||||
detail,
|
..
|
||||||
} if seen == query_id => {
|
} if seen == query_id => {
|
||||||
let _ = self.link.restart().await;
|
let _ = self.link.restart().await;
|
||||||
return Err(reason.into_fault(detail));
|
return Err(reason.into_fault());
|
||||||
}
|
}
|
||||||
_ => {
|
_ => {
|
||||||
let _ = self.link.restart().await;
|
let _ = self.link.restart().await;
|
||||||
|
|
@ -501,13 +520,11 @@ impl SdkInferencePort {
|
||||||
fn link_fault(error: SidecarLinkError) -> InferenceFault {
|
fn link_fault(error: SidecarLinkError) -> InferenceFault {
|
||||||
match error {
|
match error {
|
||||||
SidecarLinkError::Closed => {
|
SidecarLinkError::Closed => {
|
||||||
InferenceFault::new("the SDK sidecar exited during a query".to_string())
|
InferenceFault::new("the SDK sidecar exited during a query")
|
||||||
}
|
}
|
||||||
SidecarLinkError::Spawn(detail) => {
|
SidecarLinkError::Spawn(_) => InferenceFault::new("the SDK sidecar could not start"),
|
||||||
InferenceFault::new(format!("the SDK sidecar could not start: {detail}"))
|
SidecarLinkError::Codec(_) => {
|
||||||
}
|
InferenceFault::integrity_violation("the SDK sidecar's framing was invalid")
|
||||||
SidecarLinkError::Codec(detail) => {
|
|
||||||
InferenceFault::integrity_violation(format!("SDK sidecar framing: {detail}"))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -526,9 +543,9 @@ pub(super) fn lower_query(
|
||||||
if let Some(effort) = request.reasoning_effort.as_deref()
|
if let Some(effort) = request.reasoning_effort.as_deref()
|
||||||
&& !SDK_EFFORT_LEVELS.contains(&effort)
|
&& !SDK_EFFORT_LEVELS.contains(&effort)
|
||||||
{
|
{
|
||||||
return Err(InferenceFault::integrity_violation(format!(
|
return Err(InferenceFault::integrity_violation(
|
||||||
"reasoning effort `{effort}` has no SDK counterpart"
|
"the requested reasoning effort has no SDK counterpart",
|
||||||
)));
|
));
|
||||||
}
|
}
|
||||||
let mut items = request.input.iter();
|
let mut items = request.input.iter();
|
||||||
let Some(CodexInputItem::UserText { text: prompt }) = items.next() else {
|
let Some(CodexInputItem::UserText { text: prompt }) = items.next() else {
|
||||||
|
|
@ -641,7 +658,7 @@ fn assemble_output(
|
||||||
total_cost_usd_estimate: material.total_cost_usd_estimate,
|
total_cost_usd_estimate: material.total_cost_usd_estimate,
|
||||||
};
|
};
|
||||||
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
let receipt_bytes = rmp_serde::to_vec_named(&receipt)
|
||||||
.map_err(|error| InferenceFault::new(error.to_string()))?;
|
.map_err(|_| InferenceFault::new("the SDK receipt could not be encoded"))?;
|
||||||
Ok(InferenceOutput {
|
Ok(InferenceOutput {
|
||||||
events: lowered,
|
events: lowered,
|
||||||
receipt_digest: format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
receipt_digest: format!("sha256:{:x}", Sha256::digest(&receipt_bytes)),
|
||||||
|
|
@ -763,9 +780,9 @@ impl InferencePort for RoutedInferencePort {
|
||||||
let model = request.provider_model().to_owned();
|
let model = request.provider_model().to_owned();
|
||||||
match self.route(&model) {
|
match self.route(&model) {
|
||||||
Some(port) => port.prepare(request),
|
Some(port) => port.prepare(request),
|
||||||
None => Err(InferenceFault::integrity_violation(format!(
|
None => Err(InferenceFault::integrity_violation(
|
||||||
"no configured inference backend claims the model `{model}`"
|
"no configured inference backend claims the requested model",
|
||||||
))),
|
)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -773,9 +790,9 @@ impl InferencePort for RoutedInferencePort {
|
||||||
let model = request.invocation.request.model.clone();
|
let model = request.invocation.request.model.clone();
|
||||||
match self.route(&model) {
|
match self.route(&model) {
|
||||||
Some(port) => port.infer(request).await,
|
Some(port) => port.infer(request).await,
|
||||||
None => Err(InferenceFault::integrity_violation(format!(
|
None => Err(InferenceFault::integrity_violation(
|
||||||
"no configured inference backend claims the model `{model}`"
|
"no configured inference backend claims the requested model",
|
||||||
))),
|
)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -1482,7 +1499,7 @@ mod tests {
|
||||||
fn soul_every_sidecar_reason_carries_its_named_disposition() {
|
fn soul_every_sidecar_reason_carries_its_named_disposition() {
|
||||||
use SidecarFaultReason::*;
|
use SidecarFaultReason::*;
|
||||||
for reason in [RateLimited, Overloaded, ServerError, ApiTimeout] {
|
for reason in [RateLimited, Overloaded, ServerError, ApiTimeout] {
|
||||||
let fault = reason.into_fault("detail".into());
|
let fault = reason.into_fault();
|
||||||
assert!(
|
assert!(
|
||||||
!fault.requires_recovery() && !fault.integrity_was_violated(),
|
!fault.requires_recovery() && !fault.integrity_was_violated(),
|
||||||
"{reason:?} was not retryable"
|
"{reason:?} was not retryable"
|
||||||
|
|
@ -1500,13 +1517,13 @@ mod tests {
|
||||||
Unknown,
|
Unknown,
|
||||||
] {
|
] {
|
||||||
assert!(
|
assert!(
|
||||||
reason.into_fault("detail".into()).requires_recovery(),
|
reason.into_fault().requires_recovery(),
|
||||||
"{reason:?} was not recovery-required"
|
"{reason:?} was not recovery-required"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
for reason in [ProtocolViolation, ToolRegistrationFailed, TurnCapRefused] {
|
for reason in [ProtocolViolation, ToolRegistrationFailed, TurnCapRefused] {
|
||||||
assert!(
|
assert!(
|
||||||
reason.into_fault("detail".into()).integrity_was_violated(),
|
reason.into_fault().integrity_was_violated(),
|
||||||
"{reason:?} did not violate integrity"
|
"{reason:?} did not violate integrity"
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
@ -1684,6 +1701,91 @@ mod tests {
|
||||||
assert!(fault.integrity_was_violated(), "{fault:?}");
|
assert!(fault.integrity_was_violated(), "{fault:?}");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn assert_names_no(fault: &InferenceFault, canary: &str) {
|
||||||
|
let rendered = format!("{fault} | {fault:?}");
|
||||||
|
assert!(!rendered.contains(canary), "{canary} leaked into {rendered}");
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The SDK port's share of the no-input-text rule: text the sidecar sent,
|
||||||
|
/// the link's own error text, a request's effort and a request's model
|
||||||
|
/// name each arrive as a canary, and none reaches the fault. The fault
|
||||||
|
/// type carries only fixed text, so this fails if a port ever routes a
|
||||||
|
/// computed string into it (through `Box::leak`, say).
|
||||||
|
#[tokio::test]
|
||||||
|
async fn an_sdk_fault_carries_no_text_it_was_handed() {
|
||||||
|
const CANARY: &str = "CANARY-sdk-91b6d0";
|
||||||
|
|
||||||
|
// The sidecar's own detail, for every reason it may name.
|
||||||
|
for reason in [
|
||||||
|
SidecarFaultReason::RateLimited,
|
||||||
|
SidecarFaultReason::Overloaded,
|
||||||
|
SidecarFaultReason::ServerError,
|
||||||
|
SidecarFaultReason::ApiTimeout,
|
||||||
|
SidecarFaultReason::AuthenticationFailed,
|
||||||
|
SidecarFaultReason::OrgNotAllowed,
|
||||||
|
SidecarFaultReason::BillingError,
|
||||||
|
SidecarFaultReason::InvalidRequest,
|
||||||
|
SidecarFaultReason::ModelNotFound,
|
||||||
|
SidecarFaultReason::MaxOutputTokens,
|
||||||
|
SidecarFaultReason::MaxBudgetUsd,
|
||||||
|
SidecarFaultReason::ExecutionError,
|
||||||
|
SidecarFaultReason::Unknown,
|
||||||
|
SidecarFaultReason::ProtocolViolation,
|
||||||
|
SidecarFaultReason::ToolRegistrationFailed,
|
||||||
|
SidecarFaultReason::TurnCapRefused,
|
||||||
|
] {
|
||||||
|
let link = ScriptedLink::new(vec![SidecarFrame::Fault {
|
||||||
|
query_id: 1,
|
||||||
|
reason,
|
||||||
|
detail: CANARY.into(),
|
||||||
|
}]);
|
||||||
|
let port = SdkInferencePort::new(link, TEST_RUNTIME);
|
||||||
|
let prepared = prose_prepared(&port, "Say something true.");
|
||||||
|
let fault = port.infer(prepared).await.expect_err("a fault frame produced an output");
|
||||||
|
assert_names_no(&fault, CANARY);
|
||||||
|
}
|
||||||
|
|
||||||
|
// The link's own error text.
|
||||||
|
assert_names_no(&link_fault(SidecarLinkError::Spawn(CANARY.into())), CANARY);
|
||||||
|
assert_names_no(&link_fault(SidecarLinkError::Codec(CANARY.into())), CANARY);
|
||||||
|
|
||||||
|
// An effort the SDK has no counterpart for.
|
||||||
|
let port = SdkInferencePort::new(ScriptedLink::new(Vec::new()), TEST_RUNTIME);
|
||||||
|
let mut prepared = prose_prepared(&port, "Say something true.");
|
||||||
|
prepared.invocation.request.reasoning_effort = Some(CANARY.into());
|
||||||
|
let fault = lower_query(1, &prepared, 1).expect_err("an unmapped effort lowered");
|
||||||
|
assert_names_no(&fault, CANARY);
|
||||||
|
|
||||||
|
// A model no backend claims, prepared and inferred through the router.
|
||||||
|
let directory = tempfile::tempdir().unwrap();
|
||||||
|
let entry = directory.path().join("main.js");
|
||||||
|
std::fs::write(&entry, "// sidecar").unwrap();
|
||||||
|
let routed = open_inference(
|
||||||
|
None,
|
||||||
|
Some(SdkBinding {
|
||||||
|
sidecar_entry: entry,
|
||||||
|
caller_runtime_id: TEST_RUNTIME.into(),
|
||||||
|
model_prefix: DEFAULT_SDK_MODEL_PREFIX.into(),
|
||||||
|
}),
|
||||||
|
None,
|
||||||
|
&models(TEST_MODEL).each(),
|
||||||
|
)
|
||||||
|
.expect("the SDK model routes");
|
||||||
|
let request = tool_request(
|
||||||
|
CommandId::new(),
|
||||||
|
0,
|
||||||
|
InferencePurpose::Persona,
|
||||||
|
CANARY,
|
||||||
|
"Respond only in natural prose.",
|
||||||
|
vec![CodexInputItem::UserText { text: "Say something true.".into() }],
|
||||||
|
Vec::<CodexToolDefinition>::new(),
|
||||||
|
RequestShape { max_output_tokens: 1_200, parallel_tool_calls: false },
|
||||||
|
)
|
||||||
|
.expect("the request builds");
|
||||||
|
let fault = routed.prepare(request).err().expect("an unroutable model prepared");
|
||||||
|
assert_names_no(&fault, CANARY);
|
||||||
|
}
|
||||||
|
|
||||||
/// Soul: the lowering's own gate table, every row.
|
/// Soul: the lowering's own gate table, every row.
|
||||||
#[test]
|
#[test]
|
||||||
fn soul_lower_query_refuses_the_shapes_the_gate_table_names() {
|
fn soul_lower_query_refuses_the_shapes_the_gate_table_names() {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue