fix(server): filter benign read-idle-reap WARN on streaming follows

src/main/scala/ccas/server/CcasServer.scala

@@ -58,9 +58,14 @@ object CcasServer extends ZIOAppDefault {
         JobRunner.live,
         JobScheduler.live,
         // Read-idle reaper for keep-alive connections (zio-http's `Server.Config.default` leaves `idleTimeout=None`),
-        // so a client that vanishes without FIN/RST can't pin a Netty channel + fd forever. Set above
-        // `JobLogStream.KeepAliveInterval` (20s) so a streaming `/api/jobs/{id}/logs` follow's keepalive resets the
-        // timer before it fires; only genuinely dead connections get reaped.
+        // so a client that vanishes without FIN/RST can't pin a Netty channel + fd forever. This installs a *read-only*
+        // Netty `ReadTimeoutHandler` (`ServerChannelInitializer`): it resets on inbound (client→server) reads only, never
+        // on writes. A streaming `/api/jobs/{id}/{logs,progress}` follow is write-only server→client once the GET is sent,
+        // so its `JobLogStream` keepalive ticks (outbound) can't reset this timer — every live follow is reaped on the 60s
+        // schedule and the follower transparently reconnects (#161). That reap's `ReadTimeoutException` surfaces as a
+        // "Fatal exception in Netty" WARN which is benign noise, filtered at the logger (`ProgressDisplay.isBenignReadIdleReap`);
+        // this reaper's real remaining job is bounding idle *non-streaming* keep-alive fds. zio-http exposes no per-route
+        // idleTimeout nor a server `SO_KEEPALIVE` option, so one global read-idle timeout is the only knob available.
         Server.defaultWith(_.binding(host, port).idleTimeout(60.seconds))
       )
     }

src/main/scala/ccas/utils/ProgressDisplay.scala

@@ -7,6 +7,7 @@ import java.time.{Instant, ZoneId}
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.regex.Pattern
 
+import io.netty.handler.timeout.ReadTimeoutException
 import zio.stream.SubscriptionRef
 import zio.{Cause, FiberId, FiberRef, FiberRefs, LogLevel, LogSpan, Ref, Runtime, Scope, Trace, UIO, Unsafe, URIO, URLayer, ZIO, ZLayer, ZLogger}
 
@@ -204,8 +205,13 @@ final class ProgressDisplay private[utils] (
       val threshold = context.getOrDefault(FiberRef.currentLogLevel)
       if (level >= threshold) {
         try {
-          val sink = context.getOrDefault(JobLogSink.currentSink)
-          logAboveBarsSync(sink, ProgressDisplay.format(level, message(), Instant.now(), cause, spans, annotations))
+          // Force the message thunk once, inside the try so a throwing thunk stays caught (see `testLoggerSwallowsThrow`),
+          // and reuse it for both the read-idle-reap filter and the formatter.
+          val msg = message()
+          if (!ProgressDisplay.isBenignReadIdleReap(level, msg, cause)) {
+            val sink = context.getOrDefault(JobLogSink.currentSink)
+            logAboveBarsSync(sink, ProgressDisplay.format(level, msg, Instant.now(), cause, spans, annotations))
+          }
         } catch { case t: Throwable => t.printStackTrace(err) }
       }
     }
@@ -460,6 +466,30 @@ object ProgressDisplay {
   private def formatTime(when: Instant): String =
     when.atZone(Zone).toLocalTime.format(TimeFormat)
 
+  /** zio-http's `ServerInboundHandler.exceptionCaught` logs every fatal channel exception under this exact message. */
+  private val FatalNettyExceptionMessage = "Fatal exception in Netty"
+
+  /** True for the benign WARN zio-http emits when the server's read-idle reaper closes a live streaming follow.
+    *
+    * `CcasServer` sets `Server.Config#idleTimeout`, which installs a read-only Netty `ReadTimeoutHandler`. A job-log /
+    * progress follow (`GET /api/jobs/{id}/{logs,progress}`) is write-only server→client once the request is sent, so the
+    * read timer is never reset and the handler reaps every live follow on schedule — surfacing as zio-http's
+    * `ZIO.logWarningCause("Fatal exception in Netty", Cause.die(ReadTimeoutException))`. The follow's client transparently
+    * reconnects (#161), so the reap is benign transport noise, not a failure, and we drop exactly this WARN.
+    *
+    * The match is deliberately over-specified — the exact zio-http message AND a cause that is *solely* a
+    * `ReadTimeoutException` defect (no typed failure, no other defect). This scopes it to the server reaper rather than
+    * any bare read-timeout WARN elsewhere, and makes both coordinates fail *open*: if a zio-http upgrade renames the
+    * message or wraps the exception, the filter simply stops matching and the (still benign) WARN reappears — noise
+    * returns, no genuine signal is ever hidden.
+    */
+  private def isBenignReadIdleReap(level: LogLevel, message: String, cause: Cause[Any]): Boolean =
+    level == LogLevel.Warning &&
+      message == FatalNettyExceptionMessage &&
+      cause.failures.isEmpty &&
+      cause.defects.nonEmpty &&
+      cause.defects.forall(_.isInstanceOf[ReadTimeoutException])
+
   private def format(
     level: LogLevel,
     msg: String,

src/test/scala/ccas/utils/TestProgressBar.scala

@@ -2,8 +2,9 @@ package ccas.utils
 
 import java.io.{ByteArrayOutputStream, PrintStream}
 
+import io.netty.handler.timeout.ReadTimeoutException
 import zio.stream.SubscriptionRef
-import zio.{LogLevel, Promise, Ref, ZIO}
+import zio.{Cause, LogLevel, Promise, Ref, ZIO}
 import zio.test.{assertCompletes, assertTrue, Spec, ZIOSpecDefault}
 
 object TestProgressBar extends ZIOSpecDefault {
@@ -25,7 +26,8 @@ object TestProgressBar extends ZIOSpecDefault {
     testCurrentLogLevelEnablesDebug,
     testLoggerSwallowsThrow,
     testSourcedRendersPrefixTag,
-    testStripSourceTag
+    testStripSourceTag,
+    testBenignReadIdleReapFiltered
   )
   // No `@@ TestAspect.sequential`: each test owns its capture buffers and installs loggers/sinks via fiber-scoped
   // FiberRefs (`currentLoggers`, `currentSink`), so the suite holds no process-global state to serialise. (Cross-suite
@@ -356,6 +358,31 @@ object TestProgressBar extends ZIOSpecDefault {
     )
   }
 
+  /** The server's read-idle reaper closes every live streaming follow on schedule (a read-only `ReadTimeoutHandler`
+    * can't be reset by the outbound keepalive), which zio-http logs as a benign `Cause.die(ReadTimeoutException)` WARN.
+    * The logger drops exactly that WARN (`ProgressDisplay.isBenignReadIdleReap`), but any other fatal-Netty WARN — a
+    * defect of a different type — must still be rendered. Both are emitted here; only the second should survive.
+    */
+  private def testBenignReadIdleReapFiltered =
+    test("read-idle-reap WARN is filtered while other fatal-Netty WARNs still log") {
+      val effect =
+        ZIO.logWarningCause("Fatal exception in Netty", Cause.die(ReadTimeoutException.INSTANCE)) *>
+          ZIO.logWarningCause("genuine netty fault", Cause.die(new RuntimeException("kaboom"))) *>
+          ZIO.logWarningCause("read timeout we DO want", Cause.die(ReadTimeoutException.INSTANCE))
+      withLogCapture(effect).map { case (_, out) =>
+        val clean = stripAnsi(out)
+        assertTrue(
+          // The exact zio-http server-reaper WARN (its message + a bare ReadTimeoutException) is dropped...
+          !clean.contains("Fatal exception in Netty"),
+          // ...but a different fatal-Netty WARN survives...
+          clean.contains("genuine netty fault"),
+          clean.contains("kaboom"),
+          // ...and so does a bare ReadTimeoutException under a different message — the filter is scoped, not blanket.
+          clean.contains("read timeout we DO want")
+        )
+      }
+    }
+
   private def testLoggerSwallowsThrow = test("asLogger swallows exceptions from a throwing message thunk") {
     ZIO.suspendSucceed {
       val outBaos = new ByteArrayOutputStream