Session-state audit fix: thread session_id end-to-end, advance substantive-only, use monotonic seq counter

Four interlocking bugs that together made the listening layer fragile
on the live demo:

1. demo/app.py hardcoded session_id='demo' in FastDemoPipeline.run /
run_streaming. The fresh uuid generated by the 'New conversation'
button (and by every page-load via gr.State) was NEVER passed to
the core. Every conversation across users / refreshes / resets
shared the same 'demo' state entry in EmpathRAGCore, causing
state accumulation forever. Now session_id flows from session_state
through respond() into both run() and run_streaming().

2. reset_session_handler created a new uuid but did not clear the
core's lingering state for the OLD session_id. Added a call to
pipeline.reset_session(session_id=prev_sid) that the button now
invokes via wiring through the session_state input.

3. tier_history is a rolling window of size 3, so the
tier_history-based turn_index saturates at 4 forever. That made
the consent-flow recency check (current_turn - offer_turn in 1..2)
silently fail past the fourth turn. Added session_seq — a
monotonic per-session counter that does not cap — and use it for
open_offer.offer_turn, acknowledged_turn, delivered_turn, and the
_open_offer_recent diff. turn_index keeps its semantics for the
safety tracker.

4. Greetings / hedges / incomplete-message clarifies consumed
listening-layer stage slots because decide_stage was driven by
raw turn_index. A user who opened with 'Hi' then said something
substantive landed at PERMISSION, not LISTEN — invisible to the
listening layer demo. Added session_substantive_count which only
increments on LISTEN / PERMISSION / OFFER stages, and decide_stage
now reads effective_turn_index from that counter. CLARIFY turns
no longer consume the slot.

Verified with six end-to-end scenarios: greeting-then-loop, clean
5-turn, many-trailing-affirms (no infinite loop), parallel users (no
leak), crisis fresh, and hedge-then-loop.

demo/app.py

@@ -1218,17 +1218,27 @@ class FastDemoPipeline:
         self._crisis_locked = False
         self._last_escalation_reason = ""
 
-    def run(self, user_message: str, audience_mode: str = "student") -> dict:
+    def run(
+        self,
+        user_message: str,
+        audience_mode: str = "student",
+        session_id: str = "demo",
+    ) -> dict:
         core_result = self.core.run_turn(
             message=user_message,
-            session_id="demo",
+            session_id=session_id,
             audience_mode=audience_mode,
             resource_profile="umd",
             backend_mode="hybrid_ml",
         ).to_dict()
         return self._enrich_result(core_result)
 
-    def run_streaming(self, user_message: str, audience_mode: str = "student"):
+    def run_streaming(
+        self,
+        user_message: str,
+        audience_mode: str = "student",
+        session_id: str = "demo",
+    ):
         """Generator wrapping ``EmpathRAGCore.run_turn_streaming``.
 
         Yields ``("token", text)`` for each streamed chunk and ``("done",
@@ -1236,7 +1246,7 @@ class FastDemoPipeline:
         """
         for event in self.core.run_turn_streaming(
             message=user_message,
-            session_id="demo",
+            session_id=session_id,
             audience_mode=audience_mode,
             resource_profile="umd",
             backend_mode="hybrid_ml",
@@ -1380,12 +1390,12 @@ class FastDemoPipeline:
     def tracker_trajectory(self) -> str:
         return "stable"
 
-    def reset_session(self) -> None:
+    def reset_session(self, session_id: str = "demo") -> None:
         self._turn = 0
         self._tier_history = []
         self._crisis_locked = False
         self._last_escalation_reason = ""
-        self.core.reset_session("demo")
+        self.core.reset_session(session_id)
 
     def _result(
         self,
@@ -2558,13 +2568,15 @@ def respond(message, chat_history, session_state, audience_mode, rephrase_mode="
             def fast_check(text, threshold=0.5, skip_ig=False):
                 return original_check(text, threshold=threshold, skip_ig=True)
             active_pipeline.guardrail.check = fast_check
-            result = active_pipeline.run(message)
+            result = active_pipeline.run(message, session_id=session_id)
             active_pipeline.guardrail.check = original_check
             session_state["tracker_history"] = active_pipeline.tracker.history()
             session_state["conv_history"] = list(active_pipeline.conv_history)
         elif use_real_streaming:
             result = None
-            for ev in active_pipeline.run_streaming(message, audience_mode=audience_mode or "student"):
+            for ev in active_pipeline.run_streaming(
+                message, audience_mode=audience_mode or "student", session_id=session_id
+            ):
                 if ev[0] == "token":
                     chat_history[-1] = (message, ev[1])
                     yield (
@@ -2579,7 +2591,9 @@ def respond(message, chat_history, session_state, audience_mode, rephrase_mode="
             session_state["tracker_history"] = session_state.get("tracker_history", []) + [result["emotion"]]
             session_state["conv_history"] = session_state.get("conv_history", [])
         else:
-            result = active_pipeline.run(message, audience_mode=audience_mode or "student")
+            result = active_pipeline.run(
+                message, audience_mode=audience_mode or "student", session_id=session_id
+            )
             session_state["tracker_history"] = session_state.get("tracker_history", []) + [result["emotion"]]
             session_state["conv_history"] = session_state.get("conv_history", [])
 
@@ -2739,7 +2753,26 @@ def export_support_plan_pdf(session_state):
         return export_support_plan_md(session_state)
 
 
-def reset_session_handler():
+def reset_session_handler(prev_session_state=None):
+    """Reset to a brand-new session. Crucially, also clears any lingering
+    EmpathRAGCore state keyed to the previous session_id (tier_history,
+    open-offer slot, last-stage marker, intl flags, message history).
+    Without this, clicking "New conversation" only swapped the UI label
+    while the core kept growing state under the old key — which is why a
+    "fresh" run on the demo could still see prior turns in context.
+    """
+    prev_sid = (prev_session_state or {}).get("session_id")
+    with pipeline_lock:
+        try:
+            pipeline = get_pipeline()
+            if hasattr(pipeline, "reset_session") and prev_sid:
+                pipeline.reset_session(session_id=prev_sid)
+            elif hasattr(pipeline, "reset_session"):
+                pipeline.reset_session()
+        except Exception:
+            # Reset must never block UI; failed resets degrade silently
+            # because the new session_id alone gives a clean state entry.
+            pass
     session_state = new_session_state()
     return (
         [],
@@ -3090,11 +3123,11 @@ with gr.Blocks(theme=theme, title="EmpathRAG Studio", css=APP_CSS, js=_CHATBOT_A
         outputs=submit_outputs,
     ).then(_clear_input, outputs=msg_box)
 
-    def reset_with_chrome():
-        base = reset_session_handler()
+    def reset_with_chrome(prev_session_state):
+        base = reset_session_handler(prev_session_state)
         return base + (gr.update(visible=True),)
 
-    reset_btn.click(reset_with_chrome, outputs=submit_outputs)
+    reset_btn.click(reset_with_chrome, inputs=[session_state], outputs=submit_outputs)
 
     export_pdf_btn.click(export_support_plan_pdf, inputs=[session_state], outputs=[support_plan_file])
     export_md_btn.click(export_support_plan_md, inputs=[session_state], outputs=[support_plan_file])

src/pipeline/core.py

@@ -18,7 +18,9 @@ from .output_guard import validate_output
 from .response_planner import (
     CLARIFY,
     INTERNATIONAL_SOURCE_HINT,
+    LISTEN,
     OFFER,
+    PERMISSION,
     build_response_plan,
     classify_intl_topic,
     decide_stage,
@@ -180,6 +182,19 @@ class EmpathRAGCore:
         # to widen things out), while an affirm after OFFER should go
         # into the consent flow (initial -> acknowledged -> delivered).
         self.session_last_stage: dict[str, str] = {}
+        # Count of "substantive" turns (LISTEN / PERMISSION / OFFER)
+        # rendered for this session. Non-substantive turns (greeting,
+        # goodbye, meta, minimal-affirm clarify, incomplete-message
+        # clarify) do not increment this. Used so a user who opens with
+        # "Hi" then says something real lands at LISTEN, not PERMISSION
+        # — the greeting shouldn't consume the listening-layer slot.
+        self.session_substantive_count: dict[str, int] = {}
+        # Monotonic per-session turn counter that does NOT cap. Used by
+        # the consent flow to measure "how many turns ago was the open
+        # offer rendered" — the tier_history-based turn_index saturates
+        # at 4 because tier_history is a rolling deque of size 3, which
+        # makes diff-based recency checks fail past turn 4.
+        self.session_seq: dict[str, int] = {}
         # Rolling per-session message log so the rephraser has continuity.
         # Only the last few user-assistant pairs are passed to the LLM; full
         # history is kept here for diagnostics. Each entry: {"role", "content"}.
@@ -194,6 +209,8 @@ class EmpathRAGCore:
             self.session_last_specific_route.pop(session_id, None)
             self.session_open_offer.pop(session_id, None)
             self.session_last_stage.pop(session_id, None)
+            self.session_substantive_count.pop(session_id, None)
+            self.session_seq.pop(session_id, None)
             self.session_message_history.pop(session_id, None)
         else:
             self.tier_history.clear()
@@ -203,6 +220,8 @@ class EmpathRAGCore:
             self.session_last_specific_route.clear()
             self.session_open_offer.clear()
             self.session_last_stage.clear()
+            self.session_substantive_count.clear()
+            self.session_seq.clear()
             self.session_message_history.clear()
 
     # Input length cap: defends against (1) accidental wall-of-text pastes
@@ -326,6 +345,12 @@ class EmpathRAGCore:
     ) -> _TurnPlan:
         if turn_index is None:
             turn_index = len(self.tier_history.get(session_id, [])) + 1
+        # Monotonic sequence — used wherever the diff-from-prior-turn
+        # comparison matters (consent recency, last-stage marker). The
+        # tier_history-based turn_index above caps because the safety
+        # tracker only keeps the last 3 entries.
+        self.session_seq[session_id] = self.session_seq.get(session_id, 0) + 1
+        seq = self.session_seq[session_id]
         t_total = time.perf_counter()
         latency: dict[str, float] = {}
 
@@ -450,7 +475,13 @@ class EmpathRAGCore:
                 audience_mode,
                 international_concern_override=intl_active,
             )
-            stage = decide_stage(message, route_label, safety_tier.value, turn_index)
+            # Stage progression should count *substantive* turns only.
+            # Greetings / goodbyes / meta / minimal-affirms / incomplete
+            # fragments do not consume the listening layer slot — a
+            # student who opens with "Hi" then says something real should
+            # land at LISTEN, not PERMISSION.
+            effective_turn_index = self.session_substantive_count.get(session_id, 0) + 1
+            stage = decide_stage(message, route_label, safety_tier.value, effective_turn_index)
             # Only classify F-1 sub-topic when the framing is actively in play.
             # After 2 silent turns, decayed; don't pin the planner to OPT/RCL/etc.
             intl_topic = classify_intl_topic(message) if intl_active else ""
@@ -492,7 +523,7 @@ class EmpathRAGCore:
                 minimal_kind == "affirm"
                 and self.session_last_stage.get(session_id) == "permission"
                 and not _open_offer_recent(
-                    self.session_open_offer.get(session_id), turn_index
+                    self.session_open_offer.get(session_id), seq
                 )
             ):
                 # The student said yes to widening out after a PERMISSION
@@ -505,7 +536,7 @@ class EmpathRAGCore:
                 recommended_action = response_plan.recommended_action
                 stage = OFFER
             elif minimal_kind == "affirm" and _open_offer_recent(
-                self.session_open_offer.get(session_id), turn_index
+                self.session_open_offer.get(session_id), seq
             ):
                 # The student just said yes/ok to an OFFER we made on a
                 # recent turn. Re-rendering the same template at this point
@@ -522,7 +553,7 @@ class EmpathRAGCore:
                         offer_state.get("question", ""),
                     )
                     offer_state["stage"] = "acknowledged"
-                    offer_state["acknowledged_turn"] = turn_index
+                    offer_state["acknowledged_turn"] = seq
                     self.session_open_offer[session_id] = offer_state
                     stage = OFFER
                 else:
@@ -530,7 +561,7 @@ class EmpathRAGCore:
                         offer_state.get("route", "")
                     )
                     offer_state["stage"] = "delivered"
-                    offer_state["delivered_turn"] = turn_index
+                    offer_state["delivered_turn"] = seq
                     self.session_open_offer[session_id] = offer_state
                     stage = OFFER
                 recommended_action = response_plan.recommended_action
@@ -560,8 +591,8 @@ class EmpathRAGCore:
             # would reset the consent progression.
             existing_offer = self.session_open_offer.get(session_id) or {}
             advanced_this_turn = (
-                existing_offer.get("acknowledged_turn") == turn_index
-                or existing_offer.get("delivered_turn") == turn_index
+                existing_offer.get("acknowledged_turn") == seq
+                or existing_offer.get("delivered_turn") == seq
             )
             if (
                 stage == OFFER
@@ -572,7 +603,7 @@ class EmpathRAGCore:
                 self.session_open_offer[session_id] = {
                     "route": route_label,
                     "question": response_plan.follow_up_question.strip(),
-                    "offer_turn": turn_index,
+                    "offer_turn": seq,
                     "stage": "initial",
                 }
             elif (
@@ -590,6 +621,15 @@ class EmpathRAGCore:
             # PERMISSION, consent-flow after OFFER).
             if not should_intercept:
                 self.session_last_stage[session_id] = stage
+                # Substantive stages (LISTEN / PERMISSION / OFFER) move
+                # the listening-layer counter forward. CLARIFY turns
+                # (greeting, goodbye, meta, minimal-affirm fallback,
+                # incomplete-message handler) do not — they're outside
+                # the listening loop's stage progression.
+                if stage in (LISTEN, PERMISSION, OFFER):
+                    self.session_substantive_count[session_id] = (
+                        self.session_substantive_count.get(session_id, 0) + 1
+                    )
 
         return _TurnPlan(
             message=message,