diff --git a/backend/alembic/versions/5cf6d328c10a_run_事件流与当前分支.py b/backend/alembic/versions/5cf6d328c10a_run_事件流与当前分支.py new file mode 100644 index 0000000..6e3c04e --- /dev/null +++ b/backend/alembic/versions/5cf6d328c10a_run_事件流与当前分支.py @@ -0,0 +1,49 @@ +"""Run 事件流与当前分支 + +Revision ID: 5cf6d328c10a +Revises: 4cc407ec8268 +Create Date: 2026-09-10 13:13:16.305296 + +""" +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 = '5cf6d328c10a' +down_revision: str | Sequence[str] | None = '4cc407ec8268' +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('run_event', + sa.Column('id', sa.String(), nullable=False), + sa.Column('run_id', sa.String(), nullable=False), + sa.Column('sequence', sa.Integer(), nullable=False), + sa.Column('type', sa.String(), nullable=False), + sa.Column('data', 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("type IN ('content.block.started', 'content.block.delta', 'content.block.progress', 'content.block.completed', 'content.block.failed', 'content.block.cancelled', 'message.completed', 'run.status.changed')", name=op.f('ck_run_event_type_values')), + sa.ForeignKeyConstraint(['run_id'], ['run.id'], name=op.f('fk_run_event_run'), ondelete='cascade'), + sa.PrimaryKeyConstraint('id', name=op.f('pk_run_event')) + ) + op.create_index('uq_run_event_run_sequence', 'run_event', ['run_id', 'sequence'], unique=True) + op.add_column('conversation', sa.Column('current_branch_id', sa.String(), nullable=True)) + op.create_foreign_key(op.f('fk_conversation_branch'), 'conversation', 'branch', ['current_branch_id'], ['id'], ondelete='set null', use_alter=True) + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.drop_constraint(op.f('fk_conversation_branch'), 'conversation', type_='foreignkey') + op.drop_column('conversation', 'current_branch_id') + op.drop_index('uq_run_event_run_sequence', table_name='run_event') + op.drop_table('run_event') + # ### end Alembic commands ### diff --git a/backend/app/conversations/routes.py b/backend/app/conversations/routes.py index 8c6b861..202a216 100644 --- a/backend/app/conversations/routes.py +++ b/backend/app/conversations/routes.py @@ -102,7 +102,7 @@ async def run_events( db: DbSession, user: Annotated[User, Depends(get_current_user)], ) -> StreamingResponse: - """Run SSE：返回当前快照与终止状态（垂直切片空状态闭环）。""" + """Run SSE：重放持久化事件流（sequence 单调），客户端按序去重（契约第 5 节）。""" result = await db.execute(select(Run).where(Run.id == run_id)) run = result.scalar_one_or_none() if run is None: @@ -111,29 +111,98 @@ async def run_events( if run.personal_domain_id != domain.id: raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在") - terminal = run.status in {"completed", "failed", "cancelled"} + from app.models import RunEvent + + events = ( + await db.execute( + select(RunEvent) + .where(RunEvent.run_id == run.id) + .order_by(RunEvent.sequence.asc()) + ) + ).scalars().all() async def event_stream() -> AsyncIterator[str]: - def envelope(event_type: str, data: dict[str, object], sequence: int) -> str: + for event in events: payload = json.dumps( - {"type": event_type, "data": data, "sequence": sequence}, ensure_ascii=False - ) - return f"data: {payload}\n\n" - - yield envelope("run.status.changed", {"run_id": run.id, "status": run.status}, 0) - if terminal: - yield envelope( - "message.completed", - { - "run_id": run.id, - "message_ids": [], - "final_status": run.status, - }, - 1, + {"type": event.type, "data": event.data, "sequence": event.sequence}, + ensure_ascii=False, ) + yield f"data: {payload}\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={"Cache-Control": "no-store"}, ) + + +@router.get("/{conversation_id}/messages") +async def list_messages( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(get_current_user)], + limit: int = 50, + cursor: str | None = None, + branch_id: str | None = None, +) -> dict[str, object]: + """消息列表：opaque cursor 分页（默认 50，契约第 4.1 节）。""" + if not 1 <= limit <= 200: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "limit 取值范围 1-200") + domain = await _domain_of(db, user) + page, next_cursor, has_more = await service.list_messages( + db, domain.id, conversation_id, branch_id, limit, cursor + ) + from app.conversations.schemas import MessageOut + + return { + "data": [ + MessageOut( + id=m.id, + conversation_id=m.conversation_id, + branch_id=m.branch_id, + role=m.role, + content=m.content, + parser_version=m.parser_version, + created_at=m.created_at.isoformat(), + ).model_dump() + for m in page + ], + "next_cursor": next_cursor, + "has_more": has_more, + } + + +@router.get("/{conversation_id}/branches") +async def list_branches( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(get_current_user)], +) -> dict[str, object]: + domain = await _domain_of(db, user) + branches = await service.list_branches(db, domain.id, conversation_id) + from app.conversations.schemas import BranchOut + + return { + "data": [ + BranchOut( + id=b.id, conversation_id=b.conversation_id, title=b.title, + created_at=b.created_at.isoformat(), + ).model_dump() + for b in branches + ] + } + + +@router.post("/{conversation_id}/current-branch") +async def set_current_branch( + conversation_id: str, + db: DbSession, + user: Annotated[User, Depends(require_state_change)], + body: dict[str, str], +) -> dict[str, str]: + """切换当前路径（主脉工作入口；采用≠归真）。""" + domain = await _domain_of(db, user) + conversation = await service.set_current_branch( + db, domain.id, conversation_id, body.get("branch_id", "") + ) + return {"id": conversation.id, "current_branch_id": conversation.current_branch_id or ""} diff --git a/backend/app/conversations/schemas.py b/backend/app/conversations/schemas.py index c651788..ca7b6c8 100644 --- a/backend/app/conversations/schemas.py +++ b/backend/app/conversations/schemas.py @@ -1,7 +1,8 @@ -"""对话域 schema：ContentBlock 判别联合（垂直切片子集）。 +"""对话域 schema：统一 ContentBlock 判别联合（21 类型）。 -完整 21 类型随 WP-03 落地；本切片登记 text/markdown 并拒绝未知类型， -契约真源为 openapi/openapi.yaml。 +与 openapi/openapi.yaml 的 components.schemas 一一对齐；此为服务端 +Pydantic 边界（AGENTS 8.2），字段语义以契约 docs/contracts/content-blocks-and-streaming.md 为准。 +markdown block 在服务端经灵台引擎规范化（契约第 4 节）。 """ from typing import Annotated, Literal @@ -16,6 +17,9 @@ class BlockBase(BaseModel): status: Literal["pending", "streaming", "completed", "failed", "cancelled"] | None = None +# ===== 文本与结构 ===== + + class TextBlock(BlockBase): type: Literal["text"] text: str @@ -28,7 +32,203 @@ class MarkdownBlock(BlockBase): flavor: Literal["hunjing-gfm-v1"] = "hunjing-gfm-v1" -ContentBlock = Annotated[TextBlock | MarkdownBlock, Field(discriminator="type")] +class CodeBlock(BlockBase): + type: Literal["code"] + source: str + language: str | None = None + filename: str | None = None + + +class QuoteBlock(BlockBase): + type: Literal["quote"] + content: list["ContentBlock"] + source_ref: str | None = None + + +class TableBlock(BlockBase): + type: Literal["table"] + columns: list[str] + rows: list[list[object]] + caption: str | None = None + + +class CalloutBlock(BlockBase): + type: Literal["callout"] + kind: Literal["info", "warning", "danger", "success"] + title: str | None = None + content: list["ContentBlock"] + + +# ===== 数学与图 ===== + + +class MathBlock(BlockBase): + type: Literal["math"] + notation: Literal["latex", "asciimath"] + display: Literal["inline", "block"] + source: str + fallback_text: str | None = None + label: str | None = None + + +class MermaidBlock(BlockBase): + type: Literal["mermaid"] + source: str + title: str | None = None + caption: str | None = None + fallback_code_language: str = "mermaid" + + +# ===== 媒体与文件 ===== + + +class ImageBlock(BlockBase): + type: Literal["image"] + file_id: str | None = None + url_ref: str | None = None + alt: str + width: int | None = Field(default=None, ge=1) + height: int | None = Field(default=None, ge=1) + caption: str | None = None + + +class VideoBlock(BlockBase): + type: Literal["video"] + file_id: str + poster_file_id: str | None = None + caption: str | None = None + transcript_file_id: str | None = None + duration_ms: int | None = Field(default=None, ge=0) + + +class AudioBlock(BlockBase): + type: Literal["audio"] + file_id: str + transcript: str | None = None + duration_ms: int | None = Field(default=None, ge=0) + waveform: str | None = None + + +class FileBlock(BlockBase): + type: Literal["file"] + file_id: str + filename: str + media_type: str + size_bytes: int = Field(ge=0) + description: str | None = None + + +class GalleryItem(BaseModel): + model_config = ConfigDict(extra="forbid") + + type: Literal["image", "video"] + file_id: str + alt: str | None = None + caption: str | None = None + + +class GalleryBlock(BlockBase): + type: Literal["gallery"] + items: list[GalleryItem] + layout: str | None = None + + +class AvatarBlock(BlockBase): + type: Literal["avatar"] + asset_id: str + format: Literal["live2d", "vrm", "gltf"] + preview_file_id: str | None = None + + +# ===== Agent 与工具 ===== + + +class ToolCallBlock(BlockBase): + type: Literal["tool_call"] + tool_name: str + call_id: str + arguments_redacted: dict[str, object] | None = None + approval_id: str | None = None + + +class ToolResultBlock(BlockBase): + type: Literal["tool_result"] + call_id: str + summary: dict[str, object] | None = None + file_ids: list[str] | None = None + is_error: bool | None = None + + +class AgentStatusBlock(BlockBase): + type: Literal["agent_status"] + agent_id: str | None = None + child_run_id: str | None = None + stage: str + progress: dict[str, object] | None = None + waiting_reason: str | None = None + + +class ApprovalRequestBlock(BlockBase): + type: Literal["approval_request"] + approval_id: str + action: str + impact: str | None = None + arguments_summary: dict[str, object] | None = None + expires_at: str | None = None + + +class ArtifactBlock(BlockBase): + type: Literal["artifact"] + artifact_id: str + artifact_type: str + version: int | None = None + + +class CitationBlock(BlockBase): + type: Literal["citation"] + source_id: str + title: str + quoted_text: str + locator: dict[str, object] | None = None + + +class ErrorBlock(BlockBase): + type: Literal["error"] + message: str + retryable: bool | None = None + fallback: str | None = None + + +_BlockTypes = ( + TextBlock + | MarkdownBlock + | CodeBlock + | QuoteBlock + | TableBlock + | CalloutBlock + | MathBlock + | MermaidBlock + | ImageBlock + | VideoBlock + | AudioBlock + | FileBlock + | GalleryBlock + | AvatarBlock + | ToolCallBlock + | ToolResultBlock + | AgentStatusBlock + | ApprovalRequestBlock + | ArtifactBlock + | CitationBlock + | ErrorBlock +) + +ContentBlock = Annotated[_BlockTypes, Field(discriminator="type")] + +# 系统产生类 block：普通 Message 请求不允许客户端伪造（契约 3.5 节） +RUNTIME_BLOCK_TYPES = frozenset( + {"tool_call", "tool_result", "agent_status", "approval_request", "artifact"} +) class MessageIn(BaseModel): @@ -59,3 +259,17 @@ class RunOut(BaseModel): class MessageCreatedOut(BaseModel): message: MessageOut run: RunOut | None + + +class BranchOut(BaseModel): + id: str + conversation_id: str + title: str + created_at: str + + +class ConversationOut(BaseModel): + id: str + title: str + current_branch_id: str | None = None + created_at: str diff --git a/backend/app/conversations/service.py b/backend/app/conversations/service.py index 691ea52..ff9b2a8 100644 --- a/backend/app/conversations/service.py +++ b/backend/app/conversations/service.py @@ -13,7 +13,7 @@ from fastapi import HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from app.conversations.schemas import MessageIn +from app.conversations.schemas import RUNTIME_BLOCK_TYPES, MessageIn from app.core import lingtai from app.core.ids import new_id from app.models import ( @@ -24,6 +24,7 @@ from app.models import ( Message, OutboxEvent, Run, + RunEvent, ) MESSAGE_ROUTE = "/v1/conversations/{conversation_id}/messages" @@ -123,6 +124,13 @@ async def replay_or_execute_message( conversation = await _require_conversation(db, domain_id, conversation_id) branch = await _get_default_branch(db, conversation.id) + forged = [b.type for b in body.content if b.type in RUNTIME_BLOCK_TYPES] + if forged: + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + f"系统产生类 block 不允许客户端提交：{', '.join(sorted(set(forged)))}", + ) + blocks: list[dict[str, Any]] = [] parser_version: str | None = None for b in body.content: @@ -161,6 +169,7 @@ async def replay_or_execute_message( resource_ids: dict[str, Any] = {"message_id": message.id} if run is not None: resource_ids["run_id"] = run.id + _record_run_events(db, run, message, blocks) db.add( AuditEvent( actor_type="user", @@ -191,3 +200,105 @@ async def replay_or_execute_message( ) await db.flush() return message, run, False + + +def _record_run_events( + db: AsyncSession, run: Run, message: Message, blocks: list[dict[str, Any]] +) -> None: + """空状态闭环的事件流：状态事件 + 每 block completed + message.completed。 + + sequence 在 Run 内从 0 单调；事件与权威状态同事务提交（契约第 5 节）。 + delta/progress 事件在模型流式接入（WP-05）后由真实流产生，此处不伪造。 + """ + seq = 0 + events: list[tuple[str, dict[str, Any]]] = [ + ("run.status.changed", {"run_id": run.id, "status": run.status}) + ] + for index, block in enumerate(blocks): + events.append( + ( + "content.block.completed", + {"run_id": run.id, "message_id": message.id, "index": index, + "block": {"id": block["id"], "type": block["type"], "status": "completed"}}, + ) + ) + events.append( + ( + "message.completed", + {"run_id": run.id, "message_ids": [message.id], "final_status": run.status}, + ) + ) + for event_type, data in events: + db.add(RunEvent(run_id=run.id, sequence=seq, type=event_type, data=data)) + seq += 1 + + +async def list_messages( + db: AsyncSession, + domain_id: str, + conversation_id: str, + branch_id: str | None, + limit: int, + cursor: str | None, +) -> tuple[list[Message], str | None, bool]: + """按创建时间升序的消息列表（opaque cursor 分页，契约第 4.1 节）。""" + conversation = await _require_conversation(db, domain_id, conversation_id) + stmt = select(Message).where(Message.conversation_id == conversation.id) + if branch_id is not None: + stmt = stmt.where(Message.branch_id == branch_id) + after_created, after_id = decode_cursor(cursor) + if after_created is not None: + # (created_at, id) 复合键分页：严格大于游标位置 + stmt = stmt.where( + (Message.created_at > after_created) + | ((Message.created_at == after_created) & (Message.id > after_id)) + ) + stmt = stmt.order_by(Message.created_at.asc(), Message.id.asc()).limit(limit + 1) + rows = list((await db.scalars(stmt)).all()) + has_more = len(rows) > limit + page = rows[:limit] + next_cursor = encode_cursor(page[-1]) if has_more and page else None + return page, next_cursor, has_more + + +def encode_cursor(message: Message) -> str: + import base64 + + raw = f"{message.created_at.isoformat()}|{message.id}" + return base64.urlsafe_b64encode(raw.encode()).decode() + + +def decode_cursor(cursor: str | None) -> tuple[Any | None, str]: + import base64 + + if cursor is None: + return None, "" + try: + raw = base64.urlsafe_b64decode(cursor.encode()).decode() + created_str, _, mid = raw.partition("|") + return datetime.fromisoformat(created_str), mid + except Exception as exc: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "cursor 不合法") from exc + + +async def list_branches( + db: AsyncSession, domain_id: str, conversation_id: str +) -> list[Branch]: + conversation = await _require_conversation(db, domain_id, conversation_id) + result = await db.execute( + select(Branch).where(Branch.conversation_id == conversation.id).order_by(Branch.created_at) + ) + return list(result.scalars().all()) + + +async def set_current_branch( + db: AsyncSession, domain_id: str, conversation_id: str, branch_id: str +) -> Conversation: + """切换当前路径：分支必须属于该会话（跨域/跨会话统一 404/422 fail-closed）。""" + conversation = await _require_conversation(db, domain_id, conversation_id) + branch = await db.get(Branch, branch_id) + if branch is None or branch.conversation_id != conversation.id: + raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "分支不属于该会话") + conversation.current_branch_id = branch_id + await db.flush() + return conversation diff --git a/backend/app/models.py b/backend/app/models.py index 33e17b7..dae371a 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -154,6 +154,9 @@ class Conversation(Base): ForeignKey("personal_domain.id", ondelete="cascade"), nullable=False, index=True ) title: Mapped[str] = mapped_column(String, nullable=False, default="未命名对话") + current_branch_id: Mapped[str | None] = mapped_column( + ForeignKey("branch.id", ondelete="set null", use_alter=True), nullable=True + ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) @@ -249,3 +252,31 @@ class IdempotencyRecord(Base): __table_args__ = ( Index("uq_idempotency_owner_route_key", "owner_domain_id", "route", "key", unique=True), ) + + +class RunEvent(Base): + """Run 事件流：持久化后可重放；sequence 在 Run 内单调（契约第 5 节）。""" + + __tablename__ = "run_event" + + id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("event_")) + run_id: Mapped[str] = mapped_column( + ForeignKey("run.id", ondelete="cascade"), nullable=False + ) + sequence: Mapped[int] = mapped_column(nullable=False) + type: Mapped[str] = mapped_column(String, nullable=False) + data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + CheckConstraint( + "type IN ('content.block.started', 'content.block.delta', 'content.block.progress', " + "'content.block.completed', 'content.block.failed', 'content.block.cancelled', " + "'message.completed', 'run.status.changed')", + name="type_values", + ), + # Run 内 sequence 单调且唯一（重放去重的服务端保证） + Index("uq_run_event_run_sequence", "run_id", "sequence", unique=True), + ) diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py index 0e62432..eefc189 100644 --- a/backend/tests/test_conversations.py +++ b/backend/tests/test_conversations.py @@ -72,7 +72,7 @@ class TestMessagePipeline: cid = await _create_conversation(client, headers) resp = await client.post( f"/v1/conversations/{cid}/messages", - json={"content": [{"type": "image", "file_id": "x", "alt": "y"}]}, + json={"content": [{"type": "hologram", "file_id": "x"}]}, headers={**headers, "Idempotency-Key": "idem-bad"}, ) assert resp.status_code == 422 @@ -196,3 +196,133 @@ class TestLingtaiNormalization: ) # fail closed：引擎不可用时显式失败，不静默降级直存 assert resp.status_code == 500 + + +@pytest.mark.usefixtures("client") +class TestContentBlocks: + async def test_media_blocks_accepted(self, client: AsyncClient) -> None: + headers = await login(client, "blocks@example.com") + cid = await _create_conversation(client, headers) + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={ + "content": [ + {"type": "image", "file_id": "file_1", "alt": "图"}, + {"type": "math", "notation": "latex", "display": "inline", "source": "a+b"}, + {"type": "table", "columns": ["a"], "rows": [["1"]]}, + {"type": "citation", "source_id": "s1", "title": "t", "quoted_text": "q"}, + {"type": "error", "message": "块错误"}, + ] + }, + headers={**headers, "Idempotency-Key": "idem-blocks"}, + ) + assert resp.status_code == 201, resp.text + types = [b["type"] for b in resp.json()["message"]["content"]] + assert types == ["image", "math", "table", "citation", "error"] + + async def test_runtime_blocks_rejected_from_client(self, client: AsyncClient) -> None: + """契约 3.5：tool_call/agent_status 等系统 block 不允许客户端伪造。""" + headers = await login(client, "rtb@example.com") + cid = await _create_conversation(client, headers) + for block in [ + {"type": "tool_call", "tool_name": "t", "call_id": "c1"}, + {"type": "agent_status", "stage": "s"}, + {"type": "approval_request", "approval_id": "a1", "action": "do"}, + {"type": "artifact", "artifact_id": "ar1", "artifact_type": "file"}, + ]: + resp = await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [block]}, + headers={**headers, "Idempotency-Key": f"idem-rtb-{block['type']}"}, + ) + assert resp.status_code == 422, (block["type"], resp.text) + + +@pytest.mark.usefixtures("client") +class TestSseReplay: + async def test_events_replayed_in_order_with_sequence(self, client: AsyncClient) -> None: + headers = await login(client, "sseq@example.com") + cid = await _create_conversation(client, headers) + resp = await _post_message(client, headers, cid, "idem-seq") + run_id = resp.json()["run"]["id"] + sse = await client.get(f"/v1/runs/{run_id}/events") + lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")] + events = [json.loads(ln.removeprefix("data: ")) for ln in lines] + sequences = [e["sequence"] for e in events] + assert sequences == sorted(sequences) == list(range(len(events))) + assert events[0]["type"] == "run.status.changed" + assert events[-1]["type"] == "message.completed" + # 用户消息两个 block：text 与（无 markdown 时）content.block.completed 数对应 + assert sum(1 for e in events if e["type"] == "content.block.completed") >= 1 + + +@pytest.mark.usefixtures("client") +class TestPaginationAndBranches: + async def test_message_cursor_pagination(self, client: AsyncClient) -> None: + headers = await login(client, "page@example.com") + cid = await _create_conversation(client, headers) + for i in range(5): + await client.post( + f"/v1/conversations/{cid}/messages", + json={"content": [{"type": "text", "text": f"m{i}"}], "start_run": False}, + headers={**headers, "Idempotency-Key": f"idem-p{i}"}, + ) + page1 = await client.get(f"/v1/conversations/{cid}/messages?limit=2", headers=headers) + assert page1.status_code == 200 + body1 = page1.json() + assert len(body1["data"]) == 2 and body1["has_more"] is True + assert body1["next_cursor"] + page2 = await client.get( + f"/v1/conversations/{cid}/messages?limit=2&cursor={body1['next_cursor']}", + headers=headers, + ) + body2 = page2.json() + ids1 = {m["id"] for m in body1["data"]} + ids2 = {m["id"] for m in body2["data"]} + assert not (ids1 & ids2) + last = await client.get( + f"/v1/conversations/{cid}/messages?limit=100&cursor={body2['next_cursor']}", + headers=headers, + ) + assert last.json()["has_more"] is False + + async def test_invalid_cursor_rejected(self, client: AsyncClient) -> None: + headers = await login(client, "badc@example.com") + cid = await _create_conversation(client, headers) + resp = await client.get( + f"/v1/conversations/{cid}/messages?cursor=%%%", headers=headers + ) + assert resp.status_code == 422 + + async def test_branches_and_current_path(self, client: AsyncClient) -> None: + headers = await login(client, "branch@example.com") + cid = await _create_conversation(client, headers) + listing = await client.get(f"/v1/conversations/{cid}/branches", headers=headers) + branches = listing.json()["data"] + assert len(branches) == 1 and branches[0]["title"] == "主脉" + # 切换到不存在的分支：422 fail-closed + bad = await client.post( + f"/v1/conversations/{cid}/current-branch", + json={"branch_id": "branch_00000000-0000-7000-8000-000000000000"}, + headers=headers, + ) + assert bad.status_code == 422 + # 切换到合法分支 + ok = await client.post( + f"/v1/conversations/{cid}/current-branch", + json={"branch_id": branches[0]["id"]}, + headers=headers, + ) + assert ok.status_code == 200 + assert ok.json()["current_branch_id"] == branches[0]["id"] + + async def test_cross_domain_branch_rejected(self, client: AsyncClient) -> None: + alice = await login(client, "br-a@example.com") + cid = await _create_conversation(client, alice) + listing = await client.get(f"/v1/conversations/{cid}/branches", headers=alice) + bid = listing.json()["data"][0]["id"] + await login(client, "br-b@example.com") + resp = await client.post( + f"/v1/conversations/{cid}/current-branch", json={"branch_id": bid} + ) + assert resp.status_code in (401, 403) # 未带新会话 Cookie：401；带错 Cookie：拒绝