chenxin/hunjing
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
new
hunjing/backend/alembic/versions/140fe7181a6e_垂直切片_会话字段与会话树_运行_幂等记录.py
T
chenxin ba502297ac ✨ feat: 阶段 0 收尾——认证边界、请求上下文与垂直切片（WP-01 第 5-7 步） 
- 认证：Web Session（__Host- Cookie、HttpOnly/Secure/SameSite=Lax、CSPRNG opaque token 数据库只存哈希）、CSRF 双提交、Origin 校验、dev 登录/登出端点；未认证 401、缺 CSRF 403
- 中间件：Hunjing-Request-ID 回显/生成（request_ UUIDv4）+ W3C traceparent 校验透传；JSON 结构化日志与敏感头脱敏
- 迁移 2：conversation/branch/message/run/idempotency_record 五表（含检查约束与复合索引）；Session 补 token/csrf 哈希列
- 垂直切片：POST /v1/conversations/{id}/messages 按 Idempotency-Key 幂等（同键同体 200 重放、异体 409），同事务原子写 Message+Run+Audit+Outbox+幂等记录；ContentBlock text/markdown 判别联合（未知类型 422）
- SSE：GET /v1/runs/{id}/events 返回快照与终止状态事件 envelope
- 测试 26 项全绿：IDOR 跨域 404 fail-closed、失败注入回滚无孤儿（Run 崩溃后 message/run/audit/outbox/幂等全零）、SSE 事件序、auth 边界；ruff/mypy strict 干净
- 修复真实问题：asyncpg 跨事件循环（anyio/pytest-asyncio 双 loop、模块级 engine 改为 app 实例持有）、httpx 拒存 http 下 Secure Cookie
2026-09-10 10:13:45 +08:00

177 lines
8.3 KiB
Python
Raw Permalink Blame History

"""垂直切片：会话字段与会话树、运行、幂等记录
Revision ID: 140fe7181a6e
Revises: e16a40566b45
Create Date: 2026-09-10 10:06:24.354536
"""
from collections.abc import Sequence
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
from alembic import op
# revision identifiers, used by Alembic.
revision: str = '140fe7181a6e'
down_revision: str | Sequence[str] | None = 'e16a40566b45'
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('conversation',
sa.Column('id', sa.String(), nullable=False),
sa.Column('personal_domain_id', sa.String(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['personal_domain_id'], ['personal_domain.id'], name=op.f('fk_conversation_personal_domain'), ondelete='cascade'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_conversation'))
)
op.create_index(op.f('ix_conversation_personal_domain_id'), 'conversation', ['personal_domain_id'], unique=False)
op.create_table('idempotency_record',
sa.Column('key', sa.String(), nullable=False),
sa.Column('owner_domain_id', sa.String(), nullable=False),
sa.Column('route', sa.String(), nullable=False),
sa.Column('request_hash', sa.String(), nullable=False),
sa.Column('response_status', sa.Integer(), nullable=False),
sa.Column('resource_ids', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['owner_domain_id'], ['personal_domain.id'], name=op.f('fk_idempotency_record_personal_domain'), ondelete='cascade'),
sa.PrimaryKeyConstraint('key', name=op.f('pk_idempotency_record'))
)
op.create_index('uq_idempotency_owner_route_key', 'idempotency_record', ['owner_domain_id', 'route', 'key'], unique=True)
op.create_table('branch',
sa.Column('id', sa.String(), nullable=False),
sa.Column('conversation_id', sa.String(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_branch_conversation'), ondelete='cascade'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_branch'))
)
op.create_index(op.f('ix_branch_conversation_id'), 'branch', ['conversation_id'], unique=False)
op.create_table('run',
sa.Column('id', sa.String(), nullable=False),
sa.Column('personal_domain_id', sa.String(), nullable=False),
sa.Column('conversation_id', sa.String(), nullable=True),
sa.Column('status', sa.String(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True),
sa.CheckConstraint("status IN ('queued', 'running', 'waiting_approval', 'completed', 'failed', 'cancelled')", name=op.f('ck_run_status_values')),
sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_run_conversation'), ondelete='cascade'),
sa.ForeignKeyConstraint(['personal_domain_id'], ['personal_domain.id'], name=op.f('fk_run_personal_domain'), ondelete='cascade'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_run'))
)
op.create_index(op.f('ix_run_conversation_id'), 'run', ['conversation_id'], unique=False)
op.create_index('ix_run_domain_status', 'run', ['personal_domain_id', 'status'], unique=False)
op.create_index(op.f('ix_run_personal_domain_id'), 'run', ['personal_domain_id'], unique=False)
op.create_table('message',
sa.Column('id', sa.String(), nullable=False),
sa.Column('conversation_id', sa.String(), nullable=False),
sa.Column('branch_id', sa.String(), nullable=False),
sa.Column('role', sa.String(), nullable=False),
sa.Column('content', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.CheckConstraint("role IN ('user', 'assistant', 'system', 'tool')", name=op.f('ck_message_role_values')),
sa.ForeignKeyConstraint(['branch_id'], ['branch.id'], name=op.f('fk_message_branch'), ondelete='cascade'),
sa.ForeignKeyConstraint(['conversation_id'], ['conversation.id'], name=op.f('fk_message_conversation'), ondelete='cascade'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_message'))
)
op.create_index(op.f('ix_message_branch_id'), 'message', ['branch_id'], unique=False)
op.create_index(op.f('ix_message_conversation_id'), 'message', ['conversation_id'], unique=False)
op.drop_table_comment(
'audit_event',
existing_comment='不可变审计链：与权威状态、Outbox 同事务提交',
schema=None
)
op.drop_table_comment(
'outbox_event',
existing_comment='Transactional Outbox：已提交领域事实的可靠投递队列',
schema=None
)
op.drop_table_comment(
'personal_domain',
existing_comment='个人数据域：数据归属与授权边界，绑定固定数据区域',
schema=None
)
op.drop_table_comment(
'service_account',
existing_comment='服务账号：API key 明文只显示一次，只存哈希与前缀',
schema=None
)
op.add_column('session', sa.Column('token_hash', sa.String(), nullable=False))
op.add_column('session', sa.Column('csrf_token_hash', sa.String(), nullable=False))
op.create_unique_constraint(op.f('uq_session_token_hash'), 'session', ['token_hash'])
op.drop_table_comment(
'session',
existing_comment='Web 会话：ID 为 CSPRNG opaque，仅存校验值',
schema=None
)
op.drop_table_comment(
'user',
existing_comment='用户（WP-01）：多凭证与账户根',
schema=None
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table_comment(
'user',
'用户（WP-01）：多凭证与账户根',
existing_comment=None,
schema=None
)
op.create_table_comment(
'session',
'Web 会话：ID 为 CSPRNG opaque，仅存校验值',
existing_comment=None,
schema=None
)
op.drop_constraint(op.f('uq_session_token_hash'), 'session', type_='unique')
op.drop_column('session', 'csrf_token_hash')
op.drop_column('session', 'token_hash')
op.create_table_comment(
'service_account',
'服务账号：API key 明文只显示一次，只存哈希与前缀',
existing_comment=None,
schema=None
)
op.create_table_comment(
'personal_domain',
'个人数据域：数据归属与授权边界，绑定固定数据区域',
existing_comment=None,
schema=None
)
op.create_table_comment(
'outbox_event',
'Transactional Outbox：已提交领域事实的可靠投递队列',
existing_comment=None,
schema=None
)
op.create_table_comment(
'audit_event',
'不可变审计链：与权威状态、Outbox 同事务提交',
existing_comment=None,
schema=None
)
op.drop_index(op.f('ix_message_conversation_id'), table_name='message')
op.drop_index(op.f('ix_message_branch_id'), table_name='message')
op.drop_table('message')
op.drop_index(op.f('ix_run_personal_domain_id'), table_name='run')
op.drop_index('ix_run_domain_status', table_name='run')
op.drop_index(op.f('ix_run_conversation_id'), table_name='run')
op.drop_table('run')
op.drop_index(op.f('ix_branch_conversation_id'), table_name='branch')
op.drop_table('branch')
op.drop_index('uq_idempotency_owner_route_key', table_name='idempotency_record')
op.drop_table('idempotency_record')
op.drop_index(op.f('ix_conversation_personal_domain_id'), table_name='conversation')
op.drop_table('conversation')
# ### end Alembic commands ###
Reference in new issue View Git Blame Copy Permalink