mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-28 17:06:05 +00:00
e3ddf4e331
* feat(persistence): wire alembic migrations + bootstrap schema on startup Closes #3682. Pre-#3658 DBs lack the `runs.token_usage_by_model` column because alembic was never wired up — startup only ran `create_all`, which never ALTERs existing tables. Adds a hybrid bootstrap in FastAPI lifespan (replaces bare `create_all`): - empty DB → create_all + stamp head - legacy DB → stamp 0001_baseline + upgrade head - versioned DB → upgrade head Concurrency: Postgres `pg_advisory_lock` (cross-process); SQLite per-engine `asyncio.Lock` + 30s `PRAGMA busy_timeout` on both prod and alembic engines. Column revisions use `safe_add_column` / `safe_drop_column` idempotent helpers as fallback. Other bits: - 0001 baseline (chain root) + 0002 add `runs.token_usage_by_model` - `include_object` filter so alembic ignores LangGraph checkpointer tables - `make migrate-rev MSG="..."` for authoring new revisions (no migrate/stamp targets — startup is the only execution path) - Tests: three-branch decision, concurrency, #3682 regression, env filter, blocking-IO gate anchor - CLAUDE.md: new "Schema migrations" section * fix(style): fix lint error * perf(persistence): address review feedback on alembic bootstrap Behavioural fixes - _SQLITE_LOCKS now keyed via WeakKeyDictionary so id-reuse after GC cannot return a stale, loop-bound lock and the cache cannot leak one entry per disposed engine. - safe_add_column compares nullable / server_default against the desired column when the name already exists and emits a warning on drift, surfacing manual-ALTER workarounds instead of silently no-op'ing. - _postgres_lock issues SET LOCAL idle_in_transaction_session_timeout=0 before pg_advisory_lock, so managed Postgres cannot kill the idle lock-holding session mid-upgrade and silently release the advisory lock. - legacy branch now backfills missing baseline tables via a restricted create_all (Base.metadata.create_all scoped to _BASELINE_TABLE_NAMES). Restores pre-#1930 upgraders whose channel_* tables were never provisioned, without pre-empting future create_table revisions for newly-added models. Schema parity - runs.token_usage_by_model gains server_default=text("'{}'") in both the ORM model and the 0001_baseline create_table, matching what 0002 adds via ALTER. create_all and alembic-upgrade paths now produce identical column definitions. - New parity test compares Base.metadata.create_all output against a pure alembic upgrade base->head, asserting column-set, nullable, and server_default agree across all tables (normalized through the same helper safe_add_column's drift check uses). Guards - test_baseline_table_names_constant_matches_0001 pins _BASELINE_TABLE_NAMES to 0001_baseline.upgrade()'s actual output -- the constant cannot drift silently when someone edits 0001. - test_legacy_backfill_skips_non_baseline_tables verifies the restricted backfill does not create a phantom table on Base.metadata, modelling a future revision that would otherwise collide on op.create_table. Doc residuals - Three-branch decision table is now consistent across bootstrap.py top docstring, engine.py comment, test module docstring, and CLAUDE.md. - Stale test anchor in blocking_io/test_persistence_engine_sqlite.py docstring now points at the real file. * fix(style): fix lint error * fix(persistence): close drift detection holes - _check_column_drift compares column type via a family equivalence allowlist ({JSON, JSONB}). Catches the wrong-type workaround `TEXT NOT NULL DEFAULT '{}'` that previously slipped through silently, while keeping Postgres JSON/JSONB dialect reflection quiet. Reflected and desired type are also echoed in every drift warning's payload for operator triage. - Extract _escape_url_for_alembic so bootstrap._alembic_safe_url and scripts/_autogen_revision share the ConfigParser % escape rule instead of duplicating it. - backend/README.md: add `make migrate-rev MSG=...` to Commands and a Schema Migrations section per the repo's README/CLAUDE.md sync policy. - test_base_to_dict.py: scope the test ORM class to an isolated MetaData so the create_all-vs-alembic parity test (added in the previous commit) is not polluted by the phantom table on the full pytest session.
51 lines
2.3 KiB
Python
51 lines
2.3 KiB
Python
"""ORM model for run metadata."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import JSON, DateTime, Index, String, Text, text
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from deerflow.persistence.base import Base
|
|
|
|
|
|
class RunRow(Base):
|
|
__tablename__ = "runs"
|
|
|
|
run_id: Mapped[str] = mapped_column(String(64), primary_key=True)
|
|
thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
|
|
assistant_id: Mapped[str | None] = mapped_column(String(128))
|
|
user_id: Mapped[str | None] = mapped_column(String(64), index=True)
|
|
status: Mapped[str] = mapped_column(String(20), default="pending")
|
|
# "pending" | "running" | "success" | "error" | "timeout" | "interrupted"
|
|
|
|
model_name: Mapped[str | None] = mapped_column(String(128))
|
|
multitask_strategy: Mapped[str] = mapped_column(String(20), default="reject")
|
|
metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
|
|
kwargs_json: Mapped[dict] = mapped_column(JSON, default=dict)
|
|
error: Mapped[str | None] = mapped_column(Text)
|
|
|
|
# Convenience fields (for listing pages without querying RunEventStore)
|
|
message_count: Mapped[int] = mapped_column(default=0)
|
|
first_human_message: Mapped[str | None] = mapped_column(Text)
|
|
last_ai_message: Mapped[str | None] = mapped_column(Text)
|
|
|
|
# Token usage (accumulated in-memory by RunJournal, written on run completion)
|
|
total_input_tokens: Mapped[int] = mapped_column(default=0)
|
|
total_output_tokens: Mapped[int] = mapped_column(default=0)
|
|
total_tokens: Mapped[int] = mapped_column(default=0)
|
|
llm_call_count: Mapped[int] = mapped_column(default=0)
|
|
lead_agent_tokens: Mapped[int] = mapped_column(default=0)
|
|
subagent_tokens: Mapped[int] = mapped_column(default=0)
|
|
middleware_tokens: Mapped[int] = mapped_column(default=0)
|
|
token_usage_by_model: Mapped[dict] = mapped_column(JSON, default=dict, server_default=text("'{}'"))
|
|
|
|
# Follow-up association
|
|
follow_up_to_run_id: Mapped[str | None] = mapped_column(String(64))
|
|
|
|
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
|
|
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
|
|
|
|
__table_args__ = (Index("ix_runs_thread_status", "thread_id", "status"),)
|