1

server-rs/crates/spacetime-module/src/lib.rs

@@ -15,6 +15,21 @@ use module_assets::{
     INITIAL_ASSET_OBJECT_VERSION, validate_asset_entity_binding_fields,
     validate_asset_object_fields,
 };
+use module_big_fish::{
+    BigFishAgentMessageKind, BigFishAgentMessageRole, BigFishAgentMessageSnapshot,
+    BigFishAssetGenerateInput, BigFishAssetKind, BigFishAssetSlotSnapshot, BigFishAssetStatus,
+    BigFishCreationStage, BigFishDraftCompileInput, BigFishMessageSubmitInput, BigFishPublishInput,
+    BigFishRunGetInput, BigFishRunInputSubmitInput, BigFishRunProcedureResult, BigFishRunStartInput,
+    BigFishRunStatus, BigFishRuntimeSnapshot, BigFishSessionCreateInput, BigFishSessionGetInput,
+    BigFishSessionProcedureResult, BigFishSessionSnapshot,
+    advance_runtime_snapshot, build_asset_coverage, build_generated_asset_slot,
+    build_initial_runtime_snapshot, compile_default_draft, deserialize_anchor_pack,
+    deserialize_draft, deserialize_runtime_snapshot, empty_anchor_pack, infer_anchor_pack,
+    serialize_anchor_pack, serialize_asset_coverage, serialize_draft, serialize_runtime_snapshot,
+    validate_asset_generate_input, validate_draft_compile_input, validate_message_submit_input,
+    validate_publish_input, validate_run_get_input, validate_run_input_submit_input,
+    validate_run_start_input, validate_session_create_input, validate_session_get_input,
+};
 use module_combat::{
     BATTLE_STATE_ID_PREFIX, BattleMode, BattleStateInput, BattleStateProcedureResult,
     BattleStateQueryInput, BattleStateSnapshot, BattleStatus, CombatOutcome,
@@ -131,6 +146,8 @@ use shared_kernel::format_timestamp_micros;
 use serde_json::{Map as JsonMap, Value as JsonValue, json};
 use spacetimedb::{ProcedureContext, ReducerContext, SpacetimeType, Table, Timestamp};
 
+mod puzzle;
+
 // 这层输入只服务 NPC 开战编排；普通聊天、援手、招募继续走已有 resolve_npc_interaction 接口。
 #[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
 pub struct ResolveNpcBattleInteractionInput {
@@ -162,6 +179,77 @@ pub struct NpcBattleInteractionProcedureResult {
     pub error_message: Option<String>,
 }
 
+#[spacetimedb::table(
+    accessor = big_fish_creation_session,
+    index(accessor = by_big_fish_session_owner_user_id, btree(columns = [owner_user_id]))
+)]
+pub struct BigFishCreationSession {
+    #[primary_key]
+    session_id: String,
+    owner_user_id: String,
+    seed_text: String,
+    current_turn: u32,
+    progress_percent: u32,
+    stage: BigFishCreationStage,
+    anchor_pack_json: String,
+    draft_json: Option<String>,
+    asset_coverage_json: String,
+    last_assistant_reply: Option<String>,
+    publish_ready: bool,
+    created_at: Timestamp,
+    updated_at: Timestamp,
+}
+
+#[spacetimedb::table(
+    accessor = big_fish_agent_message,
+    index(accessor = by_big_fish_message_session_id, btree(columns = [session_id]))
+)]
+pub struct BigFishAgentMessage {
+    #[primary_key]
+    message_id: String,
+    session_id: String,
+    role: BigFishAgentMessageRole,
+    kind: BigFishAgentMessageKind,
+    text: String,
+    created_at: Timestamp,
+}
+
+#[spacetimedb::table(
+    accessor = big_fish_asset_slot,
+    index(accessor = by_big_fish_asset_session_id, btree(columns = [session_id]))
+)]
+pub struct BigFishAssetSlot {
+    #[primary_key]
+    slot_id: String,
+    session_id: String,
+    asset_kind: BigFishAssetKind,
+    level: Option<u32>,
+    motion_key: Option<String>,
+    status: BigFishAssetStatus,
+    asset_url: Option<String>,
+    prompt_snapshot: String,
+    updated_at: Timestamp,
+}
+
+#[spacetimedb::table(
+    accessor = big_fish_runtime_run,
+    index(accessor = by_big_fish_run_owner_user_id, btree(columns = [owner_user_id])),
+    index(accessor = by_big_fish_run_session_id, btree(columns = [session_id]))
+)]
+pub struct BigFishRuntimeRun {
+    #[primary_key]
+    run_id: String,
+    session_id: String,
+    owner_user_id: String,
+    status: BigFishRunStatus,
+    snapshot_json: String,
+    last_input_x: f32,
+    last_input_y: f32,
+    tick: u64,
+    created_at: Timestamp,
+    updated_at: Timestamp,
+}
+
 #[spacetimedb::table(
     accessor = asset_object,
     index(accessor = by_bucket_object_key, btree(columns = [bucket, object_key]))
@@ -1491,6 +1579,177 @@ pub fn get_story_session_state(
     }
 }
 
+#[spacetimedb::procedure]
+pub fn create_big_fish_session(
+    ctx: &mut ProcedureContext,
+    input: BigFishSessionCreateInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| create_big_fish_session_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn get_big_fish_session(
+    ctx: &mut ProcedureContext,
+    input: BigFishSessionGetInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| get_big_fish_session_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn submit_big_fish_message(
+    ctx: &mut ProcedureContext,
+    input: BigFishMessageSubmitInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| submit_big_fish_message_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn compile_big_fish_draft(
+    ctx: &mut ProcedureContext,
+    input: BigFishDraftCompileInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| compile_big_fish_draft_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn generate_big_fish_asset(
+    ctx: &mut ProcedureContext,
+    input: BigFishAssetGenerateInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| generate_big_fish_asset_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn publish_big_fish_game(
+    ctx: &mut ProcedureContext,
+    input: BigFishPublishInput,
+) -> BigFishSessionProcedureResult {
+    match ctx.try_with_tx(|tx| publish_big_fish_game_tx(tx, input.clone())) {
+        Ok(session) => BigFishSessionProcedureResult {
+            ok: true,
+            session: Some(session),
+            error_message: None,
+        },
+        Err(message) => BigFishSessionProcedureResult {
+            ok: false,
+            session: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn start_big_fish_run(
+    ctx: &mut ProcedureContext,
+    input: BigFishRunStartInput,
+) -> BigFishRunProcedureResult {
+    match ctx.try_with_tx(|tx| start_big_fish_run_tx(tx, input.clone())) {
+        Ok(run) => BigFishRunProcedureResult {
+            ok: true,
+            run: Some(run),
+            error_message: None,
+        },
+        Err(message) => BigFishRunProcedureResult {
+            ok: false,
+            run: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn submit_big_fish_input(
+    ctx: &mut ProcedureContext,
+    input: BigFishRunInputSubmitInput,
+) -> BigFishRunProcedureResult {
+    match ctx.try_with_tx(|tx| submit_big_fish_input_tx(tx, input.clone())) {
+        Ok(run) => BigFishRunProcedureResult {
+            ok: true,
+            run: Some(run),
+            error_message: None,
+        },
+        Err(message) => BigFishRunProcedureResult {
+            ok: false,
+            run: None,
+            error_message: Some(message),
+        },
+    }
+}
+
+#[spacetimedb::procedure]
+pub fn get_big_fish_run(
+    ctx: &mut ProcedureContext,
+    input: BigFishRunGetInput,
+) -> BigFishRunProcedureResult {
+    match ctx.try_with_tx(|tx| get_big_fish_run_tx(tx, input.clone())) {
+        Ok(run) => BigFishRunProcedureResult {
+            ok: true,
+            run: Some(run),
+            error_message: None,
+        },
+        Err(message) => BigFishRunProcedureResult {
+            ok: false,
+            run: None,
+            error_message: Some(message),
+        },
+    }
+}
+
 // Stage 6 先把 Agent 会话骨架写入 SpacetimeDB，LLM 采集与卡片生成后续再接入。
 #[spacetimedb::procedure]
 pub fn create_custom_world_agent_session(
@@ -1658,6 +1917,465 @@ fn get_story_session_state_tx(
     Ok((session_snapshot, events))
 }
 
+fn create_big_fish_session_tx(
+    ctx: &ReducerContext,
+    input: BigFishSessionCreateInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    validate_session_create_input(&input).map_err(|error| error.to_string())?;
+    if ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .is_some()
+    {
+        return Err("big_fish_creation_session.session_id 已存在".to_string());
+    }
+    if ctx
+        .db
+        .big_fish_agent_message()
+        .message_id()
+        .find(&input.welcome_message_id)
+        .is_some()
+    {
+        return Err("big_fish_agent_message.message_id 已存在".to_string());
+    }
+
+    let created_at = Timestamp::from_micros_since_unix_epoch(input.created_at_micros);
+    let anchor_pack = infer_anchor_pack(&input.seed_text, None);
+    let asset_coverage = build_asset_coverage(None, &[]);
+    ctx.db
+        .big_fish_creation_session()
+        .insert(BigFishCreationSession {
+            session_id: input.session_id.clone(),
+            owner_user_id: input.owner_user_id.clone(),
+            seed_text: input.seed_text.trim().to_string(),
+            current_turn: 0,
+            progress_percent: 20,
+            stage: BigFishCreationStage::CollectingAnchors,
+            anchor_pack_json: serialize_anchor_pack(&anchor_pack)
+                .map_err(|error| error.to_string())?,
+            draft_json: None,
+            asset_coverage_json: serialize_asset_coverage(&asset_coverage)
+                .map_err(|error| error.to_string())?,
+            last_assistant_reply: Some(input.welcome_message_text.clone()),
+            publish_ready: false,
+            created_at,
+            updated_at: created_at,
+        });
+    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
+        message_id: input.welcome_message_id,
+        session_id: input.session_id.clone(),
+        role: BigFishAgentMessageRole::Assistant,
+        kind: BigFishAgentMessageKind::Chat,
+        text: input.welcome_message_text,
+        created_at,
+    });
+
+    get_big_fish_session_tx(
+        ctx,
+        BigFishSessionGetInput {
+            session_id: input.session_id,
+            owner_user_id: input.owner_user_id,
+        },
+    )
+}
+
+fn get_big_fish_session_tx(
+    ctx: &ReducerContext,
+    input: BigFishSessionGetInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    validate_session_get_input(&input).map_err(|error| error.to_string())?;
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+
+    build_big_fish_session_snapshot(ctx, &session)
+}
+
+fn submit_big_fish_message_tx(
+    ctx: &ReducerContext,
+    input: BigFishMessageSubmitInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    validate_message_submit_input(&input).map_err(|error| error.to_string())?;
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    if ctx
+        .db
+        .big_fish_agent_message()
+        .message_id()
+        .find(&input.user_message_id)
+        .is_some()
+    {
+        return Err("big_fish_agent_message.user_message_id 已存在".to_string());
+    }
+    if ctx
+        .db
+        .big_fish_agent_message()
+        .message_id()
+        .find(&input.assistant_message_id)
+        .is_some()
+    {
+        return Err("big_fish_agent_message.assistant_message_id 已存在".to_string());
+    }
+
+    let submitted_at = Timestamp::from_micros_since_unix_epoch(input.submitted_at_micros);
+    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
+        message_id: input.user_message_id,
+        session_id: input.session_id.clone(),
+        role: BigFishAgentMessageRole::User,
+        kind: BigFishAgentMessageKind::Chat,
+        text: input.user_message_text.trim().to_string(),
+        created_at: submitted_at,
+    });
+
+    let anchor_pack = infer_anchor_pack(&session.seed_text, Some(&input.user_message_text));
+    let assistant_text = "我已经把这版方向收束成 4 个高杠杆锚点，可以继续细化，也可以直接编译第一版玩法草稿。"
+        .to_string();
+    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
+        message_id: input.assistant_message_id,
+        session_id: input.session_id.clone(),
+        role: BigFishAgentMessageRole::Assistant,
+        kind: BigFishAgentMessageKind::Summary,
+        text: assistant_text.clone(),
+        created_at: submitted_at,
+    });
+
+    let next_session = BigFishCreationSession {
+        session_id: session.session_id.clone(),
+        owner_user_id: session.owner_user_id.clone(),
+        seed_text: session.seed_text.clone(),
+        current_turn: session.current_turn.saturating_add(1),
+        progress_percent: 60,
+        stage: BigFishCreationStage::CollectingAnchors,
+        anchor_pack_json: serialize_anchor_pack(&anchor_pack).map_err(|error| error.to_string())?,
+        draft_json: session.draft_json.clone(),
+        asset_coverage_json: session.asset_coverage_json.clone(),
+        last_assistant_reply: Some(assistant_text),
+        publish_ready: session.publish_ready,
+        created_at: session.created_at,
+        updated_at: submitted_at,
+    };
+    replace_big_fish_session(ctx, &session, next_session);
+
+    get_big_fish_session_tx(
+        ctx,
+        BigFishSessionGetInput {
+            session_id: input.session_id,
+            owner_user_id: input.owner_user_id,
+        },
+    )
+}
+
+fn compile_big_fish_draft_tx(
+    ctx: &ReducerContext,
+    input: BigFishDraftCompileInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    validate_draft_compile_input(&input).map_err(|error| error.to_string())?;
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    let anchor_pack =
+        deserialize_anchor_pack(&session.anchor_pack_json).map_err(|error| error.to_string())?;
+    let draft = compile_default_draft(&anchor_pack);
+    let asset_slots = list_big_fish_asset_slots(ctx, &session.session_id);
+    let coverage = build_asset_coverage(Some(&draft), &asset_slots);
+    let compiled_at = Timestamp::from_micros_since_unix_epoch(input.compiled_at_micros);
+    let reply = "第一版玩法草稿已编译完成，可以在结果页逐级生成主图、动作和场地背景。".to_string();
+
+    let next_session = BigFishCreationSession {
+        session_id: session.session_id.clone(),
+        owner_user_id: session.owner_user_id.clone(),
+        seed_text: session.seed_text.clone(),
+        current_turn: session.current_turn,
+        progress_percent: 80,
+        stage: BigFishCreationStage::DraftReady,
+        anchor_pack_json: session.anchor_pack_json.clone(),
+        draft_json: Some(serialize_draft(&draft).map_err(|error| error.to_string())?),
+        asset_coverage_json: serialize_asset_coverage(&coverage)
+            .map_err(|error| error.to_string())?,
+        last_assistant_reply: Some(reply.clone()),
+        publish_ready: coverage.publish_ready,
+        created_at: session.created_at,
+        updated_at: compiled_at,
+    };
+    replace_big_fish_session(ctx, &session, next_session);
+    append_big_fish_system_message(
+        ctx,
+        &input.session_id,
+        format!("big-fish-message-compile-{}", input.compiled_at_micros),
+        reply,
+        input.compiled_at_micros,
+    );
+
+    get_big_fish_session_tx(
+        ctx,
+        BigFishSessionGetInput {
+            session_id: input.session_id,
+            owner_user_id: input.owner_user_id,
+        },
+    )
+}
+
+fn generate_big_fish_asset_tx(
+    ctx: &ReducerContext,
+    input: BigFishAssetGenerateInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    let draft = session
+        .draft_json
+        .as_deref()
+        .ok_or_else(|| "big_fish.draft 尚未编译".to_string())
+        .and_then(|value| deserialize_draft(value).map_err(|error| error.to_string()))?;
+    validate_asset_generate_input(&input, &draft).map_err(|error| error.to_string())?;
+
+    let slot = build_generated_asset_slot(
+        &input.session_id,
+        &draft,
+        input.asset_kind,
+        input.level,
+        input.motion_key.clone(),
+        input.generated_at_micros,
+    )
+    .map_err(|error| error.to_string())?;
+    upsert_big_fish_asset_slot(ctx, slot);
+
+    let asset_slots = list_big_fish_asset_slots(ctx, &session.session_id);
+    let coverage = build_asset_coverage(Some(&draft), &asset_slots);
+    let updated_at = Timestamp::from_micros_since_unix_epoch(input.generated_at_micros);
+    let reply = match input.asset_kind {
+        BigFishAssetKind::LevelMainImage => "本级主图已生成并设为正式资产。",
+        BigFishAssetKind::LevelMotion => "本级动作已生成并设为正式资产。",
+        BigFishAssetKind::StageBackground => "活动区域背景已生成并设为正式资产。",
+    }
+    .to_string();
+    let next_stage = if coverage.publish_ready {
+        BigFishCreationStage::ReadyToPublish
+    } else {
+        BigFishCreationStage::AssetRefining
+    };
+    let next_session = BigFishCreationSession {
+        session_id: session.session_id.clone(),
+        owner_user_id: session.owner_user_id.clone(),
+        seed_text: session.seed_text.clone(),
+        current_turn: session.current_turn,
+        progress_percent: if coverage.publish_ready { 96 } else { 88 },
+        stage: next_stage,
+        anchor_pack_json: session.anchor_pack_json.clone(),
+        draft_json: session.draft_json.clone(),
+        asset_coverage_json: serialize_asset_coverage(&coverage)
+            .map_err(|error| error.to_string())?,
+        last_assistant_reply: Some(reply.clone()),
+        publish_ready: coverage.publish_ready,
+        created_at: session.created_at,
+        updated_at,
+    };
+    replace_big_fish_session(ctx, &session, next_session);
+    append_big_fish_system_message(
+        ctx,
+        &input.session_id,
+        format!("big-fish-message-asset-{}", input.generated_at_micros),
+        reply,
+        input.generated_at_micros,
+    );
+
+    get_big_fish_session_tx(
+        ctx,
+        BigFishSessionGetInput {
+            session_id: input.session_id,
+            owner_user_id: input.owner_user_id,
+        },
+    )
+}
+
+fn publish_big_fish_game_tx(
+    ctx: &ReducerContext,
+    input: BigFishPublishInput,
+) -> Result<BigFishSessionSnapshot, String> {
+    validate_publish_input(&input).map_err(|error| error.to_string())?;
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    let draft = session
+        .draft_json
+        .as_deref()
+        .ok_or_else(|| "big_fish.draft 尚未编译".to_string())
+        .and_then(|value| deserialize_draft(value).map_err(|error| error.to_string()))?;
+    let coverage = build_asset_coverage(Some(&draft), &list_big_fish_asset_slots(ctx, &session.session_id));
+    if !coverage.publish_ready {
+        return Err(format!("big_fish 发布校验未通过：{}", coverage.blockers.join("；")));
+    }
+
+    let published_at = Timestamp::from_micros_since_unix_epoch(input.published_at_micros);
+    let next_session = BigFishCreationSession {
+        session_id: session.session_id.clone(),
+        owner_user_id: session.owner_user_id.clone(),
+        seed_text: session.seed_text.clone(),
+        current_turn: session.current_turn,
+        progress_percent: 100,
+        stage: BigFishCreationStage::Published,
+        anchor_pack_json: session.anchor_pack_json.clone(),
+        draft_json: session.draft_json.clone(),
+        asset_coverage_json: serialize_asset_coverage(&coverage)
+            .map_err(|error| error.to_string())?,
+        last_assistant_reply: Some("玩法已发布，可以进入测试运行态。".to_string()),
+        publish_ready: true,
+        created_at: session.created_at,
+        updated_at: published_at,
+    };
+    replace_big_fish_session(ctx, &session, next_session);
+
+    get_big_fish_session_tx(
+        ctx,
+        BigFishSessionGetInput {
+            session_id: input.session_id,
+            owner_user_id: input.owner_user_id,
+        },
+    )
+}
+
+fn start_big_fish_run_tx(
+    ctx: &ReducerContext,
+    input: BigFishRunStartInput,
+) -> Result<BigFishRuntimeSnapshot, String> {
+    validate_run_start_input(&input).map_err(|error| error.to_string())?;
+    if ctx
+        .db
+        .big_fish_runtime_run()
+        .run_id()
+        .find(&input.run_id)
+        .is_some()
+    {
+        return Err("big_fish_runtime_run.run_id 已存在".to_string());
+    }
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&input.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    let draft = session
+        .draft_json
+        .as_deref()
+        .ok_or_else(|| "big_fish.draft 尚未编译".to_string())
+        .and_then(|value| deserialize_draft(value).map_err(|error| error.to_string()))?;
+    let snapshot = build_initial_runtime_snapshot(
+        input.run_id.clone(),
+        input.session_id.clone(),
+        &draft,
+        input.started_at_micros,
+    );
+    let now = Timestamp::from_micros_since_unix_epoch(input.started_at_micros);
+    ctx.db.big_fish_runtime_run().insert(BigFishRuntimeRun {
+        run_id: input.run_id,
+        session_id: input.session_id,
+        owner_user_id: input.owner_user_id,
+        status: snapshot.status,
+        snapshot_json: serialize_runtime_snapshot(&snapshot).map_err(|error| error.to_string())?,
+        last_input_x: 0.0,
+        last_input_y: 0.0,
+        tick: snapshot.tick,
+        created_at: now,
+        updated_at: now,
+    });
+
+    Ok(snapshot)
+}
+
+fn submit_big_fish_input_tx(
+    ctx: &ReducerContext,
+    input: BigFishRunInputSubmitInput,
+) -> Result<BigFishRuntimeSnapshot, String> {
+    validate_run_input_submit_input(&input).map_err(|error| error.to_string())?;
+    let run = ctx
+        .db
+        .big_fish_runtime_run()
+        .run_id()
+        .find(&input.run_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_runtime_run 不存在".to_string())?;
+    let session = ctx
+        .db
+        .big_fish_creation_session()
+        .session_id()
+        .find(&run.session_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
+    let draft = session
+        .draft_json
+        .as_deref()
+        .ok_or_else(|| "big_fish.draft 尚未编译".to_string())
+        .and_then(|value| deserialize_draft(value).map_err(|error| error.to_string()))?;
+    let current_snapshot =
+        deserialize_runtime_snapshot(&run.snapshot_json).map_err(|error| error.to_string())?;
+    let next_snapshot = advance_runtime_snapshot(
+        current_snapshot,
+        &draft.runtime_params,
+        input.input_x,
+        input.input_y,
+        input.submitted_at_micros,
+    );
+    replace_big_fish_run(
+        ctx,
+        &run,
+        BigFishRuntimeRun {
+            run_id: run.run_id.clone(),
+            session_id: run.session_id.clone(),
+            owner_user_id: run.owner_user_id.clone(),
+            status: next_snapshot.status,
+            snapshot_json: serialize_runtime_snapshot(&next_snapshot)
+                .map_err(|error| error.to_string())?,
+            last_input_x: input.input_x,
+            last_input_y: input.input_y,
+            tick: next_snapshot.tick,
+            created_at: run.created_at,
+            updated_at: Timestamp::from_micros_since_unix_epoch(input.submitted_at_micros),
+        },
+    );
+
+    Ok(next_snapshot)
+}
+
+fn get_big_fish_run_tx(
+    ctx: &ReducerContext,
+    input: BigFishRunGetInput,
+) -> Result<BigFishRuntimeSnapshot, String> {
+    validate_run_get_input(&input).map_err(|error| error.to_string())?;
+    let run = ctx
+        .db
+        .big_fish_runtime_run()
+        .run_id()
+        .find(&input.run_id)
+        .filter(|row| row.owner_user_id == input.owner_user_id)
+        .ok_or_else(|| "big_fish_runtime_run 不存在".to_string())?;
+
+    deserialize_runtime_snapshot(&run.snapshot_json).map_err(|error| error.to_string())
+}
+
 fn create_custom_world_agent_session_tx(
     ctx: &ReducerContext,
     input: CustomWorldAgentSessionCreateInput,
@@ -7908,3 +8626,155 @@ fn build_npc_state_snapshot_from_row(row: &NpcState) -> NpcStateSnapshot {
         updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
     }
 }
+
+fn build_big_fish_session_snapshot(
+    ctx: &ReducerContext,
+    row: &BigFishCreationSession,
+) -> Result<BigFishSessionSnapshot, String> {
+    let anchor_pack =
+        deserialize_anchor_pack(&row.anchor_pack_json).unwrap_or_else(|_| empty_anchor_pack());
+    let draft = row
+        .draft_json
+        .as_deref()
+        .map(deserialize_draft)
+        .transpose()
+        .map_err(|error| format!("big_fish.draft_json 非法: {error}"))?;
+    let asset_slots = list_big_fish_asset_slots(ctx, &row.session_id);
+    let asset_coverage = build_asset_coverage(draft.as_ref(), &asset_slots);
+    let mut messages = ctx
+        .db
+        .big_fish_agent_message()
+        .iter()
+        .filter(|message| message.session_id == row.session_id)
+        .map(|message| BigFishAgentMessageSnapshot {
+            message_id: message.message_id,
+            session_id: message.session_id,
+            role: message.role,
+            kind: message.kind,
+            text: message.text,
+            created_at_micros: message.created_at.to_micros_since_unix_epoch(),
+        })
+        .collect::<Vec<_>>();
+    messages.sort_by_key(|message| (message.created_at_micros, message.message_id.clone()));
+
+    Ok(BigFishSessionSnapshot {
+        session_id: row.session_id.clone(),
+        owner_user_id: row.owner_user_id.clone(),
+        seed_text: row.seed_text.clone(),
+        current_turn: row.current_turn,
+        progress_percent: row.progress_percent,
+        stage: row.stage,
+        anchor_pack,
+        draft,
+        asset_slots,
+        asset_coverage,
+        messages,
+        last_assistant_reply: row.last_assistant_reply.clone(),
+        publish_ready: row.publish_ready,
+        created_at_micros: row.created_at.to_micros_since_unix_epoch(),
+        updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
+    })
+}
+
+fn list_big_fish_asset_slots(
+    ctx: &ReducerContext,
+    session_id: &str,
+) -> Vec<BigFishAssetSlotSnapshot> {
+    let mut slots = ctx
+        .db
+        .big_fish_asset_slot()
+        .iter()
+        .filter(|slot| slot.session_id == session_id)
+        .map(|slot| BigFishAssetSlotSnapshot {
+            slot_id: slot.slot_id,
+            session_id: slot.session_id,
+            asset_kind: slot.asset_kind,
+            level: slot.level,
+            motion_key: slot.motion_key,
+            status: slot.status,
+            asset_url: slot.asset_url,
+            prompt_snapshot: slot.prompt_snapshot,
+            updated_at_micros: slot.updated_at.to_micros_since_unix_epoch(),
+        })
+        .collect::<Vec<_>>();
+    slots.sort_by_key(|slot| {
+        (
+            slot.level.unwrap_or(0),
+            slot.asset_kind.as_str().to_string(),
+            slot.motion_key.clone().unwrap_or_default(),
+            slot.slot_id.clone(),
+        )
+    });
+    slots
+}
+
+fn replace_big_fish_session(
+    ctx: &ReducerContext,
+    current: &BigFishCreationSession,
+    next: BigFishCreationSession,
+) {
+    ctx.db
+        .big_fish_creation_session()
+        .session_id()
+        .delete(&current.session_id);
+    ctx.db.big_fish_creation_session().insert(next);
+}
+
+fn replace_big_fish_run(ctx: &ReducerContext, current: &BigFishRuntimeRun, next: BigFishRuntimeRun) {
+    ctx.db
+        .big_fish_runtime_run()
+        .run_id()
+        .delete(&current.run_id);
+    ctx.db.big_fish_runtime_run().insert(next);
+}
+
+fn upsert_big_fish_asset_slot(ctx: &ReducerContext, slot: BigFishAssetSlotSnapshot) {
+    if let Some(existing) = ctx
+        .db
+        .big_fish_asset_slot()
+        .slot_id()
+        .find(&slot.slot_id)
+    {
+        ctx.db
+            .big_fish_asset_slot()
+            .slot_id()
+            .delete(&existing.slot_id);
+    }
+    ctx.db.big_fish_asset_slot().insert(BigFishAssetSlot {
+        slot_id: slot.slot_id,
+        session_id: slot.session_id,
+        asset_kind: slot.asset_kind,
+        level: slot.level,
+        motion_key: slot.motion_key,
+        status: slot.status,
+        asset_url: slot.asset_url,
+        prompt_snapshot: slot.prompt_snapshot,
+        updated_at: Timestamp::from_micros_since_unix_epoch(slot.updated_at_micros),
+    });
+}
+
+fn append_big_fish_system_message(
+    ctx: &ReducerContext,
+    session_id: &str,
+    message_id: String,
+    text: String,
+    created_at_micros: i64,
+) {
+    if ctx
+        .db
+        .big_fish_agent_message()
+        .message_id()
+        .find(&message_id)
+        .is_some()
+    {
+        return;
+    }
+    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
+        message_id,
+        session_id: session_id.to_string(),
+        role: BigFishAgentMessageRole::Assistant,
+        kind: BigFishAgentMessageKind::ActionResult,
+        text,
+        created_at: Timestamp::from_micros_since_unix_epoch(created_at_micros),
+    });
+}