Procedure Catalog Extensibility¶

Procedure Onboarding Checklist (provider-independent)¶

When a new procedure type is onboarded (e.g., Pediatric PDA Closure, Breast Cancer Treatment, Liver Transplant), the work spans multiple subsystems that must be updated together. Provider-specific data (which hospitals offer it, costs, doctor lists) is a separate concern and lives in provider_procedures + Neo4j OFFERS. A procedure can exist in the catalog with zero providers — that's how we capture demand signal before sourcing a hospital.

This checklist runs before any provider can offer the procedure:

The hybrid catalog system designed below collapses items 1, 5, and parts of 4 into a single YAML edit + sync. Items 2, 3, 6, 7, 8 still need separate steps because they touch different stores or runtimes.

The provider-procedure linking flow (provider_procedures rows + Neo4j OFFERS relationships + cost data per provider) is a separate checklist that runs after this one.

Problem¶

The conversational agent has procedure-specific knowledge hardcoded in several places. Adding a new procedure category requires editing Python code and redeploying:

The data layer (Neo4j REQUIRES_TEST, Postgres procedure_requirements) is already extensible — new procedures can be seeded without code changes. But the conversational layer is not.

Goals¶

1. Operations team can add procedures without engineering involvement.
2. Configuration cannot drift — code, YAML, and database stay in sync.
3. Backwards compatible — existing seeded procedures keep working during migration.
4. Type-safe at runtime — validation catches schema errors before they reach a patient.
5. Auditable — every change is versioned and reviewable.

Design — Hybrid YAML + Database¶

The hybrid model uses YAML as the source of truth in development and Postgres as the runtime store in production, with bidirectional sync to prevent drift.

Layer 1: YAML config (dev / source control)¶

File: config/procedure_catalog.yaml

Single source of truth checked into git. Engineers and operations review changes via PR. CI validates the schema and runs sync checks.

version: 1

procedures:
  - code: "27447"
    cpt_code: "27447"
    icd10_codes: ["M17.11", "M17.12"]
    name: "Total Knee Replacement"
    slug: "total-knee-replacement"
    category: "orthopedic"
    keywords:
      - "total knee replacement"
      - "knee replacement"
      - "knee arthroplasty"
      - "tkr"
      - "tka"
    body_parts: ["knee"]
    pain_severity: "high"
    emotional_default: "discomfort"
    is_pediatric: false
    typical_records:
      - "Knee X-ray (AP/Lateral)"
      - "Knee MRI"
      - "CBC + metabolic panel"
    common_questions:
      - q: "How long is recovery?"
        a: "Typically 6-12 weeks for daily activities, 4-6 months for full recovery."

  - code: "C50"
    cpt_code: null
    icd10_codes: ["C50.0", "C50.9"]
    name: "Breast Cancer Treatment"
    slug: "breast-cancer-treatment"
    category: "oncology"
    keywords: ["breast cancer", "mastectomy", "lumpectomy"]
    body_parts: ["breast"]
    pain_severity: "moderate"
    emotional_default: "severe_diagnosis"  # auto-elevates tone
    is_pediatric: false
    typical_records:
      - "Mammogram"
      - "Biopsy report"
      - "Pathology report"
      - "Treatment plan from oncologist"

Layer 2: Postgres table (production runtime)¶

Table: procedure_catalog

Mirrors the YAML schema. Production reads from this table at runtime. Operations team can edit via an admin UI without touching git.

CREATE TABLE procedure_catalog (
    id              UUID PRIMARY KEY,
    code            VARCHAR(20) UNIQUE NOT NULL,
    cpt_code        VARCHAR(10),
    icd10_codes     JSONB,
    name            VARCHAR(255) NOT NULL,
    slug            VARCHAR(255) UNIQUE NOT NULL,
    category        VARCHAR(50) NOT NULL,
    keywords        JSONB NOT NULL,
    body_parts      JSONB NOT NULL,
    pain_severity   VARCHAR(20),
    emotional_default VARCHAR(50),
    is_pediatric    BOOLEAN DEFAULT FALSE,
    typical_records JSONB,
    common_questions JSONB,
    yaml_hash       VARCHAR(64),  -- SHA256 of YAML row, for sync detection
    source          VARCHAR(20) NOT NULL,  -- "yaml" | "admin_ui"
    created_at      TIMESTAMP WITH TIME ZONE NOT NULL,
    updated_at      TIMESTAMP WITH TIME ZONE NOT NULL,
    last_synced_at  TIMESTAMP WITH TIME ZONE
);

CREATE INDEX idx_procedure_catalog_keywords ON procedure_catalog USING gin (keywords);
CREATE INDEX idx_procedure_catalog_category ON procedure_catalog (category);

Layer 3: Loader service¶

File: app/services/procedure_catalog.py

class ProcedureCatalog:
    """Loads procedure definitions from Postgres (with YAML fallback)."""

    def __init__(self):
        self._cache: dict[str, ProcedureDef] | None = None
        self._cache_loaded_at: float = 0
        self._cache_ttl = 300  # 5 minutes

    async def get_all(self, db: AsyncSession) -> list[ProcedureDef]:
        """Returns all procedures, loading from DB on cache miss."""
        ...

    async def match_keyword(self, db: AsyncSession, message: str) -> ProcedureDef | None:
        """Returns the first procedure whose keyword matches (word-boundary aware)."""
        ...

    async def get_by_code(self, db: AsyncSession, code: str) -> ProcedureDef | None:
        ...

    async def get_emotional_default(self, db: AsyncSession, code: str) -> str | None:
        """Returns the procedure's default emotional state (e.g. 'severe_diagnosis')."""
        ...

    def invalidate_cache(self):
        self._cache = None

The loader supports a graceful degradation chain:

1. Try Postgres procedure_catalog table (production)
2. Fallback to config/procedure_catalog.yaml (dev or empty DB)
3. Fallback to current hardcoded PROCEDURE_CODES dict (legacy code paths)

This means we can roll out the new system gradually — the agent works at every step.

Sync Strategy (the hard part)¶

The challenge with hybrid approaches is keeping YAML and DB in sync. Here's how we prevent drift:

Sync direction 1: YAML → Postgres (deploy time)¶

Trigger: every Railway deploy where config/procedure_catalog.yaml changed.

Mechanism: post-deploy hook runs python -m app.sync_procedure_catalog.

What it does:

1. Parse procedure_catalog.yaml
2. For each YAML row:
3. Compute SHA256 of the row's content → yaml_hash
4. Look up DB row by code
5. If DB row missing → INSERT with source="yaml", set yaml_hash
6. If DB row exists with source="yaml" AND yaml_hash differs → UPDATE (YAML is source of truth for yaml-sourced rows)
7. If DB row exists with source="admin_ui" → SKIP and log a warning ("admin override exists for code X")
8. For DB rows where source="yaml" but the code is NOT in current YAML → mark deleted_at (soft delete; never hard delete production data)
9. Write a sync report to events table

Sync direction 2: Postgres → YAML (export workflow)¶

Trigger: manual command after operations team makes changes via admin UI.

Mechanism: python -m app.export_procedure_catalog > config/procedure_catalog.yaml.new

What it does:

1. SELECT all rows from procedure_catalog
2. Convert to YAML format
3. Write to procedure_catalog.yaml.new
4. Engineer reviews diff, commits to git
5. Next deploy triggers Sync direction 1, which now sees yaml_hash matches

This makes the round trip explicit — admin UI changes don't silently diverge from git. Operations + engineering coordinate via PRs.

Conflict resolution¶

The source column is the key — it tells us who "owns" the row. Yaml rows are eng-owned; admin_ui rows are ops-owned.

CI Validation¶

Before any YAML change reaches production, CI runs:

# tests/test_procedure_catalog_yaml.py

def test_yaml_loads():
    catalog = load_procedure_catalog_yaml()
    assert catalog.version == 1

def test_no_duplicate_codes():
    codes = [p.code for p in catalog.procedures]
    assert len(codes) == len(set(codes))

def test_no_duplicate_keywords():
    seen = {}
    for p in catalog.procedures:
        for kw in p.keywords:
            assert kw not in seen, f"Duplicate keyword: {kw} in {p.code} and {seen[kw]}"
            seen[kw] = p.code

def test_required_fields_present():
    for p in catalog.procedures:
        assert p.name and p.code and p.keywords and p.category

def test_emotional_default_is_known_state():
    valid = {"neutral", "discomfort", "anxious", "frustrated", "hopeful",
             "grieving", "severe_diagnosis", "pediatric_caregiver"}
    for p in catalog.procedures:
        if p.emotional_default:
            assert p.emotional_default in valid

def test_keywords_are_safe_for_word_boundary_match():
    """No 3-char keywords (false positive risk)."""
    for p in catalog.procedures:
        for kw in p.keywords:
            assert len(kw) >= 4 or kw.isupper(), f"Short keyword: {kw}"

def test_yaml_matches_db_state():
    """If running against staging/prod DB, verify yaml_hash matches."""
    # Skipped in dev, runs in CI deploy job
    ...

Migration Plan¶

Phase A: YAML loader, hardcoded fallback (Sprint 1)¶

1. Create config/procedure_catalog.yaml with all 14 current procedures
2. emotional defaults
3. Create app/services/procedure_catalog.py loader with file-based reading
4. Refactor _handle_procedure_identification to call PROCEDURE_CATALOG.match_keyword() instead of inline dict
5. Keep the inline dict as a hardcoded fallback (deleted in Phase B)
6. CI tests validate the YAML schema
7. No DB changes yet — pure YAML refactor
8. Deploy → verify same behavior

Phase B: Postgres table + sync (Sprint 2)¶

1. Alembic migration: create procedure_catalog table
2. python -m app.sync_procedure_catalog script that does YAML → DB
3. Loader updated to read from DB first, YAML fallback
4. Run sync as part of Railway deploy hook
5. Delete the hardcoded inline dict
6. Deploy → verify same behavior

Phase C: Admin UI (post-MVP)¶

1. Admin endpoint: GET/POST/PATCH/DELETE /api/v1/admin/procedure-catalog
2. Operations dashboard page for catalog editing
3. python -m app.export_procedure_catalog for the round-trip workflow
4. Audit log on every admin edit (existing pattern)

Emotional Patterns — Why NOT Hybrid¶

The same hybrid pattern would technically work for emotional state regex patterns, but I recommend keeping them in code for now:

• They change less often than procedures (we've added 1 state in 6 months)
• They require careful tuning against the LLM (regex + prompt + tests)
• Engineering review is essential — operations team shouldn't tune EQ rules
• The 6 current states cover most patient types

If we eventually need procedure-specific emotional defaults (e.g., "all oncology procedures default to severe_diagnosis"), the procedure catalog's emotional_default field handles that without exposing the regex layer.

Cost & Risk Analysis¶

Phase A (YAML only)¶

• Effort: ~3 hours
• Risk: Low — refactor of existing keyword match, easy to verify
• Reversible: Yes (delete one file, revert one function)
• Unblocks: Adding procedures via PR review without touching Python

Phase B (DB sync)¶

• Effort: ~6 hours (migration + sync script + tests)
• Risk: Medium — sync logic has edge cases (drift, conflicts)
• Reversible: Yes (drop table, fall back to YAML)
• Unblocks: Production procedures editable without redeploy

Phase C (Admin UI)¶

• Effort: ~8 hours (UI + endpoints + audit + export script)
• Risk: Low if Phases A+B are solid
• Reversible: Yes (disable feature flag)
• Unblocks: Operations team self-service

Open Questions¶

1. Where does Neo4j fit? Procedures already exist as Neo4j nodes (Procedure {code, name, slug}). Should the catalog be the source of truth for Neo4j too, or are they independent? Recommendation: make the catalog the source of truth, sync to Neo4j on insert/update.
2. What about translations? A patient in UAE might say "ركبتي تؤلمني" instead of "my knee hurts". Should keywords be multilingual? Recommendation: start English-only, add a keywords_locale: {ar: [...]} field in Phase C.
3. Should the keyword matcher be moved to LLM-only? With Claude Haiku handling procedure identification, the keyword fast-path saves 3-5s but adds maintenance burden. Recommendation: keep the keyword fast-path but ALSO send the catalog to the LLM as context, so it can match procedures the keyword list misses.

Acceptance Criteria¶

Phase A is done when:

• All 14 current procedures move to procedure_catalog.yaml
• case_orchestrator.py reads from the loader, not the inline dict
• All existing tests pass (pytest tests/test_agent_pipeline.py)
• New test tests/test_procedure_catalog_yaml.py validates schema
• Adding a 15th procedure requires only YAML edit (verified by adding a test procedure and verifying the agent identifies it)

Phase B is done when:

• DB table exists, YAML rows are synced on deploy
• Manual edits via admin endpoint persist + survive restart
• python -m app.export_procedure_catalog round-trip produces a clean diff
• Sync conflict scenarios all handled (yaml→db, db→yaml, both→both)

Status¶