@@ -248,6 +262,36 @@ function ScreenJournal({ onNavigate, state, setState }) {
))}
+ {/* Threads & Callbacks (#120): scheduled quest-evolution echoes — a resolved
+ quest's pending "this thread will return" callback. Display-only. */}
+ {threads.length > 0 && (
+ <>
+
+
+
+ {w?.label || "Bond fracturing"}
+
+ {typeof w?.attitude_value === "number" ? `${w.attitude_value} / band ${band}` : band}
+
+
+
+ {w?.note || "This companion is approaching a breaking point."}
+
+ {w?.decision_active && (
+
+ A recorded choice has deepened the rift.
+
+ )}
+
+ );
+}
+
function NPCDetail({ n, onNavigate }) {
return (
@@ -284,6 +313,10 @@ function NPCDetail({ n, onNavigate }) {
)}
+ {/* Betrayal-warning band (#118): advisory telegraph when a companion's bond has
+ soured into the engine's danger band. Read-only — surfaced from the engine's
+ own `betrayal_warning`; never an action. */}
+ {n.betrayalWarning &&
Banter
@@ -574,4 +607,4 @@ const NPCS = [
},
];
-Object.assign(window, { ScreenRelations, FactionDetail, NPCDetail, RepBar, DispositionDot, FACTIONS, NPCS });
+Object.assign(window, { ScreenRelations, FactionDetail, NPCDetail, BetrayalWarning, RepBar, DispositionDot, FACTIONS, NPCS });
diff --git a/viewer/server.py b/viewer/server.py
index 05bb7c0d..b535c2f5 100644
--- a/viewer/server.py
+++ b/viewer/server.py
@@ -1829,6 +1829,11 @@ def _journal_quests(snapshot: dict) -> list[dict]:
loc = locs.get(location_id)
if isinstance(loc, dict):
region = _text(loc.get("region")) or _text(loc.get("name"), location_id)
+ # Rule-of-three evolution (#120): a quest carrying an `evolves_to` hook/seed will
+ # echo back as a scheduled callback once resolved. Surface the badge display-only.
+ evolves_to = _text(row.get("evolves_to"))
+ callback_raw = _num(row.get("callback_in_days"))
+ callback_in_days = int(callback_raw) if callback_raw is not None else 0
out.append({
"id": _text(qid),
"title": _text(row.get("title"), _text(qid, "Quest")),
@@ -1841,6 +1846,8 @@ def _journal_quests(snapshot: dict) -> list[dict]:
"objectives": objectives,
"entries": [],
"location_id": location_id,
+ "evolvesTo": evolves_to,
+ "callbackInDays": callback_in_days if evolves_to else 0,
})
return out
@@ -1874,6 +1881,62 @@ def _journal_hooks(snapshot: dict) -> list[dict]:
return out
+# The deterministic Consequence.note tag the engine writes when a resolved quest with an
+# `evolves_to` hook schedules its follow-on (server._evolution_note -> "evolves_from:
").
+# It's a stable contract — both authored by and guarded on by the engine — so the journal
+# can project scheduled evolutions read-only by matching this prefix.
+_EVOLUTION_NOTE_PREFIX = "evolves_from:"
+
+
+def _journal_evolutions(snapshot: dict) -> list[dict]:
+ """Project scheduled quest-evolution callbacks (#120) — the "this thread will return"
+ threads. Reads `consequences` whose `note` is the engine's deterministic
+ ``evolves_from:`` tag (server._maybe_schedule_quest_evolution), links each
+ back to its resolved quest, and marks it due / pending against the current in-world
+ day. Display-only: it never schedules, fires, or mutates anything. World-sim background
+ beats (a non-empty ``thread_id``) are NOT evolutions and are skipped."""
+ cons = snapshot.get("consequences")
+ quests = snapshot.get("quests") if isinstance(snapshot.get("quests"), dict) else {}
+ out: list[dict] = []
+ if not isinstance(cons, list):
+ return out
+ day = _num(snapshot.get("day"))
+ day = int(day) if day is not None else 0
+ for con in cons:
+ if not isinstance(con, dict):
+ continue
+ note = _text(con.get("note"))
+ if not note.startswith(_EVOLUTION_NOTE_PREFIX):
+ continue
+ if _text(con.get("thread_id")):
+ continue # a worldsim background beat, not a quest evolution
+ quest_id = note[len(_EVOLUTION_NOTE_PREFIX):].strip()
+ q = quests.get(quest_id) if isinstance(quests.get(quest_id), dict) else {}
+ trigger_day = _num(con.get("trigger_day"))
+ trigger_day = int(trigger_day) if trigger_day is not None else day
+ fired = bool(con.get("fired"))
+ due = (not fired) and trigger_day <= day
+ out.append({
+ "id": _text(con.get("id"), note),
+ "questId": quest_id,
+ "questTitle": _text(q.get("title"), quest_id or "a resolved thread"),
+ "evolvesTo": _text(q.get("evolves_to")),
+ "triggerDay": trigger_day,
+ "fired": fired,
+ "due": due,
+ "status": "fired" if fired else ("due" if due else "pending"),
+ "label": "Echo paid" if fired else ("Echo due" if due else "Echo pending"),
+ # Player-facing telegraph — never the engine's DM-only "weave a follow-on beat"
+ # prompt text. Display-only.
+ "note": (
+ f"A resolved thread waits to return"
+ + (f" on day {trigger_day}." if not due and not fired else
+ (" now." if due else "; it has already echoed back."))
+ ),
+ })
+ return out
+
+
def build_journal_surface(
snapshot: dict,
*,
@@ -1893,6 +1956,8 @@ def build_journal_surface(
"world": _text(snapshot.get("world_id"), "unknown"),
"dayLabel": _openworlds_day_label(snapshot),
"quests": quests,
+ # Scheduled quest-evolution callbacks (#120) — the "Threads & Callbacks" sub-list.
+ "threads": _journal_evolutions(snapshot),
"directorAdvisory": advisory,
"live": bool(live),
"is_live_view": bool(is_live_view),
@@ -2321,6 +2386,66 @@ def _attitude_disposition(ch: dict) -> str:
return "neutral"
+# Mirror of companion_arc.ATTITUDE_WARN_{LOW,HIGH} (the engine's danger band). A LIVE
+# (unfired) `attitude_below` agenda whose companion sits in [-40, -20] AND below the
+# agenda's breaking point is "approaching a fracture" — the engine emits an advisory
+# `betrayal_warning` from `evaluate()`; here we recompute the SAME band read-only from the
+# snapshot so the relations screen can telegraph it. Display-only: never mutates, never
+# fires anything, reads only the approval gauge + the (engine-set) decision_flag presence.
+_ATTITUDE_WARN_HIGH = -20 # upper edge: the bond has clearly soured
+_ATTITUDE_WARN_LOW = -40 # lower edge: below this it's already deep-red / near-snap
+
+
+def _betrayal_warning(ch: dict, snapshot: dict) -> dict | None:
+ """Advisory "approaching a breaking point" telegraph for a companion, recomputed
+ read-only from the snapshot (mirrors companion_arc._betrayal_warning).
+
+ Returns a small advisory dict ONLY when the companion carries a LIVE (unfired)
+ ``attitude_below`` agenda AND its ``attitude_value`` sits in the danger band
+ [_ATTITUDE_WARN_LOW, _ATTITUDE_WARN_HIGH] AND has crossed below the agenda's
+ breaking point (``value``). None otherwise. Reads the sealed agenda's TRIGGER/VALUE/
+ FIRED only to decide *whether* a warning is live — it NEVER surfaces the agenda's
+ private intent (`note`/`decision_flag` name), so no DM-only fiction leaks. The
+ ``decision_active`` flag is a plain bool: True when the agenda names a content flag
+ that is present+True in ``Campaign.flags`` (a recorded choice has already spiked the
+ odds), so the screen can foreshadow harder."""
+ arc = ch.get("arc")
+ if not isinstance(arc, dict):
+ return None
+ agenda = arc.get("agenda")
+ if not isinstance(agenda, dict):
+ return None
+ if agenda.get("trigger") != "attitude_below" or bool(agenda.get("fired")):
+ return None
+ threshold = _num(agenda.get("value"))
+ av = _num(ch.get("attitude_value"))
+ if threshold is None or av is None:
+ return None
+ threshold = int(threshold)
+ av = int(av)
+ # Only warn while in the band AND actually below the agenda's breaking point (an
+ # agenda whose threshold is even lower isn't live yet).
+ if not (_ATTITUDE_WARN_LOW <= av <= _ATTITUDE_WARN_HIGH):
+ return None
+ if av >= threshold:
+ return None
+ flags = snapshot.get("flags") if isinstance(snapshot.get("flags"), dict) else {}
+ decision_flag = _text(agenda.get("decision_flag"))
+ decision_active = bool(flags.get(decision_flag)) if decision_flag else False
+ return {
+ "attitude_value": av,
+ "threshold": threshold,
+ "band": [_ATTITUDE_WARN_LOW, _ATTITUDE_WARN_HIGH],
+ "decision_active": decision_active,
+ "label": "Bond fracturing",
+ "note": (
+ "This companion is approaching a breaking point — their bond has soured into "
+ "the danger band."
+ + (" A choice you made has deepened the rift." if decision_active else "")
+ ),
+ }
+
+
def _relations_npcs(snapshot: dict) -> list[dict]:
"""Project NPCs the party has actually met (kind=='npc') + companions, with attitude
and (for companions) the dossier's banter/relationship facts + arc state."""
@@ -2360,6 +2485,9 @@ def _relations_npcs(snapshot: dict) -> list[dict]:
"faction": "",
"disposition": _attitude_disposition(ch),
"approval": int(approval) if approval is not None else None,
+ # Advisory telegraph (#118): present (a dict) only for a companion whose live
+ # attitude_below agenda sits in the danger band; None otherwise. Display-only.
+ "betrayalWarning": _betrayal_warning(ch, snapshot) if is_companion else None,
"attitude": _text(ch.get("attitude")),
"body": _text(ch.get("backstory")) or _text(ch.get("personality")) or _text(ch.get("notes")) or "Little is known of them yet.",
"banter_tags": [str(t) for t in dossier.get("banter_tags", []) if str(t)] if isinstance(dossier.get("banter_tags"), list) else [],
diff --git a/viewer/tests/test_readmodel_surfaces.py b/viewer/tests/test_readmodel_surfaces.py
index 93f6e914..92700fc0 100644
--- a/viewer/tests/test_readmodel_surfaces.py
+++ b/viewer/tests/test_readmodel_surfaces.py
@@ -11,6 +11,7 @@
Campaign Director advisory exercises the engine.director detection path.
"""
+import copy
import http.client
import importlib.util
import json
@@ -195,6 +196,7 @@ def test_journal_surface_empty_without_snapshot(self):
self.assertEqual(status, 200)
self.assertEqual(surface["campaign_id"], "")
self.assertEqual(surface["quests"], [])
+ self.assertEqual(surface["threads"], [])
self.assertEqual(surface["directorAdvisory"]["debts"], [])
def test_director_advisory_uses_engine_detection_path(self):
@@ -212,6 +214,77 @@ def test_director_advisory_heuristic_matches_on_nonconformant_snapshot(self):
self.assertEqual(advisory["source"], "viewer.heuristic")
self.assertIn("hook_untracked", {d["kind"] for d in advisory["debts"]})
+ def test_journal_surface_quest_carries_evolution_badge_and_threads_callback(self):
+ # Quest-evolution / callback (#120): a resolved quest carrying `evolves_to` +
+ # `callback_in_days`, plus the engine's scheduled `evolves_from:` Consequence,
+ # surfaces as both a per-quest badge AND a "Threads & Callbacks" thread row.
+ snap = copy.deepcopy(_SNAPSHOT)
+ snap["quests"]["q2"]["evolves_to"] = "h-reckoning" # q2 is the completed quest
+ snap["quests"]["q2"]["callback_in_days"] = 3
+ # the engine schedules this on resolve (note == "evolves_from:"); day=12
+ snap["consequences"].append({
+ "id": "c_evo", "trigger_day": 15, "fired": False, "thread_id": "",
+ "text": "Bring back / evolve the resolved thread 'The Ferryman's Tab'.",
+ "note": "evolves_from:q2",
+ })
+ self._write("camp_marches", snap)
+ status, surface = self._get_json("/journal-surface?campaign=camp_marches")
+ self.assertEqual(status, 200)
+
+ # (a) per-quest badge fields
+ q2 = {q["id"]: q for q in surface["quests"]}["q2"]
+ self.assertEqual(q2["evolvesTo"], "h-reckoning")
+ self.assertEqual(q2["callbackInDays"], 3)
+ # a quest WITHOUT an evolves_to hook carries neither
+ q1 = {q["id"]: q for q in surface["quests"]}["q1"]
+ self.assertEqual(q1["evolvesTo"], "")
+ self.assertEqual(q1["callbackInDays"], 0)
+
+ # (b) the Threads & Callbacks sub-list projects the scheduled evolution
+ threads = {t["id"]: t for t in surface["threads"]}
+ self.assertIn("c_evo", threads)
+ thread = threads["c_evo"]
+ self.assertEqual(thread["questId"], "q2")
+ self.assertEqual(thread["questTitle"], "The Ferryman's Tab")
+ self.assertEqual(thread["evolvesTo"], "h-reckoning")
+ self.assertEqual(thread["triggerDay"], 15)
+ self.assertFalse(thread["fired"])
+ self.assertFalse(thread["due"]) # trigger_day 15 > current day 12
+ self.assertEqual(thread["status"], "pending")
+ self.assertTrue(thread["note"])
+ self.assert_no_private_keys(surface)
+
+ def test_journal_threads_marks_due_and_skips_worldsim_beats(self):
+ # A pending evolution whose trigger_day has arrived is `due`; a worldsim background
+ # beat (a non-empty thread_id, even with an evolves_from-looking note) is NOT an
+ # evolution and must be skipped.
+ snap = copy.deepcopy(_SNAPSHOT) # day = 12
+ snap["quests"]["q2"]["evolves_to"] = "h-reckoning"
+ snap["consequences"].append({
+ "id": "c_due", "trigger_day": 12, "fired": False, "thread_id": "",
+ "text": "It returns now.", "note": "evolves_from:q2",
+ })
+ snap["consequences"].append({
+ "id": "c_ws", "trigger_day": 12, "fired": False, "thread_id": "standing-war",
+ "text": "A world beat.", "note": "evolves_from:q2",
+ })
+ self._write("camp_marches", snap)
+ _status, surface = self._get_json("/journal-surface?campaign=camp_marches")
+ threads = {t["id"]: t for t in surface["threads"]}
+ self.assertIn("c_due", threads)
+ self.assertTrue(threads["c_due"]["due"])
+ self.assertEqual(threads["c_due"]["status"], "due")
+ self.assertNotIn("c_ws", threads) # worldsim beat excluded
+ self.assertEqual(len(surface["threads"]), 1)
+
+ def test_journal_threads_empty_when_no_quest_evolution_scheduled(self):
+ # The baseline conformant snapshot schedules no evolution -> no threads, and the
+ # completed quest carries an empty evolvesTo.
+ self._write("camp_marches", _SNAPSHOT)
+ _status, surface = self._get_json("/journal-surface?campaign=camp_marches")
+ self.assertEqual(surface["threads"], [])
+ self.assertEqual({q["id"]: q for q in surface["quests"]}["q2"]["evolvesTo"], "")
+
# ── character ───────────────────────────────────────────────────────────────
def test_character_surface_projects_full_party_sheets(self):
@@ -283,6 +356,72 @@ def test_relations_surface_projects_factions_npcs_and_arcs(self):
self.assertEqual(arcs["a1"]["companion"], "Mira of the Inkstain")
self.assertEqual([s["status"] for s in arcs["a1"]["stages"]], ["active", "locked"])
self.assert_no_private_keys(surface)
+ # baseline companion (no arc) carries no betrayal warning
+ self.assertIsNone(npcs["mira"]["betrayalWarning"])
+
+ def _snapshot_with_companion_agenda(self, *, attitude_value, threshold, fired=False,
+ trigger="attitude_below", decision_flag="", flags=None):
+ """A model-conformant copy of _SNAPSHOT where the companion `mira` carries a sealed
+ attitude_below agenda + a given attitude, so the betrayal-warning band (#118) can be
+ exercised. Round-trips through the engine Campaign model (strict), matching the
+ established conformant-snapshot pattern."""
+ snap = copy.deepcopy(_SNAPSHOT)
+ snap["characters"]["mira"]["attitude_value"] = attitude_value
+ agenda = {"trigger": trigger, "value": threshold, "fired": fired, "note": "sealed: turns on the party"}
+ if decision_flag:
+ agenda["decision_flag"] = decision_flag
+ snap["characters"]["mira"]["arc"] = {"arc_gates": [], "agenda": agenda}
+ if flags is not None:
+ snap["flags"] = flags
+ return snap
+
+ def test_relations_surface_betrayal_warning_present_when_companion_in_danger_band(self):
+ # mira at -28 with a live attitude_below agenda (threshold -10) sits in the engine's
+ # danger band [-40, -20] AND below the breaking point -> the advisory surfaces.
+ snap = self._snapshot_with_companion_agenda(attitude_value=-28, threshold=-10)
+ self._write("camp_marches", snap)
+ status, surface = self._get_json("/relations-surface?campaign=camp_marches")
+ self.assertEqual(status, 200)
+ npcs = {n["id"]: n for n in surface["npcs"]}
+ warning = npcs["mira"]["betrayalWarning"]
+ self.assertIsNotNone(warning)
+ self.assertEqual(warning["attitude_value"], -28)
+ self.assertEqual(warning["threshold"], -10)
+ self.assertEqual(warning["band"], [-40, -20])
+ self.assertFalse(warning["decision_active"])
+ self.assertTrue(warning["note"])
+ # the sealed agenda's private intent never leaks into the surface
+ self.assert_no_private_keys(surface)
+
+ def test_relations_surface_betrayal_warning_flags_a_recorded_decision(self):
+ # A decision_flag set+True in Campaign.flags marks the rift as choice-deepened.
+ snap = self._snapshot_with_companion_agenda(
+ attitude_value=-35, threshold=-10, decision_flag="took_bribe", flags={"took_bribe": True})
+ self._write("camp_marches", snap)
+ _status, surface = self._get_json("/relations-surface?campaign=camp_marches")
+ warning = {n["id"]: n for n in surface["npcs"]}["mira"]["betrayalWarning"]
+ self.assertIsNotNone(warning)
+ self.assertTrue(warning["decision_active"])
+
+ def test_relations_surface_omits_betrayal_warning_outside_the_band(self):
+ # Every off-band case must omit the warning (mirror companion_arc._betrayal_warning):
+ # - attitude above the band (not yet fracturing)
+ # - attitude in band but still at/above the agenda's breaking point
+ # - agenda already fired (the betrayal is the event, not a warning)
+ # - a non-attitude_below trigger sitting in the band
+ cases = [
+ ("above_band", dict(attitude_value=-10, threshold=-5)),
+ ("at_or_above_threshold", dict(attitude_value=-30, threshold=-35)),
+ ("already_fired", dict(attitude_value=-30, threshold=-10, fired=True)),
+ ("wrong_trigger", dict(attitude_value=-30, threshold=20, trigger="day_reached")),
+ ]
+ for name, kwargs in cases:
+ with self.subTest(case=name):
+ cid = f"camp_{name}" # a distinct campaign dir per case (no rewrite clash)
+ snap = self._snapshot_with_companion_agenda(**kwargs)
+ self._write(cid, snap)
+ _status, surface = self._get_json(f"/relations-surface?campaign={cid}")
+ self.assertIsNone({n["id"]: n for n in surface["npcs"]}["mira"]["betrayalWarning"])
# ── parley ───────────────────────────────────────────────────────────────