Skip to navigation
Account Payable to Fluxx HLD

Design for AP Autoextract(Documents AI)-Intake requirements for Fluxx Intake service

High‑Level Activity Summaries (from workflow_intake.py and intake_activities.py that I inspected)

Activities in the Intake Workflow execution order:

  1. initialize_workflow (from workflow_activities)
  • Creates a workflow record (links document hash/invoiceID, client, pdf path/prefix). Pure side‑effect; no branching logic.
  1. check_duplicate_document (intake version)
  • Updates substatus to “Duplicate Check”.
  • Builds a Cosmos query from client-config ingest_config.unique_identifier_fields (supports dotted paths by flattening invoice document).
  • If >1 matching result found: marks workflow_status=failed, substatus=‘Duplicate Document Found’ and stores duplicate_invoice.
  • Otherwise succeeded.
  1. handle_sales_tax
  • Substatus “Handling Sales Tax”.
  • Reads feature flag negate_sales_tax (FeatureFlagService).
  • If enabled and invoiceTax > 0, appends a negative “Sales Tax” line item to neutralize tax at header level and updates invoice.
  • Always succeeds unless exception.
  1. validate_invoice
  • Substatus “Validating Invoice”.
  • Builds Validator object (struct of potential issues).
  • Sets workflow_status failed if subtotal, total, line item, or (optionally) missing PO number when intake_config.verify_po_number is true.
  • Records validator + includes_rounding flag.
  1. complete_workflow / fail_workflow
  • Marks child workflow status accordingly (completed or failed) after prior steps.

(Other available but not called here: run_change_stage.)

Configuration JSON (enable/disable, ordering, skip logic)

{
  "version": "1.0",
  "workflow_name": "invoice_intake",
  "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": [
      "initialize_workflow",
      "check_duplicate_document",
      "handle_sales_tax",
      "validate_invoice",
      "complete_workflow"
    ]
  },
  "activities": {
    "initialize_workflow": {
      "enabled": true,
      "timeout_ms": 15000,
      "retry_policy": { "max_attempts": 2 },
      "mutations": {
        "invoice_document.substatus": ""
      },
      "required_inputs": ["workflow_id", "current_workflow_id", "invoiceID", "client_id", "pdf_path", "pdf_prefix"],
      "skip_if": "state.workflow_initialized == true",
      "outputs_map": { "workflow_initialized": "true" }
    },
    "check_duplicate_document": {
      "enabled": true,
      "depends_on": ["initialize_workflow"],
      "mutations": {
        "invoice_document.substatus": "Duplicate Check"
      },
      "skip_if": "config.ingest_config.unique_identifier_fields LENGTH == 0",
      "failure_behavior": "halt"  // stop workflow if duplicate (status=failed)
    },
    "handle_sales_tax": {
      "enabled": true,
      "depends_on": ["check_duplicate_document"],
      "mutations": {
        "invoice_document.substatus": "Handling Sales Tax"
      },
      "skip_if": "NOT feature_flag('negate_sales_tax') OR NOT state.invoice_document.invoice.invoiceTax",
      "params": {
        "feature_flag": "negate_sales_tax"
      },
      "failure_behavior": "halt"
    },
    "validate_invoice": {
      "enabled": true,
      "depends_on": ["handle_sales_tax"],
      "mutations": {
        "invoice_document.substatus": "Validating Invoice"
      },
      "params": {
        "enforce_po": "$.config.intake_config.verify_po_number"
      },
      "skip_if": "state.invoice_document.invoice IS NULL",
      "failure_behavior": "continue"   // optional: allow finalize even if failed validation
    },
    "complete_workflow": {
      "enabled": true,
      "depends_on": ["validate_invoice"],
      "skip_if": "any_failed(['check_duplicate_document','handle_sales_tax'])",
      "mutations": { "invoice_document.substatus": "" }
    },
    "fail_workflow": {
      "enabled": true,
      "trigger_if": "any_failed(['check_duplicate_document','handle_sales_tax'])",
      "exclusive_with": ["complete_workflow"]
    }
  },
  "conditional_sets": {
    "fast_validation_only": {
      "override": {
        "activities.handle_sales_tax.enabled": false
      }
    },
    "skip_duplicates_check": {
      "override": {
        "activities.check_duplicate_document.enabled": false
      }
    }
  },
  "idempotency": {
    "activity_key_pattern": "{run_id}:{activity}:{attempt}"
  },
  "expressions": {
    "helpers": {
      "any_failed(list)": "returns true if any listed activity ended status=failed",
      "feature_flag(name)": "queries feature flag cache/context"
    }
  },
  "observability": {
    "emit_activity_events": true,
    "log_level_per_activity": {
      "validate_invoice": "DEBUG"
    }
  }
}

Notes:

  • skip_if / trigger_if expressions mirror earlier pattern.
  • fail_workflow acts as a conditional terminal path; orchestrator chooses complete_workflow vs fail_workflow.
  • feature_flag() expression corresponds to existing FeatureFlagService.
  • failure_behavior=continue lets you accept validation errors but still mark workflow completed (optional toggle).