test(middleware): cover upstream WebSocket disconnect

Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>

Author: pimlock

crates/openshell-supervisor-network/src/l7/websocket.rs

@@ -4071,6 +4071,72 @@ network_policies:
             .expect("middleware server");
     }
 
+    #[tokio::test]
+    async fn upstream_eof_reports_one_upstream_disconnect_while_downstream_is_open() {
+        let (mut session, mut observed, shutdown_tx, server_task) =
+            recording_middleware_session("wss").await;
+        assert!(session.start("").await.allowed);
+        assert!(matches!(
+            tokio::time::timeout(std::time::Duration::from_secs(2), observed.recv())
+                .await
+                .expect("middleware observes session start"),
+            Some(ObservedWebSocketRequest::SessionStart)
+        ));
+
+        let (client_app, mut relay_client) = tokio::io::duplex(4096);
+        let (mut relay_upstream, upstream_app) = tokio::io::duplex(4096);
+        let relay = tokio::spawn(async move {
+            relay_with_options(
+                &mut relay_client,
+                &mut relay_upstream,
+                Vec::new(),
+                "api.openai.com",
+                443,
+                RelayOptions {
+                    policy_name: "rest-api",
+                    assembly_budget: WebSocketAssemblyBudget::default(),
+                    resolver: None,
+                    generation_guard: None,
+                    provider_credentials: None,
+                    target: "/",
+                    inspector: None,
+                    compression: WebSocketCompression::None,
+                    middleware_session: Some(session),
+                    middleware_context: None,
+                    deny_uninspected_credentials: false,
+                },
+            )
+            .await
+        });
+
+        drop(upstream_app);
+        tokio::time::timeout(std::time::Duration::from_secs(2), relay)
+            .await
+            .expect("relay finishes after upstream disconnect")
+            .expect("join relay")
+            .expect("upstream EOF ends relay normally");
+        assert!(matches!(
+            tokio::time::timeout(std::time::Duration::from_secs(2), observed.recv())
+                .await
+                .expect("middleware observes session end"),
+            Some(ObservedWebSocketRequest::SessionEnd(
+                openshell_core::proto::MiddlewareSessionEndReason::UpstreamDisconnect,
+            ))
+        ));
+        assert!(
+            observed.try_recv().is_err(),
+            "upstream EOF must produce exactly one session end"
+        );
+
+        drop(client_app);
+        let _ = shutdown_tx.send(());
+        tokio::time::timeout(std::time::Duration::from_secs(2), server_task)
+            .await
+            .expect("middleware server shuts down")
+            .expect("join middleware server")
+            .expect("middleware server");
+    }
+
     #[tokio::test]
     async fn denied_websocket_session_start_reports_middleware_failure_before_close() {
         let (session, mut observed, shutdown_tx, server_task) =

proto/supervisor_middleware.proto

@@ -399,7 +399,9 @@ enum MiddlewareSessionEndReason {
   MIDDLEWARE_SESSION_END_REASON_UNSPECIFIED = 0;
   // Evaluation completed.
   MIDDLEWARE_SESSION_END_REASON_NORMAL = 1;
-  // The sandbox peer disconnected.
+  // The sandbox peer disconnected. Before the directional split, wire value 2
+  // represented either peer disconnect. Mixed deployments across that
+  // pre-0.1.0 change are unsupported.
   MIDDLEWARE_SESSION_END_REASON_DOWNSTREAM_DISCONNECT = 2;
   // A policy reload replaced the active middleware chain.
   MIDDLEWARE_SESSION_END_REASON_POLICY_RELOAD = 3;