diff --git a/backend/alembic/versions/0fb7d605080a_模型网关_逻辑模型_端点_turn_与_modelcall.py b/backend/alembic/versions/0fb7d605080a_模型网关_逻辑模型_端点_turn_与_modelcall.py
new file mode 100644
index 0000000..396d1fb
--- /dev/null
+++ b/backend/alembic/versions/0fb7d605080a_模型网关_逻辑模型_端点_turn_与_modelcall.py
@@ -0,0 +1,94 @@
+"""模型网关：逻辑模型、端点、Turn 与 ModelCall
+
+Revision ID: 0fb7d605080a
+Revises: 6292e21950cc
+Create Date: 2026-09-11 09:22:23.431933
+
+"""
+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 = '0fb7d605080a'
+down_revision: str | Sequence[str] | None = '6292e21950cc'
+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('logical_model',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('capabilities', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('description', sa.String(), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_logical_model'))
+    )
+    op.create_table('provider_endpoint',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('logical_model_id', sa.String(), nullable=False),
+    sa.Column('provider', sa.String(), nullable=False),
+    sa.Column('name', sa.String(), nullable=False),
+    sa.Column('base_url', sa.String(), nullable=False),
+    sa.Column('model_id', sa.String(), nullable=False),
+    sa.Column('api_key_env', sa.String(), nullable=False),
+    sa.Column('priority', sa.Integer(), nullable=False),
+    sa.Column('enabled', sa.Boolean(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.CheckConstraint("provider IN ('openai_compatible', 'anthropic')", name=op.f('ck_provider_endpoint_provider_values')),
+    sa.ForeignKeyConstraint(['logical_model_id'], ['logical_model.id'], name=op.f('fk_provider_endpoint_logical_model'), ondelete='cascade'),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_provider_endpoint'))
+    )
+    op.create_index(op.f('ix_provider_endpoint_logical_model_id'), 'provider_endpoint', ['logical_model_id'], unique=False)
+    op.create_table('turn',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('run_id', sa.String(), nullable=False),
+    sa.Column('index', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['run_id'], ['run.id'], name=op.f('fk_turn_run'), ondelete='cascade'),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_turn'))
+    )
+    op.create_index(op.f('ix_turn_run_id'), 'turn', ['run_id'], unique=False)
+    op.create_index('uq_turn_run_index', 'turn', ['run_id', 'index'], unique=True)
+    op.create_table('model_call',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('turn_id', sa.String(), nullable=False),
+    sa.Column('endpoint_id', sa.String(), nullable=True),
+    sa.Column('attempt', sa.Integer(), nullable=False),
+    sa.Column('status', sa.String(), nullable=False),
+    sa.Column('request_summary', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('usage', postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    sa.Column('ttft_ms', sa.Integer(), nullable=True),
+    sa.Column('error', sa.String(), nullable=True),
+    sa.Column('started_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('completed_at', sa.DateTime(timezone=True), nullable=True),
+    sa.CheckConstraint("status IN ('pending', 'streaming', 'succeeded', 'failed')", name=op.f('ck_model_call_status_values')),
+    sa.ForeignKeyConstraint(['endpoint_id'], ['provider_endpoint.id'], name=op.f('fk_model_call_provider_endpoint'), ondelete='set null'),
+    sa.ForeignKeyConstraint(['turn_id'], ['turn.id'], name=op.f('fk_model_call_turn'), ondelete='cascade'),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_model_call'))
+    )
+    op.create_index(op.f('ix_model_call_turn_id'), 'model_call', ['turn_id'], unique=False)
+    op.create_index('uq_model_call_turn_attempt', 'model_call', ['turn_id', 'attempt'], unique=True)
+    op.create_foreign_key(op.f('fk_character_character_version'), 'character', 'character_version', ['current_version_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_character_character_version'), 'character', type_='foreignkey')
+    op.drop_index('uq_model_call_turn_attempt', table_name='model_call')
+    op.drop_index(op.f('ix_model_call_turn_id'), table_name='model_call')
+    op.drop_table('model_call')
+    op.drop_index('uq_turn_run_index', table_name='turn')
+    op.drop_index(op.f('ix_turn_run_id'), table_name='turn')
+    op.drop_table('turn')
+    op.drop_index(op.f('ix_provider_endpoint_logical_model_id'), table_name='provider_endpoint')
+    op.drop_table('provider_endpoint')
+    op.drop_table('logical_model')
+    # ### end Alembic commands ###
diff --git a/backend/alembic/versions/1c1f7edb7cbe_消息关联_run.py b/backend/alembic/versions/1c1f7edb7cbe_消息关联_run.py
new file mode 100644
index 0000000..1033150
--- /dev/null
+++ b/backend/alembic/versions/1c1f7edb7cbe_消息关联_run.py
@@ -0,0 +1,36 @@
+"""消息关联 Run
+
+Revision ID: 1c1f7edb7cbe
+Revises: 0fb7d605080a
+Create Date: 2026-09-11 09:24:35.912343
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '1c1f7edb7cbe'
+down_revision: str | Sequence[str] | None = '0fb7d605080a'
+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.add_column('message', sa.Column('run_id', sa.String(), nullable=True))
+    op.create_index(op.f('ix_message_run_id'), 'message', ['run_id'], unique=False)
+    op.create_foreign_key(op.f('fk_message_run'), 'message', 'run', ['run_id'], ['id'], ondelete='set null')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(op.f('fk_message_run'), 'message', type_='foreignkey')
+    op.drop_index(op.f('ix_message_run_id'), table_name='message')
+    op.drop_column('message', 'run_id')
+    # ### end Alembic commands ###
diff --git a/backend/app/conversations/routes.py b/backend/app/conversations/routes.py
index 0a1172f..115aaaa 100644
--- a/backend/app/conversations/routes.py
+++ b/backend/app/conversations/routes.py
@@ -4,7 +4,7 @@ import json
 from collections.abc import AsyncIterator
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
+from fastapi import APIRouter, BackgroundTasks, Depends, Header, HTTPException, Request, status
 from fastapi.responses import JSONResponse, StreamingResponse
 from sqlalchemy import select
 
@@ -51,6 +51,7 @@ async def post_message(
     db: DbSession,
     user: Annotated[User, Depends(require_state_change)],
     request: Request,
+    background_tasks: BackgroundTasks,
     body: MessageIn,
     idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
 ) -> MessageCreatedOut | JSONResponse:
@@ -90,6 +91,17 @@ async def post_message(
     if replayed:
         # 重放成功按契约返回 200 与原资源（避开装饰器固定 201）
         return JSONResponse(status_code=200, content=json.loads(payload.model_dump_json()))
+    if run is not None:
+        # 先显式提交：确保 Run 在后台任务执行前已落库（与依赖清理顺序解耦）
+        await db.commit()
+        from app.gateway import execute_run
+
+        background_tasks.add_task(
+            execute_run,
+            request.app.state.session_factory,
+            run.id,
+            transport=getattr(request.app.state, "gw_transport", None),
+        )
     return payload
 
 
@@ -111,23 +123,40 @@ async def run_events(
     if run.personal_domain_id != domain.id:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "Run 不存在")
 
+    import asyncio as _asyncio
+
     from app.models import RunEvent
 
-    events = (
-        await db.execute(
-            select(RunEvent)
-            .where(RunEvent.run_id == run.id)
-            .order_by(RunEvent.sequence.asc())
-        )
-    ).scalars().all()
+    terminal = {"completed", "failed", "cancelled"}
 
     async def event_stream() -> AsyncIterator[str]:
-        for event in events:
-            payload = json.dumps(
-                {"type": event.type, "data": event.data, "sequence": event.sequence},
-                ensure_ascii=False,
+        last_seq = -1
+        deadline = _asyncio.get_event_loop().time() + 30.0
+        while True:
+            rows = list(
+                (
+                    await db.execute(
+                        select(RunEvent)
+                        .where(RunEvent.run_id == run.id, RunEvent.sequence > last_seq)
+                        .order_by(RunEvent.sequence.asc())
+                    )
+                ).scalars().all()
             )
-            yield f"data: {payload}\n\n"
+            for event in rows:
+                payload = json.dumps(
+                    {"type": event.type, "data": event.data, "sequence": event.sequence},
+                    ensure_ascii=False,
+                )
+                yield f"data: {payload}\n\n"
+                last_seq = event.sequence
+            if not rows:
+                await db.rollback()  # 释放快照，下一轮可见新写入
+                await db.refresh(run)
+                if run.status in terminal:
+                    break
+                if _asyncio.get_event_loop().time() > deadline:
+                    break
+                await _asyncio.sleep(0.05)
 
     return StreamingResponse(
         event_stream(),
diff --git a/backend/app/conversations/service.py b/backend/app/conversations/service.py
index da8775b..b8e03a4 100644
--- a/backend/app/conversations/service.py
+++ b/backend/app/conversations/service.py
@@ -24,7 +24,6 @@ from app.models import (
     Message,
     OutboxEvent,
     Run,
-    RunEvent,
 )
 
 MESSAGE_ROUTE = "/v1/conversations/{conversation_id}/messages"
@@ -157,20 +156,19 @@ async def replay_or_execute_message(
 
     run = None
     if body.start_run:
-        # 垂直切片空状态闭环：不调用模型，Run 原子创建并直接进入终态 completed。
+        # Run 创建为 queued，由网关执行器在后台推进至终态（WP-05）
         run = Run(
             personal_domain_id=domain_id,
             conversation_id=conversation.id,
-            status="completed",
-            completed_at=datetime.now(UTC),
+            status="queued",
         )
         db.add(run)
         await db.flush()
+        message.run_id = run.id
 
     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",
@@ -203,36 +201,6 @@ async def replay_or_execute_message(
     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,
diff --git a/backend/app/gateway/__init__.py b/backend/app/gateway/__init__.py
new file mode 100644
index 0000000..748e598
--- /dev/null
+++ b/backend/app/gateway/__init__.py
@@ -0,0 +1,6 @@
+"""模型网关：路由与 Run 执行（WP-05）。"""
+
+from .adapters import AdapterRequest, GatewayError, stream_chat
+from .runner import execute_run
+
+__all__ = ["AdapterRequest", "GatewayError", "stream_chat", "execute_run"]
diff --git a/backend/app/gateway/adapters.py b/backend/app/gateway/adapters.py
new file mode 100644
index 0000000..b2b7d2f
--- /dev/null
+++ b/backend/app/gateway/adapters.py
@@ -0,0 +1,107 @@
+"""OpenAI-compatible 流式适配器（WP-05）。
+
+- 只消费逻辑请求（messages + 逻辑模型），输出标准化的流事件与 usage。
+- 密钥从环境变量读取（endpoint.api_key_env），不落库、不进日志。
+- Provider 未返回的 usage 字段保持 None（ADR-003：不估算伪装）。
+- transport 可注入，供测试与故障注入。
+"""
+
+import json
+import os
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+
+class GatewayError(RuntimeError):
+    """网关错误：稳定分类，供 ModelCall 失败记录与上层重试决策。"""
+
+
+@dataclass
+class StreamChunk:
+    kind: str  # "delta" | "usage" | "done"
+    text: str | None = None
+    usage: dict[str, int] | None = None
+    finish_reason: str | None = None
+
+
+@dataclass
+class AdapterRequest:
+    messages: list[dict[str, str]]  # {"role", "content"}：纯文本最小集
+    model: str  # 上游模型 ID
+
+
+def _parse_sse_line(line: str) -> dict[str, Any] | None:
+    if not line.startswith("data:"):
+        return None
+    payload = line.removeprefix("data:").strip()
+    if not payload or payload == "[DONE]":
+        return None
+    try:
+        parsed: dict[str, Any] = json.loads(payload)
+        return parsed
+    except json.JSONDecodeError:
+        return None
+
+
+def _usage_from_final(chunk: dict[str, Any]) -> dict[str, int] | None:
+    usage = chunk.get("usage")
+    if not isinstance(usage, dict):
+        return None
+    result: dict[str, int] = {}
+    keys = (
+        "prompt_tokens",
+        "completion_tokens",
+        "total_tokens",
+        "cache_creation_input_tokens",
+        "cache_read_input_tokens",
+    )
+    for key in keys:
+        value = usage.get(key)
+        if isinstance(value, int):
+            result[key] = value
+    return result or None
+
+
+async def stream_chat(
+    base_url: str,
+    model_id: str,
+    api_key_env: str,
+    request: AdapterRequest,
+    *,
+    request_timeout_s: float = 120.0,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> AsyncIterator[StreamChunk]:
+    """流式调用 OpenAI-compatible /chat/completions，产出标准化块。"""
+    api_key = os.environ.get(api_key_env, "")
+    if not api_key:
+        raise GatewayError(f"环境变量 {api_key_env} 未配置 Provider 密钥")
+    url = base_url.rstrip("/") + "/chat/completions"
+    body = {
+        "model": model_id,
+        "messages": request.messages,
+        "stream": True,
+        "stream_options": {"include_usage": True},
+    }
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    async with httpx.AsyncClient(timeout=request_timeout_s, transport=transport) as client:
+        async with client.stream("POST", url, json=body, headers=headers) as resp:
+            if resp.status_code != 200:
+                detail = (await resp.aread()).decode("utf-8", "replace")[:500]
+                raise GatewayError(f"上游返回 {resp.status_code}: {detail}")
+            async for line in resp.aiter_lines():
+                chunk = _parse_sse_line(line)
+                if chunk is None:
+                    continue
+                for choice in chunk.get("choices", []):
+                    text = choice.get("delta", {}).get("content")
+                    if text:
+                        yield StreamChunk(kind="delta", text=text)
+                    finish = choice.get("finish_reason")
+                    if finish:
+                        yield StreamChunk(kind="done", finish_reason=finish)
+                usage = _usage_from_final(chunk)
+                if usage is not None:
+                    yield StreamChunk(kind="usage", usage=usage)
diff --git a/backend/app/gateway/runner.py b/backend/app/gateway/runner.py
new file mode 100644
index 0000000..5c555e8
--- /dev/null
+++ b/backend/app/gateway/runner.py
@@ -0,0 +1,250 @@
+"""Run 执行器：调用模型、产生真实流事件与 assistant 消息（WP-05）。
+
+- 后台任务执行：独立数据库会话（不复用请求会话）。
+- 事件序列（契约第 5 节）：run.status.changed(running) → content.block.started
+  → content.block.delta* → content.block.completed → message.completed →
+  run.status.changed(completed/failed)。
+- 首个用户可见 delta 前的失败允许显式 fail；无可用端点时 Run 失败并留 error 事件。
+"""
+
+import time
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from app.core.ids import new_id
+from app.gateway.adapters import AdapterRequest, GatewayError, stream_chat
+from app.models import (
+    AuditEvent,
+    Message,
+    ModelCall,
+    OutboxEvent,
+    ProviderEndpoint,
+    Run,
+    RunEvent,
+    Turn,
+)
+
+
+async def _next_sequence(db: AsyncSession, run_id: str) -> int:
+    last = await db.execute(
+        select(RunEvent.sequence)
+        .where(RunEvent.run_id == run_id)
+        .order_by(RunEvent.sequence.desc())
+        .limit(1)
+    )
+    row = last.scalar_one_or_none()
+    return (row + 1) if row is not None else 0
+
+
+def _emit(db: AsyncSession, run_id: str, seq: int, event_type: str, data: dict[str, Any]) -> int:
+    db.add(RunEvent(run_id=run_id, sequence=seq, type=event_type, data=data))
+    return seq + 1
+
+
+async def execute_run(
+    factory: async_sessionmaker[AsyncSession],
+    run_id: str,
+    transport: Any | None = None,
+) -> None:
+    """执行一个 Run 至终态。transport 注入仅供测试。"""
+    async with factory() as db:
+        try:
+            run = await db.get(Run, run_id)
+            if run is None:
+                return
+            seq = await _next_sequence(db, run_id)
+            # 状态事件：running
+            run.status = "running"
+            seq = _emit(
+                db, run_id, seq, "run.status.changed",
+                {"run_id": run_id, "status": "running"},
+            )
+            await db.commit()
+
+            # 路由：按逻辑模型选择启用端点（优先级升序）
+            endpoint = (
+                await db.execute(
+                    select(ProviderEndpoint)
+                    .where(
+                        ProviderEndpoint.enabled.is_(True),
+                        ProviderEndpoint.logical_model_id == "default-chat",
+                    )
+                    .order_by(ProviderEndpoint.priority.asc())
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+
+            turn = Turn(run_id=run_id, index=0)
+            db.add(turn)
+            await db.flush()
+
+            if endpoint is None:
+                # fail closed：无可用端点时 Run 显式失败，不静默降级
+                call = ModelCall(
+                    turn_id=turn.id, status="failed",
+                    request_summary={}, error="无可用 Provider 端点（default-chat）",
+                    started_at=datetime.now(UTC),
+                    completed_at=datetime.now(UTC),
+                )
+                db.add(call)
+                run.status = "failed"
+                run.completed_at = datetime.now(UTC)
+                seq = _emit(db, run_id, seq, "content.block.failed", {
+                    "run_id": run_id, "message_id": None, "index": 0,
+                    "block": {"id": new_id("block_"), "type": "text", "status": "failed"},
+                    "error": "no_endpoint_available",
+                })
+                seq = _emit(db, run_id, seq, "message.completed",
+                            {"run_id": run_id, "message_ids": [], "final_status": "failed"})
+                seq = _emit(db, run_id, seq, "run.status.changed",
+                            {"run_id": run_id, "status": "failed"})
+                await db.commit()
+                return
+
+            # 对话历史（最小：当前分支文本消息）
+            messages = (
+                await db.execute(
+                    select(Message)
+                    .where(
+                        Message.conversation_id == run.conversation_id,
+                        Message.role.in_(["user", "assistant"]),
+                    )
+                    .order_by(Message.created_at.asc(), Message.id.asc())
+                    .limit(50)
+                )
+            ).scalars().all()
+            prompt = [
+                {
+                    "role": m.role,
+                    "content": "".join(
+                        str(b.get("text", b.get("source", "")))
+                        for b in m.content
+                        if isinstance(b, dict)
+                    ),
+                }
+                for m in messages
+            ]
+
+            call = ModelCall(
+                turn_id=turn.id, endpoint_id=endpoint.id, status="streaming",
+                request_summary={"messages": len(prompt), "model_id": endpoint.model_id},
+                started_at=datetime.now(UTC),
+            )
+            db.add(call)
+            await db.commit()
+
+            block_id = new_id("block_")
+            seq = _emit(db, run_id, seq, "content.block.started", {
+                "run_id": run_id, "index": 0,
+                "block": {"id": block_id, "type": "text", "status": "streaming"},
+            })
+            await db.commit()
+
+            # 流式消费：真实 delta 事件
+            started = time.monotonic()
+            ttft_ms: int | None = None
+            full_text: list[str] = []
+            usage: dict[str, int] | None = None
+            finish_reason: str | None = None
+            delta_index = 0
+            try:
+                adapter_req = AdapterRequest(messages=prompt, model=endpoint.model_id)
+                async for chunk in stream_chat(
+                    endpoint.base_url, endpoint.model_id, endpoint.api_key_env,
+                    adapter_req, transport=transport,
+                ):
+                    if chunk.kind == "delta" and chunk.text:
+                        if ttft_ms is None:
+                            ttft_ms = int((time.monotonic() - started) * 1000)
+                        full_text.append(chunk.text)
+                        seq = _emit(db, run_id, seq, "content.block.delta", {
+                            "run_id": run_id, "block_id": block_id,
+                            "delta_index": delta_index,
+                            "delta": {"type": "text_append", "text": chunk.text},
+                        })
+                        delta_index += 1
+                        await db.commit()
+                    elif chunk.kind == "usage" and chunk.usage:
+                        usage = chunk.usage
+                    elif chunk.kind == "done" and chunk.finish_reason:
+                        finish_reason = chunk.finish_reason
+            except GatewayError as exc:
+                failed_call = await db.get(ModelCall, call.id)
+                assert failed_call is not None
+                call.status = "failed"
+                call.error = str(exc)[:500]
+                call.completed_at = datetime.now(UTC)
+                run.status = "failed"
+                run.completed_at = datetime.now(UTC)
+                seq = _emit(db, run_id, seq, "content.block.failed", {
+                    "run_id": run_id, "block_id": block_id, "index": 0,
+                    "error": "upstream_error",
+                })
+                seq = _emit(db, run_id, seq, "message.completed",
+                            {"run_id": run_id, "message_ids": [], "final_status": "failed"})
+                seq = _emit(db, run_id, seq, "run.status.changed",
+                            {"run_id": run_id, "status": "failed"})
+                await db.commit()
+                return
+
+            text = "".join(full_text)
+            seq = _emit(db, run_id, seq, "content.block.completed", {
+                "run_id": run_id, "index": 0,
+                "block": {"id": block_id, "type": "text", "status": "completed", "text": text},
+            })
+
+            # assistant 消息（权威快照）
+            assistant = Message(
+                conversation_id=run.conversation_id,
+                branch_id=(
+                    await db.execute(
+                        select(Message.branch_id).where(Message.run_id == run_id).limit(1)
+                    )
+                ).scalar_one_or_none() or new_id("branch_"),
+                role="assistant",
+                realm="reality",
+                content=[{"id": block_id, "type": "text", "text": text, "status": "completed"}],
+                run_id=run_id,
+            )
+            db.add(assistant)
+            await db.flush()
+            seq = _emit(db, run_id, seq, "message.completed", {
+                "run_id": run_id, "message_ids": [assistant.id], "final_status": "completed",
+            })
+            run.status = "completed"
+            run.completed_at = datetime.now(UTC)
+            seq = _emit(db, run_id, seq, "run.status.changed",
+                        {"run_id": run_id, "status": "completed"})
+
+            success_call = await db.get(ModelCall, call.id)
+            assert success_call is not None
+            success_call.status = "succeeded"
+            # Provider 未返回的 usage 保持 None：不估算伪装（ADR-003）
+            success_call.usage = usage
+            success_call.ttft_ms = ttft_ms
+            success_call.completed_at = datetime.now(UTC)
+            if finish_reason:
+                success_call.request_summary = {
+                    **success_call.request_summary, "finish_reason": finish_reason
+                }
+            db.add(
+                AuditEvent(
+                    actor_type="system", actor_id="gateway", action="model_call.completed",
+                    aggregate_type="run", aggregate_id=run_id,
+                    payload={"endpoint": endpoint.name, "usage": usage, "ttft_ms": ttft_ms},
+                )
+            )
+            db.add(
+                OutboxEvent(
+                    aggregate_type="message", aggregate_id=assistant.id,
+                    event_type="message.created",
+                    payload={"message_id": assistant.id, "run_id": run_id, "role": "assistant"},
+                )
+            )
+            await db.commit()
+        except Exception:
+            await db.rollback()
+            raise
diff --git a/backend/app/models.py b/backend/app/models.py
index eacbb82..f1da2cd 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -207,6 +207,9 @@ class Message(Base):
         ForeignKey("message.id", ondelete="set null"), nullable=True
     )
     variant_kind: Mapped[str] = mapped_column(String, nullable=False, default="origin")
+    run_id: Mapped[str | None] = mapped_column(
+        ForeignKey("run.id", ondelete="set null"), nullable=True, index=True
+    )
     content: Mapped[list[dict[str, Any]]] = mapped_column(JSONB, nullable=False)
     parser_version: Mapped[str | None] = mapped_column(String, nullable=True)
     created_at: Mapped[datetime] = mapped_column(
@@ -374,3 +377,98 @@ class Character(Base):
             name="origin_values",
         ),
     )
+
+
+class LogicalModel(Base):
+    """逻辑模型：业务只引用能力与逻辑名，不感知 Provider（契约第 5 节分面）。"""
+
+    __tablename__ = "logical_model"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)  # kebab-case，如 default-chat
+    capabilities: Mapped[list[str]] = mapped_column(JSONB, nullable=False, default=list)
+    description: Mapped[str | None] = mapped_column(String, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+class ProviderEndpoint(Base):
+    """Provider 端点：密钥只存环境变量名，永不落库明文。"""
+
+    __tablename__ = "provider_endpoint"
+
+    id: Mapped[str] = mapped_column(
+        String, primary_key=True, default=lambda: new_id("endpoint_")
+    )
+    logical_model_id: Mapped[str] = mapped_column(
+        ForeignKey("logical_model.id", ondelete="cascade"), nullable=False, index=True
+    )
+    provider: Mapped[str] = mapped_column(String, nullable=False)
+    name: Mapped[str] = mapped_column(String, nullable=False)
+    base_url: Mapped[str] = mapped_column(String, nullable=False)
+    model_id: Mapped[str] = mapped_column(String, nullable=False)
+    api_key_env: Mapped[str] = mapped_column(String, nullable=False)
+    priority: Mapped[int] = mapped_column(nullable=False, default=100)
+    enabled: Mapped[bool] = mapped_column(nullable=False, default=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = (
+        CheckConstraint(
+            "provider IN ('openai_compatible', 'anthropic')", name="provider_values"
+        ),
+    )
+
+
+class Turn(Base):
+    """Turn：Run 内一次模型决策及其工具交互阶段。"""
+
+    __tablename__ = "turn"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True, default=lambda: new_id("turn_"))
+    run_id: Mapped[str] = mapped_column(
+        ForeignKey("run.id", ondelete="cascade"), nullable=False, index=True
+    )
+    index: Mapped[int] = mapped_column(nullable=False, default=0)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+    __table_args__ = (
+        Index("uq_turn_run_index", "run_id", "index", unique=True),
+    )
+
+
+class ModelCall(Base):
+    """每次上游 attempt 单独持久化：重试/故障转移不覆盖（ADR-003）。
+
+    Provider 未返回的 usage 字段保持 null，不估算伪装。
+    """
+
+    __tablename__ = "model_call"
+
+    id: Mapped[str] = mapped_column(
+        String, primary_key=True, default=lambda: new_id("model_call_")
+    )
+    turn_id: Mapped[str] = mapped_column(
+        ForeignKey("turn.id", ondelete="cascade"), nullable=False, index=True
+    )
+    endpoint_id: Mapped[str | None] = mapped_column(
+        ForeignKey("provider_endpoint.id", ondelete="set null"), nullable=True
+    )
+    attempt: Mapped[int] = mapped_column(nullable=False, default=1)
+    status: Mapped[str] = mapped_column(String, nullable=False, default="pending")
+    request_summary: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict)
+    usage: Mapped[dict[str, Any] | None] = mapped_column(JSONB, nullable=True)
+    ttft_ms: Mapped[int | None] = mapped_column(nullable=True)
+    error: Mapped[str | None] = mapped_column(String, nullable=True)
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    __table_args__ = (
+        CheckConstraint(
+            "status IN ('pending', 'streaming', 'succeeded', 'failed')", name="status_values"
+        ),
+        Index("uq_model_call_turn_attempt", "turn_id", "attempt", unique=True),
+    )
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 02c9b2b..178a60a 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -46,6 +46,28 @@ async def client(db):
                 raise
 
     app.dependency_overrides[db_session.get_db] = override
+    # 后台 Run 任务使用测试库会话工厂与 mock 流（生产为独立工厂与真实传输）
+    app.state.session_factory = db
+    app.state.gw_transport = _mock_openai_stream()
+    import os as _os
+
+    _os.environ.setdefault("HJ_MOCK_KEY", "test-key")
+    from app.models import LogicalModel, ProviderEndpoint
+
+    async with db() as session:
+        session.add(LogicalModel(id="default-chat", capabilities=["chat"]))
+        session.add(
+            ProviderEndpoint(
+                logical_model_id="default-chat",
+                provider="openai_compatible",
+                name="mock-endpoint",
+                base_url="http://mock.internal",
+                model_id="mock-chat",
+                api_key_env="HJ_MOCK_KEY",
+                priority=1,
+            )
+        )
+        await session.commit()
     transport = ASGITransport(app=app, raise_app_exceptions=False)
     async with AsyncClient(transport=transport, base_url="https://test") as c:
         yield c
@@ -56,3 +78,25 @@ async def login(client, email: str) -> dict[str, str]:
     resp = await client.post("/v1/auth/sessions", json={"email": email})
     assert resp.status_code == 201, resp.text
     return {"Hunjing-CSRF-Token": resp.json()["csrf_token"]}
+
+
+def _mock_openai_stream():
+    """OpenAI-compatible 流式 mock：真实 SSE 字节流（data: {...} 分块 + usage）。"""
+    import httpx
+
+    sse = (
+        b'data: {"choices":[{"delta":{"content":"\\u4f60\\u597d"}}]}\n\n'
+        b'data: {"choices":[{"delta":{"content":","}}]}\n\n'
+        b'data: {"choices":[{"delta":{"content":"\\u9b42\\u9559"}}]}\n\n'
+        b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\n'
+        b'data: {"choices":[],"usage":{"prompt_tokens":12,'
+        b'"completion_tokens":6,"total_tokens":18}}\n\n'
+        b'data: [DONE]\n\n'
+    )
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            200, headers={"content-type": "text/event-stream"}, content=sse
+        )
+
+    return httpx.MockTransport(handler)
diff --git a/backend/tests/test_conversations.py b/backend/tests/test_conversations.py
index 19381de..a8aaa13 100644
--- a/backend/tests/test_conversations.py
+++ b/backend/tests/test_conversations.py
@@ -20,6 +20,22 @@ async def _create_conversation(client: AsyncClient, headers: dict) -> str:
     return resp.json()["id"]
 
 
+async def _wait_run_terminal(db, run_id: str, deadline_s: float = 10.0) -> str:
+    """轮询等待 Run 终态（后台任务推进）。"""
+    import asyncio
+
+    from app.models import Run as RunModel
+
+    deadline = asyncio.get_event_loop().time() + deadline_s
+    while asyncio.get_event_loop().time() < deadline:
+        async with db() as session:
+            run = await session.get(RunModel, run_id)
+            if run is not None and run.status in {"completed", "failed", "cancelled"}:
+                return run.status
+        await asyncio.sleep(0.05)
+    raise AssertionError(f"Run {run_id} 未在 {deadline_s}s 内到达终态")
+
+
 async def _post_message(client: AsyncClient, headers: dict, cid: str, key: str) -> object:
     return await client.post(
         f"/v1/conversations/{cid}/messages",
@@ -29,7 +45,7 @@ async def _post_message(client: AsyncClient, headers: dict, cid: str, key: str)
 
 
 class TestMessagePipeline:
-    async def test_create_and_post_message(self, client: AsyncClient) -> None:
+    async def test_create_and_post_message(self, client: AsyncClient, db) -> None:
         headers = await login(client, "alice@example.com")
         cid = await _create_conversation(client, headers)
         resp = await _post_message(client, headers, cid, "idem-1")
@@ -37,7 +53,9 @@ class TestMessagePipeline:
         body = resp.json()
         assert body["message"]["id"].startswith("message_")
         assert body["message"]["content"][0]["id"].startswith("block_")
-        assert body["run"]["status"] == "completed"  # 空状态闭环
+        assert body["run"]["status"] == "queued"  # 创建即返回；终态由后台执行推进
+        final_status = await _wait_run_terminal(db, body["run"]["id"])
+        assert final_status == "completed"
 
     async def test_idempotent_replay_returns_original(self, client: AsyncClient) -> None:
         headers = await login(client, "bob@example.com")
@@ -95,19 +113,25 @@ class TestMessagePipeline:
 
 
 class TestRunSse:
-    async def test_sse_snapshot_and_terminal(self, client: AsyncClient) -> None:
+    async def test_sse_snapshot_and_terminal(self, client: AsyncClient, db) -> None:
         headers = await login(client, "grace@example.com")
         cid = await _create_conversation(client, headers)
         resp = await _post_message(client, headers, cid, "idem-sse")
         run_id = resp.json()["run"]["id"]
+        await _wait_run_terminal(db, run_id)
         sse = await client.get(f"/v1/runs/{run_id}/events")
         assert sse.status_code == 200
         assert sse.headers["content-type"].startswith("text/event-stream")
         lines = [ln for ln in sse.text.splitlines() if ln.startswith("data: ")]
         events = [json.loads(ln.removeprefix("data: ")) for ln in lines]
         assert events[0]["type"] == "run.status.changed"
-        assert events[0]["data"]["status"] == "completed"
-        assert events[-1]["type"] == "message.completed"
+        assert events[0]["data"]["status"] == "running"
+        assert events[-1]["type"] == "run.status.changed"
+        assert events[-1]["data"]["status"] == "completed"
+        types = [e["type"] for e in events]
+        assert "content.block.started" in types
+        assert "content.block.delta" in types  # 真实流产生
+        assert "message.completed" in types
 
     async def test_cross_domain_run_returns_404(self, client: AsyncClient) -> None:
         headers = await login(client, "heidi@example.com")
@@ -240,19 +264,24 @@ class TestContentBlocks:
 
 @pytest.mark.usefixtures("client")
 class TestSseReplay:
-    async def test_events_replayed_in_order_with_sequence(self, client: AsyncClient) -> None:
+    async def test_events_replayed_in_order_with_sequence(
+        self, client: AsyncClient, db
+    ) -> 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"]
+        await _wait_run_terminal(db, 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 events[-1]["type"] == "run.status.changed"
+        # delta 序号单调（契约 (block_id, delta_index) 去重基础）
+        deltas = [e for e in events if e["type"] == "content.block.delta"]
+        assert [d["data"]["delta_index"] for d in deltas] == list(range(len(deltas)))
         assert sum(1 for e in events if e["type"] == "content.block.completed") >= 1
 
 
@@ -607,3 +636,75 @@ class TestCharacters:
         await login(client, "char8-evil@example.com")
         resp = await client.get(f"/v1/characters/{c['id']}")
         assert resp.status_code in (401, 403, 404)
+
+
+@pytest.mark.usefixtures("client")
+class TestGatewayExecution:
+    async def test_model_call_usage_and_assistant_message(self, client, db) -> None:
+        """端到端：mock 流 → ModelCall usage/TTFT + assistant 权威消息。"""
+        from sqlalchemy import select as sa
+
+        from app.models import Message as M
+        from app.models import ModelCall
+
+        headers = await login(client, "gw1@example.com")
+        cid = await _create_conversation(client, headers)
+        resp = await _post_message(client, headers, cid, "idem-gw1")
+        run_id = resp.json()["run"]["id"]
+        assert await _wait_run_terminal(db, run_id) == "completed"
+        async with db() as session:
+            calls = list((await session.execute(sa(ModelCall))).scalars().all())
+            assert len(calls) == 1
+            assert calls[0].status == "succeeded"
+            assert calls[0].usage == {
+                "prompt_tokens": 12, "completion_tokens": 6, "total_tokens": 18,
+            }
+            assert calls[0].ttft_ms is not None
+            asst = list(
+                (
+                    await session.execute(
+                        sa(M).where(M.run_id == run_id, M.role == "assistant")
+                    )
+                ).scalars().all()
+            )
+            assert len(asst) == 1
+            assert asst[0].content[0]["type"] == "text"
+            assert asst[0].content[0]["status"] == "completed"
+
+    async def test_no_endpoint_fails_closed(self, client, db) -> None:
+        """无可用端点：Run 显式失败并留 error 事件，不静默降级（ADR-016）。"""
+        from sqlalchemy import update
+
+        from app.models import ProviderEndpoint
+
+        async with db() as session:
+            await session.execute(update(ProviderEndpoint).values(enabled=False))
+            await session.commit()
+        try:
+            headers = await login(client, "gw2@example.com")
+            cid = await _create_conversation(client, headers)
+            resp = await _post_message(client, headers, cid, "idem-gw2")
+            run_id = resp.json()["run"]["id"]
+            assert await _wait_run_terminal(db, run_id) == "failed"
+        finally:
+            async with db() as session:
+                await session.execute(update(ProviderEndpoint).values(enabled=True))
+                await session.commit()
+        from sqlalchemy import select as sa_select
+
+        from app.models import RunEvent as RunEventModel
+
+        async with db() as session:
+            evs = list(
+                (
+                    await session.execute(
+                        sa_select(RunEventModel)
+                        .where(RunEventModel.run_id == run_id)
+                        .order_by(RunEventModel.sequence)
+                    )
+                ).scalars().all()
+            )
+        types = [e.type for e in evs]
+        assert "content.block.failed" in types
+        failed = next(e for e in evs if e.type == "content.block.failed")
+        assert failed.data["error"] == "no_endpoint_available"