Merged in bugfix/molina_ut_dynamic_primary (pull request #991)

Bugfix/molina ut dynamic primary

* Prompt changes for dynamic primary

* Route cover-sheet-only files to ERRORS.csv instead of leaking phantom rows

When every page of a contract was filtered out as a cover sheet / quick-review
form, process_file silently returned a FILE_NAME-only DataFrame. Because the
runner routes by checking for an "error" column, that file landed in
RESULTS.csv as a near-empty row and no ERRORS.csv was generated for the run.

- saas/file_processing.py: raise ValueError when text_dict is empty after
  cover-sheet filtering, so safe_process_file produces a proper error row.
- runner.py: add _is_phantom_result defense-in-depth — promote any result
  with no extracted fields beyond FILE_NAME to error_results with
  error_type=PhantomSuccess.

* Merged dev into bugfix/molina_ut_dynamic_primary

* Tighten PRODUCT prompt: restrict to valid_values, prune LOB/PROGRAM examples

* Merge branch 'bugfix/molina_ut_dynamic_primary' of bitbucket.org:aarete/doczy.ai into bugfix/molina_ut_dynamic_primary

* Add PROVIDER_NAME and DISCOUNT_TERM to FIELD_FORMAT_MAPPING

* Revert "Route cover-sheet-only files to ERRORS.csv instead of leaking phantom rows"

This reverts commit de79a1391511acaf78c92a6bee53697f419d3996.

* Drop internal-only columns from final output

Four columns were leaking into RESULTS.csv that have no entry in
FIELD_FORMAT_MAPPING and shouldn't ship to production:
  - DOCUMENT_TYPE: from the document classification stage, not a contract field
  - REIMB_PAGE / EXHIBIT_TEXT: internal extraction-trace columns
  - CODE_MAPPING_SOURCE: debug column

Drop them in standard_postprocess just before reorder_columns. Done after
attach_sid_column so AARETE_DERIVED_SID (which reads DOCUMENT_TYPE) still
gets populated correctly. Uses errors='ignore' so the drop is a no-op
when a column isn't present.

* Strict-filter columns to FIELD_FORMAT_MAPPING and tighten PRODUCT prompt

* Merged dev into bugfix/molina_ut_dynamic_primary

* Dedupe values within LOB/PROGRAM/PRODUCT/NETWORK fields

* Move hard-codes to within function


Approved-by: Katon Minhas

.gitignore

@@ -119,3 +119,6 @@ TEST*.xml
 
 # Local code index
 .claude/index/
+
+# Local output / offline study artifacts
+out/

src/constants/investment_columns.py

@@ -39,6 +39,7 @@ FIELD_FORMAT_MAPPING = {
     "PROV_GROUP_TIN": "list[str]",
     "PROV_GROUP_NPI": "list[str]",
     "PROV_GROUP_NAME_FULL": "list[str]",
+    "PROVIDER_NAME": "str",
     "AARETE_DERIVED_PROVIDER_NAME": "str",
     "PROV_OTHER_TIN": "list[str]",
     "PROV_OTHER_NPI": "list[str]",
@@ -165,6 +166,7 @@ FIELD_FORMAT_MAPPING = {
     "STOP_LOSS_PCT_RATE_ON_EXCESS_CHARGES": "str",
     "STOP_LOSS_EXCLUSION_CD": "list[str]",
     "STOP_LOSS_EXCLUSION_CD_DESC": "list[str]",
+    "DISCOUNT_TERM": "str",
     "DISCOUNT_PERCENT_RATE": "str",
     "DISCOUNT_RATE_CHANGE_INTERVAL": "str",
     "DISCOUNT_START_DT": "str",

src/pipelines/shared/postprocessing/postprocess.py

@@ -148,7 +148,11 @@ def standard_postprocess(df, constants: Constants):
     # Remove N/A values from all columns (standard cleaning for all outputs)
     df = postprocessing_funcs.clean_na_values(df)
 
+    # Dedupe values within LOB / PROGRAM / PRODUCT / NETWORK (and their AARETE_DERIVED_*).
+    df = postprocessing_funcs.deduplicate_dynamic_primary_fields(df)
+
     # Standardize output column order - this should ALWAYS be the final postprocessing step
+    # reorder_columns also drops any column not in FIELD_FORMAT_MAPPING.
     df = postprocessing_funcs.reorder_columns(df, FIELD_FORMAT_MAPPING)
     return df
 

src/pipelines/shared/postprocessing/postprocessing_funcs.py

@@ -483,7 +483,7 @@ def reorder_columns(
     Steps:
     1. Adds any missing columns from `column_order` to the DataFrame at once, filled with empty strings.
     2. Reorders the columns of the DataFrame to match `column_order`.
-    3. Appends any columns in the DataFrame that are not in `column_order` to the end.
+    3. Drops any columns in the DataFrame that are not in `column_order`.
 
     Args:
         df (pd.DataFrame): The input DataFrame.
@@ -510,13 +510,9 @@ def reorder_columns(
             [df_copy, pd.DataFrame(empty_cols, index=df_copy.index)], axis=1
         )
 
-    # Reorder columns to match column_order
+    # Keep only columns in column_order; drop any extras not in the mapping
     final_column_order = [col for col in column_order if col in df_copy.columns]
 
-    # Append any extra columns not in column_order (e.g. client-specific fields)
-    extra_columns = [col for col in df_copy.columns if col not in column_order]
-    final_column_order.extend(extra_columns)
-
     # Return the DataFrame with reordered columns
     return df_copy[final_column_order]
 
@@ -1182,6 +1178,88 @@ def _strip_wrapping_single_quotes(s: str) -> str:
     return s
 
 
+
+
+
+def _dedupe_list_value(val):
+    """Deduplicate values within a single cell, preserving first-seen order.
+
+    Handles list cells, JSON-array string cells, and pipe-delimited strings.
+    Scalar/empty values are returned unchanged. Comparison is case-insensitive
+    on the stripped string form so "Medicare" and "medicare " collapse.
+    """
+    if val is None or (isinstance(val, float) and pd.isna(val)):
+        return val
+
+    items = None
+    output_kind = None  # "list" | "json" | "pipe"
+
+    if isinstance(val, list):
+        items = val
+        output_kind = "list"
+    elif isinstance(val, str):
+        text = val.strip()
+        if text.startswith("[") and text.endswith("]"):
+            try:
+                parsed = json.loads(text)
+                if isinstance(parsed, list):
+                    items = parsed
+                    output_kind = "json"
+            except (json.JSONDecodeError, ValueError):
+                return val
+        elif "|" in text:
+            items = [p for p in text.split("|")]
+            output_kind = "pipe"
+        else:
+            return val
+    else:
+        return val
+
+    seen = set()
+    deduped = []
+    for item in items:
+        if item is None:
+            continue
+        key = str(item).strip().lower()
+        if not key or key in seen:
+            continue
+        seen.add(key)
+        deduped.append(item if isinstance(item, str) else str(item))
+
+    if output_kind == "list":
+        return deduped
+    if output_kind == "json":
+        return json.dumps(deduped)
+    return "|".join(deduped)
+
+
+# HOTFIX: remove once upstream dedupes LOB/PROGRAM/PRODUCT/NETWORK at source.
+# Duplicates in these fields are a contract violation in dynamic_assignment /
+# fill_na_mapping / merge_one_to_one_into_one_to_n; this is a postprocess
+# bandaid so output is clean while the real fix is scoped.
+def deduplicate_dynamic_primary_fields(df):
+    """Deduplicate values inside LOB / PROGRAM / PRODUCT / NETWORK list cells.
+
+    Applies to both the raw and AARETE_DERIVED variants. Order-preserving
+    (case-insensitive). No-op when a column is missing.
+    """
+    DYNAMIC_PRIMARY_DEDUP_FIELDS = [
+        "LOB",
+        "PROGRAM",
+        "PRODUCT",
+        "NETWORK",
+        "AARETE_DERIVED_LOB",
+        "AARETE_DERIVED_PROGRAM",
+        "AARETE_DERIVED_PRODUCT",
+        "AARETE_DERIVED_NETWORK",
+    ]
+
+    for col in DYNAMIC_PRIMARY_DEDUP_FIELDS:
+        if col in df.columns:
+            df[col] = df[col].apply(_dedupe_list_value)
+    return df
+
+
 def format_as_json_list(val):
     """
     Format a value as a JSON list string. Returns blank for empty lists

src/prompts/investment_prompts.json

@@ -131,7 +131,7 @@
         "field_name": "PRODUCT",
         "relationship": "one_to_n",
         "field_type": "dynamic_primary",
-        "prompt": "The brand name of a health plan or product offered by the Payer. Do NOT include the name of the Payer itself (such as 'Molina', 'Centene' etc) - rather, look for the Product that these healthplans offer. Valid products include, but are not limited to, the following: {valid_values}. If any of the values in this list are mentioned, return them exactly as written in the list. Use reasonable judgment for obvious synonyms and equivalent terms.",
+        "prompt": "The brand name of a health plan or product offered by the Payer. Choose Products ONLY from the following values: {valid_values}. Choose ONLY from the valid values presented; if no value from the list is explicitly mentioned, return 'N/A'. Do NOT include the name of the Payer itself (such as 'Molina', 'Centene' etc) - rather, look for the Product that these healthplans offer. Do NOT extract LOBs (such as 'Medicaid', 'Medicare', 'Marketplace', 'Duals') or Programs (such as 'Medicaid Managed Care', 'Medicare Advantage', 'ACA') as products - those belong in the LOB and PROGRAM fields respectively. Return values exactly as written in the valid values list.",
         "valid_values": "VALID_PRODUCTS",
         "retrieval_question": "What Product does this contract apply to?"
     },