Skip to content

symfonic.agent.checkpointer.serde

serde

Resilient checkpoint serializer (v7.27.0).

The LangGraph graph state (:class:symfonic.core.state.BaseAgentState) carries RUNTIME-INJECTED channels that are not meant to be persisted: deps (the :class:~symfonic.core.deps.BaseAgentDeps capability registry) and _callback_manager (the per-invocation callback manager). These objects are re-injected into input_state on EVERY run / stream turn, so their checkpointed value is never read back -- the engine never resumes deps from a checkpoint.

Pre-v7.27.0 this was latent: the checkpointer was only wired under ask_user_enabled and no test drove a successful full-run checkpoint write, so the fact that MemorySaver (langgraph >= 0.6) calls dumps_typed on every channel value -- and BaseAgentDeps is not msgpack-serializable -- never surfaced. v7.27.0 wires the checkpointer for transcript persistence (without ask_user), exercising that write path and exposing the TypeError: Type is not msgpack serializable: BaseAgentDeps crash.

SafeCheckpointSerde wraps the default JsonPlusSerializer and substitutes a None payload for any channel value that fails to serialize, instead of letting the whole checkpoint write blow up. This is correct precisely because the offending channels are runtime-only and re-injected each turn; persisting None for them loses nothing. All serializable channels (messages, query, scope ids, ...) round-trip unchanged through the wrapped serializer.

make_safe_serde 

make_safe_serde() -> Any

Return a JsonPlusSerializer subclass that never raises on dump.

Built lazily (the langgraph import is optional / extra-gated) so this module imports cleanly even when the checkpointer extras are absent.

Source code in src/symfonic/agent/checkpointer/serde.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def make_safe_serde() -> Any:
    """Return a ``JsonPlusSerializer`` subclass that never raises on dump.

    Built lazily (the langgraph import is optional / extra-gated) so this
    module imports cleanly even when the checkpointer extras are absent.
    """
    from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

    class SafeCheckpointSerde(JsonPlusSerializer):  # type: ignore[misc]
        """JsonPlusSerializer that degrades non-serializable values to None.

        Works for both serialization granularities:

        * **Per-channel** savers (Postgres / SQLite call ``dumps_typed`` on each
          channel value): a failing value degrades to ``None`` directly.
        * **Whole-checkpoint** savers (``MongoDBSaver`` calls ``dumps_typed`` on
          the entire checkpoint at once): a naive "whole thing -> None" would
          nuke the whole checkpoint (including ``messages``) because one channel
          (``deps`` / ``_callback_manager``) is non-serializable. Instead, when
          the failing object is a checkpoint, sanitize its ``channel_values``
          per entry (nulling only the offenders) and retry — so ``messages`` and
          the rest survive. Without this, get_transcript / restart-resume return
          empty on Mongo (v8.12.0 fix).
        """

        def dumps_typed(self, obj: Any) -> tuple[str, bytes]:
            try:
                return super().dumps_typed(obj)
            except (TypeError, ValueError) as exc:
                # Whole-checkpoint case: keep the serializable channels, null
                # only the offenders, then retry the whole-checkpoint dump.
                if isinstance(obj, dict) and "channel_values" in obj:
                    sanitized = self._sanitize_checkpoint(obj)
                    try:
                        return super().dumps_typed(sanitized)
                    except (TypeError, ValueError):
                        pass  # fall through to the null-payload degrade
                # Per-channel case (or a checkpoint still unserializable after
                # sanitizing): the value is runtime-only and re-injected each
                # turn, so persisting None loses nothing.
                logger.debug(
                    "SafeCheckpointSerde: dropping non-serializable value of "
                    "type %s (persisted as None): %s",
                    type(obj).__name__, exc,
                )
                return _NULL_TYPED

        def _sanitize_checkpoint(self, checkpoint: dict[str, Any]) -> dict[str, Any]:
            """Copy ``checkpoint`` with non-serializable channel values nulled."""
            safe_values: dict[str, Any] = {}
            for name, value in checkpoint.get("channel_values", {}).items():
                try:
                    super().dumps_typed(value)
                    safe_values[name] = value
                except (TypeError, ValueError):
                    logger.debug(
                        "SafeCheckpointSerde: nulling non-serializable channel "
                        "%r (type %s) inside the checkpoint.",
                        name, type(value).__name__,
                    )
                    safe_values[name] = None
            sanitized = dict(checkpoint)
            sanitized["channel_values"] = safe_values
            return sanitized

    return SafeCheckpointSerde()