Design for AP AutoExtract to Fluxx Extraction requirements

Activities (high level)

generate_invoice_id

Purpose: Create canonical invoiceID (content hash of PDF) and initialize invoice document record with initial stage/status/stepper/resolutions.
Inputs: pdf_path, client_id, client_name, workflow_id, pdf_prefix.
External calls: Blob download (get_blob_content), client config fetch (get_client), Cosmos insert (store_document).
Outputs/State: Sets data.invoiceID, data.invoice_document, workflow_status/substatus.
Idempotency: Hash is deterministic; re-run should detect existing record if logic added.

generate_images

Purpose: Produce per-page JPEG images from source PDF and store in blob; persist image URLs.
Inputs: invoiceID (from prior), pdf_path (unless pre-generated image_urls provided), pdf_prefix.
External calls: Blob get (PDF), blob upload (each page), SAS URL generation, Cosmos update (image_urls, substatus).
Outputs: data.image_urls, data.image_prefixes, invoice document updated.
Optimizations: Skips work if images already present.
Constraints: Page count limited by config.MAX_PAGES.

get_bounding_regions

Purpose: Run Document Intelligence (AI extraction) to produce structured JSON (layout / bounding regions) and persist JSON blob reference.
Inputs: pdf_path, pdf_prefix, client_id, invoiceID.
External calls: SAS URL gen (optional), analyze_pdf(), blob_exists(), blob upload for DI JSON, Cosmos update (di_json_url).
Outputs: di_json_url, di_json_prefix in state.
Idempotency: Skips re-analysis if JSON blob already exists.

extract_invoice

Purpose: AI (OpenAIExtractor) semantic extraction (key header fields, line items, totals) from PDF + DI rotations.
Inputs: pdf_path, di_json_prefix, client_config.
External calls: Blob get (PDF + DI JSON), OpenAIExtractor invocation.
Outputs: ai_extracted_content dict stored in state.
Fallback: On extractor failure returns empty dict (allows downstream transform path).

map_invoice

Purpose: Convert AI extracted raw dict into internal Invoice model (simple mapping only AI output).
Inputs: ai_extracted_content, invoiceID.
External calls: Cosmos update (invoice field).
Outputs: invoice JSON (only AI mapping), updates invoice_document.

transform_invoice

Purpose: Parse DI JSON to build Invoice model; optionally merge with AI extracted data (combine header totals & line items).
Inputs: di_json_url, ai_extracted_content (optional), invoiceID.
External calls: Blob get (DI JSON), combine/transform functions, Cosmos update (invoice).
Merge Logic: Prefers AI-derived financial fields & PO number if DI missing; fuzzy merging of line items.

post_processing

Purpose: Apply client-specific normalization rules (freight/tariff redistribution), adjust invoice subtotal/shipping/tariff amounts, optionally push to mocked ERP.
Inputs: invoice document (with lineItems), client_config.ingest_config patterns.
External calls: Cosmos update (adjusted invoice), HTTP POST to mocked ERP if provider == ‘cla-mocked’.
Logic: Detect freight/tariff line items via vendor-specific patterns, move amounts to invoice-level fields, remove matched line items.

check_duplicate_document

Purpose: Detect potential duplicate invoice based on configured unique_identifier_fields.
Inputs: invoice document, ingest_config.unique_identifier_fields.
External calls: Cosmos query (via DatastoreInterface), Cosmos update (duplicate_invoice if found).
Outputs: Status failed + substatus if duplicate; else success.

Example workflow configuration JSON with per‑activity enable/disable, ordering, retry, skip logic, and per‑activity parameters:

{
  "version": "1.0",
  "workflow_name": "invoice_ingest",
  "defaults": {
    "retry_policy": { "max_attempts": 3, "initial_delay_ms": 1000, "backoff_factor": 2.0, "jitter": true },
    "timeout_ms": 60000,
    "failure_mode": "halt"  // halt | continue
  },
  "execution": {
    "order": [
      "generate_invoice_id",
      "check_duplicate_document",
      "generate_images",
      "get_bounding_regions",
      "extract_invoice",
      "map_invoice",
      "transform_invoice",
      "post_processing"
    ]
  },
  "activities": {
    "generate_invoice_id": {
      "enabled": true,
      "timeout_ms": 15000,
      "retry_policy": { "max_attempts": 2 },
      "skip_if": "state.invoiceID EXISTS",
      "mutations": {
        "invoice_document.substatus": "Generating Invoice ID"
      },
      "outputs_map": {
        "invoice_id": "$.result.invoiceID"
      },
      "required_inputs": ["pdf_path", "client_id", "client_name", "pdf_prefix"]
    },
    "check_duplicate_document": {
      "enabled": true,
      "depends_on": ["generate_invoice_id"],
      "skip_if": "config.ingest_config.unique_identifier_fields LENGTH == 0",
      "failure_behavior": "continue",  // continue even if duplicate (marks status internally)
      "mutations": {
        "invoice_document.substatus": "Duplicate Check"
      }
    },
    "generate_images": {
      "enabled": true,
      "depends_on": ["generate_invoice_id"],
      "skip_if": "state.image_urls EXISTS",
      "retry_policy": { "max_attempts": 3, "initial_delay_ms": 1500 },
      "mutations": {
        "invoice_document.substatus": "Generating Invoice Images"
      },
      "params": {
        "output_folder_pattern": "jpg/{invoice_id}/page{page_number}/{pdf_prefix}-page-{page_number}.jpg",
        "max_pages": "${config.MAX_PAGES}"
      }
    },
    "get_bounding_regions": {
      "enabled": true,
      "depends_on": ["generate_images"],
      "skip_if": "state.di_json_url EXISTS",
      "timeout_ms": 90000,
      "retry_policy": { "max_attempts": 4, "initial_delay_ms": 2000 },
      "mutations": {
        "invoice_document.substatus": "Getting Bounding Regions Data"
      },
      "params": {
        "blob_path_pattern": "json/{client_id}/{invoice_id}/{pdf_prefix}.json"
      }
    },
    "extract_invoice": {
      "enabled": true,
      "depends_on": ["get_bounding_regions"],
      "skip_if": "state.ai_extracted_content EXISTS",
      "mutations": {
        "invoice_document.substatus": "Extracting Invoice Data"
      },
      "params": {
        "fallback_on_error": true
      }
    },
    "map_invoice": {
      "enabled": true,
      "depends_on": ["extract_invoice"],
      "skip_if": "NOT state.ai_extracted_content",
      "parallel_group": "mapping_phase",
      "mutations": {
        "invoice_document.substatus": "Map Invoice Data"
      }
    },
    "transform_invoice": {
      "enabled": true,
      "depends_on": ["extract_invoice"],
      "parallel_group": "mapping_phase",
      "mutations": {
        "invoice_document.substatus": "Transform Invoice Data"
      },
      "params": {
        "prefer_ai_financials": true,
        "fuzzy_merge_threshold": 90
      }
    },
    "post_processing": {
      "enabled": true,
      "depends_on": ["map_invoice", "transform_invoice"],
      "wait_for_parallel_group": "mapping_phase",
      "skip_if": "state.invoice_document.invoice.lineItems LENGTH == 0",
      "mutations": {
        "invoice_document.substatus": "Post Processing"
      },
      "retry_policy": { "max_attempts": 2 },
      "failure_behavior": "continue",
      "params": {
        "freight_patterns": "$.config.ingest_config.freight_patterns",
        "tariff_patterns": "$.config.ingest_config.tariff_patterns",
        "freight_as_line_item": "$.config.ingest_config.freight_as_line_item",
        "tariff_as_line_item": "$.config.ingest_config.tariff_as_line_item",
        "erp_push_condition": "$.config.connections.ERP.provider == 'cla-mocked'"
      }
    }
  },
  "conditional_sets": {
    "minimal_mode": {
      "override": {
        "activities.generate_images.enabled": false,
        "activities.map_invoice.enabled": false
      }
    },
    "ai_disabled": {
      "override": {
        "activities.extract_invoice.enabled": false,
        "activities.map_invoice.enabled": false
      }
    }
  },
  "idempotency": {
    "invoice_key": "{client_id}:{pdf_hash}",
    "activity_key_pattern": "{run_id}:{activity}:{attempt}"
  },
  "conflict_resolution": {
    "parallel_groups": {
      "mapping_phase": {
        "merge_strategy": "prefer_transform_invoice",  // last_write_wins | deep_merge | prefer_<activity>
        "deep_merge_fields": ["invoice.lineItems"]
      }
    }
  },
  "observability": {
    "emit_stage_events": true,
    "log_level_per_activity": {
      "generate_invoice_id": "INFO",
      "transform_invoice": "DEBUG"
    }
  }
}

Key Helper Functions (brief)

create_images: Render PDF pages to JPEG buffers.
transform / transform_items: Build Invoice & LineItem objects from DI JSON schema.
map_openai_extraction: Map OpenAI extraction object to internal Invoice.
combine_extractions / combine_line_items / fuzzy matching helpers: Reconcile DI vs AI line items using fuzzy description & amount heuristics.
_score_line_item: Consistency score for amount vs quantity*unit_price.
strip_vendor_name: Normalizes vendor string for pattern matching.
Separate mapping vs merging responsibilities for clarity.