fix: remediate consumer findings CF-001..004 (alkhttp ledger)

- CF-004: services_schema_handler now applies the same visibility +
  AccessControl gates as invoke() (identity resolution mirrors invoke:
  handler_identity under internal). Restricted ops return spec-404 NOT_FOUND
  — matches "restricted ops don't exist" and leaks nothing about the
  restricted surface. Closes the unauthenticated /call-path disclosure.
- CF-003: publish_schema compiled at registration time (both
  OperationRegistry::register and OperationRegistryBuilder::store);
  un-compilable schemas are a registration error — an unvalidated ingest
  path can no longer be constructed. Compiled validator cached per-op
  (publish_validator) and consumed by dispatch; per-request compile gone.
  BEHAVIOR CHANGE: register/builder reject un-compilable publish_schema.
- CF-002: demux TooLarge skip streams through a fixed 64 KiB buffer
  instead of allocating the peer-declared length (u32, up to ~4 GiB);
  cumulative 256 MiB skipped-bytes budget tears down dribbling peers.
  Existing resync test passes unchanged.
- CF-001: new retryable CallError::connection_closed (CONNECTION_CLOSED)
  applied only where the call is provably undelivered — request-frame
  write failures on all consumer paths (call/subscribe/publish, both
  stream modes; publish pump tags write stages). Mid-publish failures and
  producer-side fail_all stay non-retryable INTERNAL (delivery ambiguous).
  New code string is additive; retryable flag is the machine-readable
  signal.

Verification: cargo test (558 pass, 15 new), clippy --all-targets -D
warnings, fmt --check, wasm32-unknown-unknown check.

CHANGELOG.md

@@ -4,6 +4,49 @@ All notable changes to this crate are documented here. The format is
 based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
 this crate adheres to [Semantic Versioning](https://semver.org/).
 
+## [Unreleased]
+
+Consumer-findings remediation (CF-001..004 from
+`docs/reviews/consumer-findings-ledger.md` — alkhttp as the first real
+consumer). One behavior change noted below; otherwise additive.
+
+### Changed
+
+- **Un-compilable `publish_schema` values are rejected at registration
+  time** (CF-003). `OperationRegistry::register` and every
+  `OperationRegistryBuilder` method now return `Err` when a Pub op's
+  `publish_schema` fails to compile. Previously the bad schema was
+  accepted and the dispatch path failed **open** (warn log, chunks
+  unvalidated). The compiled validator is now cached per-op
+  (`OperationRegistry::publish_validator`) and consumed by the dispatch
+  path — the per-request compile is gone. Code that registered
+  un-compilable schemas will now get a registration error instead of a
+  silently-unvalidated op.
+
+### Fixed
+
+- **`services/schema` no longer discloses Internal or ACL-restricted op
+  specs** (CF-004). The schema handler now applies the same gates as
+  `invoke()` — Internal-visibility rejection and `AccessControl::check`
+  with matching identity resolution. Restricted ops return spec-404
+  (`NOT_FOUND`), consistent with "restricted ops don't exist" elsewhere;
+  no information leaks about the restricted surface. The gate is inside
+  the handler, so every transport is covered.
+- **Demux `TooLarge` skip no longer allocates from the peer's header**
+  (CF-002). The skip now streams through a fixed 64 KiB buffer, and a
+  cumulative 256 MiB skipped-bytes budget tears down connections that
+  loop oversized headers. Previously the buffer was sized from the
+  untrusted `u32` length (up to ~4 GiB pinned per dribbling peer).
+- **Write failures before request delivery are retryable** (CF-001).
+  New `CallError::connection_closed` (`CONNECTION_CLOSED`, `retryable:
+  true`) is returned when the `call.requested` frame write fails on any
+  consumer path (`call`/`subscribe`/`publish`, both stream modes) — the
+  call provably never reached the producer, so reconnect/retry is safe.
+  Mid-publish write failures and producer-side connection-teardown
+  failures remain non-retryable `INTERNAL` (delivery ambiguous). The
+  new code string is additive; the `retryable` flag is the
+  machine-readable signal for consumers.
+
 ## [0.1.1] - 2026-08-17
 
 A minor release that renames the ALPN prefix from `alknet/` to `alk/`

docs/reviews/consumer-findings-ledger.md

@@ -11,141 +11,101 @@ Format: date | found-in (alkhttp context) | severity | status.
 
 ## Open
 
-### CF-004 — `services_schema_handler` discloses Internal/ACL-restricted op specs — no visibility or AccessControl check (2026-08-30)
+(none)
+
+---
+
+## Resolved
+
+### CF-004 — `services_schema_handler` discloses Internal/ACL-restricted op specs — no visibility or AccessControl check (2026-08-30) — RESOLVED 2026-08-31
 
 - **Found in:** alkhttp Review 002, finding PRJ-16
   (`alkhttp/docs/reviews/002-post-remediation-review.md`, Part F',
   verified against tree `91483a7` + alkcall source). Filed to this
   ledger 2026-08-30.
   `src/registry/discovery.rs:327-343` (`services_schema_handler`) —
-  the handler does a bare `registry.registration(&name)` and returns
+  the handler did a bare `registry.registration(&name)` and returned
   `spec_to_json(&reg.spec)` verbatim, with **no `Visibility` check and
-  no `AccessControl::check(peer_identity)`**. The built-in op itself is
-  External with default ACL, so alkhttp's per-route pre-checks (which
-  guard only the *outer* op name `services/schema`) cannot see the
-  inner name a caller smuggles in the input: `POST /call
-  {"operation":"services/schema","input":{"name":"secret/op"}}` returns
-  the full spec — input/output/error schemas and
-  `access_control.required_scopes` — of any Internal or admin-scoped
-  op, unauthenticated. The GET `/schema` route and the MCP `schema`
-  tool both enforce internal-op + ACL pre-checks on the same
-  underlying operation; the `/call` path (and the MCP `call` tool
-  dispatching through it) is the hole.
-- **Impact:** unauthenticated disclosure of the complete contract of
-  every Internal operation, including the exact scopes needed for
-  escalation (the "reads the scopes needed to escalate" shape review
-  001's PRJ-06 called out). Disclosure only — the op's handler is not
-  executed.
-- **Suggested direction:** enforce inside the handler (the complete
-  fix, covers every transport): resolve the peer identity from
-  `OperationContext`, then apply the same two checks the HTTP route
-  does — `is_internal_op`-equivalent visibility rejection (spec-404)
-  and `AccessControl::check` → FORBIDDEN envelope. Defensive note: some
-  existing wire tests exercise `services/schema` with default-ACL ops
-  and would still pass; the new behavior is only observable against
-  Internal/ACL-restricted registrations.
-- **alkhttp side:** an alkhttp-local mitigation (input pre-check in
-  `call_handler`'s dispatch path) is planned as review-002 remediation
-  regardless — but it is a per-transport patch; the handler-level fix
-  here is the one that closes the hole for the wire and every future
-  transport.
-- **Status:** open — no alkcall change made.
+  no `AccessControl::check(peer_identity)`**.
+- **Fix:** the handler now applies the same two gates as
+  `OperationRegistry::invoke` — the Internal-visibility rejection
+  (`!ctx.internal` → spec-404) and `AccessControl::check` with the same
+  identity resolution invoke() uses (`handler_identity` when internal,
+  `identity` otherwise). Restricted ops return `NOT_FOUND` (spec-404)
+  rather than `FORBIDDEN` — no information leaks about the restricted
+  surface's existence or shape. The gate is in the handler itself, so
+  every transport (wire `/call`, HTTP routes, MCP tools) is covered.
+- **Status:** resolved — 2026-08-30.
 
-### CF-003 — `publish_schema` compile failure is fail-open on the wire dispatch path (2026-08-30)
+### CF-003 — `publish_schema` compile failure is fail-open on the wire dispatch path (2026-08-30) — RESOLVED 2026-08-30
 
 - **Found in:** alkhttp Review 001 post-remediation task
   `review-001-publish-schema-validation-robust` (the GW-01/#1 follow-up
   work). While fixing alkhttp's HTTP-side instance of the pattern, the
   identical instance was verified on the wire dispatch path:
   `src/protocol/dispatch.rs:352-366` (`Dispatcher::dispatch`,
-  `OperationType::Pub` arm) — when a Pub op's `publish_schema` fails to
-  **compile** (`jsonschema::options().build(&schema)` errors, e.g. a
-  non-array `"required"`, a non-string `"type"`, an invalid `pattern`
-  regex), the pump logs `warn!("publish_schema failed to compile;
-  chunks will not be validated")` and proceeds with `validator: None`.
-- **Impact:** silent unvalidated ingest, per-request. A Pub op (local
-  registration or remote import) whose schema is un-compilable by the
-  pinned jsonschema dialect silently receives arbitrary JSON over the
-  wire that the registration *declares* validated — the handler's
-  schema guarantee is transport-dependent and quietly false. One broken
-  schema in the registry turns that op into an unvalidated ingest path
-  with only a warn log as the tell. (Per-request recompile cost is the
-  secondary issue; alkhttp's gateway cache eliminated both.)
-- **Suggested direction:** fail-closed — on compile failure, reject the
-  chunk (or abort the sink dispatch) with `INTERNAL`-class
-  `CallError` naming the compile failure in the log only (untrusted
-  schema text off the wire), never a silent skip; cache the compiled
-  validator per registration (hot-reload by schema-value comparison;
-  see alkhttp's `src/gateway/schema_cache.rs` for the worked shape).
-  Complementary hygiene: reject un-compilable `publish_schema` values
-  at **registration time** (`OperationRegistry::register` observing
-  `spec.publish_schema`, or `with_publish_schema` validating eagerly)
-  so the failure never reaches traffic.
-- **alkhttp side:** fixed in alkhttp `1572a9d` (gateway cache + lazy
-  resolution + stream-terminal fail-closed). The wire path is out of
-  this crate's scope (AGENTS.md: no forking alkcall types; the pump is
-  alkcall-internal), so the same invariant currently holds only on the
-  HTTP path.
-- **Status:** open — no alkcall change made.
+  `OperationType::Pub` arm) — when a Pub op's `publish_schema` failed to
+  **compile**, the pump logged a warn and proceeded with
+  `validator: None` (silent unvalidated ingest, plus per-request
+  recompile cost).
+- **Fix:** fail-closed at the source — `publish_schema` is now compiled
+  at **registration time** in both insertion points
+  (`OperationRegistry::register` and `OperationRegistryBuilder::store`);
+  an un-compilable schema is a registration error, so an
+  unvalidated-ingest path can never be constructed. The compiled
+  validator is cached per-op (`OperationRegistry::publish_validator`)
+  and the dispatch path consumes the cache — the per-request compile is
+  gone. The `from_call` import path inherits the guarantee (imports
+  register through the same choke point); forwarded-chunk validation on
+  the producer side is the remote peer's dispatch path and now inherits
+  the same fail-closed property.
+- **Behavior change (semver-relevant):** `register`/builder methods
+  return `Err` for an un-compilable `publish_schema` (previously:
+  accepted). Code that registered garbage schemas will now get a
+  registration error instead of a silently-unvalidated op.
+- **Status:** resolved — 2026-08-30.
 
-### CF-002 — demux `TooLarge` skip allocates the full peer-declared length up front (up to ~4 GiB from an 8-byte header) (2026-08-30)
+### CF-002 — demux `TooLarge` skip allocates the full peer-declared length up front (up to ~4 GiB from an 8-byte header) (2026-08-30) — RESOLVED 2026-08-30
 
 - **Found in:** alkhttp Review 001, finding WS-12 (cross-crate;
   `docs/reviews/001-initial-implementation-review.md`, Part B,
   verified against tree `4a825d3`). Filed to this ledger 2026-08-30.
-  `src/channels/adapter.rs:144` (`run_demux_loop_for_client`) — when a
-  parsed chunk header carries `length > MAX_CHUNK_LEN`, the
-  `ChunkError::TooLarge` arm allocates
+  `src/channels/adapter.rs:144` (`run_demux_loop_for_client`) — the
+  `ChunkError::TooLarge` arm allocated
   `let mut discard = vec![0u8; length as usize];` **before** reading:
-  the buffer is sized from the peer's untrusted 8-byte header, and
-  `length` is a `u32` (`0xFFFFFFFF` ≈ 4 GiB).
-- **Impact:** memory-shape DoS, not a correctness bug (the resync/skip
-  itself is correct — alkcall's own resync test covers it). Any
-  authenticated peer can send header `[ch][len = 0xFFFFFFFF]` and then
-  dribble; the allocation is pinned for as long as the payload takes to
-  arrive (a dribbling peer can hold it indefinitely). Via the alkhttp WS
-  path this is trivially reachable by any authenticated browser; K
-  concurrent connections × 4 GiB headers → OOM before the dribble even
-  matters. The *normal* payload arm (`:159`, `vec![0u8;
-  header.length as usize]`) is safe — `parse_header` rejects
-  `length > MAX_CHUNK_LEN`, so it is bounded at 16 MiB; only the
-  `TooLarge` skip arm is unbounded.
-- **Suggested direction:** stream-skip with a bounded fixed buffer
-  (e.g. a 64 KiB stack/fixed buffer, `read_exact` loop consuming
-  `length` bytes) instead of the pre-sized allocation. Optionally bound
-  cumulative skipped bytes per connection before tearing it down, so a
-  peer cannot loop `TooLarge` headers to burn bandwidth/CPU forever.
-- **alkhttp side:** nothing to change there — this is demux-internal.
-  alkhttp's own read-side hardening (review-001-ws-pump-consolidation:
-  explicit inbound `max_message_size`/`max_frame_size`, idle-read
-  timeout) bounds the WS ingress, but a header that survives those caps
-  still hits this arm, so the fix belongs on the alkcall side.
-- **Status:** open — no alkcall change made.
+  the buffer was sized from the peer's untrusted 8-byte header (`u32`
+  `length`, up to ~4 GiB).
+- **Fix:** stream-skip with a fixed 64 KiB buffer + `read_exact` loop
+  consuming `length` bytes — no allocation sized from peer input.
+  Plus a cumulative skipped-bytes budget (256 MiB) that tears down the
+  connection when a peer loops `TooLarge` headers to burn bandwidth/CPU.
+  The budget resets on each valid chunk, so legitimate isolated
+  oversized chunks (the resync path, covered by the existing test) are
+  unaffected. The existing `demux_resyncs_after_oversized_chunk` test
+  passes unchanged.
+- **Status:** resolved — 2026-08-30.
 
-### CF-001 — `call_single_stream` write-failure maps to non-retryable `INTERNAL` (2026-08-29)
+### CF-001 — `call_single_stream` write-failure maps to non-retryable `INTERNAL` (2026-08-29) — RESOLVED 2026-08-30
 
 - **Found in:** alkhttp `from_wss` drop-monitor race tests (WS-02/CON-02,
-  review-001-ws-eof-signal). When the transport mux dies mid-call, the
-  write path fails fast and alkcall's `call_single_stream` write-failure
-  mapping produces a non-retryable
-  `INTERNAL: failed to write request frame` — even though the call never
-  reached the producer and a reconnect/retry would be safe.
-- **Impact:** consumers that drop connections under load (WS EOF, network
-  flaps) get non-retryable errors for in-flight calls that are provably
-  not processed. Callers must either tolerate-or-discriminate two
-  envelope shapes for the same transport-death race, or miss retry
-  opportunities.
-- **Suggested direction:** classify write failures that occur before any
-  response frame could arrive as retryable (`CONNECTION_CLOSED` or
-  equivalent) — or expose the distinction so consumers can decide.
-- **alkhttp side:** the race test tolerates both outcomes for now
-  (asserts retryable only when the call is provably in-flight). If/when
-  alkcall fixes the mapping, alkhttp's ws-eof-signal test should tighten
-  back to retryable-only.
-- **Status:** open — no alkcall change made.
-
----
-
-## Resolved
-
-(empty)
+  review-001-ws-eof-signal). When the transport mux died mid-call, the
+  write path failed fast and the write-failure mapping produced a
+  non-retryable `INTERNAL: failed to write request frame` — even though
+  the call never reached the producer and a reconnect/retry would be
+  safe.
+- **Fix:** new retryable `CallError::connection_closed` constructor
+  (`CONNECTION_CLOSED` code, `retryable: true`), applied **only** to
+  write failures where the call is provably undelivered:
+  `call.requested` frame write failures on all consumer paths
+  (`call`/`subscribe`/`publish`, single-stream and stream-per-request;
+  the publish pump tags its write stages so only the request-frame
+  stage is retryable). Mid-publish and completed-frame write failures
+  stay `INTERNAL` (delivery ambiguous — retry unsafe), and the
+  producer-side `fail_all(...)` on connection close stays `INTERNAL`.
+  The new code string is additive to the wire error vocabulary;
+  `CONNECTION_CLOSED` is new — consumers should treat unknown codes per
+  their existing policy (the `retryable` flag is the machine-readable
+  signal).
+- **Status:** resolved — 2026-08-30. alkhttp follow-up: the
+  `review-001-ws-eof-signal` race test can now tighten back to
+  retryable-only asserts.

src/channels/adapter.rs

@@ -128,6 +128,15 @@ impl ChannelsAdapter {
         reader: Box<dyn tokio::io::AsyncRead + Send + Unpin>,
         policy: Option<&Arc<dyn ChannelLifecyclePolicy>>,
     ) {
+        // Stack buffer for skipping oversized payloads — sized once, never
+        // from peer-declared lengths (a peer-declared `length` is untrusted
+        // up to u32::MAX ≈ 4 GiB; pre-sizing off it is an allocation-shape
+        // DoS). A cumulative cap tears down dribbling peers that loop
+        // TooLarge headers to burn bandwidth indefinitely.
+        const SKIP_BUF_LEN: usize = 64 * 1024;
+        const MAX_CONSECUTIVE_SKIPPED_BYTES: u64 = 256 * 1024 * 1024;
+        let mut skip_buf = vec![0u8; SKIP_BUF_LEN];
+        let mut skipped_since_last_valid: u64 = 0;
         let mut reader = reader;
         let mut header_buf = [0u8; CHUNK_HEADER_LEN];
         loop {
@@ -141,10 +150,23 @@ impl ChannelsAdapter {
                                 max = super::wire::MAX_CHUNK_LEN,
                                 "demux: chunk too large, skipping payload bytes"
                             );
-                            let mut discard = vec![0u8; length as usize];
-                            if let Err(e) = reader.read_exact(&mut discard).await {
+                            let mut remaining = length as u64;
+                            while remaining > 0 {
+                                let take = remaining.min(SKIP_BUF_LEN as u64) as usize;
+                                if let Err(e) = reader.read_exact(&mut skip_buf[..take]).await {
                                     warn!(error = %e, "demux: failed to skip oversized payload");
-                                break;
+                                    return;
+                                }
+                                remaining -= take as u64;
+                            }
+                            skipped_since_last_valid += length as u64;
+                            if skipped_since_last_valid > MAX_CONSECUTIVE_SKIPPED_BYTES {
+                                warn!(
+                                    skipped_since_last_valid,
+                                    max = MAX_CONSECUTIVE_SKIPPED_BYTES,
+                                    "demux: skipped-bytes budget exceeded; tearing down connection"
+                                );
+                                return;
                             }
                             continue;
                         }
@@ -170,6 +192,7 @@ impl ChannelsAdapter {
                         }
                     };
                     manager.route_payload(header.channel_id, payload).await;
+                    skipped_since_last_valid = 0;
                 }
                 Err(e) => {
                     if e.kind() == std::io::ErrorKind::UnexpectedEof {

src/protocol/connection.rs

@@ -235,7 +235,7 @@ impl CallConnection {
         };
 
         if let Err(err) = self.write_request(send, &request_id, payload).await {
-            let call_error = CallError::internal(err);
+            let call_error = CallError::connection_closed(err);
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -276,7 +276,8 @@ impl CallConnection {
 
         let envelope = EventEnvelope::requested(&request_id, payload);
         if let Err(err) = writer.write_frame(&envelope).await {
-            let call_error = CallError::internal(format!("failed to write request frame: {err}"));
+            let call_error =
+                CallError::connection_closed(format!("failed to write request frame: {err}"));
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -353,7 +354,7 @@ impl CallConnection {
         };
 
         if let Err(err) = self.write_request(send, &request_id, payload).await {
-            let call_error = CallError::internal(err);
+            let call_error = CallError::connection_closed(err);
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -385,7 +386,8 @@ impl CallConnection {
 
         let envelope = EventEnvelope::requested(&request_id, payload);
         if let Err(err) = writer.write_frame(&envelope).await {
-            let call_error = CallError::internal(format!("failed to write request frame: {err}"));
+            let call_error =
+                CallError::connection_closed(format!("failed to write request frame: {err}"));
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -498,8 +500,17 @@ impl CallConnection {
         });
 
         let write_result = pump_publish_to_wire(send, &request_id, payload, stream).await;
-        if let Err(err) = write_result {
-            let call_error = CallError::internal(err);
+        if let Err((stage, err)) = write_result {
+            // A write failure before the request frame could be delivered is
+            // provably undelivered — retry is safe. Mid-publish failures leave
+            // delivery ambiguous; the INTERNAL mapping (non-retryable) is the
+            // safe default there.
+            let call_error = match stage {
+                PublishWriteStage::Request => CallError::connection_closed(err),
+                PublishWriteStage::Published | PublishWriteStage::Completed => {
+                    CallError::internal(err)
+                }
+            };
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -533,7 +544,8 @@ impl CallConnection {
 
         let requested = EventEnvelope::requested(&request_id, payload);
         if let Err(err) = writer.write_frame(&requested).await {
-            let call_error = CallError::internal(format!("failed to write request frame: {err}"));
+            let call_error =
+                CallError::connection_closed(format!("failed to write request frame: {err}"));
             self.pending
                 .lock()
                 .handle_error(&request_id, call_error.clone());
@@ -603,33 +615,48 @@ impl CallConnection {
     }
 }
 
+/// Which frame of a publish pump failed to write. The `Request` stage is
+/// provably undelivered (the producer never received `call.requested`, so a
+/// reconnect/retry is safe); the later stages leave delivery ambiguous.
+enum PublishWriteStage {
+    Request,
+    Published,
+    Completed,
+}
+
 async fn pump_publish_to_wire<W>(
     send: W,
     request_id: &str,
     payload: Value,
     mut stream: Pin<Box<dyn Stream<Item = Value> + Send>>,
-) -> Result<(), String>
+) -> Result<(), (PublishWriteStage, String)>
 where
     W: AsyncWrite + Unpin,
 {
     let mut writer = FrameFramedWriter::new(send);
     let requested = EventEnvelope::requested(request_id, payload);
-    writer
-        .write_frame(&requested)
-        .await
-        .map_err(|e| format!("failed to write request frame: {e}"))?;
+    writer.write_frame(&requested).await.map_err(|e| {
+        (
+            PublishWriteStage::Request,
+            format!("failed to write request frame: {e}"),
+        )
+    })?;
     while let Some(chunk) = stream.next().await {
         let envelope = EventEnvelope::published(request_id, chunk);
-        writer
-            .write_frame(&envelope)
-            .await
-            .map_err(|e| format!("failed to write published frame: {e}"))?;
+        writer.write_frame(&envelope).await.map_err(|e| {
+            (
+                PublishWriteStage::Published,
+                format!("failed to write published frame: {e}"),
+            )
+        })?;
     }
     let completed = EventEnvelope::completed(request_id);
-    writer
-        .write_frame(&completed)
-        .await
-        .map_err(|e| format!("failed to write completed frame: {e}"))?;
+    writer.write_frame(&completed).await.map_err(|e| {
+        (
+            PublishWriteStage::Completed,
+            format!("failed to write completed frame: {e}"),
+        )
+    })?;
     Ok(())
 }
 
@@ -1880,6 +1907,144 @@ mod tests {
         assert_eq!(read, envelope);
     }
 
+    // CF-001 — a transport write failure before the request frame could be
+    // delivered maps to retryable CONNECTION_CLOSED (the call provably
+    // never reached the producer). Mid-publish failures stay INTERNAL.
+
+    /// AsyncWrite that always errors — simulates a dead mux mid-call.
+    struct BrokenWrite;
+    impl AsyncWrite for BrokenWrite {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &[u8],
+        ) -> Poll<std::io::Result<usize>> {
+            Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "mux dead",
+            )))
+        }
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    fn broken_write_bi_stream() -> crate::core::types::BiStream {
+        // The joined stream holds the read end; the write half always errors.
+        let (read_end, _write_end) = tokio::io::duplex(1024);
+        crate::core::types::BiStream::from_joined(read_end, BrokenWrite)
+    }
+
+    #[tokio::test]
+    async fn single_stream_call_write_failure_is_retryable_connection_closed() {
+        let bi_stream = broken_write_bi_stream();
+        let (writer, reader) = split_single_stream(bi_stream);
+        let conn = CallConnection::new_single_stream(stub_connection(), writer);
+        drop(reader);
+
+        let response = conn.call("test/echo", serde_json::json!({})).await;
+        match response.result {
+            Err(e) => {
+                assert_eq!(e.code, "CONNECTION_CLOSED");
+                assert!(e.retryable);
+            }
+            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn multi_stream_call_write_request_failure_is_retryable() {
+        // A `Connection` whose open_bi succeeds but whose write half is
+        // dead: the request write fails, the call provably undelivered.
+        use crate::core::types::StreamError;
+        use crate::core::types::{BiStream, BidiStreamSource, Connection};
+        use std::net::SocketAddr;
+
+        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
+        #[async_trait::async_trait]
+        impl BidiStreamSource for OneBiStream {
+            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
+                Err(StreamError::ConnectionClosed)
+            }
+            async fn open_bi(&self) -> Result<BiStream, StreamError> {
+                let stream = self.0.lock().expect("open once").take();
+                stream.ok_or(StreamError::StreamClosed)
+            }
+            fn remote_addr(&self) -> Option<SocketAddr> {
+                None
+            }
+            fn close(&self, _code: u32, _reason: &str) {}
+        }
+
+        let conn = Connection::from_source(
+            OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))),
+            b"alk/call".to_vec(),
+        );
+        let call_conn = CallConnection::new(conn);
+        let response = call_conn.call("test/echo", serde_json::json!({})).await;
+        match response.result {
+            Err(e) => {
+                assert_eq!(e.code, "CONNECTION_CLOSED");
+                assert!(e.retryable);
+            }
+            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn multi_stream_publish_request_frame_write_failure_is_retryable() {
+        use crate::core::types::StreamError;
+        use crate::core::types::{BiStream, BidiStreamSource, Connection};
+        use std::net::SocketAddr;
+
+        struct OneBiStream(std::sync::Mutex<Option<crate::core::types::BiStream>>);
+        #[async_trait::async_trait]
+        impl BidiStreamSource for OneBiStream {
+            async fn accept_bi(&self) -> Result<BiStream, StreamError> {
+                Err(StreamError::ConnectionClosed)
+            }
+            async fn open_bi(&self) -> Result<BiStream, StreamError> {
+                let stream = self.0.lock().expect("open once").take();
+                stream.ok_or(StreamError::StreamClosed)
+            }
+            fn remote_addr(&self) -> Option<SocketAddr> {
+                None
+            }
+            fn close(&self, _code: u32, _reason: &str) {}
+        }
+
+        let conn = Connection::from_source(
+            OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))),
+            b"alk/call".to_vec(),
+        );
+        let call_conn = CallConnection::new(conn);
+        let response = call_conn
+            .publish(
+                "fs/upload",
+                serde_json::json!({}),
+                Box::pin(futures::stream::iter(vec![serde_json::json!({"c": 1})])),
+            )
+            .await;
+        match response.result {
+            Err(e) => {
+                // The pump died on the *request* frame — nothing delivered.
+                assert_eq!(e.code, "CONNECTION_CLOSED");
+                assert!(e.retryable);
+            }
+            other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn connection_closed_error_is_retryable() {
+        let err = CallError::connection_closed("test");
+        assert_eq!(err.code, "CONNECTION_CLOSED");
+        assert!(err.retryable);
+    }
+
     #[tokio::test]
     async fn shared_frame_writer_concurrent_writes_do_not_interleave() {
         let (client_end, server_end) = tokio::io::duplex(64 * 1024);

src/protocol/dispatch.rs

@@ -349,21 +349,7 @@ impl Dispatcher {
                         Ok(h) => h,
                         Err(envelope) => return DispatchResult::Once(envelope),
                     };
-                let publish_validator = self
-                    .registry
-                    .registration(&operation_name)
-                    .and_then(|r| r.spec.publish_schema.as_ref())
-                    .and_then(|schema| match jsonschema::options().build(schema) {
-                        Ok(v) => Some(v),
-                        Err(e) => {
-                            warn!(
-                                operation = %operation_name,
-                                error = %e,
-                                "publish_schema failed to compile; chunks will not be validated",
-                            );
-                            None
-                        }
-                    });
+                let publish_validator = self.registry.publish_validator(&operation_name);
                 let (chunk_tx, chunk_rx) =
                     mpsc::channel::<Result<Value, CallError>>(PUBLISH_CHANNEL_BUFFER);
                 let publish_stream: PublishStream = Box::pin(chunk_rx);

src/protocol/wire.rs

@@ -113,6 +113,10 @@ impl CallError {
         Self::new("TIMEOUT", message, true)
     }
 
+    pub fn connection_closed(message: impl Into<String>) -> Self {
+        Self::new("CONNECTION_CLOSED", message, true)
+    }
+
     pub fn invalid_operation_type(message: impl Into<String>) -> Self {
         Self::new("INVALID_OPERATION_TYPE", message, false)
     }

src/registry/discovery.rs

@@ -4,7 +4,7 @@ use serde_json::{json, Value};
 
 use super::context::OperationContext;
 use super::registration::{Handler, OperationRegistry};
-use super::spec::{AccessControl, OperationSpec, OperationType, Visibility};
+use super::spec::{AccessControl, AccessResult, OperationSpec, OperationType, Visibility};
 use crate::protocol::wire::{CallError, ResponseEnvelope};
 
 const NAME_SERVICES_LIST: &str = "services/list";
@@ -337,13 +337,35 @@ pub fn services_schema_handler(registry: Arc<OperationRegistry>) -> Handler {
                     );
                 }
             };
-            match registry.registration(&name) {
-                Some(reg) => {
-                    let spec_json = spec_to_json(&reg.spec);
+            let registration = match registry.registration(&name) {
+                Some(reg) => reg,
+                None => return ResponseEnvelope::not_found(ctx.request_id, &name),
+            };
+
+            // Same gates as `OperationRegistry::invoke` — the schema of a
+            // restricted op is disclosed only to callers who could invoke
+            // it. Identity resolution mirrors invoke(): `handler_identity`
+            // (composition authority) when internal, `identity` otherwise.
+            if registration.spec.visibility == Visibility::Internal && !ctx.internal {
+                return ResponseEnvelope::not_found(ctx.request_id, &name);
+            }
+            let identity = if ctx.internal {
+                ctx.handler_identity
+                    .as_ref()
+                    .and_then(|ca| ca.as_identity())
+            } else {
+                ctx.identity.clone()
+            };
+            if let AccessResult::Forbidden(_) = registration.spec.access_control.check(
+                identity.as_ref(),
+                None,
+                ctx.ownership.as_deref(),
+            ) {
+                return ResponseEnvelope::not_found(ctx.request_id, &name);
+            }
+
+            let spec_json = spec_to_json(&registration.spec);
             ResponseEnvelope::ok(ctx.request_id, spec_json)
-                }
-                None => ResponseEnvelope::not_found(ctx.request_id, &name),
-            }
         })
     })
 }
@@ -720,6 +742,81 @@ mod tests {
         }
     }
 
+    // CF-004 — the schema handler must apply the same visibility + ACL
+    // gates as invoke(). Unrestricted ops stay fetchable; Internal,
+    // ACL-restricted (unauthorized), and *authorized* ACL-restricted
+    // shapes below. Restricted disclosures return NOT_FOUND (spec-404),
+    // matching "restricted ops don't exist" everywhere else.
+
+    #[tokio::test]
+    async fn services_schema_hides_internal_op_from_external_caller() {
+        let registry = registry_with_access_controlled_ops();
+        let handler = services_schema_handler(Arc::clone(&registry));
+        let ctx = root_context("req-cf4-1");
+        let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
+            other => panic!("expected NOT_FOUND for internal op, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn services_schema_shows_internal_op_to_internal_caller() {
+        let registry = registry_with_access_controlled_ops();
+        let handler = services_schema_handler(Arc::clone(&registry));
+        let mut ctx = root_context("req-cf4-2");
+        ctx.internal = true;
+        ctx.handler_identity = Some(CompositionAuthority::new("agent-chat", []));
+        let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await;
+        let spec = response.result.expect("internal caller sees internal op");
+        assert_eq!(spec.get("name"), Some(&json!("internal/hidden")));
+    }
+
+    #[tokio::test]
+    async fn services_schema_hides_acl_restricted_op_from_unauthorized_caller() {
+        let registry = registry_with_access_controlled_ops();
+        let handler = services_schema_handler(Arc::clone(&registry));
+        let ctx = root_context_with_identity(
+            "req-cf4-3",
+            Some(identity_with_scopes("regular-peer", &["user"])),
+        );
+        let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await;
+        match response.result {
+            Err(e) => assert_eq!(e.code, "NOT_FOUND"),
+            other => panic!("expected NOT_FOUND for unauthorized ACL, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn services_schema_shows_acl_restricted_op_to_authorized_caller() {
+        let registry = registry_with_access_controlled_ops();
+        let handler = services_schema_handler(Arc::clone(&registry));
+        let ctx = root_context_with_identity(
+            "req-cf4-4",
+            Some(identity_with_scopes("admin-peer", &["admin"])),
+        );
+        let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await;
+        let spec = response.result.expect("authorized caller sees the spec");
+        assert_eq!(spec.get("name"), Some(&json!("admin/secret")));
+        assert_eq!(
+            spec.get("access_control")
+                .and_then(|a| a.get("required_scopes")),
+            Some(&json!(["admin"]))
+        );
+    }
+
+    #[tokio::test]
+    async fn services_schema_unrestricted_op_fetchable_without_identity() {
+        let registry = registry_with_access_controlled_ops();
+        let handler = services_schema_handler(Arc::clone(&registry));
+        let ctx = root_context("req-cf4-5");
+        let response = handler(serde_json::json!({ "name": "public/echo" }), ctx).await;
+        let spec = response
+            .result
+            .expect("default-ACL op fetchable unauthenticated");
+        assert_eq!(spec.get("name"), Some(&json!("public/echo")));
+    }
+
     #[tokio::test]
     async fn services_list_handler_registered_and_invocable_via_registry() {
         let registry = registry_with_ops();

src/registry/registration.rs

@@ -96,15 +96,44 @@ impl HandlerRegistration {
 
 pub struct OperationRegistry {
     operations: HashMap<String, HandlerRegistration>,
+    validators: HashMap<String, Option<jsonschema::Validator>>,
 }
 
 impl OperationRegistry {
     pub fn new() -> Self {
         Self {
             operations: HashMap::new(),
+            validators: HashMap::new(),
         }
     }
 
+    /// Compile a spec's `publish_schema` (CF-003). Un-compilable schemas
+    /// are rejected at registration time — registering one is an error,
+    /// never a silent unvalidated ingest path. A compiled validator is
+    /// cached so the dispatch path does zero per-request work; the cached
+    /// value also distinguishes "no schema declared" (`None`) from a
+    /// schema that failed to compile, though the latter can no longer
+    /// reach this registry.
+    fn validate_and_cache_publish_schema(
+        validators: &mut HashMap<String, Option<jsonschema::Validator>>,
+        spec: &OperationSpec,
+    ) -> Result<(), String> {
+        let compiled = match &spec.publish_schema {
+            None => None,
+            Some(schema) => match jsonschema::options().build(schema) {
+                Ok(v) => Some(v),
+                Err(e) => {
+                    return Err(format!(
+                        "publish_schema for operation {:?} failed to compile: {e}",
+                        spec.name
+                    ));
+                }
+            },
+        };
+        validators.insert(spec.name.clone(), compiled);
+        Ok(())
+    }
+
     pub fn register(&mut self, registration: HandlerRegistration) -> Result<(), String> {
         let expected = match registration.spec.op_type {
             OperationType::Query | OperationType::Mutation => "Once",
@@ -122,6 +151,7 @@ impl OperationRegistry {
                 registration.spec.op_type, expected, actual
             ));
         }
+        Self::validate_and_cache_publish_schema(&mut self.validators, &registration.spec)?;
         self.operations
             .insert(registration.spec.name.clone(), registration);
         Ok(())
@@ -131,6 +161,14 @@ impl OperationRegistry {
         self.operations.get(name)
     }
 
+    /// The cached per-chunk validator for a Pub op's `publish_schema`
+    /// (ADR-046 §4). `None` when the op declares no schema. Compiled at
+    /// registration time — a schema that fails to compile never reaches
+    /// this registry, so the dispatch path cannot fail open (CF-003).
+    pub fn publish_validator(&self, name: &str) -> Option<jsonschema::Validator> {
+        self.validators.get(name).cloned().flatten()
+    }
+
     pub fn list_operations(&self) -> Vec<&OperationSpec> {
         self.operations
             .values()
@@ -371,21 +409,34 @@ impl Default for OperationRegistry {
 
 pub struct OperationRegistryBuilder {
     operations: HashMap<String, HandlerRegistration>,
+    validators: HashMap<String, Option<jsonschema::Validator>>,
 }
 
 impl OperationRegistryBuilder {
     pub fn new() -> Self {
         Self {
             operations: HashMap::new(),
+            validators: HashMap::new(),
         }
     }
 
     fn store(mut self, registration: HandlerRegistration) -> Result<Self, String> {
+        OperationRegistry::validate_and_cache_publish_schema(
+            &mut self.validators,
+            &registration.spec,
+        )?;
         let name = registration.spec.name.clone();
         self.operations.insert(name, registration);
         Ok(self)
     }
 
+    pub fn build(self) -> OperationRegistry {
+        OperationRegistry {
+            operations: self.operations,
+            validators: self.validators,
+        }
+    }
+
     fn wrap_once(spec: &OperationSpec, handler: Handler) -> Result<HandlerKind, String> {
         match spec.op_type {
             OperationType::Query | OperationType::Mutation => Ok(HandlerKind::Once(handler)),
@@ -572,12 +623,6 @@ impl OperationRegistryBuilder {
     pub fn with(self, registration: HandlerRegistration) -> Result<Self, String> {
         self.store(registration)
     }
-
-    pub fn build(self) -> OperationRegistry {
-        OperationRegistry {
-            operations: self.operations,
-        }
-    }
 }
 
 impl Default for OperationRegistryBuilder {
@@ -1852,6 +1897,129 @@ mod tests {
         )
     }
 
+    // --- CF-003: publish_schema validated at registration time -------------
+
+    fn make_sink_noop() -> SinkHandler {
+        make_sink_handler(|_input, ctx, _stream| async move {
+            ResponseEnvelope::ok(ctx.request_id, serde_json::json!({}))
+        })
+    }
+
+    fn registering_pub_with_schema(schema: Value) -> Result<(), String> {
+        let mut registry = OperationRegistry::new();
+        registry.register(HandlerRegistration::new(
+            pub_spec("fs/upload", AccessControl::default()).with_publish_schema(schema),
+            HandlerKind::Sink(make_sink_noop()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ))
+    }
+
+    #[test]
+    fn register_rejects_uncompilable_publish_schema() {
+        // `required` must be an array — jsonschema rejects this at compile.
+        let err = registering_pub_with_schema(serde_json::json!({
+            "type": "object",
+            "required": "not-an-array"
+        }))
+        .expect_err("un-compilable schema must be rejected at registration");
+        assert!(
+            err.contains("failed to compile"),
+            "error names the failure: {err}"
+        );
+        assert!(err.contains("fs/upload"), "error names the op: {err}");
+    }
+
+    #[test]
+    fn register_rejects_invalid_type_and_pattern_schema() {
+        assert!(registering_pub_with_schema(serde_json::json!({
+            "type": 123
+        }))
+        .is_err());
+
+        assert!(registering_pub_with_schema(serde_json::json!({
+            "type": "object",
+            "properties": { "x": { "pattern": "(unclosed" } }
+        }))
+        .is_err());
+    }
+
+    #[test]
+    fn register_accepts_compilable_publish_schema_and_caches_validator() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()).with_publish_schema(
+                    serde_json::json!({
+                        "type": "object",
+                        "properties": { "bytes": { "type": "string" } },
+                        "required": ["bytes"]
+                    }),
+                ),
+                HandlerKind::Sink(make_sink_noop()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .expect("compilable schema registers");
+        assert!(
+            registry.publish_validator("fs/upload").is_some(),
+            "validator cached at registration"
+        );
+    }
+
+    #[test]
+    fn publish_validator_none_when_no_schema_declared() {
+        let mut registry = OperationRegistry::new();
+        registry
+            .register(HandlerRegistration::new(
+                pub_spec("fs/upload", AccessControl::default()),
+                HandlerKind::Sink(make_sink_noop()),
+                OperationProvenance::Local,
+                None,
+                None,
+                Capabilities::new(),
+            ))
+            .expect("no-schema pub registers");
+        assert!(registry.publish_validator("fs/upload").is_none());
+        assert!(registry.publish_validator("no/such").is_none());
+    }
+
+    #[test]
+    fn registry_builder_rejects_uncompilable_publish_schema() {
+        let result = OperationRegistryBuilder::new().with_local_sink(
+            pub_spec("fs/upload", AccessControl::default())
+                .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })),
+            make_sink_noop(),
+            None,
+            None,
+            Capabilities::new(),
+        );
+        assert!(result.is_err(), "builder path validates too");
+    }
+
+    #[test]
+    fn failed_validation_leaves_no_validator_cache_entry() {
+        let mut registry = OperationRegistry::new();
+        let result = registry.register(HandlerRegistration::new(
+            pub_spec("fs/upload", AccessControl::default())
+                .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })),
+            HandlerKind::Sink(make_sink_noop()),
+            OperationProvenance::Local,
+            None,
+            None,
+            Capabilities::new(),
+        ));
+        assert!(result.is_err());
+        assert!(
+            registry.registration("fs/upload").is_none(),
+            "op not registered on schema failure"
+        );
+    }
+
     fn internal_pub_spec(acl: AccessControl) -> OperationSpec {
         OperationSpec::new(
             "fs/upload",