fix(test):sse stream integration

AGENTS.md

@@ -164,11 +164,11 @@ alike (`?environment={name}`, never `?env=`). Nothing outside the `environment`
 environment ids.
 
 ### Pipelines
-Build pipelines run as Kubernetes Jobs with init containers:
-1. **clone**: Clones source code from git (or downloads ZIP via curl for ZIP source type)
+Build pipelines run as Kubernetes Jobs with init containers. The container names are what the step bar and the `steps` SSE event show, so they are part of the contract the integration suite pins — `fetch`, `dockerfile`, `compile`, `publish`, then the trailing `done` main container:
+1. **fetch** (`CloneContainer`): Clones source code from git (or downloads ZIP via curl for ZIP source type)
 2. **dockerfile** (optional): Writes an inline `USER` Dockerfile into the workspace
-3. **build** (optional): Runs custom build commands
-4. **push**: Builds and pushes Docker image using Buildah, with the `overlay` storage driver over the `container-storage` emptyDir that `ContainerStorageVolume` mounts at `/var/lib/containers`. Both halves matter: the store defaults into the container's writable layer, where every write pays an overlayfs copy-up, and `vfs` (used until the volume existed) copies the entire tree per layer — together they made a build of any size stall for minutes with no log output after its last line. Buildah's own image declares a `VOLUME` for this path, but Kubernetes ignores image `VOLUME` declarations, so the mount has to be explicit.
+3. **compile** (optional, `CompileContainer`): Runs custom build commands
+4. **publish** (`PublishContainer`): Builds and pushes Docker image using Buildah, with the `overlay` storage driver over the `container-storage` emptyDir that `ContainerStorageVolume` mounts at `/var/lib/containers`. Both halves matter: the store defaults into the container's writable layer, where every write pays an overlayfs copy-up, and `vfs` (used until the volume existed) copies the entire tree per layer — together they made a build of any size stall for minutes with no log output after its last line. Buildah's own image declares a `VOLUME` for this path, but Kubernetes ignores image `VOLUME` declarations, so the mount has to be explicit.
 
 Two source types exist: `GIT` (default) and `ZIP`. ZIP uploads use presigned S3 URLs via `BuildSourceObjectStorageService` — the frontend gets a presigned PUT URL from `POST .../deployments/source-upload`, uploads the file, then triggers the pipeline. ZIP builds use `oops.pipeline.image.zip` (defaults to `alpine/curl:8.17.0`) to download the archive.
 

src/main/java/com/github/wellch4n/oops/infrastructure/config/SecurityConfiguration.java

@@ -1,5 +1,6 @@
 package com.github.wellch4n.oops.infrastructure.config;
 
+import jakarta.servlet.DispatcherType;
 import jakarta.servlet.http.HttpServletResponse;
 import java.util.List;
 import org.springframework.context.annotation.Bean;
@@ -43,6 +44,14 @@ public SecurityFilterChain filterChain(HttpSecurity http) throws Exception {
                                 response.sendError(HttpServletResponse.SC_FORBIDDEN))
                 )
                 .authorizeHttpRequests(auth -> auth
+                        // An SSE response (pipeline logs, application events) completes on an
+                        // ASYNC dispatch that walks this chain again. JwtAuthFilter, like any
+                        // OncePerRequestFilter, sits that dispatch out, so nothing is
+                        // authenticated by then and "/api/**" would deny a request it already
+                        // let through — with the response committed, the only effect is the
+                        // connection being cut before the final chunk. The ASYNC dispatch is
+                        // only ever reached through an authorized REQUEST dispatch.
+                        .dispatcherTypeMatchers(DispatcherType.ASYNC).permitAll()
                         .requestMatchers("/api/auth/**").permitAll()
                         .requestMatchers("/api/auth/feishu/**").permitAll()
                         .requestMatchers("/api/health").permitAll()

src/main/java/com/github/wellch4n/oops/interfaces/rest/GlobalExceptionHandler.java

@@ -6,6 +6,7 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.RestControllerAdvice;
+import org.springframework.web.context.request.async.AsyncRequestNotUsableException;
 
 @RestControllerAdvice
 public class GlobalExceptionHandler {
@@ -17,6 +18,18 @@ public Result<Void> handleBizException(BizException exception) {
         return Result.failure(exception.getMessage());
     }
 
+    /**
+     * A browser closing an {@code EventSource} mid-stream — leaving the pipeline page, switching to
+     * another step's log — reaches here as the container's error dispatch for the SSE request.
+     * Nothing is wrong and nobody is listening: the catch-all below would log it as an error and
+     * then fail again trying to write a JSON envelope onto a {@code text/event-stream} response.
+     * Returning nothing marks the response as handled, which for a dead connection is exactly right.
+     */
+    @ExceptionHandler(AsyncRequestNotUsableException.class)
+    public void handleDisconnectedClient(AsyncRequestNotUsableException exception) {
+        log.debug("Client disconnected from a streaming response: {}", exception.getMessage());
+    }
+
     @ExceptionHandler(Exception.class)
     public Result<Void> handleException(Exception exception) {
         log.error("Unhandled exception", exception);

tests/integration/README.md

@@ -1,6 +1,6 @@
 # Integration suite
 
-Black-box tests against the OOPS HTTP and WebSocket API. They import nothing from
+Black-box tests against the OOPS HTTP, SSE and WebSocket API. They import nothing from
 the backend and know nothing about its language, so the same suite is the
 acceptance criteria for the current Java service, a Go or Rust rewrite, or a
 GraalVM native build.
@@ -15,7 +15,7 @@ observable behaviour while a correct implementation still exists to check agains
 | Tier | Needs | Covers |
 |---|---|---|
 | contract | the compose middleware | the wire contract, users and authorisation, domains, application configuration round trips, environment management, object storage presigning, git branch listing |
-| `cluster` | the above plus k3s | build and deploy pipelines, pipeline operations, both registry credential paths, git clone with and without credentials, ingress routes and basic auth, deletion in the cluster, sandboxes, pod filesystem, log and terminal sockets |
+| `cluster` | the above plus k3s | build and deploy pipelines, pipeline operations, both registry credential paths, git clone with and without credentials, ingress routes and basic auth, deletion in the cluster, sandboxes, pod filesystem, pipeline step and log event streams, pod log and terminal sockets |
 
 Organised by scenario rather than by endpoint: what a rewrite has to preserve is
 behaviour, and every defect this suite found needed a full journey to surface.

tests/integration/oops_client.py

@@ -1,4 +1,4 @@
-"""HTTP and WebSocket client for the OOPS API.
+"""HTTP, WebSocket and server-sent event client for the OOPS API.
 
 Deliberately knows nothing about the backend implementation — it speaks only the
 wire protocol, so the same suite validates the current Java service, a Go or Rust
@@ -246,6 +246,73 @@ def websocket(self, path: str, timeout: int = 60) -> websocket.WebSocket:
         connection = websocket.create_connection(url, timeout=timeout)
         return connection
 
+    # -- server-sent events -------------------------------------------------
+
+    def sse(self, path: str, timeout: int = 60,
+            headers: Optional[dict] = None) -> Iterator[SseEvent]:
+        """Open an event stream and yield its events until the server ends the
+        response or `timeout` seconds pass with nothing arriving.
+
+        A GET like any other, so it is recorded as one for coverage. `headers`
+        is for what a browser's EventSource sends on its own — `Last-Event-ID`
+        on a reconnect — since that is the only input these streams take beyond
+        the URL.
+        """
+        record("GET", path)
+        request_headers = {"Accept": "text/event-stream"}
+        if self.token:
+            request_headers["Authorization"] = f"Bearer {self.token}"
+        if headers:
+            request_headers.update(headers)
+        response = self.session.get(self.endpoint + path, headers=request_headers,
+                                    stream=True, timeout=timeout)
+        try:
+            if response.status_code != 200:
+                raise ApiError(
+                    f"GET {path} returned {response.status_code} instead of an "
+                    f"event stream: {response.text[:400]!r}")
+            content_type = response.headers.get("Content-Type", "")
+            if not content_type.startswith("text/event-stream"):
+                raise ApiError(
+                    f"GET {path} answered with {content_type!r}, not text/event-stream")
+            yield from parse_sse(response.iter_lines())
+        finally:
+            response.close()
+
+
+@dataclass
+class SseEvent:
+    """One server-sent event: the `event:` name, the `id:` if the server set one,
+    and the `data:` lines joined back together."""
+
+    event: str
+    id: Optional[str]
+    data: str
+
+
+def parse_sse(lines: Iterator[bytes]) -> Iterator[SseEvent]:
+    """Decode the wire format: `field:value` lines, a blank line ends an event,
+    a leading `:` is a comment. Yields events in arrival order."""
+    event, event_id, data = "message", None, []
+    for raw in lines:
+        line = raw.decode("utf-8", errors="replace")
+        if line == "":
+            if data or event != "message":
+                yield SseEvent(event, event_id, "\n".join(data))
+            event, event_id, data = "message", None, []
+            continue
+        if line.startswith(":"):
+            continue
+        field, _, value = line.partition(":")
+        if value.startswith(" "):
+            value = value[1:]
+        if field == "event":
+            event = value
+        elif field == "id":
+            event_id = value
+        elif field == "data":
+            data.append(value)
+
 
 def wait_until(predicate: Callable[[], Any], timeout: int = 300, interval: float = 3.0,
                description: str = "condition") -> Any:
@@ -279,17 +346,36 @@ def read_until_closed(connection: websocket.WebSocket,
 
     Log sockets answer a text "ping" with "pong"; terminal sockets must not be
     pinged, because their stdin is live.
+
+    Frames are read one at a time rather than through `recv()`, which swallows
+    control frames and waits on for a data frame with a fresh timeout each time
+    — and every handler sends a protocol-level ping every few seconds, so an
+    idle socket would never time out. The pings are answered here instead, and
+    the deadline is the wall clock.
     """
     deadline = time.time() + timeout
-    while time.time() < deadline:
+    while True:
+        remaining = deadline - time.time()
+        if remaining <= 0:
+            return
+        connection.settimeout(remaining)
         try:
-            frame = connection.recv()
+            frame = connection.recv_frame()
         except websocket.WebSocketTimeoutException:
             return
         except websocket.WebSocketConnectionClosedException:
             return
-        if frame is None or frame == "":
+        if frame is None or frame.opcode == websocket.ABNF.OPCODE_CLOSE:
             return
-        if isinstance(frame, bytes):
-            frame = frame.decode("utf-8", errors="replace")
-        yield frame
+        if frame.opcode == websocket.ABNF.OPCODE_PING:
+            connection.pong(frame.data)
+            continue
+        if frame.opcode not in (websocket.ABNF.OPCODE_TEXT,
+                                websocket.ABNF.OPCODE_BINARY):
+            continue
+        data = frame.data
+        if isinstance(data, bytes):
+            data = data.decode("utf-8", errors="replace")
+        if data == "":
+            continue
+        yield data

tests/integration/routes.json

@@ -227,6 +227,18 @@
     "key": "GET /api/namespaces/{}/applications/{}/pipelines/{}",
     "controller": "PipelineController"
   },
+  {
+    "method": "GET",
+    "path": "/api/namespaces/{namespace}/applications/{name}/pipelines/{id}/log",
+    "key": "GET /api/namespaces/{}/applications/{}/pipelines/{}/log",
+    "controller": "PipelineController"
+  },
+  {
+    "method": "GET",
+    "path": "/api/namespaces/{namespace}/applications/{name}/pipelines/{id}/steps/watch",
+    "key": "GET /api/namespaces/{}/applications/{}/pipelines/{}/steps/watch",
+    "controller": "PipelineController"
+  },
   {
     "method": "GET",
     "path": "/api/namespaces/{namespace}/applications/{name}/pods/{pod}/files",

tests/integration/test_operations.py

@@ -60,36 +60,23 @@ def test_pipelines_can_be_searched_for_an_application(client, namespace,
 
 @pytest.mark.cluster
 def test_the_status_stream_opens_and_sends_something(client, namespace,
-                                                     environment, endpoint,
-                                                     application):
+                                                     environment, application):
     """`/status/watch` is Server-Sent Events, not JSON.
 
-    The only endpoint in the API that answers with a stream rather than the
-    result envelope, which is exactly the sort of thing a reimplementation turns
-    back into a plain GET without noticing.
+    An endpoint that answers with a stream rather than the result envelope is
+    exactly the sort of thing a reimplementation turns back into a plain GET
+    without noticing. The client refuses anything but `text/event-stream`, so
+    reaching an event at all is the proof; one is enough, since the status
+    stream follows the workload for as long as the viewer stays.
     """
-    import requests
-
-    url = (f"{endpoint}/api/namespaces/{namespace}/applications/{application}"
-           f"/status/watch?environment={environment}")
-    with requests.get(url, headers={"Authorization": f"Bearer {client.token}"},
-                      stream=True, timeout=30) as response:
-        assert response.status_code == 200, (
-            f"the status stream answered HTTP {response.status_code}")
-        content_type = response.headers.get("Content-Type", "")
-        assert "text/event-stream" in content_type, (
-            f"the status stream is served as {content_type!r} rather than "
-            f"text/event-stream, so a browser EventSource would refuse it")
-
-        # One frame is enough: it proves the emitter is wired to something.
-        for line in response.iter_lines(decode_unicode=True):
-            if line:
-                break
-    # Recorded so the coverage check sees the route, which the raw request above
-    # bypasses.
-    from oops_client import record
-    record("GET", f"/api/namespaces/{namespace}/applications/{application}"
-                  f"/status/watch")
+    stream = client.sse(
+        f"/api/namespaces/{namespace}/applications/{application}"
+        f"/status/watch?environment={environment}", timeout=30)
+    try:
+        first = next(stream, None)
+    finally:
+        stream.close()
+    assert first is not None, "the status stream sent nothing"
 
 
 @pytest.mark.cluster