fix(persistence): normalize non-finite floats before every JSON column write (#7733)

* fix(persistence): normalize non-finite floats before every JSON column write

The shared ORM engine serialized JSON columns with the lenient json.dumps
default (allow_nan=True), and the sync agent-storage engine set no
json_serializer at all. A non-finite float reaching a JSON column — run
metadata merges caller-supplied values, goal_verdict and
token_usage_by_model carry model/provider numbers, and json.loads('1e400')
yields inf — was emitted as the bare NaN/Infinity token, which PostgreSQL
jsonb rejects as invalid JSON, failing the whole write with DataError.
SQLite silently accepts and round-trips the invalid token, so the failure
was postgres-only and environment-dependent.

Adds dumps_json_safe/json_safe_value to persistence/json_compat.py (same
None-normalization the canonical display serializer applies since #7670,
which explicitly does not cover persisted state) and wires it into both
engines' json_serializer. Regressions pin: nested non-finite floats
normalize everywhere, finite values/bools pass through unchanged, the
async engine kwargs and the sync agent engine (postgres only; SQLite takes
no json_serializer) both route through it.

* fix(persistence): normalize JSON across synchronous engines

* docs: keep frontend guidance within the CI size budget

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
hataa 2026-10-31 18:12:32 +08:00 committed by GitHub
parent bee97022b1
commit 3027ab6714
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 204 additions and 6 deletions

View File

@ -133,6 +133,8 @@ That prompt is intended for coding agents. It tells the agent to clone the repo
### Configuration
Application SQL JSON writes normalize non-finite floats (`NaN`, `Infinity`, `-Infinity`) to `null` on SQLite and PostgreSQL, including DB-backed agent definitions and skill-mutation records. Existing rows are not rewritten.
With `run_events.backend: db`, caller-supplied ISO 8601 event timestamps are normalized to UTC before storage, preserving their instant when SQLite drops timezone information. Timestamps without an offset are treated as UTC. Previously stored timestamps are not rewritten.
Run event streams include `run.terminal` for the owning worker's confirmed final

View File

@ -8,6 +8,8 @@ Acquire that lock by polling `pg_try_advisory_lock`, never with a blocking `pg_a
When `database.postgres_schema` is configured, both async ORM connections and the synchronous SQLAlchemy connections used by DB-backed custom agents and managed subagents must use the same `search_path`; preserve this invariant when adding another persistence entry point.
Every JSON column write goes through `dumps_json_safe` (`json_compat.py`): non-finite floats normalize to `None` because PostgreSQL jsonb rejects the bare `NaN`/`Infinity` token (SQLite would silently accept it). The shared async engine, sync agent-storage engine, and sync skill-mutation engine wire it on both SQLite and PostgreSQL; keep new JSON-writing persistence entry points on it too.
Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned by the bootstrap critical section until the worker finishes. Drain those `asyncio.to_thread()` calls across host cancellation before releasing the in-process SQLite bootstrap lock or PostgreSQL advisory lock; otherwise another bootstrap can overlap a still-running migration worker.
On SQLite, `BEGIN IMMEDIATE` takes the database-wide write lock, so every unrelated writer (run status, thread metadata, the scheduler) waits for the transaction and fails with `database is locked` after `busy_timeout` (30s). Do slow work such as document conversion before opening a locked transaction; lock only to revalidate and publish, as `ProjectDocumentRepository.publish_under_live_lock` does. `tests/test_project_document_tools.py::TestConversionSerialization` pins this.

View File

@ -34,6 +34,7 @@ from deerflow.persistence.agents.base import (
parse_agent_config,
)
from deerflow.persistence.agents.model import AgentRow
from deerflow.persistence.json_compat import dumps_json_safe
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
@ -49,7 +50,8 @@ _engines_lock = threading.Lock()
def _build_engine(url: str) -> Engine:
connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {}
engine = create_engine(url, future=True, pool_pre_ping=True, connect_args=connect_args)
# Keep JSON columns consistent with the async engine on both backends.
engine = create_engine(url, future=True, pool_pre_ping=True, connect_args=connect_args, json_serializer=dumps_json_safe)
if url.startswith("sqlite"):
# Mirror the async engine's per-connection PRAGMAs (persistence/engine.py).
# journal_mode=WAL is persistent on the DB file (the async bootstrap sets

View File

@ -11,11 +11,11 @@ None and fall back to in-memory implementations.
from __future__ import annotations
import asyncio
import json
import logging
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
from deerflow.persistence.json_compat import dumps_json_safe
from deerflow.utils.file_io import await_drained
# Recycle pooled Postgres connections before stale idle sockets can hang
@ -25,8 +25,15 @@ POSTGRES_COMMAND_TIMEOUT_SECONDS = 30
def _json_serializer(obj: object) -> str:
"""JSON serializer with ensure_ascii=False for Chinese character support."""
return json.dumps(obj, ensure_ascii=False)
"""JSON serializer for every JSON column written through the ORM engine.
ensure_ascii=False keeps Chinese character support; non-finite floats are
normalized to ``None`` first — PostgreSQL jsonb rejects the bare
``NaN`` / ``Infinity`` token as invalid JSON, so a lenient dump turns a
reachable non-finite value (caller metadata, model verdict JSON, provider
usage numbers) into a ``DataError`` on the whole write.
"""
return dumps_json_safe(obj)
def _postgres_engine_kwargs(

View File

@ -2,6 +2,8 @@
from __future__ import annotations
import json
import math
import re
from collections.abc import Mapping
from dataclasses import dataclass
@ -14,6 +16,33 @@ from sqlalchemy.sql.expression import ColumnElement
from sqlalchemy.sql.visitors import InternalTraversal
from sqlalchemy.types import Boolean, TypeEngine
def json_safe_value(obj: Any) -> Any:
"""Recursively replace non-finite floats with ``None``.
``NaN`` / ``inf`` / ``-inf`` are not valid JSON (RFC 8259) but reach
persistence from ordinary runtime data — caller-supplied run metadata,
model-authored ``goal_verdict`` JSON, or provider usage numbers
(``json.loads("1e400")`` yields ``inf``). PostgreSQL's jsonb rejects the
bare ``NaN`` / ``Infinity`` token as invalid JSON, turning the column
write into a ``DataError``; SQLite silently round-trips it instead. This
mirrors the canonical display serializer's normalization (#7670) for the
persisted surfaces that one explicitly does not cover.
"""
if isinstance(obj, float):
return obj if math.isfinite(obj) else None
if isinstance(obj, dict):
return {key: json_safe_value(value) for key, value in obj.items()}
if isinstance(obj, (list, tuple)):
return [json_safe_value(value) for value in obj]
return obj
def dumps_json_safe(obj: Any) -> str:
"""Serialize JSON column payloads without ever emitting bare NaN/Infinity."""
return json.dumps(json_safe_value(obj), ensure_ascii=False)
# Key is interpolated into compiled SQL; restrict charset to prevent injection.
_KEY_CHARSET_RE = re.compile(r"^[A-Za-z0-9_\-]+$")

View File

@ -10,6 +10,8 @@ from sqlalchemy import create_engine, event
from sqlalchemy.engine import make_url
from sqlalchemy.orm import sessionmaker
from deerflow.persistence.json_compat import dumps_json_safe
class PublicationTopology:
"""Operator attests local single host; SQLite additionally enforces one host.
@ -64,7 +66,7 @@ def mutation_session_factory(url: str, *, postgres_schema: str = ""):
args = {"connect_timeout": 5, "options": f"{options} -c lock_timeout=5000 -c statement_timeout=30000 -c synchronous_commit=on".strip()}
else:
raise HostCapabilityError("UNSUPPORTED_TOPOLOGY")
engine = create_engine(url, connect_args=args, pool_size=4, max_overflow=0, pool_timeout=5, pool_pre_ping=True)
engine = create_engine(url, connect_args=args, pool_size=4, max_overflow=0, pool_timeout=5, pool_pre_ping=True, json_serializer=dumps_json_safe)
if parsed.get_backend_name() == "sqlite":
@event.listens_for(engine, "connect")

View File

@ -0,0 +1,154 @@
"""PostgreSQL JSON columns must never receive bare NaN/Infinity tokens.
The shared ORM engine's ``json_serializer`` (:func:`deerflow.persistence.engine._json_serializer`)
and the sync agent-storage and skill-mutation engines all use the same serializer.
Without it they serialize JSON columns with the lenient ``json.dumps`` default
(``allow_nan=True``). A non-finite float reaching a JSON column — run metadata
merges caller-supplied values, ``goal_verdict`` and ``token_usage_by_model``
carry model/provider numbers — is emitted as the bare ``NaN`` / ``Infinity``
token, which PostgreSQL's jsonb rejects as invalid JSON, turning the write
into a ``DataError``. SQLite silently accepts and round-trips the invalid
token instead, so the failure is postgres-only and environment-dependent.
The display/wire surfaces got the same normalization in the canonical
serializer (#7670), which explicitly does not cover persisted state; the
Gateway already uses ``allow_nan=False`` at three hand-picked sites. This pins
the shared persistence layer to the same contract: non-finite floats become
``None`` before any JSON column is written.
"""
from __future__ import annotations
import json
from deerflow.persistence.engine import _json_serializer, _postgres_engine_kwargs
from deerflow.persistence.json_compat import dumps_json_safe, json_safe_value
def _strict_loads(payload: str):
"""json.loads, but bare NaN/Infinity/-Infinity tokens fail loudly."""
def _reject(token):
raise AssertionError(f"bare {token} token reached the wire: {payload}")
return json.loads(payload, parse_constant=_reject)
def test_non_finite_floats_become_none_everywhere():
payload = dumps_json_safe({"a": float("inf"), "b": float("-inf"), "c": float("nan"), "nested": [float("inf"), {"d": float("nan")}], "keep": 1.5})
parsed = _strict_loads(payload)
assert parsed == {"a": None, "b": None, "c": None, "nested": [None, {"d": None}], "keep": 1.5}
def test_finite_values_and_bools_pass_through_unchanged():
value = {"s": "文", "i": 3, "b": True, "f": 0.25, "n": None, "l": [1, False]}
assert _strict_loads(dumps_json_safe(value)) == value
assert json_safe_value(value) == value
def test_async_postgres_engine_serializer_is_safe():
kwargs = _postgres_engine_kwargs(echo=False, pool_size=5)
assert kwargs["json_serializer"] is _json_serializer
assert _strict_loads(kwargs["json_serializer"]({"x": float("inf")})) == {"x": None}
def test_sync_agent_engine_wires_the_safe_serializer_for_postgres(monkeypatch):
import deerflow.persistence.agents.sql as agents_sql
captured: dict[str, object] = {}
def fake_create_engine(url, **kwargs):
captured["url"] = url
captured["kwargs"] = kwargs
engine = type("FakeEngine", (), {"dialect": type("D", (), {"name": "postgresql"})()})()
return engine
monkeypatch.setattr(agents_sql, "create_engine", fake_create_engine)
agents_sql._build_engine("postgresql+psycopg://u:p@localhost/db")
assert captured["url"].startswith("postgresql")
serializer = captured["kwargs"].get("json_serializer")
assert serializer is not None, "the sync postgres engine must set a json serializer"
assert _strict_loads(serializer({"x": float("nan")})) == {"x": None}
def test_sync_agent_engine_normalizes_sqlite_json_columns(tmp_path):
from sqlalchemy import select, text
from sqlalchemy.orm import Session
from deerflow.persistence.agents.model import AgentRow
from deerflow.persistence.agents.sql import _build_engine
engine = _build_engine(f"sqlite:///{tmp_path / 'agents.db'}")
value = {"label": "文", "keep": [1.5, True, None], "nested": [float("nan"), {"positive": float("inf"), "negative": float("-inf")}]}
expected = {"label": "文", "keep": [1.5, True, None], "nested": [None, {"positive": None, "negative": None}]}
try:
AgentRow.__table__.create(engine)
with Session(engine) as session, session.begin():
session.add(AgentRow(id="agent", user_id="owner", name="test", config=value))
with Session(engine) as session:
raw = session.scalar(text("SELECT config FROM agents WHERE id = 'agent'"))
assert _strict_loads(raw) == expected
assert session.scalar(select(AgentRow.config)) == expected
finally:
engine.dispose()
def test_sync_skill_mutation_engine_wires_the_safe_serializer_for_postgres(monkeypatch):
from types import SimpleNamespace
from deerflow.skills.mutations import topology
captured = {}
def fake_create_engine(url, **kwargs):
captured.update(kwargs)
return SimpleNamespace()
monkeypatch.setattr(topology, "create_engine", fake_create_engine)
topology.mutation_session_factory("postgresql+psycopg://u:p@localhost/db")
serializer = captured.get("json_serializer")
assert serializer is not None, "the skill-mutation engine must set a json serializer"
assert _strict_loads(serializer({"x": float("nan")})) == {"x": None}
def test_sync_skill_mutation_engine_normalizes_sqlite_json_columns(tmp_path):
from sqlalchemy import select, text
from deerflow.persistence.skill_mutations.model import SkillProposalRow
from deerflow.persistence.user.model import UserRow
from deerflow.skills.mutations.topology import mutation_session_factory
engine, sessions = mutation_session_factory(f"sqlite:///{tmp_path / 'mutations.db'}")
value = {"label": "文", "keep": [1.5, True, None], "nested": [float("nan"), {"positive": float("inf"), "negative": float("-inf")}]}
expected = {"label": "文", "keep": [1.5, True, None], "nested": [None, {"positive": None, "negative": None}]}
try:
UserRow.__table__.create(engine)
SkillProposalRow.__table__.create(engine)
with sessions.begin() as session:
session.add(UserRow(id="owner", email="owner@example.test"))
with sessions.begin() as session:
session.add(
SkillProposalRow(
proposal_id="proposal",
plugin_id="test",
owner_id="owner",
target_ref="target",
name="skill",
base_revision={},
sources=[],
candidate_hash="hash",
state="draft",
expires_at=1000,
check_result=value,
idempotency_key="key",
request_hash="hash",
)
)
with sessions() as session:
raw = session.scalar(text("SELECT check_result FROM skill_mutation_proposals WHERE proposal_id = 'proposal'"))
assert _strict_loads(raw) == expected
assert session.scalar(select(SkillProposalRow.check_result)) == expected
finally:
engine.dispose()

View File

@ -144,7 +144,7 @@ Run duration is run-scoped UI metadata even though the compatibility field `addi
`MessageReasoning` controls its disclosure state: completed mounts start collapsed, live streams open and close once after completion, and history expansion stays under user control. Keep this lifecycle in the workspace wrapper rather than editing generated AI elements. Duration-target selection is lazy for mounted, completed groups with a duration and cached by stable group identity; streaming updates must not re-convert unchanged historical processing/subagent groups. Changed groups must invalidate that cache.
The workspace-change card follows the same rule: it is resolved from `(threadId, runId)` alone, so every AI message of a run would render an identical copy. A run ends in more than one terminal assistant bubble whenever the model emits answer text that never gains a tool call, so `core/messages/workspace-change-anchor.ts` picks the run's last assistant bubble and `MessageListItem` renders the badge only for that anchor (#4555). Any future run-scoped display belongs in the same place — do not hang one off every message. The two anchor helpers deliberately differ in which group types they accept as a run's last position, because an anchor is only useful where the display is actually rendered: run duration is emitted by `MessageList` around every group, so it accepts any type, while the workspace-change card comes from `MessageListItem` and so restricts to `assistant`. Keep a new helper's candidate set matched to its own render site rather than unifying them.
`core/messages/workspace-change-anchor.ts` chooses the last `assistant` bubble for each `(threadId, runId)`; only that `MessageListItem` renders the workspace-change badge (#4555), even when a run has multiple terminal assistant bubbles. Keep future run-scoped displays at one anchor. Match each helper's candidate groups to its render site: duration renders around every group in `MessageList` and accepts any type; the workspace-change card renders only in assistant `MessageListItem`s. Do not unify those candidate sets.
Composer drafts are tab-scoped browser state. `core/threads/composer-draft.ts` stores text with inline reference tokens and conversation metadata in `sessionStorage`, keyed by user, agent, and logical conversation scope. New-chat pages pass the stable scope `"new"` because their runtime `threadId` is a fresh UUID on every reload; established conversations use their real thread ID. New drafts write `skillName: null`; `InputBox` waits for the enabled agent skill scope before converting an older draft selection into an inline `@` reference, and degrades a missing/disabled skill back to editable slash text. The stored draft clears through `SendMessageOptions.onSent`, which fires only once the send is dispatched: after the in-flight guard and any attachment upload, so a failed upload keeps the draft, quotes, conversation references and staged project files for a retry. If the composer moved to another conversation or unmounted meanwhile, `onSent` leaves live state alone and retires only the stored draft and staged files that send carried, keeping a later draft or attachment. Attachments, sidecar quotes, voice state, and polish undo state are not persisted.