别把 Agent 的 SSE 绑在一次提问上：用会话 ID 扛住中途插话和授权

用户提交一句话
POST /chat  （响应直接是 text/event-stream）
token 一段段推回来
模型说完，连接关掉

同一条 WSS
  ├─ 服务端：token / 工具状态 / 等授权
  └─ 客户端：纠偏 / 同意 / 拒绝

用户点「允许删除」
前端 abort 当前 EventSource
再 POST 一次，带上 approval
后端新开一条 SSE
前端重新订阅、重新拼状态

1. POST /conversations
   body: { "command_id": "cmd_...", "message": "帮我把登录页改成 Vue" }
   resp: { "conversation_id": "c_..." }
   SSE 随后推出 run_started，带上这一轮的 run_id

2. GET  /conversations/c_.../events
   Accept: text/event-stream
   （订阅这个会话的事件；连接掉了可以再挂，按 Last-Event-ID 续）

3. 用户看到方向不对
   POST /conversations/c_.../commands
   body: { "command_id": "cmd_...", "type": "redirect", "message": "先列风险" }

4. Agent 要删文件
   SSE 推出：id: 1042
             event: permission_required
             data: {"run_id":"r_...","call_id":"call_12","tool":"rm"}
   用户点允许
   POST /conversations/c_.../approvals
   body: { "command_id": "cmd_...", "call_id": "call_12", "allow": true }

命令收件箱 ──→  当前 run ──→ 事件日志 ──→ 每个 SSE 订阅者各自一份队列
                 ↑
            LLM / 工具

queued → generating → waiting_approval → executing_tool → completed
                         ↓                    ↓
                      rejected         cancellation_requested

模型吐出 tool_calls
循环看见这是高风险工具
不执行，先往事件日志推：
  id: 1042
  event: permission_required
  data: {"run_id":"r_...","call_id":"call_12","tool":"rm","args_digest":"..."}
然后 await pending_approvals[call_id]  这一把 Future

pending_approvals: dict[call_id, Future]

from dataclasses import dataclass, field
import asyncio, json, time, uuid

@dataclass
class Event:
    event_id: int
    name: str
    data: dict


class Conversation:
    def __init__(self, conversation_id: str, owner_id: str):
        self.id = conversation_id
        self.owner_id = owner_id
        self.closed = False
        self.event_log: list[Event] = []
        self.subscribers: set[asyncio.Queue] = set()
        self.inbox: asyncio.Queue = asyncio.Queue()
        self.seen_commands: set[str] = set()
        self.pending_approvals: dict[str, asyncio.Future] = {}
        self.generation: asyncio.Task | None = None
        self.current_run_id: str | None = None

    def publish(self, name: str, data: dict) -> Event:
        event = Event(len(self.event_log) + 1, name, data)
        self.event_log.append(event)
        for q in list(self.subscribers):
            q.put_nowait(event)
        return event

    def subscribe(self, last_event_id: int) -> asyncio.Queue:
        q: asyncio.Queue = asyncio.Queue()
        for event in self.event_log:
            if event.event_id > last_event_id:
                q.put_nowait(event)
        self.subscribers.add(q)
        return q

    def request_approval(self, call_id: str, payload: dict) -> asyncio.Future:
        loop = asyncio.get_running_loop()
        fut = loop.create_future()
        self.pending_approvals[call_id] = fut
        self.publish("permission_required", payload)
        return fut

    def resolve_approval(self, call_id: str, allow: bool, actor_id: str) -> dict:
        fut = self.pending_approvals.get(call_id)
        if fut is None:
            return {"status": "unknown_or_done"}
        if fut.done():
            return {"status": "already_resolved", "allow": fut.result()}
        # 这里还应校验 actor_id、参数摘要、过期时间
        fut.set_result(allow)
        return {"status": "ok", "allow": allow}

    async def loop(self):
        while not self.closed:
            cmd = await self.inbox.get()
            if cmd["type"] == "close":
                break
            await self.start_run(cmd)

            while self.generation and not self.generation.done():
                get_cmd = asyncio.create_task(self.inbox.get())
                done, _ = await asyncio.wait(
                    {self.generation, get_cmd},
                    return_when=asyncio.FIRST_COMPLETED,
                )
                if self.generation in done:
                    get_cmd.cancel()
                    break
                nxt = get_cmd.result()
                if nxt["type"] in {"redirect", "close", "cancel"}:
                    self.generation.cancel()
                    try:
                        await self.generation
                    except asyncio.CancelledError:
                        pass
                    self.publish("generation_interrupted", {
                        "run_id": self.current_run_id,
                    })
                    if nxt["type"] == "redirect":
                        await self.start_run(nxt)
                    else:
                        self.closed = nxt["type"] == "close"
                        break
                else:
                    await self.inbox.put(nxt)

    async def start_run(self, cmd: dict):
        self.current_run_id = str(uuid.uuid4())
        self.publish("run_started", {"run_id": self.current_run_id})
        self.generation = asyncio.create_task(self.generate(cmd))

    async def generate(self, cmd: dict):
        run_id = self.current_run_id
        try:
            # async for chunk in provider.stream(...):
            #     self.publish("token", {"run_id": run_id, "text": chunk})
            # 需要授权时：
            # allow = await self.request_approval(call_id, payload)
            self.publish("run_completed", {"run_id": run_id})
        except asyncio.CancelledError:
            # 这里 close 的是你这边的 HTTP 读流。
            # 供应商会不会停，取决于它支不支持服务端取消。
            self.publish("run_cancelled", {"run_id": run_id})
            raise

from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()
conversations: dict[str, Conversation] = {}


def require_owner(conv: Conversation, request: Request) -> None:
    user_id = getattr(request.state, "user_id", None)
    if user_id is None or user_id != conv.owner_id:
        raise HTTPException(403, "forbidden")


class CreateConversation(BaseModel):
    command_id: str
    message: str


class Command(BaseModel):
    command_id: str
    type: str = "redirect"
    message: str


class Approval(BaseModel):
    command_id: str
    call_id: str
    allow: bool


@app.post("/conversations")
async def create_conversation(body: CreateConversation, request: Request):
    conv = Conversation(str(uuid.uuid4()), request.state.user_id)
    conversations[conv.id] = conv
    asyncio.create_task(conv.loop())
    await conv.inbox.put({"type": "user", **body.model_dump()})
    # run_id 由随后的 run_started 事件带出来，这里不抢一个可能还是 None 的值
    return {"conversation_id": conv.id}


@app.post("/conversations/{conversation_id}/commands")
async def inject_command(conversation_id: str, body: Command, request: Request):
    conv = conversations.get(conversation_id)
    if not conv:
        raise HTTPException(404)
    require_owner(conv, request)
    if body.command_id in conv.seen_commands:
        return {"ok": True, "deduped": True}
    conv.seen_commands.add(body.command_id)
    await conv.inbox.put(body.model_dump())
    return {"ok": True}


@app.post("/conversations/{conversation_id}/approvals")
async def approve(conversation_id: str, body: Approval, request: Request):
    conv = conversations.get(conversation_id)
    if not conv:
        raise HTTPException(404)
    require_owner(conv, request)
    if body.command_id in conv.seen_commands:
        return {"ok": True, "deduped": True}
    conv.seen_commands.add(body.command_id)
    return conv.resolve_approval(body.call_id, body.allow, request.state.user_id)


@app.get("/conversations/{conversation_id}/events")
async def events(
    conversation_id: str,
    request: Request,
    last_event_id: str | None = Header(default=None, alias="Last-Event-ID"),
):
    conv = conversations.get(conversation_id)
    if not conv:
        raise HTTPException(404)
    require_owner(conv, request)
    after = int(last_event_id) if last_event_id else 0
    q = conv.subscribe(after)

    async def gen():
        try:
            while not conv.closed:
                try:
                    item = await asyncio.wait_for(q.get(), timeout=15)
                except asyncio.TimeoutError:
                    yield ": ping\n\n"
                    continue
                payload = json.dumps(item.data, ensure_ascii=False)
                yield f"id: {item.event_id}\nevent: {item.name}\ndata: {payload}\n\n"
        finally:
            conv.subscribers.discard(q)

    return StreamingResponse(
        gen(),
        media_type="text/event-stream; charset=utf-8",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

const { conversation_id } = await fetch("/conversations", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ command_id: crypto.randomUUID(), message }),
}).then((r) => r.json());

const es = new EventSource(`/conversations/${conversation_id}/events`);
es.addEventListener("token", (e) => appendText(JSON.parse(e.data).text));
es.addEventListener("permission_required", (e) => showApproval(JSON.parse(e.data)));
es.addEventListener("run_completed", () => markIdle());
es.addEventListener("generation_interrupted", () => markInterrupted());

await fetch(`/conversations/${conversation_id}/commands`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({
    command_id: crypto.randomUUID(),
    type: "redirect",
    message: "先列风险，先别改代码",
  }),
});

事件为主、回程稀疏     →  HTTP 命令 + 应用层会话 ID + SSE 订阅
两头持续互发           →  WSS
问一次、不插话、不授权 →  请求体直接带流也行，别过度设计

HTTP           →  往 conversation 里扔带 command_id 的命令
conversation   →  养活上下文、当前 run、审批 Future、事件日志
SSE            →  订阅事件日志；不因交互主动拆，断了按 event_id 续