use super::state::{DispatchInner, DispatchState, slot_key_named, slot_key_serving_task};
use super::transport::serves_session;
use super::*;
use onlyne_proto::adapter::HandoffArgs;
use onlyne_proto::{DETAILS_MAX_BYTES, ErrorCode, Report, ResBody};
pub(super) fn session_key_of_connection(inner: &DispatchInner, io: &AdapterIo) -> Option<String> {
let named = inner
.transports
.iter()
.find(|(_, (transport, _))| transport.same_connection(io))
.map(|(session, _)| session.clone())
.or_else(|| {
inner
.tools_mounts
.iter()
.find(|(_, bound)| bound.same_connection(io))
.map(|(key, _)| key.clone())
})?;
slot_key_named(inner, &named)
}
impl DispatchState {
pub const TOOLS_GONE_MESSAGE: &'static str = "token names no live session";
pub fn tools_gone() -> ResBody {
ResBody::err(
ErrorCode::Unauthorized,
Self::TOOLS_GONE_MESSAGE,
Some("token".to_string()),
)
}
pub fn stamp_tools_task(
&self,
io: &AdapterIo,
task_id: &mut String,
) -> std::result::Result<(), ResBody> {
let Some(scope) = self.tools_scope(io) else {
return Err(Self::tools_gone());
};
let Some(open) = scope.task_id else {
return Err(ResBody::err(
ErrorCode::Invalid,
"this session holds no open task",
Some("task_id".to_string()),
));
};
if task_id.is_empty() {
*task_id = open;
return Ok(());
}
if *task_id != open {
return Err(ResBody::err(
ErrorCode::Forbidden,
format!("this session serves task {open}, not {task_id}"),
Some("task_id".to_string()),
));
}
Ok(())
}
pub fn stamp_tools_send(
&self,
io: &AdapterIo,
envelope: &mut Envelope,
) -> std::result::Result<(), ResBody> {
let Some(scope) = self.tools_scope(io) else {
return Err(Self::tools_gone());
};
if scope.task_id.is_none() {
return Err(ResBody::err(
ErrorCode::Invalid,
"this session serves no delivery, so a send it makes belongs to no work",
None,
));
}
envelope.from = Principal::role(self.inner.lock().role.clone());
if envelope.kind == MsgKind::Task {
envelope.op_id = Some(onlyne_proto::new_op_id());
envelope.causality = Some(Causality::root(onlyne_proto::new_task_id()));
} else {
envelope.op_id = None;
envelope.causality = None;
}
Ok(())
}
}
pub(super) fn record_delivery(inner: &mut DispatchInner, key: &str, to: &Principal) {
let Some(role) = to.role_name() else {
return;
};
if let Some(slot) = inner.sessions.get_mut(key) {
slot.delivered_roles.insert(role.to_string());
}
}
fn relay_refusal(inner: &DispatchInner, key: &str) -> Option<String> {
let slot = inner.sessions.get(key)?;
if inner.required_targets.is_empty() {
return None;
}
let origin = slot.origin.as_ref().and_then(Principal::role_name);
let missing: Vec<&str> = inner
.required_targets
.iter()
.filter(|role| {
Some(role.as_str()) != origin && !slot.delivered_roles.contains(role.as_str())
})
.map(String::as_str)
.collect();
if missing.is_empty() {
return None;
}
let delivered = if slot.delivered_roles.is_empty() {
"none".to_string()
} else {
slot.delivered_roles
.iter()
.cloned()
.collect::<Vec<_>>()
.join(", ")
};
Some(format!(
"relay guard: missing handoff to: {} (this session delivered to: {delivered})",
missing.join(", "),
))
}
fn shape_refusal(report: &Report) -> Option<(String, &'static str)> {
let Report::Complete { details, files, .. } = report else {
return None;
};
if let Some(details) = details {
if details.len() > DETAILS_MAX_BYTES {
return Some((
format!("details exceeds the {DETAILS_MAX_BYTES}-byte cap"),
"details",
));
}
}
for path in files {
if !Path::new(path).is_absolute() {
return Some((format!("files must name absolute paths: {path}"), "files"));
}
}
None
}
impl DispatchState {
pub fn completion_refusal(&self, from: Option<&AdapterIo>, report: &Report) -> Option<ResBody> {
if let Some((message, field)) = shape_refusal(report) {
return Some(ResBody::err(
ErrorCode::Invalid,
message,
Some(field.to_string()),
));
}
let Report::Complete { .. } = report else {
return None;
};
let io = from?;
let inner = self.inner.lock();
let key = session_key_of_connection(&inner, io)?;
let message = relay_refusal(&inner, &key)?;
Some(ResBody::err(ErrorCode::Invalid, message, None))
}
pub fn handoff_refusal(&self, io: &AdapterIo, args: &HandoffArgs) -> Option<ResBody> {
let inner = self.inner.lock();
let key = slot_key_serving_task(&inner, &args.task_id)?;
if !serves_session(&inner, &key, io) {
return None;
}
hop_refusal(&inner, &key, "handoff")
}
}
fn hop_refusal(inner: &DispatchInner, key: &str, what: &str) -> Option<ResBody> {
let slot = inner.sessions.get(key)?;
let budget = slot.causality.hop_budget?;
let next = slot.causality.hop.saturating_add(1);
(next > budget).then(|| {
ResBody::err(
ErrorCode::Invalid,
format!("hop budget exhausted: this {what} would sit at hop {next} of {budget}"),
None,
)
})
}
#[cfg(test)]
mod tests {
use super::shape_refusal;
use onlyne_proto::{DETAILS_MAX_BYTES, Outcome, Report};
#[test]
fn a_completion_at_the_cap_passes_and_one_byte_over_it_is_refused() {
let at_cap = "x".repeat(DETAILS_MAX_BYTES);
let over = "x".repeat(DETAILS_MAX_BYTES + 1);
let at = shape_refusal(&complete(Some(at_cap.clone()), &[]));
assert_eq!(at, None, "the cap itself is inside the cap");
let over = shape_refusal(&complete(Some(over), &[]));
let (message, field) = over.expect("one byte over the cap is refused");
assert_eq!(field, "details", "the refusal names the field");
assert!(
message.contains(&DETAILS_MAX_BYTES.to_string()),
"the refusal names the cap it measured against, so the model can \
count its own bytes: {message}"
);
assert_eq!(shape_refusal(&complete(None, &[])), None);
}
#[test]
fn a_file_that_is_not_an_absolute_path_is_refused_by_name() {
let refused = shape_refusal(&complete(
None,
&["relative/out.txt".to_string(), "/abs/ok.txt".to_string()],
));
let (message, field) = refused.expect("a relative path is refused");
assert_eq!(field, "files", "the refusal names the field");
assert!(
message.contains("relative/out.txt"),
"the refusal quotes the path: {message}"
);
assert_eq!(
shape_refusal(&complete(None, &["/abs/a.png".to_string()])),
None,
"an absolute path is the shape the contract asks for"
);
assert_eq!(
shape_refusal(&complete(None, &[])),
None,
"no files, no rule"
);
}
#[test]
fn a_report_that_is_not_a_completion_is_never_refused_for_shape() {
let ready = Report::Ready {
task_id: "t-1".to_string(),
session_id: "s-1".to_string(),
generation: 1,
seq: 7,
cluster_ref: None,
};
assert_eq!(shape_refusal(&ready), None);
}
fn complete(details: Option<String>, files: &[String]) -> Report {
Report::Complete {
task_id: "t-1".to_string(),
outcome: Outcome::Done,
head: Some("done".to_string()),
details,
files: files.to_vec(),
reply_to: None,
cluster_ref: None,
}
}
}