//! Producer half of the channels integration — the `register_openable`
//! helper and the `TtyOpenHandler` factory (ADR-047 §3, amended by
//! alkcall ADR-049 / alktty ADR-010 with an establishment phase).
//!
//! This is the path where TTY rides inside `alk/channels` rather than
//! directly on a `alk/tty` ALPN connection. The assembly layer (an
//! `alk/channels` acceptor's `install_channel_zero` hook) builds a
//! per-connection [`ChannelCore`], then calls
//! [`register_openable`] to register the `channels/tty/sub` open op
//! on the per-connection overlay registry. When a consumer (a
//! [`crate::session::TtySession`] or any `ChannelClient` caller)
//! invokes `channels/tty/sub`, the alkcall channels wrapper does
//! `check_open` → `open_channel` → **await the establisher bounded**
//! (alkcall ADR-049 §1) → spawn the `TtyOpenHandler` → respond with
//! `{ channel_id }`. The establisher performs the semantic validation
//! (full `NegotiateRequest` parse, `carriage == "raw"`, backend
//! lookup, ADR-050 ownership check) and fails the open op with
//! `channel:open_failed` on rejection — no phantom channel. The
//! `TtyOpenHandler` receives the channel's `Connection` (data-plane
//! ALPN `alk/tty`), calls `accept_bi()` to get the channel's
//! `BiStream`, and runs the session pumps on it — the same three-pump
//! session driver as the direct-ALPN path, minus the wire-frame
//! negotiation phase (the open op's `input` is the negotiation,
//! ADR-009). Backend allocation runs in the establisher (alkcall 0.6
//! R-01: the allocated `TtyHandle` crosses via the `Establishment`
//! plan payload); no negotiation error frame is ever written on the
//! channels path by a registered producer.
//!
//! The access control (scope-gate + ownership) is wired into the
//! `OperationSpec`'s `AccessControl` and enforced by the registry's
//! `invoke`/`invoke_streaming` before the establisher runs — the
//! establisher itself only runs the ADR-050 ownership check (the
//! scope gate is the registry's); this is the difference from the
//! direct-ALPN `TtyAdapter`, which does its own ad-hoc scope check;
//! the channels path gets ACL for free from alkcall's registry.
//!
//! [`ChannelCore`]: alkcall::channels::operations::ChannelCore

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use alkcall::channels::operations::{
    ChannelCore, ChannelPlan, Establishment, EstablishmentError, OpenEstablisher, OpenHandler,
};
use alkcall::core::auth::AuthContext;
use alkcall::core::ownership::OwnershipProvider;
use alkcall::core::Connection;
use alkcall::registry::spec::{
    AccessControl, ChannelOpenSpec, ErrorDefinition, OperationSpec, OperationType, Visibility,
};
use serde_json::{json, Value};
use tracing::debug;

use crate::adapter::{drive_session_pre_negotiated, TTY_OPEN_SCOPE};
use crate::backend::TtyBackend;
use crate::negotiation::{NegotiateRequest, NegotiationWriter};

/// The per-ALPN open operation name (`channels/<alpn>/sub` convention,
/// ADR-047). TTY is consumer-opens (the client requests a shell), so
/// the op is `sub` (subscribe), not `pub` (publish).
pub const OP_TTY_OPEN: &str = "channels/tty/sub";

/// The data-plane ALPN the channel carries. Matches the ALPN returned
/// by [`crate::adapter::TtyAdapter`]'s `alpn()`.
pub const TTY_ALPN: &str = "alk/tty";

/// Register the `channels/tty/sub` open op on a per-connection
/// `OperationRegistry` (ADR-047 §3, as amended 2026-08-13 —
/// per-connection registration).
///
/// The assembly layer (an `alk/channels` acceptor's
/// `install_channel_zero` hook) builds a [`ChannelCore`] from the
/// adapter-supplied `ChannelManager` + `ChannelLifecyclePolicy`,
/// constructs a fresh per-connection `OperationRegistry`, and calls
/// this helper. The helper builds the [`OperationSpec`] for
/// `channels/tty/sub` with the `channel_open` marker set
/// (`ChannelOpenSpec::new("alk/tty")`), an `AccessControl` carrying
/// the `tty:open` scope gate, and the partial `NegotiateRequest`
/// input schema (`carriage`/`backend`/`cmd` required — alkcall 0.4
/// enforces it at dispatch; the full parse and semantic validation
/// runs in the handler). It then wraps the
/// `TtyOpenHandler` factory and calls
/// [`ChannelCore::register_openable`].
///
/// `backends` is the same backend map the direct-ALPN `TtyAdapter`
/// holds — `HashMap<String, Arc<dyn TtyBackend>>` keyed by the
/// negotiation frame's `backend` string. The assembly layer typically
/// shares one backend map between the direct-ALPN adapter and the
/// channels `register_openable` registration.
///
/// `ownership` is the optional `OwnershipProvider` for the ADR-050
/// resource-ownership check. `None` = scope-gate only. The provider
/// is consulted inside `drive_session_pre_negotiated` (after the
/// backend is selected and `resource_id` is extracted from
/// `backend_params`), not by the channels wrapper — the wrapper's
/// `AccessControl` carries only the scope gate. This mirrors the
/// direct-ALPN `TtyAdapter` shape: the scope gate is enforced once at
/// the entry point (here by the registry's `AccessControl`), the
/// ownership check is enforced once at the backend-selection point.
///
/// `auth` is the peer's `AuthContext`, the install-time context passed
/// through to `register_openable_with_establisher`. The establisher
/// and pump handler read their identity from the per-call context
/// alkcall derives (alkcall 0.7, CF-006 corollary) — the
/// dispatch-resolved opener, the same view the registry's ACL gate
/// checked.
///
/// # Errors
///
/// Returns an error if `ChannelCore::register_openable` rejects the
/// spec (e.g., the spec has no `channel_open` marker — won't happen
/// with the spec this helper builds, but the underlying call is
/// fallible).
pub fn register_openable(
    core: &ChannelCore,
    backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
    ownership: Option<Arc<dyn OwnershipProvider>>,
    registry: &mut alkcall::registry::registration::OperationRegistry,
    auth: AuthContext,
) -> Result<(), String> {
    let spec = tty_open_spec();
    let establisher = make_tty_establisher(Arc::clone(&backends), ownership);
    let open_handler = make_tty_open_handler(backends);
    core.register_openable_with_establisher(
        spec,
        Some(establisher),
        open_handler,
        registry,
        auth,
        None,
    )
}

/// Build the [`OperationSpec`] for `channels/tty/sub`.
///
/// The op is `Sub`-typed (consumer-opens — the initiator opens a
/// channel and the data plane flows on the channel's `BiStream`, not
/// in the call response stream). The `channel_open` marker is set to
/// `ChannelOpenSpec::new("alk/tty")` so the channels wrapper allocates
/// a data channel. The `AccessControl` carries the `tty:open` scope
/// gate (the same scope the direct-ALPN `TtyAdapter` checks).
///
/// The input schema is the shared part of the `NegotiateRequest` shape
/// (`carriage`/`backend`/`cmd` required — alkcall 0.4 enforces it at
/// dispatch). The schema is deliberately partial: `tty`, `cwd`, `env`,
/// and backend-specific selector fields are free-form (`additionalProperties`
/// defaults to `true` in raw JSON Schema), because the
/// `NegotiateRequest` wire shape is owned by `negotiation.rs` and the
/// backend params are opaque (ADR-053) — duplicating their full schemas
/// here would create a second definition to drift. The registry check
/// catches structurally-broken opens (missing command, wrong types);
/// the full parse and semantic validation (`carriage == "raw"`,
/// backend lookup, ownership) and backend allocation run in the
/// establisher (ADR-010, as amended for alkcall 0.6 — failures are
/// `channel:open_failed`, including `allocate_failed` as
/// `dial_failed`).
///
/// The op declares an [`ErrorDefinition`] for `channel:open_failed`
/// (ADR-016 — establishment failures are disclosed so `services/schema`
/// shows the failure contract; reason ∈ `unknown_resource` /
/// `handler_error` / `timeout`, per the ADR-049 vocabulary — see
/// `make_tty_establisher` for the mapping) and carries a
/// `description` (review 006 E-02 — discovery enrichment).
pub fn tty_open_spec() -> OperationSpec {
    let open_failed_schema = json!({
        "type": "object",
        "properties": {
            "reason": {
                "type": "string",
                "enum": ["dial_failed", "unknown_resource", "handler_error", "timeout"]
            },
            "message": { "type": "string" }
        },
        "required": ["reason", "message"]
    });
    OperationSpec::new(
        OP_TTY_OPEN,
        OperationType::Sub,
        Visibility::External,
        json!({
            "type": "object",
            "properties": {
                "carriage": { "type": "string" },
                "backend": { "type": "string" },
                "cmd": {
                    "type": "array",
                    "items": { "type": "string" }
                }
            },
            "required": ["carriage", "backend", "cmd"]
        }),
        json!({
            "type": "object",
            "properties": {
                "channel_id": { "type": "integer", "minimum": 0 }
            }
        }),
        vec![ErrorDefinition {
            code: "channel:open_failed".to_string(),
            description: "Establishment failed after channel allocation: the establisher \
                          rejected the negotiation (malformed negotiation, unknown backend, \
                          ownership denial), the backend allocation failed (dial_failed), or \
                          the establishment deadline expired. Details: { reason, message }. \
                          No channel_id is returned."
                .to_string(),
            schema: open_failed_schema,
            http_status: None,
        }],
        AccessControl {
            required_scopes: vec![TTY_OPEN_SCOPE.to_string()],
            required_scopes_any: None,
            resource_type: None,
            resource_action: None,
        },
        None,
    )
    .with_description(
        "Open a terminal session (alk/tty). The open op's input is the \
         NegotiateRequest (ADR-009); the data plane is TTY's 5-byte chunk \
         format on the allocated channel.",
    )
    .with_channel_open(ChannelOpenSpec::new(TTY_ALPN))
}

/// Build the [`OpenEstablisher`] for `channels/tty/sub` — the awaited
/// establishment phase (alkcall ADR-049 §1, alktty ADR-010).
///
/// The establisher runs the semantic validation that used to be
/// post-open error frames inside the pump handler (ADR-009 R4): the
/// full `NegotiateRequest` parse of the registry-schema-validated
/// `input` (the schema is deliberately partial so the opaque ADR-053
/// backend params pass through), `carriage == "raw"`, non-empty
/// `cmd`, the backend lookup, the ADR-050 ownership check, and —
/// since alkcall 0.6 (ADR-049 amendment 2, review 007 R-01) gave
/// `Establishment` its plan payload — the backend allocation
/// (ADR-010 as amended: the allocated `TtyHandle` crosses to the pump
/// handler via the plan). Any rejection resolves
/// `Err(EstablishmentError)` and the wrapper replies
/// `channel:open_failed` with `details.reason` — the SSH contract:
/// the channel never exists opener-side, no `channel_id` is ever
/// returned.
///
/// The `tty:open` scope gate is NOT checked here — the registry's
/// `AccessControl` enforced it before the establisher ran. The
/// establisher's `auth` is the per-call context (alkcall 0.7, CF-006
/// corollary): its identity is the dispatch-resolved opener — the same
/// identity view the registry's ACL gate checked — and is the
/// ownership-check subject (the same split as the old handler-side
/// check).
///
/// Mapping into the ADR-049 reason vocabulary (no new codes — the
/// vocabulary is alkcall's one-way wire surface):
///
/// - `NegotiateRequest` parse failure, `carriage != "raw"`, empty
///   `cmd` → [`EstablishmentError::HandlerError`] (`malformed_
///   negotiation` semantics — the request shape is wrong).
/// - Unknown backend → [`EstablishmentError::UnknownResource`].
/// - Ownership denial → [`EstablishmentError::HandlerError`] (an ACL
///   outcome, not a dial outcome; the registry's scope gate stays the
///   pre-allocation `FORBIDDEN` path).
/// - `backend.allocate` failure (ADR-005's `AllocFailed`: PTY
///   couldn't be allocated, docker exec failed to start, SSH channel
///   request rejected) → [`EstablishmentError::DialFailed`] — the
///   target refused or the backend lacked capacity; the review-007
///   verification gate's "allocate in the establisher →
///   `allocate_failed` is a call error, no in-band frame" shape.
///
/// The allocation is handed to the pump handler through the
/// establishment plan: a private `AllocatedHandle` slot (the
/// `TtyHandle` itself is not `Sync` — its fields are boxed dyn
/// streams — so it cannot be the `ChannelPlan` directly). The slot is
/// created per establisher invocation (per open), so the
/// shared-slot same-resource race alkcall's R-01 test pins is
/// unreachable by construction; the wrapper moves the plan to this
/// open's handler and the slot is `take`n exactly once. The
/// establisher-before-handler ordering is alkcall's wrapper
/// guarantee (ADR-049 §1).
fn make_tty_establisher(
    backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
    ownership: Option<Arc<dyn OwnershipProvider>>,
) -> OpenEstablisher {
    Arc::new(move |input: Value, auth: AuthContext| {
        let backends = Arc::clone(&backends);
        let ownership = ownership.clone();
        let identity = auth.identity.clone();
        Box::pin(async move {
            let req: NegotiateRequest = match serde_json::from_value(input) {
                Ok(r) => r,
                Err(e) => {
                    debug!("tty: channels establisher: invalid negotiate params: {e}");
                    return Err(EstablishmentError::HandlerError {
                        message: format!("malformed negotiation: {e}"),
                    });
                }
            };
            if req.carriage != "raw" {
                return Err(EstablishmentError::HandlerError {
                    message: "malformed negotiation: carriage must be 'raw'".to_string(),
                });
            }
            if req.cmd.is_empty() {
                return Err(EstablishmentError::HandlerError {
                    message: "malformed negotiation: cmd must be non-empty".to_string(),
                });
            }
            let backend = match backends.get(&req.backend) {
                Some(b) => Arc::clone(b),
                None => {
                    return Err(EstablishmentError::UnknownResource {
                        message: format!("unknown backend: {}", req.backend),
                    })
                }
            };
            let params = crate::backend::TtyParams::from(req);
            if let Some(provider) = ownership {
                if let Some((kind, id)) = backend.resource_id(&params) {
                    let owns = identity
                        .as_ref()
                        .map(|id_ref| provider.owns(id_ref, kind, &id, "tty"))
                        .unwrap_or(false);
                    if !owns {
                        debug!("tty: channels establisher: ownership denied");
                        return Err(EstablishmentError::HandlerError {
                            message: "forbidden: caller does not own the requested tty \
                                          resource"
                                .to_string(),
                        });
                    }
                }
            }
            let handle = match backend.allocate(&params).await {
                Ok(h) => h,
                Err(e) => {
                    debug!("tty: channels establisher: allocation failed: {e}");
                    return Err(EstablishmentError::DialFailed {
                        message: e.to_string(),
                    });
                }
            };
            Ok(Establishment::new(Arc::new(AllocatedHandle {
                handle: Mutex::new(Some(handle)),
            }) as ChannelPlan))
        })
    })
}

/// The establishment plan payload for the channels path: the
/// `TtyHandle` the establisher allocated, behind a one-shot slot (the
/// handle's boxed dyn fields make the struct non-`Sync`, so it cannot
/// be the `ChannelPlan` directly). Created per open; `take`n by the
/// pump handler.
struct AllocatedHandle {
    handle: Mutex<Option<crate::backend::TtyHandle>>,
}

impl AllocatedHandle {
    fn take(&self) -> Option<crate::backend::TtyHandle> {
        self.handle.lock().unwrap_or_else(|e| e.into_inner()).take()
    }
}

/// Build the [`OpenHandler`] for `channels/tty/sub`.
///
/// The handler receives the open op's `input` — the `NegotiateRequest`
/// params, validated (and allocated against) by the establisher before
/// the handler was spawned (alkcall ADR-049 §1: the establisher is
/// awaited bounded, before the reply and before the pump handler) —
/// plus the establisher's `plan`: the [`AllocatedHandle`] slot
/// carrying the backend-allocated `TtyHandle` (alkcall 0.6 ADR-049
/// amendment 2, review 007 R-01). It re-parses `input`, calls
/// `accept_bi()` to get the channel's [`BiStream`], splits it into
/// read/write halves (the stdlib `tokio::io::split` idiom — the same
/// split the direct-ALPN `TtyAdapter::handle` does), and runs the
/// pre-allocated session driver on them — the same three-pump session
/// driver as the direct-ALPN path.
///
/// With a plan the handler never allocates: the establisher already
/// ran `backend.allocate` (ADR-010 as amended), so no negotiation
/// error frame is reachable on the registered path. With `plan: None`
/// (a no-establisher registration — defense-in-depth, the ADR-010
/// pre-0.3.0 shape) the handler re-runs the full inline
/// validate-and-allocate, whose negotiation error frames
/// (`malformed_negotiation`, `allocate_failed`, ...) remain the only
/// failure surface for that registration shape.
///
/// The returned task awaits the session driver inline (R-02 lifetime
/// contract): the `JoinHandle`'s completion is the wrapper's teardown
/// trigger, so the pumps must finish before the task returns — never
/// spawn-and-forget.
///
/// The `tty:open` scope gate is enforced by the registry's `AccessControl`
/// before this handler runs — the handler does not re-check it.
fn make_tty_open_handler(backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>) -> OpenHandler {
    Arc::new(
        move |input: Value,
              plan: Option<ChannelPlan>,
              channel_conn: Connection,
              auth: AuthContext| {
            let backends = Arc::clone(&backends);
            let identity = auth.identity.clone();

            tokio::spawn(async move {
                let req: NegotiateRequest = match serde_json::from_value(input) {
                    Ok(r) => r,
                    Err(e) => {
                        debug!("tty: channels open: invalid negotiate params: {e}");
                        let stream = match channel_conn.accept_bi().await {
                            Ok(s) => s,
                            Err(e) => {
                                debug!("tty: channels open: accept_bi failed: {e}");
                                return;
                            }
                        };
                        let (_, mut client_write) = tokio::io::split(stream);
                        crate::adapter::send_negotiation_error(
                            NegotiationWriter::new(&mut client_write),
                            "malformed_negotiation",
                            &[("message", &e.to_string())],
                        )
                        .await;
                        return;
                    }
                };
                let stream = match channel_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(e) => {
                        debug!("tty: channels open: accept_bi failed: {e}");
                        return;
                    }
                };
                let (client_read, client_write) = tokio::io::split(stream);
                let allocated = plan
                    .as_ref()
                    .and_then(|p| p.downcast_ref::<AllocatedHandle>())
                    .and_then(|slot| slot.take());
                match allocated {
                    Some(handle) => {
                        crate::adapter::drive_session_pre_allocated(
                            client_write,
                            client_read,
                            handle,
                        )
                        .await;
                    }
                    None => {
                        drive_session_pre_negotiated(
                            client_write,
                            client_read,
                            req,
                            backends,
                            None,
                            identity,
                        )
                        .await;
                    }
                }
            })
        },
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::{MockBackend, TtyError};
    use crate::testing::wire_client_and_server;
    use alkcall::channels::operations::ChannelCore;
    use alkcall::core::auth::Identity;
    use alkcall::registry::registration::OperationRegistry;
    use std::collections::HashMap as StdHashMap;
    use tokio::io::duplex;

    /// The R4 unit-testable seam, retained as a defense-in-depth gate:
    /// feed `make_tty_open_handler` a hand-built `input` the (partial)
    /// schema would accept but the full `NegotiateRequest` parse rejects
    /// (`cwd` typed as a number). Post-ADR-010 the establisher rejects
    /// this class before the handler ever spawns — this test pins the
    /// handler's own parse arm, which must still write a
    /// `0x00`-prefixed `malformed_negotiation` error frame rather than
    /// silently EOFing if that input ever reaches it (e.g. a
    /// no-establisher registration). The read side is the consumer's
    /// exact post-open sequence (`from_halves_raw`): peek the first
    /// byte, see `0x00`, read the length-prefixed error frame.
    #[tokio::test]
    async fn open_handler_writes_error_frame_on_schema_bypassing_input() {
        use alkcall::core::types::{BiStream, BidiStreamSource, StreamError};
        use tokio::io::AsyncReadExt;

        struct YieldOnce {
            stream: tokio::sync::Mutex<Option<BiStream>>,
        }
        #[async_trait::async_trait]
        impl BidiStreamSource for YieldOnce {
            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
                self.stream
                    .lock()
                    .await
                    .take()
                    .ok_or(StreamError::ConnectionClosed)
            }
            async fn open_bi(&self) -> Result<BiStream, StreamError> {
                Err(StreamError::StreamClosed)
            }
            fn remote_addr(&self) -> Option<std::net::SocketAddr> {
                None
            }
            fn close(&self, _code: u32, _reason: &str) {
                // The take-on-drop of the wrapped stream is not needed
                // for this test; the connection is never closed.
            }
        }

        let (client_end, server_end) = duplex(64 * 1024);
        let (server_read, server_write) = tokio::io::split(server_end);
        let bidi = BiStream::from_joined(server_read, server_write);
        let channel_conn = Connection::from_source(
            YieldOnce {
                stream: tokio::sync::Mutex::new(Some(bidi)),
            },
            TTY_ALPN.as_bytes().to_vec(),
        );

        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let handler = make_tty_open_handler(Arc::new(backends));

        let task = handler(
            serde_json::json!({
                "carriage": "raw",
                "backend": "mock",
                "cmd": ["true"],
                "cwd": 42
            }),
            None,
            channel_conn,
            AuthContext::anonymous(b"test"),
        );
        task.await.expect("handler task");

        let mut read = client_end;
        let mut first = [0u8; 1];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            read.read_exact(&mut first),
        )
        .await
        .expect("no byte from handler")
        .expect("read first byte");
        assert_eq!(first[0], 0x00, "error frame length prefix starts with 0x00");

        let mut len_rest = [0u8; 3];
        read.read_exact(&mut len_rest).await.expect("read len rest");
        let len = u32::from_be_bytes([first[0], len_rest[0], len_rest[1], len_rest[2]]) as usize;
        let mut body = vec![0u8; len];
        read.read_exact(&mut body).await.expect("read error body");
        let v: serde_json::Value = serde_json::from_slice(&body).expect("parse error frame");
        assert_eq!(v["error"], "malformed_negotiation");
        assert!(v["message"].as_str().is_some_and(|m| !m.is_empty()));
    }

    #[test]
    fn op_tty_open_is_channels_tty_sub() {
        assert_eq!(OP_TTY_OPEN, "channels/tty/sub");
    }

    // --- ADR-010: the establisher (semantic validation → channel:open_failed)

    fn mock_backends_one() -> Arc<HashMap<String, Arc<dyn TtyBackend>>> {
        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        Arc::new(backends)
    }

    async fn establisher_result(
        backends: Arc<HashMap<String, Arc<dyn TtyBackend>>>,
        ownership: Option<Arc<dyn OwnershipProvider>>,
        input: Value,
    ) -> Result<Establishment, EstablishmentError> {
        let establisher = make_tty_establisher(backends, ownership);
        establisher(input, AuthContext::anonymous(b"test")).await
    }

    #[tokio::test]
    async fn establisher_ok_on_valid_params() {
        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
        )
        .await;
        assert!(result.is_ok(), "valid params establish, got {result:?}");
    }

    #[tokio::test]
    async fn establisher_rejects_schema_bypassing_parse() {
        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"], "cwd": 42 }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "handler_error");
                assert!(e.message().contains("malformed negotiation"));
            }
            Ok(_) => panic!("schema-bypassing input must be rejected by the establisher"),
        }
    }

    #[tokio::test]
    async fn establisher_rejects_non_raw_carriage() {
        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "line", "backend": "mock", "cmd": ["true"] }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "handler_error");
                assert!(e.message().contains("carriage must be 'raw'"));
            }
            Ok(_) => panic!("non-raw carriage must be rejected"),
        }
    }

    #[tokio::test]
    async fn establisher_rejects_empty_cmd() {
        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "raw", "backend": "mock", "cmd": [] }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "handler_error");
                assert!(e.message().contains("cmd must be non-empty"));
            }
            Ok(_) => panic!("empty cmd must be rejected"),
        }
    }

    #[tokio::test]
    async fn establisher_rejects_unknown_backend_as_unknown_resource() {
        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "raw", "backend": "nope", "cmd": ["true"] }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "unknown_resource");
                assert_eq!(e.message(), "unknown backend: nope");
            }
            Ok(_) => panic!("unknown backend must be rejected"),
        }
    }

    /// The ADR-050 ownership check runs in the establisher (moved from
    /// the pump handler's `validate_and_allocate`); a denial is
    /// `handler_error` (an ACL outcome, not a dial outcome).
    #[tokio::test]
    async fn establisher_rejects_ownership_denial() {
        use alkcall::core::OwnershipStore;

        let store = alkcall::core::ownership::InMemoryOwnershipStore::new();
        // The requested resource is owned by someone else (or nobody).
        store
            .record(
                &Identity {
                    id: "mallory".to_string(),
                    scopes: vec![],
                    resources: HashMap::new(),
                },
                "session",
                "s-42",
            )
            .await
            .expect("record");
        let provider: Arc<dyn OwnershipProvider> = Arc::new(store);

        struct OwnedBackend;
        #[async_trait::async_trait]
        impl TtyBackend for OwnedBackend {
            async fn allocate(
                &self,
                _params: &crate::backend::TtyParams,
            ) -> Result<crate::backend::TtyHandle, crate::backend::TtyError> {
                panic!("allocate must not run during establishment");
            }
            fn resource_id(
                &self,
                _params: &crate::backend::TtyParams,
            ) -> Option<(&'static str, String)> {
                Some(("session", "s-42".to_string()))
            }
        }
        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(OwnedBackend));

        let result = establisher_result(
            Arc::new(backends),
            Some(provider),
            json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "handler_error");
                assert!(e.message().contains("forbidden"));
            }
            Ok(_) => panic!("ownership denial must be rejected by the establisher"),
        }
    }

    /// Allocation runs in the establisher (alkcall 0.6 — the
    /// `TtyHandle` crosses to the pump handler via the `Establishment`
    /// plan payload, ADR-010 as amended). Its failure maps to
    /// `DialFailed` (`details.reason == "dial_failed"` — the target
    /// refused or the backend lacked capacity), and a successful
    /// allocation lands in the plan slot the handler takes.
    #[tokio::test]
    async fn establisher_allocates_and_maps_failure_to_dial_failed() {
        struct AllocFailBackend;
        #[async_trait::async_trait]
        impl TtyBackend for AllocFailBackend {
            async fn allocate(
                &self,
                _params: &crate::backend::TtyParams,
            ) -> Result<crate::backend::TtyHandle, crate::backend::TtyError> {
                Err(crate::backend::TtyError::AllocFailed {
                    message: "out of ptys".to_string(),
                })
            }
        }
        let mut failing: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        failing.insert("mock".to_string(), Arc::new(AllocFailBackend));
        let result = establisher_result(
            Arc::new(failing),
            None,
            json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
        )
        .await;
        match result {
            Err(e) => {
                assert_eq!(e.reason(), "dial_failed");
                assert!(e.message().contains("out of ptys"));
            }
            Ok(_) => panic!("allocation failure must be rejected by the establisher"),
        }

        let result = establisher_result(
            mock_backends_one(),
            None,
            json!({ "carriage": "raw", "backend": "mock", "cmd": ["true"] }),
        )
        .await;
        let establishment = result.expect("successful allocate carries the handle");
        let slot = establishment
            .plan
            .as_ref()
            .and_then(|p| p.downcast_ref::<AllocatedHandle>())
            .expect("plan is the AllocatedHandle slot")
            .take()
            .expect("the allocated handle is in the slot");
        // The handle is live: its exit_code resolves (MockBackend
        // resolves immediately with the configured code).
        let code = slot.exit_code.await.expect("exit_code resolves");
        assert_eq!(code, 0);
    }

    /// E-02: the spec carries a description (disclosed by
    /// `services/list`) and declares the `channel:open_failed` error
    /// schema (ADR-016 — the establishment-failure contract is
    /// discoverable via `services/schema`).
    #[test]
    fn tty_open_spec_declares_description_and_open_failed_error() {
        let spec = tty_open_spec();
        assert!(
            spec.description.is_some(),
            "description set for services/list disclosure (review 006 E-02)"
        );
        let def = spec
            .error_schemas
            .iter()
            .find(|e| e.code == "channel:open_failed")
            .expect("channel:open_failed ErrorDefinition declared");
        let enum_values = def.schema["properties"]["reason"]["enum"]
            .as_array()
            .expect("reason enum");
        let reasons: Vec<&str> = enum_values.iter().filter_map(|v| v.as_str()).collect();
        // The establisher produces these four (resource_shortage has
        // no TTY producer path — ADR-010; allocation failure maps to
        // dial_failed, the amended mapping).
        assert_eq!(
            reasons,
            vec![
                "dial_failed",
                "unknown_resource",
                "handler_error",
                "timeout"
            ]
        );
    }

    #[test]
    fn tty_alpn_is_alk_tty() {
        assert_eq!(TTY_ALPN, "alk/tty");
    }

    #[test]
    fn tty_open_spec_has_channel_open_marker_for_alk_tty() {
        let spec = tty_open_spec();
        assert_eq!(spec.name, OP_TTY_OPEN);
        assert_eq!(spec.op_type, OperationType::Sub);
        assert_eq!(spec.visibility, Visibility::External);
        let marker = spec.channel_open.expect("channel_open marker set");
        assert_eq!(marker.alpn, TTY_ALPN);
    }

    #[test]
    fn tty_open_spec_requires_tty_open_scope() {
        let spec = tty_open_spec();
        assert_eq!(spec.access_control.required_scopes, vec![TTY_OPEN_SCOPE]);
        assert!(spec.access_control.has_restrictions());
    }

    #[test]
    fn tty_open_spec_input_schema_has_required_fields() {
        let spec = tty_open_spec();
        let schema = spec.input_schema;
        let required = schema
            .get("required")
            .and_then(|v| v.as_array())
            .expect("required array");
        let required_names: Vec<&str> = required.iter().filter_map(|v| v.as_str()).collect();
        assert!(required_names.contains(&"carriage"));
        assert!(required_names.contains(&"backend"));
        assert!(required_names.contains(&"cmd"));
    }

    /// `register_openable` registers the op on the registry. The op
    /// must be discoverable by name afterwards.
    #[tokio::test]
    async fn register_openable_registers_op_on_registry() {
        use alkcall::channels::policy::default_policy;
        let (_client, server) = duplex(1024);
        let (_reader, writer) = tokio::io::split(server);
        let (handle, _runner) = alkcall::channels::mux::MuxRunner::new(Box::new(writer));
        let manager = alkcall::channels::manager::ChannelManager::with_defaults(handle, None);
        let core = ChannelCore::new(manager, default_policy());

        let backends: Arc<HashMap<String, Arc<dyn TtyBackend>>> = Arc::new(HashMap::new());
        let mut registry = OperationRegistry::new();
        register_openable(
            &core,
            backends,
            None,
            &mut registry,
            AuthContext::anonymous(b"alk/channels"),
        )
        .expect("register_openable");
        assert!(registry.registration(OP_TTY_OPEN).is_some());
    }

    /// End-to-end: `ChannelClient::call_open_op("channels/tty/sub")`
    /// returns `{ channel_id }` when the server-side has registered via
    /// `register_openable` with a `MockBackend`. The open op increments
    /// the channel count; the `channel_id` is non-zero (channel 0 is
    /// the call channel, ADR-036).
    #[tokio::test]
    async fn end_to_end_open_op_returns_channel_id() {
        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let backends = Arc::new(backends);

        let identity = Identity {
            id: "alice".to_string(),
            scopes: vec![TTY_OPEN_SCOPE.to_string()],
            resources: StdHashMap::new(),
        };
        let client = wire_client_and_server(backends, None, Some(identity)).await;

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                OP_TTY_OPEN,
                json!({
                    "carriage": "raw",
                    "backend": "mock",
                    "cmd": ["true"],
                }),
            ),
        )
        .await
        .expect("open op timed out");

        let out = response.result.expect("open op should succeed");
        let channel_id = out
            .get("channel_id")
            .and_then(|v| v.as_u64())
            .expect("channel_id in response");
        assert!(
            channel_id > 0,
            "channel_id should be non-zero (channel 0 is the call channel), got {channel_id}"
        );
    }

    /// The open op denies when the caller lacks `tty:open`. The
    /// `AccessControl::required_scopes` gate runs in the registry's
    /// `invoke_streaming` before the `TtyOpenHandler` spawns.
    #[tokio::test]
    async fn end_to_end_open_op_denies_without_tty_open_scope() {
        let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
        let backends = Arc::new(backends);

        let identity = Identity {
            id: "alice".to_string(),
            scopes: vec![],
            resources: StdHashMap::new(),
        };
        let client = wire_client_and_server(backends, None, Some(identity)).await;

        let response = tokio::time::timeout(
            std::time::Duration::from_secs(10),
            client.call_open_op(
                OP_TTY_OPEN,
                json!({
                    "carriage": "raw",
                    "backend": "mock",
                    "cmd": ["true"],
                }),
            ),
        )
        .await
        .expect("open op timed out");

        let err = response.result.expect_err("open op should be denied");
        assert!(
            err.message.contains("scope")
                || err.code.contains("FORBIDDEN")
                || err.code.contains("AUTH"),
            "error should mention scope/forbidden/auth, got code={} message={}",
            err.code,
            err.message
        );
    }

    /// Sanity that the `MockBackend` we use in the channels tests
    /// compiles as a `TtyBackend` and resolves to the configured exit
    /// code. Belt-and-suspenders — the backend module already tests
    /// this, but the channels tests depend on it.
    #[tokio::test]
    async fn mock_backend_resolves_configured_exit_code() {
        let backend = MockBackend::with_exit_code(7);
        let params = crate::backend::TtyParams {
            terminal: None,
            cmd: vec!["true".to_string()],
            cwd: None,
            env: HashMap::new(),
            backend_params: serde_json::Map::new(),
        };
        let handle = backend.allocate(&params).await.expect("allocate");
        let code = handle.exit_code.await.expect("exit_code resolves");
        assert_eq!(code, 7);
        let _: Result<i32, TtyError> = Ok(code);
    }
}