Fix bytes serialization

Co-authored-by: Copilot <copilot@github.com>

Author: codeSamuraii

pyfuse/worker/remote.py

@@ -572,7 +572,18 @@ async def _handle_task(
     # If the client cancelled mid-execution, it already stored a cancelled
     # result envelope (via Result.cancel) which the client reads first.
     await flush()
-    await backend.send_result(task.task_id, envelope.to_json())
+    try:
+        result_json = envelope.to_json()
+    except Exception as exc:
+        # Result serialization failed (e.g. unsupported return type).
+        # Surface the failure as an error envelope rather than letting
+        # the task hang forever from the client's point of view.
+        logger.exception(
+            "Failed to serialize result for task %s", task.task_id,
+        )
+        envelope = ResultEnvelope.failure(task.task_id, exc)
+        result_json = envelope.to_json()
+    await backend.send_result(task.task_id, result_json)
     await backend.notify_result(task.task_id)
 
     _log_task_result(task, envelope, elapsed_ms, worker)

pyfuse/worker/result.py

@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from collections.abc import Generator
 
+from pyfuse.core.task import _TaskEncoder, _resolve
 from pyfuse.core.errors import RemoteError, TaskStalled, TaskCancelled, ThrottleError
 from pyfuse.core.progress import ProgressInfo
 from pyfuse.worker.backends.base import Backend
@@ -58,7 +59,12 @@ def failure(cls, task_id: str, exc: BaseException) -> Self:
         )
 
     def to_json(self) -> str:
-        """Serialize to JSON string."""
+        """Serialize to JSON string.
+
+        Result payloads are encoded with the same sentinel-based encoder
+        used for task arguments, so ``bytes``, ``datetime``, ``Decimal``,
+        ``Path`` etc. round-trip transparently.
+        """
         d: dict[str, Any] = {
             "task_id": self.task_id,
             "status": self.status,
@@ -69,16 +75,17 @@ def to_json(self) -> str:
             d["error_type"] = self.error_type
             d["error_message"] = self.error_message
             d["error_traceback"] = self.error_traceback
-        return json.dumps(d)
+        return json.dumps(d, cls=_TaskEncoder)
 
     @classmethod
     def from_json(cls, raw: str | bytes) -> Self:
         """Deserialize from a JSON string or bytes."""
         data = json.loads(raw)
+        result = _resolve(data.get("result"), {}) if data.get("status") == "ok" else None
         return cls(
             task_id=data["task_id"],
             status=data["status"],
-            result=data.get("result"),
+            result=result,
             error_type=data.get("error_type"),
             error_message=data.get("error_message"),
             error_traceback=data.get("error_traceback"),