From bcbdbcf661b0558b6930ab61ed60dacbf5a61572 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Mon, 6 Jul 2026 00:04:15 +0530
Subject: [PATCH 1/5] feat(python): add payload codecs and codec serializer

---
 sdks/python/taskito/__init__.py       |  14 +++
 sdks/python/taskito/codecs.py         | 174 ++++++++++++++++++++++++++
 sdks/python/taskito/exceptions.py     |   4 +
 sdks/python/tests/core/test_codecs.py | 162 ++++++++++++++++++++++++
 4 files changed, 354 insertions(+)
 create mode 100644 sdks/python/taskito/codecs.py
 create mode 100644 sdks/python/tests/core/test_codecs.py

diff --git a/sdks/python/taskito/__init__.py b/sdks/python/taskito/__init__.py
index d4de1916..cd77151e 100644
--- a/sdks/python/taskito/__init__.py
+++ b/sdks/python/taskito/__init__.py
@@ -7,11 +7,19 @@
     BatchResultTypeError,
 )
 from taskito.canvas import Signature, chain, chord, chunks, group, starmap
+from taskito.codecs import (
+    AesGcmCodec,
+    CodecSerializer,
+    GzipCodec,
+    HmacCodec,
+    PayloadCodec,
+)
 from taskito.context import LogLevel, current_job
 from taskito.events import EventType
 from taskito.exceptions import (
     CircuitBreakerOpenError,
     CircularDependencyError,
+    CryptoError,
     JobNotFoundError,
     MaxRetriesExceededError,
     NotesValidationError,
@@ -53,14 +61,19 @@
 
 __all__ = [
     "MAX_NOTE_FIELDS",
+    "AesGcmCodec",
     "BatchItemResult",
     "BatchPartialFailureError",
     "BatchResultTypeError",
     "CircuitBreakerOpenError",
     "CircularDependencyError",
     "CloudpickleSerializer",
+    "CodecSerializer",
+    "CryptoError",
     "EncryptedSerializer",
     "EventType",
+    "GzipCodec",
+    "HmacCodec",
     "Inject",
     "InterceptionError",
     "InterceptionReport",
@@ -74,6 +87,7 @@
     "MsgPackSerializer",
     "NoProxy",
     "NotesValidationError",
+    "PayloadCodec",
     "PredicateRejectedError",
     "ProxyCleanupError",
     "ProxyReconstructionError",
diff --git a/sdks/python/taskito/codecs.py b/sdks/python/taskito/codecs.py
new file mode 100644
index 00000000..c2bbdb95
--- /dev/null
+++ b/sdks/python/taskito/codecs.py
@@ -0,0 +1,174 @@
+"""Payload codecs: reversible byte transforms layered over a serializer.
+
+A :class:`PayloadCodec` transforms serialized bytes on the way out (producer)
+and reverses the transform on the way in (worker) — compression, encryption,
+signing. Codecs compose: a chain encodes in list order and decodes in reverse,
+so ``[gzip, hmac]`` verifies integrity *before* decompressing.
+
+Wire formats match the Java and Node SDKs byte-for-byte, so codec-framed
+payloads are cross-SDK compatible.
+
+Enable globally via ``Queue(codec=...)`` (wraps the queue serializer, covers
+payloads and results) or per task via ``Queue(codecs={"name": codec})`` +
+``@queue.task(codecs=["name"])`` (payload only; results stay on the queue
+serializer). Jobs persisted before a codec chain is enabled cannot be decoded
+through it — drain the queue before turning codecs on.
+"""
+
+from __future__ import annotations
+
+import gzip
+import hashlib
+import hmac
+import os
+import zlib
+from collections.abc import Sequence
+from typing import Any, Protocol, runtime_checkable
+
+from .exceptions import CryptoError, SerializationError
+from .serializers import Serializer
+
+_GZIP_WBITS = 16 + zlib.MAX_WBITS  # gzip framing for zlib.decompressobj
+_HMAC_DIGEST_SIZE = 32  # SHA-256 output length in bytes
+_AES_NONCE_SIZE = 12  # GCM standard nonce length in bytes
+
+
+@runtime_checkable
+class PayloadCodec(Protocol):
+    """Protocol for reversible payload byte transforms."""
+
+    def encode(self, data: bytes) -> bytes:
+        """Transform serialized bytes on the producer (compress, encrypt, sign)."""
+        ...
+
+    def decode(self, data: bytes) -> bytes:
+        """Reverse the transform on the worker."""
+        ...
+
+
+class GzipCodec:
+    """Gzip compression codec.
+
+    Decompression is capped at ``max_decompressed_bytes`` (default 64 MiB) so a
+    malicious or corrupt payload cannot expand into a zip bomb.
+    """
+
+    _DEFAULT_MAX_DECOMPRESSED_BYTES = 64 * 1024 * 1024
+
+    def __init__(self, max_decompressed_bytes: int = _DEFAULT_MAX_DECOMPRESSED_BYTES):
+        if max_decompressed_bytes <= 0:
+            raise ValueError(
+                f"max_decompressed_bytes must be positive, got {max_decompressed_bytes}"
+            )
+        self._max_decompressed_bytes = max_decompressed_bytes
+
+    def encode(self, data: bytes) -> bytes:
+        return gzip.compress(data)
+
+    def decode(self, data: bytes) -> bytes:
+        # stdlib gzip.decompress has no output bound; stream through a
+        # decompressobj with max_length so the cap holds mid-stream.
+        decompressor = zlib.decompressobj(wbits=_GZIP_WBITS)
+        try:
+            output = decompressor.decompress(data, self._max_decompressed_bytes)
+            if decompressor.unconsumed_tail:
+                raise SerializationError(
+                    f"decompressed payload exceeds the {self._max_decompressed_bytes}-byte limit"
+                )
+            output += decompressor.flush()
+        except zlib.error as exc:
+            raise SerializationError(f"gzip decompression failed: {exc}") from exc
+        if len(output) > self._max_decompressed_bytes:
+            raise SerializationError(
+                f"decompressed payload exceeds the {self._max_decompressed_bytes}-byte limit"
+            )
+        return output
+
+
+class AesGcmCodec:
+    """AES-GCM encryption codec.
+
+    Wire format: ``[12-byte random nonce][ciphertext || 16-byte GCM tag]`` —
+    identical to the Java and Node SDK codecs. Requires the ``encryption``
+    extra (``pip install taskito[encryption]``). Key must be 16, 24, or 32
+    bytes (AES-128/192/256).
+    """
+
+    def __init__(self, key: bytes):
+        if not isinstance(key, bytes):
+            raise TypeError(f"key must be bytes, got {type(key).__name__}")
+        if len(key) not in (16, 24, 32):
+            raise ValueError(
+                f"key must be 16, 24, or 32 bytes for AES-128/192/256, got {len(key)} bytes"
+            )
+
+        from cryptography.exceptions import InvalidTag
+        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+        self._aesgcm = AESGCM(key)
+        # Cache the exception class so ``decode`` doesn't re-import per call.
+        self._invalid_tag = InvalidTag
+
+    def encode(self, data: bytes) -> bytes:
+        nonce = os.urandom(_AES_NONCE_SIZE)
+        return bytes(nonce + self._aesgcm.encrypt(nonce, data, None))
+
+    def decode(self, data: bytes) -> bytes:
+        if len(data) < _AES_NONCE_SIZE:
+            raise CryptoError("encrypted payload is too short")
+        nonce, ciphertext = data[:_AES_NONCE_SIZE], data[_AES_NONCE_SIZE:]
+        try:
+            return bytes(self._aesgcm.decrypt(nonce, ciphertext, None))
+        except self._invalid_tag as exc:
+            raise CryptoError("decryption failed") from exc
+
+
+class HmacCodec:
+    """HMAC-SHA256 signing codec (authenticates, does not encrypt).
+
+    Wire format: ``[32-byte mac][body]`` — identical to the Java and Node SDK
+    codecs.
+    """
+
+    def __init__(self, key: bytes):
+        if not isinstance(key, bytes):
+            raise TypeError(f"key must be bytes, got {type(key).__name__}")
+        self._key = key
+
+    def encode(self, data: bytes) -> bytes:
+        mac = hmac.new(self._key, data, hashlib.sha256).digest()
+        return mac + data
+
+    def decode(self, data: bytes) -> bytes:
+        if len(data) < _HMAC_DIGEST_SIZE:
+            raise CryptoError("signed payload is too short")
+        mac, body = data[:_HMAC_DIGEST_SIZE], data[_HMAC_DIGEST_SIZE:]
+        expected = hmac.new(self._key, body, hashlib.sha256).digest()
+        if not hmac.compare_digest(mac, expected):
+            raise CryptoError("signature mismatch")
+        return body
+
+
+class CodecSerializer:
+    """Serializer decorator applying a codec chain around a delegate.
+
+    ``dumps`` serializes with the delegate then encodes codecs in list order;
+    ``loads`` decodes in reverse order then deserializes. Used internally when
+    ``Queue(codec=...)`` is set — the chain then covers every payload and
+    result flowing through the queue serializer.
+    """
+
+    def __init__(self, delegate: Serializer, codecs: Sequence[PayloadCodec]):
+        self._delegate = delegate
+        self._codecs = list(codecs)
+
+    def dumps(self, obj: Any) -> bytes:
+        data = self._delegate.dumps(obj)
+        for codec in self._codecs:
+            data = codec.encode(data)
+        return data
+
+    def loads(self, data: bytes) -> Any:
+        for codec in reversed(self._codecs):
+            data = codec.decode(data)
+        return self._delegate.loads(data)
diff --git a/sdks/python/taskito/exceptions.py b/sdks/python/taskito/exceptions.py
index b8d03eb2..bc386aad 100644
--- a/sdks/python/taskito/exceptions.py
+++ b/sdks/python/taskito/exceptions.py
@@ -29,6 +29,10 @@ class SerializationError(TaskitoError):
     """Raised on serialization or deserialization failures."""
 
 
+class CryptoError(SerializationError):
+    """Raised when a payload codec fails to decrypt or verify a payload."""
+
+
 class CircuitBreakerOpenError(TaskitoError):
     """Raised when a task's circuit breaker is open."""
 
diff --git a/sdks/python/tests/core/test_codecs.py b/sdks/python/tests/core/test_codecs.py
new file mode 100644
index 00000000..b452c943
--- /dev/null
+++ b/sdks/python/tests/core/test_codecs.py
@@ -0,0 +1,162 @@
+"""Unit tests for payload codecs and the codec serializer."""
+
+import gzip
+
+import pytest
+
+from taskito import (
+    AesGcmCodec,
+    CodecSerializer,
+    CryptoError,
+    GzipCodec,
+    HmacCodec,
+    JsonSerializer,
+    PayloadCodec,
+    SerializationError,
+    Serializer,
+    SmartSerializer,
+)
+
+HMAC_KEY = b"codec-hmac-secret"
+AES_KEY = b"0123456789abcdef0123456789abcdef"  # 32 bytes -> AES-256
+
+
+def make_aes_codec() -> AesGcmCodec:
+    pytest.importorskip("cryptography")
+    return AesGcmCodec(AES_KEY)
+
+
+class TestGzipCodec:
+    def test_round_trip(self) -> None:
+        codec = GzipCodec()
+        data = b"hello codec world" * 100
+        encoded = codec.encode(data)
+        assert encoded != data
+        assert len(encoded) < len(data)
+        assert codec.decode(encoded) == data
+
+    def test_round_trip_empty(self) -> None:
+        codec = GzipCodec()
+        assert codec.decode(codec.encode(b"")) == b""
+
+    def test_decode_rejects_payload_exceeding_cap(self) -> None:
+        big = b"a" * 1024
+        encoded = GzipCodec().encode(big)
+        with pytest.raises(SerializationError, match="exceeds"):
+            GzipCodec(max_decompressed_bytes=16).decode(encoded)
+
+    def test_decode_at_exact_cap_succeeds(self) -> None:
+        data = b"a" * 64
+        encoded = GzipCodec().encode(data)
+        assert GzipCodec(max_decompressed_bytes=64).decode(encoded) == data
+
+    def test_decode_rejects_corrupt_stream(self) -> None:
+        with pytest.raises(SerializationError, match="decompression failed"):
+            GzipCodec().decode(b"not gzip data")
+
+    def test_constructor_rejects_non_positive_cap(self) -> None:
+        with pytest.raises(ValueError, match="positive"):
+            GzipCodec(max_decompressed_bytes=0)
+        with pytest.raises(ValueError, match="positive"):
+            GzipCodec(max_decompressed_bytes=-1)
+
+
+class TestAesGcmCodec:
+    def test_round_trip_and_hides_plaintext(self) -> None:
+        codec = make_aes_codec()
+        data = b"secret payload"
+        encoded = codec.encode(data)
+        assert data not in encoded
+        assert codec.decode(encoded) == data
+
+    def test_wire_format_nonce_prefix(self) -> None:
+        # [12-byte nonce][ciphertext || 16-byte tag] — cross-SDK contract.
+        codec = make_aes_codec()
+        data = b"x" * 10
+        encoded = codec.encode(data)
+        assert len(encoded) == 12 + len(data) + 16
+
+    def test_decode_rejects_tampered_payload(self) -> None:
+        codec = make_aes_codec()
+        encoded = bytearray(codec.encode(b"secret payload"))
+        encoded[-1] ^= 0xFF
+        with pytest.raises(CryptoError, match="decryption failed"):
+            codec.decode(bytes(encoded))
+
+    def test_decode_rejects_short_payload(self) -> None:
+        codec = make_aes_codec()
+        with pytest.raises(CryptoError, match="too short"):
+            codec.decode(b"short")
+
+    def test_constructor_rejects_bad_key_length(self) -> None:
+        with pytest.raises(ValueError, match="16, 24, or 32"):
+            AesGcmCodec(b"short-key")
+
+    def test_constructor_rejects_non_bytes_key(self) -> None:
+        with pytest.raises(TypeError, match="bytes"):
+            AesGcmCodec("string-key")  # type: ignore[arg-type]
+
+
+class TestHmacCodec:
+    def test_round_trip(self) -> None:
+        codec = HmacCodec(HMAC_KEY)
+        data = b"authenticated payload"
+        encoded = codec.encode(data)
+        assert encoded[32:] == data  # [32-byte mac][body] — cross-SDK contract
+        assert codec.decode(encoded) == data
+
+    def test_decode_rejects_tampered_payload(self) -> None:
+        codec = HmacCodec(HMAC_KEY)
+        encoded = bytearray(codec.encode(b"authenticated payload"))
+        encoded[-1] ^= 0xFF
+        with pytest.raises(CryptoError, match="signature mismatch"):
+            codec.decode(bytes(encoded))
+
+    def test_decode_rejects_wrong_key(self) -> None:
+        encoded = HmacCodec(HMAC_KEY).encode(b"payload")
+        with pytest.raises(CryptoError, match="signature mismatch"):
+            HmacCodec(b"different-key").decode(encoded)
+
+    def test_decode_rejects_short_payload(self) -> None:
+        with pytest.raises(CryptoError, match="too short"):
+            HmacCodec(HMAC_KEY).decode(b"short")
+
+    def test_constructor_rejects_non_bytes_key(self) -> None:
+        with pytest.raises(TypeError, match="bytes"):
+            HmacCodec("string-key")  # type: ignore[arg-type]
+
+
+class TestPayloadCodecProtocol:
+    def test_built_in_codecs_satisfy_protocol(self) -> None:
+        assert isinstance(GzipCodec(), PayloadCodec)
+        assert isinstance(HmacCodec(HMAC_KEY), PayloadCodec)
+
+
+class TestCodecSerializer:
+    @pytest.mark.parametrize("delegate", [SmartSerializer(), JsonSerializer()])
+    def test_chain_is_reversible_in_reverse_order(self, delegate: Serializer) -> None:
+        pytest.importorskip("cryptography")
+        chain: list[PayloadCodec] = [GzipCodec(), make_aes_codec(), HmacCodec(HMAC_KEY)]
+        serializer = CodecSerializer(delegate, chain)
+        obj = {"numbers": [1, 2, 3], "text": "hello" * 50}
+        assert serializer.loads(serializer.dumps(obj)) == obj
+
+    def test_encoded_bytes_are_codec_framed(self) -> None:
+        serializer = CodecSerializer(SmartSerializer(), [GzipCodec()])
+        encoded = serializer.dumps({"key": "value"})
+        assert encoded[:2] == b"\x1f\x8b"  # gzip magic
+        assert gzip.decompress(encoded) == SmartSerializer().dumps({"key": "value"})
+
+    def test_tamper_detected_before_decompression(self) -> None:
+        # decode runs in reverse: HMAC verifies before gzip touches the bytes.
+        serializer = CodecSerializer(SmartSerializer(), [GzipCodec(), HmacCodec(HMAC_KEY)])
+        encoded = bytearray(serializer.dumps("payload"))
+        encoded[-1] ^= 0xFF
+        with pytest.raises(CryptoError, match="signature mismatch"):
+            serializer.loads(bytes(encoded))
+
+    def test_empty_chain_is_transparent(self) -> None:
+        serializer = CodecSerializer(SmartSerializer(), [])
+        obj = (1, "two", [3])
+        assert serializer.loads(serializer.dumps(obj)) == obj
+        assert serializer.dumps(obj) == SmartSerializer().dumps(obj)

From 477b212a1eaa293e76764473f30d07c9a36909c3 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Mon, 6 Jul 2026 00:20:04 +0530
Subject: [PATCH 2/5] fix(python): write results through the queue serializer

---
 .../src/native_async/task_executor.rs         | 11 +++-
 crates/taskito-python/src/py_worker.rs        | 12 +++-
 sdks/python/taskito/app.py                    | 10 +++
 sdks/python/taskito/async_support/executor.py |  6 +-
 sdks/python/taskito/prefork/child.py          |  2 +-
 .../taskito/workflows/saga/orchestrator.py    |  2 +-
 .../tests/core/test_result_serialization.py   | 64 +++++++++++++++++++
 sdks/python/tests/worker/test_native_async.py | 19 ++++--
 8 files changed, 111 insertions(+), 15 deletions(-)
 create mode 100644 sdks/python/tests/core/test_result_serialization.py

diff --git a/crates/taskito-python/src/native_async/task_executor.rs b/crates/taskito-python/src/native_async/task_executor.rs
index ea955cf4..e57cb8a7 100644
--- a/crates/taskito-python/src/native_async/task_executor.rs
+++ b/crates/taskito-python/src/native_async/task_executor.rs
@@ -149,11 +149,18 @@ fn run_task(py: Python<'_>, task_registry: &Py<PyAny>, job: &Job) -> PyResult<Op
     let _ = context_mod.call_method0("_clear_context");
     let result = result?;
 
+    // Serialize result using the queue-level serializer (with any codec
+    // chain); fall back to raw cloudpickle when no queue is registered.
     if result.is_none() {
         Ok(None)
     } else {
-        let pickled = cloudpickle.call_method1("dumps", (result,))?;
-        let bytes: Vec<u8> = pickled.extract()?;
+        let queue_ref = context_mod.getattr("_queue_ref")?;
+        let serialized = if !queue_ref.is_none() {
+            queue_ref.call_method1("_serialize_result", (&job.task_name, result))?
+        } else {
+            cloudpickle.call_method1("dumps", (result,))?
+        };
+        let bytes: Vec<u8> = serialized.extract()?;
         Ok(Some(bytes))
     }
 }
diff --git a/crates/taskito-python/src/py_worker.rs b/crates/taskito-python/src/py_worker.rs
index 3047433e..37fde72d 100644
--- a/crates/taskito-python/src/py_worker.rs
+++ b/crates/taskito-python/src/py_worker.rs
@@ -207,12 +207,18 @@ pub fn execute_task(
 
     let result = result?;
 
-    // Serialize result
+    // Serialize result using the queue-level serializer (with any codec
+    // chain); fall back to raw cloudpickle when no queue is registered.
     if result.is_none() {
         Ok(None)
     } else {
-        let pickled = cloudpickle.call_method1("dumps", (result,))?;
-        let bytes: Vec<u8> = pickled.extract()?;
+        let queue_ref = context_mod.getattr("_queue_ref")?;
+        let serialized = if !queue_ref.is_none() {
+            queue_ref.call_method1("_serialize_result", (&job.task_name, result))?
+        } else {
+            cloudpickle.call_method1("dumps", (result,))?
+        };
+        let bytes: Vec<u8> = serialized.extract()?;
         Ok(Some(bytes))
     }
 }
diff --git a/sdks/python/taskito/app.py b/sdks/python/taskito/app.py
index dc9309d4..912a1f36 100644
--- a/sdks/python/taskito/app.py
+++ b/sdks/python/taskito/app.py
@@ -440,6 +440,16 @@ def _deserialize_payload(self, task_name: str, payload: bytes) -> tuple:
         args, kwargs = self._get_serializer(task_name).loads(payload)
         return tuple(args), kwargs
 
+    def _serialize_result(self, task_name: str, result: Any) -> bytes:
+        """Serialize a task result with the queue-level serializer.
+
+        Results always use the queue serializer, never per-task serializers:
+        consumers (``JobResult.get``, workflow trackers) read results back
+        with only the queue-level configuration. ``task_name`` is part of the
+        contract with the Rust worker paths, which call this per job.
+        """
+        return self._serializer.dumps(result)
+
     def enqueue(
         self,
         task_name: str,
diff --git a/sdks/python/taskito/async_support/executor.py b/sdks/python/taskito/async_support/executor.py
index b2267baf..5685ead4 100644
--- a/sdks/python/taskito/async_support/executor.py
+++ b/sdks/python/taskito/async_support/executor.py
@@ -9,8 +9,6 @@
 import traceback
 from typing import TYPE_CHECKING, Any
 
-import cloudpickle
-
 from taskito.async_support.context import clear_async_context, set_async_context
 from taskito.context import current_job
 from taskito.exceptions import TaskCancelledError
@@ -159,7 +157,9 @@ async def _execute(
                 result = await fn(*args, **kwargs)
 
                 # Serialize and report success
-                result_bytes = cloudpickle.dumps(result) if result is not None else None
+                result_bytes = (
+                    queue._serialize_result(task_name, result) if result is not None else None
+                )
                 wall_ns = time.monotonic_ns() - start_ns
                 self._sender.report_success(job_id, task_name, result_bytes, wall_ns)
 
diff --git a/sdks/python/taskito/prefork/child.py b/sdks/python/taskito/prefork/child.py
index 1bcee847..d4331741 100644
--- a/sdks/python/taskito/prefork/child.py
+++ b/sdks/python/taskito/prefork/child.py
@@ -119,7 +119,7 @@ def _execute_job(
     try:
         args, kwargs = queue._deserialize_payload(task_name, payload)
         result = run_maybe_async(wrapper(*args, **kwargs))
-        result_bytes = queue._serializer.dumps(result) if result is not None else None
+        result_bytes = queue._serialize_result(task_name, result) if result is not None else None
         wall_time_ns = time.monotonic_ns() - start_ns
 
         return {
diff --git a/sdks/python/taskito/workflows/saga/orchestrator.py b/sdks/python/taskito/workflows/saga/orchestrator.py
index 3872d7f7..3c686c79 100644
--- a/sdks/python/taskito/workflows/saga/orchestrator.py
+++ b/sdks/python/taskito/workflows/saga/orchestrator.py
@@ -764,7 +764,7 @@ def _load_forward_payload(
         result_bytes = forward_job.result_bytes
         if result_bytes:
             try:
-                forward_result = self._queue._get_serializer(task_name).loads(result_bytes)
+                forward_result = self._queue._serializer.loads(result_bytes)
             except Exception:
                 logger.exception(
                     "saga: failed to deserialize forward result for %s/%s — "
diff --git a/sdks/python/tests/core/test_result_serialization.py b/sdks/python/tests/core/test_result_serialization.py
new file mode 100644
index 00000000..dc2eb765
--- /dev/null
+++ b/sdks/python/tests/core/test_result_serialization.py
@@ -0,0 +1,64 @@
+"""Regression tests: results are written with the queue serializer.
+
+Before the fix, worker paths wrote results as raw cloudpickle regardless of
+the configured serializer, while ``JobResult.get()`` read them back with the
+queue serializer — silently broken for any non-default serializer.
+"""
+
+import json
+import threading
+from collections.abc import Generator
+from pathlib import Path
+
+import cloudpickle
+import pytest
+
+from taskito import JsonSerializer, Queue, SmartSerializer
+
+
+@pytest.fixture
+def json_queue(tmp_path: Path) -> Queue:
+    return Queue(db_path=str(tmp_path / "json.db"), workers=1, serializer=JsonSerializer())
+
+
+@pytest.fixture
+def run_json_worker(json_queue: Queue) -> Generator[threading.Thread]:
+    thread = threading.Thread(target=json_queue.run_worker, daemon=True)
+    thread.start()
+    yield thread
+    json_queue._inner.request_shutdown()
+    thread.join(timeout=5)
+
+
+def test_result_round_trips_with_json_serializer(
+    json_queue: Queue, run_json_worker: threading.Thread
+) -> None:
+    @json_queue.task()
+    def build_report(name: str) -> dict:
+        return {"report": name, "rows": [1, 2, 3]}
+
+    result = build_report.delay("weekly")
+    assert result.result(timeout=10) == {"report": "weekly", "rows": [1, 2, 3]}
+
+
+def test_result_bytes_use_queue_serializer(
+    json_queue: Queue, run_json_worker: threading.Thread
+) -> None:
+    @json_queue.task()
+    def echo(value: str) -> str:
+        return value
+
+    result = echo.delay("hello")
+    result.result(timeout=10)
+
+    job = json_queue._inner.get_job(result.id)
+    assert job is not None and job.result_bytes is not None
+    # Valid JSON proves the queue serializer produced the bytes, not cloudpickle.
+    assert json.loads(job.result_bytes.decode("utf-8")) == "hello"
+
+
+def test_smart_serializer_reads_legacy_cloudpickle_results() -> None:
+    # Results persisted before the fix are raw cloudpickle; the default
+    # serializer's untagged-bytes fallback keeps them readable after upgrade.
+    legacy = cloudpickle.dumps({"legacy": True})
+    assert SmartSerializer().loads(legacy) == {"legacy": True}
diff --git a/sdks/python/tests/worker/test_native_async.py b/sdks/python/tests/worker/test_native_async.py
index c90ca8ab..e4e1a211 100644
--- a/sdks/python/tests/worker/test_native_async.py
+++ b/sdks/python/tests/worker/test_native_async.py
@@ -152,6 +152,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -195,6 +196,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -237,6 +239,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -276,6 +279,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -327,6 +331,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -383,6 +388,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -424,6 +430,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -470,6 +477,7 @@ class FakeWrapper:
 
     queue_ref = MagicMock()
     queue_ref._deserialize_payload.side_effect = lambda _name, data: cloudpickle.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: cloudpickle.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -504,10 +512,9 @@ def test_async_concurrency_default(tmp_path: Path) -> None:
 
 def test_async_executor_honors_per_task_serializer(poll_until: PollUntil) -> None:
     """Regression: the async executor deserializes via
-    ``queue._deserialize_payload`` (honoring ``@task(serializer=...)``) rather
-    than a hardcoded ``cloudpickle.loads``."""
-    import cloudpickle
-
+    ``queue._deserialize_payload`` (honoring ``@task(serializer=...)``) and
+    serializes results via ``queue._serialize_result`` rather than hardcoded
+    cloudpickle in either direction."""
     from taskito.async_support.executor import AsyncTaskExecutor
     from taskito.serializers import JsonSerializer
 
@@ -526,6 +533,7 @@ class FakeWrapper:
     # Route deserialization through a NON-cloudpickle serializer; a hardcoded
     # cloudpickle.loads would raise on this JSON payload.
     queue_ref._deserialize_payload.side_effect = lambda _name, data: serializer.loads(data)
+    queue_ref._serialize_result.side_effect = lambda _name, result: serializer.dumps(result)
     queue_ref._interceptor = None
     queue_ref._proxy_registry = None
     queue_ref._test_mode_active = False
@@ -544,5 +552,6 @@ class FakeWrapper:
     executor.stop()
 
     queue_ref._deserialize_payload.assert_called_once_with("mod.add", payload)
-    result = cloudpickle.loads(sender.report_success.call_args[0][2])
+    queue_ref._serialize_result.assert_called_once_with("mod.add", 5)
+    result = serializer.loads(sender.report_success.call_args[0][2])
     assert result == 5

From cc7a01cac718dae87bcd62443dcdb2c166059224 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Mon, 6 Jul 2026 00:41:46 +0530
Subject: [PATCH 3/5] feat(python): wire codec chain and per-task codecs into
 queue

---
 sdks/python/taskito/app.py                    |  81 +++++++++++---
 sdks/python/taskito/mixins/decorators.py      |  28 ++++-
 sdks/python/taskito/mixins/predicates.py      |   4 +-
 sdks/python/taskito/workflows/builder.py      |   5 +-
 .../taskito/workflows/tracker/fan_out.py      |   9 +-
 sdks/python/tests/core/test_codecs.py         | 105 ++++++++++++++++++
 6 files changed, 208 insertions(+), 24 deletions(-)

diff --git a/sdks/python/taskito/app.py b/sdks/python/taskito/app.py
index 912a1f36..4ea7785d 100644
--- a/sdks/python/taskito/app.py
+++ b/sdks/python/taskito/app.py
@@ -21,14 +21,16 @@
 import hashlib
 import logging
 import os
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Any
 
 from taskito._taskito import PyQueue
 from taskito.async_support.mixins import AsyncQueueMixin
 from taskito.batching import BatchAccumulator, BatchConfig
+from taskito.codecs import CodecSerializer, PayloadCodec
 from taskito.events import EventBus, EventType
+from taskito.exceptions import SerializationError
 from taskito.interception import ArgumentInterceptor
 from taskito.interception.built_in import build_default_registry
 from taskito.interception.metrics import InterceptionMetrics
@@ -122,6 +124,8 @@ def __init__(
         default_priority: int = 0,
         result_ttl: int | None = None,
         serializer: Serializer | None = None,
+        codec: PayloadCodec | Sequence[PayloadCodec] | None = None,
+        codecs: dict[str, PayloadCodec] | None = None,
         middleware: list[TaskMiddleware] | None = None,
         backend: str = "sqlite",
         db_url: str | None = None,
@@ -159,6 +163,17 @@ def __init__(
                 seconds. None disables auto-cleanup.
             serializer: Serializer for task payloads. Defaults to SmartSerializer
                 (msgpack with cloudpickle fallback).
+            codec: Global payload codec chain — a single :class:`PayloadCodec`
+                or a sequence applied in order after serialization (and in
+                reverse before deserialization). Wraps the queue serializer,
+                so it covers every payload and result. Jobs persisted before
+                the chain was enabled cannot be decoded through it. Note:
+                nondeterministic codecs (e.g. ``AesGcmCodec``) make
+                ``idempotent=True`` auto-keys nondeterministic per call —
+                pass an explicit ``idempotency_key`` instead.
+            codecs: Named codec registry for per-task codecs. Tasks opt in
+                via ``@queue.task(codecs=["name", ...])``; applies to task
+                payloads only (results stay on the queue serializer).
             middleware: List of global middleware instances applied to all tasks.
             backend: Storage backend — ``"sqlite"`` (default) or ``"postgres"``.
             db_url: PostgreSQL connection URL (required when backend is ``"postgres"``).
@@ -245,8 +260,18 @@ def __init__(
             "on_success": [],
             "on_failure": [],
         }
-        self._serializer: Serializer = serializer or SmartSerializer()
+        base_serializer: Serializer = serializer or SmartSerializer()
+        self._codec_chain: list[PayloadCodec] = (
+            [codec] if isinstance(codec, PayloadCodec) else list(codec or [])
+        )
+        self._serializer: Serializer = (
+            CodecSerializer(base_serializer, self._codec_chain)
+            if self._codec_chain
+            else base_serializer
+        )
+        self._codecs: dict[str, PayloadCodec] = dict(codecs or {})
         self._task_serializers: dict[str, Serializer] = {}
+        self._task_codecs: dict[str, list[str]] = {}
         self._task_idempotent: dict[str, bool] = {}
         self._task_compensates: dict[str, str] = {}
         self._task_batch_configs: dict[str, BatchConfig] = {}
@@ -315,6 +340,20 @@ def _get_serializer(self, task_name: str) -> Serializer:
         """Get the serializer for a task (per-task or queue-level fallback)."""
         return self._task_serializers.get(task_name, self._serializer)
 
+    def _apply_task_codecs(self, task_name: str, payload: bytes) -> bytes:
+        """Encode a serialized payload with the task's named codecs, in order."""
+        for name in self._task_codecs.get(task_name, ()):
+            codec = self._codecs.get(name)
+            if codec is None:
+                raise ValueError(f"no codec registered named {name!r}")
+            payload = codec.encode(payload)
+        return payload
+
+    def _encode_payload(self, task_name: str, args: tuple, kwargs: dict) -> bytes:
+        """Serialize ``(args, kwargs)`` and apply the task's named codecs."""
+        payload = self._get_serializer(task_name).dumps((args, kwargs))
+        return self._apply_task_codecs(task_name, payload)
+
     def _check_payload_size(self, task_name: str, size: int) -> None:
         """Reject a serialized payload that exceeds ``max_payload_bytes``."""
         limit = self.max_payload_bytes
@@ -389,8 +428,7 @@ def _dispatch_batched_payload(
         handle (one per item that landed in this batch).
         """
         del config  # currently unused at dispatch time; reserved for telemetry
-        task_serializer = self._get_serializer(task_name)
-        payload = task_serializer.dumps(((items,), {}))
+        payload = self._encode_payload(task_name, (items,), {})
         self._check_payload_size(task_name, len(payload))
         py_job = self._inner.enqueue(
             task_name=task_name,
@@ -437,6 +475,13 @@ def _deserialize_payload(self, task_name: str, payload: bytes) -> tuple:
         ``args`` straight to PyO3, which requires a real tuple, so the shape is
         normalized back to ``(tuple, dict)`` here regardless of serializer.
         """
+        task_codecs = self._task_codecs.get(task_name)
+        if task_codecs:
+            for name in reversed(task_codecs):
+                codec = self._codecs.get(name)
+                if codec is None:
+                    raise SerializationError(f"no codec registered named {name!r}")
+                payload = codec.decode(payload)
         args, kwargs = self._get_serializer(task_name).loads(payload)
         return tuple(args), kwargs
 
@@ -571,6 +616,18 @@ def enqueue(
 
         task_serializer = self._get_serializer(task_name)
         payload = task_serializer.dumps((final_args, final_kwargs))
+
+        # Dedup keys hash the pre-codec bytes so nondeterministic codecs
+        # (AES-GCM's random nonce) can't break idempotency for codec tasks.
+        unique_key = self._resolve_unique_key(
+            task_name=task_name,
+            payload=payload,
+            unique_key=unique_key,
+            idempotency_key=idempotency_key,
+            idempotent=idempotent,
+        )
+
+        payload = self._apply_task_codecs(task_name, payload)
         self._check_payload_size(task_name, len(payload))
 
         # Evaluate enqueue-time predicate (if registered). Outcome may
@@ -589,14 +646,6 @@ def enqueue(
                 delay=delay,
             )
 
-        unique_key = self._resolve_unique_key(
-            task_name=task_name,
-            payload=payload,
-            unique_key=unique_key,
-            idempotency_key=idempotency_key,
-            idempotent=idempotent,
-        )
-
         dep_ids = None
         if depends_on is not None:
             dep_ids = [depends_on] if isinstance(depends_on, str) else list(depends_on)
@@ -795,10 +844,10 @@ def enqueue_many(
             payloads = [
                 task_serializer.dumps((a, kw)) for a, kw in zip(args_list, kw_list, strict=True)
             ]
-        for payload in payloads:
-            self._check_payload_size(task_name, len(payload))
         task_names = [task_name] * count
 
+        # Dedup keys hash the pre-codec bytes so nondeterministic codecs
+        # (AES-GCM's random nonce) can't break idempotency for codec tasks.
         per_job_unique_keys = [
             self._resolve_unique_key(
                 task_name=task_name,
@@ -810,6 +859,10 @@ def enqueue_many(
             for i in range(count)
         ]
 
+        payloads = [self._apply_task_codecs(task_name, payload) for payload in payloads]
+        for payload in payloads:
+            self._check_payload_size(task_name, len(payload))
+
         # Evaluate enqueue-time predicate per row. Cancel raises for the
         # whole batch — all-or-nothing semantics. Defer adjusts that row's
         # delay only.
diff --git a/sdks/python/taskito/mixins/decorators.py b/sdks/python/taskito/mixins/decorators.py
index 11f88311..6d701f1b 100644
--- a/sdks/python/taskito/mixins/decorators.py
+++ b/sdks/python/taskito/mixins/decorators.py
@@ -21,6 +21,7 @@
     BatchResultTypeError,
     is_batch_item_result_list,
 )
+from taskito.codecs import CodecSerializer
 from taskito.context import _clear_context, current_job
 from taskito.dashboard.middleware_store import MiddlewareDisableStore
 from taskito.events import EventType
@@ -37,6 +38,7 @@
 )
 
 if TYPE_CHECKING:
+    from taskito.codecs import PayloadCodec
     from taskito.interception import ArgumentInterceptor
     from taskito.middleware import TaskMiddleware
     from taskito.predicates import Predicate
@@ -96,6 +98,9 @@ class QueueDecoratorMixin:
     _periodic_configs: list[dict[str, Any]]
     _hooks: dict[str, list[Callable]]
     _task_serializers: dict[str, Serializer]
+    _codec_chain: list[PayloadCodec]
+    _codecs: dict[str, PayloadCodec]
+    _task_codecs: dict[str, list[str]]
     _task_idempotent: dict[str, bool]
     _task_compensates: dict[str, str]
     _task_batch_configs: dict[str, Any]
@@ -369,6 +374,7 @@ def task(
         retry_delays: list[float] | None = None,
         inject: list[str] | None = None,
         serializer: Serializer | None = None,
+        codecs: list[str] | None = None,
         max_retry_delay: int | None = None,
         max_concurrent: int | None = None,
         idempotent: bool = False,
@@ -402,6 +408,10 @@ def task(
                 ``max_retries``.
             inject: List of resource names to inject as keyword arguments.
             serializer: Per-task serializer. Falls back to the queue-level serializer.
+            codecs: Names of payload codecs (registered via ``Queue(codecs=...)``)
+                applied in order to this task's serialized payload on enqueue
+                and reversed on the worker. Payload only — results stay on the
+                queue-level serializer.
             max_retry_delay: Maximum backoff delay in seconds. Defaults to 300
                 (5 minutes) if not set.
             max_concurrent: Maximum number of concurrent running instances of
@@ -507,9 +517,19 @@ def decorator(fn: Callable) -> TaskWrapper:
             if middleware:
                 self._task_middleware[task_name] = middleware
 
-            # Store per-task serializer
+            # Store per-task serializer, wrapped in the global codec chain so a
+            # per-task override can't silently bypass queue-wide codecs.
             if serializer is not None:
-                self._task_serializers[task_name] = serializer
+                self._task_serializers[task_name] = (
+                    CodecSerializer(serializer, self._codec_chain)
+                    if self._codec_chain
+                    else serializer
+                )
+
+            # Store per-task codec names (validated lazily against the
+            # registry at enqueue/decode time).
+            if codecs:
+                self._task_codecs[task_name] = list(codecs)
 
             # Store per-task idempotency flag (auto-derives unique_key on enqueue)
             if idempotent:
@@ -657,7 +677,7 @@ def _wf_launcher() -> str:
                     run = proxy.submit()
                     return f"submitted workflow run {run.id}"
 
-                payload = self._get_serializer(launcher_name).dumps(((), {}))  # type: ignore[attr-defined]
+                payload = self._encode_payload(launcher_name, (), {})  # type: ignore[attr-defined]
                 self._periodic_configs.append(
                     {
                         "name": launcher_name,
@@ -674,7 +694,7 @@ def _wf_launcher() -> str:
             wrapper = self.task(name=name, queue=queue)(fn)
 
             # Store periodic config for registration at worker startup
-            payload = self._get_serializer(wrapper.name).dumps((args, kwargs or {}))  # type: ignore[attr-defined]
+            payload = self._encode_payload(wrapper.name, args, kwargs or {})  # type: ignore[attr-defined]
             self._periodic_configs.append(
                 {
                     "name": name or f"{_resolve_module_name(fn.__module__)}.{fn.__qualname__}",
diff --git a/sdks/python/taskito/mixins/predicates.py b/sdks/python/taskito/mixins/predicates.py
index 193162c5..fc84e259 100644
--- a/sdks/python/taskito/mixins/predicates.py
+++ b/sdks/python/taskito/mixins/predicates.py
@@ -59,6 +59,7 @@ class QueuePredicateMixin:
     # Supplied by other mixins on the composed Queue.
     _emit_event: Callable[..., None]
     _get_serializer: Callable[[str], Serializer]
+    _encode_payload: Callable[..., bytes]
 
     def _init_predicate_state(self) -> None:
         """Initialise predicate-related instance state.
@@ -216,8 +217,7 @@ def _reenqueue_after_defer(
         middleware or re-evaluating the predicate (which would create an
         infinite ping-pong).
         """
-        serializer = self._get_serializer(task_name)
-        payload = serializer.dumps((tuple(args), dict(kwargs)))
+        payload = self._encode_payload(task_name, tuple(args), dict(kwargs))
         self._inner.enqueue(
             task_name=task_name,
             payload=payload,
diff --git a/sdks/python/taskito/workflows/builder.py b/sdks/python/taskito/workflows/builder.py
index d275291a..a24bea9a 100644
--- a/sdks/python/taskito/workflows/builder.py
+++ b/sdks/python/taskito/workflows/builder.py
@@ -473,8 +473,9 @@ def _compile(
                 and step.fan_in is None
                 and step.sub_workflow is None
             ):
-                serializer = queue._get_serializer(step.task_name)
-                node_payloads[step.name] = serializer.dumps((step.args, step.kwargs))
+                node_payloads[step.name] = queue._encode_payload(
+                    step.task_name, step.args, step.kwargs
+                )
 
         dag_bytes, step_metadata_json = builder.serialize()
         return (
diff --git a/sdks/python/taskito/workflows/tracker/fan_out.py b/sdks/python/taskito/workflows/tracker/fan_out.py
index ac1bd668..8d50ce82 100644
--- a/sdks/python/taskito/workflows/tracker/fan_out.py
+++ b/sdks/python/taskito/workflows/tracker/fan_out.py
@@ -61,7 +61,10 @@ def expand_fan_out(
     task_name = meta["task_name"]
     serializer = tracker._queue._get_serializer(task_name)
     child_names = [f"{fan_out_node}[{i}]" for i in range(len(items))]
-    child_payloads = [build_child_payload(item, serializer) for item in items]
+    child_payloads = [
+        tracker._queue._apply_task_codecs(task_name, build_child_payload(item, serializer))
+        for item in items
+    ]
 
     queue_name = meta.get("queue") or "default"
     max_retries = int_or(meta.get("max_retries"), 3)
@@ -164,7 +167,9 @@ def create_fan_in_job(
 
     task_name = meta["task_name"]
     serializer = tracker._queue._get_serializer(task_name)
-    payload = build_fan_in_payload(results, serializer)
+    payload = tracker._queue._apply_task_codecs(
+        task_name, build_fan_in_payload(results, serializer)
+    )
 
     queue_name = meta.get("queue") or "default"
     max_retries = int_or(meta.get("max_retries"), 3)
diff --git a/sdks/python/tests/core/test_codecs.py b/sdks/python/tests/core/test_codecs.py
index b452c943..4d51ef45 100644
--- a/sdks/python/tests/core/test_codecs.py
+++ b/sdks/python/tests/core/test_codecs.py
@@ -1,6 +1,10 @@
 """Unit tests for payload codecs and the codec serializer."""
 
 import gzip
+import threading
+from collections.abc import Generator
+from contextlib import AbstractContextManager, contextmanager
+from pathlib import Path
 
 import pytest
 
@@ -12,6 +16,7 @@
     HmacCodec,
     JsonSerializer,
     PayloadCodec,
+    Queue,
     SerializationError,
     Serializer,
     SmartSerializer,
@@ -160,3 +165,103 @@ def test_empty_chain_is_transparent(self) -> None:
         obj = (1, "two", [3])
         assert serializer.loads(serializer.dumps(obj)) == obj
         assert serializer.dumps(obj) == SmartSerializer().dumps(obj)
+
+
+class TestQueueCodecIntegration:
+    @staticmethod
+    def _run_worker(queue: Queue) -> AbstractContextManager[None]:
+        @contextmanager
+        def _ctx() -> Generator[None]:
+            thread = threading.Thread(target=queue.run_worker, daemon=True)
+            thread.start()
+            try:
+                yield
+            finally:
+                queue._inner.request_shutdown()
+                thread.join(timeout=5)
+
+        return _ctx()
+
+    def test_global_chain_round_trips_through_worker(self, tmp_path: Path) -> None:
+        queue = Queue(
+            db_path=str(tmp_path / "chain.db"),
+            workers=1,
+            codec=[GzipCodec(), HmacCodec(HMAC_KEY)],
+        )
+
+        @queue.task()
+        def double(x: int) -> int:
+            return x * 2
+
+        with self._run_worker(queue):
+            result = double.delay(21)
+            assert result.result(timeout=10) == 42
+            # Both the payload and the result are codec-framed in storage.
+            job = queue._inner.get_job(result.id)
+            assert job is not None and job.result_bytes is not None
+            hmac_codec = HmacCodec(HMAC_KEY)
+            assert gzip.decompress(hmac_codec.decode(bytes(job.payload_bytes)))
+            assert gzip.decompress(hmac_codec.decode(bytes(job.result_bytes)))
+
+    def test_single_codec_accepted_without_list(self, tmp_path: Path) -> None:
+        queue = Queue(db_path=str(tmp_path / "single.db"), workers=1, codec=GzipCodec())
+        assert len(queue._codec_chain) == 1
+
+    def test_per_task_codec_round_trips_through_worker(self, tmp_path: Path) -> None:
+        queue = Queue(
+            db_path=str(tmp_path / "pertask.db"),
+            workers=1,
+            codecs={"gzip": GzipCodec()},
+        )
+
+        @queue.task(codecs=["gzip"])
+        def shout(text: str) -> str:
+            return text.upper()
+
+        @queue.task()
+        def plain(text: str) -> str:
+            return text
+
+        with self._run_worker(queue):
+            result = shout.delay("hello")
+            assert result.result(timeout=10) == "HELLO"
+            job = queue._inner.get_job(result.id)
+            assert job is not None and job.result_bytes is not None
+            # Payload is gzip-framed; the result skips per-task codecs and
+            # stays on the plain queue serializer.
+            assert bytes(job.payload_bytes[:2]) == b"\x1f\x8b"
+            assert queue._serializer.loads(bytes(job.result_bytes)) == "HELLO"
+
+            plain_result = plain.delay("untouched")
+            assert plain_result.result(timeout=10) == "untouched"
+            plain_job = queue._inner.get_job(plain_result.id)
+            assert plain_job is not None
+            assert bytes(plain_job.payload_bytes[:2]) != b"\x1f\x8b"
+
+    def test_unregistered_codec_name_raises_at_enqueue(self, tmp_path: Path) -> None:
+        queue = Queue(db_path=str(tmp_path / "missing.db"), workers=1)
+
+        @queue.task(codecs=["nope"])
+        def orphan() -> None:
+            return None
+
+        with pytest.raises(ValueError, match="no codec registered named 'nope'"):
+            orphan.delay()
+
+    def test_idempotency_key_stable_under_encryption_codec(self, tmp_path: Path) -> None:
+        pytest.importorskip("cryptography")
+        queue = Queue(
+            db_path=str(tmp_path / "idem.db"),
+            workers=1,
+            codecs={"aes": make_aes_codec()},
+        )
+
+        @queue.task(codecs=["aes"], idempotent=True)
+        def once(x: int) -> int:
+            return x
+
+        # AES adds a random nonce per encode; dedup keys hash the pre-codec
+        # bytes, so the second enqueue dedups onto the first job.
+        first = once.delay(7)
+        second = once.delay(7)
+        assert first.id == second.id

From 378fa10c7ba31aabcaf3e5e2c11e15bad05ca347 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Mon, 6 Jul 2026 00:56:07 +0530
Subject: [PATCH 4/5] docs(python): keep codec docs SDK-neutral

---
 sdks/python/taskito/codecs.py | 13 ++++++-------
 1 file changed, 6 insertions(+), 7 deletions(-)

diff --git a/sdks/python/taskito/codecs.py b/sdks/python/taskito/codecs.py
index c2bbdb95..455fbcae 100644
--- a/sdks/python/taskito/codecs.py
+++ b/sdks/python/taskito/codecs.py
@@ -5,8 +5,8 @@
 signing. Codecs compose: a chain encodes in list order and decodes in reverse,
 so ``[gzip, hmac]`` verifies integrity *before* decompressing.
 
-Wire formats match the Java and Node SDKs byte-for-byte, so codec-framed
-payloads are cross-SDK compatible.
+Wire formats are part of the cross-SDK contract, so codec-framed payloads
+decode from any Taskito SDK.
 
 Enable globally via ``Queue(codec=...)`` (wraps the queue serializer, covers
 payloads and results) or per task via ``Queue(codecs={"name": codec})`` +
@@ -89,9 +89,9 @@ class AesGcmCodec:
     """AES-GCM encryption codec.
 
     Wire format: ``[12-byte random nonce][ciphertext || 16-byte GCM tag]`` —
-    identical to the Java and Node SDK codecs. Requires the ``encryption``
-    extra (``pip install taskito[encryption]``). Key must be 16, 24, or 32
-    bytes (AES-128/192/256).
+    the cross-SDK contract. Requires the ``encryption`` extra
+    (``pip install taskito[encryption]``). Key must be 16, 24, or 32 bytes
+    (AES-128/192/256).
     """
 
     def __init__(self, key: bytes):
@@ -126,8 +126,7 @@ def decode(self, data: bytes) -> bytes:
 class HmacCodec:
     """HMAC-SHA256 signing codec (authenticates, does not encrypt).
 
-    Wire format: ``[32-byte mac][body]`` — identical to the Java and Node SDK
-    codecs.
+    Wire format: ``[32-byte mac][body]`` — the cross-SDK contract.
     """
 
     def __init__(self, key: bytes):

From 40e577aaf8720bb83353939bb7ddd75fb7f2b3c0 Mon Sep 17 00:00:00 2001
From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com>
Date: Mon, 6 Jul 2026 01:11:44 +0530
Subject: [PATCH 5/5] fix(python): reject empty HMAC codec keys

---
 sdks/python/taskito/codecs.py         | 2 ++
 sdks/python/tests/core/test_codecs.py | 4 ++++
 2 files changed, 6 insertions(+)

diff --git a/sdks/python/taskito/codecs.py b/sdks/python/taskito/codecs.py
index 455fbcae..615e9ce9 100644
--- a/sdks/python/taskito/codecs.py
+++ b/sdks/python/taskito/codecs.py
@@ -132,6 +132,8 @@ class HmacCodec:
     def __init__(self, key: bytes):
         if not isinstance(key, bytes):
             raise TypeError(f"key must be bytes, got {type(key).__name__}")
+        if not key:
+            raise ValueError("key must not be empty")
         self._key = key
 
     def encode(self, data: bytes) -> bytes:
diff --git a/sdks/python/tests/core/test_codecs.py b/sdks/python/tests/core/test_codecs.py
index 4d51ef45..90c025fd 100644
--- a/sdks/python/tests/core/test_codecs.py
+++ b/sdks/python/tests/core/test_codecs.py
@@ -130,6 +130,10 @@ def test_constructor_rejects_non_bytes_key(self) -> None:
         with pytest.raises(TypeError, match="bytes"):
             HmacCodec("string-key")  # type: ignore[arg-type]
 
+    def test_constructor_rejects_empty_key(self) -> None:
+        with pytest.raises(ValueError, match="must not be empty"):
+            HmacCodec(b"")
+
 
 class TestPayloadCodecProtocol:
     def test_built_in_codecs_satisfy_protocol(self) -> None: