bugfixes: transaction parameter collector, generic formatting messages, guardrails, prompts. Testing contas

agent_framework_oci/libs/agent_framework/src/agent_framework/checkpoints/langgraph_saver.py

@@ -26,11 +26,104 @@ from .checkpoint_repository import create_checkpoint_repository
 
 
 def _jsonable(value: Any) -> Any:
-    try:
-        json.dumps(value, default=str)
+    """Convert a value to a JSON-safe tree without collapsing containers to strings.
+
+    The previous implementation used ``json.dumps(..., default=str)`` only as a
+    probe and then returned the *original* object.  Because ``default=str`` makes
+    virtually every object serializable, the fallback branch was never reached.
+    JSON repositories could therefore stringify arbitrary nested LangGraph objects
+    later, and restored checkpoints would contain strings where mappings were
+    required (for example ``metadata``, ``versions_seen`` or ``checkpoint_map``).
+
+    Keep dict/list structure durable and stringify only unsupported leaf objects.
+    """
+    if value is None or isinstance(value, (str, int, float, bool)):
         return value
-    except TypeError:
-        return json.loads(json.dumps(value, default=str))
+    if isinstance(value, dict):
+        return {str(k): _jsonable(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple, set, frozenset)):
+        return [_jsonable(v) for v in value]
+
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        try:
+            return _jsonable(model_dump(mode="python"))
+        except TypeError:
+            return _jsonable(model_dump())
+        except Exception:
+            pass
+
+    # Unsupported runtime objects are not durable.  At this point the object is a
+    # leaf; converting only the leaf to text cannot destroy an enclosing mapping.
+    return str(value)
+
+
+def _mapping(value: Any, *, default: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Recover a mapping from native or legacy JSON-string values.
+
+    Legacy rows written with permissive ``default=str`` serializers may contain a
+    JSON object encoded as text.  Parse that representation when possible; if the
+    value is an opaque string, return a safe empty/default mapping instead of
+    letting LangGraph fail with ``'str' object has no attribute 'items'``.
+    """
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            decoded = json.loads(value)
+        except Exception:
+            decoded = None
+        if isinstance(decoded, dict):
+            return decoded
+    return dict(default or {})
+
+
+def _normalize_checkpoint(checkpoint: Any) -> dict[str, Any]:
+    """Normalize LangGraph checkpoint mapping fields after JSON persistence."""
+    cp = _mapping(checkpoint)
+    if not cp:
+        return {}
+    cp = dict(cp)
+    cp["channel_values"] = _mapping(cp.get("channel_values"))
+    cp["channel_versions"] = _mapping(cp.get("channel_versions"))
+
+    versions_seen = _mapping(cp.get("versions_seen"))
+    cp["versions_seen"] = {
+        str(node): _mapping(versions) for node, versions in versions_seen.items()
+    }
+
+    pending_sends = cp.get("pending_sends")
+    if pending_sends is None:
+        cp["pending_sends"] = []
+    elif not isinstance(pending_sends, list):
+        cp["pending_sends"] = [pending_sends]
+    return cp
+
+
+def _durable_config(config: dict[str, Any] | None) -> dict[str, Any] | None:
+    """Return the persistable subset of a LangGraph RunnableConfig.
+
+    LangGraph injects private execution objects under ``configurable`` (notably
+    ``__pregel_runtime``). Those objects are valid only for the current graph
+    invocation and must never be persisted by a checkpointer. JSON-backed
+    repositories stringify them; on restore LangGraph then sees that string as
+    its runtime object and calls ``.override(...)``, producing errors such as
+    ``AttributeError: 'str' object has no attribute 'override'`` before the first
+    workflow node runs.
+
+    Durable checkpoint identity lives in the normal configurable keys
+    (thread_id/checkpoint_ns/checkpoint_id). User-defined configurable values are
+    preserved; only LangGraph's private ``__pregel_*`` execution keys are removed.
+    """
+    if config is None:
+        return None
+    cleaned = dict(_mapping(config))
+    configurable = dict(_mapping(cleaned.get("configurable")))
+    for key in list(configurable):
+        if str(key).startswith("__pregel_"):
+            configurable.pop(key, None)
+    cleaned["configurable"] = configurable
+    return cleaned
 
 
 def _thread_id(config: dict[str, Any] | None) -> str:
@@ -85,6 +178,13 @@ class RepositoryCheckpointSaver(BaseCheckpointSaver):
     """Checkpoint saver nativo para LangGraph usando os repositories do framework."""
 
     def __init__(self, settings, repository=None):
+        # BaseCheckpointSaver initializes LangGraph serializer state in current
+        # checkpoint releases. Keep compatibility with the lightweight fallback
+        # class used by framework-only unit tests.
+        try:
+            super().__init__()
+        except TypeError:
+            pass
         self.settings = settings
         self.repository = repository or create_checkpoint_repository(settings)
         self._loop: asyncio.AbstractEventLoop | None = None
@@ -103,10 +203,10 @@ class RepositoryCheckpointSaver(BaseCheckpointSaver):
     def _make_tuple(self, payload: dict[str, Any] | None):
         if not payload:
             return None
-        config = payload.get("config") or {"configurable": {"thread_id": payload.get("thread_id")}}
-        checkpoint = payload.get("checkpoint") or {}
-        metadata = payload.get("metadata") or {}
-        parent_config = payload.get("parent_config")
+        config = _durable_config(payload.get("config")) or {"configurable": {"thread_id": payload.get("thread_id")}}
+        checkpoint = _normalize_checkpoint(payload.get("checkpoint"))
+        metadata = _mapping(payload.get("metadata"))
+        parent_config = _durable_config(payload.get("parent_config"))
         pending_writes = _normalize_pending_writes(payload.get("pending_writes") or [])
         try:
             from langgraph.checkpoint.base import CheckpointTuple
@@ -129,11 +229,13 @@ class RepositoryCheckpointSaver(BaseCheckpointSaver):
     async def aput(self, config: dict[str, Any], checkpoint: dict[str, Any], metadata: dict[str, Any] | None = None, new_versions: dict[str, Any] | None = None):
         thread_id = _thread_id(config)
         checkpoint_id = _checkpoint_id(checkpoint)
+        base_config = _durable_config(config) or {}
         next_config = {
-            **(config or {}),
+            **base_config,
             "configurable": {
-                **((config or {}).get("configurable") or {}),
+                **_mapping(base_config.get("configurable")),
                 "thread_id": thread_id,
+                "checkpoint_ns": str(_mapping(base_config.get("configurable")).get("checkpoint_ns") or ""),
                 "checkpoint_id": checkpoint_id,
             },
         }

agent_framework_oci/tests/unit/test_langgraph_checkpoint_runtime_config.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+from agent_framework.checkpoints.langgraph_saver import (
+    RepositoryCheckpointSaver,
+    _durable_config,
+)
+
+
+class _Repo:
+    def __init__(self) -> None:
+        self.saved = None
+
+    async def put(self, thread_id, checkpoint):
+        self.saved = (thread_id, checkpoint)
+
+    async def get_latest(self, thread_id):
+        return self.saved[1] if self.saved and self.saved[0] == thread_id else None
+
+
+def test_private_pregel_runtime_is_not_durable() -> None:
+    config = {
+        "tags": ["x"],
+        "configurable": {
+            "thread_id": "t1",
+            "tenant": "default",
+            "__pregel_runtime": "stringified-runtime",
+            "__pregel_store": "stringified-store",
+        },
+    }
+
+    cleaned = _durable_config(config)
+
+    assert cleaned == {
+        "tags": ["x"],
+        "configurable": {"thread_id": "t1", "tenant": "default"},
+    }
+    assert "__pregel_runtime" in config["configurable"]
+
+
+def test_saver_strips_private_runtime_before_put_and_restore() -> None:
+    repo = _Repo()
+    saver = RepositoryCheckpointSaver(SimpleNamespace(), repository=repo)
+    config = {
+        "configurable": {
+            "thread_id": "t1",
+            "__pregel_runtime": "Runtime(...) persisted by JSON backend",
+        }
+    }
+    checkpoint = {"id": "cp1", "v": 1}
+
+    returned = asyncio.run(saver.aput(config, checkpoint, {}, {}))
+    assert returned["configurable"] == {"thread_id": "t1", "checkpoint_id": "cp1"}
+    assert repo.saved is not None
+    persisted = repo.saved[1]
+    assert "__pregel_runtime" not in persisted["config"]["configurable"]
+
+    # Also protects legacy records already stored with a stringified runtime.
+    persisted["config"]["configurable"]["__pregel_runtime"] = "legacy-string"
+    restored = saver._make_tuple(persisted)
+    restored_config = restored.config if hasattr(restored, "config") else restored["config"]
+    assert "__pregel_runtime" not in restored_config["configurable"]