diff --git a/README.md b/README.md index 87ccdbe72f..ce39624736 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index a96fd657ab..e200e7e3ea 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -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. diff --git a/backend/packages/harness/deerflow/persistence/agents/sql.py b/backend/packages/harness/deerflow/persistence/agents/sql.py index 543a6940c0..93fd3e6059 100644 --- a/backend/packages/harness/deerflow/persistence/agents/sql.py +++ b/backend/packages/harness/deerflow/persistence/agents/sql.py @@ -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 diff --git a/backend/packages/harness/deerflow/persistence/engine.py b/backend/packages/harness/deerflow/persistence/engine.py index 4c99ab392a..cd552fe687 100644 --- a/backend/packages/harness/deerflow/persistence/engine.py +++ b/backend/packages/harness/deerflow/persistence/engine.py @@ -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( diff --git a/backend/packages/harness/deerflow/persistence/json_compat.py b/backend/packages/harness/deerflow/persistence/json_compat.py index 6cd57b8656..cf1afb6542 100644 --- a/backend/packages/harness/deerflow/persistence/json_compat.py +++ b/backend/packages/harness/deerflow/persistence/json_compat.py @@ -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_\-]+$") diff --git a/backend/packages/harness/deerflow/skills/mutations/topology.py b/backend/packages/harness/deerflow/skills/mutations/topology.py index 25d6fa4a6e..d4aaf78092 100644 --- a/backend/packages/harness/deerflow/skills/mutations/topology.py +++ b/backend/packages/harness/deerflow/skills/mutations/topology.py @@ -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") diff --git a/backend/tests/test_persistence_json_safe_dumps.py b/backend/tests/test_persistence_json_safe_dumps.py new file mode 100644 index 0000000000..41bc14a9cd --- /dev/null +++ b/backend/tests/test_persistence_json_safe_dumps.py @@ -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() diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 2b46f5cf65..0f39f3991d 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -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.