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/main.py

@@ -3,6 +3,7 @@ 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")
 
@@ -22,6 +23,12 @@ def step(request: StepRequest) -> StepResponse:
     # 检查并触发阵营事件
     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)
 
@@ -32,7 +39,8 @@ def step(request: StepRequest) -> StepResponse:
         world_state=state,
         actions=actions,
         global_event=global_event_result,
-        triggered_faction_event=faction_event_result
+        triggered_faction_event=faction_event_result,
+        story_event=story_event_result
     )
 
 

engine-python/app/models.py

@@ -62,6 +62,23 @@ class FactionEventResult(BaseModel):
     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):
     emotion: Emotion = Emotion.CALM
     goal: str = ""
@@ -84,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):
@@ -120,3 +138,4 @@ class StepResponse(BaseModel):
     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, Factions, FactionData
+from .story_arcs import get_default_story_arcs
 
 STATE_FILE = Path(__file__).parent.parent / "state.json"
 
@@ -33,6 +34,7 @@ def get_default_state() -> WorldState:
             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

@@ -348,3 +348,36 @@ def test_triggered_faction_event_in_response(client):
     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"]