games-development-ai/tier2/gen-worker/service/app.py

"""service/app.py —— tier2 富游戏自治线 · Agent Service 服务壳(U1;on AgentScope v2.0.2)。

把「本地 CLI 直驱裸 Agent」升级成经官方 `agentscope.app.create_app` 的 Agent Service:多租户 / 多会话 /
REST 触发(POST /chat,fire-and-forget)+ SSE 事件流(GET /sessions/{sid}/stream),session 状态与
跨会话 MessageBus 走 Redis(mini-infra)。

════════════════════════════════════════════════════════════════════════════
【create_app 的真实形态(源码核验 app/_app.py:33 + examples/agent_service/main.py)】
  create_app(storage, message_bus, workspace_manager, *, extra_credentials, extra_middlewares,
             extra_agent_middlewares, extra_agent_tools, custom_subagent_templates,
             custom_agent_cls, title, version) -> fastapi.FastAPI
  关键事实(决定本壳怎么接,逐条源码核验):
   1. **create_app 不收 agent 实例,也不收 agent 工厂**。agent 由框架在【每个 chat 回合】内部装配
      (_service/_chat.py:_run_impl 第 317 行 `self._agent_cls(...)`)。我们能注入的只有装配的「零件」:
        · custom_agent_cls —— Agent 子类(默认就用内置 Agent,本壳传 None);
        · extra_agent_tools —— 异步工厂 (user_id, agent_id, session_id) -> list[ToolBase],每回合调一次,
          产出的工具加进 toolkit 的 "basic" 组(_app.py:104-111,_chat.py:240-251 的 extra_factory);
        · extra_agent_middlewares —— 异步工厂同签名,产出的中间件接在框架中间件(InboxMiddleware /
          StateChangeMiddleware / ToolOffloadMiddleware)之后(_app.py:95-103,_chat.py:273-280);
        · custom_subagent_templates —— 工作室 Team 的 worker 蓝图(本壳暂留空,见下「工作室 Team 现状」)。
   2. **模型不在 create_app 配,在 session 的 chat_model_config 配**,且必须先注册成 AgentScope
      Credential。_chat.py:297-303 从 `session_record.config.chat_model_config` 经 get_model 解析模型,
      凭据来自存储里的 CredentialRecord。**故 M3(MiniMax-M3 经 new-api 走 Anthropic 原生)要先
      POST /credential 注册成 anthropic_credential,再在建 session 时引用**(端点/口令见 bootstrap.py 与
      docs/内网凭据与端点.md;凭据档由 new-api 网关托管,base_url=host 根、key=NEWAPI_KEY)。
   3. **系统提示词 / context_config / react_config 在 AgentRecord 配**(POST /agent),_chat.py:319/323/324
      从 agent_record.data 取。故 tier2 单写 agent 的 system_prompt(roles.writer_system)、历史压缩
      (config.build_context_config)、ReAct 轮数(ReActConfig(max_iters=...))经建 agent 时写进 AgentRecord。
   4. **session 状态(AgentState:context/summary/cur_iter/permission/tool/tasks)持久化在 Redis**,每回合
      reload(_chat.py:315)→ 跑 → 回写(_chat.py:441)。这就是 session「续接」的原生机制,无需自写。
   5. **storage 与 message_bus 都只有 Redis 实现,无 in-memory/SQLite 版本**,生产必须配 Redis 且需 AUTH。
   6. **无 CLI / serve 命令**:启动 = 在本文件里调 uvicorn.run(见文末 main + deploySteps)。

【与 CLI 单写主链(agent_loop.studio.run_studio)的关系 —— 诚实差距,见 followups】
  run_studio 是「外层 Python 编排」:它在 Agent 的 ReAct 循环【之外】套了一圈有界 resume(agent 过早
  停下就带 verdict 反馈踹回去续修)、阶段 1 工作室多 agent 设计、收口后的 L2/L3 软检、成本台账、落库寻址。
  而 create_app 的 chat 是【每回合一次 agent.reply_stream】、由前端经 REST 决定要不要再发一轮——
  外层编排的位置变成了「服务消费方 / 前端 / 控制面」。**所以本壳不可能、也不应该把 run_studio 整个塞进
  create_app**(那会与框架的 per-turn 模型打架)。本壳做的是:**复用 run_studio 用的同一批装配零件**
  (roles 系统提示词 / build_toolkit 九工具 / 四熔断 + trace 中间件 / BYPASS 权限 / 历史压缩),把它们
  接到 create_app 的扩展点上,**绝不重写任何生成逻辑**;有界 resume / 设计团队 / L2/L3 / 成本 / 落库
  这些「循环外」能力如何在服务态落位,作为 followups 明确标差距(见文末)。

【session 三路初始化 —— 汇到同一 agent 装配点】(实现见 bootstrap.py)
  三路全部经 create_app 现成的 REST 端点达成(不自造端点、不旁路框架),最终都走 _chat.py 的同一
  装配点(custom_agent_cls + extra_agent_tools + extra_agent_middlewares + session 的模型/系统提示词):
    ① 新建(模板起手):POST /agent → POST /sessions(新 workspace_id)→ POST /chat(kick 文本引导 agent
       scaffold_init 起手)。
    ② 续接会话:对已存在的 (agent_id, session_id) 直接再 POST /chat;框架自动 reload 该 session 的
       AgentState(含已写历史 / cur_iter / tasks)续跑——即「从 SessionRecord.state resume」。
    ③ 加载已有工程迭代:绑到「工程所在 workdir 对应的 agent_id」再开 session。AgentScope 的
       LocalWorkspaceManager workdir = basedir/agent_id(_local_workspace_manager.py:116,**按 agent_id 不按
       workspace_id**),故一款游戏工程 = 一个 agent_id;迭代 = 在该 agent_id 上新开 / 续用 session。
       (注:tier2 九工具自己的工程目录另在 game-runtime/games/_tier2-gen/<game_id>,与 AgentScope
       workspace 解耦——见下「九工具 workdir 与 AgentScope workspace 的关系」。)

【九工具 workdir 与 AgentScope workspace 的关系(重要,避免误解)】
  tier2 九工具(scaffold_init/write_source/build/run_gates/finish 等)是 worker.toolkit.build_toolkit 产出的
  纯 Python 闭包,它们读写自己的工程目录 game-runtime/games/_tier2-gen/<game_id>(run._workdir),
  **不依赖 AgentScope 的 workspace.workdir**。本壳把 game_id 绑成 AgentScope 的 session_id,九工具据此
  各自管文件。AgentScope 的 LocalWorkspaceManager 是 create_app 的硬性必填项(每个 chat run 都要它),
  但本线生成产物落在九工具自己的目录,AgentScope workspace 基本只承载框架内置的 filesystem 工具
  (本壳不靠它们生成)。这是有意为之:九工具是 spike 已验证的生成主链,零改接进来。

【工作室 Team(多 agent 设计)现状】
  run_studio 阶段 1 的工作室星形多 agent 设计在 CLI 线是「纯库 import Agent」实现(design_team.py:13 明示
  studio.py 无 create_app、拿不到部署态 Team 原语)。create_app 自带部署态 Team(AgentCreate/TeamCreate/
  TeamSay + SubAgentTemplate),但语义与 CLI 线的 design_team 不同构。本壳先把 custom_subagent_templates
  留空(单写 agent 主链先服务化跑通),把「设计阶段在服务态如何落位(用部署态 Team 重写 or 设计阶段作为
  前置 REST 调用产出 design_text 再塞进单写 agent 的 system_prompt)」列为 followup。

【惰性 import 红线】agentscope[full] 的 app、redis 客户端在 6c6g 未装。本模块顶层【绝不】import
  agentscope.app / redis;所有重依赖只在 build_app / 工厂函数体内 import,保证 6c6g py_compile + import 过。
  worker.config 在顶层 import agentscope(模型客户端),故对它也惰性 import(在工厂体内),避免顶层连带炸。
════════════════════════════════════════════════════════════════════════════
"""

from __future__ import annotations

import os
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any

# infra_config 不 import 重依赖,顶层 import 安全(6c6g 可用)。
from . import infra_config

# 仅类型检查期引用 FastAPI / agentscope 类型,运行时不 import(6c6g 无这些包也能 py_compile)。
if TYPE_CHECKING:  # pragma: no cover
    from fastapi import FastAPI


# ── 包内/直跑兼容:让顶层包 `worker` / `observability` 可解析(与 studio.py 同款兜底)──
#   本文件在 tier2/gen-worker/service/app.py;把 gen-worker/ 加进 sys.path,使 `worker.*` 可 import
#   (目录名 gen-worker 含连字符不可直接 import,但子目录 worker/ 是合法包名)。
_GEN_WORKER_DIR = Path(__file__).resolve().parents[1]
if str(_GEN_WORKER_DIR) not in sys.path:
    sys.path.insert(0, str(_GEN_WORKER_DIR))


# 服务级常量:tier2 Agent Service 的标题(OpenAPI docs 显示)。
SERVICE_TITLE = "tier2-rich-game-agent-service"


def _extract_function_tools(toolkit: Any) -> list:
    """从 worker.toolkit.build_toolkit 返回的 Toolkit 里抠出 FunctionTool 列表(供 extra_agent_tools 工厂用)。

    为什么要抠:create_app 的 extra_agent_tools 工厂约定返回 `list[ToolBase]`(_types.py:21),框架会把
    它们加进【框架自建的】toolkit 的 basic 组;而 tier2 的 build_toolkit 直接返回一个【已组装好的】
    Toolkit 实例。本壳不想绕过框架自建 toolkit(那会丢掉 workspace 工具 / planning / schedule 等),所以
    取折中:从 tier2 Toolkit 里取出九个 FunctionTool 实例,交给框架合并。

    2.0.2 Toolkit 内部把工具存在 `tools`(dict[str, ToolBase])或等价结构;不同小版本字段名可能微调,
    故按几个候选名探测,全失败则诚实抛(让启动期就暴露,而非运行期静默少工具)。
    """
    # 2.0.2 Toolkit 把注册的工具按【组】存在 self.tool_groups:每个 ToolGroup 的 .tools 是 list[ToolBase];
    #   build_toolkit 传给 Toolkit(tools=...) 的九工具被 __init__ 包进默认 "basic" 组(源码核验
    #   tool/_toolkit.py:127-135 建 basic 组 tools=tools or []、:515 for tool in group.tools)。
    #   故从所有组收 .tools 即得九个 FunctionTool 实例(MCP 客户端在 group.mcps、不走这里——服务态 MCP 另议)。
    #   2026-06-24 实证修复:此前按 tools/_tools/function_tools 扁平探测全空 → 每个 chat run 起手即抛、
    #   服务态生成零产出(B1 控制面真驱时暴露;P4 只验了服务能启,从没真跑过一回合 chat)。
    groups = getattr(toolkit, "tool_groups", None)
    if groups:
        extracted: list = []
        for group in groups:
            extracted.extend(getattr(group, "tools", None) or [])
        if extracted:
            return extracted
    # 兜底:旧扁平探测路径(不同小版本若把工具放扁平映射时仍可用)。
    for attr in ("tools", "_tools", "function_tools"):
        container = getattr(toolkit, attr, None)
        if isinstance(container, dict) and container:
            return list(container.values())
        if isinstance(container, (list, tuple)) and container:
            return list(container)
    raise RuntimeError(
        "无法从 tier2 Toolkit 抠出 FunctionTool 列表(tool_groups[*].tools 与 tools/_tools/function_tools 均空):"
        "AgentScope 2.0.2 Toolkit 内部结构可能与预期不符,请核对 build_toolkit 产物结构后调整 _extract_function_tools。"
    )


async def _tier2_tools_factory(user_id: str, agent_id: str, session_id: str) -> list:
    """extra_agent_tools 工厂:每个 chat 回合产出 tier2 九工具(绑定 game_id = session_id)。

    签名严格对齐 AgentToolFactory((user_id, agent_id, session_id) -> Awaitable[list[ToolBase]],_types.py:21)。
    把 AgentScope 的 session_id 当成 tier2 的 game_id —— 九工具据此各自管自己的工程目录
    (game-runtime/games/_tier2-gen/<session_id>),与 AgentScope workspace 解耦(见文件头说明)。

    重依赖(worker.toolkit → 经 worker.config 连带 import agentscope)在此惰性 import,保证 6c6g import app.py 不炸。

    Args:
        user_id: 框架注入的用户 id(本线暂不按用户隔离工程目录,留作 followup;先用 session_id 作 game_id)。
        agent_id: 框架注入的 agent id。
        session_id: 框架注入的 session id —— 即本款游戏工程的 game_id。

    Returns:
        list[ToolBase]:九个 FunctionTool 实例(scaffold_init/write_source/validate_datatable/build/
        run_gates/read_verdict/finish 等;具体面见 worker.toolkit.build_toolkit)。
    """
    # 惰性 import:worker.toolkit 顶层经 worker.config 会 import agentscope(6c6g 无);故只在回合内 import。
    from worker.toolkit import Tier2Session, build_toolkit  # noqa: PLC0415 —— 惰性 import 红线

    # game_id 绑 session_id:一个 AgentScope session = 一款游戏工程(九工具据 game_id 管目录)。
    #   play_spec 在服务态由控制面/前端经后续接口下发(spike 期先 None;真玩门 run_gates 仍可跑,
    #   只是无定制驱动规格——按品类注册表默认)。这是与 CLI 的差异点之一,列入 followup。
    session = Tier2Session(session_id)
    toolkit = build_toolkit(session)
    return _extract_function_tools(toolkit)


async def _tier2_middlewares_factory(user_id: str, agent_id: str, session_id: str) -> list:
    """extra_agent_middlewares 工厂:每个 chat 回合产出 tier2 的四熔断 + trace 中间件。

    签名严格对齐 AgentMiddlewareFactory((user_id, agent_id, session_id) -> Awaitable[list[MiddlewareBase]])。
    产出的中间件接在框架中间件之后(_chat.py:273-280);trace 列在 breaker 前 → 它是更外层洋葱
    (先 ingest 事件再进熔断巡检,与 studio.py 同序)。trace_id 用 session_id 贯穿本款生成。

    重依赖(worker.middleware → 经包顶层连带 agentscope)在此惰性 import。

    Returns:
        list[MiddlewareBase]:[Tier2TraceMiddleware(trace_id=session_id), CircuitBreakerMiddleware()]。
    """
    # 惰性 import:worker.middleware 经 worker 包顶层会牵出 agentscope;故只在回合内 import。
    from worker.middleware import (  # noqa: PLC0415 —— 惰性 import 红线
        CircuitBreakerMiddleware,
        Tier2TraceMiddleware,
    )

    # trace 在外、breaker 在内(与 studio.py middlewares=[tracer, breaker] 同序);sink=None → 步留内存,
    #   真落库 sink 随控制面 phase-1 接(同 studio 现状)。每回合一组新实例(熔断计数按回合,非跨回合累计——
    #   这是与 CLI 单局累计的差异点,列入 followup:跨回合累计 ¥ 硬闸需把计数挂到 session 维度)。
    tracer = Tier2TraceMiddleware(trace_id=session_id)
    breaker = CircuitBreakerMiddleware()
    return [tracer, breaker]


def build_app(*, title: str = SERVICE_TITLE) -> "FastAPI":
    """组装 tier2 Agent Service 的 FastAPI app(create_app + Redis storage/message_bus + 本地 workspace)。

    这是部署入口要调的工厂:`from service.app import build_app; app = build_app()`,再 uvicorn 起(见文末 main)。
    所有重依赖(agentscope.app / redis 客户端,经 LocalWorkspaceManager 还要 fastapi)在此惰性 import,
    保证 6c6g 仅 import 本模块(不调 build_app)时不炸——真正连基建在 mini-desktop 调 build_app 时才发生。

    装配口径(逐条对应 create_app 扩展点,见文件头):
      · storage        = RedisStorage(**infra.redis storage 库)        —— session/agent/team 记录持久化
      · message_bus    = RedisMessageBus(**infra.redis message_bus 库)  —— 跨会话事件 / inbox / run-lock
      · workspace_manager = LocalWorkspaceManager(basedir=...)          —— 框架必填(本线产物另落九工具目录)
      · extra_agent_tools       = _tier2_tools_factory                  —— 注入九工具(per-turn)
      · extra_agent_middlewares = _tier2_middlewares_factory            —— 注入四熔断 + trace(per-turn)
      · custom_subagent_templates = []  （工作室 Team 暂留空,见文件头 followup）
      · custom_agent_cls = None         （用内置 Agent;tier2 装配靠零件注入,不必子类化）

    Returns:
        fastapi.FastAPI:create_app 组装好的 app,直接交 uvicorn。
    """
    # ── 惰性 import 重依赖(6c6g 无;只在真正建 app 时 import)──
    from agentscope.app import create_app  # noqa: PLC0415
    from agentscope.app.message_bus import RedisMessageBus  # noqa: PLC0415
    from agentscope.app.storage import RedisStorage  # noqa: PLC0415
    from agentscope.app.workspace_manager import LocalWorkspaceManager  # noqa: PLC0415

    # Redis 连接参数从 infra.yaml 取(storage 与 message_bus 走不同逻辑库隔离键空间;均带 password)。
    storage_params = infra_config.redis_params(for_message_bus=False)
    bus_params = infra_config.redis_params(for_message_bus=True)

    # 本地 workspace 根目录:放在 gen-worker 下的 _service-workspaces(与九工具的 _tier2-gen 目录分开;
    #   这里只承载框架内置工具的工作区,本线生成产物不落这里——见文件头)。可被 env 覆盖换盘。
    ws_basedir = os.environ.get(
        "TIER2_SERVICE_WORKSPACES",
        str(_GEN_WORKER_DIR / "_service-workspaces"),
    )

    # 可追溯启动日志(口令脱敏):连的哪个 Redis、workspace 落哪。
    print(
        f"[tier2-service] build_app: redis storage={storage_params['host']}:{storage_params['port']}"
        f"/db{storage_params['db']} bus=db{bus_params['db']} "
        f"auth={'on' if storage_params['password'] else 'off'} ws_basedir={ws_basedir}",
        flush=True,
    )

    return create_app(
        storage=RedisStorage(**storage_params),
        message_bus=RedisMessageBus(**bus_params),
        workspace_manager=LocalWorkspaceManager(basedir=ws_basedir),
        # tier2 装配零件经扩展点注入(每个 chat 回合调一次)。
        extra_agent_tools=_tier2_tools_factory,
        extra_agent_middlewares=_tier2_middlewares_factory,
        # 工作室 Team worker 蓝图暂留空(单写主链先服务化;设计阶段服务态落位见文件头 followup)。
        custom_subagent_templates=[],
        # 用内置 Agent(tier2 单写 agent 靠零件注入装配,不必子类化)。
        custom_agent_cls=None,
        title=title,
    )


def main() -> None:
    """部署入口:起 uvicorn 跑 tier2 Agent Service(只在 mini-desktop 装好 agentscope[full]+redis 后跑)。

    端口 / host 经 env 调(默认 0.0.0.0:8200,避开 wg1/其它服务常用口)。reload=False(生产形态;
    调试可 export TIER2_SERVICE_RELOAD=1)。注意 reload 模式需用 import 字符串而非 app 对象,故两分支。
    """
    import uvicorn  # noqa: PLC0415 —— 惰性 import(6c6g 无 uvicorn)

    host = os.environ.get("TIER2_SERVICE_HOST", "0.0.0.0")
    port = int(os.environ.get("TIER2_SERVICE_PORT", "8200"))
    reload = os.environ.get("TIER2_SERVICE_RELOAD", "0").strip().lower() in ("1", "true", "yes", "on")

    print(f"[tier2-service] 启动 Agent Service:http://{host}:{port}  reload={reload}", flush=True)
    if reload:
        # reload 模式必须传 import 字符串(uvicorn 要能在 worker 子进程重导入);指向本模块的 module-level `app`。
        #   子进程重导入 service.app 时,module-level `app` 只在 TIER2_SERVICE_EAGER_APP=1 时才建——故在此
        #   先显式置位,保证 reload worker 进程导入即建出真 app(否则会服务到 app=None)。非 reload 分支不需要。
        os.environ["TIER2_SERVICE_EAGER_APP"] = "1"
        uvicorn.run("service.app:app", host=host, port=port, reload=True)
    else:
        # 生产形态:直接传 app 对象(此处才真正连 Redis 建 app)。
        uvicorn.run(build_app(), host=host, port=port)


# reload 模式 / `uvicorn service.app:app` 直起时用的 module-level app。
#   ⚠️ 仅当显式设置 TIER2_SERVICE_EAGER_APP=1 时才在 import 期建 app(会连 Redis)——默认不建,
#   保证 6c6g `import service.app` 不触发重依赖 import / 不连基建(惰性红线)。mini-desktop 用
#   `uvicorn service.app:app` 起时,设这个 env 让 module-level app 就绪;或直接 `python -m service.app`
#   走 main()(推荐,无需 eager)。
app = None
if os.environ.get("TIER2_SERVICE_EAGER_APP", "0").strip().lower() in ("1", "true", "yes", "on"):
    app = build_app()  # pragma: no cover —— 仅 mini-desktop 显式开启时执行


if __name__ == "__main__":
    main()