diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md index 4301a60..29f2406 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md +++ b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md @@ -1,6 +1,6 @@ # AI 费用闭环与价值证明 概念文档 -更新时间：2026-07-13 +更新时间：2026-07-14 文档路径：document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/CONCEPT.md @@ -246,6 +246,7 @@ - 旧 `ReimbursementRequest` 进入只读兼容和迁移状态，停止新增第二套业务编排。 - 现有 `risk_flags_json` 保持读取兼容，新审批、付款、归档和关系事件写入结构化表。 - API 新字段优先追加，不在同一阶段破坏现有前端契约。 +- 迁移桥接期集中维护 migration-owned 表集合；所有 legacy bootstrap 只能创建集合之外的表。标准启动在 `alembic upgrade head` 前只读核对 revision 与自有表集合，发现漂移时拒绝启动，不自动猜测、stamp 或修复。 - 所有表结构通过 Alembic 迁移，不继续在请求路径执行 DDL。 #### 版本与审计 @@ -405,7 +406,7 @@ customer_contribution_margin ```bash docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \ - local-x-financial-linux \ + x-financial-local-linux \ /tmp/x-financial-server-venv/bin/pytest -q server/tests/test_expense_case_service.py ``` @@ -500,3 +501,6 @@ docker exec -w /app -e SERVER_VENV_DIR=/tmp/x-financial-server-venv \ - 2026-07-13（AI 申请身份边界）：申请预览快速入口不再接受请求体中的 `user_id`、管理员标记、角色、员工编号或其他身份字段作为授权事实，全部强制绑定服务端会话；伪造管理员身份编辑他人退回申请会返回 400，且原申请与费用事件不发生变化。 - 2026-07-13（安全与事务验证）：容器内受影响后端定向回归 36 项、Expense Case 前端兼容测试 9 项和 Python `ruff --select F,I` 通过；覆盖本人、当前审批人、财务、管理员、无权限、跨租户、整 Case 安全摘要、草稿新建/更新、失败回滚、HTTP 创建重放、相同快照去重、不同快照留痕、Steward 重放及伪造身份越权。未修改数据库结构，未执行持久化开发数据库迁移。 - 2026-07-13（联调边界）：当前持久化开发数据库尚无 `auth_sessions`、`expense_cases`、`expense_case_links` 和 `business_events` 表，浏览器登录无法获得有效认证凭证，因此本轮未声称完成真实页面端到端联调，也未擅自执行数据库迁移。计划、消费/票据、入账、对账和复盘事件仍待后续补齐。 +- 2026-07-14（迁移所有权加固）：新增统一 `schema_ownership.py`，七个运行时初始化入口只创建 legacy 表；标准启动在 Alembic upgrade 前执行只读漂移预检，revision 与 migration-owned 表集合不一致时 fail-fast，且不会自动 stamp 或修改数据库。 +- 2026-07-14（真实迁移验证）：在主应用容器连接的一次性 tmpfs PostgreSQL 17 中完成空库升级、重复升级、关键约束/索引、真实外键级联、降级到 base、legacy 哨兵保留、无版本自有表漂移拒绝和再次升级，`test_alembic_migrations.py` 4 项通过，最终 revision 为 `20260713_0002`；临时容器已自动清理，持久化开发数据库复查仍未迁移。 +- 2026-07-14（剩余边界）：当前两条 revision 只覆盖 Expense Case、Business Event 和 Auth Session，完整 legacy schema baseline 及停止其余运行时 DDL 仍未完成；本轮受影响服务回归 46 项通过，既有员工目录历史部门归一化用例仍单独失败，未混入本次迁移安全范围。 diff --git a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md index 6d9d901..a0fa87d 100644 --- a/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md +++ b/document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md @@ -1,6 +1,6 @@ # AI 费用闭环与价值证明 开发 TODO -更新时间：2026-07-13 +更新时间：2026-07-14 文档路径：document/development/2026-07-13/feature/ai-expense-closed-loop-and-value-proof/TODO.md @@ -58,9 +58,12 @@ 证据：`expense_case.py`、`test_expense_case_endpoints.py`；用户态响应只保留安全流程摘要，关联资源 ID 和未知嵌套 payload 被递归过滤，本人、当前审批人、财务、管理员、无权限及跨租户边界测试通过。 - [ ] [CONCEPT: 权限与安全] 为 Qdrant collection/namespace、对象存储前缀和缓存键补齐租户隔离回归测试。 - [ ] [CONCEPT: 数据与契约] 建立 Alembic baseline 和正式迁移链，停止请求路径运行 DDL。 +- [x] [CONCEPT: 兼容策略] 集中 migration-owned 表所有权并在标准启动迁移前执行只读漂移预检，禁止 legacy bootstrap 越权建表。 + 证据：`schema_ownership.py`、`migration_preflight.py`、`server_start.sh`、`test_migration_preflight.py`、`test_schema_ownership.py`；无版本自有表、缺表、多表、未知/多 revision 均 fail-fast，不自动 stamp 或修改数据库。 - [x] [CONCEPT: 费用领域与编排] 新增 `ExpenseCaseService` 和费用事件查询接口，保持编排与具体职责分离。 证据：`server/src/app/services/expense_cases.py`、`server/src/app/api/v1/endpoints/expense_cases.py`、`GET /api/v1/expense-cases/by-claim/{claim_id}`；容器 OpenAPI 校验通过。 -- [ ] [CONCEPT: 数据与契约] 新增 `expense_cases`、`expense_case_links` 和 `business_events` 表及迁移。 +- [x] [CONCEPT: 数据与契约] 新增 `expense_cases`、`expense_case_links` 和 `business_events` 表及迁移。 + 证据：`20260713_0001_expense_case_business_events.py`；一次性 PostgreSQL 17 已验证表、唯一约束、复合索引、外键级联、降级和再次升级。 - [ ] [CONCEPT: 业务事件与 AI 决策] 新增 `ai_decisions`、`ai_decision_feedback`、`workflow_outcomes` 表及迁移。 - [ ] [CONCEPT: 数据与契约] 为申请、票据、草稿、提交、退回、审批、付款和归档接入统一 correlation ID。 - [ ] [CONCEPT: 业务事件与 AI 决策] 建立事务 Outbox：申请、提交、退回、审批、支付和入账状态与事件同事务提交，消费端按事件 ID 幂等处理。 @@ -73,7 +76,7 @@ - [x] [CONCEPT: 权限与安全] 将 AI 申请预览快速入口的用户、租户、角色与管理员身份强制绑定到服务端会话，拒绝请求体伪造身份编辑他人申请。 证据：`reimbursements.py`、`test_reimbursement_endpoints.py`；对抗用例修复前返回 200，修复后返回 400，且目标申请和费用事件保持不变。 - [x] [CONCEPT: 兼容策略] 建立迁移桥接：服务启动先执行 Alembic，旧 metadata bootstrap 排除 migration-owned 表。 - 证据：`server_start.sh`、`agent_foundation.py`、`20260713_0001_expense_case_business_events.py`；容器 `sh -n`、离线 upgrade/downgrade SQL 和 legacy bootstrap 排除测试通过。 + 证据：`server_start.sh`、`schema_ownership.py`、`migration_preflight.py`、`20260713_0001_expense_case_business_events.py`、`20260713_0002_auth_sessions.py`；容器 Shell/静态检查及一次性 PostgreSQL 完整 upgrade/downgrade/re-upgrade 通过。 - [ ] [CONCEPT: 兼容策略] 正式切换前以 shadow 事件校验现有 `ExpenseClaim` 映射；切换后禁止关键事件可丢弃写入。 - [ ] [CONCEPT: 兼容策略] 制定旧 `ReimbursementRequest` 只读兼容、迁移和停止新增编排的计划。 - [ ] [CONCEPT: 兼容策略] 把新增审批、付款、归档和关系事件移出 `risk_flags_json`，保留旧数据读取兼容。 @@ -155,6 +158,8 @@ - [x] [CONCEPT: 测试方案] 为 Expense Case GET 接口补充 owner、审批人、财务、管理员、无权限用户和整 Case 关联事件可见范围的 HTTP 权限测试。 证据：`test_expense_case_endpoints.py` 容器内 8 项通过，覆盖跨租户、无 Case、申请与报销关联摘要以及内部字段递归过滤。 - [ ] [CONCEPT: 测试方案] 为 Alembic baseline、升级、旧数据迁移和回滚边界补充 Postgres 集成测试。 +- [x] [CONCEPT: 测试方案] 为当前 migration-owned schema 切片补充一次性 PostgreSQL 集成测试和危险 URL 防误连门禁。 + 证据：`test_alembic_migrations.py` 默认无显式 URL 时跳过，主机和库名必须带 disposable 标记；tmpfs PostgreSQL 17 中 4 项通过，覆盖空库升级、重复升级、关键约束/索引、外键级联、base 降级、legacy 哨兵保留、漂移拒绝和再次升级；持久化开发库未被修改。完整 legacy baseline 仍保留在上一条未完成项中。 - [ ] [CONCEPT: 测试方案] 为连接器幂等、重试、回执、失败恢复、重复付款和对账补充测试。 - [ ] [CONCEPT: 测试方案] 跑通申请 → 票据 → 报销 → 预审 → 审批 → 付款 → 入账 → 归档端到端。 - [ ] [CONCEPT: 测试方案] 跑通 AI 建议 → 用户修改 → 工作流结果 → 记忆激活 → 下次建议变化闭环。 diff --git a/document/development/2026-07-14/dev-logs/bugs/migration-owned-table-bootstrap-drift.md b/document/development/2026-07-14/dev-logs/bugs/migration-owned-table-bootstrap-drift.md new file mode 100644 index 0000000..d4757e9 --- /dev/null +++ b/document/development/2026-07-14/dev-logs/bugs/migration-owned-table-bootstrap-drift.md @@ -0,0 +1,12 @@ +# 迁移自有表被运行时 bootstrap 越权创建 + +日期：2026-07-14 +文档路径：document/development/2026-07-14/dev-logs/bugs/migration-owned-table-bootstrap-drift.md + +## 修复记录 +- 09:21：记录 bug 修复：迁移自有表被运行时 bootstrap 越权创建。（bug-log:1347366b） + - Git 提交检查：fetch 成功；upstream `origin/main`；upstream 新提交：未发现；本地 ahead 提交：1347366b (HEAD -> main) feat(expenses): secure timeline and draft events；22669a90 feat(expenses): show unified expense event timeline；a616b30c fix(expenses): unify AI application submission transaction；653eda05 feat(auth): add opaque bearer sessions；661990b2 feat(expenses): add transactional expense case events。 + - 修改：新增 `schema_ownership.py` 统一四张 migration-owned 表的所有权边界，把 Agent Foundation、员工、预算、设置、系统看板、数字员工看板和演示数据初始化中的全量 `create_all` 改为只创建 legacy 表；新增 `migration_preflight.py`，在启动迁移前只读核对 Alembic revision 与自有表集合，遇到无版本表、缺表、多表、未知版本或多版本时直接拒绝，不自动 stamp 或修复。 + - 操作：调整 `server_start.sh` 的启动顺序和 `alembic.ini` 的脚本绝对定位方式；在无端口、无持久卷、tmpfs 数据目录的一次性 PostgreSQL 17 容器中执行空库升级、重复升级、降级到 base、漂移拦截和再次升级，结束后自动删除临时容器；没有执行持久化开发数据库迁移。 + - 验证：迁移与预检定向测试 `19 passed, 1 skipped`；一次性 PostgreSQL 真实迁移测试 `4 passed`，最终 revision 为 `20260713_0002`，关键唯一约束、复合索引和外键级联均通过，legacy 哨兵跨降级/再升级保留；受影响服务回归 `46 passed, 1 failed`，唯一失败为既有员工目录历史部门归一化用例，单独运行同样失败，与本次建表 helper 无关；新增文件 Ruff 全规则、旧服务 F/I/B/UP、Shell 语法、Alembic heads 和 `git diff --check` 均通过。 + - 影响：运行时 bootstrap 不再可能越权重建 Alembic 管理的表，启动会在可能破坏数据前暴露版本/表漂移；完整 legacy schema baseline 仍未建立，后续迁移旧表时必须继续小步验证。 diff --git a/server/README.md b/server/README.md index 8e105ce..02e1bf8 100644 --- a/server/README.md +++ b/server/README.md @@ -58,14 +58,33 @@ pip install -e .[dev] copy ..\\.env.example ..\\.env ``` -3. 启动服务 +3. 使用标准入口启动服务 ```bash -uvicorn app.main:app --reload --app-dir src +cd .. +./start.sh server ``` +标准入口在启动新的 FastAPI 进程前会自动执行 `alembic upgrade head`，迁移成功后才会 +启动 Uvicorn。`./start.sh all` 在需要启动后端时也会复用同一流程。不要把直接运行 +`uvicorn` 当作标准启动方式；手工调试 Uvicorn 时，需要先自行完成迁移。 + ## 迁移 ```bash -alembic upgrade head +cd server +alembic -c alembic.ini upgrade head ``` + +一次性 PostgreSQL 迁移测试默认跳过，只有显式提供 +`MIGRATION_TEST_DATABASE_URL` 时才会执行。为避免误操作开发库，测试会同时要求主机名 +和数据库名使用 `migration-probe` 或 `disposable-probe` 安全前缀，并要求数据库初始为空。 + +```bash +cd server +MIGRATION_TEST_DATABASE_URL='postgresql+psycopg://migration_probe:migration_probe_pw@x-financial-migration-probe-123:5432/migration_probe' \ + pytest -q tests/test_alembic_migrations.py +``` + +该测试覆盖首次升级、重复升级、关键表/约束/索引、外键级联、降级到 `base`、无关旧表 +及哨兵数据保留，以及再次升级。请只把它指向无持久卷的一次性 PostgreSQL 容器。 diff --git a/server/alembic.ini b/server/alembic.ini index a66ced1..f597ce4 100644 --- a/server/alembic.ini +++ b/server/alembic.ini @@ -1,6 +1,7 @@ [alembic] -script_location = alembic +script_location = %(here)s/alembic prepend_sys_path = . +path_separator = os sqlalchemy.url = postgresql+psycopg://postgres:postgres@127.0.0.1:5432/x_financial [loggers] diff --git a/server/server_start.sh b/server/server_start.sh index f34d2e7..e9f262a 100755 --- a/server/server_start.sh +++ b/server/server_start.sh @@ -379,6 +379,9 @@ ensure_dependencies() { } run_database_migrations() { + info "Checking database migration state..." + PYTHONPATH="$SCRIPT_DIR/src${PYTHONPATH:+:$PYTHONPATH}" \ + "$PYTHON_BIN" -m app.db.migration_preflight info "Applying database migrations..." "$PYTHON_BIN" -m alembic -c "$SCRIPT_DIR/alembic.ini" upgrade head info "Database migrations are up to date." diff --git a/server/src/app/db/migration_preflight.py b/server/src/app/db/migration_preflight.py new file mode 100644 index 0000000..954331c --- /dev/null +++ b/server/src/app/db/migration_preflight.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import sys +from dataclasses import dataclass + +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.engine import Connection, Engine +from sqlalchemy.exc import SQLAlchemyError + +from app.core.config import get_settings +from app.db.schema_ownership import MIGRATION_OWNED_TABLES + +# 迁移脚本新增或删除自有表时，必须同步更新该映射并补充对应测试。 +MIGRATION_OWNED_TABLES_BY_REVISION: dict[str, frozenset[str]] = { + "20260713_0001": frozenset( + { + "expense_cases", + "expense_case_links", + "business_events", + } + ), + "20260713_0002": frozenset( + { + "expense_cases", + "expense_case_links", + "business_events", + "auth_sessions", + } + ), +} +if MIGRATION_OWNED_TABLES_BY_REVISION["20260713_0002"] != MIGRATION_OWNED_TABLES: + raise RuntimeError("latest Alembic revision must own the centralized migration table set") + + +class MigrationPreflightError(RuntimeError): + """Raised when the database schema cannot be safely advanced by Alembic.""" + + +@dataclass(frozen=True) +class MigrationPreflightState: + revision: str | None + owned_tables: frozenset[str] + + +def _format_tables(table_names: frozenset[str]) -> str: + return ", ".join(sorted(table_names)) or "none" + + +def _validate_connection(connection: Connection) -> MigrationPreflightState: + table_names = frozenset(inspect(connection).get_table_names()) + owned_tables = table_names & MIGRATION_OWNED_TABLES + + if "alembic_version" not in table_names: + if owned_tables: + raise MigrationPreflightError( + "unversioned database contains migration-owned tables " + f"({_format_tables(owned_tables)}); refusing to guess, stamp, or repair" + ) + return MigrationPreflightState(revision=None, owned_tables=owned_tables) + + revisions = tuple( + str(revision) + for revision in connection.execute( + text("SELECT version_num FROM alembic_version") + ).scalars() + ) + if not revisions: + if owned_tables: + raise MigrationPreflightError( + "alembic_version has no recorded revision but migration-owned tables exist " + f"({_format_tables(owned_tables)}); refusing to guess, stamp, or repair" + ) + return MigrationPreflightState(revision=None, owned_tables=owned_tables) + + if len(revisions) > 1: + raise MigrationPreflightError( + "multiple Alembic revisions are recorded " + f"({', '.join(sorted(revisions))}); branch state is unsupported" + ) + + revision = revisions[0] + expected_tables = MIGRATION_OWNED_TABLES_BY_REVISION.get(revision) + if expected_tables is None: + raise MigrationPreflightError( + f"unknown Alembic revision {revision!r}; refusing to run migrations" + ) + + if owned_tables != expected_tables: + missing_tables = expected_tables - owned_tables + unexpected_tables = owned_tables - expected_tables + raise MigrationPreflightError( + f"migration-owned table set does not match revision {revision}: " + f"missing={_format_tables(missing_tables)}; " + f"unexpected={_format_tables(unexpected_tables)}" + ) + + return MigrationPreflightState(revision=revision, owned_tables=owned_tables) + + +def validate_migration_state(bind: Engine | Connection) -> MigrationPreflightState: + """Inspect the database without changing its schema or Alembic revision state.""" + + if isinstance(bind, Engine): + with bind.connect() as connection: + return _validate_connection(connection) + return _validate_connection(bind) + + +def main() -> int: + settings = get_settings() + engine = create_engine(settings.resolved_database_url, pool_pre_ping=True) + try: + state = validate_migration_state(engine) + except (MigrationPreflightError, SQLAlchemyError) as exc: + print(f"Database migration preflight failed: {exc}", file=sys.stderr) + return 1 + finally: + engine.dispose() + + revision = state.revision or "unversioned/base" + print( + "Database migration preflight passed: " + f"revision={revision}; owned_tables={_format_tables(state.owned_tables)}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/src/app/db/schema_ownership.py b/server/src/app/db/schema_ownership.py new file mode 100644 index 0000000..875518d --- /dev/null +++ b/server/src/app/db/schema_ownership.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +from sqlalchemy.engine import Connection, Engine + +from app.db.base import Base + +MIGRATION_OWNED_TABLES: frozenset[str] = frozenset( + { + "auth_sessions", + "expense_cases", + "expense_case_links", + "business_events", + } +) + + +def create_legacy_schema(bind: Engine | Connection) -> None: + """创建仍由旧 bootstrap 管理的表，不越过 Alembic 的表所有权边界。""" + legacy_tables = [ + table for table in Base.metadata.sorted_tables if table.name not in MIGRATION_OWNED_TABLES + ] + Base.metadata.create_all(bind=bind, tables=legacy_tables) diff --git a/server/src/app/services/agent_foundation.py b/server/src/app/services/agent_foundation.py index 1755bcd..08c1c99 100644 --- a/server/src/app/services/agent_foundation.py +++ b/server/src/app/services/agent_foundation.py @@ -2,31 +2,16 @@ from __future__ import annotations import threading -from sqlalchemy import inspect, select, text +from sqlalchemy import inspect, text from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.logging import get_logger -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.db.session import get_session_factory -from app.models.agent_asset import AgentAsset from app.services.agent_foundation_asset_helpers import AgentFoundationAssetHelperMixin from app.services.agent_foundation_asset_seed import AgentFoundationAssetSeedMixin from app.services.agent_foundation_asset_topup import AgentFoundationAssetTopUpMixin -from app.services.agent_foundation_constants import ( - ATTACHMENT_RULE_ASSET_CODE, - ATTACHMENT_RULE_RUNTIME_CONFIG, - COMPANY_COMMUNICATION_RULE_SCENARIO_JSON, - COMPANY_COMMUNICATION_RULE_VERSION, - COMPANY_TRAVEL_RULE_SCENARIO_JSON, - COMPANY_TRAVEL_RULE_VERSION, - DEMO_EXPENSE_CLAIM_SIGNATURES, - DEMO_PAYABLE_SIGNATURES, - DEMO_RECEIVABLE_SIGNATURES, - LEGACY_RULE_CODES, - PLATFORM_DESTINATION_LOCATION_RULE_CODE, - PLATFORM_DESTINATION_LOCATION_RULE_FILENAME, -) from app.services.agent_foundation_digital_employee_tasks import ( AgentFoundationDigitalEmployeeTaskMixin, ) @@ -38,12 +23,6 @@ from app.services.agent_foundation_spreadsheets import AgentFoundationSpreadshee logger = get_logger("app.services.agent_foundation") _foundation_ready_lock = threading.RLock() _foundation_ready_keys: set[str] = set() -MIGRATION_OWNED_TABLES = { - "auth_sessions", - "expense_cases", - "expense_case_links", - "business_events", -} def prepare_agent_foundation() -> None: @@ -83,12 +62,7 @@ class AgentFoundationService( def _prepare_foundation(self) -> None: try: - legacy_bootstrap_tables = [ - table - for table in Base.metadata.sorted_tables - if table.name not in MIGRATION_OWNED_TABLES - ] - Base.metadata.create_all(bind=self.db.get_bind(), tables=legacy_bootstrap_tables) + create_legacy_schema(self.db.get_bind()) self._ensure_agent_asset_schema() self._ensure_financial_record_schema() self._seed_agent_assets() diff --git a/server/src/app/services/budget.py b/server/src/app/services/budget.py index 9284b6e..833b2b1 100644 --- a/server/src/app/services/budget.py +++ b/server/src/app/services/budget.py @@ -7,7 +7,7 @@ from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.models.budget import BudgetAllocation, BudgetReservation from app.models.financial_record import ExpenseClaim from app.schemas.budget import ( @@ -32,7 +32,7 @@ class BudgetService(BudgetPaginationMixin, BudgetSupportMixin): def ensure_budget_ready(self) -> None: # 复用当前 Session 连接，避免在业务事务中通过 Engine 隐式提交已 flush 的单据。 - Base.metadata.create_all(bind=self.db.connection()) + create_legacy_schema(self.db.connection()) exists = self.db.scalar(select(BudgetAllocation.id).limit(1)) if exists: return diff --git a/server/src/app/services/demo_company_simulation_seed.py b/server/src/app/services/demo_company_simulation_seed.py index 56c47a7..dcaa582 100644 --- a/server/src/app/services/demo_company_simulation_seed.py +++ b/server/src/app/services/demo_company_simulation_seed.py @@ -10,7 +10,7 @@ from sqlalchemy import or_, select from sqlalchemy.orm import Session, selectinload from app.core.security import hash_password -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.models.budget import BudgetAllocation, BudgetReservation, BudgetTransaction from app.models.employee import Employee from app.models.financial_record import ExpenseClaim, ExpenseClaimItem @@ -78,7 +78,7 @@ class HalfYearExpenseSimulationSeeder: return self._run(apply=True) def _run(self, *, apply: bool) -> SimulationSummary: - Base.metadata.create_all(bind=self.db.get_bind()) + create_legacy_schema(self.db.get_bind()) departments = self._department_refs(apply=apply) current_employee_count = self._employee_count() planned_employees = self._build_new_employee_refs(departments, current_employee_count) diff --git a/server/src/app/services/digital_employee_dashboard.py b/server/src/app/services/digital_employee_dashboard.py index 567022e..d07ca6a 100644 --- a/server/src/app/services/digital_employee_dashboard.py +++ b/server/src/app/services/digital_employee_dashboard.py @@ -7,7 +7,7 @@ from sqlalchemy import or_, select from sqlalchemy.orm import Session, selectinload from app.core.agent_enums import AgentName, AgentRunSource -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.models.agent_run import AgentRun, AgentToolCall from app.schemas.digital_employee_dashboard import DigitalEmployeeDashboardRead @@ -186,7 +186,7 @@ class DigitalEmployeeDashboardService: ) def _ensure_storage_ready(self) -> None: - Base.metadata.create_all(bind=self.db.get_bind()) + create_legacy_schema(self.db.get_bind()) def _fetch_runs(self, *, start: datetime, limit: int) -> list[AgentRun]: stmt = ( diff --git a/server/src/app/services/employee.py b/server/src/app/services/employee.py index dcb6bd8..05fecc3 100644 --- a/server/src/app/services/employee.py +++ b/server/src/app/services/employee.py @@ -1,8 +1,8 @@ from __future__ import annotations -from collections import Counter -from datetime import UTC, date, datetime import threading +from collections import Counter +from datetime import UTC, datetime from typing import Any from sqlalchemy import select @@ -11,7 +11,7 @@ from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.logging import get_logger from app.core.security import hash_password -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.db.session import get_session_factory from app.models.employee import Employee from app.models.employee_change_log import EmployeeChangeLog @@ -28,11 +28,9 @@ from app.schemas.employee import ( EmployeeStatusSummaryRead, EmployeeUpdate, ) -from app.services.employee_import import EmployeeImportCoordinator from app.services.employee_bank_info import apply_default_bank_info +from app.services.employee_import import EmployeeImportCoordinator from app.services.employee_schema import ensure_employee_schema -from app.services.employee_serialization import serialize_employee -from app.services.employee_spreadsheet import build_import_template_bytes from app.services.employee_seed import ( CANONICAL_DEPARTMENT_CODES, EMPLOYEE_DEFINITIONS, @@ -44,6 +42,8 @@ from app.services.employee_seed import ( ROLE_PERMISSION_MAP, normalize_organization_unit_code, ) +from app.services.employee_serialization import serialize_employee +from app.services.employee_spreadsheet import build_import_template_bytes from app.services.employee_time import ( format_date, format_datetime, @@ -108,7 +108,7 @@ class EmployeeService: def _ensure_directory_ready_uncached(self) -> None: try: - Base.metadata.create_all(bind=self.db.get_bind()) + create_legacy_schema(self.db.get_bind()) ensure_employee_schema(self.db) self._prune_extra_seed_employees() self._seed_roles() diff --git a/server/src/app/services/settings.py b/server/src/app/services/settings.py index a4e3c05..3b02c8c 100644 --- a/server/src/app/services/settings.py +++ b/server/src/app/services/settings.py @@ -8,16 +8,20 @@ from datetime import datetime from sqlalchemy import inspect, text from sqlalchemy.orm import Session -from app.core.admin_secret import legacy_admin_secret_to_password_hash, read_admin_secret, verify_admin_secret +from app.core.admin_secret import ( + legacy_admin_secret_to_password_hash, + read_admin_secret, + verify_admin_secret, +) from app.core.config import get_settings from app.core.secret_box import decrypt_secret, encrypt_secret from app.core.security import hash_password, verify_password -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.db.session import get_session_factory +from app.models.hermes_config import HermesTaskConfig from app.models.system_model_setting import SystemModelSetting from app.models.system_setting import SystemSetting from app.models.system_setting_secret import SystemSettingSecret -from app.models.hermes_config import HermesTaskConfig from app.repositories.settings import SETTINGS_ROW_ID, SettingsRepository from app.schemas.settings import SettingsRead, SettingsWrite from app.services.hermes_sync import ( @@ -161,7 +165,7 @@ class SettingsService: if cache_key not in self._schema_ready_keys: with self._schema_ready_lock: if cache_key not in self._schema_ready_keys: - Base.metadata.create_all(bind=self.db.get_bind()) + create_legacy_schema(self.db.get_bind()) self._ensure_settings_schema() self._schema_ready_keys.add(cache_key) diff --git a/server/src/app/services/system_dashboard.py b/server/src/app/services/system_dashboard.py index 83e9e70..0931c2e 100644 --- a/server/src/app/services/system_dashboard.py +++ b/server/src/app/services/system_dashboard.py @@ -8,7 +8,7 @@ from typing import Any from sqlalchemy import or_, select from sqlalchemy.orm import Session -from app.db.base import Base +from app.db.schema_ownership import create_legacy_schema from app.models.agent_feedback import AgentOperationFeedback from app.models.agent_run import AgentRun, AgentToolCall from app.models.user_session_metric import UserSessionMetric @@ -143,7 +143,7 @@ class SystemDashboardService: ) def _ensure_storage_ready(self) -> None: - Base.metadata.create_all(bind=self.db.get_bind()) + create_legacy_schema(self.db.get_bind()) def _fetch_runs(self, start: datetime, *, before: datetime | None = None) -> list[_DashboardRun]: stmt = ( diff --git a/server/tests/test_alembic_migrations.py b/server/tests/test_alembic_migrations.py new file mode 100644 index 0000000..4afa09e --- /dev/null +++ b/server/tests/test_alembic_migrations.py @@ -0,0 +1,364 @@ +from __future__ import annotations + +import os +import re +from collections.abc import Iterator +from pathlib import Path + +import pytest +from alembic.config import Config +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.engine import Engine, make_url +from sqlalchemy.pool import NullPool + +from alembic import command +from app.core.config import get_settings +from app.db.migration_preflight import MigrationPreflightError, validate_migration_state +from app.db.schema_ownership import MIGRATION_OWNED_TABLES + +MIGRATION_TEST_DATABASE_URL = os.getenv("MIGRATION_TEST_DATABASE_URL", "").strip() +LEGACY_PROBE_TABLE = "legacy_migration_probe_records" +HEAD_REVISION = "20260713_0002" +SERVER_DIR = Path(__file__).resolve().parents[1] +ALEMBIC_INI_PATH = SERVER_DIR / "alembic.ini" + +def _normalize_probe_component(value: str) -> str: + return re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") + + +def _is_disposable_probe_host(value: str) -> bool: + markers = ("migration-probe", "disposable-probe") + return value in markers or any( + value.startswith(f"{marker}-") or value.startswith(f"x-financial-{marker}-") + for marker in markers + ) + + +def _is_disposable_probe_database(value: str) -> bool: + markers = ("migration-probe", "disposable-probe") + return value in markers or any(value.startswith(f"{marker}-") for marker in markers) + + +def _require_disposable_probe_url(raw_url: str) -> str: + try: + parsed = make_url(raw_url) + except Exception as exc: # pragma: no cover - SQLAlchemy 提供具体解析异常 + raise RuntimeError("MIGRATION_TEST_DATABASE_URL 不是有效的数据库 URL") from exc + + if parsed.get_backend_name() != "postgresql": + raise RuntimeError("迁移测试只允许连接 PostgreSQL 一次性数据库") + + host = _normalize_probe_component(parsed.host or "") + database = _normalize_probe_component(parsed.database or "") + if not _is_disposable_probe_host(host): + raise RuntimeError( + "迁移测试数据库主机名必须使用 migration-probe 或 disposable-probe 前缀" + ) + if not _is_disposable_probe_database(database): + raise RuntimeError( + "迁移测试数据库名必须使用 migration-probe 或 disposable-probe 前缀" + ) + + return raw_url + + +@pytest.fixture(scope="module") +def migration_database_url() -> Iterator[str]: + if not MIGRATION_TEST_DATABASE_URL: + pytest.skip("仅在显式配置 MIGRATION_TEST_DATABASE_URL 时运行一次性 PostgreSQL 迁移测试") + database_url = _require_disposable_probe_url(MIGRATION_TEST_DATABASE_URL) + previous_database_url = os.environ.get("DATABASE_URL") + os.environ["DATABASE_URL"] = database_url + get_settings.cache_clear() + + try: + resolved_url = get_settings().resolved_database_url + if make_url(resolved_url) != make_url(database_url): + raise RuntimeError("运行时数据库 URL 未解析到 MIGRATION_TEST_DATABASE_URL") + yield database_url + finally: + if previous_database_url is None: + os.environ.pop("DATABASE_URL", None) + else: + os.environ["DATABASE_URL"] = previous_database_url + get_settings.cache_clear() + + +def _alembic_config(database_url: str) -> Config: + config = Config(str(ALEMBIC_INI_PATH)) + config.set_main_option("sqlalchemy.url", database_url.replace("%", "%%")) + return config + + +def _upgrade_head(database_url: str) -> None: + get_settings.cache_clear() + command.upgrade(_alembic_config(database_url), "head") + + +def _downgrade_base(database_url: str) -> None: + get_settings.cache_clear() + command.downgrade(_alembic_config(database_url), "base") + + +def _table_names(engine: Engine) -> set[str]: + return set(inspect(engine).get_table_names(schema="public")) + + +def _assert_unique_constraint( + engine: Engine, + table_name: str, + constraint_name: str, + expected_columns: tuple[str, ...], +) -> None: + constraints = { + str(item["name"]): tuple(item["column_names"]) + for item in inspect(engine).get_unique_constraints(table_name, schema="public") + } + assert constraints.get(constraint_name) == expected_columns + + +def _assert_indexes( + engine: Engine, + table_name: str, + expected_indexes: dict[str, tuple[str, ...]], +) -> None: + indexes = { + str(item["name"]): tuple(item["column_names"]) + for item in inspect(engine).get_indexes(table_name, schema="public") + } + for index_name, expected_columns in expected_indexes.items(): + assert indexes.get(index_name) == expected_columns + + +def _assert_cascade_foreign_key(engine: Engine, table_name: str) -> None: + foreign_keys = inspect(engine).get_foreign_keys(table_name, schema="public") + matching = [ + item + for item in foreign_keys + if item["constrained_columns"] == ["expense_case_id"] + and item["referred_table"] == "expense_cases" + and item["referred_columns"] == ["id"] + ] + assert len(matching) == 1 + assert str(matching[0].get("options", {}).get("ondelete", "")).upper() == "CASCADE" + + +def _assert_head_schema(engine: Engine) -> None: + names = _table_names(engine) + assert MIGRATION_OWNED_TABLES.issubset(names) + assert "alembic_version" in names + + with engine.connect() as connection: + assert connection.scalar(text("SELECT version_num FROM alembic_version")) == HEAD_REVISION + + _assert_unique_constraint( + engine, + "expense_cases", + "uq_expense_cases_tenant_case_no", + ("tenant_id", "case_no"), + ) + _assert_unique_constraint( + engine, + "expense_case_links", + "uq_expense_case_links_resource", + ("resource_type", "resource_id"), + ) + _assert_unique_constraint( + engine, + "business_events", + "uq_business_event_idempotency", + ("tenant_id", "aggregate_type", "aggregate_id", "event_type", "idempotency_key"), + ) + _assert_unique_constraint( + engine, + "auth_sessions", + "uq_auth_sessions_token_hash", + ("token_hash",), + ) + + _assert_indexes( + engine, + "expense_cases", + { + "ix_expense_cases_tenant_stage": ("tenant_id", "current_stage"), + "ix_expense_cases_tenant_status": ("tenant_id", "status"), + }, + ) + _assert_indexes( + engine, + "expense_case_links", + {"ix_expense_case_links_tenant_case": ("tenant_id", "expense_case_id")}, + ) + _assert_indexes( + engine, + "business_events", + { + "ix_business_events_aggregate": ("aggregate_type", "aggregate_id"), + "ix_business_events_outbox": ("delivery_status", "occurred_at"), + "ix_business_events_tenant_case_time": ( + "tenant_id", + "expense_case_id", + "occurred_at", + ), + }, + ) + _assert_indexes( + engine, + "auth_sessions", + { + "ix_auth_sessions_principal_active": ( + "principal_type", + "revoked_at", + "expires_at", + ), + "ix_auth_sessions_tenant_username": ("tenant_id", "username"), + }, + ) + _assert_cascade_foreign_key(engine, "expense_case_links") + _assert_cascade_foreign_key(engine, "business_events") + + +def _assert_runtime_cascade(engine: Engine) -> None: + with engine.begin() as connection: + connection.execute( + text( + """ + INSERT INTO expense_cases ( + id, tenant_id, case_no, scene_code, title, current_stage, status + ) VALUES ( + 'migration-probe-case', 'migration-probe', 'CASE-MIGRATION-PROBE', + 'reimbursement', '迁移级联验证', 'claiming', 'active' + ) + """ + ) + ) + connection.execute( + text( + """ + INSERT INTO expense_case_links ( + id, tenant_id, expense_case_id, resource_type, resource_id, relation_type + ) VALUES ( + 'migration-probe-link', 'migration-probe', 'migration-probe-case', + 'expense_claim', 'migration-probe-claim', 'claim' + ) + """ + ) + ) + connection.execute( + text( + """ + INSERT INTO business_events ( + id, tenant_id, expense_case_id, aggregate_type, aggregate_id, + event_type, event_version, idempotency_key, correlation_id, + actor_id, actor_type, payload_json, delivery_status, delivery_attempts + ) VALUES ( + 'migration-probe-event', 'migration-probe', 'migration-probe-case', + 'expense_claim', 'migration-probe-claim', 'claim_draft_created', 1, + 'migration-probe-idempotency', 'migration-probe-correlation', + 'migration-probe-user', 'user', '{}', 'pending', 0 + ) + """ + ) + ) + connection.execute( + text("DELETE FROM expense_cases WHERE id = 'migration-probe-case'") + ) + assert connection.scalar( + text("SELECT COUNT(*) FROM expense_case_links WHERE id = 'migration-probe-link'") + ) == 0 + assert connection.scalar( + text("SELECT COUNT(*) FROM business_events WHERE id = 'migration-probe-event'") + ) == 0 + + +def _create_legacy_sentinel(engine: Engine) -> None: + with engine.begin() as connection: + connection.execute( + text( + f""" + CREATE TABLE {LEGACY_PROBE_TABLE} ( + id VARCHAR(64) PRIMARY KEY, + payload VARCHAR(255) NOT NULL + ) + """ + ) + ) + connection.execute( + text( + f""" + INSERT INTO {LEGACY_PROBE_TABLE} (id, payload) + VALUES ('legacy-sentinel', 'must-survive-migration-cycle') + """ + ) + ) + + +def _assert_legacy_sentinel(engine: Engine) -> None: + assert LEGACY_PROBE_TABLE in _table_names(engine) + with engine.connect() as connection: + payload = connection.scalar( + text( + f"SELECT payload FROM {LEGACY_PROBE_TABLE} " + "WHERE id = 'legacy-sentinel'" + ) + ) + assert payload == "must-survive-migration-cycle" + + +def _assert_base_schema(engine: Engine) -> None: + names = _table_names(engine) + assert not MIGRATION_OWNED_TABLES.intersection(names) + assert "alembic_version" in names + with engine.connect() as connection: + assert connection.scalar(text("SELECT COUNT(*) FROM alembic_version")) == 0 + + +@pytest.mark.parametrize( + "database_url", + [ + "sqlite+pysqlite:///:memory:", + "postgresql+psycopg://postgres:postgres@x-financial-local-postgres:5432/x_financial", + "postgresql+psycopg://probe:probe@migration-probe-123:5432/x_financial", + ], +) +def test_disposable_database_guard_rejects_unsafe_urls(database_url: str) -> None: + with pytest.raises(RuntimeError): + _require_disposable_probe_url(database_url) + + +def test_alembic_migration_cycle_on_disposable_postgres( + migration_database_url: str, +) -> None: + engine = create_engine(migration_database_url, poolclass=NullPool) + try: + assert _table_names(engine) == set(), "迁移测试必须从全新空库开始" + assert validate_migration_state(engine).revision is None + + _upgrade_head(migration_database_url) + _assert_head_schema(engine) + assert validate_migration_state(engine).revision == HEAD_REVISION + + _upgrade_head(migration_database_url) + _assert_head_schema(engine) + _assert_runtime_cascade(engine) + + _create_legacy_sentinel(engine) + _downgrade_base(migration_database_url) + + _assert_base_schema(engine) + _assert_legacy_sentinel(engine) + assert validate_migration_state(engine).revision is None + + with engine.begin() as connection: + connection.execute(text("CREATE TABLE expense_cases (id VARCHAR(36) PRIMARY KEY)")) + with pytest.raises(MigrationPreflightError, match="migration-owned tables exist"): + validate_migration_state(engine) + with engine.begin() as connection: + connection.execute(text("DROP TABLE expense_cases")) + assert validate_migration_state(engine).revision is None + + _upgrade_head(migration_database_url) + _assert_head_schema(engine) + _assert_legacy_sentinel(engine) + finally: + engine.dispose() diff --git a/server/tests/test_migration_preflight.py b/server/tests/test_migration_preflight.py new file mode 100644 index 0000000..3a9d8dd --- /dev/null +++ b/server/tests/test_migration_preflight.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest +from sqlalchemy import Column, Integer, MetaData, Table, create_engine, text +from sqlalchemy.engine import Engine + +from app.db.migration_preflight import ( + MIGRATION_OWNED_TABLES_BY_REVISION, + MigrationPreflightError, + validate_migration_state, +) + + +@pytest.fixture +def engine() -> Engine: + database = create_engine("sqlite+pysqlite:///:memory:") + try: + yield database + finally: + database.dispose() + + +def _create_tables(engine: Engine, table_names: set[str] | frozenset[str]) -> None: + metadata = MetaData() + for table_name in table_names: + Table(table_name, metadata, Column("id", Integer, primary_key=True)) + metadata.create_all(engine) + + +def _create_version_table(engine: Engine, *revisions: str) -> None: + with engine.begin() as connection: + connection.execute(text("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")) + for revision in revisions: + connection.execute( + text("INSERT INTO alembic_version (version_num) VALUES (:revision)"), + {"revision": revision}, + ) + + +def test_unversioned_database_without_migration_owned_tables_is_safe(engine: Engine) -> None: + state = validate_migration_state(engine) + + assert state.revision is None + assert state.owned_tables == frozenset() + + +@pytest.mark.parametrize( + "owned_table", + sorted(MIGRATION_OWNED_TABLES_BY_REVISION["20260713_0002"]), +) +def test_unversioned_database_with_any_migration_owned_table_is_rejected( + engine: Engine, + owned_table: str, +) -> None: + _create_tables(engine, {owned_table}) + + with pytest.raises(MigrationPreflightError, match="unversioned database contains"): + validate_migration_state(engine) + + +@pytest.mark.parametrize( + ("revision", "expected_tables"), + list(MIGRATION_OWNED_TABLES_BY_REVISION.items()), +) +def test_known_revision_requires_and_accepts_its_exact_owned_table_set( + engine: Engine, + revision: str, + expected_tables: frozenset[str], +) -> None: + _create_tables(engine, expected_tables) + _create_version_table(engine, revision) + + state = validate_migration_state(engine) + + assert state.revision == revision + assert state.owned_tables == expected_tables + + +@pytest.mark.parametrize( + ("revision", "actual_tables"), + [ + ("20260713_0001", frozenset({"expense_cases", "expense_case_links"})), + ("20260713_0001", MIGRATION_OWNED_TABLES_BY_REVISION["20260713_0002"]), + ( + "20260713_0002", + MIGRATION_OWNED_TABLES_BY_REVISION["20260713_0002"] - {"auth_sessions"}, + ), + ], +) +def test_known_revision_with_missing_or_unexpected_owned_tables_is_rejected( + engine: Engine, + revision: str, + actual_tables: frozenset[str], +) -> None: + _create_tables(engine, actual_tables) + _create_version_table(engine, revision) + + with pytest.raises(MigrationPreflightError, match="does not match revision"): + validate_migration_state(engine) + + +def test_unknown_revision_is_rejected(engine: Engine) -> None: + _create_version_table(engine, "20990101_unknown") + + with pytest.raises(MigrationPreflightError, match="unknown Alembic revision"): + validate_migration_state(engine) + + +def test_multiple_revisions_are_rejected(engine: Engine) -> None: + _create_version_table(engine, "20260713_0001", "20260713_0002") + + with pytest.raises(MigrationPreflightError, match="multiple Alembic revisions"): + validate_migration_state(engine) + + +def test_empty_version_table_is_safe_only_when_owned_tables_are_absent(engine: Engine) -> None: + _create_version_table(engine) + assert validate_migration_state(engine).revision is None + + _create_tables(engine, {"expense_cases"}) + with pytest.raises(MigrationPreflightError, match="no recorded revision"): + validate_migration_state(engine) + + +def test_server_start_runs_preflight_before_alembic_upgrade() -> None: + script_path = Path(__file__).resolve().parents[1] / "server_start.sh" + script = script_path.read_text(encoding="utf-8") + + preflight = '"$PYTHON_BIN" -m app.db.migration_preflight' + upgrade = '"$PYTHON_BIN" -m alembic -c "$SCRIPT_DIR/alembic.ini" upgrade head' + + assert 'PYTHONPATH="$SCRIPT_DIR/src${PYTHONPATH:+:$PYTHONPATH}"' in script + assert script.index(preflight) < script.index(upgrade) diff --git a/server/tests/test_schema_ownership.py b/server/tests/test_schema_ownership.py new file mode 100644 index 0000000..2f9db24 --- /dev/null +++ b/server/tests/test_schema_ownership.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from sqlalchemy import create_engine, inspect + +from app.db.schema_ownership import MIGRATION_OWNED_TABLES, create_legacy_schema + + +def test_create_legacy_schema_never_creates_migration_owned_tables() -> None: + engine = create_engine("sqlite+pysqlite:///:memory:") + try: + create_legacy_schema(engine) + + table_names = set(inspect(engine).get_table_names()) + finally: + engine.dispose() + + assert MIGRATION_OWNED_TABLES == frozenset( + { + "auth_sessions", + "business_events", + "expense_case_links", + "expense_cases", + } + ) + assert table_names + assert table_names.isdisjoint(MIGRATION_OWNED_TABLES)