python: fix handling of base64 output in JobWatcher

Problem: If there is binary output, encoded in base64 in the job
output eventlog, the JobWatcher class and thus flux-watch(1) throws
errors about bytes not being encodable as JSON.

First, ensure that base64 encoded output is decoded to utf-8 with
surrogates, so that the data can be encoded back to binary without
data loss.

Then, in the JobWatcher class, ensure all output streams are capable
of handling the surrogates (and utf-8) by reconfiguring them (in
python 3.7+) or reopening them with the proper error handling.

Fixes #5702

Author: grondo

src/bindings/python/flux/job/output.py

@@ -113,7 +113,9 @@ def __init__(self, entry, labelio=False):
             data = self.context["data"]
             if "encoding" in self.context:
                 if self.context["encoding"] == "base64":
-                    data = base64.b64decode(data)
+                    data = base64.b64decode(data).decode(
+                        "utf-8", errors="surrogateescape"
+                    )
             if "repeat" in self.context:
                 data *= self.context["repeat"]
             if labelio:

src/bindings/python/flux/job/watcher.py

@@ -367,8 +367,8 @@ def __init__(
         self.t0 = starttime
         self.log_events = log_events
         self.log_status = log_status
-        self.stdout = stdout
-        self.stderr = stderr
+        self.stdout = self._reopen(stdout)
+        self.stderr = self._reopen(stderr)
         self.labelio = labelio
         self.exitcode = 0
         self.show_progress = progress
@@ -379,6 +379,22 @@ def __init__(
         if jobs:
             self.add_jobs(*jobs)
 
+    @staticmethod
+    def _reopen(stream):
+        """reconfigure/reopen stream with correct encoding and error handling"""
+        try:
+            # reconfigure() only available in >=3.7
+            stream.reconfigure(encoding="utf-8", errors="surrogateescape")
+            return stream
+        except AttributeError:
+            return open(
+                stream.fileno(),
+                mode="w",
+                encoding="utf-8",
+                errors="surrogateescape",
+                closefd=False,
+            )
+
     @staticmethod
     def _status_to_exitcode(status):
         """Calculate exitcode from job status"""
@@ -420,9 +436,9 @@ def add_jobs(self, *jobs, stdout=None, stderr=None, wait="clean"):
         self.progress.add_jobs(*jobs)
 
         if stdout is None:
-            stdout = self.stdout
+            stdout = self._reopen(self.stdout)
         if stderr is None:
-            stderr = self.stderr
+            stderr = self._reopen(self.stderr)
 
         for job in jobs:
             if not self.t0 or job.t_submit < self.t0: