Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 6 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -271,6 +271,12 @@ private_intra_doc_links = "warn"
redundant_explicit_links = "warn"
unescaped_backticks = "warn"

# Temporary rmcp decoder fix for custom extension results carrying protocol-generic
# `resultType` / `_meta`. Pinned to one immutable fork commit; remove once the
# equivalent fix ships in an upstream rmcp release.
[patch.crates-io]
rmcp = { git = "https://github.com/dinglebear-ai/rust-sdk", rev = "c9c3e518ce94b8d715519657c88195515978bda4" }

[profile.release]
lto = "thin"
codegen-units = 1
Expand Down
49 changes: 49 additions & 0 deletions crates/labby-gateway/src/gateway/shared.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,3 +57,52 @@ pub fn make_mcp_runtime_owner(subject: Option<&str>) -> UpstreamRuntimeOwner {
raw,
}
}

/// Whether transport-owned runtime provenance belongs in this action's params.
///
/// Runtime owner/origin are internal transport metadata, not universal gateway
/// parameters. Only actions that actually consume them should receive them.
/// Keeping this list shared prevents strict read-only parameter schemas (such as
/// `gateway.skills.list`) from rejecting otherwise valid API or MCP requests.
pub fn action_accepts_runtime_owner(action: &str) -> bool {
matches!(
action,
"gateway.add"
| "gateway.update"
| "gateway.remove"
| "gateway.reload"
| "gateway.import_tombstones.restore"
| "gateway.mcp.enable"
| "gateway.mcp.disable"
)
}

#[cfg(test)]
mod tests {
use super::action_accepts_runtime_owner;

#[test]
fn runtime_owner_metadata_is_limited_to_actions_that_consume_it() {
for action in [
"gateway.add",
"gateway.update",
"gateway.remove",
"gateway.reload",
"gateway.import_tombstones.restore",
"gateway.mcp.enable",
"gateway.mcp.disable",
] {
assert!(action_accepts_runtime_owner(action), "{action}");
}

for action in [
"gateway.skills.list",
"gateway.list",
"gateway.status",
"gateway.code_mode.get",
"gateway.oauth.resource_lease.create",
] {
assert!(!action_accepts_runtime_owner(action), "{action}");
}
}
}
4 changes: 2 additions & 2 deletions crates/labby-gateway/src/upstream/pool/skills.rs
Original file line number Diff line number Diff line change
Expand Up @@ -230,7 +230,7 @@ impl UpstreamPool {

match self.fetch_upstream_skills(&config.name, &peer).await {
Ok(skills) => {
let count = skills.skills.len();
let discovered_count = skills.discovered_count;
let skill_names = skills
.skills
.iter()
Expand All @@ -249,7 +249,7 @@ impl UpstreamPool {
let mut catalog = self.catalog.write().await;
if let Some(catalog_entry) = catalog.get_mut(&config.name) {
catalog_entry.supports_skills = Some(true);
catalog_entry.skill_count = count;
catalog_entry.skill_count = discovered_count;
catalog_entry.skill_names = skill_names;
}
}
Expand Down
49 changes: 36 additions & 13 deletions crates/labby-gateway/src/upstream/pool/skills_list.rs
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ use super::logging::{UpstreamRequestLog, log_upstream_request_start};
/// One upstream's validated skills plus what was dropped getting there.
#[derive(Debug, Clone, Default)]
pub(super) struct UpstreamSkills {
/// Skill candidates observed across the upstream pages fetched for this snapshot,
/// before host validation or exposure policy is applied.
pub(super) discovered_count: usize,
/// Skills that passed ingest validation.
pub(super) skills: Vec<ValidatedSkill>,
/// Skills dropped for integrity or budget reasons, by cause. Operators see
Expand Down Expand Up @@ -127,14 +130,28 @@ fn custom_result_value(result: ServerResult) -> Result<serde_json::Value, String
}
}

/// Validate and accumulate one page of entries, honoring the per-upstream cap.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SkillIngestCap {
ValidatedSkills,
Candidates,
}

/// Validate and accumulate one page of entries, honoring the per-upstream caps.
///
/// Returns `true` when the cap stopped accumulation, so the caller can stop
/// walking rather than fetching pages whose contents would be discarded.
fn ingest_page(entries: Vec<SkillEntry>, out: &mut UpstreamSkills) -> bool {
/// Returns the cap that stopped accumulation, so the caller can stop walking
/// rather than fetching pages whose contents would be discarded.
fn ingest_page(entries: Vec<SkillEntry>, out: &mut UpstreamSkills) -> Option<SkillIngestCap> {
// Discovery is what the upstream advertised, not what the host later accepts.
// Count the fetched page before validation so operator status can distinguish
// discovered candidates from the validated/exposed subset.
out.discovered_count = out.discovered_count.saturating_add(entries.len());
for entry in entries {
let processed_candidates = out.skills.len().saturating_add(out.excluded.len());
if processed_candidates >= limits::MAX_SKILL_CANDIDATES_PER_UPSTREAM {
return Some(SkillIngestCap::Candidates);
}
if out.skills.len() >= limits::MAX_SKILLS_PER_UPSTREAM {
return true;
return Some(SkillIngestCap::ValidatedSkills);
}
let uri = entry.uri.clone();
match validate_skill_entry(&entry) {
Expand All @@ -157,7 +174,7 @@ fn ingest_page(entries: Vec<SkillEntry>, out: &mut UpstreamSkills) -> bool {
Err(reason) => out.excluded.push((reason, uri)),
}
}
false
None
}

impl UpstreamPool {
Expand Down Expand Up @@ -218,14 +235,20 @@ impl UpstreamPool {
(current, next) => current.or(next),
};

let capped = ingest_page(result.skills, &mut out);
if capped {
if let Some(cap) = ingest_page(result.skills, &mut out) {
out.truncated = true;
tracing::warn!(
upstream = %upstream_name,
cap = limits::MAX_SKILLS_PER_UPSTREAM,
"upstream published more skills than the per-upstream cap — snapshot truncated"
);
match cap {
SkillIngestCap::ValidatedSkills => tracing::warn!(
upstream = %upstream_name,
cap = limits::MAX_SKILLS_PER_UPSTREAM,
"upstream published more validated skills than the per-upstream cap — snapshot truncated"
),
SkillIngestCap::Candidates => tracing::warn!(
upstream = %upstream_name,
cap = limits::MAX_SKILL_CANDIDATES_PER_UPSTREAM,
"upstream published more skill candidates than the validation cap — snapshot truncated"
),
}
break;
}

Expand Down
88 changes: 87 additions & 1 deletion crates/labby-gateway/src/upstream/pool/skills_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -168,7 +168,16 @@ impl ServerHandler for SkillsServer {
"skills/get" => {
self.get_calls.fetch_add(1, Ordering::SeqCst);
match self.get_entry.as_ref() {
Some(entry) => Ok(CustomResult::new(json!({ "skill": entry }))),
Some(entry) => Ok(CustomResult::new(json!({
"resultType": "complete",
"skill": entry,
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "depot-shaped-skills-server",
"version": "1.0.0"
}
}
}))),
None => Err(ErrorData::new(
ErrorCode::INVALID_PARAMS,
"unknown skill".to_string(),
Expand Down Expand Up @@ -210,6 +219,38 @@ async fn capability_detection_reads_the_declared_extension() {
));
}

#[tokio::test]
async fn skills_list_with_result_type_and_meta_stays_a_custom_result() {
let server = SkillsServer::new(vec![json!({
"resultType": "complete",
"skills": [entry("up", "alpha")],
"ttlMs": 60000,
"cacheScope": "private",
"_meta": {
"io.modelcontextprotocol/serverInfo": {
"name": "depot-shaped-skills-server",
"version": "1.0.0"
}
}
})]);
let pool = catalog_pool_with_server("up", server).await;

let skills = pool
.fetch_upstream_skills("up", &peer_for(&pool, "up").await)
.await
.expect("skills/list custom result remains intact");

assert_eq!(skills.skills.len(), 1);
assert_eq!(
skills.skills[0]
.entry
.frontmatter
.get("name")
.and_then(Value::as_str),
Some("alpha")
);
}

#[tokio::test]
async fn walks_every_page_and_stops_when_the_cursor_clears() {
let server = SkillsServer::new(vec![
Expand Down Expand Up @@ -302,6 +343,11 @@ async fn the_skill_cap_stops_the_walk_before_the_next_request() {
.expect("walk terminates");

assert!(skills.truncated);
assert_eq!(
skills.discovered_count,
limits::MAX_SKILLS_PER_UPSTREAM + 5,
"every candidate on the fetched page was discovered before the host cap engaged"
);
assert_eq!(skills.skills.len(), limits::MAX_SKILLS_PER_UPSTREAM);
assert_eq!(
calls.load(Ordering::SeqCst),
Expand All @@ -310,6 +356,45 @@ async fn the_skill_cap_stops_the_walk_before_the_next_request() {
);
}

#[tokio::test]
async fn the_candidate_cap_bounds_invalid_skill_floods() {
let big: Vec<Value> = (0..limits::MAX_SKILL_CANDIDATES_PER_UPSTREAM + 5)
.map(|i| {
let name = format!("invalid-{i}");
json!({
"uri": format!("skill://up/{name}/SKILL.md"),
"frontmatter": { "name": name, "description": "d" }
})
})
.collect();
let server = SkillsServer::new(vec![json!({ "skills": big, "nextCursor": "more" })]);
let calls = Arc::clone(&server.list_calls);
let pool = catalog_pool_with_server("up", server).await;

let skills = pool
.fetch_upstream_skills("up", &peer_for(&pool, "up").await)
.await
.expect("walk terminates");

assert!(skills.truncated);
assert_eq!(
skills.discovered_count,
limits::MAX_SKILL_CANDIDATES_PER_UPSTREAM + 5,
"the fetched page remains visible to operator discovery accounting"
);
assert!(skills.skills.is_empty());
assert_eq!(
skills.excluded_count(),
limits::MAX_SKILL_CANDIDATES_PER_UPSTREAM,
"invalid entries stop consuming rejection memory at the candidate cap"
);
assert_eq!(
calls.load(Ordering::SeqCst),
1,
"the next page is never fetched"
);
}

#[tokio::test]
async fn one_malformed_skill_does_not_sink_the_upstream() {
let good = entry("up", "alpha");
Expand All @@ -334,6 +419,7 @@ async fn one_malformed_skill_does_not_sink_the_upstream() {
.await
.expect("the upstream survives its own bad skills");

assert_eq!(skills.discovered_count, 3);
assert_eq!(skills.skills.len(), 1);
assert_eq!(skills.skills[0].name, "alpha");
assert_eq!(skills.excluded_count(), 2);
Expand Down
3 changes: 2 additions & 1 deletion crates/labby-gateway/src/upstream/types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -344,7 +344,8 @@ pub struct UpstreamEntry {
pub prompt_count: usize,
/// Last successfully discovered upstream resource count.
pub resource_count: usize,
/// Number of validated skills last enumerated from this upstream (SEP-2640).
/// Number of skill candidates observed during the last successful `skills/list` walk,
/// before Labby validation or exposure policy is applied.
pub skill_count: usize,
/// Cached validated skill names from the last successful skills/list walk.
pub skill_names: Vec<String>,
Expand Down
8 changes: 8 additions & 0 deletions crates/labby-runtime/src/skills/limits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,14 @@ use std::time::Duration;
/// Maximum skills retained from one upstream's `skills/list`, across all pages.
pub const MAX_SKILLS_PER_UPSTREAM: usize = 256;

/// Maximum skill candidates validated from one upstream's `skills/list`, across all pages.
///
/// This is deliberately higher than [`MAX_SKILLS_PER_UPSTREAM`]: malformed or
/// policy-rejected entries still consume validation work and rejection memory,
/// so an invalid-skill flood needs its own bound instead of relying on the
/// retained-valid-skill ceiling.
pub const MAX_SKILL_CANDIDATES_PER_UPSTREAM: usize = 1024;

/// Maximum entries in a single skill's `resources` manifest.
pub const MAX_RESOURCES_PER_SKILL: usize = 64;

Expand Down
Loading