games-development-ai/game-cloud/scripts/wg1_stub_worker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WG1 stub worker（P3 W-G1 派发面·3b-A，2026-06-14）——证全 plumbing 用的极简桩 worker。

【用途】把 P3 派发面端到端打通：执行器 POST §6.1 job → 本 stub 收 job → 读已知 pong bundle 作 engineBundle →
  把 §6.1 result-out 经回调入参 POST 回 /admin-api/aigc/dify/callback-internal（@PermitAll 内网回调子路由）→
  后端 handleCallback 建版本/组包/落包入 feed。**不调任何 LLM、不打包**——只回固定 pong bundle，专证管线。
  真 worker（L1 run.py：裸 openai SDK 写 GameHostFactory → esbuild 打 __GameBundle → 自检）留 3b-B。

【与 §6.1 接缝的关系】本 stub 实现 §6.1 job-in（收）/ result-out（回）两个 JSON 契约：
  - 收 job：解 job_id/brief/gameId/templateId/callback.target（其余字段透传/忽略）。
  - 回 result：但回调入参形态对齐【后端 DifyCallbackReqVO】（traceId/status/templateId/gameConfig/engineBundle），
    而非直接回 §6.1 result-out 全形态——因为 3b-A 复用 HTTP 回调路（worker→/dify/callback-internal），
    后端回调入参 VO 即落点；§6.1 result-out 的 selfCheck/costYuan 等字段是 worker↔pipeline「队列路」语义，
    HTTP 回调路下由后端从 DifyCallbackReqVO 取 engineBundle（resolveEngineBundleText）落包，二者等价。

【gameConfig 占位（§5.2 占位框）】engine 路下游戏逻辑全在 engineBundle 内，gameConfig 仍不可空（GamePackage
  required + additionalProperties:false）；故回最小合法占位 {templateId, title, theme, engineDriven:true}。

【鉴权（§6.5 auth 孤儿解法）】worker 是内部服务无 RBAC token，回调走后端新开的 @PermitAll 内网回调子路由
  /admin-api/aigc/dify/callback-internal（复用 handleCallback 唯一写入路径）。本 stub 因此【不带 Authorization 头】。

【运行（mini-desktop）】
  python3 game-cloud/scripts/wg1_stub_worker.py [--port 9301] \
      [--bundle game-runtime/games/_wg1-gen/pong/bundle.iife.js] \
      [--callback http://localhost:48080/admin-api/aigc/dify/callback-internal]
  执行器 application-staging.yaml 配 aigc.executor.worker-url=http://localhost:9301/wg1/generate 指向本 stub。

@author 造梦AI（P3 派发面·stub）
"""

import argparse
import json
import os
import sys
import threading
import urllib.request
import urllib.error
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# ---------- 默认常量（mini-desktop staging 口径）----------
# pong bundle 固定产物（brick-2 已放此路径，267KB iife 顶层全局名 __GameBundle）
DEFAULT_BUNDLE = "game-runtime/games/_wg1-gen/pong/bundle.iife.js"
# 内网免鉴权回调子路由（后端 AdminAigcTaskController.difyCallbackInternal，@PermitAll）
DEFAULT_CALLBACK = "http://localhost:48080/admin-api/aigc/dify/callback-internal"
DEFAULT_PORT = 9301


def log(msg):
    """简单带前缀日志（stdout，便于 nohup 落盘排障）。"""
    print(f"[wg1-stub-worker] {msg}", flush=True)


class StubWorkerState:
    """进程级状态（bundle 文本一次读入常驻 + 配置）。"""

    def __init__(self, bundle_text, callback_url):
        self.bundle_text = bundle_text   # pong bundle 全文（engineBundle 真值）
        self.callback_url = callback_url  # 回调子路由 URL


def build_callback_payload(job, bundle_text):
    """
    据收到的 §6.1 job 组后端回调入参（DifyCallbackReqVO 形态）。

    :param job: 收到的 §6.1 job-in（dict）
    :param bundle_text: pong bundle 全文（engineBundle 真值）
    :return: 回调入参 dict（traceId/status/templateId/gameConfig/engineBundle）
    """
    trace_id = job.get("traceId") or job.get("job_id")  # traceId 贯穿全链（与 job_id 同值）
    template_id = job.get("templateId") or "generic"
    brief = job.get("brief") or ""
    # gameConfig 最小合法占位（§5.2）：engine 路承载 meta，非填参玩法。title 由 brief 派生（截断防超 META_TITLE_MAX）。
    title = (brief[:20] if brief else "弹球(stub)")
    game_config = {
        "templateId": template_id,   # 须与包级 templateId 一致（validateSucceededPayload 一致性校验）
        "title": title,              # 供 meta 段派生 cover/summary
        "theme": "stub-pong",        # 占位主题
        "engineDriven": True,        # 标记本款走引擎 bundle 路（非填参 GameConfig）
    }
    return {
        "traceId": trace_id,
        "status": "succeeded",
        "templateId": template_id,
        "gameConfig": game_config,
        "engineBundle": bundle_text,  # 回调入参携 pong bundle 全文 → 后端 resolveEngineBundleText 取值落包
        "assets": [],
        # qualityScore/failureReason 省略（succeeded 路不需要）
    }


def post_callback(callback_url, payload):
    """
    把回调入参 POST 回后端内网回调子路由（不带 Authorization；@PermitAll 内网可达）。

    :param callback_url: 回调 URL
    :param payload: 回调入参 dict
    :return: (http_status, body_text)
    """
    data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        callback_url, data=data, method="POST",
        headers={"Content-Type": "application/json", "tenant-id": "0"},  # tenant-id 头兜底（@TenantIgnore 已忽略，带上无害）
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            body = resp.read().decode("utf-8", "replace")
            return resp.status, body
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", "replace") if e.fp else ""
        return e.code, body
    except Exception as e:  # 网络异常：返回 -1 + 异常文本（不抛，便于日志记真因）
        return -1, f"{type(e).__name__}: {e}"


def handle_job_async(state, job):
    """
    异步处理一个 job（读 bundle → 组回调入参 → POST 回调）。
    放后台线程，模拟「worker 异步生成、执行器投递握手后即返回、任务留 RUNNING 等回调」的真实时序。
    """
    trace_id = job.get("traceId") or job.get("job_id")
    game_id = job.get("gameId")
    log(f"开始处理 job trace_id={trace_id}, gameId={game_id}, bundleLen={len(state.bundle_text)}")
    payload = build_callback_payload(job, state.bundle_text)
    status, body = post_callback(state.callback_url, payload)
    # 截断 body 防刷屏
    body_brief = body[:300] if body else ""
    log(f"回调完成 trace_id={trace_id}, callbackHttp={status}, body={body_brief}")


def make_handler(state):
    """工厂：绑定 state 的 HTTP handler 类。"""

    class Handler(BaseHTTPRequestHandler):
        # 静默默认访问日志（用自定义 log）
        def log_message(self, fmt, *args):
            return

        def _send_json(self, code, obj):
            data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
            self.send_response(code)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            self.wfile.write(data)

        def do_GET(self):
            # 健康探针：GET /health → 200
            if self.path.rstrip("/") in ("/health", "/wg1/health"):
                self._send_json(200, {"ok": True, "bundleLen": len(state.bundle_text)})
            else:
                self._send_json(404, {"error": "not found"})

        def do_POST(self):
            # 派发入口：POST /wg1/generate ← 执行器投递 §6.1 job
            if self.path.rstrip("/") != "/wg1/generate":
                self._send_json(404, {"error": "not found", "path": self.path})
                return
            length = int(self.headers.get("Content-Length", "0") or "0")
            raw = self.rfile.read(length) if length > 0 else b""
            try:
                job = json.loads(raw.decode("utf-8")) if raw else {}
            except Exception as e:
                log(f"job 解析失败：{e}")
                self._send_json(400, {"accepted": False, "error": "bad job json"})
                return
            trace_id = job.get("traceId") or job.get("job_id")
            log(f"收到 job trace_id={trace_id}, templateId={job.get('templateId')}, gameId={job.get('gameId')}")
            # 投递握手：立即回 202（投递成功），后台异步「生成」+回调（模拟真实 worker 异步时序）
            threading.Thread(target=handle_job_async, args=(state, job), daemon=True).start()
            self._send_json(202, {"accepted": True, "job_id": job.get("job_id"), "traceId": trace_id})

    return Handler


def main():
    parser = argparse.ArgumentParser(description="WG1 stub worker（P3 派发面 3b-A：回固定 pong bundle 证管线）")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"监听端口（默认 {DEFAULT_PORT}）")
    parser.add_argument("--bundle", default=DEFAULT_BUNDLE, help="engineBundle 真值文件（默认 pong bundle）")
    parser.add_argument("--callback", default=DEFAULT_CALLBACK, help="后端内网回调子路由 URL")
    args = parser.parse_args()

    # 读 bundle（相对路径基于 cwd；建议在 repo 根跑）
    bundle_path = args.bundle
    if not os.path.isfile(bundle_path):
        log(f"❌ bundle 文件不存在：{bundle_path}（cwd={os.getcwd()}）")
        sys.exit(2)
    with open(bundle_path, "r", encoding="utf-8") as f:
        bundle_text = f.read()
    # 出厂自检：bundle 必含 __GameBundle 全局名（宿主据此取工厂；缺则后端落包后宿主取不到 → demo 兜底）
    if "__GameBundle" not in bundle_text:
        log(f"❌ bundle 缺 __GameBundle 全局名：{bundle_path}（宿主无法取 GameHostFactory）")
        sys.exit(2)
    log(f"bundle 已加载：{bundle_path}（{len(bundle_text)} 字符，含 __GameBundle ✓）")
    log(f"回调子路由：{args.callback}")

    state = StubWorkerState(bundle_text, args.callback)
    server = ThreadingHTTPServer(("0.0.0.0", args.port), make_handler(state))
    log(f"stub worker 已启动 → http://0.0.0.0:{args.port}/wg1/generate（Ctrl-C 退出）")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        log("收到中断，退出")
        server.shutdown()


if __name__ == "__main__":
    main()