Add a world-state edit mode for overriding live RPG values

Players/authors can now directly correct the live values of stats the
schema already defines (health, trust, flags, milestones, etc.) without
waiting for the AI to emit a delta. New apply_override() sets values
absolutely rather than adding deltas, and — unlike the AI-facing
apply_delta() — bypasses cooldown/max_delta_per_turn and lets
milestones be un-set, since this is a deliberate correction rather
than a turn to police. Exposed via PUT /adventures/{id}/world-state
and an edit toggle in the World State drawer. Also fixes the
schema-editor stat-kind dropdown and free-text initial-value input
to size consistently with the numeric fields next to them.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

backend/app/routers/adventures.py

@@ -190,6 +190,28 @@ def get_world_state(
     }
 
 
+@router.put("/{adventure_id}/world-state")
+def override_world_state(
+    adventure_id: int,
+    overrides: dict = Body(...),
+    db: Session = Depends(get_db),
+    user: models.User = CurrentUser,
+):
+    """Directly edit the live RPG values (a manual correction, not a turn).
+    `overrides` maps paths (e.g. "player.hp", "npc.gwen.trust", "flags.x",
+    "milestones.y") to their new absolute value. Unknown paths/wrong types are
+    rejected individually; the rest still apply."""
+    adventure = get_adventure_or_404(adventure_id, db, user)
+    schema = adventure.scenario.stat_schema if adventure.scenario else None
+    if not worldstate.has_schema(schema):
+        raise HTTPException(400, "This adventure has no RPG world-state layer")
+    state = adventure.world_state if isinstance(adventure.world_state, dict) else {}
+    new_state, report = worldstate.apply_override(state, schema, overrides)
+    adventure.world_state = new_state
+    db.commit()
+    return {"state": new_state, "report": report}
+
+
 def snapshot_state(adventure: models.Adventure) -> dict:
     """Deep copy of the shared script_state, to staple onto an action so undo/
     retry can restore it. Independent of later hook mutations."""

backend/app/worldstate/__init__.py

@@ -4,6 +4,7 @@ module validates and clamps them against a scenario's stat_schema."""
 from .engine import (
     EMIT_RULE,
     apply_delta,
+    apply_override,
     band_label,
     extract_delta,
     has_schema,
@@ -17,6 +18,7 @@ from .engine import (
 __all__ = [
     "EMIT_RULE",
     "apply_delta",
+    "apply_override",
     "band_label",
     "extract_delta",
     "has_schema",

backend/app/worldstate/engine.py

@@ -255,6 +255,120 @@ def _apply_text_stat(container: dict, key: str, stat_def: dict, change,
     report["applied"].append({"path": path, "old": old, "new": new})
 
 
+def apply_override(world_state: dict, stat_schema: dict, overrides: dict) -> tuple[dict, dict]:
+    """Directly set live values — a manual author/admin edit, not an AI turn.
+
+    Unlike `apply_delta`: numeric stats are SET rather than added to, and
+    `cooldown`/`max_delta_per_turn`/the counter-can't-decrease rule are all
+    ignored (a deliberate correction, not an AI move to police). Milestones
+    can be toggled either way, not only marked reached. Values are still
+    validated against the schema (unknown path/type is rejected) and numeric
+    values still clamp to min/max."""
+    ws = copy.deepcopy(world_state) if isinstance(world_state, dict) else {}
+    if not ws:
+        ws = instantiate(stat_schema)
+    report: dict = {"applied": [], "rejected": []}
+
+    if not isinstance(overrides, dict):
+        return ws, report
+
+    milestones = stat_schema.get("milestones") or {}
+    flag_defs = stat_schema.get("flags") or {}
+    npcs = stat_schema.get("npcs") or {}
+
+    def set_stat(container: dict, key: str, stat_def: dict, value, path: str) -> None:
+        if stat_def.get("type") == "text":
+            if not isinstance(value, str):
+                report["rejected"].append({"path": path, "reason": "not a string"})
+                return
+            new = value.strip()
+            max_len = stat_def.get("max_length")
+            if isinstance(max_len, int) and max_len > 0 and len(new) > max_len:
+                new = new[:max_len]
+            old = container.get(key, stat_def.get("initial", ""))
+            container[key] = new
+            report["applied"].append({"path": path, "old": old, "new": new})
+            return
+
+        num = _coerce_number(value)
+        if num is None:
+            report["rejected"].append({"path": path, "reason": "not a number"})
+            return
+        old = container.get(key, stat_def.get("initial", 0))
+        lo, hi = stat_def.get("min"), stat_def.get("max")
+        if lo is not None and num < lo:
+            num = lo
+        if hi is not None and num > hi:
+            num = hi
+        if isinstance(old, int) and float(num).is_integer():
+            num = int(num)
+        container[key] = num
+        report["applied"].append({"path": path, "old": old, "new": num})
+
+    for raw_path, value in overrides.items():
+        path = str(raw_path)
+        parts = path.split(".")
+
+        if parts[0] == "flags" and len(parts) == 2:
+            fid = parts[1]
+            if fid not in flag_defs:
+                report["rejected"].append({"path": path, "reason": "unknown flag"})
+                continue
+            if not isinstance(value, bool):
+                report["rejected"].append({"path": path, "reason": "not a boolean"})
+                continue
+            flags = ws.setdefault("flags", {})
+            old = bool(flags.get(fid, False))
+            flags[fid] = value
+            report["applied"].append({"path": path, "old": old, "new": value})
+            continue
+
+        if parts[0] == "milestones" and len(parts) == 2:
+            mid = parts[1]
+            if mid not in milestones:
+                report["rejected"].append({"path": path, "reason": "unknown milestone"})
+                continue
+            if not isinstance(value, bool):
+                report["rejected"].append({"path": path, "reason": "not a boolean"})
+                continue
+            reached = ws.setdefault("milestones", {})
+            old = bool(reached.get(mid, {}).get("reached"))
+            if value:
+                reached[mid] = {"reached": True}
+            else:
+                reached.pop(mid, None)
+            report["applied"].append({"path": path, "old": old, "new": value})
+            continue
+
+        if parts[0] in STAT_SECTIONS and len(parts) == 2:
+            stat_def = (stat_schema.get(parts[0]) or {}).get(parts[1])
+            if not isinstance(stat_def, dict):
+                report["rejected"].append({"path": path, "reason": "unknown stat"})
+                continue
+            container = ws.setdefault(parts[0], {})
+            set_stat(container, parts[1], stat_def, value, path)
+            continue
+
+        if parts[0] == "npc" and len(parts) == 3:
+            ndef = npcs.get(parts[1])
+            if not isinstance(ndef, dict):
+                report["rejected"].append({"path": path, "reason": "unknown npc"})
+                continue
+            stat_defs = ndef.get("stats") or {}
+            stat_def = stat_defs.get(parts[2])
+            if not isinstance(stat_def, dict):
+                report["rejected"].append({"path": path, "reason": "unknown npc stat"})
+                continue
+            npc_state = ws.setdefault("npc", {})
+            container = npc_state.setdefault(parts[1], _initials(stat_defs))
+            set_stat(container, parts[2], stat_def, value, path)
+            continue
+
+        report["rejected"].append({"path": path, "reason": "unknown path"})
+
+    return ws, report
+
+
 def apply_delta(world_state: dict, stat_schema: dict, delta: dict,
                 action_index: int) -> tuple[dict, dict]:
     """Validate/clamp `delta` against `stat_schema` and apply to a copy of

backend/tests/test_worldstate.py

@@ -149,6 +149,57 @@ def test_flag_rejects_non_bool_and_unknown():
     assert ws["flags"]["has_key"] is False
 
 
+def test_override_sets_absolute_value_bypassing_cap():
+    # A manual override isn't policed by max_delta_per_turn like a turn is —
+    # it sets the value directly (still clamped to min/max).
+    ws, report = w.apply_override(fresh(), SCHEMA, {"player.hp": 10})
+    assert ws["player"]["hp"] == 10
+    assert report["applied"] == [{"path": "player.hp", "old": 100, "new": 10}]
+    ws, report = w.apply_override(ws, SCHEMA, {"player.hp": 999})
+    assert ws["player"]["hp"] == 100  # still clamps to max
+
+
+def test_override_bypasses_cooldown_and_counter_rule():
+    ws, _ = w.apply_override(fresh(), SCHEMA, {"npc.gwen.trust": 10})
+    # a second override immediately after would be blocked by cooldown under
+    # apply_delta, but override ignores cooldown entirely.
+    ws, report = w.apply_override(ws, SCHEMA, {"npc.gwen.trust": -5})
+    assert ws["npc"]["gwen"]["trust"] == -5
+    assert report["applied"]
+    # counters can be set down directly too (a correction, not a turn).
+    ws, report = w.apply_override(ws, SCHEMA, {"world.day": 1})
+    assert ws["world"]["day"] == 1
+    assert report["applied"]
+
+
+def test_override_text_stat_replaces():
+    ws, report = w.apply_override(fresh(), SCHEMA, {"player.outfit": "knight's plate"})
+    assert ws["player"]["outfit"] == "knight's plate"
+    assert report["applied"]
+
+
+def test_override_milestone_toggles_both_ways():
+    ws, _ = w.apply_override(fresh(), SCHEMA, {"milestones.rescue_gwen": True})
+    assert ws["milestones"]["rescue_gwen"]["reached"] is True
+    # unlike apply_delta, override can un-set a milestone.
+    ws, report = w.apply_override(ws, SCHEMA, {"milestones.rescue_gwen": False})
+    assert "rescue_gwen" not in ws["milestones"]
+    assert report["applied"]
+
+
+def test_override_rejects_unknown_and_bad_type():
+    ws, report = w.apply_override(fresh(), SCHEMA, {
+        "player.hp": "not a number",
+        "npc.bogus.trust": 5,
+        "flags.nope": True,
+    })
+    reasons = {r["path"]: r["reason"] for r in report["rejected"]}
+    assert reasons["player.hp"] == "not a number"
+    assert reasons["npc.bogus.trust"] == "unknown npc"
+    assert reasons["flags.nope"] == "unknown flag"
+    assert ws["player"]["hp"] == 100  # untouched
+
+
 def test_reference_includes_desc_and_bands_independently():
     guide = w.render_reference(SCHEMA)
     # hp has both a description and a band ladder.

backend/tests/test_worldstate_integration.py

@@ -163,6 +163,22 @@ def test_world_state_endpoint(client):
     assert body["state"]["player"]["hp"] == 70
 
 
+def test_override_world_state_endpoint(client):
+    r = client.put(f"/api/adventures/{client.adv_id}/world-state",
+                    json={"player.hp": 5, "flags.alarm": True, "npc.bogus.trust": 1})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["state"]["player"]["hp"] == 5
+    assert body["state"]["flags"]["alarm"] is True
+    assert body["report"]["rejected"][0]["reason"] == "unknown npc"
+    # persisted to the DB, not just the response.
+    assert _world(client.adv_id)["player"]["hp"] == 5
+
+    # bypasses max_delta_per_turn (30) — a direct correction, not a turn.
+    r = client.put(f"/api/adventures/{client.adv_id}/world-state", json={"player.hp": 100})
+    assert r.json()["state"]["player"]["hp"] == 100
+
+
 def test_undo_reverts_world_state(client):
     _play(client)
     assert _world(client.adv_id)["player"]["hp"] == 70