feat(worldgraph): add atlas graph metadata skeleton

servers/engine/content.py

@@ -33,6 +33,9 @@
     Quest,
     RegionControl,
     StrategicClock,
+    WorldGraph,
+    WorldGraphEdge,
+    WorldGraphNode,
     WorldState,
 )
 from store import safe_path_segment  # path-containment guard for world/adventure ids
@@ -934,6 +937,68 @@ def _seed_faction_arcs(c: Campaign, world: dict) -> None:
     _seed_faction_arcs_block(c, world.get("faction_arcs"), where="world faction_arcs block")
 
 
+def _seed_world_graph(c: Campaign, world: dict) -> None:
+    """Seed optional WorldGraph metadata from world.json.
+
+    Graph nodes/edges are player-facing metadata only. They are skipped unless
+    they refer to existing locations and edges already authorized by
+    ``Location.connections``.
+    """
+    raw = world.get("world_graph")
+    if raw is None:
+        return
+    if not isinstance(raw, dict):
+        print("[content] skipping malformed world_graph block (not an object)")
+        return
+
+    graph = WorldGraph(
+        seed=str(raw.get("seed", "")),
+        provenance=str(raw.get("provenance") or "authored"),
+    )
+    nodes = raw.get("nodes")
+    node_values = nodes.values() if isinstance(nodes, dict) else _as_list_lenient(raw, "nodes")
+    for entry in node_values:
+        if not isinstance(entry, dict):
+            print("[content] skipping world_graph node (not an object)")
+            continue
+        try:
+            node = WorldGraphNode.model_validate(entry)
+        except (ValidationError, ValueError, TypeError):
+            print("[content] skipping malformed world_graph node")
+            continue
+        if node.location_id not in c.locations:
+            print(f"[content] skipping world_graph node {node.location_id!r}: unknown location")
+            continue
+        graph.nodes[node.location_id] = node
+
+    for entry in _as_list_lenient(raw, "edges"):
+        if not isinstance(entry, dict):
+            print("[content] skipping world_graph edge (not an object)")
+            continue
+        try:
+            edge = WorldGraphEdge.model_validate(entry)
+        except (ValidationError, ValueError, TypeError):
+            print("[content] skipping malformed world_graph edge")
+            continue
+        src = c.locations.get(edge.from_id)
+        dst = c.locations.get(edge.to_id)
+        if src is None or dst is None:
+            print(
+                "[content] skipping world_graph edge "
+                f"{edge.from_id!r}->{edge.to_id!r}: unknown location"
+            )
+            continue
+        if edge.to_id not in src.connections and edge.from_id not in dst.connections:
+            print(
+                "[content] skipping world_graph edge "
+                f"{edge.from_id!r}->{edge.to_id!r}: not a canonical connection"
+            )
+            continue
+        graph.edges.append(edge)
+
+    c.world_graph = graph
+
+
 def _seed_campaign_backlog(c: Campaign, world: dict) -> None:
     """Seed the PROACTIVE living-world backlog (P0) — the world's own off-screen to-do, so the
     campaign advances when in-fiction time passes (P1 tick_backlog) instead of only reacting to
@@ -1252,6 +1317,7 @@ def seed_world(world: dict, start_at: str = "", ending: str = "") -> Campaign:
             faction.id = fac["id"]
         c.factions[faction.id] = faction
 
+    _seed_world_graph(c, world)
     _seed_strategic_state(c, world)
 
     # Faction-growth questlines (Quest & Arc engine, faction arcs / #127). Runs AFTER factions are

servers/engine/models.py

@@ -786,6 +786,50 @@ class Location(_StrictModel):
     travel_times: dict[str, int] = Field(default_factory=dict)  # connected location id -> walk minutes
 
 
+class WorldGraphNode(_StrictModel):
+    """Player-facing spatial metadata for an existing Location.
+
+    ``Campaign.locations`` remains canonical. This metadata may enrich Atlas and
+    travel displays, but it cannot create or authorize locations by itself.
+    """
+
+    location_id: str
+    x: Optional[float] = None
+    y: Optional[float] = None
+    biome: str = ""
+    terrain: str = ""
+    danger: int = Field(0, ge=0, le=10)
+    discovered: bool = True
+    atlas_layer: Literal["region", "settlement", "site", "dungeon", "route"] = "site"
+    tags: list[str] = Field(default_factory=list)
+
+
+class WorldGraphEdge(_StrictModel):
+    """Route metadata for an already-authorized Location connection."""
+
+    from_id: str
+    to_id: str
+    route_kind: Literal["street", "road", "trail", "sea", "river", "passage", "portal"] = "road"
+    minutes: Optional[int] = Field(None, ge=1)
+    distance: Optional[float] = Field(None, ge=0)
+    difficulty: Literal["easy", "normal", "hard", "hazardous"] = "normal"
+    danger: int = Field(0, ge=0, le=10)
+    tags: list[str] = Field(default_factory=list)
+
+
+class WorldGraph(_StrictModel):
+    """Additive strategic-map metadata owned by the engine.
+
+    This graph is deliberately subordinate to ``Location.connections``: edges
+    enrich existing travel links, they never authorize movement on their own.
+    """
+
+    nodes: dict[str, WorldGraphNode] = Field(default_factory=dict)
+    edges: list[WorldGraphEdge] = Field(default_factory=list)
+    seed: str = ""
+    provenance: str = "authored"
+
+
 class Faction(_StrictModel):
     id: str = Field(default_factory=lambda: _new_id("fac"))
     name: str
@@ -1353,6 +1397,7 @@ class Campaign(_StrictModel):
     # strict sibling of consequences/worldsim threads (never consumed by consequences.due).
     campaign_backlog: CampaignBacklog = Field(default_factory=CampaignBacklog)
     strategic_state: StrategicState = Field(default_factory=StrategicState)
+    world_graph: WorldGraph = Field(default_factory=WorldGraph)
     # Persistent camp-beat memory (#69). Read by camp_scene/scheduler, written only by an
     # explicit record path so prompt generation never advances campaign state by accident.
     camp_beats: CampBeatState = Field(default_factory=CampBeatState)

servers/engine/tests/test_content.py

@@ -214,6 +214,67 @@ def test_seed_world_seeds_strategic_state_additively(capsys):
     assert "skipping strategic project" in out
 
 
+def test_seed_world_seeds_world_graph_metadata_without_authorizing_travel(capsys):
+    world = {
+        "id": "graph-test",
+        "name": "Graph Test",
+        "premise": "A compact graph fixture.",
+        "era": "now",
+        "regions": [
+            {"id": "loc-harbor", "name": "Harbor", "description": "docks", "connections": ["loc-hill"]},
+            {"id": "loc-hill", "name": "Hill", "description": "watchpost", "connections": ["loc-harbor"]},
+            {"id": "loc-sealed", "name": "Sealed", "description": "locked", "connections": []},
+        ],
+        "factions": [],
+        "npc_roster": [],
+        "history": [],
+        "standing_threads": [],
+        "starting_options": [{"location_id": "loc-harbor", "framing": "Start at the harbor."}],
+        "world_graph": {
+            "seed": "graph-fixture",
+            "provenance": "authored-test",
+            "nodes": [
+                {
+                    "location_id": "loc-harbor",
+                    "biome": "coast",
+                    "terrain": "docks",
+                    "danger": 2,
+                    "atlas_layer": "settlement",
+                    "tags": ["port"],
+                },
+                {"location_id": "loc-missing", "biome": "void"},
+            ],
+            "edges": [
+                {
+                    "from_id": "loc-harbor",
+                    "to_id": "loc-hill",
+                    "route_kind": "road",
+                    "minutes": 45,
+                    "difficulty": "easy",
+                    "danger": 1,
+                    "tags": ["patrolled"],
+                },
+                {"from_id": "loc-harbor", "to_id": "loc-sealed", "route_kind": "trail"},
+                {"from_id": "loc-harbor", "to_id": "loc-missing", "route_kind": "road"},
+            ],
+        },
+    }
+
+    c = content.seed_world(world)
+
+    assert set(c.world_graph.nodes) == {"loc-harbor"}
+    assert c.world_graph.nodes["loc-harbor"].biome == "coast"
+    assert len(c.world_graph.edges) == 1
+    assert c.world_graph.edges[0].from_id == "loc-harbor"
+    assert c.world_graph.edges[0].to_id == "loc-hill"
+    assert "loc-sealed" not in c.locations["loc-harbor"].connections
+
+    out = capsys.readouterr().out
+    assert "skipping world_graph node" in out
+    assert "not a canonical connection" in out
+    assert "unknown location" in out
+
+
 def test_seed_world_rejects_unknown_start(tmp_path, monkeypatch):
     w = content.load_world_data("sundered-reach")
     with pytest.raises(ValueError, match="not a region"):

viewer/server.py

@@ -1720,20 +1720,26 @@ def _atlas_tags(row: dict) -> list[str]:
     return out
 
 
+def _safe_str_list(raw: object) -> list[str]:
+    return [_text(item) for item in raw if _text(item)] if isinstance(raw, list) else []
+
+
 def _atlas_known_locations(snapshot: dict) -> list[dict]:
     locs = _atlas_locations(snapshot)
     visible_ids = _atlas_visible_location_ids(snapshot)
     current_id = _text(snapshot.get("current_location_id"))
+    graph = _atlas_world_graph(snapshot)
     out: list[dict] = []
     for idx, loc_id in enumerate(visible_ids):
         row = locs.get(loc_id)
         if not isinstance(row, dict):
             continue
+        node_meta = _atlas_node_meta(graph, loc_id)
         x, y = _atlas_hex_position(row, idx)
         name = _text(row.get("name"), loc_id)
         connections = row.get("connections")
         connections = connections if isinstance(connections, list) else []
-        out.append({
+        item = {
             "id": loc_id,
             "name": name,
             "description": _text(row.get("description")),
@@ -1744,13 +1750,66 @@ def _atlas_known_locations(snapshot: dict) -> list[dict]:
             "y": y,
             "tags": _atlas_tags(row),
             "connections": [_text(c) for c in connections if _text(c) in visible_ids],
-        })
+        }
+        if node_meta:
+            item.update({
+                "biome": _text(node_meta.get("biome")),
+                "terrain": _text(node_meta.get("terrain")),
+                "danger": node_meta.get("danger") if isinstance(node_meta.get("danger"), int) else 0,
+                "atlas_layer": _text(node_meta.get("atlas_layer"), "site"),
+                "graph_tags": _safe_str_list(node_meta.get("tags")),
+            })
+        out.append(item)
     out.sort(key=lambda loc: (not loc["current"], loc["name"]))
     return out
 
 
-def _atlas_edges(locations: list[dict]) -> list[dict]:
+def _atlas_world_graph(snapshot: dict) -> dict:
+    graph = snapshot.get("world_graph")
+    return graph if isinstance(graph, dict) else {}
+
+
+def _atlas_node_meta(graph: dict, loc_id: str) -> dict:
+    nodes = graph.get("nodes")
+    row = nodes.get(loc_id) if isinstance(nodes, dict) else None
+    return row if isinstance(row, dict) else {}
+
+
+def _atlas_edge_meta(graph: dict, src: str, dst: str) -> dict:
+    edges = graph.get("edges")
+    if not isinstance(edges, list):
+        return {}
+    for row in edges:
+        if not isinstance(row, dict):
+            continue
+        a = _text(row.get("from_id"))
+        b = _text(row.get("to_id"))
+        if {a, b} == {src, dst}:
+            return row
+    return {}
+
+
+def _atlas_edge_payload(meta: dict) -> dict:
+    if not meta:
+        return {}
+    out: dict = {}
+    for key in ("route_kind", "difficulty"):
+        value = _text(meta.get(key))
+        if value:
+            out[key] = value
+    for key in ("minutes", "danger"):
+        value = meta.get(key)
+        if isinstance(value, int) and not isinstance(value, bool):
+            out[key] = value
+    tags = _safe_str_list(meta.get("tags"))
+    if tags:
+        out["tags"] = tags
+    return out
+
+
+def _atlas_edges(locations: list[dict], snapshot: dict | None = None) -> list[dict]:
     known = {loc["id"] for loc in locations}
+    graph = _atlas_world_graph(snapshot or {})
     seen: set[tuple[str, str]] = set()
     out: list[dict] = []
     for loc in locations:
@@ -1762,7 +1821,9 @@ def _atlas_edges(locations: list[dict]) -> list[dict]:
             if key in seen or src == dst:
                 continue
             seen.add(key)
-            out.append({"from": key[0], "to": key[1]})
+            item = {"from": key[0], "to": key[1]}
+            item.update(_atlas_edge_payload(_atlas_edge_meta(graph, src, dst)))
+            out.append(item)
     return out
 
 
@@ -1786,6 +1847,7 @@ def _atlas_travel_options(snapshot: dict, locations: list[dict], *, live: bool,
         return []
     known = {loc["id"]: loc for loc in locations}
     travel_times = current.get("travel_times") if isinstance(current.get("travel_times"), dict) else {}
+    graph = _atlas_world_graph(snapshot)
     disabled = _atlas_move_reason(snapshot, live=live, is_live_view=is_live_view)
     out: list[dict] = []
     for dst in current.get("connections", []) if isinstance(current.get("connections"), list) else []:
@@ -1795,13 +1857,19 @@ def _atlas_travel_options(snapshot: dict, locations: list[dict], *, live: bool,
             continue
         minutes = travel_times.get(dst_id)
         minutes = minutes if isinstance(minutes, int) and not isinstance(minutes, bool) else None
+        edge_meta = _atlas_edge_meta(graph, current_id, dst_id)
+        edge_payload = _atlas_edge_payload(edge_meta)
+        minutes = minutes if minutes is not None else edge_payload.get("minutes")
         item = {
             "to": dst_id,
             "name": target["name"],
             "minutes": minutes,
             "available": disabled is None,
             "disabled_reason": disabled,
         }
+        for key in ("route_kind", "difficulty", "danger", "tags"):
+            if key in edge_payload:
+                item[key] = edge_payload[key]
         if disabled is None:
             item["move"] = {"kind": "do", "text": f"Travel to {target['name']}"}
         out.append(item)
@@ -1933,7 +2001,7 @@ def build_atlas_surface(
         "dayLabel": _openworlds_day_label(snapshot),
         "current_location": current or {"id": "", "name": "Unknown location", "tags": []},
         "known_locations": locations,
-        "edges": _atlas_edges(locations),
+        "edges": _atlas_edges(locations, snapshot),
         "travel_options": travel_options,
         "quest_markers": _atlas_quest_markers(snapshot, visible_ids),
         "strategic_clocks": clocks,