alkdev/alkcall
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(review 008 Unit 1): establisher reply projection (U-2, ADR-049 amendment 3)

Browse Source 
The establisher can now contribute to the open-op success reply — the
establisher → opener direction amendment 2 left empty. First consumer:
alktunnels ADR-008's bind-first listen establisher, whose observed
OS-chosen bound address rides the reply as an additive `bound` field
(the SOCKS5 BIND reply#1 BND.ADDR fidelity ask).

- Establishment gains reply_fields (private, builder-constructed):
  with_reply_field / with_reply_fields / reply_fields. The map shape
  generalizes beyond `bound` without a fourth amendment; the
  #[non_exhaustive] carrier (amendment 2) makes the extension additive
  — no construction-site break.
- run_open_wrapper merges contributed fields into the success output
  after reserving `channel_id`. The key is wrapper-owned: an
  establisher supplying it fails the open loudly
  (channel:open_failed, reason handler_error, message naming the
  reserved key), tearing the just-allocated channel down — never
  shadowing. Absent fields leave the reply byte-identical to the
  pre-amendment shape (asserted exactly).
- ChannelClient::open_channel_with_reply returns
  (channel_id, reply, send, recv) — the full success output — so
  consumers read `bound` first-class; open_channel delegates and
  discards the fields, signature unchanged.

Tests (all gates from the review + plan):
- projection produces { channel_id, bound } on the wire; no-fields
  replies are byte-identical (establisher and no-establisher shapes)
- reserved-key establisher fails with reason handler_error; channel
  torn down, ledger decremented, pump handler never spawned
- registry-level register_openable_with_establisher projects fields
- e2e over a real channels connection: bound reaches
  open_channel_with_reply; open_channel unchanged against the same
  accept side
- existing establishment tests (establisher-success round-trip,
  plan-flow, no-establisher compat) pass unchanged

Docs: ADR-049 amendment 3 (projection, reservation, read path,
compatibility posture, door type).

Verification: cargo test 637 passed; clippy -D warnings clean;
cargo fmt --check clean; cargo doc --no-deps clean.
This commit is contained in:
glm-5.3-flash
2026-09-16 10:27:12 +00:00
parent 1bc937c061
commit 82ddddf986
3 changed files with 560 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture/decisions/049-channel-open-establishment-phase.md
+68 -4
View File
@@ -6,7 +6,8 @@ Accepted — implemented in alkcall 0.5.0 (Unit 1: E-01 + N-1; see the
"Amendment (Unit 1 implementation, 2026-09-06)" at the bottom).
Amends ADR-047 §3 — the open-op wrapper gains an awaited
establishment phase ahead of the spawned pump handler; resolves review
006 E-01 and N-1
006 E-01 and N-1. Amendment 3 (2026-09-16, review 008 U-2) adds the
establisher reply projection — see "Amendment 3" below.
## Context
@@ -379,8 +380,8 @@ is unfixable within that shape; alktty documented the same wall
(backend `allocate` cannot cross, so failure classes stayed in-band —
the phantom-channel shape ADR-049 removed, alive one layer down).
The plan is now real: `Establishment { plan: Option<ChannelPlan> }`
with `ChannelPlan = Arc<dyn Any + Send + Sync>` — **typed-opaque, not
The plan is now real: `Establishment { plan: Option<ChannelPlan> }` with
`ChannelPlan = Arc<dyn Any + Send + Sync>` — **typed-opaque, not
`serde_json::Value`**. The review's `Option<Value>` sketch could not
satisfy its own verification gate ("establisher dials, `plan` carries
the handle"): the payloads establishers actually hand off are live
@@ -414,4 +415,67 @@ and the registration entry points, plus a `debug!` telemetry line in
`run_open_wrapper` when a handler exits without having accepted the
channel's `BiStream` (the birth-teardown hint; the accept is
observable in-process via the yield-once source). §6's pinned
EOF-shaped panic semantics are unchanged.
EOF-shaped panic semantics are unchanged.
## Amendment 3 (establisher reply projection, 2026-09-16 — review 008 U-2)
Amendment 2 filled the establisher → handler direction (the plan);
the establisher → opener direction stayed empty — the wrapper
hardcodes the success reply to `channel_id`
(`run_open_wrapper`'s `ResponseEnvelope::ok`). alktunnels ADR-008's
bind-first listen establisher ends establishment at bind time, and the
observed OS-chosen bound address must ride the open-op reply as an
additive `"bound"` field (the SOCKS5 BIND reply#1 `BND.ADDR` fidelity
ask) — the alternative (an out-of-band query op) is a new wire
surface, strictly worse than an optional reply field.
**The decision: `Establishment` gains optional reply fields the
wrapper merges into the success output.**
- The carrier: `Establishment.reply_fields: Option<Map<String,
Value>>` (private, builder-constructed) —
`Establishment::new(plan).with_reply_field("bound", json!({...}))`,
plus `with_reply_fields(map)` (batch) and `reply_fields()` (read).
The map shape generalizes beyond `bound` without a fourth
amendment; `#[non_exhaustive]` (amendment 2) made the carrier
extension additive — no construction-site break.
- The merge: on establisher success the wrapper extends the success
output with the contributed fields AFTER reserving `channel_id`.
**The `channel_id` reservation:** the key is wrapper-owned; an
establisher supplying it is an establisher bug — the wrapper fails
the open loudly (`channel:open_failed`, reason `handler_error`,
message naming the reserved key) and tears the just-allocated
channel down, never silently shadowing its own value. Contributed
fields otherwise merge flat; the output schema remains the op's own
concern (the producing crate documents its optional fields —
alktunnels' listen-op spec documents `bound`).
- The client read path: `ChannelClient::open_channel_with_reply`
returns `(channel_id, reply, send, recv)` — the full success
output — so a consumer reads `bound` without dropping to
`call_open_op` + manual `adopt_channel`;
`open_channel` delegates and discards the extra fields, signature
unchanged.
- Compatibility: absent reply fields leave the reply byte-identical
to the pre-amendment shape (`{ channel_id }`) — verified by test.
Old consumers (`open_channel` extracting `channel_id`, ignoring
unknown fields) are unaffected by a NEW field; new consumers reading
`bound` against an OLD alkcall see the field absent — the additive
posture alktunnels ADR-008 §2 pins.
Door type: the reply fields are an additive call-plane JSON surface
(like `channel:open_failed`'s `details` in §3 — consumers ignore
unknown fields by the envelope's own parse posture); the field set an
op may contribute is the producing crate's schema concern, not a
protocol vocabulary. No data-plane change.
Implemented surface (alkcall 0.7.2): `Establishment::with_reply_field`
/ `with_reply_fields` / `reply_fields`, the wrapper's
reservation check (`RESERVED_REPLY_KEY` = `channel_id`) +
`merge_reply_fields`, `ChannelClient::open_channel_with_reply`.
Verification gates landed as tests: the projection produces
`{ channel_id, bound }` on the wire; no-fields (establisher or not)
produces exactly `{ channel_id }` (byte-identical); the reserved-key
establisher fails with reason `handler_error`, channel torn down,
ledger decremented, handler never spawned; the e2e gate carries
`bound` through a real channels connection to
`open_channel_with_reply` while `open_channel` stays unchanged.
src/channels/client.rs
+143 -1
View File
@@ -346,6 +346,26 @@ impl ChannelClient {
input: Value,
alpn: &str,
) -> Result<(u32, MpscSendStream, MpscRecvStream), ChannelOpenError> {
let (channel_id, _reply, send, recv) = self
.open_channel_with_reply(operation_id, input, alpn)
.await?;
Ok((channel_id, send, recv))
}
/// Open a channel and return the open-op success reply's extra
/// fields alongside the streams (ADR-049 amendment 3). The reply
/// is the full success output — `channel_id` plus any
/// establisher-contributed fields (e.g. a bind-first listener's
/// `bound`); a producer that contributed none returns just
/// `{ channel_id }`, so the payload's `Value` is the old reply.
/// [`ChannelClient::open_channel`] delegates here and discards the
/// reply, keeping its signature unchanged.
pub async fn open_channel_with_reply(
&self,
operation_id: &str,
input: Value,
alpn: &str,
) -> Result<(u32, Value, MpscSendStream, MpscRecvStream), ChannelOpenError> {
let response = self.call_open_op(operation_id, input).await;
let out = response
.result
@@ -359,7 +379,7 @@ impl ChannelClient {
.adopt_channel(channel_id, alpn, None)
.await
.map_err(ChannelOpenError::AdoptFailed)
.map(|(send, recv)| (channel_id, send, recv))
.map(|(send, recv)| (channel_id, out, send, recv))
}
/// Take the `CallConnection` — used by the consumer to register
@@ -1625,6 +1645,128 @@ mod tests {
assert_eq!(&data, b"ping", "handler received ping post-establishment");
}
/// ADR-049 amendment 3 (review 008 U-2) acceptance gate: the
/// establisher contributes a reply field (`bound`, the bind-first
/// listener shape) and the client's `open_channel_with_reply`
/// receives it end-to-end over a real channels connection, while
/// `open_channel` keeps its pre-amendment shape (fields discarded).
#[tokio::test]
async fn establisher_reply_fields_reach_open_channel_with_reply_end_to_end() {
use crate::channels::operations::{
ChannelCore, Establishment, OpenEstablisher, OpenHandler,
};
use crate::channels::policy::NoCap;
use crate::registry::spec::ChannelOpenSpec;
let establisher: OpenEstablisher = Arc::new(|_input, _auth| {
Box::pin(async {
Ok(Establishment::default().with_reply_field(
"bound",
serde_json::json!({ "host": "203.0.113.9", "port": 42113 }),
))
})
});
let open_handler: OpenHandler =
Arc::new(|_input, _plan, _channel_conn, _auth| tokio::spawn(async {}));
let establisher_for_hook = Arc::clone(&establisher);
let open_handler_for_hook = Arc::clone(&open_handler);
let install_hook: crate::channels::adapter::InstallChannelZero =
Arc::new(move |manager, channel0_conn, auth| {
let establisher = Arc::clone(&establisher_for_hook);
let open_handler = Arc::clone(&open_handler_for_hook);
tokio::spawn(async move {
let channel0_bidi = match channel0_conn.accept_bi().await {
Ok(s) => s,
Err(_) => return,
};
let (writer, reader) = split_single_stream(channel0_bidi);
let core = ChannelCore::new(manager, Arc::new(NoCap));
let registry = crate::registry::registration::OperationRegistry::new();
let spec = OperationSpec::new(
"channels/tty/sub",
OperationType::Sub,
Visibility::External,
serde_json::json!({}),
serde_json::json!({
"type": "object",
"properties": {
"channel_id": { "type": "integer" },
"bound": { "type": "object" }
}
}),
vec![],
AccessControl::default(),
None,
)
.with_channel_open(ChannelOpenSpec::new("alk/tty"));
core.register_openable_with_establisher(
spec,
Some(establisher),
open_handler,
&registry,
auth.clone(),
None,
)
.expect("register_openable_with_establisher");
let registry = Arc::new(registry);
let provider: Arc<dyn IdentityProvider> = Arc::new(NoopIdProvider);
let call_connection = Arc::new(CallConnection::new_single_stream(
channel0_conn,
Arc::clone(&writer),
));
let dp = Dispatcher::new(registry, provider);
dp.run_loop_single_stream(call_connection, reader, writer)
.await;
})
});
let (client_end, server_end) = tokio::io::duplex(64 * 1024);
let client_conn =
Connection::from_bidi(client_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let server_conn =
Connection::from_bidi(server_end, b"alk/channels".to_vec(), Some(TEST_ADDR));
let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
let auth = AuthContext::anonymous(b"alk/channels");
let _server_handle = tokio::spawn(async move {
let _ = crate::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
});
let client = ChannelClient::from_connection(client_conn)
.await
.expect("channel client init");
let (channel_id, reply, mut send, _recv) = client
.open_channel_with_reply(
"channels/tty/sub",
serde_json::json!({ "container": "abc" }),
"alk/tty",
)
.await
.expect("open_channel_with_reply");
assert!(channel_id > 0, "channel_id should be non-zero");
assert_eq!(reply["channel_id"], serde_json::json!(channel_id));
assert_eq!(
reply["bound"],
serde_json::json!({ "host": "203.0.113.9", "port": 42113 }),
"the establisher's reply field survives the wire to the consumer"
);
// The streams work as always — the projection is reply-plane only.
send.write_all(b"ping").await.expect("write ping");
drop(send);
// `open_channel` (fields discarded) still opens cleanly against
// the same accept side.
let (channel_id2, _send2, _recv2) = client
.open_channel("channels/tty/sub", serde_json::json!({}), "alk/tty")
.await
.expect("open_channel unchanged");
assert!(channel_id2 > 0);
}
// --- review 004 Unit 3 acceptance gates (F-04 serving half) -----------
/// F-04 gate 1: hub→consumer call over an existing `ChannelClient`
src/channels/operations.rs
+349 -9
View File
@@ -17,7 +17,7 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use futures::future::BoxFuture;
use serde_json::{json, Value};
use serde_json::{json, Map, Value};
use crate::core::auth::{AuthContext, Identity};
use crate::core::types::{Capabilities, Connection};
@@ -374,14 +374,18 @@ pub type OpenHandler = Arc<
pub type ChannelPlan = Arc<dyn std::any::Any + Send + Sync>;
/// The establishment-phase result (ADR-049 §1, as filled by ADR-049
/// amendment 2 — R-01). Carries the channel plan the wrapper threads
/// to the pump handler's `plan` parameter ([`OpenHandler`]). The
/// plan is optional: an establisher that only validates (TTY's
/// lookup/ownership shape) returns [`Establishment::default`].
/// amendment 2 — R-01; extended by amendment 3 — review 008 U-2).
/// Carries the channel plan the wrapper threads to the pump handler's
/// `plan` parameter ([`OpenHandler`]) and optional reply fields the
/// wrapper merges into the open-op success reply (the establisher →
/// opener direction). The plan is optional: an establisher that only
/// validates (TTY's lookup/ownership shape) returns
/// [`Establishment::default`].
///
/// `#[non_exhaustive]` so a future carrier change is not another
/// breaking release. Construct with [`Establishment::new`] (plan) or
/// [`Establishment::default`] (no plan).
/// [`Establishment::default`] (no plan); chain
/// [`Establishment::with_reply_field`] to contribute reply fields.
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct Establishment {
@@ -389,12 +393,49 @@ pub struct Establishment {
/// establisher and the handler agree on the concrete type
/// ([`ChannelPlan`]); alkcall never inspects it.
pub plan: Option<ChannelPlan>,
pub(crate) reply_fields: Option<Map<String, Value>>,
}
/// The wrapper-owned success-reply key (ADR-049 amendment 3): the
/// wrapper reserves `channel_id` — an establisher contributing a reply
/// field with this key is an establisher bug and fails the open loudly
/// (`handler_error`) rather than shadowing the wrapper's value.
pub(crate) const RESERVED_REPLY_KEY: &str = "channel_id";
impl Establishment {
/// A result carrying `plan` to the pump handler.
pub fn new(plan: ChannelPlan) -> Self {
Self { plan: Some(plan) }
Self {
plan: Some(plan),
reply_fields: None,
}
}
/// Contribute one reply field (ADR-049 amendment 3 — the
/// establisher → opener direction). Merged into the open-op
/// success reply by the wrapper, after its `channel_id`
/// reservation. Establisher-supplied `channel_id` is rejected —
/// the check runs at merge time in the wrapper, not here, so the
/// failure carries the wire error path. Builder-style.
pub fn with_reply_field(mut self, key: impl Into<String>, value: Value) -> Self {
self.reply_fields
.get_or_insert_with(Map::new)
.insert(key.into(), value);
self
}
/// Contribute reply fields from a map (the batch form of
/// [`Establishment::with_reply_field`]). Builder-style.
pub fn with_reply_fields(mut self, fields: Map<String, Value>) -> Self {
self.reply_fields
.get_or_insert_with(Map::new)
.extend(fields);
self
}
/// The contributed reply fields, for inspection.
pub fn reply_fields(&self) -> Option<&Map<String, Value>> {
self.reply_fields.as_ref()
}
}
@@ -742,6 +783,36 @@ fn establishment_timeout_call_error() -> CallError {
}))
}
/// The `channel:open_failed` reply for an establisher that supplied
/// the wrapper-owned `channel_id` reply key (ADR-049 amendment 3) —
/// reason `handler_error`, the establisher-internal-failure class.
fn reserved_reply_key_call_error() -> CallError {
CallError::new(
CHANNEL_OPEN_FAILED,
"establisher supplied the reserved reply key `channel_id`",
false,
)
.with_details(json!({
"reason": "handler_error",
"message": "establisher supplied the reserved reply key `channel_id`",
}))
}
/// Merge the establisher's reply fields into the success output
/// (ADR-049 amendment 3): called after the `channel_id` reservation
/// has been checked, so the merge is a flat extend — absent fields
/// leave the reply byte-identical to the pre-amendment shape.
fn merge_reply_fields(mut output: Value, reply_fields: Option<Map<String, Value>>) -> Value {
if let Some(fields) = reply_fields {
if let Some(obj) = output.as_object_mut() {
for (key, value) in fields {
obj.insert(key, value);
}
}
}
output
}
/// Run the open-op wrapper's shared post-ACL steps: `check_open` →
/// `open_channel` → **await the establisher bounded** (ADR-049 §1/§2 —
/// no-op when no establisher is registered) → build the channel
@@ -759,6 +830,12 @@ fn establishment_timeout_call_error() -> CallError {
/// opener-side" property: the consumer's open resolves `Err` and no
/// `channel_id` was ever returned.
///
/// On establishment success the wrapper merges the establisher's
/// contributed reply fields into the success output (ADR-049 amendment
/// 3) after reserving `channel_id` — an establisher supplying that key
/// fails the open with reason `handler_error`. Absent reply fields
/// leave the reply byte-identical to the pre-amendment shape.
///
/// `input` is the open op's input (params), passed through to the
/// establisher (when registered) and the ALPN's `OpenHandler`.
/// `call_identity` is the dispatch-resolved caller identity — `Some`
@@ -802,6 +879,7 @@ async fn run_open_wrapper(
return ResponseEnvelope::error(request_id, map_channel_error_to_call_error(&channel_err));
}
let mut reply_fields: Option<Map<String, Value>> = None;
let channel_id = match manager
.open_channel(alpn, opener_identity.id.clone(), None)
.await
@@ -839,7 +917,31 @@ async fn run_open_wrapper(
establishment_error_to_call_error(&e),
);
}
Ok(Ok(establishment)) => establishment.plan,
Ok(Ok(establishment)) => {
// The `channel_id` reservation (ADR-049
// amendment 3): the wrapper owns the key.
// An establisher supplying it is a bug —
// fail loudly (same open_failed path) so a
// reserved-key collision never silently
// shadows the wrapper's value.
if establishment
.reply_fields
.as_ref()
.is_some_and(|m| m.contains_key(RESERVED_REPLY_KEY))
{
tracing::debug!(
channel_id = id,
"open wrapper: establisher supplied the reserved reply key; tearing down channel"
);
teardown_failed_channel(manager, policy, id, &opener_identity);
return ResponseEnvelope::error(
request_id,
reserved_reply_key_call_error(),
);
}
reply_fields = establishment.reply_fields;
establishment.plan
}
}
}
None => None,
@@ -942,7 +1044,10 @@ async fn run_open_wrapper(
}
};
ResponseEnvelope::ok(request_id, json!({ "channel_id": channel_id }))
ResponseEnvelope::ok(
request_id,
merge_reply_fields(json!({ "channel_id": channel_id }), reply_fields),
)
}
/// Map a `ChannelError` (from `ChannelLifecyclePolicy::check_open`) to
@@ -2378,4 +2483,239 @@ mod tests {
let h = plan.downcast_ref::<TestHandle>().expect("typed");
assert_eq!(h.0, 7);
}
// --- ADR-049 amendment 3 (review 008 U-2): the reply projection -------
#[test]
fn establishment_builder_accumulates_reply_fields() {
let e = Establishment::default()
.with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 }))
.with_reply_field("extra", json!(true));
let fields = e.reply_fields().expect("fields");
assert_eq!(fields.len(), 2);
assert_eq!(fields["bound"]["port"], json!(42113));
assert_eq!(fields["extra"], json!(true));
let batch: Map<String, Value> = [("a".to_string(), json!(1)), ("b".to_string(), json!(2))]
.into_iter()
.collect();
let e = Establishment::new(Arc::new(TestHandle(1)) as ChannelPlan).with_reply_fields(batch);
assert_eq!(e.reply_fields().expect("fields").len(), 2);
assert!(e.plan.is_some(), "the plan rides alongside the fields");
}
#[tokio::test]
async fn run_open_wrapper_projects_establisher_reply_field_onto_success_output() {
let manager = make_manager().await;
let policy = super::super::policy::default_policy();
let establishing: OpenEstablisher = Arc::new(|_input, _auth| {
Box::pin(async {
Ok(Establishment::default()
.with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 })))
})
});
let open_handler: OpenHandler =
Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
let auth = AuthContext::anonymous(b"alk/call");
let env = run_open_wrapper(
&manager,
&policy,
hook(establishing, None).as_ref(),
&open_handler,
&auth,
"alk/tty",
json!({}),
Some(identity("alice")),
"req-reply-projection".to_string(),
None,
)
.await;
match env.result {
Ok(v) => {
let channel_id = v["channel_id"].as_u64().expect("channel_id present");
assert!(channel_id > 0);
assert_eq!(
v["bound"],
json!({ "host": "203.0.113.9", "port": 42113 }),
"the establisher's reply field rides the success output"
);
assert_eq!(
v.as_object().expect("object").len(),
2,
"channel_id + bound, nothing else"
);
}
Err(e) => panic!("projecting establisher should succeed, got: {e:?}"),
}
}
#[tokio::test]
async fn run_open_wrapper_without_reply_fields_is_byte_identical_to_pre_amendment() {
let manager = make_manager().await;
let policy = super::super::policy::default_policy();
let open_handler: OpenHandler =
Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
let auth = AuthContext::anonymous(b"alk/call");
// With an establisher contributing nothing:
let env = run_open_wrapper(
&manager,
&policy,
hook(ok_establisher(), None).as_ref(),
&open_handler,
&auth,
"alk/tty",
json!({}),
Some(identity("alice")),
"req-no-fields-est".to_string(),
None,
)
.await;
match env.result {
Ok(v) => assert_eq!(
v,
json!({ "channel_id": v["channel_id"] }),
"establisher, no fields: the reply is exactly channel_id"
),
Err(e) => panic!("open should succeed, got: {e:?}"),
}
// And with no establisher at all:
let env = run_open_wrapper(
&manager,
&policy,
None,
&open_handler,
&auth,
"alk/tty",
json!({}),
Some(identity("alice")),
"req-no-fields-none".to_string(),
None,
)
.await;
match env.result {
Ok(v) => assert_eq!(
v,
json!({ "channel_id": v["channel_id"] }),
"no establisher: the reply is exactly channel_id"
),
Err(e) => panic!("open should succeed, got: {e:?}"),
}
}
#[tokio::test]
async fn run_open_wrapper_establisher_channel_id_reply_key_fails_loudly() {
let manager = make_manager().await;
let concrete_policy = Arc::new(super::super::policy::PerIdentityChannelPolicy::new(8));
let policy: Arc<dyn ChannelLifecyclePolicy> =
Arc::clone(&concrete_policy) as Arc<dyn ChannelLifecyclePolicy>;
let spawned = Arc::new(AtomicBool::new(false));
let spawned_clone = Arc::clone(&spawned);
let open_handler: OpenHandler = Arc::new(move |_input, _plan, _conn, _auth| {
spawned_clone.store(true, Ordering::SeqCst);
tokio::spawn(async {})
});
let establishing: OpenEstablisher = Arc::new(|_input, _auth| {
Box::pin(async {
Ok(Establishment::default().with_reply_field("channel_id", json!(999_u64)))
})
});
let auth = AuthContext::anonymous(b"alk/call");
let env = run_open_wrapper(
&manager,
&policy,
hook(establishing, None).as_ref(),
&open_handler,
&auth,
"alk/tty",
json!({}),
Some(identity("alice")),
"req-reserved-key".to_string(),
None,
)
.await;
match env.result {
Err(e) => {
assert_eq!(e.code, "channel:open_failed");
let details = e.details.expect("details carry the reason");
assert_eq!(details["reason"], "handler_error");
assert!(
details["message"]
.as_str()
.expect("message")
.contains("channel_id"),
"the message names the reserved key"
);
}
Ok(v) => panic!("reserved-key establisher must fail the open, got: {v:?}"),
}
assert!(
manager.channel_ids().is_empty(),
"the just-allocated channel was torn down"
);
assert_eq!(
concrete_policy.count_for(&identity("alice")),
0,
"ledger take + policy.on_close restored the cap count"
);
assert!(
!spawned.load(Ordering::SeqCst),
"pump handler must not spawn on the reserved-key rejection"
);
}
#[tokio::test]
async fn register_openable_with_establisher_replies_projected_fields_end_to_end() {
let manager = make_manager().await;
let policy = super::super::policy::default_policy();
let core = ChannelCore::new(manager.clone(), policy);
let spec = OperationSpec::new(
"channels/tty/sub",
OperationType::Mutation,
Visibility::External,
json!({}),
json!({
"type": "object",
"properties": {
"channel_id": { "type": "integer" },
"bound": { "type": "object" }
}
}),
vec![],
AccessControl::default(),
None,
)
.with_channel_open(ChannelOpenSpec::new("alk/tty"));
let establishing: OpenEstablisher = Arc::new(|_input, _auth| {
Box::pin(async {
Ok(Establishment::default()
.with_reply_field("bound", json!({ "host": "203.0.113.9", "port": 42113 })))
})
});
let open_handler: OpenHandler =
Arc::new(|_input, _plan, _conn, _auth| tokio::spawn(async {}));
let registry = OperationRegistry::new();
core.register_openable_with_establisher(
spec,
Some(establishing),
open_handler,
&registry,
AuthContext::anonymous(b"alk/call"),
None,
)
.expect("register");
let env = registry
.invoke(
"channels/tty/sub",
json!({}),
test_context("reg-projection"),
)
.await;
match env.result {
Ok(v) => {
assert!(v["channel_id"].as_u64().is_some());
assert_eq!(v["bound"]["port"], json!(42113));
}
Err(e) => panic!("projected open should succeed, got: {e:?}"),
}
}
}