"""助手提交的东西,到底变成了哪一行 —— 把「调用返回了」和「真的成了」分开。 `AssistantActionProposal.status = executed` 今天只证明一件事:**那次调用返回 了**。它不证明飞书审批过了、Provisioning 跑完了、Provider 真的生效了。而员工问 的从来只有后者。于是助手能说「已提交」,却答不了「现在到哪一步、最后成没成」。 这个迁移加的是中间那个指针:一次执行创建了**哪一行业务对象**。有了它,状态才 能回原领域去读 —— 而不是在这里再抄一份状态机。 两处刻意的形状: * `(resource_type, resource_id)` 是多态对,**没有外键**。它跨九张表、九个域, 多态对是诚实的形状。代价明确接受:数据库层拦不住悬空指针,所以解析器必须 容忍目标行已经不在,并如实说出来,而不是崩掉。 * `phase` 是**缓存**,给漏斗查询用,永远不是回答员工的那个答案。抄一份领域 状态过来就等于养第二套状态机,而一份写着 approved 的过期缓存比没有缓存更坏。 同时给 `assistant_action_proposals` 加三列尝试可见性。失败会被回落成 `pending` (刻意的:卡片还能按、员工能重试),代价是「按了但失败」和「从没按过」变得除了 一个没人统计的 JSON 之外**每一列都同形**。生产上 14 条提案全部读作未执行,而其 中**两条其实被按过** —— 都栽在模型无法预先满足的参数契约上(`create_email_alias` 的 `alias_suffix` 要裸域名、`submit_data_access_personal_grant` 的两个参数是 JSON Schema 表达不了的二选一),两条都只在执行时才炸。这三列就是那个差别, 可查询的形式。它们是计数器和时间戳,**不是第二个状态字段**。 Revision ID: 572 Revises: 571 """ from __future__ import annotations import sqlalchemy as sa from alembic import op from app.core.migration_helpers import id_type_for revision = "572" down_revision = "571" branch_labels = None depends_on = None LINKS = "assistant_object_links" PROPOSALS = "assistant_action_proposals" _NEW_PROPOSAL_COLUMNS = ( ( "attempt_count", sa.Column( "attempt_count", sa.Integer(), nullable=False, server_default="0", comment="确认按钮被真正按下并执行过几次。0 = 从没按过。", ), ), ( "last_attempted_at", sa.Column( "last_attempted_at", sa.DateTime(timezone=True), nullable=True, comment="最后一次执行尝试的时间,成功与否都记。", ), ), ( "last_error_code", sa.Column( "last_error_code", sa.String(64), nullable=True, comment=( "最后一次失败的异常类名。NULL 且 attempt_count>0 = 按过且成功;" "非 NULL = 按过但失败 —— 这两者过去在库里长得一模一样。" ), ), ), ( "link_miss_reason", sa.Column( "link_miss_reason", sa.String(32), nullable=True, comment=( "执行成功却没有链上业务对象的原因(no_locator / id_key_absent / …)。" "单独成列是为了能直接数出来 —— 埋在 JSON 里的契约缺口," "从漏斗上看跟「根本没人提交」一模一样。" ), ), ), ) def _has_column(conn: sa.engine.Connection, table: str, column: str) -> bool: # DATABASE()-qualified:不限定库名的 information_schema 探测会匹配到**任何** # schema 里的同名表,而这台机器上并排跑着上百个 helix_test_* 库。 return bool( conn.execute( sa.text( "SELECT COUNT(*) FROM information_schema.columns " "WHERE table_schema = DATABASE() " "AND table_name = :t AND column_name = :c", ), {"t": table, "c": column}, ).scalar() ) def upgrade() -> None: conn = op.get_bind() inspector = sa.inspect(conn) if inspector.has_table(PROPOSALS): for name, column in _NEW_PROPOSAL_COLUMNS: if not _has_column(conn, PROPOSALS, name): op.add_column(PROPOSALS, column) if not inspector.has_table(LINKS): op.create_table( LINKS, sa.Column("id", sa.CHAR(36), primary_key=True), # own PK, no parent to match sa.Column( "proposal_id", id_type_for(PROPOSALS), nullable=False, comment="哪一次确认执行产生了这个对象。", ), sa.Column( "conversation_id", id_type_for("assistant_conversations"), nullable=True, ), sa.Column( "service_request_id", id_type_for("support_service_requests"), nullable=True, ), sa.Column( "person_id", id_type_for("identity_persons"), nullable=False, comment=( "动作以谁的身份跑的。**不是授权** —— 它只是给解析器一个主体," "读目标行的权限每次重新按领域自己的规则推导。" ), ), sa.Column("action_key", sa.String(64), nullable=False), sa.Column( "source_tool", sa.String(64), nullable=True, comment=( "真正产出这个 payload 的 MCP 工具。派生动作与 action_key 相同;" "策展动作转调时不同(回购跑的是 submit_approval_request)。" "NULL = 策展动作直接调领域服务,没有经过工具。" ), ), sa.Column( "resource_type", sa.String(48), nullable=False, comment="employee_request / support_ticket / … 共九种。", ), sa.Column("resource_id", sa.String(64), nullable=False), sa.Column( "resolver_key", sa.String(48), nullable=False, comment="谁去读这一行的实时状态。", ), sa.Column( "resource_number", sa.String(64), nullable=True, comment="员工嘴里会说的那个号(REQ-1042)。没有号的域为 NULL。", ), sa.Column( "deep_link", sa.String(255), nullable=True, comment="服务端按模板拼的,模型从不参与 URL 的决定。", ), sa.Column( "covers_count", sa.Integer(), nullable=True, comment=( "一个 id 代表一批时的行数(云授权续期覆盖 N 条)。有了它," "投影就不会因为第一条成了而宣布整批完成。" ), ), sa.Column( "phase", sa.String(24), nullable=True, comment="最近一次投影出的阶段。**缓存**,不是答案的来源。", ), sa.Column("phase_detail", sa.JSON(), nullable=True), sa.Column("phase_at", sa.DateTime(timezone=True), nullable=True), sa.Column( "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.Column( "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now(), ), sa.ForeignKeyConstraint(["proposal_id"], [f"{PROPOSALS}.id"]), sa.ForeignKeyConstraint( ["conversation_id"], ["assistant_conversations.id"], ), sa.ForeignKeyConstraint( ["service_request_id"], ["support_service_requests.id"], ), sa.ForeignKeyConstraint(["person_id"], ["identity_persons.id"]), # 重放安全:重复的飞书回调、重试的 confirm、再跑一次的解析器,都必须 # 落回同一行,而不是长出第二行。 sa.UniqueConstraint( "proposal_id", "resource_type", "resource_id", name="uq_assistant_link_proposal_resource", ), ) op.create_index( "idx_assistant_link_person", LINKS, ["person_id", "created_at"], ) op.create_index( "idx_assistant_link_resource", LINKS, ["resource_type", "resource_id"], ) op.create_index("idx_assistant_link_request", LINKS, ["service_request_id"]) op.create_index("idx_assistant_link_phase", LINKS, ["phase", "phase_at"]) def downgrade() -> None: conn = op.get_bind() inspector = sa.inspect(conn) if inspector.has_table(LINKS): op.drop_table(LINKS) if inspector.has_table(PROPOSALS): for name, _column in reversed(_NEW_PROPOSAL_COLUMNS): if _has_column(conn, PROPOSALS, name): op.drop_column(PROPOSALS, name)