fix(docker,flows): surface truncated/systemerr exec streams, bound ftester drain, tidy failure logging

- demuxExecStdout errors on a header cut short mid-frame (was a silent EOF that
  dropped the tail) and surfaces a docker systemerr frame instead of discarding it.
- ftester drains telemetry on exit through the bounded observer.Drain instead of
  two unbounded ForceFlush calls, so an unreachable collector can't hang it at exit.
- container-listing failures are sorted for a deterministic skipped-entries preview,
  logged per-entry at Debug (the detail is already in the response and the endpoint
  is hit on every navigation) with names quoted so control bytes in a hostile
  filename can't inject into a log line; corrected the stat-failure comment
  (a dangling symlink lstats fine, it doesn't fail).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/cmd/ftester/main.go

@@ -63,21 +63,11 @@ func main() {
 	if err != nil && !errors.Is(err, obs.ErrNotConfigured) {
 		log.Fatalf("Unable to create langfuse client: %v\n", err)
 	}
-	defer func() {
-		if lfclient != nil {
-			lfclient.ForceFlush(context.Background())
-		}
-	}()
 
 	otelclient, err := obs.NewTelemetryClient(ctx, cfg)
 	if err != nil && !errors.Is(err, obs.ErrNotConfigured) {
 		log.Fatalf("Unable to create telemetry client: %v\n", err)
 	}
-	defer func() {
-		if otelclient != nil {
-			otelclient.ForceFlush(context.Background())
-		}
-	}()
 
 	obs.InitObserver(ctx, lfclient, otelclient, []logrus.Level{
 		logrus.DebugLevel,
@@ -85,6 +75,13 @@ func main() {
 		logrus.WarnLevel,
 		logrus.ErrorLevel,
 	})
+	// Drain telemetry on exit, bounded — an unreachable collector must not hang
+	// the tester on the SDK's per-provider timeouts.
+	defer func() {
+		drainCtx, cancelDrain := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancelDrain()
+		_ = obs.Observer.Drain(drainCtx)
+	}()
 
 	// Initialize database connection
 	db, err := sql.Open("postgres", cfg.DatabaseURL)

backend/cmd/pentagi/main.go

@@ -57,8 +57,8 @@ func main() {
 		logrus.SetLevel(logrus.InfoLevel)
 	}
 
-	// Telemetry is optional: degrade to a no-op observer on init failure instead
-	// of killing the process, so an unreachable collector can't take the app down.
+	// Telemetry is optional — degrade to a no-op observer on init failure so an
+	// unreachable collector can't take the app down.
 	lfclient, err := obs.NewLangfuseClient(ctx, cfg)
 	if err != nil && !errors.Is(err, obs.ErrNotConfigured) {
 		logrus.WithError(err).Warn("langfuse telemetry disabled: client init failed")

backend/pkg/docker/client.go

@@ -705,24 +705,34 @@ func demuxExecStdout(r io.Reader, maxStdout int) ([]byte, error) {
 	header := make([]byte, 8)
 	for {
 		if _, err := io.ReadFull(r, header); err != nil {
-			if err == io.EOF || err == io.ErrUnexpectedEOF {
-				break
+			if err == io.EOF {
+				break // clean end at a frame boundary
 			}
-			return nil, err
+			// A header cut short (ErrUnexpectedEOF) means the stream was truncated
+			// mid-frame — the listing is incomplete, so fail rather than silently
+			// dropping the tail.
+			return nil, fmt.Errorf("truncated exec stream: %w", err)
 		}
 		size := int64(binary.BigEndian.Uint32(header[4:8]))
 		if size == 0 {
 			continue
 		}
-		sink := io.Writer(io.Discard)
-		if header[0] == 1 { // stdout
-			sink = &stdout
-		}
-		if _, err := io.CopyN(sink, r, size); err != nil {
-			return nil, err
-		}
-		if stdout.Len() > maxStdout {
-			return nil, fmt.Errorf("listing output exceeded %d bytes", maxStdout)
+		switch header[0] {
+		case 1: // stdout
+			if _, err := io.CopyN(&stdout, r, size); err != nil {
+				return nil, err
+			}
+			if stdout.Len() > maxStdout {
+				return nil, fmt.Errorf("listing output exceeded %d bytes", maxStdout)
+			}
+		case 3: // systemerr — a daemon-level error injected mid-stream; surface it
+			var msg bytes.Buffer
+			_, _ = io.CopyN(&msg, r, size)
+			return nil, fmt.Errorf("docker exec systemerr: %s", strings.TrimSpace(msg.String()))
+		default: // stderr and anything else — discard
+			if _, err := io.CopyN(io.Discard, r, size); err != nil {
+				return nil, err
+			}
 		}
 	}
 	return stdout.Bytes(), nil

backend/pkg/docker/client_listing_test.go

@@ -53,3 +53,17 @@ func TestDemuxExecStdout_StdoutOnly_AndByteCap(t *testing.T) {
 		t.Fatalf("want cap error, got %v", err)
 	}
 }
+
+func TestDemuxExecStdout_TruncatedAndSystemerr(t *testing.T) {
+	// a header cut short mid-frame must error, not silently drop the tail
+	torn := append(listingFrame(1, "a.txt\x00"), 0x01, 0x00, 0x00) // 3 stray header bytes
+	if _, err := demuxExecStdout(bytes.NewReader(torn), 1<<20); err == nil || !strings.Contains(err.Error(), "truncated") {
+		t.Fatalf("want truncated-stream error, got %v", err)
+	}
+
+	// a systemerr (stream id 3) daemon error must surface, not be discarded
+	sys := append(listingFrame(1, "ok"), listingFrame(3, "daemon connection reset")...)
+	if _, err := demuxExecStdout(bytes.NewReader(sys), 1<<20); err == nil || !strings.Contains(err.Error(), "systemerr") {
+		t.Fatalf("want systemerr, got %v", err)
+	}
+}

backend/pkg/server/services/flow_files.go

@@ -1108,6 +1108,15 @@ func (s *FlowFileService) GetFlowContainerFiles(c *gin.Context) {
 		return allFiles[i].ModifiedAt.Before(allFiles[j].ModifiedAt)
 	})
 
+	// Failures are sorted too so the dialog's first-N skipped-entries preview is
+	// deterministic rather than in raw find/readdir order.
+	sort.Slice(allFailures, func(i, j int) bool {
+		if allFailures[i].Path != allFailures[j].Path {
+			return allFailures[i].Path < allFailures[j].Path
+		}
+		return allFailures[i].Name < allFailures[j].Name
+	})
+
 	// Backward-compat: when exactly one container path was queried, echo it back
 	// as Path so existing callers receive the same field value as before.
 	responsePath := ""
@@ -1115,27 +1124,22 @@ func (s *FlowFileService) GetFlowContainerFiles(c *gin.Context) {
 		responsePath = containerPaths[0]
 	}
 
-	// A per-entry stat failure (dangling symlink, a file removed mid-listing,
-	// transient /proc entries) degrades the listing but must not blank it: the
-	// readable files are returned and each skipped entry is logged and carried
-	// back in Failures so an operator can tell what and how much was skipped.
+	// A per-entry stat failure (a file removed mid-listing, a transient /proc pid,
+	// a symlink loop) degrades the listing but must not blank it: the readable
+	// files are returned and each skipped entry is carried back in Failures so the
+	// caller can tell what and how much was skipped.
 	if len(allFailures) > 0 {
 		log := logger.FromContext(c)
-		const maxLoggedFailures = 20
-		for i, fe := range allFailures {
-			if i >= maxLoggedFailures {
-				log.WithFields(map[string]any{
-					"flow_id":    flowID,
-					"suppressed": len(allFailures) - maxLoggedFailures,
-				}).Warn("additional container entries skipped (log capped)")
-				break
-			}
+		for _, fe := range allFailures {
+			// Detail is already in the Failures response, so log per-entry at Debug,
+			// not Warn — this endpoint is hit on every navigation. Quote the names so
+			// control bytes in a hostile filename can't inject into the log line.
 			log.WithFields(map[string]any{
 				"flow_id":    flowID,
-				"entry":      fe.Name,
-				"entry_path": fe.Path,
+				"entry":      strconv.Quote(fe.Name),
+				"entry_path": strconv.Quote(fe.Path),
 				"error":      rawFailureMessages[fe.Path],
-			}).Warn("container entry skipped: could not stat")
+			}).Debug("container entry skipped: could not stat")
 		}
 		log.WithFields(map[string]any{
 			"flow_id": flowID,