From 2e3617865b8ccc390dd7bb0f70ad1d2851f409a1 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 6 Nov 2026 14:18:47 -0300 Subject: [PATCH] fix(loops,ui,data): resolve adversarial loop-review audit findings (DL-1, P-3, P-4, P-8, P-9, P-11) - Allow end_conversation to resume 100-turn conversations via allow_at_limit=True, skip maxed conversations in find_client_conversation, and strip/sync orphan trailing player turns on failed LLM calls (F1, F11) - Invert roll-under advantage/disadvantage keep direction (2d20kl1/2d20kh1) in _format_dice_spec and Resolver._execute_roll (F2) - Avoid doubling dice-pool size when calc_modifier returns stat_value (vampire.json) and parse count-successes pool thresholds without concatenating multi-band PbtA strings (F3) - Defer SceneLoop._run_combat while awaiting a player roll on propose_roll or forced_narrative_pushback, populate pending_roll on pushback, and pass only_resource_deltas=True with case-insensitive HP matching on combat persistence (F4, F8, F12) - Keep WebSocket active on llm_provider_error, include restored dice_request, guard JSON/int parsing, and handle scene_ended phase (F5, F9) - Exclude disable/enabled from st_extensions overwrite on SillyTavern lorebook export and expose include_inactive on GET /lorebook/entries (F6) - Guard PostgresClient.connect with a per-loop asyncio.Lock and close the active loop's pool in close() (F7) --- .../src/monitor_agents/loops/scene_loop.py | 20 ++- .../loops/scene_orchestrator_support.py | 4 +- .../agents/src/monitor_agents/resolver.py | 162 ++++++++++-------- .../services/persistence_service.py | 56 +++--- .../tests/test_resolver_dice_authority.py | 89 ++++++++++ .../tests/test_scene_state_carryover.py | 25 +++ .../src/monitor_data/db/postgres.py | 68 +++++--- .../interop/sillytavern_lorebook.py | 6 +- .../data-layer/tests/test_db/test_postgres.py | 45 +++++ .../tests/test_sillytavern_lorebook.py | 23 +++ .../routers/character_conversation.py | 61 ++++++- .../ui/backend/src/monitor_ui/routers/chat.py | 55 +++++- .../src/monitor_ui/routers/lorebook.py | 10 +- .../tests/test_character_conversation.py | 83 +++++++++ 14 files changed, 560 insertions(+), 147 deletions(-) diff --git a/packages/agents/src/monitor_agents/loops/scene_loop.py b/packages/agents/src/monitor_agents/loops/scene_loop.py index 9da531c7..1dd2a9fc 100644 --- a/packages/agents/src/monitor_agents/loops/scene_loop.py +++ b/packages/agents/src/monitor_agents/loops/scene_loop.py @@ -568,7 +568,7 @@ async def resolve_action(state: SceneState) -> dict[str, Any]: # When the roll is resolved (not pending), clear it. new_pending_roll = state.pending_roll res_type = resolution.get("resolution_type") - if res_type == "propose_roll": + if res_type in ("propose_roll", "forced_narrative_pushback"): new_pending_roll = { "stat": resolution.get("stat"), "dc": resolution.get("difficulty_class"), @@ -576,7 +576,7 @@ async def resolve_action(state: SceneState) -> dict[str, Any]: "attribute_value": resolution.get("attribute_value"), "roll_mode": resolution.get("roll_mode") or state.roll_mode, "action": state.user_input, - "resolution_type": "propose_roll", + "resolution_type": res_type, } elif res_type not in _KEEPS_PENDING_ROLL: new_pending_roll = None @@ -1660,10 +1660,16 @@ async def run( resolution = result.get("resolution") or {} prior_combat = result.get("combat_state") has_active_combat = isinstance(prior_combat, dict) and bool(prior_combat.get("combat_active")) - if ( - has_active_combat - or resolution.get("subsystem_hint") == "combat" - or resolution.get("action_type") == "combat" + awaiting_roll_choice = resolution.get("resolution_type") in ( + "propose_roll", + "forced_narrative_pushback", + ) + if has_active_combat or ( + not awaiting_roll_choice + and ( + resolution.get("subsystem_hint") == "combat" + or resolution.get("action_type") == "combat" + ) ): combat_result = await _run_combat( scene_id=self.scene_id, @@ -1720,6 +1726,7 @@ async def run( memories_to_persist=[], actor_context=getattr(self, "actor_context", None), universe_id=result.get("universe_id") or getattr(self, "universe_id", None), + only_resource_deltas=True, ) if combat_persist.get("working_state") is not None: result["working_state"] = combat_persist["working_state"] @@ -2115,6 +2122,7 @@ def _extract_combat_resource_deltas( "resource_key": "HP", "resource_name": "HP", "delta": hp_delta, + "max_value": int(combatant.get("hp_max") or 10), "source": "combat", "reason": f"Combat damage: HP {pre_hp_val} → {post_hp}", } diff --git a/packages/agents/src/monitor_agents/loops/scene_orchestrator_support.py b/packages/agents/src/monitor_agents/loops/scene_orchestrator_support.py index 1247d79b..b1422855 100644 --- a/packages/agents/src/monitor_agents/loops/scene_orchestrator_support.py +++ b/packages/agents/src/monitor_agents/loops/scene_orchestrator_support.py @@ -17,9 +17,9 @@ def _format_dice_spec( ) -> str: """Return the client dice spec for a pending d20 check.""" if roll_mode == "advantage": - base = "2d20kh1" + base = "2d20kl1" if roll_under else "2d20kh1" elif roll_mode == "disadvantage": - base = "2d20kl1" + base = "2d20kh1" if roll_under else "2d20kl1" else: base = "1d20" if roll_under: diff --git a/packages/agents/src/monitor_agents/resolver.py b/packages/agents/src/monitor_agents/resolver.py index 6b2a3cd0..79dac995 100644 --- a/packages/agents/src/monitor_agents/resolver.py +++ b/packages/agents/src/monitor_agents/resolver.py @@ -217,43 +217,68 @@ def _uses_roll_under(gsr: GameSystemRuntime | None) -> bool: return "roll under" in text or "under or equal" in text or "<=" in text +def _pool_die_sides(core: dict[str, Any]) -> int: + formula = str(core.get("formula") or "") + sides_match = re.search(r"d(\d+)", formula, re.IGNORECASE) + return max(2, int(sides_match.group(1))) if sides_match else 10 + + +def _parse_pool_threshold(threshold_val: Any, *, die_sides: int = 10) -> int: + """Extract a valid per-die success threshold (never concatenating multi-band text).""" + default = min(6, die_sides) + try: + if isinstance(threshold_val, str): + m = re.search(r"\d+", threshold_val) + parsed = int(m.group(0)) if m else default + else: + parsed = int(threshold_val) + except (ValueError, TypeError): + return default + if parsed < 1 or parsed > die_sides: + return default + return parsed + + +def _is_count_successes_pool(core: dict[str, Any] | None) -> bool: + """Return True only for genuine count-successes dice pools (excluding 2d6+stat PbtA / 4dF Fate).""" + if not core or "dice_pool" not in str(core.get("type", "")).lower(): + return False + formula = str(core.get("formula") or "") + if "df" in formula.lower(): + return False + die_sides = _pool_die_sides(core) + raw_threshold = core.get("success_threshold") + if isinstance(raw_threshold, str): + m = re.search(r"\d+", raw_threshold) + if m and int(m.group(0)) > die_sides: + return False + elif isinstance(raw_threshold, (int, float)) and int(raw_threshold) > die_sides: + return False + return True + + def _pool_proposal_fields( gsr: GameSystemRuntime | None, stat_value: int | None, - modifier: int, + pool_bonus: int, ) -> dict[str, Any]: """Return dice-pool metadata when ``gsr`` uses a pool mechanic, else ``{}``.""" - if gsr is None or not gsr._sd.core: + if gsr is None or not _is_count_successes_pool(gsr._sd.core): return {} core = gsr._sd.core - if "dice_pool" not in str(core.get("type", "")).lower(): - return {} - formula = str(core.get("formula") or "") - sides_match = re.search(r"d(\d+)", formula, re.IGNORECASE) - die_sides = max(2, int(sides_match.group(1))) if sides_match else 10 - - threshold = 6 - threshold_val = core.get("success_threshold", 6) - try: - if isinstance(threshold_val, str): - threshold = int("".join(filter(str.isdigit, threshold_val)) or 6) - else: - threshold = int(threshold_val) - except (ValueError, TypeError): - threshold = 6 - if threshold < 1: - threshold = 6 + die_sides = _pool_die_sides(core) + threshold = _parse_pool_threshold(core.get("success_threshold", 6), die_sides=die_sides) if stat_value is not None: try: - pool_size = int(stat_value) + int(modifier) + pool_size = int(stat_value) + int(pool_bonus) except (ValueError, TypeError): pool_size = 1 else: cnt_match = re.match(r"^\s*(\d+)d\d+", formula, re.IGNORECASE) base_pool = int(cnt_match.group(1)) if cnt_match else 1 - pool_size = base_pool + int(modifier) + pool_size = base_pool + int(pool_bonus) if pool_size < 1: pool_size = 1 @@ -1240,11 +1265,12 @@ async def resolve_turn( # The Narrator will frame the invitation for the player to accept. stat_value, base_modifier = self._actor_stat(context, gsr, stat_name) sc_eval = await self._evaluate_scenery_and_conditions(context, user_input or "", roll_mode, gsr) - modifier = base_modifier + sc_eval["total_modifier"] + dynamic_mod = int(sc_eval["total_modifier"]) + modifier = base_modifier + dynamic_mod actual_roll_mode = sc_eval.get("roll_mode") or roll_mode or "normal" if actual_roll_mode not in ("normal", "advantage", "disadvantage"): actual_roll_mode = "normal" - pool_fields = _pool_proposal_fields(gsr, stat_value, modifier) + pool_fields = _pool_proposal_fields(gsr, stat_value, dynamic_mod) if pool_fields: actual_roll_mode = "normal" breakdown = ( @@ -1412,10 +1438,6 @@ def _execute_roll( """ sc_eval = sc_eval or {"cond_modifier": 0, "scenery_modifier": 0, "reasons": []} - m_type = "d20" - if gsr and gsr._sd.core: - m_type = gsr._sd.core.get("type", "d20").lower() - explain_parts = [] if sc_eval.get("cond_modifier", 0) != 0: explain_parts.append(f"conditions: {sc_eval['cond_modifier']}") @@ -1427,33 +1449,33 @@ def _execute_roll( if explain_parts: explain_str = f" [base: {base_modifier}, " + ", ".join(explain_parts) + "]" - if "dice_pool" in m_type: + if gsr and _is_count_successes_pool(gsr._sd.core): from monitor_agents.dice import default_dice_registry - engine = default_dice_registry.get("pool") - - threshold = 6 - if gsr and gsr._sd.core: - threshold_val = gsr._sd.core.get("success_threshold", 6) - try: - if isinstance(threshold_val, str): - threshold = int("".join(filter(str.isdigit, threshold_val)) or 6) - else: - threshold = int(threshold_val) - except (ValueError, TypeError): - threshold = 6 - + core = gsr._sd.core + die_sides = _pool_die_sides(core) + threshold = _parse_pool_threshold(core.get("success_threshold", 6), die_sides=die_sides) + pool_bonus = int(modifier) - int(base_modifier) try: - pool_size = int(stat_value) + modifier + pool_size = int(stat_value) + pool_bonus if stat_value is not None else 1 + pool_bonus except (ValueError, TypeError): pool_size = 1 if pool_size < 1: pool_size = 1 - roll_res = engine.resolve_check(pool_size=pool_size, difficulty=threshold) - total = roll_res.successes + if die_sides == 10: + engine = default_dice_registry.get("pool") + roll_res = engine.resolve_check(pool_size=pool_size, difficulty=threshold) + total = roll_res.successes + botched = roll_res.botches > 0 and total == 0 + raw_rolls = roll_res.raw_rolls + else: + pool_res = roll_dice(f"{pool_size}d{die_sides}") + total = sum(1 for r in pool_res.rolls if r >= threshold) + botched = any(r == 1 for r in pool_res.rolls) and total == 0 + raw_rolls = pool_res.rolls - if roll_res.botches > 0 and total == 0: + if botched: success_level = "critical_failure" elif total >= 1: success_level = "success" @@ -1462,20 +1484,21 @@ def _execute_roll( mode_label = "normal" actual_roll_mode = "normal" - roll_formula = f"{pool_size}d10" - rolls = roll_res.raw_rolls - kept_rolls = roll_res.raw_rolls + roll_formula = f"{pool_size}d{die_sides}" + rolls = raw_rolls + kept_rolls = raw_rolls - spec = f"{pool_size}d10" + spec = f"{pool_size}d{die_sides}" roll_breakdown = f"Pool {pool_size} ({rolls}): {total} successes (diff {threshold}){explain_str}" roll_reason = f"{stat_name} check (pool {pool_size}, diff {threshold}){explain_str}" else: - # Select roll formula based on actual_roll_mode + roll_under = _uses_roll_under(gsr) + # Select roll formula based on actual_roll_mode (roll-under inverts kh/kl) if actual_roll_mode == "advantage": - roll_formula = "2d20kh1" + roll_formula = "2d20kl1" if roll_under else "2d20kh1" mode_label = "advantage" elif actual_roll_mode == "disadvantage": - roll_formula = "2d20kl1" + roll_formula = "2d20kh1" if roll_under else "2d20kl1" mode_label = "disadvantage" else: roll_formula = "1d20" @@ -1487,7 +1510,7 @@ def _execute_roll( natural = kept_rolls[0] if kept_rolls else (rolls[0] if rolls else roll.total) total = roll.total + modifier - if _uses_roll_under(gsr): + if roll_under: target = dc + modifier if natural == 1: success_level = "critical_success" @@ -1511,11 +1534,12 @@ def _execute_roll( else: success_level = "failure" - spec = f"1d20+{modifier}" if modifier >= 0 else f"1d20{modifier}" - if actual_roll_mode != "normal": + if roll_under: + spec = roll_formula + else: spec = f"{roll_formula}+{modifier}" if modifier >= 0 else f"{roll_formula}{modifier}" - if _uses_roll_under(gsr): + if roll_under: target = dc + modifier roll_breakdown = f"{roll_formula}({natural}) <= {dc} + {modifier} = {target}{explain_str}" roll_reason = f"{stat_name} check (target {target}, roll-under, {mode_label}){explain_str}" @@ -1689,14 +1713,10 @@ async def resolve_opposed_check( threshold = 6 if gsr and gsr._sd.core: - threshold_val = gsr._sd.core.get("success_threshold", 6) - try: - if isinstance(threshold_val, str): - threshold = int("".join(filter(str.isdigit, threshold_val)) or 6) - else: - threshold = int(threshold_val) - except (ValueError, TypeError): - threshold = 6 + threshold = _parse_pool_threshold( + gsr._sd.core.get("success_threshold", 6), + die_sides=_pool_die_sides(gsr._sd.core), + ) engine = default_dice_registry.get("pool") # Resources aren't in the resolver signature today, so we @@ -1968,21 +1988,17 @@ async def resolve_check( success = total >= dc details = f"Rolled {formula} ({base_roll.rolls}) + Mod {modifier} = {total} (DC {dc})" - elif "dice_pool" in m_type: + elif _is_count_successes_pool(core_mechanic): # Sub-plan 4: route pool-vs-difficulty games through # the VtM V20 dice engine. Provides hunger dice, # willpower reroll, and bestial-failure detection. from monitor_agents.dice import default_dice_registry pool_size = stat_value - threshold_val = core_mechanic.get("success_threshold", 6) - try: - if isinstance(threshold_val, str): - threshold = int("".join(filter(str.isdigit, threshold_val)) or 6) - else: - threshold = int(threshold_val) - except (ValueError, TypeError): - threshold = 6 + threshold = _parse_pool_threshold( + core_mechanic.get("success_threshold", 6), + die_sides=_pool_die_sides(core_mechanic), + ) engine = default_dice_registry.get("pool") dice_res = engine.resolve_check( pool_size=pool_size, diff --git a/packages/agents/src/monitor_agents/services/persistence_service.py b/packages/agents/src/monitor_agents/services/persistence_service.py index 83f582db..ab1ba9f1 100644 --- a/packages/agents/src/monitor_agents/services/persistence_service.py +++ b/packages/agents/src/monitor_agents/services/persistence_service.py @@ -156,6 +156,7 @@ async def persist_working_state( memories_to_persist: list[dict[str, Any]] | None = None, actor_context: dict[str, Any] | None = None, universe_id: UUID | None = None, + only_resource_deltas: bool = False, ) -> dict[str, Any]: """Upsert scene-scoped working state and return a UI-friendly snapshot. @@ -241,17 +242,21 @@ async def persist_working_state( current_stats = dict(working_state.current_stats or base_stats or {}) state_id = working_state.state_id - deltas, condition_tags = await derive_state_deltas( - resolution, - user_input, - narrative_text, - resources=merged_resources, - game_context=game_context, - ) - if condition_tags: - current_stats["conditions"] = condition_tags - current_stats["narrative_pressure"] = str(resolution.get("narrative_pressure") or "steady") - current_stats["last_success_level"] = str(resolution.get("success_level") or "success") + if only_resource_deltas: + deltas: dict[str, Any] = {} + condition_tags: list[str] = list(current_stats.get("conditions") or []) + else: + deltas, condition_tags = await derive_state_deltas( + resolution, + user_input, + narrative_text, + resources=merged_resources, + game_context=game_context, + ) + if condition_tags: + current_stats["conditions"] = condition_tags + current_stats["narrative_pressure"] = str(resolution.get("narrative_pressure") or "steady") + current_stats["last_success_level"] = str(resolution.get("success_level") or "success") for key, delta in list(deltas.items()): if key not in merged_resources: @@ -266,21 +271,26 @@ async def persist_working_state( if delta: merged_resources[key] = apply_resource_delta(merged_resources[key], delta) - # Apply ResourceEngine deltas (earn_on_failure, session_start, etc.) + # Apply ResourceEngine deltas (earn_on_failure, session_start, combat, etc.) engine_modifications: list[Any] = [] for rd in resource_deltas: key = str(rd.get("resource_key", "")) delta = int(rd.get("delta", 0)) if not key or not delta: continue - if key not in merged_resources: - merged_resources[key] = normalise_resource_value(key, 0, 0) + matched_key = next( + (k for k in merged_resources if k.lower() == key.lower()), + key, + ) + if matched_key not in merged_resources: + default_cap = int(rd.get("max_value") or (10 if key.lower() == "hp" else 0)) + merged_resources[matched_key] = normalise_resource_value(matched_key, default_cap, default_cap) if delta: - merged_resources[key] = apply_resource_delta(merged_resources[key], delta) - engine_modifications.append((key, delta, f"ResourceEngine: {rd.get('source', 'engine')}")) + merged_resources[matched_key] = apply_resource_delta(merged_resources[matched_key], delta) + engine_modifications.append((matched_key, delta, f"ResourceEngine: {rd.get('source', 'engine')}")) # Award XP based on the turn's resolution outcome (P-21 progression) - xp_award = _award_xp(resolution, game_context) + xp_award = 0 if only_resource_deltas else _award_xp(resolution, game_context) if xp_award > 0: current_xp = int(working_state.xp or 0) if working_state else 0 current_stats["xp"] = current_xp + xp_award @@ -333,8 +343,12 @@ def _add_modification(key: str, delta: int, src: str = source_text) -> Any: } new_pending_proposals: list[dict[str, Any]] = [] - add_tags = canonical_state_tags(condition_tags, deltas, merged_resources, game_context) - if any(delta for delta in deltas.values()) or condition_tags or add_tags: + combined_changes = {key: delta for key, delta in deltas.items() if delta} + for key, delta, _src in engine_modifications: + if key != "xp" and delta: + combined_changes[key] = combined_changes.get(key, 0) + delta + add_tags = canonical_state_tags(condition_tags, combined_changes, merged_resources, game_context) + if combined_changes or condition_tags or add_tags: turn_uuid = coerce_uuid( turn_id, seed=f"monitor://scene/{scene_id}/turn/{turn_id or uuid4()}", @@ -346,12 +360,12 @@ def _add_modification(key: str, delta: int, src: str = source_text) -> Any: "turn_id": str(turn_uuid), "add_tags": add_tags, "remove_tags": [], - "resource_changes": {key: delta for key, delta in deltas.items() if delta}, + "resource_changes": combined_changes, "resources": merged_resources, "condition_tags": condition_tags, "checkpoint_summary": checkpoint["summary"], } - proposal_summary = PersistenceService.build_state_change_summary(deltas, condition_tags, add_tags) + proposal_summary = PersistenceService.build_state_change_summary(combined_changes, condition_tags, add_tags) try: created_proposal = await anyio.to_thread.run_sync( diff --git a/packages/agents/tests/test_resolver_dice_authority.py b/packages/agents/tests/test_resolver_dice_authority.py index e8f6fe84..68e25d50 100644 --- a/packages/agents/tests/test_resolver_dice_authority.py +++ b/packages/agents/tests/test_resolver_dice_authority.py @@ -267,3 +267,92 @@ async def test_player_sees_the_pushback_prompt_and_can_roll() -> None: assert out.response_text == "Bold claim. Roll Dexterity." assert out.metadata["dice_request"]["spec"] == "1d20+3" assert session["pending_dice_request"]["difficulty_class"] == 25 + + +def test_roll_under_advantage_and_disadvantage_invert_keep_direction(): + """F2: In roll-under systems (lower is better), advantage keeps lowest (2d20kl1).""" + from monitor_agents.loops.scene_orchestrator_support import _format_dice_spec + + assert _format_dice_spec(0, roll_under=True, roll_mode="advantage") == "2d20kl1" + assert _format_dice_spec(0, roll_under=True, roll_mode="disadvantage") == "2d20kh1" + + r = _resolver(None) + gsr_under = _gsr(core={"type": "roll_under", "formula": "1d20 <= STAT"}) + roll_kw: dict[str, Any] = { + "scene_id": "s", + "user_input": "u", + "action_type": "action", + "roll_type": "check", + "intent_target": "", + "requires_clarification": False, + "subsystem_hint": None, + "roll_necessity": "roll", + } + with patch("monitor_data.utils.dice.random.randint", side_effect=[15, 4]): + res = r._execute_roll( + **roll_kw, + stat_name="BODY", + stat_value=13, + base_modifier=0, + modifier=0, + dc=13, + gsr=gsr_under, + actual_roll_mode="advantage", + ) + assert res["roll_total"] == 4 + assert res["roll_detail"]["spec"] == "2d20kl1" + assert res["success"] is True + + +def test_dice_pool_does_not_double_stat_or_concatenate_pbta_bands(): + """F3: vampire.json-style 1..5 attributes do not double pool_size; PbtA bands don't become threshold 10796.""" + from monitor_agents.resolver import ( + _is_count_successes_pool, + _parse_pool_threshold, + _pool_proposal_fields, + ) + + # PbtA / Apocalypse World 2d6+stat with multi-band string is not a count-successes pool + pbta_cm = { + "type": "dice_pool", + "formula": "2d6 + stat", + "success_threshold": "10+ full success, 7-9 partial/mixed, 6- fail", + } + assert _is_count_successes_pool(pbta_cm) is False + assert _parse_pool_threshold(pbta_cm["success_threshold"], die_sides=6) == 6 + + # Vampire-style d10 pool where calc_modifier returns stat_value (base_modifier == stat_value == 4) + vtm_gsr = _gsr( + core={ + "type": "dice_pool", + "formula": "Pool of d10s (Attribute + Skill)", + "success_threshold": "6+ on each die is a success", + } + ) + fields = _pool_proposal_fields(vtm_gsr, stat_value=4, pool_bonus=0) + assert fields["pool_size"] == 4 + assert fields["success_threshold"] == 6 + + r = _resolver(None) + with patch("monitor_agents.dice.pool.random.randint", side_effect=[8, 6, 3, 9]): + res = r._execute_roll( + scene_id="s", + user_input="u", + action_type="action", + roll_type="check", + intent_target="", + requires_clarification=False, + subsystem_hint=None, + roll_necessity="roll", + stat_name="DEX", + stat_value=4, + base_modifier=4, + modifier=4, + dc=2, + gsr=vtm_gsr, + actual_roll_mode="normal", + ) + assert len(res["roll_detail"]["rolls"]) == 4 + assert res["roll_total"] == 3 # 8, 6, 9 are >= 6 + assert res["success"] is True + diff --git a/packages/agents/tests/test_scene_state_carryover.py b/packages/agents/tests/test_scene_state_carryover.py index acb8d571..4aa8eef8 100644 --- a/packages/agents/tests/test_scene_state_carryover.py +++ b/packages/agents/tests/test_scene_state_carryover.py @@ -512,6 +512,7 @@ async def narrate_turn(self, **_kw: Any) -> dict[str, Any]: "resource_key": "HP", "resource_name": "HP", "delta": -3, + "max_value": 10, "source": "combat", "reason": "Combat damage: HP 10 → 7", } @@ -521,6 +522,7 @@ async def narrate_turn(self, **_kw: Any) -> dict[str, Any]: ] assert len(combat_persist_calls) == 1 assert combat_persist_calls[0].kwargs["resource_deltas"][0]["delta"] == -3 + assert combat_persist_calls[0].kwargs.get("only_resource_deltas") is True # Turn 2 resumes Round 2 with Brigand at 4 HP; Hero deals 6 -> Brigand drops to 0 and combat ends! t2 = await loop.run("I finish the brigand") @@ -528,3 +530,26 @@ async def narrate_turn(self, **_kw: Any) -> dict[str, Any]: assert t2["combat_result"]["victory_side"] == "pc" assert t2["combat_state"] is None assert isinstance(loop, sl.SceneLoop) and isinstance(cl.CombatLoop, type) + + +@pytest.mark.asyncio +async def test_propose_roll_and_pushback_defer_combat_and_set_pending_roll(world: _World) -> None: + """F4 / F12a: Combat does not auto-run while awaiting a player roll on propose_roll or pushback.""" + loop = _loop() + world.resolutions = [ + { + "resolution_type": "forced_narrative_pushback", + "stat": "Dexterity", + "modifier": 2, + "difficulty_class": 25, + "reason": "Too ambitious without a roll", + } + ] + combat_spy = AsyncMock(return_value={"combat_result": {"combat_active": True}}) + with patch.object(sl, "_run_combat", combat_spy): + res = await loop.run("I decapitate the warlord in one swing") + + assert res["pending_roll"]["resolution_type"] == "forced_narrative_pushback" + assert res["pending_roll"]["dc"] == 25 + combat_spy.assert_not_awaited() + diff --git a/packages/data-layer/src/monitor_data/db/postgres.py b/packages/data-layer/src/monitor_data/db/postgres.py index 7f82e18c..aeeadeec 100644 --- a/packages/data-layer/src/monitor_data/db/postgres.py +++ b/packages/data-layer/src/monitor_data/db/postgres.py @@ -432,12 +432,22 @@ def __init__( self._pools_by_loop: weakref.WeakKeyDictionary[ asyncio.AbstractEventLoop, asyncpg.Pool ] = weakref.WeakKeyDictionary() + self._locks_by_loop: weakref.WeakKeyDictionary[ + asyncio.AbstractEventLoop, asyncio.Lock + ] = weakref.WeakKeyDictionary() self._schema_applied: bool = False # ------------------------------------------------------------------ # Connection lifecycle # ------------------------------------------------------------------ + def _loop_lock(self, loop: asyncio.AbstractEventLoop) -> asyncio.Lock: + lock = self._locks_by_loop.get(loop) + if lock is None: + lock = asyncio.Lock() + self._locks_by_loop[loop] = lock + return lock + async def connect(self) -> None: """Open the connection pool and apply schema DDL (idempotent).""" try: @@ -451,24 +461,25 @@ async def connect(self) -> None: self._pool = cached self._pool_loop = running return - - if ( - self._pool is not None - and running is not None - and self._pool_loop is not None - and self._pool_loop is not running - ): - pool = await asyncpg.create_pool( - self._dsn, - min_size=self._min_size, - max_size=self._max_size, - ) - self._pools_by_loop[running] = pool - self._pool = pool - self._pool_loop = running - if not self._schema_applied: - await self._apply_schema() - return + if self._pool is not None and self._pool_loop is None: + return + async with self._loop_lock(running): + cached = self._pools_by_loop.get(running) + if cached is not None: + self._pool = cached + self._pool_loop = running + return + pool = await asyncpg.create_pool( + self._dsn, + min_size=self._min_size, + max_size=self._max_size, + ) + self._pools_by_loop[running] = pool + self._pool = pool + self._pool_loop = running + if not self._schema_applied: + await self._apply_schema(pool) + return if self._pool is None: pool = await asyncpg.create_pool( @@ -477,11 +488,9 @@ async def connect(self) -> None: max_size=self._max_size, ) self._pool = pool - self._pool_loop = running - if running is not None: - self._pools_by_loop[running] = pool + self._pool_loop = None if not self._schema_applied: - await self._apply_schema() + await self._apply_schema(pool) async def close(self) -> None: """Close all pooled connections gracefully.""" @@ -489,16 +498,23 @@ async def close(self) -> None: running = asyncio.get_running_loop() except RuntimeError: running = None + if running is not None: + loop_pool = self._pools_by_loop.pop(running, None) + if loop_pool is not None: + await loop_pool.close() + if self._pool is loop_pool: + self._pool = None + self._pool_loop = None + return if self._pool is not None: if self._pool_loop in (None, running): await self._pool.close() - if running is not None: - self._pools_by_loop.pop(running, None) self._pool = None self._pool_loop = None - async def _apply_schema(self) -> None: - async with self._pool.acquire() as conn, conn.transaction(): # type: ignore[union-attr] + async def _apply_schema(self, pool: asyncpg.Pool | None = None) -> None: + target_pool = pool or self._pool + async with target_pool.acquire() as conn, conn.transaction(): # type: ignore[union-attr] await conn.execute(_SCHEMA_SQL) self._schema_applied = True diff --git a/packages/data-layer/src/monitor_data/interop/sillytavern_lorebook.py b/packages/data-layer/src/monitor_data/interop/sillytavern_lorebook.py index 45ef2d26..8be77ca5 100644 --- a/packages/data-layer/src/monitor_data/interop/sillytavern_lorebook.py +++ b/packages/data-layer/src/monitor_data/interop/sillytavern_lorebook.py @@ -72,6 +72,7 @@ "uid", "displayIndex", "addMemo", + "disable", "enabled", "content", "id", @@ -411,8 +412,9 @@ def build_st_lorebook_entry(entry: LorebookEntryCreate | BaseModel | dict[str, A "addMemo": True, } - # If explicit ST field values are present in extensions, let them win. - out.update(st_extensions) + # If explicit ST field values are present in extensions, let them win — + # except active/disabled state, which is authoritative on ``is_active``. + out.update({k: v for k, v in st_extensions.items() if k not in ("disable", "enabled")}) return out diff --git a/packages/data-layer/tests/test_db/test_postgres.py b/packages/data-layer/tests/test_db/test_postgres.py index f63eb995..a52658db 100644 --- a/packages/data-layer/tests/test_db/test_postgres.py +++ b/packages/data-layer/tests/test_db/test_postgres.py @@ -101,6 +101,51 @@ async def fake_create_pool(*_args, **_kwargs): loop2.close() +@pytest.mark.asyncio +async def test_concurrent_same_loop_connect_creates_single_pool(mock_pool): + import asyncio + + pool, _conn = mock_pool + + async def slow_create_pool(*_args, **_kwargs): + await asyncio.sleep(0.01) + return pool + + c = PostgresClient(dsn="postgresql://test") + with patch("monitor_data.db.postgres.asyncpg.create_pool", side_effect=slow_create_pool) as create_mock: + p1, p2 = await asyncio.gather(c._get_pool(), c._get_pool()) + assert p1 is pool and p2 is pool + assert create_mock.call_count == 1 + + +def test_close_releases_current_loop_pool_even_after_other_loop_connects(mock_pool): + import asyncio + + pool1, _conn1 = mock_pool + pool2 = AsyncMock() + pool2.acquire = pool1.acquire + pool2.close = AsyncMock() + created = [pool1, pool2] + + async def fake_create_pool(*_args, **_kwargs): + return created.pop(0) + + c = PostgresClient(dsn="postgresql://test") + loop1 = asyncio.new_event_loop() + loop2 = asyncio.new_event_loop() + try: + with patch("monitor_data.db.postgres.asyncpg.create_pool", side_effect=fake_create_pool): + loop1.run_until_complete(c._get_pool()) + loop2.run_until_complete(c._get_pool()) + # Now c._pool points at pool2, but closing on loop1 must still close pool1 + loop1.run_until_complete(c.close()) + pool1.close.assert_awaited_once() + finally: + loop1.close() + loop2.close() + + + @pytest.mark.asyncio async def test_connect_skips_if_pool_exists(mock_pool): pool, conn = mock_pool diff --git a/packages/data-layer/tests/test_sillytavern_lorebook.py b/packages/data-layer/tests/test_sillytavern_lorebook.py index f72852ef..715cb788 100644 --- a/packages/data-layer/tests/test_sillytavern_lorebook.py +++ b/packages/data-layer/tests/test_sillytavern_lorebook.py @@ -305,3 +305,26 @@ def test_round_trip_keeps_disabled_entries_off(self): assert len(parsed_entries) == 1 assert parsed_entries[0]["is_active"] is False assert parsed_entries[0]["order"] == 15 + + def test_toggling_is_active_after_import_not_overwritten_by_st_extensions(self): + from monitor_data.interop.sillytavern_lorebook import ( + build_character_book, + build_st_lorebook_entry, + parse_st_lorebook_entry, + ) + + imported = parse_st_lorebook_entry( + {"content": "Secret vault", "keys": ["vault"], "disable": False, "enabled": True} + ) + assert "disable" not in imported.get("st_extensions", {}) + # User toggles is_active off in UI; even if legacy st_extensions had disable/enabled, + # current is_active is authoritative on export. + imported["is_active"] = False + imported["st_extensions"] = {"disable": False, "enabled": True, "custom": 7} + st_out = build_st_lorebook_entry(imported, uid=0) + assert st_out["disable"] is True + assert st_out["custom"] == 7 + cb_out = build_character_book([imported])["entries"][0] + assert cb_out["enabled"] is False + assert cb_out["disable"] is True + diff --git a/packages/ui/backend/src/monitor_ui/routers/character_conversation.py b/packages/ui/backend/src/monitor_ui/routers/character_conversation.py index b5aae5fc..9d3b4646 100644 --- a/packages/ui/backend/src/monitor_ui/routers/character_conversation.py +++ b/packages/ui/backend/src/monitor_ui/routers/character_conversation.py @@ -23,6 +23,7 @@ from __future__ import annotations import asyncio +import contextlib import hashlib import uuid from asyncio import Lock as _AsyncLock @@ -577,12 +578,19 @@ def _find() -> dict[str, Any] | None: "client_session.persona_character_id": persona_id, "client_session.fingerprints": fingerprint, }, - {"conversation_id": 1}, + {"conversation_id": 1, "turns": 1}, sort=[("updated_at", -1)], ) doc = await asyncio.to_thread(_find) - return str(doc["conversation_id"]) if doc else None + if not doc: + return None + player_turns = sum( + 1 for t in (doc.get("turns") or []) if isinstance(t, dict) and t.get("speaker_role") == "player" + ) + if player_turns >= 100: + return None + return str(doc["conversation_id"]) async def remember_client_conversation( @@ -615,7 +623,12 @@ def _save() -> None: await asyncio.to_thread(_save) -async def resume_conversation(character_id: str, conversation_id: str) -> Any | None: +async def resume_conversation( + character_id: str, + conversation_id: str, + *, + allow_at_limit: bool = False, +) -> Any | None: """Rebuild an in-memory DIRECT loop from its persisted MongoDB transcript. Light-RP loops are process-local; a backend restart orphans active @@ -652,7 +665,7 @@ async def resume_conversation(character_id: str, conversation_id: str) -> Any | ) loop._apply(await load_npc_context(loop.state)) loop.state.lorebook_character_ids = [character_id] - loop.state.turns = [ + restored_turns = [ { "turn_index": t.get("turn_index", i), "speaker_role": t.get("speaker_role"), @@ -661,9 +674,18 @@ async def resume_conversation(character_id: str, conversation_id: str) -> Any | } for i, t in enumerate(doc.get("turns", [])) ] + # Drop any trailing unanswered player turns left by a failed/interrupted LLM call. + had_trailing_player = False + while restored_turns and restored_turns[-1].get("speaker_role") == "player": + restored_turns.pop() + had_trailing_player = True + if had_trailing_player: + with contextlib.suppress(Exception): + coll.update_one({"conversation_id": conversation_id}, {"$set": {"turns": restored_turns}}) + loop.state.turns = restored_turns # turns_count tracks completed player exchanges, not individual messages. loop.state.turns_count = sum(1 for t in loop.state.turns if t.get("speaker_role") == "player") - if loop.state.turns_count >= getattr(loop.state, "max_turns", 100): + if not allow_at_limit and loop.state.turns_count >= getattr(loop.state, "max_turns", 100): return None # Restore the durable proposal outbox written after each turn (crash-safe). loop.state.pending_proposals = list(doc.get("pending_proposals") or []) @@ -702,8 +724,22 @@ async def send_message( raise KeyError(conversation_id) loop = cached - if loop is None and character_id: - loop = await resume_conversation(character_id, conversation_id) + if loop is None: + resolved_char_id = character_id + if not resolved_char_id: + with contextlib.suppress(Exception): + from monitor_data.db.mongodb import get_mongodb_client + + doc = get_mongodb_client().get_collection("conversations").find_one( + {"conversation_id": conversation_id}, + {"client_session": 1}, + ) + if isinstance(doc, dict): + cs = doc.get("client_session") + if isinstance(cs, dict) and cs.get("character_id"): + resolved_char_id = str(cs["character_id"]) + if resolved_char_id: + loop = await resume_conversation(resolved_char_id, conversation_id) if loop is None: raise KeyError(conversation_id) @@ -718,6 +754,15 @@ async def send_message( except ConversationClosedError as exc: pop_loop(conversation_id) raise KeyError(conversation_id) from exc + except Exception: + with contextlib.suppress(Exception): + from monitor_data.db.mongodb import get_mongodb_client + + get_mongodb_client().get_collection("conversations").update_one( + {"conversation_id": conversation_id}, + {"$set": {"turns": list(getattr(loop.state, "turns", []) or [])}}, + ) + raise finally: # Always reset the transient flag, even if step() raised. if include_cross_incarnation and getattr(loop, "state", None) is not None: @@ -825,7 +870,7 @@ async def end_conversation(conversation_id: str, character_id: str | None = None loop = cached if loop is None and character_id: - loop = await resume_conversation(character_id, conversation_id) + loop = await resume_conversation(character_id, conversation_id, allow_at_limit=True) if loop is None: return {"ended": True, "proposals": 0} proposals: list[Any] = [] diff --git a/packages/ui/backend/src/monitor_ui/routers/chat.py b/packages/ui/backend/src/monitor_ui/routers/chat.py index 7faae529..a4b13ca9 100644 --- a/packages/ui/backend/src/monitor_ui/routers/chat.py +++ b/packages/ui/backend/src/monitor_ui/routers/chat.py @@ -1270,8 +1270,15 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None: try: while True: raw = await websocket.receive_text() - data = json.loads(raw) - msg_type: str = data.get("type", "message") + try: + data = json.loads(raw) + if not isinstance(data, dict): + raise ValueError("WebSocket frame must be a JSON object.") + except (ValueError, TypeError) as exc: + if not await _send_ws_payload(websocket, {"type": "error", "detail": f"Invalid JSON payload: {exc}"}): + break + continue + msg_type: str = str(data.get("type", "message") or "message") raw_content = data.get("content", "") content = raw_content if isinstance(raw_content, str) else "" @@ -1298,8 +1305,16 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None: # Handle dice_result if msg_type == "dice_result": - spec: str = data.get("spec", "") - value: int = int(data.get("value", 0)) + spec: str = str(data.get("spec") or "") + try: + value: int = int(data.get("value", 0)) + except (TypeError, ValueError): + if not await _send_ws_payload( + websocket, + {"type": "error", "detail": "dice_result value must be an integer."}, + ): + break + continue # reason: `list` is the bare generic missing type parameters; `data` is `Any` (json.loads return), so `.get("rolls", [])` is `Any` and cannot be parameterized without narrowing rolls: list = data.get("rolls", []) # type: ignore reason: str = data.get("reason", spec) @@ -1503,6 +1518,29 @@ async def on_event(kind: str, data): # type: ignore session_game_system_doc=_session_game_system_doc, gsr_available=_GSR_AVAILABLE, ) + elif session.get("phase") == "scene_ended": + story_id, scene_id, _ = _bootstrap_story_scene(session) + _db_save_session(session) + if scene_id: + narrative = "The world settles. A new scene opens before you — what do you do?" + meta = { + "type": "scene_resume", + "phase": "active_play", + "story_id": story_id, + "scene_id": scene_id, + } + else: + narrative, meta = await _run_scene_turn( + session_id, + content, + sessions=_SESSIONS, + messages=_MESSAGES, + db_save_session=_db_save_session, + db_load_messages=_db_load_messages, + bootstrap_story_scene=_bootstrap_story_scene, + session_game_system_doc=_session_game_system_doc, + gsr_available=_GSR_AVAILABLE, + ) else: narrative, meta = await _run_scene_turn( session_id, @@ -1528,15 +1566,18 @@ async def on_event(kind: str, data): # type: ignore # client-side timeouts, instead of streaming a fake GM # message — the player should see "the AI hit a limit, # retry", not a blank or placeholder turn. - error_payload = { + error_payload: dict[str, Any] = { "type": "error", "message_id": gm_id, "detail": narrative, "error_class": meta.get("error_class"), } - await _send_ws_payload(websocket, error_payload) + if meta.get("dice_request"): + error_payload["dice_request"] = meta["dice_request"] + if not await _send_ws_payload(websocket, error_payload): + break await fanout_event(session_id, error_payload, exclude=websocket) - break + continue if not streamed_tokens: for word in narrative.split(): diff --git a/packages/ui/backend/src/monitor_ui/routers/lorebook.py b/packages/ui/backend/src/monitor_ui/routers/lorebook.py index 34540195..79cb6277 100644 --- a/packages/ui/backend/src/monitor_ui/routers/lorebook.py +++ b/packages/ui/backend/src/monitor_ui/routers/lorebook.py @@ -75,10 +75,16 @@ async def list_entries( description="Sort field: priority, trigger_count, created_at", ), ascending: bool = Query(default=False), + include_inactive: bool = Query(default=False), ) -> list[LorebookEntry]: - """List all active lorebook entries for a character.""" + """List lorebook entries for a character (active only by default).""" # reason: mongodb_get_lorebook_entries returns list[LorebookEntry] but the Mongo cursor bridges Any through motor; FastAPI/Pydantic v2 returns Any for the response_model inference without explicit Annotated[] - return mongodb_get_lorebook_entries(character_id=character_id, sort_by=sort_by, ascending=ascending) # type: ignore + return mongodb_get_lorebook_entries( + character_id=character_id, + sort_by=sort_by, # type: ignore[arg-type] + ascending=ascending, + include_inactive=include_inactive, + ) @router.get("/lorebook/entries/{entry_id}", response_model=LorebookEntry) diff --git a/packages/ui/backend/tests/test_character_conversation.py b/packages/ui/backend/tests/test_character_conversation.py index 83f60188..e0c4605b 100644 --- a/packages/ui/backend/tests/test_character_conversation.py +++ b/packages/ui/backend/tests/test_character_conversation.py @@ -1343,3 +1343,86 @@ def test_closed_loop_evicts_and_maps_to_keyerror(self): with pytest.raises(KeyError): asyncio.run(cc.send_message("conv-ceil", "extra", character_id="char-1")) assert cc.get_loop("conv-ceil") is None + + def test_end_conversation_can_resume_at_turn_limit_and_find_skips_maxed(self): + """F1: A conversation at 100 player turns can still be resumed by end_conversation, + while find_client_conversation skips it so Risu starts a fresh conversation.""" + from uuid import uuid4 + + entity_id, universe_id = str(uuid4()), str(uuid4()) + conv_id = str(uuid4()) + turns_100 = [] + for i in range(100): + turns_100.append({"turn_index": i * 2, "speaker_role": "player", "text": f"p{i}"}) + turns_100.append({"turn_index": i * 2 + 1, "speaker_role": "npc", "text": f"n{i}"}) + + doc = { + "conversation_id": conv_id, + "status": "active", + "universe_id": universe_id, + "npc_ids": [entity_id], + "turns": turns_100, + "pending_proposals": [{"change_type": "fact"}], + } + mongo = MagicMock() + col = mongo.get_collection.return_value + col.find_one.return_value = doc + col.find.return_value.sort.return_value = [doc] + + backing = {"entity_id": entity_id, "universe_id": universe_id, "version_id": "v1"} + fake_loop = MagicMock() + fake_loop.state = SimpleNamespace( + turns=[], turns_count=0, pending_proposals=[{"change_type": "fact"}], lorebook_character_ids=["char-1"] + ) + fake_loop.finish = AsyncMock(return_value=[{"change_type": "fact"}]) + + with ( + patch("monitor_data.db.mongodb.get_mongodb_client", return_value=mongo), + patch.object(cc, "ensure_character_backed", new=AsyncMock(return_value=backing)), + patch("monitor_agents.loops.conversation_loop.ConversationLoop", return_value=fake_loop), + patch("monitor_agents.loops.conversation_loop.load_npc_context", new=AsyncMock(return_value={})), + ): + # Regular resume_conversation rejects >= 100 player turns by returning None + assert asyncio.run(cc.resume_conversation("char-1", conv_id)) is None + # find_client_conversation skips maxed conversation + assert asyncio.run(cc.find_client_conversation("char-1", None, "Evening.")) is None + # end_conversation succeeds via allow_at_limit=True + ended = asyncio.run(cc.end_conversation(conv_id, character_id="char-1")) + assert ended == {"ended": True, "proposals": 1} + fake_loop.finish.assert_awaited_once() + + def test_resume_strips_orphan_trailing_player_turns(self): + """F11a: Trailing unanswered player turns from a failed LLM call are stripped on resume.""" + from uuid import uuid4 + + entity_id, universe_id = str(uuid4()), str(uuid4()) + conv_id = str(uuid4()) + doc = { + "conversation_id": conv_id, + "status": "active", + "universe_id": universe_id, + "npc_ids": [entity_id], + "turns": [ + {"turn_index": 0, "speaker_role": "player", "text": "Hello"}, + {"turn_index": 1, "speaker_role": "npc", "text": "Greetings."}, + {"turn_index": 2, "speaker_role": "player", "text": "Orphan failed turn"}, + ], + } + mongo = MagicMock() + mongo.get_collection.return_value.find_one.return_value = doc + backing = {"entity_id": entity_id, "universe_id": universe_id, "version_id": "v1"} + fake_loop = MagicMock() + fake_loop.state = SimpleNamespace(turns=[], turns_count=0) + + with ( + patch("monitor_data.db.mongodb.get_mongodb_client", return_value=mongo), + patch.object(cc, "ensure_character_backed", new=AsyncMock(return_value=backing)), + patch("monitor_agents.loops.conversation_loop.ConversationLoop", return_value=fake_loop), + patch("monitor_agents.loops.conversation_loop.load_npc_context", new=AsyncMock(return_value={})), + ): + loop = asyncio.run(cc.resume_conversation("char-1", conv_id)) + + assert len(loop.state.turns) == 2 + assert loop.state.turns_count == 1 + mongo.get_collection.return_value.update_one.assert_called_once() +