diff --git a/bridge/bridge_loop.py b/bridge/bridge_loop.py
index c2eb21a..2812b8a 100644
--- a/bridge/bridge_loop.py
+++ b/bridge/bridge_loop.py
@@ -51,14 +51,19 @@ def __init__(
         self.turn_interval_sec = turn_interval_sec
         self.callbacks: Dict[str, Callable] = {}
         self.running = False
-        # Wire the listener's on_change to the audit log if both are present.
-        if self.hook_listener is not None and self.audit_log is not None:
+        # Per-turn event buffer (PRD §GM-5: visible plan/goal bonuses).
+        # Maps city_id → list of {"kind": "plan"|"goal"|"redirect", "session_id": ..., "session_name": ...}
+        self._pending_events_by_city: Dict[str, List[Dict[str, Any]]] = {}
+        # Compose any existing on_change handlers with our internal hooks.
+        if self.hook_listener is not None:
             existing_callback = self.hook_listener.on_change
-            def _audit_then_existing(change):
-                self.audit_log.log_state_change(change)
+            def _on_change_chain(change):
+                self._record_event_for_city(change)
+                if self.audit_log is not None:
+                    self.audit_log.log_state_change(change)
                 if existing_callback is not None:
                     existing_callback(change)
-            self.hook_listener.on_change = _audit_then_existing
+            self.hook_listener.on_change = _on_change_chain
 
     # ──────────────────────────────────────────────────────────────────
     # Lifecycle
@@ -115,6 +120,14 @@ async def process_turn(self) -> Optional[Dict[str, Any]]:
 
         self.state.end_turn()
 
+        # Flush per-city events to the save-exchange-bound state-side
+        # buffer so SaveExchange.write_payload picks them up; clear our
+        # internal buffer for the next turn.
+        per_city_events = self._flush_per_city_events()
+        # Stash on the SessionState so SaveExchange can read it without
+        # changing its signature.
+        self.state._pending_per_city_events = per_city_events  # type: ignore[attr-defined]
+
         if self.save_exchange is not None:
             self.save_exchange.write_payload(self.state)
 
@@ -124,12 +137,58 @@ async def process_turn(self) -> Optional[Dict[str, Any]]:
         self._invoke_callbacks("turn_complete", rewards)
         if per_city:
             self._invoke_callbacks("per_city_rewards", per_city)
+        if per_city_events:
+            self._invoke_callbacks("per_city_events", per_city_events)
         return rewards
 
     # ──────────────────────────────────────────────────────────────────
     # External API
     # ──────────────────────────────────────────────────────────────────
 
+    # ──────────────────────────────────────────────────────────────────
+    # Per-turn event buffer (PRD §GM-5)
+    # ──────────────────────────────────────────────────────────────────
+
+    def _record_event_for_city(self, change: Any) -> None:
+        """Buffer plan/goal/redirect events per city for surfacing on next turn."""
+        # Only record events that the player will see as a discrete bonus.
+        kind_map = {
+            "plan_added": "plan",
+            "goal_completed": "goal",
+            "redirect": "redirect",
+        }
+        event_kind = kind_map.get(getattr(change, "event_type", None))
+        if event_kind is None:
+            return
+        # goal_completed emits 2 records (goals + momentum); only buffer the
+        # field=goals_completed one so we don't double-count.
+        if event_kind == "goal" and getattr(change, "field", None) != "goals_completed":
+            return
+        # redirect emits 1 record (field=momentum); buffer that.
+        if event_kind == "redirect" and getattr(change, "field", None) != "momentum":
+            return
+
+        session_id = getattr(change, "session_id", None)
+        if not session_id:
+            return
+        # Find the session to get its city_id and name.
+        for s in self.state.sessions:
+            if s.session_id == session_id:
+                city = s.city_id or ""
+                bucket = self._pending_events_by_city.setdefault(city, [])
+                bucket.append({
+                    "kind": event_kind,
+                    "session_id": session_id,
+                    "session_name": s.name,
+                })
+                return
+
+    def _flush_per_city_events(self) -> Dict[str, List[Dict[str, Any]]]:
+        """Snapshot the current event buffer and clear it for the next turn."""
+        events = self._pending_events_by_city
+        self._pending_events_by_city = {}
+        return events
+
     def feed_event(self, event: Dict[str, Any]) -> List[Any]:
         """Proxy a hook event into the listener (TUI / test convenience).
 
diff --git a/bridge/save_exchange.py b/bridge/save_exchange.py
index d134e5f..135cc92 100644
--- a/bridge/save_exchange.py
+++ b/bridge/save_exchange.py
@@ -76,6 +76,10 @@ def write_payload(self, state: SessionState) -> Path:
         # Surface ALL sessions (not just active) so quiet cities are visible too.
         payload["per_city_status"] = engine.calculate_per_city_status(state.sessions)
 
+        # Per-city event log for this turn (PRD §GM-5 — visible plan/goal
+        # bonus deltas the mod can surface as transient announcements).
+        payload["per_city_events"] = getattr(state, "_pending_per_city_events", {}) or {}
+
         # Embed the per-session breakdown the PRD's binding contract requires.
         for session_dict, session in zip(payload.get("sessions", []), state.sessions):
             breakdown = strategy.calculate_session_breakdown(session)
diff --git a/bridge/tests/test_per_city_events.py b/bridge/tests/test_per_city_events.py
new file mode 100644
index 0000000..962c1af
--- /dev/null
+++ b/bridge/tests/test_per_city_events.py
@@ -0,0 +1,155 @@
+"""Tests for per-turn per-city event surfacing (PRD §GM-5).
+
+The bridge buffers plan/goal/redirect events as they fire and
+flushes them on turn boundary, embedding them in the save exchange so
+the Unciv mod can surface a transient announcement per tile (e.g.
+'+15 from plan' or '+40 from goal').
+"""
+
+import json
+
+import pytest
+
+from bridge.bridge_loop import BridgeLoop
+from bridge.hook_listener import HookListener
+from bridge.save_exchange import SaveExchange
+from bridge.scoring import ScoringEngine, SimpleScoringStrategy
+from bridge.session_state import SessionState, SessionStatus
+
+
+def _build_loop(tmp_path):
+    state = SessionState(kingdom_name="Camelot")
+    s = state.add_session("Royal Court")
+    s.session_id = "sess-1"
+    s.city_id = "camelot"
+    s.status = SessionStatus.ACTIVE
+    listener = HookListener(state=state)
+    save = SaveExchange(tmp_path / "saves")
+    return state, BridgeLoop(
+        state=state,
+        scoring_engine=ScoringEngine(SimpleScoringStrategy()),
+        hook_listener=listener,
+        save_exchange=save,
+    )
+
+
+# ────────────────────────────────────────────────────────────────────────
+# Buffering between turns
+# ────────────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_plan_event_buffered_then_flushed_on_turn(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    loop.feed_event({"event_type": "SubagentStart",
+                     "payload": {"session_id": "sess-1", "type": "Plan",
+                                  "plan_id": "p1"}})
+    # Before turn: pending buffer holds 1 event for camelot
+    assert loop._pending_events_by_city == {"camelot": [
+        {"kind": "plan", "session_id": "sess-1", "session_name": "Royal Court"}
+    ]}
+    await loop.process_turn()
+    # After turn: buffer cleared
+    assert loop._pending_events_by_city == {}
+
+
+@pytest.mark.asyncio
+async def test_goal_event_records_once_per_completion(tmp_path):
+    """goal_completed emits 2 StateChange records (goals + momentum);
+    only the goals_completed one should be buffered to avoid double-count."""
+    state, loop = _build_loop(tmp_path)
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    bucket = loop._pending_events_by_city["camelot"]
+    assert len(bucket) == 1
+    assert bucket[0]["kind"] == "goal"
+
+
+@pytest.mark.asyncio
+async def test_redirect_event_recorded(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    state.sessions[0].momentum = 5
+    loop.feed_event({"event_type": "PermissionDenied",
+                     "payload": {"session_id": "sess-1"}})
+    bucket = loop._pending_events_by_city.get("camelot", [])
+    redirect_events = [e for e in bucket if e["kind"] == "redirect"]
+    assert len(redirect_events) == 1
+
+
+@pytest.mark.asyncio
+async def test_per_city_events_emitted_in_save_exchange(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    loop.feed_event({"event_type": "SubagentStart",
+                     "payload": {"session_id": "sess-1", "type": "Plan",
+                                  "plan_id": "p1"}})
+    await loop.process_turn()
+    save_path = loop.save_exchange.path
+    data = json.loads(save_path.read_text(encoding="utf-8"))
+    assert "per_city_events" in data
+    events = data["per_city_events"]["camelot"]
+    kinds = sorted(e["kind"] for e in events)
+    assert kinds == ["goal", "plan"]
+
+
+@pytest.mark.asyncio
+async def test_empty_event_buffer_emits_empty_dict(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    await loop.process_turn()
+    data = json.loads(loop.save_exchange.path.read_text(encoding="utf-8"))
+    assert data["per_city_events"] == {}
+
+
+@pytest.mark.asyncio
+async def test_events_buffer_resets_between_turns(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    await loop.process_turn()
+    # Turn 2: no new events.
+    await loop.process_turn()
+    data = json.loads(loop.save_exchange.path.read_text(encoding="utf-8"))
+    # Last write reflects only the second turn — empty events.
+    assert data["per_city_events"] == {}
+
+
+@pytest.mark.asyncio
+async def test_per_city_events_callback_fires(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    captured = []
+    loop.register_callback("per_city_events", lambda evs: captured.append(evs))
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    await loop.process_turn()
+    assert len(captured) == 1
+    assert "camelot" in captured[0]
+
+
+@pytest.mark.asyncio
+async def test_event_record_includes_session_name(tmp_path):
+    state, loop = _build_loop(tmp_path)
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    bucket = loop._pending_events_by_city["camelot"]
+    assert bucket[0]["session_name"] == "Royal Court"
+    assert bucket[0]["session_id"] == "sess-1"
+
+
+@pytest.mark.asyncio
+async def test_events_segregate_by_city_id(tmp_path):
+    """Two cities each get their own event list."""
+    state, loop = _build_loop(tmp_path)
+    s2 = state.add_session("North Camp")
+    s2.session_id = "sess-2"
+    s2.city_id = "north_fort"
+    s2.status = SessionStatus.ACTIVE
+
+    loop.feed_event({"event_type": "TaskCompleted",
+                     "payload": {"session_id": "sess-1"}})
+    loop.feed_event({"event_type": "SubagentStart",
+                     "payload": {"session_id": "sess-2", "type": "Plan",
+                                  "plan_id": "p2"}})
+    await loop.process_turn()
+    data = json.loads(loop.save_exchange.path.read_text(encoding="utf-8"))
+    assert data["per_city_events"]["camelot"][0]["kind"] == "goal"
+    assert data["per_city_events"]["north_fort"][0]["kind"] == "plan"
diff --git a/mod/ClaudeKingdoms/mod.lua b/mod/ClaudeKingdoms/mod.lua
index f24123c..9f37621 100644
--- a/mod/ClaudeKingdoms/mod.lua
+++ b/mod/ClaudeKingdoms/mod.lua
@@ -83,6 +83,11 @@ local function readKingdomSave(currentTurn)
     if parsed.per_city_status and type(parsed.per_city_status) == "table" then
         rewards.__per_city_status__ = parsed.per_city_status
     end
+    -- Per-city events (PRD §GM-5) — list of {kind, session_id, session_name}
+    -- per city for this turn, so the mod can announce "+15 from plan" etc.
+    if parsed.per_city_events and type(parsed.per_city_events) == "table" then
+        rewards.__per_city_events__ = parsed.per_city_events
+    end
 
     -- Cache the rewards for this turn
     rewardCache = rewards
@@ -106,6 +111,57 @@ local function getKingdomStatus()
     return {}
 end
 
+--[[
+    Returns the per-city event log (city_id → array of event records)
+    for the current turn (PRD §GM-5). Each record has fields
+    { kind = "plan"|"goal"|"redirect", session_id, session_name }.
+    Mod consumers iterate this to render announcement messages per tile.
+]]
+local function getKingdomEvents()
+    if rewardCache and rewardCache.__per_city_events__ then
+        return rewardCache.__per_city_events__
+    end
+    return {}
+end
+
+--[[
+    Format an event record as a player-facing announcement string.
+    PRD §GM-5: 'Plan and goal events trigger one-time visible bonuses
+    on turn resolution.' Returns a short medieval-flavor string.
+]]
+local function announcementFor(event)
+    if not event or not event.kind then return "" end
+    local who = event.session_name or "the realm"
+    if event.kind == "plan" then
+        return string.format("A new charter is drafted for '%s' (+15 production)", who)
+    elseif event.kind == "goal" then
+        return string.format("'%s' achieves a goal (+40 gold, +20 science)", who)
+    elseif event.kind == "redirect" then
+        return string.format("'%s' is redirected — momentum lost, banked tribute preserved", who)
+    end
+    return string.format("'%s' event: %s", who, tostring(event.kind))
+end
+
+--[[
+    Apply per-tile event announcements (PRD §GM-5). Tries city.announce →
+    addNotification → falls back to print so the data flow is visible
+    even without an Unciv UI surface. pcall-wrapped.
+]]
+local function applyCityEventAnnouncements(city, eventList)
+    if not city or not eventList then return end
+    local cityName = "unknown"
+    if city.getName then cityName = city.getName(city) end
+    for _, event in ipairs(eventList) do
+        local msg = announcementFor(event)
+        print(string.format("[%s] %s", cityName, msg))
+        if city.announce then
+            pcall(function() city.announce(city, msg) end)
+        elseif city.addNotification then
+            pcall(function() city.addNotification(city, msg) end)
+        end
+    end
+end
+
 --[[
     Maps the bridge's session status string to a short tile glyph the
     player can read at a glance without opening the TUI (PRD §GM-8).