Skip to content

Session N: Conversation Flow Gates — Feature Spec

Context

You are working on the Curaway backend (curaway_src repo). This is the implementation spec for Layer 1 of the conversation flow remediation plan. Read the steer doc first: docs/specs/ai-steer/conversation-flow-gates-steer.md

Also read CLAUDE.md for project conventions.

Branch

git checkout main && git pull origin main
git checkout -b fix/conversation-flow-gates-v2

What you're building

A surgical relaxation of the workflow gates so most conversations actually reach the matching stage. No new orchestrator architecture — just the gates, the intake completion criteria, and a patient-driven explicit-advance affordance, all behind a gates_v2 Flagsmith flag.

PART A — config/guardrails.yaml

Update the gating section. Three new keys, three existing values lowered:

gating:
  ehr:
    # Single matching gate. Match when EHR is at least this complete.
    completeness_for_matching: 0.4   # was 0.5
    completeness_with_meds: 0.4       # unchanged (now == main gate)
    matching_ready_threshold: 0.4     # unchanged

  matching:
    require_medications: true
    # NEW: when intake_complete is true, ALWAYS allow matching even if the
    # completeness score is below the threshold
    allow_when_intake_complete: true

  intake:
    min_answers_for_completion: 3
    # NEW: phrases the patient can use to skip remaining intake and jump
    # to matching
    explicit_advance_phrases:
      - "find providers now"
      - "show me providers"
      - "i'm ready"
      - "im ready"
      - "skip the rest"
      - "proceed"
      - "advance"
      - "let's match"
      - "lets match"

PART B — app/services/gating_config.py

Update the _DEFAULTS dict to mirror the new YAML shape (so the file works when YAML is missing). Also add a small helper:

def is_explicit_advance_phrase(message: str) -> bool:
    """Word-boundary check if the patient's message contains a phrase that
    should override remaining intake gating and route directly to matching.
    """
    import re
    msg = (message or "").lower()
    phrases = GATING.get("intake", {}).get("explicit_advance_phrases", [])
    for p in phrases:
        if re.search(r"\b" + re.escape(p.lower()) + r"\b", msg):
            return True
    return False

PART C — app/services/feature_flags.py

Make sure gates_v2 reads from Flagsmith and defaults to True when Flagsmith is unreachable. Look for the existing pattern for similar feature flags (e.g. enable_chat_cache) and follow it.

If there isn't already a helper, add:

def is_gates_v2_enabled() -> bool:
    return is_feature_enabled("gates_v2", default=True)

PART D — app/agents/case_orchestrator.py

D1. New helper: _intake_complete_v2

Insert near the existing _handle_intake helper:

def _intake_complete_v2(case) -> tuple[bool, list[str]]:
    """Layer-1 intake completion check.

    Returns (is_complete, met_criteria). The five rules are:
      1. procedure identified
      2. at least one analyzed doc OR >= min_answers_for_completion turns
      3. age + country known
      4. medications confirmed (any list) OR medications_confirmed_none
      5. allergies confirmed (any list)   OR allergies_confirmed_none
    """
    from app.services.gating_config import GATING

    met: list[str] = []
    ws = case.workflow_state or {}
    meta = case.extra_metadata or {}
    ehr = case.ehr_snapshot or {}
    demo = (ehr.get("patient_demographics") or {}) if isinstance(ehr, dict) else {}
    history = (ehr.get("medical_history") or {}) if isinstance(ehr, dict) else {}

    if case.procedure_code:
        met.append("procedure")

    analyzed_docs = ws.get("analyzed_doc_count", 0)
    min_answers = GATING.get("intake", {}).get("min_answers_for_completion", 3)
    answer_count = meta.get("intake_answer_count", 0)
    if analyzed_docs > 0 or answer_count >= min_answers:
        met.append("evidence")

    has_age = bool(demo.get("age") or meta.get("patient_age"))
    has_country = bool(
        demo.get("country") or meta.get("patient_country")
    )
    if has_age and has_country:
        met.append("demographics")

    meds = (history.get("medications") or []) or meta.get("medications") or []
    if meds or meta.get("medications_confirmed_none"):
        met.append("medications")

    allergies = (history.get("allergies") or []) or meta.get("allergies") or []
    if allergies or meta.get("allergies_confirmed_none"):
        met.append("allergies")

    return (len(met) == 5, met)

D2. Wire _intake_complete_v2 into the existing intake gate

In handle_message, find the block (currently around line 300) that gates on ws.get("intake_complete"). Add an early check before that block:

from app.services.feature_flags import is_gates_v2_enabled
from app.services.gating_config import is_explicit_advance_phrase

if is_gates_v2_enabled():
    # Explicit-advance affordance — patient asks to skip ahead
    if is_explicit_advance_phrase(message):
        await _record_explicit_advance(db, case)
        # Promote skip flags so downstream gates accept
        meta = dict(case.extra_metadata or {})
        meta.setdefault("medications_confirmed_none", True)
        meta.setdefault("allergies_confirmed_none", True)
        case.extra_metadata = meta
        ws_update = dict(case.workflow_state or {})
        ws_update["intake_complete"] = True
        case.workflow_state = ws_update
        await db.flush()
        # Re-fetch ws so the rest of handle_message sees the new flags
        ws = ws_update
        # Fall through — next branch will route to matching

    # If v2 says intake is complete, set the flag so the existing
    # branch logic short-circuits in the right place
    if not ws.get("intake_complete"):
        complete, met = _intake_complete_v2(case)
        if complete:
            ws_update = dict(case.workflow_state or {})
            ws_update["intake_complete"] = True
            case.workflow_state = ws_update
            await db.flush()
            ws = ws_update
            await _record_intake_complete(db, case, met)

D3. Lower the matching gate

In the same handle_message, find the min_info_for_matching block (currently around line 434). Replace with:

has_meds = ws.get("medications_asked", False) or bool((case.extra_metadata or {}).get("medications"))
_ehr_gate = GATING["ehr"]["completeness_for_matching"]
_ehr_with_meds = GATING["ehr"]["completeness_with_meds"]
allow_intake = (
    is_gates_v2_enabled()
    and GATING.get("matching", {}).get("allow_when_intake_complete", True)
    and ws.get("intake_complete", False)
)

min_info_for_matching = (
    allow_intake
    or ehr_completeness >= _ehr_gate
    or (ehr_completeness >= _ehr_with_meds and has_meds)
)

D4. Track answer count

Inside _handle_intake, after each substantive answer is processed, increment meta["intake_answer_count"]:

meta = dict(case.extra_metadata or {})
meta["intake_answer_count"] = meta.get("intake_answer_count", 0) + 1
case.extra_metadata = meta

A "substantive answer" is any user message that is at least 5 characters long and not in the explicit_advance_phrases list. Don't double-count.

D5. New helpers _record_explicit_advance and _record_intake_complete

Both write to the events table via the existing decision_recorder if available; otherwise log + skip silently. Pattern:

async def _record_explicit_advance(db, case):
    try:
        from app.services.decision_recorder import record_decision
        await record_decision(db, case, decision_type="gates_v2_explicit_advance",
                              payload={"case_id": str(case.id)})
    except Exception:
        logger.debug("decision_recorder unavailable, skipping")

async def _record_intake_complete(db, case, met_criteria):
    try:
        from app.services.decision_recorder import record_decision
        await record_decision(db, case, decision_type="gates_v2_intake_complete",
                              payload={"case_id": str(case.id), "met_criteria": met_criteria})
    except Exception:
        logger.debug("decision_recorder unavailable, skipping")

D6. Records-first early termination — only count answers after the patient actually answers

In the records-first branch (currently around line 361-404), the quick_questions_asked flag flips regardless of whether the patient actually answered. Move that flag flip into the _handle_intake branch that processes substantive answers, not the branch that asks the questions for the first time.

PART E — Tests

E1. tests/test_gating_config.py

Add tests for the new keys:

def test_explicit_advance_phrase_matching():
    from app.services.gating_config import is_explicit_advance_phrase
    assert is_explicit_advance_phrase("find providers now")
    assert is_explicit_advance_phrase("I'm ready to see matches")
    assert is_explicit_advance_phrase("can we proceed?")
    assert not is_explicit_advance_phrase("I have more records")
    assert not is_explicit_advance_phrase("let's read the documentation")  # 'read' should not match 'ready'

E2. tests/test_intake_complete_v2.py

def test_empty_case_not_complete()
def test_procedure_only_not_complete()
def test_procedure_plus_doc_plus_demographics_plus_skips_complete()
def test_procedure_plus_3_answers_plus_demographics_plus_meds_plus_allergies_complete()
def test_explicit_skip_meds_counts()
def test_explicit_skip_allergies_counts()
def test_age_without_country_incomplete()
def test_country_without_age_incomplete()
def test_real_world_uploaded_doc_plus_quick_answers_complete()

Each should construct a fake case object with workflow_state, extra_metadata, ehr_snapshot, procedure_code, and assert _intake_complete_v2 returns the expected (bool, list) shape.

E3. tests/test_orchestrator_gates_v2.py

End-to-end orchestrator test using the in-memory SQLite test fixture (see tests/conftest.py). Three cases:

  1. Patient says "find providers now" → orchestrator routes to matching
  2. Patient uploads 1 doc + answers 3 questions → intake_complete fires → matching is triggered next turn
  3. Patient with no docs and 1 answer → intake_complete is False → orchestrator continues asking

PART F — Verification

After implementing, run:

pytest tests/test_gating_config.py tests/test_intake_complete_v2.py \
       tests/test_orchestrator_gates_v2.py -v

All should pass. Then run the full backend suite to make sure nothing upstream broke:

pytest -q

PART G — Commit + PR + merge

Single PR titled fix(orchestrator): lower conversation gates v2. Body must reference the steer doc and the bug report (/app/case/e241e34f-6ef8-4eee-8d54-cbb02451100c). Squash-merge to main once tests pass.

FINAL CHECKLIST

  • [ ] gating.intake.explicit_advance_phrases added to YAML
  • [ ] gating.matching.allow_when_intake_complete added to YAML
  • [ ] completeness_for_matching lowered from 0.5 → 0.4
  • [ ] _DEFAULTS in gating_config.py mirrors the new YAML
  • [ ] is_explicit_advance_phrase() helper added
  • [ ] is_gates_v2_enabled() helper added (defaults true)
  • [ ] _intake_complete_v2() helper added in orchestrator
  • [ ] Orchestrator early-check sets intake_complete=True when v2 says so
  • [ ] Explicit-advance phrase routes to matching with skip flags set
  • [ ] Matching gate accepts intake_complete regardless of completeness score
  • [ ] intake_answer_count is incremented per substantive answer
  • [ ] quick_questions_asked flag only flips after substantive answers, not after the first time the questions are asked
  • [ ] decision_recorder writes new event types
  • [ ] All new tests pass + full backend suite green
  • [ ] PR merged + Flagsmith flag created with default true