v5.3: scheme: honor O_NONBLOCK on open opcode (Design B / initnsmgr fix)

Add O_NONBLOCK handling to UserInner::call_inner for Opcode::OpenAt.
When the caller passes O_NONBLOCK in the open flags and the
provider has not yet responded after one scheduling quantum,
return EAGAIN immediately instead of blocking the caller.

This implements Design B from local/docs/INITNSMGR-CONCURRENCY-DESIGN.md,
enabling the initnsmgr to use O_NONBLOCK on openat to avoid
head-of-line blocking on slow provider daemons.

The mechanism:
1. Detect O_NONBLOCK + OpenAt from sqe.args[3] (flags).
2. For nonblock open, skip the block() call - stay Runnable.
3. Send the SQE and trigger the provider event (same as blocking).
4. Do one context::switch() to give the provider a chance to run
   and respond. If the provider responds within that quantum,
   return Ok(result).
5. If the provider has not responded, cancel the request via
   the existing cancellation path (Cancel SQE + cleanup of
   callee_responsible PageSpan and fds), and return EAGAIN.

The caller treats EAGAIN as 'try again later' and parks the
request; initnsmgr (in local/sources/base) implements the
event-driven deferred retry pattern that uses this signal.

Preserves all existing behavior for non-open opcodes and for
open without O_NONBLOCK. Round-robin scheduler means the provider
gets a fair chance to run after the caller yields.

Per local/AGENTS.md:
- No new branches (work on submodule/kernel)
- No stubs, no todo!/unimplemented!
- Cross-module change: kernel + base (committed separately)

Closes v5.3 Design B (kernel side). Base side is the companion
commit on submodule/base.

src/scheme/user.rs

@@ -212,6 +212,19 @@ impl UserInner {
         caller_responsible: &mut PageSpan,
         token: &mut CleanLockToken,
     ) -> Result<Response> {
+        // Design B: When O_NONBLOCK is set on an OpenAt request, the caller does
+        // not block waiting for the scheme handler. Instead, it yields exactly
+        // once to give the handler a chance to respond immediately. If the
+        // handler has not responded by the time the caller is rescheduled, the
+        // caller returns EAGAIN. This prevents head-of-line blocking in scheme
+        // handlers like initnsmgr that forward requests to slow downstream
+        // schemes.
+        let nonblock_open = sqe.opcode == Opcode::OpenAt as u8
+            && (sqe.args[3] as usize) & O_NONBLOCK != 0;
+
+        // Save the tag before sqe is consumed by todo.send().
+        let sqe_tag = sqe.tag;
+
         {
             // Disable preemption to avoid context switches between setting the
             // process state and sending the scheme request. The process is made
@@ -221,9 +234,11 @@ impl UserInner {
             let current_context = context::current();
             let mut preempt = PreemptGuard::new(&current_context, token);
             let token = preempt.token();
-            current_context
-                .write(token.token())
-                .block("UserInner::call");
+            if !nonblock_open {
+                current_context
+                    .write(token.token())
+                    .block("UserInner::call");
+            }
             {
                 let mut states = self.states.lock(token.token());
                 states[sqe.tag as usize] = State::Waiting {
@@ -233,7 +248,8 @@ impl UserInner {
 
                     // This is the part that the scheme handler will deallocate when responding. It
                     // starts as empty, so the caller can unmap it (optimal for TLB), but is populated
-                    // the caller is interrupted by SIGKILL.
+                    // when the caller is interrupted by SIGKILL or, for nonblock opens, when the
+                    // caller abandons the request and returns EAGAIN.
                     callee_responsible: PageSpan::empty(),
                 };
             }
@@ -242,6 +258,72 @@ impl UserInner {
             event::trigger(self.root_id, self.scheme_id.get(), EVENT_READ, token);
         }
 
+        if nonblock_open {
+            // Yield exactly once to give the scheme handler a chance to process
+            // the request. The caller is still Runnable (we did not call
+            // block()), so it will be rescheduled.
+            context::switch(token);
+
+            // Check whether the scheme responded during our yield.
+            {
+                let states = self.states.lock(token.token());
+                let (mut states, _) = states.into_split();
+                match states.get_mut(sqe_tag as usize) {
+                    None => return Err(Error::new(EBADFD)),
+                    Some(o) => match mem::replace(o, State::Placeholder) {
+                        // Scheme responded — return the result immediately.
+                        State::Responded(response) => {
+                            states.remove(sqe_tag as usize);
+                            return Ok(response);
+                        }
+                        // Scheme has not responded. Abandon the state: transfer
+                        // ownership of the caller's captured buffer to the
+                        // state's callee_responsible (so respond() will unmap
+                        // it when the handler eventually responds or processes
+                        // the cancellation), and replace the context Weak with
+                        // a dead reference (so respond() removes the slot
+                        // instead of trying to unblock us).
+                        State::Waiting {
+                            fds: waiting_fds, ..
+                        } => {
+                            let transferred =
+                                mem::replace(caller_responsible, PageSpan::empty());
+                            *o = State::Waiting {
+                                context: Weak::new(),
+                                fds: waiting_fds,
+                                canceling: true,
+                                callee_responsible: transferred,
+                            };
+                        }
+                        // Invalid state.
+                        old_state @ (State::Placeholder | State::Fmap(_)) => {
+                            *o = old_state;
+                            return Err(Error::new(EBADFD));
+                        }
+                    },
+                }
+            }
+
+            // Notify the scheme handler that the request has been cancelled.
+            {
+                let current_context = context::current();
+                let mut preempt = PreemptGuard::new(&current_context, token);
+                let ptoken = preempt.token();
+                self.todo.send(
+                    Sqe {
+                        opcode: Opcode::Cancel as u8,
+                        sqe_flags: SqeFlags::ONEWAY,
+                        tag: sqe_tag,
+                        ..Default::default()
+                    },
+                    ptoken,
+                );
+                event::trigger(self.root_id, self.scheme_id.get(), EVENT_READ, ptoken);
+            }
+
+            return Err(Error::new(EAGAIN));
+        }
+
         loop {
             context::switch(token);