id: ap-three-way-match-ai-triage-gate
namespace: company.finance
description: |
Automate Accounts Payable (AP) 3-way matching across Purchase Orders (POs),
Goods Receipt Notes (GRNs), and Vendor Invoices using DuckDB SQL, generate
auto-approved ERP payment batches and held exception reports, and optionally
run Gemini AI root-cause triage, Slack alerting, and human approval gating.
triggers:
- id: nightly_ap_reconciliation
type: io.kestra.plugin.core.trigger.Schedule
description: Nightly Accounts Payable 3-way match sweep before ERP payment run.
Shipped disabled by default.
cron: "0 2 * * 1-5"
disabled: true
- id: erp_invoice_batch_webhook
type: io.kestra.plugin.core.trigger.Webhook
description: Webhook trigger for on-demand AP subledger or OCR invoice batch ingestion.
key: "{{ secret('AP_WEBHOOK_KEY') }}"
inputs:
- id: inject_discrepancies
type: BOOL
defaults: true
description: When true, evaluates the full 6-invoice demo batch including 4
realistic AP discrepancies (overbilling, price variance, missing receipt,
duplicate invoice). Set to false to simulate a 100% clean batch.
- id: price_tolerance_pct
type: FLOAT
defaults: 2.0
description: Maximum allowed positive unit price variance percentage above the
contracted PO unit price before placing an invoice on hold.
- id: qty_tolerance_pct
type: FLOAT
defaults: 0.0
description: Maximum allowed quantity overbilling percentage above accepted
Goods Receipt Note (GRN) quantity.
- id: enable_ai_triage
type: BOOL
defaults: false
description: Enable optional Google Gemini AI triage to generate vendor
debit-memo drafts and remediation actions (requires GEMINI_API_KEY
secret).
- id: send_slack_alert
type: BOOL
defaults: false
description: Enable optional Slack notification to the Accounts Payable review
channel when exceptions are held (requires SLACK_WEBHOOK_URL secret).
- id: require_human_approval
type: BOOL
defaults: false
description: Pause workflow execution for interactive Controller sign-off when
held exceptions are detected.
- id: model_name
type: STRING
defaults: gemini-2.5-flash
description: Gemini model used when enable_ai_triage is set to true.
- id: purchase_orders
type: JSON
defaults:
- po_id: "PO-2026-101"
vendor_id: "VEND-01"
vendor_name: "Apex Cloud Hardware Inc."
item_sku: "SKU-GPU-H100"
item_description: "Enterprise AI Compute Server Node"
ordered_qty: 10
po_unit_price: 2500.00
currency: "USD"
- po_id: "PO-2026-102"
vendor_id: "VEND-02"
vendor_name: "Meridian Datacenter Optics"
item_sku: "SKU-SFP-400G"
item_description: "400G QSFP-DD Optical Transceiver"
ordered_qty: 50
po_unit_price: 120.00
currency: "USD"
- po_id: "PO-2026-103"
vendor_id: "VEND-03"
vendor_name: "Vanguard Power Systems"
item_sku: "SKU-PDU-30A"
item_description: "Rackmounted Managed PDU 30A"
ordered_qty: 20
po_unit_price: 450.00
currency: "USD"
- po_id: "PO-2026-104"
vendor_id: "VEND-04"
vendor_name: "Sentinel Security Appliances"
item_sku: "SKU-FW-10G"
item_description: "Next-Gen Hardware Firewall Appliance"
ordered_qty: 5
po_unit_price: 1800.00
currency: "USD"
- po_id: "PO-2026-105"
vendor_id: "VEND-05"
vendor_name: "Kinetix Storage Solutions"
item_sku: "SKU-NVME-15T"
item_description: "15.36TB U.3 Enterprise NVMe SSD"
ordered_qty: 15
po_unit_price: 800.00
currency: "USD"
description: Approved Purchase Orders (POs) specifying contracted quantities and
unit prices.
- id: goods_receipts
type: JSON
defaults:
- grn_id: "GRN-9001"
po_id: "PO-2026-101"
item_sku: "SKU-GPU-H100"
accepted_qty: 10
received_date: "2026-10-02"
- grn_id: "GRN-9002"
po_id: "PO-2026-102"
item_sku: "SKU-SFP-400G"
accepted_qty: 40
received_date: "2026-10-03"
- grn_id: "GRN-9003"
po_id: "PO-2026-103"
item_sku: "SKU-PDU-30A"
accepted_qty: 20
received_date: "2026-10-04"
- grn_id: "GRN-9005"
po_id: "PO-2026-105"
item_sku: "SKU-NVME-15T"
accepted_qty: 15
received_date: "2026-10-05"
description: Warehouse Goods Receipt Notes (GRNs) recording physically received
and inspected quantities.
- id: vendor_invoices
type: JSON
defaults:
- invoice_id: "INV-7001"
vendor_invoice_number: "APEX-88410"
vendor_id: "VEND-01"
po_id: "PO-2026-101"
item_sku: "SKU-GPU-H100"
billed_qty: 10
billed_unit_price: 2500.00
invoice_date: "2026-10-06"
- invoice_id: "INV-7002"
vendor_invoice_number: "MER-44912"
vendor_id: "VEND-02"
po_id: "PO-2026-102"
item_sku: "SKU-SFP-400G"
billed_qty: 50
billed_unit_price: 120.00
invoice_date: "2026-10-06"
- invoice_id: "INV-7003"
vendor_invoice_number: "VAN-33109"
vendor_id: "VEND-03"
po_id: "PO-2026-103"
item_sku: "SKU-PDU-30A"
billed_qty: 20
billed_unit_price: 495.00
invoice_date: "2026-10-07"
- invoice_id: "INV-7004"
vendor_invoice_number: "SEN-19004"
vendor_id: "VEND-04"
po_id: "PO-2026-104"
item_sku: "SKU-FW-10G"
billed_qty: 5
billed_unit_price: 1800.00
invoice_date: "2026-10-07"
- invoice_id: "INV-7005"
vendor_invoice_number: "KIN-55201"
vendor_id: "VEND-05"
po_id: "PO-2026-105"
item_sku: "SKU-NVME-15T"
billed_qty: 15
billed_unit_price: 808.00
invoice_date: "2026-10-08"
- invoice_id: "INV-7006"
vendor_invoice_number: "APEX-88410"
vendor_id: "VEND-01"
po_id: "PO-2026-101"
item_sku: "SKU-GPU-H100"
billed_qty: 10
billed_unit_price: 2500.00
invoice_date: "2026-10-09"
description: Inbound Vendor Invoices to validate against Purchase Orders and
Goods Receipt Notes.
tasks:
- id: stage_ap_documents
type: io.kestra.plugin.scripts.python.Script
description: Normalizes inbound PO, GRN, and Vendor Invoice JSON payloads into
structured CSV staging files for DuckDB reconciliation.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.11-slim
inputFiles:
purchase_orders.json: "{{ inputs.purchase_orders | toJson }}"
goods_receipts.json: "{{ inputs.goods_receipts | toJson }}"
vendor_invoices.json: "{{ inputs.vendor_invoices | toJson }}"
outputFiles:
- purchase_orders.csv
- goods_receipts.csv
- vendor_invoices.csv
script: |
import csv
import json
from pathlib import Path
inject_discrepancies = "{{ inputs.inject_discrepancies }}".strip().lower() == "true"
pos = json.loads(Path("purchase_orders.json").read_text(encoding="utf-8"))
grns = json.loads(Path("goods_receipts.json").read_text(encoding="utf-8"))
invoices = json.loads(Path("vendor_invoices.json").read_text(encoding="utf-8"))
if not inject_discrepancies:
clean_invoice_ids = {"INV-7001", "INV-7005"}
filtered = [inv for inv in invoices if inv.get("invoice_id") in clean_invoice_ids]
if filtered:
invoices = filtered
def write_csv(filename: str, fieldnames: list[str], rows: list[dict]) -> None:
with open(filename, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in rows:
writer.writerow({k: row.get(k, "") for k in fieldnames})
write_csv(
"purchase_orders.csv",
["po_id", "vendor_id", "vendor_name", "item_sku", "item_description", "ordered_qty", "po_unit_price", "currency"],
pos,
)
write_csv(
"goods_receipts.csv",
["grn_id", "po_id", "item_sku", "accepted_qty", "received_date"],
grns,
)
write_csv(
"vendor_invoices.csv",
["invoice_id", "vendor_invoice_number", "vendor_id", "po_id", "item_sku", "billed_qty", "billed_unit_price", "invoice_date"],
invoices,
)
print(f"Staged {len(pos)} POs, {len(grns)} GRNs, and {len(invoices)} Vendor Invoices (inject_discrepancies={inject_discrepancies}).")
- id: reconcile_three_way_match
type: io.kestra.plugin.jdbc.duckdb.Query
description: Executes deterministic 3-way match SQL joining Purchase Orders,
aggregated Goods Receipts, and deduplicated Vendor Invoices.
url: "jdbc:duckdb:"
inputFiles:
purchase_orders.csv: "{{ outputs.stage_ap_documents.outputFiles['purchase_orders.csv'] }}"
goods_receipts.csv: "{{ outputs.stage_ap_documents.outputFiles['goods_receipts.csv'] }}"
vendor_invoices.csv: "{{ outputs.stage_ap_documents.outputFiles['vendor_invoices.csv'] }}"
fetchType: FETCH
sql: |
WITH po AS (
SELECT
CAST(po_id AS VARCHAR) AS po_id,
CAST(vendor_id AS VARCHAR) AS vendor_id,
CAST(vendor_name AS VARCHAR) AS vendor_name,
CAST(item_sku AS VARCHAR) AS item_sku,
CAST(item_description AS VARCHAR) AS item_description,
CAST(ordered_qty AS DOUBLE) AS ordered_qty,
CAST(po_unit_price AS DOUBLE) AS po_unit_price,
CAST(currency AS VARCHAR) AS currency
FROM read_csv_auto('purchase_orders.csv', header = true)
),
grn_agg AS (
SELECT
CAST(po_id AS VARCHAR) AS po_id,
CAST(item_sku AS VARCHAR) AS item_sku,
SUM(CAST(accepted_qty AS DOUBLE)) AS received_qty,
MAX(CAST(received_date AS VARCHAR)) AS last_received_date
FROM read_csv_auto('goods_receipts.csv', header = true)
GROUP BY po_id, item_sku
),
ranked_invoices AS (
SELECT
CAST(invoice_id AS VARCHAR) AS invoice_id,
CAST(vendor_invoice_number AS VARCHAR) AS vendor_invoice_number,
CAST(vendor_id AS VARCHAR) AS vendor_id,
CAST(po_id AS VARCHAR) AS po_id,
CAST(item_sku AS VARCHAR) AS item_sku,
CAST(billed_qty AS DOUBLE) AS billed_qty,
CAST(billed_unit_price AS DOUBLE) AS billed_unit_price,
CAST(invoice_date AS VARCHAR) AS invoice_date,
ROW_NUMBER() OVER (
PARTITION BY CAST(vendor_id AS VARCHAR), CAST(vendor_invoice_number AS VARCHAR), CAST(item_sku AS VARCHAR)
ORDER BY CAST(invoice_date AS VARCHAR), CAST(invoice_id AS VARCHAR)
) AS duplicate_rank
FROM read_csv_auto('vendor_invoices.csv', header = true)
),
reconciled AS (
SELECT
inv.invoice_id,
inv.vendor_invoice_number,
inv.vendor_id,
COALESCE(po.vendor_name, 'UNREGISTERED_VENDOR') AS vendor_name,
inv.po_id,
inv.item_sku,
COALESCE(po.item_description, 'UNCATALOGUED_ITEM') AS item_description,
COALESCE(po.currency, 'USD') AS currency,
COALESCE(po.ordered_qty, 0.0) AS po_ordered_qty,
COALESCE(po.po_unit_price, 0.0) AS po_unit_price,
COALESCE(grn.received_qty, 0.0) AS grn_received_qty,
inv.billed_qty,
inv.billed_unit_price,
inv.invoice_date,
inv.duplicate_rank,
ROUND(inv.billed_qty * inv.billed_unit_price, 2) AS billed_amount_usd,
ROUND(COALESCE(grn.received_qty, 0.0) * COALESCE(po.po_unit_price, 0.0), 2) AS received_po_value_usd,
ROUND(
CASE
WHEN inv.duplicate_rank > 1 THEN inv.billed_qty * inv.billed_unit_price
ELSE (inv.billed_qty * inv.billed_unit_price) - (COALESCE(grn.received_qty, 0.0) * COALESCE(po.po_unit_price, 0.0))
END,
2
) AS line_variance_usd,
ROUND(
CASE
WHEN COALESCE(po.po_unit_price, 0.0) > 0.0
THEN ((inv.billed_unit_price - po.po_unit_price) / po.po_unit_price) * 100.0
WHEN COALESCE(po.po_unit_price, 0.0) = 0.0 AND inv.billed_unit_price = 0.0
THEN 0.0
ELSE 100.0
END,
2
) AS price_variance_pct,
GREATEST(0.0, inv.billed_qty - COALESCE(grn.received_qty, 0.0)) AS overbilled_qty,
CASE
WHEN po.po_id IS NULL THEN 'UNMATCHED_PO'
WHEN inv.duplicate_rank > 1 THEN 'DUPLICATE_INVOICE'
WHEN COALESCE(grn.received_qty, 0.0) <= 0.0 THEN 'MISSING_RECEIPT'
WHEN inv.billed_qty > COALESCE(grn.received_qty, 0.0) * (1.0 + ({{ inputs.qty_tolerance_pct }} / 100.0)) THEN 'QTY_OVERBILLED'
WHEN po.po_unit_price < 0.0 OR inv.billed_unit_price < 0.0 THEN 'INVALID_UNIT_PRICE'
WHEN po.po_unit_price = 0.0 AND inv.billed_unit_price > 0.0 THEN 'PRICE_VARIANCE_EXCEEDED'
WHEN po.po_unit_price > 0.0 AND ((inv.billed_unit_price - po.po_unit_price) / po.po_unit_price) * 100.0 > {{ inputs.price_tolerance_pct }} THEN 'PRICE_VARIANCE_EXCEEDED'
ELSE 'MATCHED_AUTO_APPROVED'
END AS exception_code
FROM ranked_invoices inv
LEFT JOIN po
ON inv.po_id = po.po_id
AND inv.item_sku = po.item_sku
LEFT JOIN grn_agg grn
ON inv.po_id = grn.po_id
AND inv.item_sku = grn.item_sku
)
SELECT
invoice_id,
vendor_invoice_number,
vendor_id,
vendor_name,
po_id,
item_sku,
item_description,
currency,
po_ordered_qty,
po_unit_price,
grn_received_qty,
billed_qty,
billed_unit_price,
billed_amount_usd,
received_po_value_usd,
line_variance_usd,
price_variance_pct,
overbilled_qty,
exception_code,
CASE
WHEN exception_code = 'MATCHED_AUTO_APPROVED' THEN 'AUTO_APPROVED'
ELSE 'HELD_FOR_REVIEW'
END AS match_status,
CASE
WHEN exception_code = 'DUPLICATE_INVOICE'
THEN 'Duplicate submission of vendor invoice ' || vendor_invoice_number || ' for vendor ' || vendor_id || ' ($' || CAST(billed_amount_usd AS VARCHAR) || ' duplicate exposure blocked).'
WHEN exception_code = 'MISSING_RECEIPT'
THEN 'No Goods Receipt Note (GRN) recorded for ' || po_id || ' / ' || item_sku || ' (billed ' || CAST(CAST(billed_qty AS INTEGER) AS VARCHAR) || ' units vs 0 received).'
WHEN exception_code = 'QTY_OVERBILLED'
THEN 'Billed quantity (' || CAST(CAST(billed_qty AS INTEGER) AS VARCHAR) || ') exceeds GRN accepted quantity (' || CAST(CAST(grn_received_qty AS INTEGER) AS VARCHAR) || ') by ' || CAST(CAST(overbilled_qty AS INTEGER) AS VARCHAR) || ' units ($' || CAST(line_variance_usd AS VARCHAR) || ' variance).'
WHEN exception_code = 'INVALID_UNIT_PRICE'
THEN 'Invalid negative unit price detected (billed $' || CAST(billed_unit_price AS VARCHAR) || ' vs PO $' || CAST(po_unit_price AS VARCHAR) || ').'
WHEN exception_code = 'PRICE_VARIANCE_EXCEEDED'
THEN 'Billed unit price ($' || CAST(billed_unit_price AS VARCHAR) || ') exceeds PO unit price ($' || CAST(po_unit_price AS VARCHAR) || ') by ' || CAST(price_variance_pct AS VARCHAR) || '%, breaching the {{ inputs.price_tolerance_pct }}% tolerance ($' || CAST(line_variance_usd AS VARCHAR) || ' variance).'
WHEN exception_code = 'UNMATCHED_PO'
THEN 'Invoice references non-existent PO ' || po_id || ' or SKU ' || item_sku || '.'
ELSE '3-way match verified within PO/GRN tolerances (price variance: ' || CAST(price_variance_pct AS VARCHAR) || '%).'
END AS hold_explanation
FROM reconciled
ORDER BY invoice_id;
- id: generate_erp_and_exception_batches
type: io.kestra.plugin.scripts.python.Script
description: Partitions reconciled invoices into an ERP auto-approval CSV batch,
a held exceptions CSV report, a JSON metrics summary, and a Markdown audit
report.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
containerImage: python:3.11-slim
inputFiles:
reconciled_rows.json: "{{ outputs.reconcile_three_way_match.rows | toJson }}"
outputFiles:
- auto_approved_erp_batch.csv
- held_exceptions_report.csv
- reconciliation_metrics.json
- ap_three_way_match_audit_report.md
script: |
import csv
import json
from pathlib import Path
rows = json.loads(Path("reconciled_rows.json").read_text(encoding="utf-8"))
approved_rows = [r for r in rows if r.get("match_status") == "AUTO_APPROVED"]
held_rows = [r for r in rows if r.get("match_status") == "HELD_FOR_REVIEW"]
columns = [
"invoice_id",
"vendor_invoice_number",
"vendor_id",
"vendor_name",
"po_id",
"item_sku",
"po_ordered_qty",
"po_unit_price",
"grn_received_qty",
"billed_qty",
"billed_unit_price",
"billed_amount_usd",
"received_po_value_usd",
"line_variance_usd",
"price_variance_pct",
"overbilled_qty",
"exception_code",
"match_status",
"hold_explanation",
]
def write_batch(filename: str, dataset: list[dict]) -> None:
with open(filename, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=columns)
writer.writeheader()
for item in dataset:
writer.writerow({col: item.get(col, "") for col in columns})
write_batch("auto_approved_erp_batch.csv", approved_rows)
write_batch("held_exceptions_report.csv", held_rows)
total_billed = round(sum(float(r.get("billed_amount_usd", 0.0)) for r in rows), 2)
approved_amount = round(sum(float(r.get("billed_amount_usd", 0.0)) for r in approved_rows), 2)
held_exposure = round(sum(float(r.get("billed_amount_usd", 0.0)) for r in held_rows), 2)
held_overcharge_variance = round(sum(max(0.0, float(r.get("line_variance_usd", 0.0))) for r in held_rows), 2)
exception_breakdown = {}
for r in held_rows:
code = r.get("exception_code", "UNKNOWN")
exception_breakdown[code] = exception_breakdown.get(code, 0) + 1
metrics = {
"total_invoices_evaluated": len(rows),
"auto_approved_count": len(approved_rows),
"held_exceptions_count": len(held_rows),
"total_billed_amount_usd": total_billed,
"auto_approved_amount_usd": approved_amount,
"held_exposure_amount_usd": held_exposure,
"held_overcharge_variance_usd": held_overcharge_variance,
"exception_breakdown": exception_breakdown,
}
Path("reconciliation_metrics.json").write_text(json.dumps(metrics, indent=2), encoding="utf-8")
md_lines = [
"# Accounts Payable 3-Way Match Audit Report",
"",
"## Executive Summary",
f"- **Total Invoices Evaluated:** {metrics['total_invoices_evaluated']} (`${total_billed:,.2f}`)",
f"- **Auto-Approved for ERP Payment:** {metrics['auto_approved_count']} (`${approved_amount:,.2f}`)",
f"- **Held for AP Exception Review:** {metrics['held_exceptions_count']} (`${held_exposure:,.2f}` total invoice exposure; `${held_overcharge_variance:,.2f}` net overcharge/unbacked variance)",
f"- **Configured Tolerances:** Price Variance <= `{{ inputs.price_tolerance_pct }}%`, Quantity Overbill <= `{{ inputs.qty_tolerance_pct }}%`",
"",
"## Held Exceptions Requiring Controller / Buyer Action",
]
if held_rows:
md_lines.extend([
"| Invoice ID | Vendor Invoice # | Vendor | PO ID | SKU | Exception Code | Billed ($) | Variance ($) | Hold Explanation |",
"| :--- | :--- | :--- | :--- | :--- | :--- | ---: | ---: | :--- |",
])
for r in held_rows:
md_lines.append(
f"| `{r['invoice_id']}` | `{r['vendor_invoice_number']}` | {r['vendor_name']} | `{r['po_id']}` | `{r['item_sku']}` | **{r['exception_code']}** | ${float(r['billed_amount_usd']):,.2f} | ${float(r['line_variance_usd']):,.2f} | {r['hold_explanation']} |"
)
else:
md_lines.append("_No discrepancies detected. All invoices satisfied 3-way match tolerances._")
md_lines.extend([
"",
"## Auto-Approved ERP Payment Batch",
"| Invoice ID | Vendor Invoice # | Vendor | PO ID | SKU | Billed Qty | Unit Price ($) | Approved Total ($) | Status |",
"| :--- | :--- | :--- | :--- | :--- | ---: | ---: | ---: | :--- |",
])
for r in approved_rows:
md_lines.append(
f"| `{r['invoice_id']}` | `{r['vendor_invoice_number']}` | {r['vendor_name']} | `{r['po_id']}` | `{r['item_sku']}` | {int(float(r['billed_qty']))} | ${float(r['billed_unit_price']):,.2f} | ${float(r['billed_amount_usd']):,.2f} | `{r['match_status']}` |"
)
Path("ap_three_way_match_audit_report.md").write_text("\n".join(md_lines) + "\n", encoding="utf-8")
print("::" + json.dumps({"outputs": metrics}) + "::")
- id: expose_reconciliation_metrics
type: io.kestra.plugin.core.output.OutputValues
description: Publishes structured 3-way match counts and dollar exposure metrics
to execution outputs.
values:
total_invoices_evaluated: "{{
outputs.generate_erp_and_exception_batches.vars.total_invoices_evaluated
}}"
auto_approved_count: "{{
outputs.generate_erp_and_exception_batches.vars.auto_approved_count }}"
held_exceptions_count: "{{
outputs.generate_erp_and_exception_batches.vars.held_exceptions_count
}}"
auto_approved_amount_usd: "{{
outputs.generate_erp_and_exception_batches.vars.auto_approved_amount_usd
}}"
held_exposure_amount_usd: "{{
outputs.generate_erp_and_exception_batches.vars.held_exposure_amount_usd
}}"
held_overcharge_variance_usd: "{{
outputs.generate_erp_and_exception_batches.vars.held_overcharge_varianc\
e_usd }}"
- id: route_reconciliation_outcome
type: io.kestra.plugin.core.flow.If
description: Branches workflow between the exception quarantine path (when
held_exceptions_count > 0) and the clean ERP batch path.
condition: "{{
outputs.generate_erp_and_exception_batches.vars.held_exceptions_count > 0
}}"
then:
- id: log_exception_hold_summary
type: io.kestra.plugin.core.log.Log
description: Logs the deterministic rule-based 3-way match exception hold
summary to the execution log.
message: |
AP 3-WAY MATCH GATE: {{ outputs.expose_reconciliation_metrics.values.held_exceptions_count }} invoice(s) placed on payment hold (${{ outputs.expose_reconciliation_metrics.values.held_exposure_amount_usd }} total held exposure; ${{ outputs.expose_reconciliation_metrics.values.held_overcharge_variance_usd }} net variance).
Auto-approved batch ({{ outputs.expose_reconciliation_metrics.values.auto_approved_count }} invoices, ${{ outputs.expose_reconciliation_metrics.values.auto_approved_amount_usd }}) exported to auto_approved_erp_batch.csv.
Held exceptions exported to held_exceptions_report.csv.
- id: check_ai_triage_enabled
type: io.kestra.plugin.core.flow.If
description: Runs optional Google Gemini AI exception triage when
enable_ai_triage is set to true.
condition: "{{ inputs.enable_ai_triage == true }}"
then:
- id: ai_triage_exceptions
type: io.kestra.plugin.ai.completion.ChatCompletion
description: Uses Google Gemini to synthesize per-invoice root-cause
explanations, vendor debit-memo communication drafts, and ERP
remediation actions.
provider:
type: io.kestra.plugin.ai.provider.GoogleGemini
apiKey: "{{ secret('GEMINI_API_KEY') }}"
modelName: "{{ inputs.model_name }}"
messages:
- type: SYSTEM
content: |
You are a Senior Accounts Payable Controller and Procurement Audit Specialist.
Analyze the held 3-way match invoice exceptions exported from DuckDB.
For each held invoice:
1. Classify the root cause (Duplicate Billing, Unreceived Goods, Quantity Overbill, or Unit Price Overcharge).
2. State the exact dollar variance and recommended ERP action (Block Payment, Request Warehouse Receiving Confirmation, Issue Debit Memo, or Reject Duplicate).
3. Draft a concise, professional vendor dispute notice referencing the PO ID, Vendor Invoice Number, and exact variance figures.
- type: USER
content: |
Reconciliation Metrics:
{{ read(outputs.generate_erp_and_exception_batches.outputFiles['reconciliation_metrics.json']) }}
Held Exceptions CSV:
{{ read(outputs.generate_erp_and_exception_batches.outputFiles['held_exceptions_report.csv']) }}
- id: save_ai_triage_memo
type: io.kestra.plugin.core.storage.Write
description: Saves the Gemini AI triage memo and vendor dispute drafts to
internal storage.
content: "{{ outputs.ai_triage_exceptions.textOutput }}"
extension: .md
- id: check_slack_alert_enabled
type: io.kestra.plugin.core.flow.If
description: Sends optional Slack notification to the AP team when
send_slack_alert is set to true.
condition: "{{ inputs.send_slack_alert == true }}"
then:
- id: notify_ap_slack_channel
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Posts a rich AP 3-way match exception alert to Slack with held
exposure metrics.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": ":rotating_light: *AP 3-Way Match Exception Gate Triggered*\n*Execution:* `{{ execution.id }}`\n*Auto-Approved Invoices:* `{{ outputs.expose_reconciliation_metrics.values.auto_approved_count }}` (`${{ outputs.expose_reconciliation_metrics.values.auto_approved_amount_usd }}`)\n*Held Invoices:* `{{ outputs.expose_reconciliation_metrics.values.held_exceptions_count }}` (`${{ outputs.expose_reconciliation_metrics.values.held_exposure_amount_usd }}` exposure | `${{ outputs.expose_reconciliation_metrics.values.held_overcharge_variance_usd }}` variance)\nReview `held_exceptions_report.csv` and `ap_three_way_match_audit_report.md` in Kestra UI."
}
- id: check_human_approval_required
type: io.kestra.plugin.core.flow.If
description: Optionally pauses execution for Controller sign-off when
require_human_approval is set to true.
condition: "{{ inputs.require_human_approval == true }}"
then:
- id: await_controller_signoff
type: io.kestra.plugin.core.flow.Pause
description: Pauses the pipeline until an AP Controller reviews the held
exceptions and records a disposition decision.
onResume:
- id: controller_decision
type: SELECT
values:
- RELEASE_AUTO_APPROVED_AND_QUARANTINE_HELD
- OVERRIDE_AND_RELEASE_ALL
- REJECT_ENTIRE_BATCH
defaults: RELEASE_AUTO_APPROVED_AND_QUARANTINE_HELD
description: Select the Accounts Payable Controller disposition for the current
batch.
- id: controller_notes
type: STRING
defaults: "Verified 3-way match audit report; release clean batch and issue
vendor debit memos for held invoices."
description: Audit log note recorded by the approving Controller.
- id: log_controller_disposition
type: io.kestra.plugin.core.log.Log
description: Records the Controller's interactive approval decision and audit
note.
message: "Controller disposition recorded: {{
outputs.await_controller_signoff.onResume.controller_decision }},
Notes: {{
outputs.await_controller_signoff.onResume.controller_notes }}"
else:
- id: log_clean_batch_posting
type: io.kestra.plugin.core.log.Log
description: Logs clean audit confirmation when 100% of invoices pass 3-way
matching within tolerance.
message: "All {{
outputs.expose_reconciliation_metrics.values.auto_approved_count }}
invoice(s) (${{
outputs.expose_reconciliation_metrics.values.auto_approved_amount_usd
}}) passed 3-way match tolerances with zero held exceptions. Ready for
ERP payment run."
