feat(engine): add story arc system for narrative-driven world

- Add StoryArc and StoryEventResult models
- Create story_arcs.py with progress tracking logic
- Implement two story arcs: civil_unrest and golden_age
- Progress updates based on faction power balance
- Trigger story events when progress reaches threshold
- Add 3 tests for story arc system

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

engine-python/app/engine.py

@@ -2,12 +2,14 @@ from __future__ import annotations
 from typing import List, Optional, Tuple
 from .models import (
     WorldState, AgentState, Event, Action, Emotion, Weather,
-    GlobalEventResult, GlobalEventInfo
+    GlobalEventResult, GlobalEventInfo, FactionEventResult
 )
 from .global_events import GLOBAL_EVENT_POOL, GlobalEvent
 from .opinions import generate_opinions
 from .social_influence import apply_social_influence
-from .factions import update_factions, apply_faction_influence
+from .factions import (
+    update_factions, apply_faction_influence, check_and_trigger_faction_event
+)
 
 MAX_EVENTS = 20
 MAX_MEMORY_PER_AGENT = 3

engine-python/app/factions.py

@@ -1,40 +1,138 @@
 """派系系统 - 基于立场分类角色并影响世界"""
-from typing import Dict
-from .models import WorldState, AgentState, Factions
+from typing import Dict, Tuple
+from .models import (
+    WorldState, AgentState, Factions, FactionData,
+    FactionEventResult, WorldEffect, Emotion
+)
 
 # 派系分类阈值
 OPTIMIST_THRESHOLD = 0.6
 FEARFUL_THRESHOLD = 0.6
 
+# 阵营技能触发阈值
+FACTION_POWER_THRESHOLD = 5
+
+# 阵营事件对情绪的影响
+FACTION_EVENT_EFFECTS = {
+    "festival": 0.2,   # 乐观派事件提升情绪
+    "panic": -0.3,     # 恐惧派事件降低情绪
+}
+
 
 def classify_faction(agent: AgentState) -> str:
-    """根据 stance 分类角色所属派系"""
-    if agent.stance.optimism > OPTIMIST_THRESHOLD:
+    """根据 emotion 分类角色所属派系
+
+    规则：
+    - emotion 偏正 (happy) → optimists
+    - emotion 偏负 (anxious) → fearful
+    - 其它 (calm) → neutral
+    """
+    if agent.emotion == Emotion.HAPPY:
         return "optimists"
-    elif agent.stance.fear > FEARFUL_THRESHOLD:
+    elif agent.emotion == Emotion.ANXIOUS:
         return "fearful"
     else:
         return "neutral"
 
 
 def update_factions(state: WorldState) -> None:
-    """统计各派系人数并更新 world_state"""
-    counts = {"optimists": 0, "fearful": 0, "neutral": 0}
+    """统计各派系并更新 world_state
 
-    for agent in state.agents.values():
+    流程：
+    1. 清空 factions.members
+    2. 遍历所有 agents，根据 emotion 分类
+    3. 更新 agent.faction 字段
+    4. 将 agent 加入对应 faction.members
+    5. 每个 agent 为 faction 增加 power += 1
+    """
+    # 重置 factions
+    state.factions = Factions(
+        optimists=FactionData(power=0, members=[]),
+        fearful=FactionData(power=0, members=[])
+    )
+
+    for agent_id, agent in state.agents.items():
         faction = classify_faction(agent)
-        counts[faction] += 1
+        agent.faction = faction
 
-    state.factions = Factions(**counts)
+        if faction == "optimists":
+            state.factions.optimists.members.append(agent_id)
+            state.factions.optimists.power += 1
+        elif faction == "fearful":
+            state.factions.fearful.members.append(agent_id)
+            state.factions.fearful.power += 1
+        # neutral 不加入任何派系
 
 
 def apply_faction_influence(state: WorldState) -> None:
-    """派系分布影响世界情绪"""
-    optimists = state.factions.optimists
-    fearful = state.factions.fearful
+    """派系分布影响世界情绪（保留原有逻辑）"""
+    optimists = state.factions.optimists.power
+    fearful = state.factions.fearful.power
 
     if optimists > fearful:
         state.town_mood = min(10, state.town_mood + 1)
     elif fearful > optimists:
         state.town_mood = max(-10, state.town_mood - 1)
-    # 平局时不变化
+
+
+def check_and_trigger_faction_event(state: WorldState) -> FactionEventResult:
+    """检查并触发阵营事件
+
+    规则：
+    - 当 faction.power >= 5 时，触发对应世界事件
+    - optimists → festival（提升情绪）
+    - fearful → panic（降低情绪）
+    - 触发后该 faction.power 归零
+    """
+    result = FactionEventResult(type=None, source_faction=None)
+
+    # 检查乐观派
+    if state.factions.optimists.power >= FACTION_POWER_THRESHOLD:
+        result = FactionEventResult(type="festival", source_faction="optimists")
+        _apply_faction_event(state, "festival")
+        state.factions.optimists.power = 0
+
+    # 检查恐惧派
+    elif state.factions.fearful.power >= FACTION_POWER_THRESHOLD:
+        result = FactionEventResult(type="panic", source_faction="fearful")
+        _apply_faction_event(state, "panic")
+        state.factions.fearful.power = 0
+
+    return result
+
+
+def _apply_faction_event(state: WorldState, event_type: str) -> None:
+    """应用阵营事件效果
+
+    1. 创建 WorldEffect 写入 world_state.world_effects
+    2. 影响所有 agent 的 emotion
+    """
+    # 创建持续影响效果
+    if event_type == "festival":
+        effect = WorldEffect(
+            type="faction_event",
+            name="节日庆典",
+            intensity=1,
+            remaining_ticks=3,
+            mood_modifier=1
+        )
+    else:  # panic
+        effect = WorldEffect(
+            type="faction_event",
+            name="恐慌蔓延",
+            intensity=1,
+            remaining_ticks=3,
+            mood_modifier=-1
+        )
+
+    state.world_effects.append(effect)
+
+    # 影响所有 agent 的 stance（间接影响下一轮 emotion）
+    emotion_delta = FACTION_EVENT_EFFECTS.get(event_type, 0)
+    for agent in state.agents.values():
+        if emotion_delta > 0:
+            agent.stance.optimism = min(1.0, agent.stance.optimism + emotion_delta)
+            agent.stance.fear = max(0.0, agent.stance.fear - emotion_delta * 0.5)
+        else:
+            agent.stance.fear = min(1.0, agent.stance.fear + abs(emotion_delta))
+            agent.stance.optimism = max(0.0, agent.stance.optimism - abs(emotion_delta) * 0.5)

engine-python/app/main.py

@@ -2,6 +2,8 @@ from fastapi import FastAPI
 from .models import StepRequest, StepResponse
 from .state import load_state, save_state
 from .engine import process_events, generate_actions, check_and_trigger_global_event
+from .factions import check_and_trigger_faction_event
+from .story_arcs import update_story_progress, check_and_trigger_story_event
 
 app = FastAPI(title="AI Town Engine", version="0.1.0")
 
@@ -18,6 +20,15 @@ def step(request: StepRequest) -> StepResponse:
     # 检查并触发世界级事件
     state, global_event_result = check_and_trigger_global_event(state)
 
+    # 检查并触发阵营事件
+    faction_event_result = check_and_trigger_faction_event(state)
+
+    # 更新剧情进度
+    update_story_progress(state)
+
+    # 检查并触发剧情事件
+    story_event_result = check_and_trigger_story_event(state)
+
     # 生成 agent 行动
     actions = generate_actions(state)
 
@@ -27,7 +38,9 @@ def step(request: StepRequest) -> StepResponse:
     return StepResponse(
         world_state=state,
         actions=actions,
-        global_event=global_event_result
+        global_event=global_event_result,
+        triggered_faction_event=faction_event_result,
+        story_event=story_event_result
     )
 
 

engine-python/app/models.py

@@ -44,11 +44,39 @@ class Stance(BaseModel):
     fear: float = Field(default=0.5, ge=0.0, le=1.0)
 
 
+class FactionData(BaseModel):
+    """单个派系的数据"""
+    power: int = 0
+    members: List[str] = Field(default_factory=list)
+
+
 class Factions(BaseModel):
-    """派系分布"""
-    optimists: int = 0
-    fearful: int = 0
-    neutral: int = 0
+    """派系分布（带 power 和 members）"""
+    optimists: FactionData = Field(default_factory=FactionData)
+    fearful: FactionData = Field(default_factory=FactionData)
+
+
+class FactionEventResult(BaseModel):
+    """阵营事件触发结果"""
+    type: Optional[str] = None  # "festival" | "panic" | None
+    source_faction: Optional[str] = None  # "optimists" | "fearful"
+
+
+class StoryArc(BaseModel):
+    """剧情线"""
+    progress: float = Field(default=0.0, ge=0.0, le=1.0)
+    threshold: float = 1.0
+    active: bool = True
+    stage: int = 1  # 当前阶段
+    description: str = ""
+
+
+class StoryEventResult(BaseModel):
+    """剧情事件触发结果"""
+    triggered: bool = False
+    arc_id: Optional[str] = None
+    event_name: Optional[str] = None
+    description: Optional[str] = None
 
 
 class AgentState(BaseModel):
@@ -60,6 +88,8 @@ class AgentState(BaseModel):
     commented_effects: List[str] = Field(default_factory=list)
     # 角色立场
     stance: Stance = Field(default_factory=Stance)
+    # 所属阵营
+    faction: str = "neutral"  # "optimists" | "fearful" | "neutral"
 
 
 class WorldState(BaseModel):
@@ -71,6 +101,7 @@ class WorldState(BaseModel):
     global_meter: GlobalMeter = Field(default_factory=GlobalMeter)
     world_effects: List[WorldEffect] = Field(default_factory=list)
     factions: Factions = Field(default_factory=Factions)
+    story_arcs: Dict[str, StoryArc] = Field(default_factory=dict)
 
 
 class Event(BaseModel):
@@ -106,3 +137,5 @@ class StepResponse(BaseModel):
     world_state: WorldState
     actions: List[Action]
     global_event: GlobalEventResult = Field(default_factory=GlobalEventResult)
+    triggered_faction_event: FactionEventResult = Field(default_factory=FactionEventResult)
+    story_event: StoryEventResult = Field(default_factory=StoryEventResult)

engine-python/app/state.py

@@ -1,6 +1,7 @@
 import json
 from pathlib import Path
-from .models import WorldState, AgentState, GlobalMeter, Stance
+from .models import WorldState, AgentState, GlobalMeter, Stance, Factions, FactionData
+from .story_arcs import get_default_story_arcs
 
 STATE_FILE = Path(__file__).parent.parent / "state.json"
 
@@ -17,16 +18,23 @@ def get_default_state() -> WorldState:
                 goal="探索小镇",
                 memory=[],
                 stance=Stance(optimism=0.6, fear=0.4),
+                faction="neutral",
             ),
             "bob": AgentState(
                 emotion="calm",
                 goal="与人交流",
                 memory=[],
                 stance=Stance(optimism=0.4, fear=0.6),
+                faction="neutral",
             ),
         },
         events=[],
         global_meter=GlobalMeter(value=0, threshold=100, cooldown=0),
+        factions=Factions(
+            optimists=FactionData(power=0, members=[]),
+            fearful=FactionData(power=0, members=[])
+        ),
+        story_arcs=get_default_story_arcs(),
     )
 
 

engine-python/app/story_arcs.py

@@ -0,0 +1,115 @@
+"""剧情线系统 - 让事件形成因果链"""
+from typing import Dict, Optional
+from .models import (
+    WorldState, StoryArc, StoryEventResult, WorldEffect
+)
+
+# 剧情进度变化量
+PROGRESS_INCREMENT = 0.1
+PROGRESS_DECREMENT = 0.05
+
+# 剧情事件定义
+STORY_EVENTS = {
+    "civil_unrest": {
+        "name": "民众骚乱",
+        "description": "恐惧派势力过大，城镇爆发了骚乱",
+        "mood_effect": -3,
+        "duration": 5,
+    },
+    "golden_age": {
+        "name": "黄金时代",
+        "description": "乐观派主导了城镇，迎来繁荣期",
+        "mood_effect": 3,
+        "duration": 5,
+    },
+}
+
+
+def get_default_story_arcs() -> Dict[str, StoryArc]:
+    """获取默认剧情线"""
+    return {
+        "civil_unrest": StoryArc(
+            progress=0.0,
+            threshold=1.0,
+            active=True,
+            stage=1,
+            description="城市内部的紧张局势"
+        ),
+        "golden_age": StoryArc(
+            progress=0.0,
+            threshold=1.0,
+            active=True,
+            stage=1,
+            description="城镇繁荣的希望"
+        ),
+    }
+
+
+def update_story_progress(state: WorldState) -> None:
+    """根据阵营状态更新剧情进度"""
+    fearful_power = state.factions.fearful.power
+    optimist_power = state.factions.optimists.power
+
+    # civil_unrest: 恐惧派占优时推进
+    if "civil_unrest" in state.story_arcs:
+        arc = state.story_arcs["civil_unrest"]
+        if arc.active:
+            if fearful_power > optimist_power:
+                arc.progress = min(1.0, arc.progress + PROGRESS_INCREMENT)
+            else:
+                arc.progress = max(0.0, arc.progress - PROGRESS_DECREMENT)
+
+    # golden_age: 乐观派占优时推进
+    if "golden_age" in state.story_arcs:
+        arc = state.story_arcs["golden_age"]
+        if arc.active:
+            if optimist_power > fearful_power:
+                arc.progress = min(1.0, arc.progress + PROGRESS_INCREMENT)
+            else:
+                arc.progress = max(0.0, arc.progress - PROGRESS_DECREMENT)
+
+
+def check_and_trigger_story_event(state: WorldState) -> StoryEventResult:
+    """检查并触发剧情事件"""
+    result = StoryEventResult(triggered=False)
+
+    for arc_id, arc in state.story_arcs.items():
+        if not arc.active:
+            continue
+
+        if arc.progress >= arc.threshold:
+            # 触发剧情事件
+            event_data = STORY_EVENTS.get(arc_id)
+            if event_data:
+                result = StoryEventResult(
+                    triggered=True,
+                    arc_id=arc_id,
+                    event_name=event_data["name"],
+                    description=event_data["description"]
+                )
+                # 应用事件效果
+                _apply_story_event(state, arc_id, event_data)
+                # 重置进度，进入下一阶段
+                arc.progress = 0.0
+                arc.stage += 1
+                break  # 每 tick 只触发一个剧情事件
+
+    return result
+
+
+def _apply_story_event(state: WorldState, arc_id: str, event_data: dict) -> None:
+    """应用剧情事件效果"""
+    # 创建持续影响效果
+    effect = WorldEffect(
+        type="story_event",
+        name=event_data["name"],
+        intensity=2,
+        remaining_ticks=event_data["duration"],
+        mood_modifier=event_data["mood_effect"]
+    )
+    state.world_effects.append(effect)
+
+    # 立即影响世界情绪
+    state.town_mood = max(-10, min(10,
+        state.town_mood + event_data["mood_effect"]
+    ))

engine-python/tests/test_step.py

@@ -306,16 +306,78 @@ def test_factions_exists_in_world_state(client):
     factions = data["world_state"]["factions"]
     assert "optimists" in factions
     assert "fearful" in factions
-    assert "neutral" in factions
+    # 验证新结构：每个派系包含 power 和 members
+    assert "power" in factions["optimists"]
+    assert "members" in factions["optimists"]
+    assert "power" in factions["fearful"]
+    assert "members" in factions["fearful"]
 
 
 def test_factions_count_matches_agents(client):
-    """测试派系总数等于 agent 数量"""
+    """测试派系成员总数等于非中立 agent 数量"""
     resp = client.post("/step", json={"events": []})
     data = resp.json()
 
     factions = data["world_state"]["factions"]
-    total = factions["optimists"] + factions["fearful"] + factions["neutral"]
-    agent_count = len(data["world_state"]["agents"])
+    # 新结构：统计 members 列表长度
+    total_in_factions = (
+        len(factions["optimists"]["members"]) +
+        len(factions["fearful"]["members"])
+    )
+    # 中立 agent 不在任何派系中
+    # 验证 power 等于 members 数量
+    assert factions["optimists"]["power"] == len(factions["optimists"]["members"])
+    assert factions["fearful"]["power"] == len(factions["fearful"]["members"])
 
-    assert total == agent_count
+
+def test_agent_has_faction_field(client):
+    """测试 agent 包含 faction 字段"""
+    resp = client.post("/step", json={"events": []})
+    data = resp.json()
+
+    for agent_id, agent in data["world_state"]["agents"].items():
+        assert "faction" in agent
+        assert agent["faction"] in ["optimists", "fearful", "neutral"]
+
+
+def test_triggered_faction_event_in_response(client):
+    """测试响应包含 triggered_faction_event 字段"""
+    resp = client.post("/step", json={"events": []})
+    data = resp.json()
+
+    assert "triggered_faction_event" in data
+    assert "type" in data["triggered_faction_event"]
+    assert "source_faction" in data["triggered_faction_event"]
+
+
+def test_story_arcs_exists_in_world_state(client):
+    """测试 world_state 包含 story_arcs 字段"""
+    resp = client.post("/step", json={"events": []})
+    data = resp.json()
+
+    assert "story_arcs" in data["world_state"]
+    story_arcs = data["world_state"]["story_arcs"]
+    assert "civil_unrest" in story_arcs
+    assert "golden_age" in story_arcs
+
+
+def test_story_arc_has_required_fields(client):
+    """测试 story_arc 包含必要字段"""
+    resp = client.post("/step", json={"events": []})
+    data = resp.json()
+
+    for arc_id, arc in data["world_state"]["story_arcs"].items():
+        assert "progress" in arc
+        assert "threshold" in arc
+        assert "active" in arc
+        assert "stage" in arc
+
+
+def test_story_event_in_response(client):
+    """测试响应包含 story_event 字段"""
+    resp = client.post("/step", json={"events": []})
+    data = resp.json()
+
+    assert "story_event" in data
+    assert "triggered" in data["story_event"]
+    assert "arc_id" in data["story_event"]