src/webui/api.py
Rama main · réplica de lectura del repositorio original.
"""WebUI 后端 API -- 为六个管理页面提供 JSON 数据接口。"""
from __future__ import annotations
from src.engine.modules import adventure_runtime_state, economy_state
from src.adventures.outcomes import OutcomeError
from src.adventures.progress import ProgressError
from src.assistant_agent_limits import DEFAULT_TASK_MAX_TOKENS
from src.generation_limits import DEFAULT_TEXT_GEN_MAX_TOKENS
import asyncio
import json
import logging
import time
from copy import deepcopy
from pathlib import Path
from collections.abc import Awaitable, Callable
from typing import Any, Literal
from src.engine.character_utils import calc_hp_from_rule, get_rule_attr_config, make_default_character, parse_tavern_card, roll_attributes
from src.engine.economy import queue_proposal, resolve_auto_reward_policy
from src.engine.game_instance import GameRegistry
from src.engine.game_state import GameState
from src.engine import persistence
from src.webui.services.adventure_materialization import materialize_world_seed
from src.engine.memory_outbox import pending_memory_deliveries, pending_memory_reversals
from src.engine.participant_view import Viewer
from src.lorebook.store import LorebookStore
from src.adventures import AdventureBundleLoader, AdventureResolver
from src.adventures.scene_context import AdventureSceneContext
from src.adventures.registry import AdventureSource, AdventureSourceRegistry
from src.memory.delta import MemoryStore
from src.rules.rule_system import RuleSystem
from src.rules.loader import RuleBundleLoader
from src.rulesets.builtin import (
build_default_ruleset_registry,
default_adventure_runtime_requirement,
default_module_runtime_requirement,
)
from src.rulesets.registry import RulesetRuntimeRegistry
from src.engine.world_template import load_world_template
from src.content_modules.projection import ContentProjectionService
from src.webui.services import adventures, asr, avatars, bot_access, bot_extensions, character_cards, characters, content, content_pack_maps, game_controls, game_lifecycle, game_master, game_media, game_packages, game_queries, generated_images, generation, knowledge, kp_questions, logs, lorebooks, map_backgrounds, maps, tavern, turns, worlds, rules, ruleset_advancement, ruleset_builder, ruleset_gameplay, ruleset_rest, plugins, modules, scene_images, speech, system, tunnel, announcements, assistant, hub, legal, manual_rolls, check_reveals, seat_tokens, storytelling
from src.webui.services import lorebook_usage
from src.webui.services import standing_portraits
from src.webui.services import combat_extension as combat_extension_service
from src.webui.services import adventure_runtime
from src.game_profiles.service import ProfileService, ProfileDependencies
from src.webui.services.ai_seats import AISeatDependencies, AISeatRuntime, AISeatService
from src.webui.services.check_decisions import CheckDecisionDependencies, CheckDecisionService
from src.webui.services.game_preferences import GamePreferenceDependencies, GamePreferenceService
from src.webui.services.run_story import RunStoryDependencies, RunStoryService
from src.webui.services.resource_recovery import RecoveryDependencies, RecoveryService
from src.game_preferences import UNSET, validate_preferences
from src.webui.services.check_modes import CheckModeDependencies, CheckModeService
from src.commands.hosted_combat import HostedCombatScheduler
from src.commands import adventure_effects
from src.commands.scene_transition import SceneTransitionNarrator
from src.webui.services import ruleset_characters
from src.webui.services import character_card_import
from src.webui.services import content_sync
from src.lorebook.sync import LorebookSyncExporter, LorebookSyncImporter
from src.webui.services import memory as memory_service
from src.webui.services import module_import_adapters
from src.webui.services._common import _parse_game_key, _is_safe_world_id
# Preserve the existing public import used by callers while the service owns the value.
CANONICAL_ENTRY_DEFAULTS = lorebooks.CANONICAL_ENTRY_DEFAULTS
logger = logging.getLogger("trpg")
def can_modify_character(session_uid: str, target_uid: str, gm_uid: str, owner: bool = False) -> bool:
"""角色卡更新/删除权限：仅本人、该局 GM 或已登录 owner 可改，空身份拒绝。"""
if not session_uid:
return False
return session_uid == target_uid or session_uid == gm_uid or owner
class WebAPI:
"""WebUI 数据接口，供前端页面调用。
方法签名保持简单，便于通过 HTTP/WebSocket 暴露。
所有返回值为 JSON 可序列化的字典。
"""
async def legal_document(self, document_name: str, language: str) -> dict[str, Any]:
return await self._legal.document(document_name, language)
async def fetch_public_content_text(
self,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
) -> str:
return await self._content.fetch_text(
path,
force_refresh=force_refresh,
allow_cached=allow_cached,
)
def public_content_disk_age(self, path: str) -> float | None:
"""公共内容磁盘缓存文件的年龄（秒）；无缓存返回 None。"""
return self._content.disk_cache_age(path)
async def fetch_public_content_json(
self,
path: str,
*,
force_refresh: bool = False,
allow_cached: bool = True,
) -> dict[str, Any] | None:
return await self._content.fetch_json(
path,
force_refresh=force_refresh,
allow_cached=allow_cached,
)
async def current_legal_documents(self) -> dict[str, Any]:
return await self._legal.current_documents()
def legal_acceptance_payload(
self,
documents: dict[str, Any],
language: str,
) -> dict[str, dict[str, str]]:
return legal.acceptance_payload(documents, language)
def legal_accepted(
self,
state: dict[str, Any],
documents: dict[str, Any] | None = None,
) -> bool:
return legal.accepted(state, documents)
def record_legal_acceptance(
self,
state: dict[str, Any],
*,
acceptance: dict[str, Any],
documents: dict[str, Any],
accepted_at: str,
) -> None:
return legal.record_acceptance(
state,
acceptance=acceptance,
documents=documents,
accepted_at=accepted_at,
)
def legal_bundle_version(self, documents: dict[str, Any]) -> str:
return legal.bundle_version(documents)
def __init__(self, registry: GameRegistry, lorebook: LorebookStore,
memory: MemoryStore, rules_dir: Path,
handler=None, llm_client=None, worlds_dir: Path | None = None,
adventures_dir: Path | None = None,
character_gen_max_tokens: int = 4096,
text_gen_max_tokens: int = DEFAULT_TEXT_GEN_MAX_TOKENS, plugin_host=None, hub_client=None,
speech_service=None, asr_service=None, imagegen_service=None,
ruleset_registry: RulesetRuntimeRegistry | None = None,
content_cache_dir: Path | None = None,
content_mirrors: Any | None = None,
config_state: dict | None = None,
save_config: Callable[[], None] | None = None,
assistant_agent_max_tokens: int = 4096,
assistant_agent_task_max_tokens: int = DEFAULT_TASK_MAX_TOKENS):
self._reg = registry
self._lore = lorebook
self._mem = memory
self._rules_dir = rules_dir
self._handler = handler
self._config_state = config_state if config_state is not None else {}
self._ai_seat_runtime = AISeatRuntime(config_state=lambda: self._config_state)
self._ai_seats = AISeatService(AISeatDependencies(
get_instance=lambda game_key: self._reg.get(_parse_game_key(game_key)),
save_instance=self._reg.save,
resume=lambda game_key, uid: turns.resume_after_control_change(
self._turn_dependencies, game_key, seat_uid=uid,
),
), self._ai_seat_runtime)
if handler is not None and config_state is not None:
handler.ai_seat_runtime = self._ai_seat_runtime
self._save_config = save_config or (lambda: None)
handler_rulesets = getattr(handler, "ruleset_registry", None)
self._ruleset_registry: RulesetRuntimeRegistry = (
ruleset_registry or handler_rulesets or build_default_ruleset_registry()
)
self._check_decisions = CheckDecisionService(CheckDecisionDependencies(
get_instance=self.get_game_instance, save_instance=self._reg.save,
))
self._game_preferences = GamePreferenceService(GamePreferenceDependencies(
get_instance=self.get_game_instance, save_instance=self._reg.save,
load_rule=self._load_rule_for_game, get_runtime=self._load_runtime_for_game,
get_defaults=lambda: validate_preferences((getattr(self, '_config_state', {}) or {}).get('game_defaults', {})),
))
self.resource_recovery = RecoveryService(RecoveryDependencies(
get_instance=self.get_game_instance, save_instance=self._reg.save,
load_rule=self._load_rule_for_game,
advance_adventure_world=lambda instance: adventure_runtime.advance_adventure_world(
self._adventure_runtime_dependencies, instance,
),
))
self.run_story = RunStoryService(RunStoryDependencies(
get_instance=self.get_game_instance, save_instance=self._reg.save,
load_rule=self._load_rule_for_game, get_runtime=self._load_runtime_for_game,
get_llm=lambda: self._llm_client,
register_instance=self._reg.register,
project_scene=lambda instance, **options: self.scene_context_provider(instance, **options),
bind_carried_game=lambda candidate: self._game_lorebooks.bind_carried_game(candidate),
prepare_adventure_draft=lambda draft: self.review_adventure_script(draft),
))
if callable(getattr(handler, 'set_game_defaults_provider', None)):
handler.set_game_defaults_provider(lambda: validate_preferences((getattr(self, '_config_state', {}) or {}).get('game_defaults', {})))
self._check_modes = CheckModeService(CheckModeDependencies(
get_instance=lambda key: self._reg.get(_parse_game_key(key)), save_instance=self._reg.save,
load_rule=self._load_rule_for_game, rulesets=self._ruleset_registry,
))
legacy_world_loader = getattr(handler, "_load_world_template", None)
self._game_query_dependencies = game_queries.GameQueryDependencies(
list_instances=self._reg.list_all,
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
load_world_template=(
legacy_world_loader if callable(legacy_world_loader) else None
),
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
)
self._llm_client = llm_client
self._worlds_dir = worlds_dir or (Path(__file__).parent.parent.parent / "templates" / "worlds")
self._builtin_adventures_dir = (
Path(__file__).parent.parent.parent / "templates" / "adventures"
).resolve()
# FIX-02 §4.1：唯一的 AdventureResolver。builtin（runtime 数据目录里的
# 同步副本或发布目录）/ user（不含 builtin 标记的目录）/ plugin（模组声明）
# 三类来源由它统一解析；WebAPI 与 D&D Runtime 共用同一个实例
# （见下方 set_adventure_resolver 注入）。
self._dnd_module_catalogs: list[tuple[str, Any]] = []
self._adventure_resolver = AdventureResolver.from_directories(
self._builtin_adventures_dir,
(
adventures_dir
if adventures_dir is not None
and adventures_dir.resolve() != self._builtin_adventures_dir
else None
),
)
self._adventure_source_registry = self._adventure_resolver.registry
self._adventure_loader = self._adventure_resolver
# 让 runtime（D&D load_adventure）也走同一个 resolver，消除双解析路径。
self._ruleset_registry.set_adventure_resolver(self._adventure_resolver)
# FIX-03 §5.2：把"模组 catalog 来源"注入 runtime；gameplay catalog 的
# 链路与解析语义留在 runtime，WebAPI 不拥有它。
self._ruleset_registry.set_module_content_sources(self.module_content_sources)
# Some embedders construct GameHandler before WebAPI and provide it a
# distinct registry. Its RoundProcessor is still a production path,
# so inject the same application-owned sources there as well instead
# of letting it silently fall back to a private Adventure loader.
handler_registry = getattr(self._handler, "ruleset_registry", None)
if handler_registry is not None and handler_registry is not self._ruleset_registry:
handler_registry.set_adventure_resolver(self._adventure_resolver)
handler_registry.set_module_content_sources(self.module_content_sources)
self._character_cards_path = self._reg.save_dir.parent / "character_cards.json"
self._character_dependencies = characters.CharacterDependencies(
games=characters.CharacterGameDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
save_instance=self._reg.save,
),
rules=characters.CharacterRuleDependencies(
rules_dir=self._rules_dir,
load_rule_by_id=self._load_rule_by_id,
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
),
assets=characters.CharacterAssetDependencies(
lorebook=self._lore,
load_world_template=self._load_world_template,
avatar_file=lambda asset_id: self.avatar_file(asset_id),
generated_image_file=lambda asset_id: self.generated_image_file(
asset_id,
),
content_projection=getattr(handler, "content_projection", None),
standing_portrait_file=lambda asset_id: self.standing_portrait_file(
asset_id,
),
),
# GAL P8：投影侧校验个人音色仍存在（失效省略，保持未配置）。
voice_profile_exists=(
lambda voice: self._voice_profile_exists(voice)
),
# Table characters auto-saved into the library: never plugin cards.
save_character_card=lambda character: character_cards.save_character_card(
self._character_card_dependencies, character, from_game=True,
),
apply_economy_effects=(
handler.commit_deferred_economy_effects
if handler is not None
and callable(getattr(handler, "commit_deferred_economy_effects", None))
else None
),
schedule_economy_scene_image=(
handler.schedule_deferred_economy_scene_image
if handler is not None
and callable(
getattr(handler, "schedule_deferred_economy_scene_image", None)
)
else None
),
apply_economy_memory=(
self._mem.apply_economy_delta if self._mem is not None else None
),
reverse_economy_memory=(
self._mem.reverse_economy_delta if self._mem is not None else None
),
)
self._ruleset_character_dependencies = (
ruleset_characters.RulesetCharacterDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
save_instance=self._reg.save,
load_rule_by_id=self._load_rule_by_id,
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
read_cards=lambda: character_cards._read_cards(
self._character_card_dependencies,
),
write_cards=lambda cards: character_cards._write_cards(
self._character_card_dependencies, cards,
),
validate_portrait=lambda portrait: characters._validated_portrait(
self._character_dependencies.assets, portrait,
),
)
)
self._character_card_dependencies = character_cards.CharacterCardDependencies(
cards_path=self._character_cards_path,
ruleset_card_metadata=lambda card: ruleset_characters.runtime_metadata_for_card(
self._ruleset_character_dependencies, card,
),
normalize_ruleset_card=lambda card: ruleset_characters.normalize_character_card_blueprint(
self._ruleset_character_dependencies, card,
),
is_ruleset_card=lambda card: ruleset_characters.card_has_rules_aware_lifecycle(
self._ruleset_character_dependencies, card,
),
lorebook=self._lore,
rebuild_lorebook_index=self._rebuild_lorebook_index,
)
self._storytelling = storytelling.StorytellingService(
storytelling.StorytellingDependencies(
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
save_instance=self._reg.save,
validate_portrait=lambda portrait: characters.validate_character_portrait(
self._character_dependencies.assets, portrait,
),
)
)
self._ruleset_advancement_dependencies = (
ruleset_advancement.RulesetAdvancementDependencies(
load_rule_by_id=self._load_rule_by_id,
ruleset_registry=self._ruleset_registry,
)
)
self._card_advancement_dependencies = (
ruleset_advancement.CardAdvancementDependencies(
read_cards=lambda: character_cards._read_cards(
self._character_card_dependencies,
),
write_cards=lambda cards: character_cards._write_cards(
self._character_card_dependencies, cards,
),
load_rule_by_id=self._load_rule_by_id,
runtime_for_card=lambda card: ruleset_characters.runtime_for_card(
self._ruleset_character_dependencies, card,
),
)
)
self._live_advancement_dependencies = (
ruleset_advancement.LiveAdvancementDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
save_instance=self._reg.save,
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
)
)
self._ruleset_rest_dependencies = ruleset_rest.RulesetRestDependencies(
load_rule_by_id=self._load_rule_by_id,
ruleset_registry=self._ruleset_registry,
)
self._live_ruleset_rest_dependencies = ruleset_rest.LiveRulesetRestDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
save_instance=self._reg.save,
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
)
self._content = content.PublicContentService(
content_cache_dir,
mirror_manager=content_mirrors,
)
self._legal = legal.LegalService(
legal.LegalDependencies(
fetch_public_json=self.fetch_public_content_json,
fetch_public_text=self.fetch_public_content_text,
)
)
self._avatars = avatars.AvatarService(self._reg.save_dir.parent / "avatars")
self._standing_portraits = standing_portraits.StandingPortraitService(
self._reg.save_dir.parent / "standing-portraits",
is_referenced=lambda asset_id: self._standing_portrait_is_referenced(asset_id),
)
self._scene_images = scene_images.SceneImageService(
scene_images.SceneImageDependencies(
images_dir=self._reg.save_dir.parent / "scene-images",
load_world_template=self._load_world_template,
get_rule_template=self.get_rule_template,
generated_image_file=lambda asset_id: self.generated_image_file(asset_id),
plugin_asset_path=self.plugin_asset_path,
)
)
self._map_backgrounds = map_backgrounds.MapBackgroundService(
self._reg.save_dir.parent / "map-backgrounds",
lambda asset_id: self.generated_image_file(asset_id),
)
self._content_map_dependencies = content_pack_maps.ContentMapDependencies(
resolve_background_file=self.resolve_map_background_file,
)
self._memory_service = memory_service.MemoryService(
memory_service.MemoryDependencies(
repository=self._mem,
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
)
)
self._bot_access = bot_access.BotAccessService(
bot_access.BotAccessDependencies(
registry=self._reg,
parse_game_key=_parse_game_key,
)
)
self._game_logs = logs.GameLogService(
logs.LogDependencies(
registry=self._reg,
parse_game_key=_parse_game_key,
)
)
self._tavern_import = tavern.TavernImportService(
tavern.TavernImportDependencies(
lorebook=self._lore,
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
rebuild_lorebook_index=self._rebuild_lorebook_index,
)
)
self._game_packages = game_packages.GamePackageService(
game_packages.GamePackageDependencies(
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
state_path_for=self._reg.save_package_state_path,
import_save_zip=self._reg.import_save_zip,
resolve_scene_image_file=self.resolve_scene_image_file,
resolve_map_background_file=self.resolve_map_background_file,
save_scene_image_upload=self.save_scene_image_upload,
save_map_background_upload=self.save_map_background_upload,
)
)
self._game_controls = game_controls.GameControlService(
game_controls.GameControlDependencies(
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
save_instance=self._reg.save,
load_rule=self._load_rule_for_game,
# 延迟到调用时解析：回合依赖在下面才创建，而且这里只做组合，
# 不把 WebAPI 自己变成 service locator。
resume_after_control_change=lambda game_key, seat_uid: (
turns.resume_after_control_change(
self._turn_dependencies, game_key, seat_uid=seat_uid,
)
),
)
)
self._manual_rolls = manual_rolls.ManualRollService(manual_rolls.ManualRollDependencies(_parse_game_key, self._reg.get, self._reg.save, self._load_rule_for_game))
import secrets
self._profile_games = ProfileService(ProfileDependencies(get=self._reg.get, register=self._reg.register, save=self._reg.save, roll=lambda: secrets.randbelow(6) + 1))
self._check_reveals = check_reveals.CheckRevealService(check_reveals.CheckRevealDependencies(_parse_game_key, self._reg.get, self._reg.save))
self._seat_tokens = seat_tokens.SeatTokenService(seat_tokens.SeatTokenDependencies(_parse_game_key, self._reg.get, self._reg.save))
self._game_master = game_master.GameMasterService(
game_master.GameMasterDependencies(
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
save_instance=self._reg.save,
load_rule=self._load_rule_for_game,
load_runtime=self._load_runtime_for_game,
generate_recap=(
self._handler.generate_story_recap
if self._handler is not None
else None
),
drain_economy_outbox=self._drain_economy_outbox,
)
)
self._game_media = game_media.GameMediaService(
game_media.GameMediaDependencies(
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
save_instance=self._reg.save,
load_world_template=(
self._load_world_template if self._worlds_dir else None
),
get_lore_world=(
self._lore.get_world if self._lore is not None else None
),
refresh_lorebook_index=(
self._refresh_game_lorebook_index
if self._handler is not None
else None
),
resolve_rule_id=self._project_game_rule_id,
resolve_default_scene_image=self.resolve_default_scene_image,
materialize_scene_image=self.materialize_scene_image,
)
)
self.character_gen_max_tokens = character_gen_max_tokens
self.text_gen_max_tokens = text_gen_max_tokens
self._plugins = plugin_host
self._plugin_host_dependencies = plugins.PluginHostDependencies(
plugin_host=self._plugins,
)
self._module_dependencies = modules.ModuleDependencies(
plugin_host=self._plugins,
adventure_registry=self._adventure_source_registry,
lorebook_store=self._lore,
ruleset_registry=self._ruleset_registry,
list_instances=self._reg.list_all,
# FIX-01 §3.7：绑定存档保护必须覆盖所有持久化存档（paused / ended /
# 加载失败但元数据可读），不能只看内存 active GameInstance。
list_save_metadata=lambda: persistence.scan_save_metadata(self._reg),
# FIX-01 §3.6：destructive module action 前刷新来源注册表。
refresh_adventure_sources=self._sync_plugin_adventure_sources,
default_runtime_requirement=default_module_runtime_requirement,
# FIX-06 §8：在线模组区块复用既有插件市场索引（含 installed 判定）。
list_marketplace_plugins=(
self._plugins.marketplace_plugins if self._plugins is not None else None
),
)
self._adventure_runtime_dependencies = adventure_runtime.AdventureRuntimeDependencies(
resolve_binding=lambda instance: self._adventure_resolver.resolve_binding(
getattr(instance, "adventure_binding", None) or {},
str(getattr(instance, "language", "") or ""),
),
materialize_world_seed=materialize_world_seed,
rules_evaluator=lambda instance: self._adventure_runtime_adapter(
instance, "adventure_rules_evaluator",
),
reward_converter=lambda instance: self._adventure_runtime_adapter(
instance, "adventure_reward_converter",
),
# FIX-08 §10 步骤 12：item_reward 的权威出口（既有 reward proposal /
# GM 确认 / rollback 快照），adventure runtime 绝不直接写 inventory。
queue_reward_intents=self._queue_adventure_reward_intents,
save_instance=self._reg.save,
)
def resolve_scene_binding(instance):
self._sync_plugin_adventure_sources()
return self._adventure_runtime_dependencies.resolve_binding(instance)
def legacy_scene_context(instance, *, viewer_is_gm):
from src.rulesets.contracts import CurrentSceneRuntime
runtime = self._load_runtime_for_game(instance)
return runtime.scene_objective_context(instance, viewer_is_gm=viewer_is_gm) if isinstance(runtime, CurrentSceneRuntime) else {}
adventure_scene_context = AdventureSceneContext(resolve_scene_binding, legacy_projection=legacy_scene_context)
def scene_context(instance, *, viewer_is_gm):
from src.engine.modules import run_story
if instance.adventure_binding:
return adventure_scene_context(instance, viewer_is_gm=viewer_is_gm)
return run_story.project(instance, viewer_is_gm=viewer_is_gm)
self.scene_context_provider = scene_context
if self._handler is not None and hasattr(self._handler, "set_scene_context_provider"):
self._handler.set_scene_context_provider(scene_context)
if self._handler is not None and hasattr(self._handler, "set_adventure_world_advance"):
self._handler.set_adventure_world_advance(
lambda instance: adventure_runtime.advance_adventure_world(
self._adventure_runtime_dependencies, instance,
)
)
if self._handler is not None and hasattr(self._handler, "set_adventure_run_initializer"):
# Reset/restart build a new run of the same game: initialize v2
# progress + world seed like the creation transaction (v1 bindings
# are not re-resolved, preserving their previous restart path).
self._handler.set_adventure_run_initializer(
lambda instance: adventure_runtime.initialize_adventure_new_run(
self._adventure_runtime_dependencies, instance,
)
)
self._world_dependencies = worlds.WorldDependencies(
lorebook=self._lore,
worlds_dir=self._worlds_dir,
plugin_host=self._plugins,
llm_client=self._llm_client,
character_gen_max_tokens=self.character_gen_max_tokens,
invalidate_lorebook_index=(
self._handler.invalidate_matcher_for_world
if self._handler is not None
else None
),
list_instances=self._reg.list_all,
content_projection=getattr(self._handler, "content_projection", None),
)
self._plugin_content_dependencies = plugins.PluginContentDependencies(
plugin_host=self._plugins,
store=plugins.PluginContentStoreDependencies(
lorebook=self._lore,
list_games=self._reg.list_all,
list_character_cards=lambda: character_cards.list_character_cards(
self._character_card_dependencies,
),
save_character_card=lambda card: character_cards.save_character_card(
self._character_card_dependencies, card,
),
delete_character_card=lambda card_id: character_cards.delete_character_card(
self._character_card_dependencies, card_id,
),
save_entry=lambda entry: worlds.save_entry(
self._world_dependencies, entry,
),
save_book_entry=lambda book_id, entry: lorebooks.save_lorebook_entry(
self._lorebook_dependencies(), book_id, entry,
),
),
portraits=plugins.PluginPortraitDependencies(
plugin_asset_path=lambda plugin_id, relative_path: plugins.plugin_asset_path(
self._plugin_host_dependencies, plugin_id, relative_path,
),
avatar_file=lambda asset_id: self._avatars.file(asset_id),
generated_image_file=lambda asset_id: self._generated_images.image_file(
asset_id,
),
save_avatar_upload=lambda file_data, file_name: self._avatars.save_upload(
file_data, file_name,
),
),
)
self._plugin_lifecycle_dependencies = plugins.PluginLifecycleDependencies(
plugin_host=self._plugins,
content=self._plugin_content_dependencies,
)
self._plugin_export_dependencies = plugins.PluginExportDependencies(
plugin_host=self._plugins,
lorebook=self._lore,
rules_dir=self._rules_dir,
list_character_cards=lambda: character_cards.list_character_cards(
self._character_card_dependencies,
),
media=plugins.PluginExportMediaDependencies(
package_scene_image=lambda reference, files: self._scene_images.package(
reference, files,
),
package_content_map=lambda *args, **kwargs: content_pack_maps.package_content_map(
self._content_map_dependencies, *args, **kwargs,
),
avatar_file=lambda asset_id: self._avatars.file(asset_id),
generated_image_file=lambda asset_id: self._generated_images.image_file(
asset_id,
),
plugin_asset_path=lambda plugin_id, relative_path: plugins.plugin_asset_path(
self._plugin_host_dependencies, plugin_id, relative_path,
),
),
adventure_resolver=self._adventure_resolver,
)
self._adventure_dependencies = adventures.AdventureDependencies(
adventure_loader=self._adventure_loader,
adventure_registry=self._adventure_source_registry,
adventure_resolver=self._adventure_resolver,
list_instances=self._reg.list_all,
load_rule_by_id=self._load_rule_by_id,
ruleset_registry=self._ruleset_registry,
default_runtime_requirement=default_adventure_runtime_requirement,
builtin_adventures_dir=self._builtin_adventures_dir,
)
from src.webui.services.content_text_review import ContentReviewDependencies, ContentTextReview
self.content_text_review = ContentTextReview(ContentReviewDependencies(
self._world_dependencies, self._lorebook_dependencies(), self._character_card_dependencies,
self._adventure_dependencies, self._reg.save_dir.parent / "content-import-receipts",
))
from src.webui.services.module_resources import ModuleResourceDependencies, ModuleResources
self.module_resources_service = ModuleResources(ModuleResourceDependencies(self._module_dependencies, self._lorebook_dependencies()))
from src.webui.services.module_forks import ModuleForkDependencies, ModuleForks
self.module_forks_service = ModuleForks(ModuleForkDependencies(
self.module_resources_service, self._world_dependencies, self._lorebook_dependencies(), self._character_card_dependencies,
self._adventure_dependencies, self._avatars.save_upload, self._scene_images.save_upload,
self._avatars.preview_upload, self._scene_images.preview_upload,
))
self._ruleset_gameplay_dependencies = (
ruleset_gameplay.RulesetGameplayDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
load_rule_for_game=self._load_rule_for_game,
ruleset_registry=self._ruleset_registry,
resolve_adventure_binding=lambda adventure_id, runtime, world_id, language, source_kind="", source_id="": adventures.resolve_binding_for_runtime(
self._adventure_dependencies,
adventure_id,
runtime,
world_id,
language,
source_kind=source_kind,
source_id=source_id,
),
save_instance=self._reg.save,
complete_adventure_node=lambda instance, node_id, recipient_uid, actor_uid, choice_id="": adventure_runtime.complete_adventure_node(
self._adventure_runtime_dependencies,
instance,
node_id,
recipient_uid=recipient_uid,
actor_uid=actor_uid,
choice_id=choice_id,
),
request_adventure_choice=lambda instance, node_id, choice_id, actor_uid: adventure_runtime.request_adventure_choice(
self._adventure_runtime_dependencies, instance, node_id, choice_id, actor_uid,
),
apply_memory_delta=(
self._mem.apply_delta if self._mem is not None else None
),
resolve_llm_client=lambda: self._llm_client,
seat_runtime=self._ai_seat_runtime if config_state is not None else None,
)
)
self._hosted_combat_scheduler = None
if config_state is not None and handler is not None and hasattr(handler, "set_hosted_combat_resume"):
self._hosted_combat_scheduler = HostedCombatScheduler(
get_instance=self._reg.get,
seat_for=lambda instance: ruleset_gameplay.current_hosted_seat(self._ruleset_gameplay_dependencies, instance),
resume=lambda key, uid: ruleset_gameplay.resume_authoritative_combat(self._ruleset_gameplay_dependencies, key, uid),
)
handler.set_hosted_combat_resume(self._hosted_combat_scheduler.schedule)
self._turn_dependencies = turns.TurnDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
ruleset_registry=self._ruleset_registry,
load_rule_for_game=self._load_rule_for_game,
prepare_round_checks_ai=getattr(
self._handler, "prepare_round_checks_ai", None,
),
prepare_round_checks=getattr(
self._handler, "prepare_round_checks", None,
),
fill_ai_player_actions=getattr(
self._handler, "fill_ai_player_actions", None,
),
resume_authoritative_combat=lambda game_key, seat_uid: (
ruleset_gameplay.resume_authoritative_combat(
self._ruleset_gameplay_dependencies, game_key, seat_uid,
)
),
resolve_pending_dice=self.resolve_pending_dice_for_game,
roll_for_game=self.roll_for_game,
save_instance=self._reg.save,
process_round=getattr(self._handler, "process_round", None),
schedule_decision_timeouts=getattr(self._handler, "schedule_decision_timeouts", None),
resolve_luck_decision=self.resolve_luck_decision,
decline_pending_luck=self.decline_pending_luck,
drain_economy_outbox=self._drain_economy_outbox,
economy_auto_reward_settings=self.economy_auto_reward_settings,
resolve_reward=self.resolve_reward_as_gm,
settle_adventure_completion=self._settle_adventure_completion_for,
)
self._map_dependencies = maps.MapDependencies(
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
list_lore_entries=self._lore.list_entries,
list_map_assets=(
plugin_host.list_map_assets
if plugin_host is not None
else lambda _world_id: {
"maps": [],
"locations": [],
"icons": [],
"scenes": [],
}
),
validate_background_selection=self.validate_map_background_selection,
save_instance=self._reg.save,
load_world_template=self._load_world_template,
map_background_file=self.map_background_file,
generated_image_file=self.generated_image_file,
content_projection=getattr(handler, "content_projection", None),
)
self._assistant_dependencies = assistant.AssistantDependencies(
list_plugins=self.list_plugins,
llm_configuration_error=self._llm_configuration_error,
llm_client=self._llm_client,
data_dir=self._reg.save_dir.parent,
text_gen_max_tokens=self.text_gen_max_tokens,
)
self._connection_dependencies = generation.ConnectionDependencies(
llm_client=self._llm_client,
config_state=lambda: getattr(self, "_config_state", {}) or {},
)
self._generation_dependencies = generation.GenerationDependencies(
llm_client=self._llm_client,
llm_configuration_error=self._llm_configuration_error,
worlds_dir=self._worlds_dir,
lorebook_store=self._lore,
rules_dir=self._rules_dir,
registry=self._reg,
parse_game_key=_parse_game_key,
get_instance=self._reg.get,
load_rule_by_id=self._load_rule_by_id,
load_rule_for_game=self._load_rule_for_game,
character_gen_max_tokens=self.character_gen_max_tokens,
text_gen_max_tokens=self.text_gen_max_tokens,
)
self._rule_dependencies = rules.RuleDependencies(
rules_dir=self._rules_dir,
ruleset_registry=self._ruleset_registry,
plugin_host=plugin_host,
has_games_using_rule=lambda rule_id: any(
instance.rule_id == rule_id for instance in self._reg.list_all()
),
)
self._ruleset_builder_dependencies = ruleset_builder.RulesetBuilderDependencies(
load_rule=self._load_rule_by_id,
ruleset_registry=self._ruleset_registry,
)
self._tunnel_dependencies = tunnel.TunnelDependencies(
config_state=lambda: getattr(self, "_config_state", {}),
save_config=lambda: getattr(self, "_save_config")(),
list_plugins=self.list_plugins,
)
self._bot_extensions = bot_extensions.BotExtensionService(
bot_extensions.BotExtensionDependencies(plugin_host=plugin_host)
)
self._hub = hub_client
self._hub_service = hub.HubService(
hub.HubDependencies(
client=hub_client,
plugin_host=plugin_host,
config_state=lambda: getattr(self, "_config_state", {}),
save_config=lambda: getattr(self, "_save_config")(),
current_legal_documents=self.current_legal_documents,
legal_bundle_version=self.legal_bundle_version,
legal_acceptance_payload=self.legal_acceptance_payload,
legal_accepted=self.legal_accepted,
record_legal_acceptance=self.record_legal_acceptance,
)
)
self._speech = speech_service
self._web_speech = speech.WebSpeechService(
speech.SpeechDependencies(
backend=speech_service,
plugin_host=plugin_host,
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
)
)
self._asr = asr_service
self._imagegen = imagegen_service
self._system = system.SystemService(
system.SystemDependencies(
data_dir=self._reg.save_dir.parent,
config_state=lambda: getattr(self, "_config_state", {}) or {},
proxy_url=lambda: str(
getattr(self._llm_client, "proxy_url", "") or ""
),
mirrors=lambda: getattr(self._plugins, "mirrors", None),
)
)
self._announcements = announcements.AnnouncementService(
announcements.AnnouncementDependencies(
fetch_public_text=self.fetch_public_content_text,
disk_cache_age=self.public_content_disk_age,
)
)
self._web_asr = asr.WebAsrService(
asr.AsrDependencies(
backend=self._asr,
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
)
)
self._lore_preview = knowledge.LorePreviewService(
knowledge.LorePreviewDependencies(
lorebook=self._lore,
get_instance=self._reg.get,
parse_game_key=_parse_game_key,
content_projection=ContentProjectionService(self._lore),
)
)
kp_answerer = (
getattr(self._handler, "answer_kp_question", None)
if self._handler is not None
else None
)
self._kp_questions = kp_questions.KPQuestionService(
kp_questions.KPQuestionDependencies(
registry=self._reg,
parse_game_key=_parse_game_key,
answer_question=kp_answerer if callable(kp_answerer) else None,
)
)
self._generated_images = generated_images.GeneratedImageService(
generated_images.GeneratedImageDependencies(
imagegen=imagegen_service,
get_instance=self.get_game_instance,
save_instance=self._reg.save,
update_map_background=self.update_map_background,
avatar_file=lambda asset_id: self.avatar_file(asset_id),
llm_client=self._llm_client,
)
)
self._game_lifecycle = game_lifecycle.GameLifecycleService(
game_lifecycle.GameLifecycleDependencies(
registry=self._reg,
handler=self._handler,
rulesets=self._ruleset_registry,
lorebook=self._lore,
worlds_dir=self._worlds_dir,
rules_dir=self._rules_dir,
parse_game_key=_parse_game_key,
llm_configuration_error=self._llm_configuration_error,
load_rule_by_id=self._load_rule_by_id,
resolve_adventure_binding=lambda adventure_id, runtime, world_id, language, source_kind="", source_id="": adventures.resolve_binding_for_runtime(
self._adventure_dependencies,
adventure_id,
runtime,
world_id,
language,
source_kind=source_kind,
source_id=source_id,
),
# FIX-04 §6.5：v2 冒险的创建事务步骤（进度 + 原子世界种子）。
initialize_adventure_run=lambda instance: adventure_runtime.initialize_adventure_run(
self._adventure_runtime_dependencies, instance,
),
resolve_default_scene_image=self.resolve_default_scene_image,
list_default_scene_images=self._scene_images.default_candidates,
materialize_scene_image=self.materialize_scene_image,
validate_map_background=self.validate_map_background_selection,
create_player=lambda *args, **kwargs: self.create_player(*args, **kwargs),
cleanup_orphan_game_templates=self.cleanup_orphan_game_templates,
refresh_lorebook_index=self._refresh_game_lorebook_index,
project_rule_id=lambda instance: game_queries.projected_rule_id(
self._game_query_dependencies,
instance,
),
clean_public_narration=game_queries.clean_public_narration,
get_game_defaults=lambda: validate_preferences((getattr(self, '_config_state', {}) or {}).get('game_defaults', {})),
resume_after_creation=lambda instance: turns.resume_after_creation(
self._turn_dependencies,
"|".join(instance.game_key),
),
# issue #700：开局勾选 catalog 模块声明书时按需物化（组合根接线）。
materialize_module_book=lambda source_kind, source_id, book_id: plugins.materialize_module_book_binding(
self._plugin_content_dependencies,
source_kind,
source_id,
book_id,
),
)
)
if self._plugins and self._handler and hasattr(self._handler, "set_plugin_host"):
self._handler.set_plugin_host(self._plugins)
from src.runtime_diagnostics import assistant_runtime_log_context
from src.webui.services.assistant_content import AgentContentDependencies, AgentContentTools
from src.webui.services.assistant_agent import AgentDependencies, AgentRuntime
from src.webui.services.assistant_agent_application import AgentApplicationDependencies, application_tools
from src.webui.services.assistant_authoring import AuthoringDependencies, authoring_tools
from src.webui.services.content_recovery import ContentRecoveryDependencies, ContentRecoveryService
from src.webui.services.assistant_recovery_tools import recovery_tools
from src.webui.services.game_lorebooks import GameLorebookDependencies, GameLorebookService
from src.webui.services.assistant_game_lorebook_tools import game_lorebook_tools
from src.application_actions import ActionRegistry
from src.engine.participant_view import resolve_viewer
from src.webui.services.assistant_gameplay_tools import GameplayToolDependencies, gameplay_tools
from src.webui.services.assistant_world_tools import WorldToolDependencies, world_tools
from src.webui.services.assistant_plugin_tools import PluginActionDependencies, plugin_tools
from src.webui.services.assistant_library_tools import LibraryToolCallbacks, build_library_actions
self._game_lorebooks = GameLorebookService(GameLorebookDependencies(
store=self._lore, get_game=self.get_game_instance,
list_books=lambda: self.list_lorebooks(include_all=True), save_instance=self._reg.save,
))
from src.webui.services.content_metadata_recovery import MetadataRecoveryDependencies, MetadataRecoveryService
world_snapshot = lambda identity: worlds.authoring_recovery_snapshot(self._world_dependencies, identity)
card_snapshot = lambda identity: character_cards.authoring_recovery_snapshot(self._character_card_dependencies, identity)
self._metadata_recovery = MetadataRecoveryService(MetadataRecoveryDependencies(
store=self._lore, world_snapshot=world_snapshot,
world_update=lambda identity, patch: self.update_world(identity, **{**world_snapshot(identity)['fields'], **patch}),
card_snapshot=card_snapshot, card_update=self.update_character_card,
))
self._content_recovery = ContentRecoveryService(ContentRecoveryDependencies(
store=self._lore, list_games=self.list_games, get_game=self.get_game_instance,
world_definition=self.get_world_definition, save_entry=self.save_lorebook_entry,
), metadata=self._metadata_recovery)
agent_content = AgentContentTools(AgentContentDependencies(
generate_text=self.generate_text, review_content=self.review_content_text,
commit_content=self.commit_content_text,
))
agent_tools = application_tools(AgentApplicationDependencies(
catalogs={"worlds": self.list_worlds, "world_templates": self.list_world_templates,
"lorebooks": lambda: self.list_lorebooks(include_all=True), "adventures": self.list_adventures,
"characters": self.list_character_cards, "rules": self.list_rules, "games": self.list_games,
"plugins": self.list_plugins, "plugin_tools": self.list_plugin_tools},
get_game=self.get_game_instance, read_game=self.game_detail, create_game=self.create_game,
content=agent_content, set_player_control=self.set_player_control, submit_action=self.submit_action,
generate_rule=self.generate_rule, generate_image=self.generate_generated_image,
invoke_plugin_tool=self.invoke_plugin_tool, update_plugin=self.update_plugin_config,
runtime_logs=lambda: assistant_runtime_log_context(self._reg.save_dir.parent),
))
def agent_game_players(game_key, owner):
instance = self.get_game_instance(game_key)
if instance is None:
raise ValueError('Game does not exist')
viewer = resolve_viewer(instance, user_id=owner, owner_authenticated=True, player_preview=False)
if not viewer.is_gm:
raise PermissionError('GM access required')
return {'players': {uid: {} for uid in instance.players}}
agent_tools += authoring_tools(AuthoringDependencies(
list_worlds=self.list_worlds, update_world=self.update_world,
list_book_entries=self.list_lorebook_entries, save_book_entry=self.save_lorebook_entry,
list_cards=self.list_character_cards, update_card=self.update_character_card,
adventure_detail=self.adventure_detail, get_rule=self.get_rule_template,
write_entries_with_recovery=self._content_recovery.write_entries,
write_book_visibility_with_recovery=self._content_recovery.write_book_visibility,
preview_book_visibility_with_recovery=self._content_recovery.preview_book_visibility,
book_snapshot=self._lore.snapshot_lorebook,
book_visibility_receipt=self._content_recovery.read_book_visibility_receipt,
write_world_with_recovery=self._metadata_recovery.write_world,
write_card_with_recovery=self._metadata_recovery.write_card,
world_snapshot=world_snapshot, card_snapshot=card_snapshot,
game_players=agent_game_players,
))
agent_tools += recovery_tools(self._content_recovery)
agent_tools += game_lorebook_tools(self._game_lorebooks)
agent_tools += gameplay_tools(GameplayToolDependencies(
get_game=self.get_game_instance,
viewer_for=lambda instance, owner: resolve_viewer(
instance, user_id=owner, owner_authenticated=True, player_preview=False),
adventure_projection=lambda instance, viewer: self.game_adventure_projection(
instance.game_key, viewer_is_gm=viewer.is_gm, viewer_uid=viewer.uid),
adventure_command=lambda instance, body, uid, gm: self.ruleset_submit_intent(
instance.game_key, uid, gm, body),
run_story=self.run_story, recovery=self.resource_recovery,
rest=self.ruleset_rest_submit_live,
room_preferences=self.game_preferences, set_room_preferences=self.update_game_preferences,
available_ruleset_actions=self.ruleset_available_actions,
submit_ruleset_intent=self.ruleset_submit_intent,
))
agent_tools += world_tools(WorldToolDependencies(
get_game=self.get_game_instance,
viewer_for=lambda instance, owner: resolve_viewer(
instance, user_id=owner, owner_authenticated=True, player_preview=False),
save_instance=self._reg.save,
))
agent_tools += build_library_actions(LibraryToolCallbacks(
list_worlds=self.list_world_templates, clone_world=self.clone_world_from_template,
list_lorebooks=lambda: self.list_lorebooks(include_all=True),
read_lorebook=self._lore.get_lorebook,
list_lorebook_bindings=self.list_lorebook_bindings,
create_lorebook=self.create_lorebook, update_lorebook=self.update_lorebook,
export_lorebook=self.export_lorebook,
preview_lorebook_import=self.preview_lorebook_import,
commit_lorebook_plan=self.commit_lorebook_plan,
list_adventures=self.list_adventures, copy_adventure=self.copy_adventure,
adventure_detail=self.adventure_detail,
))
self.application_actions = ActionRegistry(agent_tools)
plugin_action_dependencies = PluginActionDependencies(
list_tools=self.list_plugin_tools, invoke=self.invoke_plugin_tool,
)
self.application_actions.register_source('plugins', lambda: plugin_tools(plugin_action_dependencies))
self._assistant_agent = AgentRuntime(AgentDependencies(
llm_client=lambda: self._llm_client, configuration_error=self._llm_configuration_error,
tools=(), registry=self.application_actions, storage_dir=self._reg.save_dir.parent / "assistant_agent",
answer=assistant.offline_answer, knowledge=assistant.agent_knowledge,
review_content=agent_content.review, max_tokens=assistant_agent_max_tokens,
task_max_tokens=assistant_agent_task_max_tokens,
))
def register_application_actions(self, actions):
"""Owning modules extend the live action directory without Agent changes."""
for action in actions:
self.application_actions.register(action)
# ---- 跨域共享辅助 ----
def _llm_configuration_status(self) -> dict[str, Any]:
"""检查当前主模型配置；兼容不暴露 providers 的测试或第三方客户端。"""
client = self._llm_client
if client is None:
return {"ready": False, "missing": ["client"], "provider": ""}
providers = getattr(client, "providers", None)
if not isinstance(providers, dict):
return {
"ready": True,
"missing": [],
"provider": str(getattr(client, "default", "") or ""),
}
default = str(getattr(client, "default", "") or "")
provider = providers.get(default)
if provider is None and providers:
provider = next(iter(providers.values()))
if provider is None:
return {"ready": False, "missing": ["provider"], "provider": default}
missing: list[str] = []
if not str(getattr(provider, "base_url", "") or "").strip():
missing.append("base_url")
if not str(getattr(provider, "model_name", "") or "").strip():
missing.append("model")
return {
"ready": not missing,
"missing": missing,
"provider": str(getattr(provider, "provider_name", default) or default),
}
def _llm_configuration_error(self, language: str = "zh-CN") -> dict[str, Any] | None:
"""为所有依赖主模型的 WebUI 域生成一致的未配置错误。"""
status = self._llm_configuration_status()
if status["ready"]:
return None
english = str(language or "").lower().startswith("en")
message = (
"The main model is not configured. Add an AI provider in Settings "
"and assign a main model before continuing."
if english
else "尚未配置主模型，请先在设置页添加 AI 服务商并选择主模型。"
)
return {
"ok": False,
"error_code": "llm_not_configured",
"error": message,
"missing": status["missing"],
}
# ---- 规则辅助 ----
def list_plugins(self) -> dict[str, Any]:
return plugins.list_plugins(self._plugin_host_dependencies)
def list_modules(self) -> dict[str, Any]:
return modules.list_modules(self._module_dependencies)
async def list_module_marketplace(
self, keyword: str = "", ruleset: str = "",
) -> dict[str, Any]:
"""Online content-module catalogue for the modules view (FIX-06 §8)."""
return await modules.module_marketplace(
self._module_dependencies, keyword=keyword, ruleset=ruleset,
)
def module_detail(self, module_id: str, *, viewer_is_gm: bool = True) -> dict[str, Any]:
return modules.module_detail(self._module_dependencies, module_id, viewer_is_gm=viewer_is_gm)
def module_cover_file(self, module_id: str) -> str:
"""已安装内容包的封面文件路径（只读本机，不出站）。"""
return modules.module_cover_path(self._module_dependencies, module_id)
def module_resources(self, module_id: str) -> dict[str, Any]:
return self.module_resources_service.list(module_id)
def module_resource(self, module_id: str, selector: dict, language: str = "") -> dict[str, Any]:
return self.module_resources_service.read(module_id, selector, language)
def module_resource_asset(self, module_id: str, selector: dict) -> Path:
return self.module_resources_service.asset_path(module_id, selector)
def preview_module_fork(self, module_id: str, selector: dict, language: str = "") -> dict[str, Any]:
return self.module_forks_service.preview(module_id, selector, language)
def review_module_fork(self, module_id: str, draft: dict) -> dict[str, Any]:
return self.module_forks_service.review(module_id, draft)
def commit_module_fork(self, module_id: str, body: dict) -> dict[str, Any]:
return self.module_forks_service.commit(module_id, body)
def module_adventures(self, module_id: str) -> dict[str, Any]:
return modules.module_adventures(self._module_dependencies, module_id)
def module_lorebooks(self, module_id: str, *, viewer_is_gm: bool = True) -> dict[str, Any]:
return modules.module_lorebooks(self._module_dependencies, module_id, viewer_is_gm=viewer_is_gm)
def module_content(
self, module_id: str, kind: str, key: str, language: str = "",
*, viewer_is_gm: bool = True,
) -> dict[str, Any]:
return modules.module_content(
self._module_dependencies, module_id, kind, key, language=language,
viewer_is_gm=viewer_is_gm,
)
def module_compatibility(self, module_id: str) -> dict[str, Any]:
return modules.module_compatibility(self._module_dependencies, module_id)
def module_usages(self, module_id: str) -> dict[str, Any]:
return modules.module_usages(self._module_dependencies, module_id)
def preview_module_import(self, payload: bytes) -> dict[str, Any]:
if self._plugins is None:
return {"ok": False, "error": "插件宿主未启用"}
# FIX-01 §3.3：预览与安装共用同一套深度校验（真实解压 + Adventure/catalog
# 真装载），所以"预览显示 blocker"与"直接 API 安装被拒"结论一致。
return self._plugins.validate_package(
payload,
lambda directory, manifest: modules.preview_module_install(
self._module_dependencies, manifest, directory=directory,
),
)
def preview_external_module_import(
self, payload: bytes, source_name: str = "",
) -> dict[str, Any]:
"""Preview a data-only external VTT module; never executes its runtime."""
return module_import_adapters.preview_external_module_import(
payload, source_name=source_name,
)
def validate_module_package(
self, directory: Any, manifest: dict[str, Any],
) -> None:
"""Install-transaction hook: refuse the package when it has blockers."""
modules.validate_module_directory(
self._module_dependencies, directory, manifest,
require_content_pack=False,
)
def guard_module_mutation(self, plugin_id: str, action: str) -> None:
"""Package-mutation hook shared by every install/uninstall path."""
modules.assert_module_action_allowed(
self._module_dependencies, plugin_id, action,
)
def attach_package_hooks(self, plugin_host: Any) -> None:
"""Wire the FIX-01 install-transaction hooks onto the plugin host.
单一接线点：bootstrap 与测试共用，保证本地导入 / 市场安装 / 覆盖安装 /
后台自动更新 / 卸载都经过同一条校验与 bound-save guard。
"""
plugin_host.set_package_hooks(
package_validator=self.validate_module_package,
content_module_guard=self.guard_module_mutation,
)
async def get_official_announcement(self, language: str = "zh-CN") -> dict[str, Any]:
return await self._announcements.fetch(language)
async def hub_preferences(self, language: str = "zh-CN") -> dict[str, Any]:
return await self._hub_service.preferences(language)
async def update_hub_preferences(
self,
telemetry_enabled: bool,
legal_acceptance: dict[str, Any] | None = None,
language: str = "zh-CN",
) -> dict[str, Any]:
return await self._hub_service.update_preferences(
telemetry_enabled, legal_acceptance, language,
)
async def delete_hub_identity(self) -> dict[str, Any]:
return await self._hub_service.delete_identity()
async def create_rendezvous_room(self, peer_count: int) -> dict[str, Any]:
return await self._hub_service.create_rendezvous_room(peer_count)
async def rendezvous_config(self) -> dict[str, Any]:
return await self._hub_service.rendezvous_config()
async def hub_plugin_detail(self, plugin_id: str) -> dict[str, Any]:
return await self._hub_service.plugin_detail(plugin_id)
async def hub_plugin_readme(self, plugin_id: str) -> dict[str, Any]:
return await self._hub_service.plugin_readme(plugin_id)
async def hub_plugin_ratings(self, plugin_id: str) -> dict[str, Any]:
return await self._hub_service.plugin_ratings(plugin_id)
async def set_hub_plugin_like(self, plugin_id: str, liked: bool) -> dict[str, Any]:
return await self._hub_service.set_plugin_like(plugin_id, liked)
async def set_hub_plugin_rating(
self, plugin_id: str, stars: int | None, tags: list[str] | None = None
) -> dict[str, Any]:
return await self._hub_service.set_plugin_rating(plugin_id, stars, tags)
async def assistant_chat(self, response, messages: list[dict], language: str = "zh-CN") -> None:
return await assistant.chat_stream(
self._assistant_dependencies,
response,
messages,
language,
)
async def assistant_agent_run(self, owner, messages, language, context, session_id, emit):
return await self._assistant_agent.run(owner, messages, language, context, session_id, emit)
async def assistant_agent_confirm(self, owner, session_id, approval_id, plan_digest, approved):
return await self._assistant_agent.confirm(owner, session_id, approval_id, plan_digest, approved)
async def assistant_agent_review(self, owner, session_id, draft_id, draft):
return await self._assistant_agent.review(owner, session_id, draft_id, draft)
def assistant_agent_session(self, owner, session_id):
return self._assistant_agent.snapshot(owner, session_id)
def assistant_agent_capabilities(self, owner, query='', domain='', offset=0):
return self._assistant_agent.capabilities(owner, query, domain, offset)
async def assistant_agent_resume(self, owner, session_id, approval_id, plan_digest):
return await self._assistant_agent.resume(owner, session_id, approval_id, plan_digest)
def assistant_agent_authorize(self, owner, session_id, mode, scope, domains, max_steps):
return self._assistant_agent.authorize(owner, session_id, mode, scope, domains, max_steps)
def assistant_agent_cancel(self, owner, session_id):
return self._assistant_agent.cancel(owner, session_id)
async def assistant_agent_plan(self, owner, tool, arguments, session_id=''):
return await self._assistant_agent.plan(owner, tool, arguments, session_id)
def assistant_content_operations(self, owner):
return self._content_recovery.list_operations(owner)
def game_lorebook_selection(self, owner, game_key):
return self._game_lorebooks.selection(game_key, owner, owner_authenticated=True)
async def close_assistant_agent(self):
await self._assistant_agent.close()
def list_plugin_types(self) -> dict[str, Any]:
return plugins.list_plugin_types()
def list_speech_voices(self) -> dict[str, Any]:
return self._web_speech.list_voices()
def list_personal_speech_profiles(self) -> dict[str, Any]:
return self._web_speech.list_personal_profiles()
def save_personal_speech_profile(
self,
profile_id: str,
values: dict[str, Any],
*,
file_data: str = "",
file_name: str = "",
) -> dict[str, Any]:
return self._web_speech.save_personal_profile(
profile_id,
values,
file_data=file_data,
file_name=file_name,
)
def delete_personal_speech_profile(self, profile_id: str) -> dict[str, Any]:
return self._web_speech.delete_personal_profile(profile_id)
async def synthesize_speech(
self,
game_key: str,
user_id: str,
text: str,
voice: str = "",
language: str = "zh-CN",
speed: float = 1.0,
owner: bool = False,
):
return await self._web_speech.synthesize(
game_key, user_id, text, voice, language, speed, owner,
)
async def test_speech(
self,
text: str,
voice: str = "",
language: str = "zh-CN",
speed: float = 1.0,
):
return await self._web_speech.test_synthesis(
text, voice, language, speed,
)
async def transcribe_speech(
self,
game_key: str,
user_id: str,
audio: bytes,
content_type: str,
language: str = "",
owner: bool = False,
):
return await self._web_asr.transcribe(
game_key, user_id, audio, content_type, language, owner,
)
async def test_transcription(self, audio: bytes, content_type: str, language: str = ""):
return await self._web_asr.test_transcription(
audio, content_type, language,
)
async def rescan_plugins(self) -> dict[str, Any]:
return await plugins.rescan_plugins(self._plugin_host_dependencies)
def publish_tunnel_url(self, plugin_id: str, url: str) -> dict[str, Any]:
return tunnel.publish_tunnel_url(self._tunnel_dependencies, plugin_id, url)
def release_tunnel_url(self, plugin_id: str) -> dict[str, Any]:
return tunnel.release_tunnel_url(self._tunnel_dependencies, plugin_id)
def tunnel_status(self) -> dict[str, Any]:
return tunnel.tunnel_status(self._tunnel_dependencies)
def authenticate_plugin_token(self, token: str) -> dict[str, Any] | None:
"""校验插件进程内部 Token，返回插件身份（复用 plugin_host）。"""
if not self._plugins:
return None
return self._plugins.authenticate_api_token(token)
def plugin_detail(self, plugin_id: str) -> dict[str, Any]:
return plugins.plugin_detail(self._plugin_host_dependencies, plugin_id)
def read_plugin_docs(self, plugin_id: str) -> dict[str, Any]:
return plugins.read_plugin_docs(self._plugin_host_dependencies, plugin_id)
async def update_plugin_config(self, plugin_id: str, changes: dict[str, Any]) -> dict[str, Any]:
if changes.get("enabled") is False:
modules.assert_module_action_allowed(
self._module_dependencies, plugin_id, "disable",
)
result = await plugins.update_plugin_config(
self._plugin_lifecycle_dependencies, plugin_id, changes,
)
if result.get("ok"):
# FIX-07：启用/禁用会改变模块贡献的内容 → 立刻刷新来源与 catalog 链，
# 否则"启用模组后马上开局"会解析不到它的冒险（解析不得依赖
# "碰巧同步过 registry"，与 §3.6 同一原则）。
self.refresh_content_sources()
return result
async def control_plugin(self, plugin_id: str, action: str) -> dict[str, Any]:
if action == "stop":
modules.assert_module_action_allowed(
self._module_dependencies, plugin_id, "disable",
)
result = await plugins.control_plugin(
self._plugin_lifecycle_dependencies, plugin_id, action,
)
if result.get("ok"):
self.refresh_content_sources()
return result
async def install_plugin(
self, payload: bytes, overwrite: bool = False, expected_plugin_type: str = "",
) -> dict[str, Any]:
result = await plugins.install_plugin(
self._plugin_lifecycle_dependencies, payload, overwrite,
expected_plugin_type=expected_plugin_type,
)
if result.get("ok"):
self.refresh_content_sources()
return result
async def list_plugin_marketplace(self) -> dict[str, Any]:
return await plugins.list_plugin_marketplace(
self._plugin_host_dependencies,
)
async def install_marketplace_plugin(
self, plugin_id: str, overwrite: bool = False, expected_plugin_type: str = "",
) -> dict[str, Any]:
result = await plugins.install_marketplace_plugin(
self._plugin_lifecycle_dependencies, plugin_id, overwrite,
expected_plugin_type=expected_plugin_type,
)
if result.get("ok"):
self.refresh_content_sources()
return result
async def import_module(self, payload: bytes, overwrite: bool = False) -> dict[str, Any]:
"""Install a local content-module package through the canonical host.
FIX-01 §3.1：模组安装面只接受 data-only content-pack——类型门在服务端
package inspection 之后强制（前端分类不可信）。
"""
return await self.install_plugin(
payload, overwrite, expected_plugin_type=modules.MODULE_PLUGIN_TYPE,
)
async def install_marketplace_module(
self, module_id: str, overwrite: bool = False,
) -> dict[str, Any]:
"""Install a marketplace content module through the canonical host."""
return await self.install_marketplace_plugin(
module_id, overwrite, expected_plugin_type=modules.MODULE_PLUGIN_TYPE,
)
async def update_marketplace_plugin(self, plugin_id: str) -> dict[str, Any]:
modules.assert_module_action_allowed(self._module_dependencies, plugin_id, "update")
result = await plugins.update_marketplace_plugin(
self._plugin_lifecycle_dependencies, plugin_id,
)
if result.get("ok"):
# 与市场安装 / 覆盖安装一致：新版本的内容和冒险立刻可解析。
self.refresh_content_sources()
return result
async def uninstall_plugin(self, plugin_id: str, delete_data: bool = False) -> dict[str, Any]:
modules.assert_module_action_allowed(self._module_dependencies, plugin_id, "uninstall")
result = await plugins.uninstall_plugin(
self._plugin_lifecycle_dependencies, plugin_id, delete_data,
)
if result.get("ok"):
self.refresh_content_sources()
return result
def list_plugin_mirrors(self) -> dict[str, Any]:
return plugins.list_plugin_mirrors(self._plugin_host_dependencies)
def add_plugin_mirror(self, data: dict[str, Any]) -> dict[str, Any]:
return plugins.add_plugin_mirror(self._plugin_host_dependencies, data)
def update_plugin_mirror(self, mirror_id: str, data: dict[str, Any]) -> dict[str, Any]:
return plugins.update_plugin_mirror(
self._plugin_host_dependencies, mirror_id, data,
)
def delete_plugin_mirror(self, mirror_id: str) -> dict[str, Any]:
return plugins.delete_plugin_mirror(
self._plugin_host_dependencies, mirror_id,
)
async def test_plugin_mirror(self, mirror_id: str = "") -> dict[str, Any]:
return await plugins.test_plugin_mirror(
self._plugin_host_dependencies, mirror_id,
)
def clear_plugin_card_cache(self, plugin_id: str) -> dict[str, Any]:
return plugins.clear_plugin_card_cache(
self._plugin_host_dependencies, plugin_id,
)
def list_plugin_contributions(self, kind: str = "") -> dict[str, Any]:
return plugins.list_plugin_contributions(
self._plugin_host_dependencies, kind,
)
def list_plugin_themes(self) -> dict[str, Any]:
return plugins.list_plugin_themes(self._plugin_host_dependencies)
def list_plugin_tools(self) -> dict[str, Any]:
return plugins.list_plugin_tools(self._plugin_host_dependencies)
async def invoke_plugin_tool(
self,
plugin_id: str,
tool_name: str,
arguments: dict[str, Any],
context: dict[str, Any] | None = None,
) -> dict[str, Any]:
return await plugins.invoke_plugin_tool(
self._plugin_host_dependencies,
plugin_id,
tool_name,
arguments,
context,
)
def list_plugin_content(self, kind: str = "", world_id: str = "", rule_id: str = "", language: str = "") -> dict[str, Any]:
return plugins.list_plugin_content(
self._plugin_host_dependencies,
kind,
world_id,
rule_id,
language,
)
def sync_plugin_lorebooks(self) -> dict[str, Any]:
"""同步已启用插件的世界模板世界书到世界书库（幂等）。"""
return plugins.sync_plugin_lorebooks(self._plugin_content_dependencies)
def cleanup_plugin_lorebook(self, plugin_id: str) -> dict[str, Any]:
"""删除某插件灌入的、未被用户改动的世界书条目。"""
return plugins.cleanup_plugin_lorebook(
self._plugin_content_dependencies, plugin_id,
)
def import_plugin_content(
self,
kind: str,
resource_id: str,
plugin_id: str = "",
target_world_id: str = "",
overwrite: bool = False,
target_book_id: str = "",
) -> dict[str, Any]:
return plugins.import_plugin_content(
self._plugin_content_dependencies,
kind,
resource_id,
plugin_id,
target_world_id,
overwrite,
target_book_id=target_book_id,
)
def import_all_plugin_content(
self,
plugin_id: str,
target_world_id: str = "",
target_book_id: str = "",
) -> dict[str, Any]:
return plugins.import_all_plugin_content(
self._plugin_content_dependencies,
plugin_id,
target_world_id,
target_book_id=target_book_id,
)
def export_content_pack(
self,
plugin_id: str,
name: str,
version: str,
description: str,
world_id: str = "",
card_ids: list[str] | None = None,
rule_id: str = "",
flat: bool = False,
include_portraits: bool = True,
include_scene_images: bool = True,
world_scene_image: dict[str, Any] | None = None,
rule_scene_image: dict[str, Any] | None = None,
include_map: bool = True,
map_background: dict[str, Any] | None = None,
map_icons: list[dict[str, Any]] | None = None,
language: str = "",
adventure_ids: list[str] | None = None,
lorebook_ids: list[str] | None = None,
adventure_refs: list[dict[str, str]] | None = None,
cover_data: str = "",
cover_filename: str = "",
supported_locales: list[str] | None = None,
) -> dict[str, Any]:
return plugins.export_content_pack(
self._plugin_export_dependencies,
plugin_id, name, version, description, world_id, card_ids, rule_id, flat,
include_portraits, include_scene_images, world_scene_image, rule_scene_image,
include_map, map_background, map_icons, language,
adventure_ids, lorebook_ids, adventure_refs,
cover_data=cover_data, cover_filename=cover_filename,
supported_locales=supported_locales,
)
def package_content_map(
self,
plugin_id: str,
pack_name: str,
world: dict[str, Any],
entries: list[dict[str, Any]],
files: dict[str, str | bytes],
*,
background_selection: dict[str, Any] | None = None,
icon_uploads: list[dict[str, Any]] | None = None,
):
"""Package map contributions through the WebAPI cross-domain facade."""
return content_pack_maps.package_content_map(
self._content_map_dependencies,
plugin_id,
pack_name,
world,
entries,
files,
background_selection=background_selection,
icon_uploads=icon_uploads,
)
def plugin_asset_path(self, plugin_id: str, relative_path: str) -> Path:
return plugins.plugin_asset_path(
self._plugin_host_dependencies, plugin_id, relative_path,
)
async def check_updates(self, include_prerelease: bool | None = None) -> dict[str, Any]:
return await self._system.check_updates(include_prerelease)
def runtime_log_status(self) -> dict[str, Any]:
return self._system.runtime_log_status()
def clear_runtime_logs(self) -> dict[str, Any]:
return self._system.clear_runtime_logs()
def export_runtime_logs(self) -> tuple[bytes, int]:
return self._system.export_runtime_logs()
def _load_world_template(self, world_id: str, language: str = "") -> dict[str, Any] | None:
"""按 world_id 读取世界模板；不存在或非法时返回 None。"""
if not self._worlds_dir:
return None
data = load_world_template(self._worlds_dir, world_id, language)
if data:
return data
if self._plugins:
return self._plugins.load_world_template(world_id, language)
return None
def _load_rule_for_game(self, inst) -> RuleSystem | None:
"""优先按存档自身规则加载；旧存档缺失时回退世界默认规则。"""
if not inst.world_id or not self._worlds_dir:
return None
language = getattr(inst, "language", "") or ""
world_data = self._load_world_template(inst.world_id, language)
if not world_data:
return None
language = language or world_data.get("active_locale") or world_data.get("language", "")
rule_id = str(
getattr(inst, "rule_id", "")
or world_data.get("default_rule")
or "freeform_fantasy"
)
return self._load_rule_by_id(rule_id, language)
def _load_runtime_for_game(self, inst):
rule = self._load_rule_for_game(inst)
return self._ruleset_registry.resolve(rule.template) if rule else None
def _adventure_runtime_adapter(self, instance: Any, hook: str) -> Any:
"""Ask the instance's ruleset runtime for an adventure adapter (FIX-04).
``rules.*`` gate 求值与 ``item_reward`` 转换都属于 ruleset 的职责
（D&D 实现见 Dnd2024Runtime）；其它 runtime 没有该能力时返回 None →
gate 证据不足 = 不放行 / reward 缺少转换器 = fail closed。
"""
try:
runtime = self._load_runtime_for_game(instance)
except Exception: # noqa: BLE001 - 适配器缺席不等于放行
return None
getter = getattr(runtime, hook, None)
if not callable(getter):
return None
try:
return getter(instance)
except Exception: # noqa: BLE001
return None
def _queue_adventure_reward_intents(
self, instance: Any, intents: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""FIX-08 §10 步骤 12：adventure item_reward → 既有奖励提案权威。
Adventure 的 ``item_reward`` 只声明"该给什么"（reward intent）；发放必须走
既有 proposal/settlement 权威，**绝不直接写 inventory**：
``kind="reward"`` 且 ``amount=0``（纯物品、无货币）、``approval_policy="gm"``
→ 提案 pending，GM 用既有的支付确认路径结算，物品才由
``characters.grant_reward`` 写入角色卡（带 before-image，整轮回滚可还原）。
幂等：``source_ref`` 由奖励内容 + 收件人确定（同一批奖励不会被排两次队），
重放/中止后同一身份会走 ``_existing_by_source`` 的既有语义。
"""
named = []
for intent in intents or []:
if str(intent.get("kind") or "") != "item_grant":
continue
name = str(intent.get("name") or "")
if not name.strip():
continue
reward = {
"name": name,
"category": str(intent.get("category") or ""),
}
# 身份映射是不透明的 metadata：通用 webui 层不解释 D&D 规则，
# 只保证它随奖励一起到达 grant_reward（缺失时保持旧形状）。
item_ref = str(intent.get("item_ref") or "")
if item_ref:
reward["item_ref"] = item_ref
named.append(reward)
if not named:
return []
recipient_uid = str((intents or [{}])[0].get("recipient_uid") or "")
if not recipient_uid:
raise ValueError("adventure reward intent has no recipient")
refs = sorted({
f"{intent.get('ref', {}).get('source', '')}:{intent.get('ref', {}).get('id', '')}"
for intent in intents or []
if isinstance(intent.get("ref"), dict)
} | {reward["name"] for reward in named})
proposal = queue_proposal(
instance,
kind="reward",
amount=0,
recipient_uid=recipient_uid,
reason="冒险奖励",
source="adventure",
source_ref=f"adventure_reward:{recipient_uid}:{'|'.join(refs)}",
approval_policy="gm",
rewards=named,
visibility="party",
)
return [{
"proposal_id": str(proposal.get("id") or ""),
"status": str(proposal.get("status") or ""),
"kind": "reward",
}]
def _project_game_rule_id(self, instance) -> str:
return game_queries.projected_rule_id(
self._game_query_dependencies,
instance,
)
def _load_rule_by_id(self, rule_id: str, language: str = "") -> RuleSystem | None:
rule_id = (rule_id or "").strip()
if not rule_id or not rules.is_valid_rule_id(rule_id):
return None
core_path = self._rules_dir / f"{rule_id}.json"
if core_path.exists():
return RuleSystem(RuleBundleLoader().load_rule(self._rules_dir, rule_id, language))
if self._plugins:
item = self._plugins.contributions.find("rule", rule_id)
if item:
localized = self._plugins.load_rule_template(
rule_id, language, plugin_id=item.plugin_id,
)
if localized:
return RuleSystem(localized)
return RuleSystem.load(item.path)
legacy_path = RuleSystem.path_for(self._rules_dir, rule_id, language)
return RuleSystem.load(legacy_path) if legacy_path.exists() else None
# ---- 游戏总览 ----
def list_games(self) -> dict[str, Any]:
return game_queries.list_games(self._game_query_dependencies)
def game_detail(
self,
game_key: str,
viewer_uid: str = "",
viewer_is_gm: bool = False,
) -> dict[str, Any] | None:
return game_queries.game_detail(
self._game_query_dependencies,
game_key,
viewer_uid,
viewer_is_gm,
)
def game_adventure_projection(
self, game_key: str, *, viewer_is_gm: bool, viewer_uid: str = "",
) -> dict[str, Any] | None:
"""Read the bound adventure through the viewer-safe projection seam."""
instance = self.get_game_instance(game_key)
if instance is None:
return None
self._sync_plugin_adventure_sources()
return adventures.game_adventure_projection(
self._adventure_dependencies, instance, viewer_is_gm=viewer_is_gm, viewer_uid=viewer_uid,
)
def get_game_instance(self, game_key: str):
"""Resolve a public game key without exposing registry/parser internals."""
return self._reg.get(_parse_game_key(game_key))
def game_profile_catalog(self):
from src.game_profiles.registry import catalog
return {'ok': True, 'profiles': catalog()}
async def create_profile_game(self, body):
return await self._profile_games.create(body)
def game_profile_view(self, game_key, uid, is_gm, actor_uid=''):
return self._profile_games.view(self.get_game_instance(game_key), uid, is_gm, actor_uid)
async def submit_profile_action(self, game_key, uid, is_gm, body):
instance = self.get_game_instance(game_key)
if instance is None: raise ValueError('game does not exist')
return await self._profile_games.act(instance, uid, is_gm, body)
def manual_roll_requests(self, game_key: str, user_id: str):
return self._manual_rolls.list(game_key, user_id)
async def create_manual_roll_request(self, game_key, user_id, body):
return await self._manual_rolls.create(game_key, user_id, body)
async def resolve_manual_roll_request(self, game_key, user_id, request_id, body):
return await self._manual_rolls.resolve(game_key, user_id, request_id, body)
async def cancel_manual_roll_request(self, game_key, user_id, request_id, body):
return await self._manual_rolls.cancel(game_key, user_id, request_id, body)
async def reveal_check(self, game_key: str, check_id: str, user_id: str):
return await self._check_reveals.reveal(game_key, check_id, user_id)
async def issue_seat_token(
self, game_key: str, uid: str, *, requester_uid: str, owner: bool, rotate: bool = False,
bound_sessions: int = 0, check_token: str = "",
) -> dict:
return await self._seat_tokens.issue_for_gm(
game_key, uid, requester_uid=requester_uid, owner=owner, rotate=rotate,
bound_sessions=bound_sessions, check_token=check_token,
)
async def claim_seat_token(self, game_key: str, *, session_uid: str) -> dict:
return await self._seat_tokens.claim(game_key, session_uid=session_uid)
async def save_game_instance(self, instance) -> None:
"""Persist an already-authorized aggregate through the application facade."""
await self._reg.save(instance)
async def generate_game_swipe(self, instance, round_number: int) -> str:
"""Generate an alternate narration without exposing the command handler."""
return await self._handler.generate_swipe(instance, round_number)
def saved_game_access(self, game_key: str) -> dict[str, Any]:
return self._game_packages.saved_game_access(game_key)
def export_game_package(self, game_key: str) -> dict[str, Any]:
return self._game_packages.export_game_package(game_key)
async def import_game_package(self, payload: bytes) -> dict[str, Any]:
return await self._game_packages.import_game_package(payload)
def delete_game(self, game_key: str) -> dict[str, Any]:
return self._game_lifecycle.delete_game(game_key)
async def get_bot_bind_token(self, game_key: str, rotate: bool = False) -> dict[str, Any]:
return await self._bot_access.get_bind_token(game_key, rotate)
async def verify_bot_bind_game(self, game_key: str, bind_token: str) -> dict[str, Any]:
return await self._bot_access.verify_bind_game(game_key, bind_token)
def bot_actor_allowed(self, game_key: str, user_id: str) -> bool:
return self._bot_access.actor_allowed(game_key, user_id)
def bot_surface(self, game_key: str) -> str:
return self._bot_access.surface_for_game(game_key)
def bot_extension_capabilities(self) -> dict[str, Any]:
return self._bot_extensions.capabilities()
async def apply_bot_extensions(self, stage: str, payload: dict[str, Any]) -> dict[str, Any]:
return await self._bot_extensions.apply(stage, payload)
def bot_extension_asset_path(self, plugin_id: str, relative_path: str) -> Path:
return self._bot_extensions.asset_path(plugin_id, relative_path)
def bot_bridge_card_path(self, name: str) -> Path:
return self._bot_extensions.bridge_card_path(name)
def multiplayer_status(self, game_key: str) -> dict[str, Any]:
return game_queries.multiplayer_status(
self._game_query_dependencies,
game_key,
)
def player_context(
self, *, preview: bool = False, delegate: bool = False, user_id: str = "",
) -> dict[str, Any]:
return game_queries.player_context(
preview=preview, delegate=delegate, user_id=user_id,
)
async def set_player_away(self, game_key: str, user_id: str, away: bool) -> dict[str, Any]:
return await self._game_controls.set_player_away(game_key, user_id, away)
async def set_player_control(self, game_key: str, user_id: str, mode: str) -> dict[str, Any]:
return await self._game_controls.set_player_control(game_key, user_id, mode)
def ai_seat_settings(self, game_key: str) -> dict[str, Any]:
return self._ai_seats.read(game_key)
def check_mode_settings(self, game_key: str, viewer: Any) -> dict[str, Any]:
return self._check_modes.read(game_key, viewer)
async def update_check_mode_settings(self, game_key: str, viewer: Any, uid: str, operation: str, body: Any) -> dict[str, Any]:
return await self._check_modes.update(game_key, viewer, uid, operation, body)
async def close_hosted_combat(self) -> None:
if self._hosted_combat_scheduler is not None:
await self._hosted_combat_scheduler.close()
async def update_ai_seat_settings(self, game_key: str, body: Any, uid: str | None = None) -> dict[str, Any]:
return await self._ai_seats.update(game_key, body, uid)
async def retry_ai_seat(self, game_key: str, uid: str) -> dict[str, Any]:
return await self._ai_seats.retry(game_key, uid)
async def set_away_control_policy(self, game_key: str, policy: str) -> dict[str, Any]:
return await self._game_controls.set_away_control_policy(game_key, policy)
async def set_player_access(self, game_key: str, open_access: bool) -> dict[str, Any]:
return await self._game_controls.set_player_access(game_key, open_access)
def check_request_for_action(
self,
game_key: str,
user_id: str,
text: str,
selected_attribute: str = "",
selected_skill: str = "",
target_text: str = "",
) -> dict[str, Any] | None:
return self._game_controls.check_request_for_action(
game_key, user_id, text, selected_attribute, selected_skill, target_text
)
def roll_for_game(self, game_key: str) -> dict[str, Any]:
return self._game_controls.roll_for_game(game_key)
async def resolve_pending_dice_for_game(self, game_key: str, user_id: str = "", source: str = "system") -> dict[str, Any]:
return await self._game_controls.resolve_pending_dice(game_key, user_id, source)
async def resolve_luck_decision(self, game_key: str, check_id: str, actor_uid: str, spend: bool) -> dict[str, Any]:
return await self._game_controls.resolve_luck_decision(
game_key, check_id, actor_uid, spend,
)
async def decline_pending_luck(self, game_key: str) -> dict[str, Any]:
return await self._game_controls.decline_pending_luck(game_key)
async def submit_action(self, game_key: str, actor_uid: str, text: str, **kwargs) -> turns.TurnResult:
return await turns.submit_action(
self._turn_dependencies,
game_key,
actor_uid,
text,
**kwargs,
)
async def ask_kp_question(
self,
game_key: str,
actor_uid: str,
question: str,
visibility: Literal["private", "party"] = "private",
) -> kp_questions.KPQuestionResult:
return await self._kp_questions.ask(
game_key, actor_uid, question, visibility,
)
async def resolve_luck_and_continue(
self,
game_key: str,
check_id: str,
actor_uid: str,
spend: bool,
**kwargs,
) -> turns.TurnResult:
return await turns.resolve_luck_and_continue(
self._turn_dependencies,
game_key,
check_id,
actor_uid,
spend,
**kwargs,
)
async def resolve_push_and_continue(self, game_key: str, check_id: str, viewer: Any, operation: str, body: Any, **kwargs) -> turns.TurnResult:
decision = await self._check_decisions.update(game_key, check_id, viewer, operation, body)
inst = self.get_game_instance(game_key)
uid = inst.gm_uid if viewer.is_gm and inst is not None else viewer.uid
return await turns.continue_after_decision(self._turn_dependencies, game_key, uid, decision, **kwargs)
def game_preferences(self, game_key: str, viewer: Any, *, include_defaults: bool = False) -> dict:
return self._game_preferences.get(game_key, viewer, include_defaults=include_defaults)
async def update_game_preferences(self, game_key: str, viewer: Any, values: Any, **kwargs) -> turns.TurnResult:
result = await self._game_preferences.update(game_key, viewer, values)
if result.get('ok') and result.get('closed_push_decisions'):
inst = self.get_game_instance(game_key)
try:
resumed = await turns.continue_after_decision(self._turn_dependencies, game_key,
inst.gm_uid if inst is not None else viewer.uid, result, **kwargs)
except Exception:
logger.exception('Preferences saved but round continuation failed: %s', game_key)
resumed = {'payload': {'ok': False, 'error_code': 'RESUME_FAILED'}}
# Preferences are already durable; a subsequent generation failure
# must not make the UI revert the switch.
return {'payload': {**result, 'resume': resumed['payload']}, 'status': 200}
status = 200 if result.get('ok') else {'GAME_NOT_FOUND': 404, 'GM_ONLY': 403,
'REWRITE_IN_PROGRESS': 409, 'STALE_RUN': 409, 'ROUND_PROCESSING': 409, 'SAVE_FAILED': 500}.get(result.get('error_code'), 422)
return {'payload': result, 'status': status}
async def advance_turn(self, game_key: str, actor_uid: str, **kwargs) -> turns.TurnResult:
return await turns.advance_round(
self._turn_dependencies,
game_key,
actor_uid,
**kwargs,
)
def private_log(self, game_key: str) -> dict[str, Any]:
return game_queries.private_log(self._game_query_dependencies, game_key)
def private_log_for_user(self, game_key: str, user_id: str) -> dict[str, Any]:
return game_queries.private_log_for_user(
self._game_query_dependencies,
game_key,
user_id,
)
def table_talk(self, game_key: str) -> dict[str, Any]:
return game_queries.table_talk(self._game_query_dependencies, game_key)
def game_health(self, game_key: str, include_resolved: bool = False) -> dict[str, Any]:
return game_queries.game_health(
self._game_query_dependencies,
game_key,
include_resolved,
)
async def set_solo_mode(self, game_key: str, solo: bool) -> dict[str, Any]:
return await self._game_controls.set_solo_mode(game_key, solo)
async def set_narrative_perspective(
self, game_key: str, perspective: str,
) -> dict[str, Any]:
return await self._game_controls.set_narrative_perspective(
game_key, perspective,
)
async def set_gm_style(self, game_key: str, raw: Any) -> dict[str, Any]:
return await self._game_controls.set_gm_style(game_key, raw)
async def mark_game_health_event(
self,
game_key: str,
event_id: str,
*,
resolved: bool = False,
ignored: bool = False,
) -> dict[str, Any]:
return await self._game_controls.mark_health_event(
game_key, event_id, resolved=resolved, ignored=ignored,
)
async def gm_command(self, game_key: str, command: str, mode: str = "note") -> dict[str, Any]:
return await self._game_master.command(game_key, command, mode)
async def rollback_round(self, game_key: str) -> dict[str, Any]:
return await self._game_master.rollback_round(game_key)
async def generate_story_recap(self, game_key: str) -> dict[str, Any]:
return await self._game_master.generate_story_recap(game_key)
async def gm_private_message(self, game_key: str, user_id: str, text: str) -> dict[str, Any]:
return await self._game_master.private_message(game_key, user_id, text)
# ---- 角色卡库 ----
def list_character_cards(self) -> dict[str, Any]:
return character_cards.list_character_cards(self._character_card_dependencies)
def list_shareable_character_cards(self) -> dict[str, Any]:
return character_cards.list_shareable_character_cards(self._character_card_dependencies)
def save_character_card(self, character: dict) -> dict[str, Any]:
return character_cards.save_character_card(
self._character_card_dependencies, character,
)
def update_character_card(self, card_id: str, patch: dict[str, Any]) -> dict[str, Any]:
return character_cards.update_character_card(
self._character_card_dependencies, card_id, patch,
)
def delete_character_card(self, card_id: str) -> dict[str, Any]:
return character_cards.delete_character_card(
self._character_card_dependencies, card_id,
)
async def import_character_card(self, file_data: str = "", file_name: str = "card.json",
target: str = "character_card", world_id: str = "",
include_character_book: bool = True,
character_uid: str = "") -> dict[str, Any]:
return await character_cards.import_character_card(
self._character_card_dependencies,
file_data,
file_name,
target,
world_id,
include_character_book=include_character_book,
character_uid=character_uid,
)
def export_character_cards(self, card_ids: list[str]) -> dict[str, Any]:
return character_cards.export_character_cards(
self._character_card_dependencies, card_ids,
)
def _card_import_dependencies(self) -> character_card_import.CardImportDependencies:
deps = self._character_card_dependencies
return character_card_import.CardImportDependencies(
read_cards=lambda: character_cards.read_library(deps),
write_cards=lambda cards: character_cards.write_library(deps, cards),
lock=lambda: character_cards.library_lock(deps),
to_card=character_cards.to_library_card,
new_card_id=character_cards.new_card_id,
is_ruleset_card=deps.is_ruleset_card,
lorebook=deps.lorebook,
)
def _content_sync_dependencies(self) -> content_sync.ContentSyncDependencies:
def importers(pushed_by_device: str) -> dict[str, Any]:
found: dict[str, Any] = {
"character-card": character_card_import.CardSyncImporter(
self._card_import_dependencies(), pushed_by_device=pushed_by_device,
),
}
if self._lore is not None:
found["lorebook-v3"] = LorebookSyncImporter(self._lore)
return found
def exporters() -> dict[str, Any]:
found: dict[str, Any] = {
"character-card": character_card_import.CardSyncExporter(self._card_import_dependencies()),
}
if self._lore is not None:
found["lorebook-v3"] = LorebookSyncExporter(self._lore)
return found
return content_sync.ContentSyncDependencies(importers=importers, exporters=exporters)
def preview_content_import(self, body: dict[str, Any], *, pushed_by_device: str = "") -> dict[str, Any]:
return content_sync.preview_import(
self._content_sync_dependencies(), body, pushed_by_device=pushed_by_device,
)
def commit_content_import(self, body: dict[str, Any], *, pushed_by_device: str = "") -> dict[str, Any]:
return content_sync.commit_import(
self._content_sync_dependencies(), body, pushed_by_device=pushed_by_device,
)
def content_status(self, body: dict[str, Any]) -> dict[str, Any]:
return content_sync.content_status(self._content_sync_dependencies(), body)
def list_content(self, body: dict[str, Any], *, max_limit: int) -> dict[str, Any]:
return content_sync.list_content(self._content_sync_dependencies(), body, max_limit=max_limit)
def export_content(self, body: dict[str, Any], *, server_instance_id: str) -> dict[str, Any]:
return content_sync.export_content(
self._content_sync_dependencies(), body, server_instance_id=server_instance_id,
)
def preview_character_card_import(self, body: dict[str, Any]) -> dict[str, Any]:
return character_card_import.preview_card_import(self._card_import_dependencies(), body)
def commit_character_card_plan(
self, body: dict[str, Any], *, pushed_by_device: str = "",
) -> dict[str, Any]:
return character_card_import.commit_card_import(
self._card_import_dependencies(), body, pushed_by_device=pushed_by_device,
)
def update_ruleset_character_card_profile(
self, card_id: str, patch: dict[str, Any],
) -> dict[str, Any]:
return ruleset_characters.update_character_card_profile(
self._ruleset_character_dependencies, card_id, patch,
)
def preview_character_card_advancement(
self, card_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return ruleset_advancement.preview_card(
self._card_advancement_dependencies, card_id, body,
)
def apply_character_card_advancement(
self, card_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return ruleset_advancement.apply_card(
self._card_advancement_dependencies, card_id, body,
)
# ---- 世界编辑器 ----
def list_worlds(self) -> dict[str, Any]:
# 确保已启用插件的世界模板世界书已同步（幂等）
if self._plugins:
try:
plugins.sync_plugin_lorebooks(self._plugin_content_dependencies)
except Exception:
logger.warning("list_worlds 同步插件世界书失败，已跳过", exc_info=True)
return worlds.list_worlds(self._world_dependencies)
def create_world(
self, name: str, description: str = "", language: str = "",
*, world_setting: str | None = None, starter_scene: str | None = None,
) -> dict[str, Any]:
return worlds.create_world(
self._world_dependencies, name, description, language,
world_setting=world_setting, starter_scene=starter_scene,
)
def update_world(
self, world_id: str, name: str, description: str = "", language: str = "",
*, world_setting: str | None = None, starter_scene: str | None = None,
) -> dict[str, Any]:
return worlds.update_world(
self._world_dependencies, world_id, name, description, language,
world_setting=world_setting, starter_scene=starter_scene,
)
def get_world_definition(self, world_id: str, language: str = "") -> dict[str, Any]:
return worlds.get_world_definition(self._world_dependencies, world_id, language)
def clone_world_from_template(self, template_id: str, name: str = "") -> dict[str, Any]:
return worlds.clone_world_from_template(
self._world_dependencies, template_id, name,
)
def update_world_gm_style(self, world_id: str, raw: dict | None = None) -> dict[str, Any]:
return worlds.update_world_gm_style(
self._world_dependencies, world_id, raw,
)
def set_user_world_scene_image(self, world_id: str, scene_image: dict | None = None) -> dict[str, Any]:
return worlds.set_user_world_scene_image(
self._world_dependencies, world_id, scene_image,
)
def list_entries(self, world_id: str, entry_type: str | None = None) -> dict[str, Any]:
return worlds.list_entries(
self._world_dependencies, world_id, entry_type,
)
def search_entries(self, world_id: str, keyword: str) -> dict[str, Any]:
return worlds.search_entries(self._world_dependencies, world_id, keyword)
def get_entry(self, entry_id: str) -> dict[str, Any] | None:
return worlds.get_entry(self._world_dependencies, entry_id)
def save_entry(self, entry: dict) -> dict[str, Any]:
return worlds.save_entry(self._world_dependencies, entry)
def import_entries(self, world_id: str, entries: list) -> dict[str, Any]:
return worlds.import_entries(self._world_dependencies, world_id, entries)
def _lorebook_dependencies(self) -> lorebooks.LorebookDependencies:
registry = getattr(self, "_reg", None)
handler = getattr(self, "_handler", None)
return lorebooks.LorebookDependencies(
lorebook=self._lore,
get_instance=lambda key: (
registry.get(_parse_game_key(key)) if registry is not None and key else None
),
get_lore_retriever=lambda: getattr(handler, "lore_retriever", None),
)
def _lorebook_usage_dependencies(self) -> lorebook_usage.LorebookUsageDependencies:
registry = getattr(self, "_reg", None)
return lorebook_usage.LorebookUsageDependencies(
lorebook=self._lore,
get_instance=lambda key: (
registry.get(_parse_game_key(key)) if registry is not None and key else None
),
)
def list_lorebooks(self, world_id: str = "", game_key: str = "", include_all: bool = False) -> dict[str, Any]:
return lorebooks.list_lorebooks(self._lorebook_dependencies(), world_id, game_key, include_all=include_all)
def lorebook_usage_summary(self, game_key: str = "", viewer: str = "gm") -> dict[str, Any]:
return lorebook_usage.usage_summary(
self._lorebook_usage_dependencies(), game_key, viewer,
)
def create_lorebook(self, book: dict[str, Any]) -> dict[str, Any]:
return lorebooks.create_lorebook(self._lorebook_dependencies(), book)
def update_lorebook(self, book_id: str, updates: dict[str, Any]) -> dict[str, Any]:
return lorebooks.update_lorebook(self._lorebook_dependencies(), book_id, updates)
def delete_lorebook(self, book_id: str) -> dict[str, Any]:
return lorebooks.delete_lorebook(self._lorebook_dependencies(), book_id)
def list_lorebook_bindings(self, book_id: str) -> dict[str, Any]:
return lorebooks.list_lorebook_bindings(self._lorebook_dependencies(), book_id)
def create_lorebook_binding(self, book_id: str, binding: dict[str, Any]) -> dict[str, Any]:
return lorebooks.create_lorebook_binding(self._lorebook_dependencies(), book_id, binding)
def update_lorebook_binding(self, binding_id: str, updates: dict[str, Any]) -> dict[str, Any]:
return lorebooks.update_lorebook_binding(self._lorebook_dependencies(), binding_id, updates)
def delete_lorebook_binding(self, binding_id: str) -> dict[str, Any]:
return lorebooks.delete_lorebook_binding(self._lorebook_dependencies(), binding_id)
def list_lorebook_entries(self, book_id: str) -> dict[str, Any]:
return lorebooks.list_lorebook_entries(self._lorebook_dependencies(), book_id)
def save_lorebook_entry(self, book_id: str, entry: dict[str, Any]) -> dict[str, Any]:
return lorebooks.save_lorebook_entry(self._lorebook_dependencies(), book_id, entry)
def move_lorebook_entry(
self, book_id: str, entry_id: str, target_book_id: str,
) -> dict[str, Any]:
return lorebooks.move_lorebook_entry(self._lorebook_dependencies(), book_id, entry_id, target_book_id)
def delete_lorebook_entry(self, book_id: str, entry_id: str) -> dict[str, Any]:
return lorebooks.delete_lorebook_entry(self._lorebook_dependencies(), book_id, entry_id)
def export_lorebook(self, book_id: str) -> dict[str, Any]:
return lorebooks.export_lorebook(self._lorebook_dependencies(), book_id)
async def lorebook_activation_preview(self, payload: dict[str, Any]) -> dict[str, Any]:
return await lorebooks.lorebook_activation_preview(self._lorebook_dependencies(), payload)
def preview_lorebook_import(self, payload: dict[str, Any]) -> dict[str, Any]:
return lorebooks.preview_lorebook_import(self._lorebook_dependencies(), payload)
def commit_lorebook_import(
self,
payload: dict[str, Any],
binding: dict[str, Any] | None = None,
book_id: str | None = None,
plan_digest: str = "",
duplicate_policy: str = "update",
) -> dict[str, Any]:
return lorebooks.commit_lorebook_import(
self._lorebook_dependencies(), payload, binding, book_id,
plan_digest, duplicate_policy,
)
def commit_lorebook_plan(self, body: dict[str, Any]) -> dict[str, Any]:
return lorebooks.commit_lorebook_plan(self._lorebook_dependencies(), body)
async def generate_lorebook_entries(
self, world_id: str, prompt: str, language: str = "", book_id: str = "", game_key: str = "",
) -> dict[str, Any]:
return await worlds.generate_lorebook_entries(
self._world_dependencies, world_id, prompt, language, book_id, game_key,
)
def update_entry(self, entry_id: str, updates: dict) -> dict[str, Any]:
return worlds.update_entry(self._world_dependencies, entry_id, updates)
def delete_entry(self, entry_id: str) -> dict[str, Any]:
return worlds.delete_entry(self._world_dependencies, entry_id)
def delete_world(self, world_id: str) -> dict[str, Any]:
return worlds.delete_world(self._world_dependencies, world_id)
def preview_lore_visibility(
self,
world_id: str,
viewer: str,
game_key: str | None = None,
) -> dict[str, Any]:
return self._lore_preview.preview(world_id, viewer, game_key)
def _rebuild_lorebook_index(self, world_id: str) -> None:
worlds.rebuild_lorebook_index(self._world_dependencies, world_id)
def _refresh_game_lorebook_index(self, world_id: str) -> None:
self._handler._last_matcher_world_id = None
self._rebuild_lorebook_index(world_id)
# ---- 角色管理 ----
def list_characters(self, game_key: str, *, viewer_is_gm: bool) -> dict[str, Any]:
return characters.list_characters(
self._character_dependencies, game_key, viewer_is_gm=viewer_is_gm,
)
def character_schema(self, rule_id: str, language: str = "") -> dict[str, Any]:
return characters.character_schema(
self._character_dependencies, rule_id, language,
)
def get_character(self, game_key: str, user_id: str) -> dict[str, Any] | None:
return characters.get_character(
self._character_dependencies, game_key, user_id,
)
async def update_character(
self, game_key: str, user_id: str, updates: dict, *, gm_authority: bool = False,
) -> dict[str, Any]:
return await characters.update_character(
self._character_dependencies, game_key, user_id, updates,
gm_authority=gm_authority,
)
async def update_ruleset_character_profile(
self, game_key: str, user_id: str, patch: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_characters.update_live_character_profile(
self._ruleset_character_dependencies, game_key, user_id, patch,
)
async def adopt_ruleset_character_card(
self, game_key: str, user_id: str, card_id: str, *, gm_authority: bool,
) -> dict[str, Any]:
"""Adopt a library card by id; classic games apply it server-side.
``gm_authority`` is the caller's: once the seat has acted a
player-side caller gets ``ADOPT_REQUIRES_GM`` on either path.
"""
card = next(
(
item for item in self.list_character_cards()["cards"]
if card_id and card_id in {str(item.get("id") or ""), str(item.get("card_id") or "")}
),
None,
)
if card is not None:
result = await characters.adopt_library_card(
self._character_dependencies, game_key, user_id, card,
gm_authority=gm_authority,
)
if result.get("error_code") != "RULESET_CHARACTER_OPERATION_REQUIRED":
return result
return await ruleset_characters.adopt_character_card(
self._ruleset_character_dependencies, game_key, user_id, card_id,
gm_authority=gm_authority,
)
def preview_live_character_advancement(
self, game_key: str, user_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return ruleset_advancement.preview_live(
self._live_advancement_dependencies, game_key, user_id, body,
)
async def apply_live_character_advancement(
self, game_key: str, user_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_advancement.apply_live(
self._live_advancement_dependencies, game_key, user_id, body,
)
def live_advancement_status(self, game_key: str) -> dict[str, Any]:
return ruleset_advancement.live_status(
self._live_advancement_dependencies, game_key,
)
async def control_live_advancement(
self, game_key: str, body: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_advancement.control_live(
self._live_advancement_dependencies, game_key, body,
)
async def update_npc_portrait(self, game_key: str, npc_id: str, portrait: Any) -> dict[str, Any]:
return await characters.update_npc_portrait(
self._character_dependencies, game_key, npc_id, portrait,
)
async def update_npc_standing_portrait(
self, game_key: str, npc_id: str,
standing_portrait: Any = characters.UNCHANGED,
expressions: Any = characters.UNCHANGED,
) -> dict[str, Any]:
return await characters.update_npc_standing_portrait(
self._character_dependencies, game_key, npc_id,
standing_portrait, expressions,
)
async def update_npc_voice(
self, game_key: str, npc_id: str, voice_profile: Any,
) -> dict[str, Any]:
return await characters.update_npc_voice(
self._character_dependencies, game_key, npc_id, voice_profile,
)
async def set_story_mode(self, game_key: str, mode: Any) -> dict[str, Any]:
return await self._storytelling.set_mode(game_key, mode)
async def set_story_speaker_portrait(
self, game_key: str, speaker_id: str, portrait: Any,
) -> dict[str, Any]:
return await self._storytelling.set_portrait(game_key, speaker_id, portrait)
def list_story_speakers(self, game_key: str) -> dict[str, Any]:
return self._storytelling.list_speakers(game_key)
async def merge_story_speakers(self, game_key: str, source: Any, target: Any) -> dict[str, Any]:
return await self._storytelling.merge_speakers(game_key, source, target)
async def resolve_payment(self, game_key: str, payment_id: str, accepted: bool, session_uid: str = "") -> dict[str, Any]:
return await characters.resolve_payment(
self._character_dependencies,
game_key,
payment_id,
accepted,
session_uid,
)
async def resolve_reward_as_gm(self, game_key: str, payment_id: str, session_uid: str = "") -> dict[str, Any]:
"""Settle one narrative reward through the standard payment-confirm path.
Used by the round-completion auto-settler: the authority chain
(ledger, deferred effects, save) is the manual confirmation path —
only the GM click is removed for qualifying rewards.
"""
return await self.resolve_payment(game_key, payment_id, True, session_uid)
def economy_auto_reward_settings(self, instance: Any = None) -> tuple[bool, int]:
"""按局解析奖励自动结算 (enabled, gold_cap)。
优先级：本局 GM 覆盖（economy_state.reward_policy）→ 规则模板
economy_defaults（D&D 金币与 COC 美元的量级不同）→ 服务器全局配置
兜底。规则模板读取失败时按无默认值处理，不影响本局覆盖与全局兜底。
"""
state = self._config_state if isinstance(self._config_state, dict) else {}
global_enabled = bool(state.get("economy_auto_reward_enabled", True))
global_cap = max(1, int(state.get("economy_auto_reward_gold_cap", 500) or 500))
if instance is None:
return (global_enabled, global_cap)
rule_template: Any = None
try:
rule = self._load_rule_for_game(instance)
rule_template = getattr(rule, "template", None)
except Exception:
logger.warning("读取规则模板奖励默认值失败，回退全局配置", exc_info=True)
return resolve_auto_reward_policy(
game_policy=economy_state.reward_policy(instance),
rule_template=rule_template,
global_enabled=global_enabled,
global_cap=global_cap,
)
async def _settle_adventure_completion_for(self, instance: Any) -> dict[str, Any] | None:
"""回合提交后结算叙事者节点完成提议（GAL 轨道 P6）。
规则未声明 ``adventure_progression`` 时不做任何事（提议保持
advisory-only）。确认走标准完成事务（gates / world ops / ending
冻结 / 奖励出口，全有或全无回滚）；场景过渡叙事为纯文本，不向
叙事者开放状态写权。
"""
try:
rule = self._load_rule_for_game(instance)
except Exception:
logger.warning("读取规则失败，跳过节点完成提议结算", exc_info=True)
return None
config = getattr(rule, "adventure_progression", None) if rule is not None else None
if config is None or not getattr(config, "auto_confirm_completion", False):
return None
async def confirm(node_id: str) -> dict[str, Any]:
# 与 GM 命令（adventure_commands.execute）同一并发边界：authoritative
# write + instance lock + STALE_RUN 检查 + 判定中拒绝；被拒时按
# settle 契约丢弃提议。
async with instance.authoritative_write() as write_entered, instance._lock:
if not write_entered:
raise RuntimeError("historical rewrite is in progress")
if self._reg.get(instance.game_key) is not instance:
raise RuntimeError("game run was replaced")
if instance.state == GameState.ACTIVE_JUDGMENT:
# Another round is being judged; confirming now could reach
# an ending under it. Keep the proposal: that round's own
# settlement retries it once the round has committed.
raise RuntimeError("a round is being judged")
before = {
"world_state": deepcopy(instance.world_state),
"adventure_progress": deepcopy(adventure_runtime_state.progress(instance)),
"economy": deepcopy(economy_state.state(instance)),
}
try:
try:
result = adventure_runtime.complete_adventure_node(
self._adventure_runtime_dependencies,
instance,
node_id,
actor_uid="narrator-auto",
)
except (
adventure_runtime.AdventureRuntimeError,
ProgressError,
OutcomeError,
# Deterministic failures of the transaction itself (e.g.
# WorldStateError from world ops) reject the proposal like
# the GM command does; retrying them would block the queue.
ValueError,
KeyError,
TypeError,
) as exc:
raise adventure_effects.AdventureCompletionRejected(
str(exc),
) from exc
# Persist the authority mutation before narration runs. A
# failed save restores the full before-image so the
# proposal remains recoverable and can be retried.
await self._reg.save(instance)
return result
except BaseException:
adventure_runtime.restore_completion_before_image(instance, before)
raise
async def discard(node_id: str) -> None:
"""Drop a rejected proposal under the same authority boundary.
A failed confirmation still mutates the proposal queue. Keep that
cleanup serialized with node completion and persist it immediately;
otherwise a restart can resurrect a proposal rejected in memory.
"""
async with instance.authoritative_write() as write_entered, instance._lock:
if not write_entered:
raise RuntimeError("historical rewrite is in progress")
if self._reg.get(instance.game_key) is not instance:
raise RuntimeError("game run was replaced")
before_progress = deepcopy(adventure_runtime_state.progress(instance))
try:
adventure_runtime.discard_completion_suggestion(instance, node_id)
await self._reg.save(instance)
except BaseException:
adventure_runtime.restore_completion_progress_before_image(
instance, before_progress,
)
raise
narration_fn = None
if getattr(config, "activation_narration", False):
narrator = SceneTransitionNarrator(
self._llm_client,
getattr(self._handler, "prompt_composer", None),
)
async def narration_fn(result: dict[str, Any], activated_nodes: list[str]) -> str:
return await narrator.narrate(
instance, completed_node_id=str(result.get("node_id") or ""),
)
settlement = await adventure_effects.settle_adventure_completion(
instance,
config=config,
confirm=confirm,
discard=discard,
narration=narration_fn,
)
if settlement is None:
return None
return {
"node_id": settlement.node_id,
"activated_nodes": settlement.activated_nodes,
"transition_narration": settlement.transition_narration,
}
async def set_economy_reward_policy(
self, game_key: str, policy: dict[str, Any],
) -> dict[str, Any]:
instance = self.get_game_instance(game_key)
if instance is None:
return {"ok": False, "error": "not found"}
try:
instance.configure_session(economy_reward_policy=policy)
except ValueError as exc:
return {"ok": False, "error": str(exc)}
await self.save_game_instance(instance)
return {"ok": True, "economy_reward_policy": economy_state.reward_policy(instance)}
async def combat_extension_scheduler_advance(self, game_key: str) -> dict[str, Any]:
"""通用战斗扩展：GM 推进 ATB 时间（gauge 累积至有人就绪）。"""
instance = self.get_game_instance(game_key)
if instance is None:
return {"ok": False, "code": "GAME_NOT_FOUND", "error": "游戏不存在"}
async with instance.authoritative_write() as entered:
if not entered:
return {
"ok": False,
"code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
if self.get_game_instance(game_key) is not instance:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
async with instance._lock:
rule = self._load_rule_for_game(instance)
result = combat_extension_service.scheduler_advance(instance, rule)
if result.get("ok"):
await self.save_game_instance(instance)
return result
async def combat_extension_action(
self, game_key: str, intent: dict[str, Any], *,
session_uid: str, viewer_is_gm: bool,
) -> dict[str, Any]:
"""通用战斗扩展：结算一次动作 intent（数值全部服务端求值）。"""
instance = self.get_game_instance(game_key)
if instance is None:
return {"ok": False, "code": "GAME_NOT_FOUND", "error": "游戏不存在"}
async with instance.authoritative_write() as entered:
if not entered:
return {
"ok": False,
"code": "REWRITE_IN_PROGRESS",
"error": "GM 正在重写历史回合，请等待完成后重试",
}
if self.get_game_instance(game_key) is not instance:
return {"ok": False, "code": "STALE_RUN", "error": "对局已重开，请刷新后重试"}
async with instance._lock:
rule = self._load_rule_for_game(instance)
result = combat_extension_service.resolve_combat_action(
instance, rule, intent,
actor_uid=session_uid, viewer_is_gm=viewer_is_gm,
)
if result.get("ok"):
await self.save_game_instance(instance)
return result
async def create_payment_proposal(
self,
game_key: str,
*,
payer_uid: str,
amount: int,
reason: str = "",
recipient_uid: str = "",
items: list[str] | None = None,
) -> dict[str, Any]:
return await characters.create_payment_proposal(
self._character_dependencies,
game_key,
payer_uid=payer_uid,
amount=amount,
reason=reason,
recipient_uid=recipient_uid,
items=items,
)
async def _drain_economy_outbox(self, instance: Any) -> bool:
return await characters.drain_economy_outbox(
self._character_dependencies, instance,
)
async def drain_economy_outbox(self, game_key: str) -> bool:
instance = self.get_game_instance(game_key)
if instance is None:
return False
return await self._drain_economy_outbox(instance)
async def recover_economy_outboxes(self, instances: list[Any]) -> int:
recovered = 0
for instance in instances:
had_pending = bool(
pending_memory_deliveries(instance)
or pending_memory_reversals(instance)
)
if had_pending and await self._drain_economy_outbox(instance):
recovered += 1
return recovered
async def delete_character(
self, game_key: str, user_id: str, *, gm_authority: bool,
) -> dict[str, Any]:
return await characters.delete_character(
self._character_dependencies, game_key, user_id,
gm_authority=gm_authority,
)
async def create_player(self, game_key: str, character: dict,
force_uid: str = "", assign_new_id: bool = False,
*, seat_token_uid: str = "", require_seat_proof: bool = False) -> dict[str, Any]:
return await characters.create_player(
self._character_dependencies,
game_key,
character,
force_uid,
assign_new_id,
seat_token_uid=seat_token_uid,
require_seat_proof=require_seat_proof,
)
def save_avatar_upload(self, file_data: str, file_name: str = "") -> dict[str, Any]:
return self._avatars.save_upload(file_data, file_name)
def avatar_file(self, asset_id: str) -> Path | None:
return self._avatars.file(asset_id)
def list_user_avatars(self) -> dict[str, Any]:
return self._avatars.list_user_avatars()
def delete_avatar(self, asset_id: str) -> dict[str, Any]:
return self._avatars.delete(asset_id)
# GAL P8：立绘资产（透明保留、content-addressed，见 standing_portraits）。
def save_standing_portrait_upload(self, file_data: str, file_name: str = "") -> dict[str, Any]:
return self._standing_portraits.save_upload(file_data, file_name)
def standing_portrait_file(self, asset_id: str) -> Path | None:
return self._standing_portraits.file(asset_id)
def delete_standing_portrait(self, asset_id: str) -> dict[str, Any]:
return self._standing_portraits.delete(asset_id)
def _standing_portrait_is_referenced(self, asset_id: str) -> bool:
"""Refuse asset deletion while any persisted/live character uses it."""
def contains(value: Any) -> bool:
if isinstance(value, dict):
if (
value.get("kind") == "upload"
and str(value.get("asset_id") or "") == asset_id
):
return True
return any(contains(item) for item in value.values())
if isinstance(value, list):
return any(contains(item) for item in value)
return False
registered_dirs = set()
try:
for instance in self._reg.list_all():
registered_dirs.add(self._reg.save_package_state_path(instance.game_key).parent)
# The whole aggregate, not only players / npcs: round snapshots
# keep deep copies of NPCs that a rollback restores. The save
# encodes only the recent log, but every loaded round can still
# be swiped, so the full in-memory log is scanned too.
if contains(instance.to_dict()) or contains(list(instance.log)):
return True
if self._unregistered_save_mentions(asset_id, registered_dirs):
return True
cards = character_cards._read_cards(self._character_card_dependencies, strict=True)
except Exception:
# A failed reference scan must not make a referenced asset deletable.
logger.exception("立绘引用扫描失败，拒绝删除: %s", asset_id)
return True
return contains(cards)
def _unregistered_save_mentions(self, asset_id: str, registered_dirs: set[Path]) -> bool:
"""True when a save the registry did not load may reference ``asset_id``.
Startup recovery registers every loadable save, so what remains here is
a save that failed to load (corrupt, or written by a newer runtime).
Its schema cannot be trusted, so match the raw text: asset ids are
random hex, and a false match only keeps an asset that could be deleted.
Unreadable files raise and the caller refuses the deletion.
"""
save_dir = self._reg.save_dir
if not save_dir.exists():
return False
for entry in save_dir.iterdir():
if not entry.is_dir() or entry in registered_dirs:
continue
for name in ("state.json", "state.backup.json", "chatlog.jsonl"):
state = entry / name
if state.is_file() and asset_id in state.read_text(encoding="utf-8", errors="replace"):
return True
return False
def _voice_profile_exists(self, voice_id: str) -> bool:
"""True when the personal voice profile still exists (GAL P8)."""
speech_service = getattr(self, "_speech", None)
if speech_service is None:
return True # 无法校验时不省略：合成端仍会 fail closed 报错。
return any(
str(item.get("id") or "") == voice_id
for item in speech_service.editable_voice_profiles()
)
def image_generation_status(self) -> dict[str, Any]:
return self._generated_images.public_config()
async def optimize_image_prompt(self, **request: Any) -> dict[str, Any]:
return await self._generated_images.optimize_prompt(**request)
def storyboard_draft(self, game_key: str, user_id: str, round_number: int = 0) -> dict[str, Any]:
return self._generated_images.storyboard_draft(game_key, user_id, round_number)
async def analyze_storyboard(self, game_key: str, user_id: str, round_number: int = 0, panel_count: int | None = None) -> dict[str, Any]:
return await self._generated_images.analyze_storyboard(game_key, user_id, round_number, panel_count)
def preview_image_prompt(self, game_key: str, user_id: str, prompt: str, panels: Any = None) -> dict[str, Any]:
return self._generated_images.preview_prompt(game_key, user_id, prompt, panels)
async def generate_generated_image(self, **request: Any) -> dict[str, Any]:
return await self._generated_images.generate_image(**request)
async def generate_current_round_image(
self, game_key: str, user_id: str, prompt: str, round_number: int,
panels: Any = None, use_avatar_references: bool = False,
panel_count: int | None = None,
combine_avatar_references: bool = True,
) -> dict[str, Any]:
return await self._generated_images.generate_current_round(
game_key, user_id, prompt, round_number, panels, use_avatar_references, panel_count,
combine_avatar_references,
)
def list_game_generated_images(
self, game_key: str, user_id: str, *, purpose: str = "", viewer_is_gm: bool = False,
) -> list[dict[str, Any]]:
return self._generated_images.list_game_images(
game_key, user_id, purpose=purpose, viewer_is_gm=viewer_is_gm,
)
async def use_generated_image_as_map_background(
self, game_key: str, user_id: str, asset_id: str,
) -> dict[str, Any]:
return await self._generated_images.use_as_map_background(
game_key, user_id, asset_id,
)
def generated_image_file(self, asset_id: str) -> Path | None:
return self._generated_images.image_file(asset_id)
def save_scene_image_upload(self, file_data: str, file_name: str = "") -> dict[str, Any]:
return self._scene_images.save_upload(file_data, file_name)
def scene_image_file(self, asset_id: str) -> Path | None:
return self._scene_images.file(asset_id)
def validate_scene_image_ref(self, reference: Any) -> dict[str, str]:
return self._scene_images.validate(reference)
def resolve_default_scene_image(
self,
world_id: str = "",
rule_id: str = "",
selection_key: str = "",
) -> dict[str, str]:
return self._scene_images.resolve_default(world_id, rule_id, selection_key)
def materialize_scene_image(self, reference: Any) -> dict[str, str]:
return self._scene_images.materialize(reference)
def resolve_scene_image_file(self, reference: Any) -> Path | None:
return self._scene_images.resolve_file(reference)
def package_scene_image(
self,
reference: Any,
files: dict[str, str | bytes],
) -> dict[str, str] | None:
return self._scene_images.package(reference, files)
def save_map_background_upload(self, file_data: str, file_name: str = "") -> dict[str, Any]:
return self._map_backgrounds.save_upload(file_data, file_name)
def map_background_file(self, asset_id: str) -> Path | None:
return self._map_backgrounds.file(asset_id)
def validate_map_background_selection(self, selection: Any) -> dict[str, str]:
return self._map_backgrounds.validate(selection)
def resolve_map_background_file(self, selection: Any) -> Path | None:
return self._map_backgrounds.resolve_file(selection)
# ---- 剧情日志 ----
def get_log(
self,
game_key: str,
page: int = 1,
per_page: int = 50,
include_internal: bool = False,
) -> dict[str, Any]:
return self._game_logs.get_log(
game_key, page, per_page, include_internal,
)
def get_statistics(self, game_key: str) -> dict[str, Any]:
return self._game_logs.get_statistics(game_key)
# ---- 规则配置 ----
def list_rules(self, language: str = "") -> dict[str, Any]:
return rules.list_rules(self._rule_dependencies, language)
def save_custom_rule(self, data: dict[str, Any]) -> dict[str, Any]:
return rules.save_custom_rule(self._rule_dependencies, data)
def get_rule_template(self, rule_id: str, language: str = "") -> dict[str, Any]:
return rules.get_rule_template(self._rule_dependencies, rule_id, language)
def update_custom_rule(self, rule_id: str, template: dict[str, Any]) -> dict[str, Any]:
return rules.update_custom_rule(
self._rule_dependencies,
rule_id,
template,
)
def delete_custom_rule(self, rule_id: str) -> dict[str, Any]:
return rules.delete_custom_rule(self._rule_dependencies, rule_id)
def ruleset_experience(self, rule_id: str, language: str = "") -> dict[str, Any]:
return ruleset_builder.experience(
self._ruleset_builder_dependencies, rule_id, language
)
def ruleset_builder_choices(
self, rule_id: str, draft: Any, language: str = "",
) -> dict[str, Any]:
return ruleset_builder.choices(
self._ruleset_builder_dependencies, rule_id, draft, language
)
def ruleset_builder_validate(
self, rule_id: str, draft: Any, language: str = "",
) -> dict[str, Any]:
return ruleset_builder.validate(
self._ruleset_builder_dependencies, rule_id, draft, language
)
def ruleset_builder_derive(
self, rule_id: str, draft: Any, language: str = "",
) -> dict[str, Any]:
return ruleset_builder.derive(
self._ruleset_builder_dependencies, rule_id, draft, language
)
def ruleset_builder_finalize(
self, rule_id: str, draft: Any, language: str = "",
) -> dict[str, Any]:
return ruleset_builder.finalize(
self._ruleset_builder_dependencies, rule_id, draft, language
)
def ruleset_progression(
self, rule_id: str, class_ref: str, start_level: int = 1,
end_level: int = 20, language: str = "",
) -> dict[str, Any]:
return ruleset_advancement.progression(
self._ruleset_advancement_dependencies,
rule_id, class_ref, start_level, end_level, language,
)
def ruleset_advancement_preview(
self, rule_id: str, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return ruleset_advancement.preview(
self._ruleset_advancement_dependencies, rule_id, body, language,
)
def ruleset_advancement_apply(
self, rule_id: str, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return ruleset_advancement.apply(
self._ruleset_advancement_dependencies, rule_id, body, language,
)
def ruleset_rest_resolve(
self, rule_id: str, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return ruleset_rest.resolve(
self._ruleset_rest_dependencies, rule_id, body, language,
)
async def ruleset_rest_resolve_live(
self, game_key: str, user_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_rest.resolve_live(
self._live_ruleset_rest_dependencies, game_key, user_id, body,
)
async def ruleset_rest_submit_live(
self, game_key: str, user_id: str, body: dict[str, Any],
) -> dict[str, Any]:
"""Shared live-rest application entry point for HTTP and Agent callers."""
return await ruleset_rest.submit_live(
self._live_ruleset_rest_dependencies, game_key, user_id, body,
)
async def ruleset_rest_resolve_live_party(
self, game_key: str, user_id: str, body: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_rest.resolve_live_party(
self._live_ruleset_rest_dependencies, game_key, user_id, body,
)
async def ruleset_available_actions(
self, game_key: str, requester_id: str, requester_is_gm: bool = False,
) -> dict[str, Any]:
return await ruleset_gameplay.available_actions(
self._ruleset_gameplay_dependencies,
game_key,
requester_id,
requester_is_gm,
)
async def ruleset_submit_intent(
self, game_key: str, requester_id: str, requester_is_gm: bool,
body: dict[str, Any],
) -> dict[str, Any]:
return await ruleset_gameplay.submit_intent(
self._ruleset_gameplay_dependencies,
game_key,
requester_id,
requester_is_gm,
body,
)
async def ruleset_plan_temporary_encounter(
self, game_key: str, requester_id: str, requester_is_gm: bool,
) -> dict[str, Any]:
return await ruleset_gameplay.plan_temporary_encounter(
self._ruleset_gameplay_dependencies,
game_key,
requester_id,
requester_is_gm,
)
# ---- 世界模板 ----
def refresh_content_sources(self) -> None:
"""Recompute plugin adventure sources + module catalogs from host state.
FIX-07：安装 / 启用 / 停用 / 卸载 / 启动都经过这一个刷新入口。解析与目录
读取都不得依赖"碰巧同步过 registry"（§3.6 同一原则）——否则"启用模组后
马上开局"或"重启后马上开局"会解析不到模组的冒险包。
"""
self._sync_plugin_adventure_sources()
self._sync_module_catalogs()
def _sync_plugin_adventure_sources(self) -> None:
"""MOD-03：把启用的 content-pack 模组声明的 adventure_packages 同步为
registry 的 plugin 来源（declared-only loader）。
只读 plugin runtime 状态，幂等重建 plugin 来源集合；插件禁用/卸载后
其来源自动消失。
"""
if self._adventure_source_registry is None or self._plugins is None:
return
sources = []
for runtime in self._plugins.plugins.values():
root = getattr(runtime, "adventure_packages_root", None)
if root is None or runtime.status == "disabled":
continue
sources.append(AdventureSource(
"plugin",
str(runtime.manifest.get("id") or ""),
AdventureBundleLoader(
root,
allowed_directory_ids=getattr(runtime, "adventure_package_directories", ()),
),
))
self._adventure_source_registry.sync_plugin_sources(sources)
def _sync_module_catalogs(self) -> None:
"""DNDMOD-04：把启用的 content-pack 模组声明的 ruleset_catalogs 装载为
模块目录来源（DNDMOD-01 契约校验，fail closed；坏包让该模块目录缺席
并记日志，不拖垮其它模块）。"""
if self._plugins is None:
return
from src.rulesets.dnd2024.content.catalog import load_catalog_dir
sources: list[tuple[str, Any]] = []
for runtime in self._plugins.plugins.values():
root = getattr(runtime, "ruleset_catalogs_root", None)
if root is None or runtime.status == "disabled":
continue
label = f"module:{runtime.manifest.get('id') or ''}"
for directory in getattr(runtime, "ruleset_catalog_directories", ()):
try:
source = load_catalog_dir(Path(root) / directory, source_label=label)
except Exception as exc:
logger.warning(
"module catalog load failed: %s: %s", label, exc,
)
continue
# catalog_from_sources / 运行时期望 {kind: {record_id: record}}；
# CatalogSource.records 是 (kind, id) 键——这里统一成嵌套形状。
nested: dict[str, dict[str, Any]] = {}
for (kind, record_id), record in source.records.items():
nested.setdefault(str(kind), {})[str(record_id)] = record
sources.append((source.label, nested))
self._dnd_module_catalogs = sources
def module_catalog(self) -> Any:
"""DNDMOD-04：合并的模块内容目录（每次调用前同步）。"""
self._sync_plugin_adventure_sources()
self._sync_module_catalogs()
from src.rulesets.dnd2024.content.catalog import catalog_from_sources
if not self._dnd_module_catalogs:
return catalog_from_sources([])
return catalog_from_sources(self._dnd_module_catalogs)
def module_content_sources(self, instance: Any = None) -> list[tuple[str, Any]]:
"""FIX-03 §5.1：模组 catalog 来源链（owning module 优先，其余按 label）。
只提供"哪些模组声明了哪些 catalog"这一事实；链路语义由 runtime 决定。
"""
self._sync_module_catalogs()
sources = list(self._dnd_module_catalogs)
owning = ""
binding = getattr(instance, "adventure_binding", None)
if isinstance(binding, dict) and str(binding.get("source_kind") or "") == "plugin":
owning = str(binding.get("source_id") or "")
if owning:
owner_label = f"module:{owning}"
sources.sort(key=lambda item: 0 if item[0] == owner_label else 1)
return sources
def list_adventures(
self, rule_id: str = "", world_id: str = "", language: str = "",
) -> dict[str, Any]:
self._sync_plugin_adventure_sources()
return adventures.list_adventures(
self._adventure_dependencies,
rule_id,
world_id,
language,
)
def adventure_detail(self, adventure_id: str, language: str = "") -> dict[str, Any]:
self._sync_plugin_adventure_sources()
return adventures.adventure_detail(
self._adventure_dependencies,
adventure_id,
language,
)
def copy_adventure(
self, adventure_id: str, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return adventures.copy_adventure(
self._adventure_dependencies,
adventure_id,
body,
language,
)
def create_adventure(
self, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return adventures.create_adventure(
self._adventure_dependencies,
body,
language,
)
def update_adventure(
self, adventure_id: str, body: dict[str, Any], language: str = "",
) -> dict[str, Any]:
return adventures.update_adventure(
self._adventure_dependencies,
adventure_id,
body,
language,
)
def delete_adventure(self, adventure_id: str) -> dict[str, Any]:
return adventures.delete_adventure(
self._adventure_dependencies,
adventure_id,
)
def export_adventure(self, adventure_id: str) -> tuple[str, bytes]:
return adventures.export_adventure(
self._adventure_dependencies,
adventure_id,
)
def import_adventure(
self, payload: bytes, directory_id: str = "",
) -> dict[str, Any]:
return adventures.import_adventure(
self._adventure_dependencies,
payload,
directory_id,
)
def preview_adventure_script(self, payload: bytes, filename: str, language: str = "") -> dict[str, Any]:
from src.webui.services import adventure_script_import
return adventure_script_import.preview(self._adventure_dependencies, payload, filename, language)
def prepare_adventure_script_source(self, payload: bytes, filename: str, language: str = "") -> dict[str, Any]:
from src.adventures.external_import import parse_import
return parse_import(payload, filename, self._adventure_dependencies.default_runtime_requirement(), language)
def preview_content_text(self, text: str, filename: str, language: str = "") -> dict[str, Any]:
return self.content_text_review.preview(text, filename, language)
def review_content_text(self, draft: dict[str, Any]) -> dict[str, Any]:
return self.content_text_review.review(draft)
def commit_content_text(self, body: dict[str, Any]) -> dict[str, Any]:
return self.content_text_review.commit(body)
def review_adventure_script(self, draft: dict[str, Any]) -> dict[str, Any]:
from src.webui.services import adventure_script_import
return adventure_script_import.review(self._adventure_dependencies, draft)
async def organize_adventure_script(
self, draft: dict[str, Any], instructions: str, language: str,
emit: Callable[[dict[str, Any]], Awaitable[None]],
) -> dict[str, Any]:
from src.webui.services.adventure_ai_import import AdventureAiImportDependencies, organize
dependencies = AdventureAiImportDependencies(self.generate_text, self.review_adventure_script)
return await organize(dependencies, draft, instructions, language, emit)
def commit_adventure_script(self, draft: dict[str, Any]) -> dict[str, Any]:
from src.webui.services import adventure_script_import
return adventure_script_import.commit(self._adventure_dependencies, draft)
def list_world_templates(self, language: str = "") -> dict[str, Any]:
# 确保已启用插件的世界模板世界书已同步（幂等）
if self._plugins:
try:
plugins.sync_plugin_lorebooks(self._plugin_content_dependencies)
except Exception:
logger.warning("list_world_templates 同步插件世界书失败，已跳过", exc_info=True)
return worlds.list_world_templates(self._world_dependencies, language)
def cleanup_orphan_game_templates(self, world_id: str = "") -> int:
return worlds.cleanup_orphan_game_templates(
self._world_dependencies, world_id,
)
# ---- 创建游戏 ----
async def create_game(self, world_id: str, game_name: str = "",
group_name: str = "Web端", rule_id: str = "",
solo: bool = False,
lorebook_world_id: str = "",
difficulty: str = "标准",
description: str = "",
create_lorebook: bool = False,
blank_lorebook: bool = False,
source_world_id: str = "",
players: list[dict] | None = None,
custom_world: bool = False,
gm_uid: str = "",
room_password: str = "",
language: str = "",
scene_image: dict[str, Any] | None = None,
map_background: dict[str, Any] | None = None,
adventure_id: str = "",
adventure_source_kind: str = "",
adventure_source_id: str = "",
play_mode: str = "",
narrative_perspective: Any = UNSET,
story_mode: Any = UNSET,
gm_style_override: Any = UNSET,
advancement_mode: Any = UNSET,
advancement_authority: Any = UNSET,
unclaimed_control_default: str = "",
world_ref: dict[str, Any] | None = None,
book_bindings: list[dict[str, Any]] | None = None) -> dict[str, Any]:
return await self._game_lifecycle.create_game(
world_id=world_id, game_name=game_name, group_name=group_name,
rule_id=rule_id, solo=solo, lorebook_world_id=lorebook_world_id,
difficulty=difficulty, description=description,
create_lorebook=create_lorebook, blank_lorebook=blank_lorebook,
source_world_id=source_world_id, players=players,
custom_world=custom_world, gm_uid=gm_uid,
room_password=room_password, language=language,
scene_image=scene_image, map_background=map_background,
adventure_id=adventure_id,
adventure_source_kind=adventure_source_kind,
adventure_source_id=adventure_source_id,
play_mode=play_mode,
narrative_perspective=narrative_perspective,
story_mode=story_mode,
gm_style_override=gm_style_override,
advancement_mode=advancement_mode,
advancement_authority=advancement_authority,
unclaimed_control_default=unclaimed_control_default,
world_ref=world_ref,
book_bindings=book_bindings,
)
# ---- 重开引用码 ----
async def reset_game(self, game_key: str) -> dict[str, Any]:
return await self._game_lifecycle.reset_game(game_key)
async def restart_game(self, game_key: str) -> dict[str, Any]:
return await self._game_lifecycle.restart_game(game_key)
async def switch_world(self, game_key: str, world_id: str) -> dict[str, Any]:
return await self._game_media.switch_world(game_key, world_id)
async def create_from_seed(self, seed_code: str, solo: bool = False,
players: list[dict] | None = None,
gm_uid: str = "",
language: str = "",
scene_image: dict[str, Any] | None = None,
narrative_perspective: str = "",
story_mode: str = "") -> dict[str, Any]:
return await self._game_lifecycle.create_from_seed(
seed_code=seed_code, solo=solo, players=players, gm_uid=gm_uid,
language=language, scene_image=scene_image,
narrative_perspective=narrative_perspective,
story_mode=story_mode,
)
async def update_scene_image(
self,
game_key: str,
reference: dict[str, Any] | None = None,
*,
use_default: bool = False,
) -> dict[str, Any]:
return await self._game_media.update_scene_image(
game_key, reference, use_default=use_default,
)
async def update_map_background(
self,
game_key: str,
selection: dict[str, Any] | None = None,
) -> dict[str, Any]:
return await maps.update_map_background(
self._map_dependencies,
game_key,
selection,
)
# ---- AI 生成 ----
async def test_connection(self, base_url: str, api_key: str,
model: str, proxy_url: str = "",
api_format: str = "openai") -> dict[str, Any]:
return await generation.test_connection(
self._connection_dependencies,
base_url,
api_key,
model,
proxy_url,
api_format,
)
async def list_models(self, base_url: str, api_key: str,
proxy_url: str = "", api_format: str = "openai") -> dict[str, Any]:
return await generation.list_models(
self._connection_dependencies,
base_url,
api_key,
proxy_url,
api_format,
)
async def generate_world(self, prompt: str, rule_id: str = "", language: str = "") -> dict[str, Any]:
return await generation.generate_world(
self._generation_dependencies,
prompt,
rule_id,
language,
)
async def generate_rule(self, prompt: str, source_rule_id: str = "", language: str = "") -> dict[str, Any]:
return await generation.generate_rule(
self._generation_dependencies,
prompt,
source_rule_id,
language,
)
async def generate_character(self, prompt: str, game_key: str = "", rule_id: str = "", language: str = "") -> dict[str, Any]:
return await generation.generate_character(
self._generation_dependencies,
prompt,
game_key,
rule_id,
language,
)
async def generate_text(self, prompt: str, system_hint: str = "", language: str = "") -> dict[str, Any]:
return await generation.generate_text(
self._generation_dependencies,
prompt,
system_hint,
language,
)
# ---- 内存 ----
def list_memories(self, game_key: str, keyword: str = "",
limit: int = 20, offset: int = 0, *,
viewer_is_gm: bool = False) -> dict[str, Any]:
"""列出记忆；默认按玩家安全面过滤（GM 私有记忆需显式 viewer_is_gm）。"""
return self._memory_service.list(
game_key, keyword, limit, offset, viewer_is_gm=bool(viewer_is_gm),
)
async def update_memory(self, game_key: str, entry_id: int, updates: dict[str, Any]) -> dict[str, Any]:
return await self._memory_service.update(game_key, entry_id, updates)
async def delete_memory(self, game_key: str, entry_id: int) -> dict[str, Any]:
return await self._memory_service.delete(game_key, entry_id)
async def import_tavern_card(self, file_path: str = "", file_data: str = "",
file_name: str = "card.png", game_key: str = "") -> dict[str, Any]:
return await self._tavern_import.import_card(
file_path, file_data, file_name, game_key,
)
# ----
def get_map_locations(self, game_key: str, *, viewer: Viewer) -> dict[str, Any]:
return maps.get_map_locations(self._map_dependencies, game_key, viewer=viewer)
def map_background_asset(self, game_key: str, asset_id: str) -> Path | None:
return maps.map_background_asset(
self._map_dependencies,
game_key,
asset_id,
)
@staticmethod
def _parse_key(game_key: str) -> tuple:
return _parse_game_key(game_key)