fix(sdk-java): report a close-aborted parked SSE read as a transport failure (#13314)

Co-authored-by: Qwen-Coder <[email protected]>

packages/sdk-java/qwencode/src/main/java/com/alibaba/qwen/code/daemon/HarnessEventStream.java

@@ -98,16 +98,22 @@ public final class HarnessEventStream implements AutoCloseable {
                 client.observeEvent(session, event);
                 return event;
             } catch (IOException e) {
-                if (closed.get()) {
-                    // close() ended the blocked read.
-                    throw new IllegalStateException(
-                            "HarnessEventStream is closed", e);
-                }
-                closeQuietly();
                 if (idleTimedOut.get()) {
+                    // The watchdog aborted the stream: the parked consumer
+                    // gets the idle-timeout signal whether or not the
+                    // watchdog's close() has landed yet.
                     throw new DaemonTransportException(
                             "Hosted Harness SSE idle timeout", e);
                 }
+                if (closed.get()) {
+                    // A caller-initiated close() ended the blocked read.
+                    // The parked consumer gets a transport failure; a
+                    // caller arriving after the close gets the closed
+                    // guard's signal instead.
+                    throw new DaemonTransportException(
+                            "HarnessEventStream is closed", e);
+                }
+                closeQuietly();
                 throw new DaemonTransportException(
                         "Hosted Harness SSE stream failed", e);
             } catch (RuntimeException e) {

packages/sdk-java/qwencode/src/test/java/com/alibaba/qwen/code/daemon/HarnessEventStreamCloseTest.java

@@ -149,9 +149,13 @@ class HarnessEventStreamCloseTest {
         reader.join(TimeUnit.SECONDS.toMillis(10));
         assertFalse(reader.isAlive(),
                 "the blocked read did not end when its stream closed");
-        // A local close is not the server ending the stream.
-        IllegalStateException closed = assertInstanceOf(
-                IllegalStateException.class, outcome.get());
+        // A local close is not the server ending the stream: the read
+        // blocked on the socket is cut at the transport layer, so the
+        // parked consumer gets a DaemonTransportException, while a
+        // caller arriving after the close still gets the closed guard's
+        // IllegalStateException.
+        DaemonTransportException closed = assertInstanceOf(
+                DaemonTransportException.class, outcome.get());
         assertEquals("HarnessEventStream is closed", closed.getMessage());
         client.close();
     }