Skip to content

Commit 2d26339

Browse files
MattJacksonMattJackson
committed
refactor(a2a): the front door's tap and gate become one screening step, in one file
Adding the tap beside the agent gate pushed `a2a/receive.rs` to 2504 lines, four over the impl-file cap `scripts/structure-lint.sh` enforces, and the gate went red for exactly that. The cap is the prompt, not the reason. The reason is that the two controls became a UNIT: a tap that saw a different projection from the gate's would be an audit record of a request nobody screened. They cannot disagree if the subject is built ONCE and both borrow it, and a shared construction with two consumers is a function. So the block moves to `a2a/screen.rs` whole, ordering and all — after admission, before the meter, the egress gate, the callback guard and the task row; tap before the verdict so a refused submission is still observed. Behaviour-preserving, and watched both ways: all five tap pins and the four gate pins stay green across the move, and deleting the new call site turns the inbound tap pin and both a2a gate pins red. receive.rs is now 2443 lines.
1 parent 65d175d commit 2d26339

3 files changed

Lines changed: 139 additions & 74 deletions

File tree

‎crates/busbar-core/src/a2a/mod.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,9 @@ pub(crate) mod relay;
9595
/// The plane's HTTP+JSON binding — the SECOND wire format, re-framed onto `ingress`'s one sequence.
9696
pub(crate) mod rest;
9797
pub(crate) mod rpcerror;
98+
/// THE FRONT DOOR'S HOOK SCREENING — the operator's tap and gate over one inbound submission, off
99+
/// ONE subject so the two controls cannot disagree about what the request was.
100+
pub(crate) mod screen;
98101
/// THIS PLANE'S REFUSAL VOCABULARY: `A2aWords`, the total match that gives every refusal
99102
/// `crate::ingress::protocol` decides a sentence in A2A's own error envelope, plus the three facts
100103
/// of its RFC 9728 document.

‎crates/busbar-core/src/a2a/receive.rs‎

Lines changed: 13 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ use super::words::{plane_absent, refuse_admission, A2aWords};
4242
use crate::state::{App, CurrentApp};
4343

4444
/// The audit action every inbound call on this plane records under.
45-
const AUDIT_ACTION: &str = "agent.call";
45+
pub(super) const AUDIT_ACTION: &str = "agent.call";
4646

4747
/// THE CREDENTIAL KIND THIS MOUNT CONFERS. `a2a_inbound` only when the plane is audience-bound;
4848
/// otherwise the empty string, which [`super::inbound::authorize`] refuses.
@@ -885,79 +885,18 @@ async fn admitted(
885885
// EVERY VERB, not only `message/send`. A gate an operator attached to an agent is a statement
886886
// about that agent, and a plane that fired it for submissions but not for the task verbs would
887887
// be a plane where the control's scope depends on which method a caller happened to use.
888-
// ── AND THE OPERATOR'S TAP, at the same point and off the same subject. ──────────────────────
889-
//
890-
// A global `kind: tap` hook observes EVERY request, and until this line it observed only the
891-
// model plane (`hooks-tap x a2a-server`, then `missing`). That was never a payload-contract problem:
892-
// a tap is shown the projection the gate below is shown, built by the same function from the
893-
// same `IrFacts`. FIRED BEFORE THE VERDICT so an audit tap sees the submissions that were
894-
// REFUSED as well as the ones that were relayed; it is spawned detached and can affect neither.
895-
let attached_gates = app.a2a_agent_gates.get(&admitted.dispatch.agent_id);
896-
if attached_gates.is_some() || !app.tap_hooks.is_empty() {
897-
// THE A2A SUBMISSION AS THE INVOKE IR: a caller names a target and hands it arguments,
898-
// which is what `ir::invoke` says it carries (`it carries A2A message/send alongside MCP
899-
// tools/call`). The target is the METHOD and the arguments are `params` — which is where a
900-
// message's `parts` live, so the prose a screening gate exists to read is inside the
901-
// projection rather than summarised beside it.
902-
let facts = crate::ir::invoke::InvokeReq {
903-
tool: super::local::method_of(&envelope).to_string(),
904-
arguments: envelope
905-
.get("params")
906-
.cloned()
907-
.unwrap_or(serde_json::Value::Null),
908-
extra: Default::default(),
909-
};
910-
let subject = crate::hooks::subject::RequestSubject {
911-
facts: &facts,
912-
container: &admitted.dispatch.agent_id,
913-
ingress_protocol: crate::plane::Plane::A2a.key(),
914-
request_id: app.next_request_id(),
915-
key: Some(key.as_ref()),
916-
};
917-
crate::hooks::tap::fire(
918-
&app.tap_hooks,
919-
&subject,
920-
key.group.as_deref(),
921-
&app.groups_registry,
922-
);
923-
if let Some(gates) = attached_gates {
924-
let verdict = crate::hooks::gate::decide(gates, &subject).await;
925-
if let crate::hooks::gate::GateVerdict::Reject {
926-
status,
927-
message,
928-
hook,
929-
} = verdict
930-
{
931-
crate::admin::audit::AUDIT.record_by(
932-
AUDIT_ACTION,
933-
&resource,
934-
crate::admin::audit::OUTCOME_REJECTED,
935-
&actor,
936-
);
937-
tracing::info!(
938-
agent = %admitted.dispatch.agent_id,
939-
hook,
940-
status,
941-
"a2a submission refused by a hook gate"
942-
);
943-
// THE HOOK'S STATUS, IN THIS PLANE'S ERROR VOCABULARY. A2A section 5.4 binds a JSON-RPC
944-
// code and a ProtoJSON body to every refusal, and a body in another plane's shape is a
945-
// body the TCK rejects by schema — so the code stays `UnsupportedOperation` (this
946-
// plane's binding for "busbar will not do this for you") and carries the hook's own
947-
// message, while the HTTP status is the gate's clamped one. Exactly what the egress
948-
// gate below already does with its own refusal.
949-
return (
950-
axum::http::StatusCode::from_u16(status)
951-
.unwrap_or(axum::http::StatusCode::FORBIDDEN),
952-
axum::Json(super::rpcerror::body(
953-
&rpc_id,
954-
super::rpcerror::A2aError::UnsupportedOperation,
955-
message,
956-
)),
957-
)
958-
.into_response();
959-
}
960-
}
888+
if let Some(refusal) = super::screen::screen_the_submission(
889+
&app,
890+
&envelope,
891+
&admitted.dispatch.agent_id,
892+
key,
893+
&rpc_id,
894+
&resource,
895+
&actor,
896+
)
897+
.await
898+
{
899+
return refusal;
961900
}
962901

963902
// 5. METER, before the work rather than after: an over-budget caller is refused instead of

‎crates/busbar-core/src/a2a/screen.rs‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// SPDX-License-Identifier: Apache-2.0
2+
// Copyright (C) 2026 Busbar Inc and contributors
3+
4+
//! THE A2A FRONT DOOR'S HOOK SCREENING — the operator's tap and the operator's gate, at one point,
5+
//! off one subject.
6+
//!
7+
//! ## Why this is a module and not eight lines in `receive.rs`
8+
//!
9+
//! It became a UNIT the moment the tap joined the gate here. Before that there was one control with
10+
//! one branch; now there are two controls that must agree, by construction, about WHAT THE REQUEST
11+
//! WAS — a tap that saw a different projection from the gate's would be an audit record of a
12+
//! request nobody screened. The way to make them unable to disagree is to build the subject ONCE
13+
//! and hand the same borrow to both, and a shared construction with two consumers is a function.
14+
//!
15+
//! It also keeps `receive.rs` under the impl-file cap that `scripts/structure-lint.sh` enforces,
16+
//! which is the same argument one level down: the front door's job is to admit, meter and relay,
17+
//! and the screening step is a thing it CALLS rather than a thing it contains.
18+
//!
19+
//! ## The ordering, which is the load-bearing part
20+
//!
21+
//! Called AFTER admission (the agent is what an attach is keyed on, so there is nothing to look up
22+
//! before it) and BEFORE the meter, the egress gate, the callback guard and the task row.
23+
//! Everything after it either spends the caller's budget, leases busbar's own credential, or mints
24+
//! durable state; a refusal must cost none of them.
25+
//!
26+
//! The TAP fires BEFORE the verdict, so an audit tap sees the submissions that were REFUSED as well
27+
//! as the ones that were relayed. It is spawned detached and can affect neither the verdict nor the
28+
//! answer.
29+
//!
30+
//! EVERY VERB, not only `message/send`. A gate an operator attached to an agent is a statement
31+
//! about that agent, and a plane that fired it for submissions but not for the task verbs would be
32+
//! a plane where the control's scope depends on which method a caller happened to use.
33+
34+
use axum::response::{IntoResponse, Response};
35+
36+
use crate::state::App;
37+
38+
/// Run the front door's taps and gate over one submission.
39+
///
40+
/// Returns `Some(response)` ONLY when a gate refused, in which case the caller must return it
41+
/// unchanged and do nothing else. `None` means "not refused" — it does not mean "no hook ran".
42+
pub(super) async fn screen_the_submission(
43+
app: &App,
44+
envelope: &serde_json::Value,
45+
agent_id: &str,
46+
key: &busbar_api::VirtualKey,
47+
rpc_id: &serde_json::Value,
48+
resource: &str,
49+
actor: &str,
50+
) -> Option<Response> {
51+
let attached_gates = app.a2a_agent_gates.get(agent_id);
52+
if attached_gates.is_none() && app.tap_hooks.is_empty() {
53+
return None;
54+
}
55+
56+
// THE A2A SUBMISSION AS THE INVOKE IR: a caller names a target and hands it arguments, which is
57+
// what `ir::invoke` says it carries (`it carries A2A message/send alongside MCP tools/call`).
58+
// The target is the METHOD and the arguments are `params` — which is where a message's `parts`
59+
// live, so the prose a screening gate exists to read is inside the projection rather than
60+
// summarised beside it.
61+
let facts = crate::ir::invoke::InvokeReq {
62+
tool: super::local::method_of(envelope).to_string(),
63+
arguments: envelope
64+
.get("params")
65+
.cloned()
66+
.unwrap_or(serde_json::Value::Null),
67+
extra: Default::default(),
68+
};
69+
// ONE subject, TWO consumers — see the module header. The tap is shown the projection the gate
70+
// is shown, built by the same function from the same `IrFacts`.
71+
let subject = crate::hooks::subject::RequestSubject {
72+
facts: &facts,
73+
container: agent_id,
74+
ingress_protocol: crate::plane::Plane::A2a.key(),
75+
request_id: app.next_request_id(),
76+
key: Some(key),
77+
};
78+
crate::hooks::tap::fire(
79+
&app.tap_hooks,
80+
&subject,
81+
key.group.as_deref(),
82+
&app.groups_registry,
83+
);
84+
85+
let gates = attached_gates?;
86+
let crate::hooks::gate::GateVerdict::Reject {
87+
status,
88+
message,
89+
hook,
90+
} = crate::hooks::gate::decide(gates, &subject).await
91+
else {
92+
return None;
93+
};
94+
95+
crate::admin::audit::AUDIT.record_by(
96+
super::receive::AUDIT_ACTION,
97+
resource,
98+
crate::admin::audit::OUTCOME_REJECTED,
99+
actor,
100+
);
101+
tracing::info!(
102+
agent = %agent_id,
103+
hook,
104+
status,
105+
"a2a submission refused by a hook gate"
106+
);
107+
// THE HOOK'S STATUS, IN THIS PLANE'S ERROR VOCABULARY. A2A section 5.4 binds a JSON-RPC code
108+
// and a ProtoJSON body to every refusal, and a body in another plane's shape is a body the TCK
109+
// rejects by schema — so the code stays `UnsupportedOperation` (this plane's binding for
110+
// "busbar will not do this for you") and carries the hook's own message, while the HTTP status
111+
// is the gate's clamped one. Exactly what the egress gate does with its own refusal.
112+
Some(
113+
(
114+
axum::http::StatusCode::from_u16(status).unwrap_or(axum::http::StatusCode::FORBIDDEN),
115+
axum::Json(super::rpcerror::body(
116+
rpc_id,
117+
super::rpcerror::A2aError::UnsupportedOperation,
118+
message,
119+
)),
120+
)
121+
.into_response(),
122+
)
123+
}

0 commit comments

Comments
 (0)