Auto-Approval Engine for APS Recommendations¶

Status: Implemented (Phases A–F). Builds on the Phase-1 Scenario Review foundation (historization.auto_approval, api.history_router, interact_auto_approval_rule / interact_recommendation_approval / interact_planning_event).

Design plan: .kilo/plans/auto-approval-engine.md

Canonical code: files/auto_approval/ · files/api/auto_approval_router.py · files/DDL/auto_approval_schema.sql · files/DDL/ch_auto_approval_schema.sql · files/scripts/retrain_auto_approval.py

1. Overview & Goals¶

The Auto-Approval Engine is a governance-gated, auditable decision layer that sits between an APS (Advanced Planning & Scheduling) recommendation generator and the human planner. Its goal is to automatically approve high-confidence, low-risk recommendations (freeing planner time for the cases that matter) while routing uncertain or risky ones for manual review.

Goals¶

#	Goal	Mechanism
G1	Reduce planner toil	`auto_approve` rules + model band `p_approve > 0.99`
G2	Never silently approve risk	`mandatory_review` rules override model; mid-band requires `review_approve`
G3	Explain every decision	matched rule `explanation_template` + SHAP top features + kNN similar history
G4	Audit every transition	status column + `interact_planning_event` + ClickHouse mirrors
G5	Learn from planner behaviour	planner-imitation classifier retrained nightly
G6	Be RL-ready	`Policy`/`RewardFunction` Protocols + reward signal populated today

Non-goals¶

Replacing the planner — the engine imitates the planner; it does not replace judgement on novel/risky cases.
Cross-tenant model sharing — one champion model per tenant DB (locked decision #6).
Writing directly to the ERP — the default StatusFlagExecutor marks status; the ERP writes back via webhook (locked decision #3).

Phase-1 foundation

The engine extends — does not fork — historization.auto_approval: evaluate_predicate, record_decision_safe, simulate_rule, and the interact_recommendation_approval / interact_auto_approval_rule / interact_planning_event tables are reused unchanged.

2. Recommendation Types¶

The eight APS recommendation types the engine handles (domain.py → RecommendationType):

Type	Meaning	Typical payload keys
`CREATE_PO`	Create a new purchase order	`qty`, `release_week`, `arrival_week`, `supplier_id`
`INCREASE_PO`	Increase an existing PO quantity	`qty_delta`, `quantity_change_percent`
`DECREASE_PO`	Decrease an existing PO quantity	`qty_delta`, `quantity_change_percent`
`RESCHEDULE_IN`	Pull an PO in (earlier)	`date_change_days`
`RESCHEDULE_OUT`	Push a PO out (later)	`date_change_days`
`TRANSFER`	Move stock between sites	`source_site_id`, `qty`
`EXPEDITE`	Expedite an order (premium freight)	`qty`, `arrival_week`
`CANCEL_ORDER`	Cancel an open order	`reason`

recommendation_type is one-hot encoded into the dense feature vector (recommendation_type_CREATE_PO … recommendation_type_CANCEL_ORDER) so the model can learn type-conditional planner behaviour.

3. Architecture¶

flowchart LR
    APS[APS / MEIO / Supply Engine] -->|POST /ingest| API[FastAPI router<br/>auto_approval_router.py]
    API -->|insert| PGrec[(interact_recommendation<br/>PG)]
    API -->|mirror| CHrec[(PIPE_recommendation<br/>CH)]

    API --> ORC[orchestrator.decide]

    ORC -->|1. route_planner| RULE[Rule engine<br/>rules/engine.py]
    ORC -->|2. build_features| FB[Feature builder<br/>features/builder.py]
    FB -->|read| CHpipe[(ClickHouse<br/>PIPE_supply_*<br/>PIPE_forecast_*<br/>PIPE_series_*<br/>PIPE_fitted_*<br/>PIPE_lead_time_*<br/>PIPE_service_*<br/>PIPE_on_hand_*<br/>)]
    FB -->|read| PGmaster[(master.item<br/>interact_planner_profile<br/>interact_recommendation_approval)]
    FB -->|persist| PGfeat[(interact_recommendation_feature)]

    ORC -->|3. evaluate_rules| RULE
    RULE -->|governance gate| RULETBL[(interact_auto_approval_rule<br/>+ interact_rule_set)]

    ORC -->|4. model_score| INF[Inference<br/>models/inference.py]
    INF -->|load champion| REG[(interact_model_artifact)]
    INF -->|score| MDL[(LightGBM / XGBoost<br/>+ isotonic calibrator)]

    ORC -->|5. apply_policy| POL[confidence.py]
    ORC -->|6. explain| EXP[explain.py<br/>SHAP + kNN]
    ORC -->|7. persist + audit| PGrec
    ORC -->|audit event| EV[(interact_planning_event)]

    POL -->|auto_approve| DONE1[status = decided]
    POL -->|review_approve / manual_review| REVIEW[status = planner_review]
    REVIEW -->|GET /pending + POST /{id}/decide| UI[Planner UI]
    UI -->|approve| EXE[executor.py<br/>StatusFlagExecutor]
    EXE -->|status = executed| PGrec
    EXE -.->|ERP webhook<br/>POST /{id}/execution-result| ERPTBL[(interact_recommendation_execution)]

    DONE1 --> OUTC[engine/outcome.py<br/>+4w / +13w]
    EXE --> OUTC
    OUTC -->|deltas + reward_signal| PGout[(interact_recommendation_outcome)]
    OUTC -->|mirror| CHout[(PIPE_recommendation_outcome<br/>RL replay buffer)]

The orchestrator (engine/orchestrator.py:decide) is the single entry point. It runs a strict ten-step pipeline per recommendation and persists every intermediate state. A deterministic rule hit short-circuits the model (locked decision #1); mandatory_review forces manual review even if the model is highly confident; planner routing runs before features so the planner-history block can use the routed planner_id (locked decision #2).

4. Decision Policy / Confidence Layer¶

The confidence layer (engine/confidence.py:decide) is the heart of the engine. It turns a (RuleHit, ModelScore) pair into a final decision + confidence band.

flowchart TD
    START([recommendation scored]) --> Q1{Rule hit?}
    Q1 -->|auto_approve| FA[FinalDecision.AUTO_APPROVE<br/>source=rule, conf=1.0, band=high]
    Q1 -->|auto_reject| FR[FinalDecision.AUTO_REJECT<br/>source=rule, conf=0.0, band=low]
    Q1 -->|auto_modify| FM[FinalDecision.AUTO_MODIFY<br/>source=rule, conf=null]
    Q1 -->|mandatory_review| MREV[FinalDecision.MANUAL_REVIEW<br/>source=rule<br/>even if model confident]
    Q1 -->|no rule / action=manual| Q2{Model champion loaded?}

    Q2 -->|no champion| MM[FinalDecision.MANUAL_REVIEW<br/>source=model, band=low]
    Q2 -->|champion loaded| Q3{p_approve > auto_thr?<br/>default 0.99}
    Q3 -->|yes| MA[FinalDecision.AUTO_APPROVE<br/>source=model, band=high]
    Q3 -->|no| Q4{p_approve > review_thr?<br/>default 0.95}
    Q4 -->|yes| MRV[FinalDecision.REVIEW_APPROVE<br/>planner confirms<br/>band=medium]
    Q4 -->|no| ML[FinalDecision.MANUAL_REVIEW<br/>source=model, band=low]

Condition	Final decision	Source	Confidence band
Rule `auto_approve` matches	`auto_approve`	`rule`	`high` (conf=1.0)
Rule `auto_reject` matches	`auto_reject`	`rule`	`low` (conf=0.0)
Rule `auto_modify` matches	`auto_modify`	`rule`	—
Rule `mandatory_review` matches	`manual_review`	`rule`	model band (if any)
No rule, `p_approve > 0.99`	`auto_approve`	`model`	`high`
No rule, `p_approve > 0.95`	`review_approve`	`model`	`medium`
No rule, `p_approve ≤ 0.95`	`manual_review`	`model`	`low`
No rule, no champion	`manual_review`	`model`	`low`

Thresholds are governance-tunable

auto_approve_threshold and review_threshold live in files/auto_approval/config.yaml. Changing them does not require a redeploy — call POST /api/auto-approval/config/reload (admin only).

5. Recommendation State Machine¶

stateDiagram-v2
    [*] --> ingested: POST /ingest
    ingested --> featured: build_features() ok
    ingested --> ingested: feature build failed (rollback)
    featured --> rule_eval: evaluate_rules()
    rule_eval --> model_scored: no rule short-circuit + champion loaded
    rule_eval --> decided: rule auto_approve/reject/modify
    model_scored --> decided: auto_approve (p>0.99)
    model_scored --> planner_review: review_approve or manual_review
    decided --> planner_review: review_approve edge case
    planner_review --> planner_decided: planner approve (POST /{id}/decide)
    planner_review --> rejected_by_planner: planner reject/modify
    planner_decided --> executed: execute() (StatusFlagExecutor)
    decided --> executed: auto_approve path
    executed --> outcome_measured: +4w / +13w elapsed
    outcome_measured --> [*]
    rejected_by_planner --> [*]
    ingested --> superseded: newer rec for same key
    featured --> superseded: newer rec for same key
    decided --> superseded: newer rec for same key

States are stored on interact_recommendation.status (a CHECK-constrained TEXT column). Every transition also appends a row to interact_planning_event (audit layer 2) so the full chain is recoverable even if the status column is mutated.

Deduplication

interact_recommendation has UNIQUE (pipeline_id, recommendation_type, recommendation_id). Re-ingesting the same external APS ref does an ON CONFLICT DO UPDATE on recommended_payload (the dedupe key is the APS-supplied recommendation_id, not our surrogate id).

6. Event Architecture & Sequence Diagrams¶

6.1 Ingest → Decision (happy path, rule short-circuit)¶

sequenceDiagram
    autonumber
    participant APS as APS
    participant API as FastAPI router
    participant DB as PG (interact_recommendation)
    participant ORC as orchestrator.decide
    participant FB as feature builder
    participant CH as ClickHouse PIPE_*
    participant RE as rule engine
    participant POL as confidence.apply_policy
    participant EV as interact_planning_event

    APS->>API: POST /ingest (defer=false)
    API->>API: require_pipeline_id(); ad=planning_monday(pid)
    API->>DB: INSERT ... ON CONFLICT DO UPDATE
    API->>ORC: decide(rec_id)
    ORC->>EV: rec_decide_start
    ORC->>RE: route_planner() (assign_planner rules)
    RE-->>ORC: planner_id
    ORC->>DB: UPDATE planner_id
    ORC->>FB: build_features()
    FB->>CH: PIPE_supply_inventory_projection, PIPE_forecast_point_values, ...
    FB->>DB: master.item, interact_planner_profile
    FB-->>ORC: feature_vector_dense
    ORC->>DB: persist features; status=featured
    ORC->>EV: feature_built
    ORC->>RE: evaluate_rules()
    RE->>RE: governance gate (governance_state='active')
    RE-->>ORC: RuleHit(action=auto_approve, rule_id=42)
    ORC->>EV: rule_evaluated (rule_id=42)
    Note over ORC,POL: rule short-circuit → skip model
    ORC->>POL: apply_policy(rule_hit, None)
    POL-->>ORC: (auto_approve, rule, 1.0, high)
    ORC->>DB: status=decided; confidence=1.0; auto_rule_id=42
    ORC->>EV: decision_emitted
    ORC-->>API: Decision
    API-->>APS: 200 {recommendation_id, decision}

6.2 Planner review path (model-scored, mid-band)¶

sequenceDiagram
    autonumber
    participant ORC as orchestrator.decide
    participant INF as inference.score
    participant EXP as explain (SHAP+kNN)
    participant POL as confidence.apply_policy
    participant DB as PG
    participant UI as Planner UI
    participant EXE as executor
    participant ERP as External ERP

    ORC->>ORC: route_planner → build_features → evaluate_rules (no hit)
    ORC->>INF: score(feature_vector_dense)
    INF->>INF: load champion (singleton, hot-reload)
    INF->>INF: predict_proba + isotonic transform
    INF-->>ORC: ModelScore(p_approve=0.96, p_reject=0.02, p_modify=0.02)
    ORC->>POL: apply_policy(None, model_score)
    POL-->>ORC: (review_approve, model, 0.96, medium)
    ORC->>EXP: explain(feature_vector_dense)
    EXP->>EXP: SHAP TreeExplainer (approve class)
    EXP->>EXP: cosine kNN over 500 candidates
    EXP-->>ORC: top_features[], similar_history_ids[]
    ORC->>DB: status=planner_review; rationale; top_features
    Note over UI: planner polls /pending (risk-first: confidence ASC)
    UI->>UI: GET /pending → sees rec with rationale + SHAP + similar cases
    UI->>DB: POST /{id}/decide {decision: approve}
    UI->>EXE: execute(rec_id)
    EXE->>DB: status=executed
    EXE-->>UI: {exec_status: success}
    ERP-->>UI: POST /{id}/execution-result {erp_ref, exec_status}
    Note over UI: outcome measured later at +4w / +13w

6.3 Learning loop + retrain¶

sequenceDiagram
    autonumber
    participant CRON as Nightly cron
    participant RT as retrain_auto_approval.py
    participant DB as PG
    participant BF as backfill.build_training_dataset
    participant TR as trainer.train
    participant RG as registry.save_artifact
    participant SH as Shadow scoring<br/>(challenger)
    participant ADM as Admin

    CRON->>RT: run
    RT->>DB: refresh interact_planner_profile<br/>(per-approver approval/override/reject rates)
    RT->>BF: build_training_dataset(lookback_days=180)
    BF->>DB: SELECT historical approvals<br/>+ features_at_decision
    BF-->>RT: X[], y[] (labels: approve/modify/reject)
    alt len(X) < min_train_rows
        RT-->>CRON: exit 0 (skip retrain)
    else enough rows
        RT->>TR: train(X, y, algorithm=lightgbm)
        TR->>TR: inverse-frequency sample_weight
        TR->>TR: time-based validation split (last 20%)
        TR->>TR: LightGBM fit + isotonic calibration
        TR-->>RT: artifact{metrics, model_uri, model_hash}
        RT->>RG: save_artifact(artifact)
        RG->>DB: INSERT interact_model_artifact<br/>status='validated'
        RT-->>CRON: saved challenger (no auto-promote)
    end
    Note over SH: For each new decision, shadow-score challenger<br/>log agreement/disagreement over shadow_weeks=2
    ADM->>DB: POST /models/{id}/promote<br/>(after agreement_on_auto_approve=0.99 + lower logloss/brier)
    RG->>DB: retire old champion, promote new (transactional)
    RG->>CH: PIPE_model_artifact_audit (best-effort)

7. Database Schema¶

erDiagram
    interact_recommendation ||--|| interact_recommendation_feature : "1 latest snapshot"
    interact_recommendation ||--o{ interact_recommendation_execution : "0..n ERP results"
    interact_recommendation ||--o{ interact_recommendation_outcome : "1 per horizon (4w,13w)"
    interact_rule_set ||--o{ interact_auto_approval_rule : "versioned children"
    interact_recommendation }o--|| interact_auto_approval_rule : "auto_rule_id (matched)"
    interact_recommendation }o--|| interact_model_artifact : "model_id (champion)"
    interact_planner_profile }o--o{ interact_recommendation : "planner_id"

    interact_recommendation {
        bigserial id PK
        bigint pipeline_id
        date archive_date
        date as_of_date
        text recommendation_type
        text recommendation_id
        text source_system
        bigint item_id
        bigint site_id
        bigint source_site_id
        bigint supplier_id
        text unique_id
        jsonb recommended_payload
        text status
        text decision_source
        bigint auto_rule_id
        bigint model_id
        float confidence
        text confidence_band
        text rationale
        jsonb top_features
        bigint_array similar_history_ids
        timestamptz decided_at
        text planner_id
    }
    interact_recommendation_feature {
        bigserial id PK
        bigint recommendation_id FK
        bigint pipeline_id
        date archive_date
        text feature_set_version
        jsonb inventory_features
        jsonb demand_features
        jsonb supply_features
        jsonb business_features
        jsonb recommendation_features
        jsonb planner_history_features
        jsonb feature_vector_dense
    }
    interact_rule_set {
        bigserial id PK
        text name
        int version
        int parent_version
        text yaml_text
        text yaml_hash
        text feature_set_version
        text status
        text created_by
        timestamptz activated_at
        timestamptz retired_at
    }
    interact_auto_approval_rule {
        bigserial id PK
        bigint rule_set_id FK
        text name
        text recommendation_type
        jsonb condition_expr
        text action
        jsonb action_payload
        int_array scope_segments
        int_array scope_pipelines
        bool is_active
        int priority
        text governance_state
        text explanation_template
        int rule_version
        text yaml_source_hash
        bigint fire_count
    }
    interact_model_artifact {
        bigserial id PK
        text name
        text algorithm
        text feature_set_version
        bigint rule_set_id
        timestamptz trained_at
        int train_rows
        int validation_rows
        jsonb metrics
        jsonb calibration_params
        text model_uri
        text model_hash
        jsonb feature_manifest
        text status
        text created_by
    }
    interact_recommendation_execution {
        bigserial id PK
        bigint recommendation_id FK
        bigint pipeline_id
        timestamptz executed_at
        text exec_status
        text erp_ref
        text error_msg
        text executor
    }
    interact_recommendation_outcome {
        bigserial id PK
        bigint recommendation_id FK
        bigint pipeline_id
        timestamptz measured_at
        smallint measurement_weeks
        jsonb baseline_kpi
        jsonb actual_kpi
        jsonb deltas
        float reward_signal
        text source
    }
    interact_planner_profile {
        text planner_id PK
        bigint total_decisions
        float approval_rate
        float override_rate
        float reject_rate
        jsonb per_type_rates
        int avg_decision_ms
        timestamptz last_refreshed_at
    }

PostgreSQL tables (additive to `pg_history_schema.sql`)¶

Table	Purpose	Key constraints
`interact_recommendation`	typed intake + decision state	`UNIQUE(pipeline_id, recommendation_type, recommendation_id)`, status CHECK
`interact_recommendation_feature`	feature store (latest per rec)	FK cascade on rec delete, JSONB per group
`interact_rule_set`	versioned YAML rule sets	`UNIQUE(name, version)`, status CHECK
`interact_auto_approval_rule`	(extended) + `rule_set_id`, `governance_state`, `explanation_template`	`governance_state` default `'active'` (migration-safe)
`interact_model_artifact`	champion/challenger registry	`UNIQUE(feature_set_version) WHERE status='champion'`
`interact_recommendation_execution`	ERP webhook results	`UNIQUE(erp_ref) WHERE erp_ref IS NOT NULL` (idempotent)
`interact_recommendation_outcome`	split-horizon KPI deltas + reward	one row per `(rec, horizon)`
`interact_planner_profile`	derived planner stats	`PK(planner_id)`, nightly upsert

Migration safety

The schema uses CREATE TABLE IF NOT EXISTS / ALTER TABLE ... ADD COLUMN IF NOT EXISTS so it is fully idempotent. The governance_state migration defaults to 'active' and runs an UPDATE ... WHERE is_active = TRUE AND governance_state = 'draft' so pre-existing firing rules are not silently deactivated.

ClickHouse mirrors¶

Append-only MergeTree mirrors (ch_auto_approval_schema.sql):

Table	Partition	Order by
`PIPE_recommendation`	`(pipeline_id, toYYYYMM(archive_date))`	`(pipeline_id, as_of_date, item_id, site_id, recommendation_type)`
`PIPE_recommendation_feature`	`(pipeline_id, toYYYYMM(archive_date))`	`(pipeline_id, archive_date, recommendation_id)`
`PIPE_recommendation_outcome`	`(pipeline_id, toYYYYMM(measured_at))`	`(pipeline_id, measured_at, recommendation_id, measurement_weeks)`
`PIPE_model_artifact_audit`	—	`(model_id, promoted_at)`

ClickHouse DEFAULT toMonday(today()) is a safety-net only

Per AGENTS.md, the CH DEFAULT cannot query the planning_date table. Every Python insert passes an explicit archive_date = planning_monday(pid). Relying on the CH DEFAULT is a bug.

8. Feature Store Design¶

Six feature groups, persisted as JSONB columns in interact_recommendation_feature. The frozen manifest (features/manifest.py, feature_set_version='v1') is shipped with every model artifact so inference and training use the identical column ordering.

flowchart LR
    subgraph Sources
        CH1[PIPE_supply_inventory_projection]
        CH2[PIPE_on_hand_snapshot]
        CH3[PIPE_forecast_point_values]
        CH4[PIPE_series_backtest_metrics]
        CH5[PIPE_series_characteristics]
        CH6[PIPE_fitted_distributions]
        CH7[PIPE_lead_time_actuals]
        CH8[PIPE_service_actuals]
        PG1[master.item<br/>price/cost/unit_cost/abc_class]
        PG2[interact_planner_profile]
        PG3[interact_recommendation_approval<br/>similar-history rate]
        PAY[recommended_payload]
    end

    subgraph Groups
        INV[inventory<br/>7 features]
        DEM[demand<br/>9 features]
        SUP[supply<br/>5 features]
        BUS[business<br/>5 features]
        REC[recommendation<br/>6 features]
        PLA[planner_history<br/>5 features]
    end

    CH1 --> INV
    CH2 --> INV
    CH3 --> DEM
    CH4 --> DEM
    CH5 --> DEM
    CH6 --> DEM
    CH7 --> SUP
    CH8 --> SUP
    PG1 --> BUS
    PAY --> BUS
    PAY --> REC
    PG2 --> PLA
    PG3 --> PLA

    INV --> VEC
    DEM --> VEC
    SUP --> VEC
    BUS --> VEC
    REC --> VEC
    PLA --> VEC

    VEC[feature_vector_dense<br/>35 numeric/categorical<br/>+ 8 one-hot recommendation_type<br/>= 43 columns]
    VEC -->|ordered by manifest| MODEL[LightGBM / XGBoost]

Group	Features
`inventory`	`stock_on_hand`, `stock_in_transit`, `safety_stock`, `projected_inventory`, `projected_shortage_qty`, `projected_shortage_days`, `days_of_supply`
`demand`	`forecast`, `forecast_accuracy` (1−smape), `forecast_bias`, `demand_trend`, `demand_variability`, `shortage_probability`, `dol_q10`, `dol_q50`, `dol_q90`
`supply`	`leadtime_mean`, `leadtime_std`, `supplier_otif`, `supplier_risk_score`, `supplier_id_bucket` (MD5 hash → 64 buckets)
`business`	`abc_class`, `customer_priority`, `inventory_value`, `margin_impact`, `revenue_at_risk`
`recommendation`	`quantity_change_percent`, `date_change_days`, `inventory_impact`, `service_level_impact`, `working_capital_impact`, `pipeline_id`
`planner_history`	`planner_id`, `planner_approval_rate`, `planner_override_rate`, `planner_reject_rate`, `similar_past_approval_rate`

Frozen manifest

feature_vector_dense ordering is fixed by _FEATURES in manifest.py. Any change → new feature_set_version → mandatory retrain. The manifest ships inside each model artifact (feature_manifest JSONB column) so a stale champion can never be scored against a newer vector.

NaN-safe

_clean_nan in builder.py converts NaN/Inf to JSON null before persistence, so PG/CH never store non-finite floats. Missing CH data → feature defaults (never a crash — feature build is best-effort).

9. Rule Engine (YAML-driven, versioned, governance-gated)¶

Rules are authored as human-readable YAML, parsed into the JSON predicate AST that historization.auto_approval.evaluate_predicate already consumes. This keeps a single canonical evaluator and makes rule sets diffable and hash-tied.

YAML schema¶

name: low-value-c-items
feature_set_version: v1
rules:
  - name: auto-approve low value C-class CREATE_PO
    recommendation_type: CREATE_PO
    action: auto_approve           # auto_approve|auto_reject|auto_modify|mandatory_review|assign_planner
    priority: 100                  # higher = evaluated first
    scope:
      pipelines: [4, 7]
      segments: [2, 3]
    condition:                     # predicate AST in YAML form
      and:
        - "==": [abc_class, C]
        - "<=": [inventory_value, 5000]
        - "<=": [abs(quantity_change_percent), 10]
    action_payload:                # optional, e.g. auto_modify cap
      cap_pct: 10
    explanation: |
      Auto-approved because {{abc_class}}-class item with inventory value
      {{inventory_value|round(0)}} and small {{quantity_change_percent|round(1)}}%
      adjustment.

Supported condition operators: and / or / not / == / != / < / <= / > / >= / in / not_in. Use {const: <str>} to force a string literal that collides with a variable name.

Governance state machine¶

stateDiagram-v2
    [*] --> draft: POST /rule-sets (new version)
    draft --> proposed: activate
    proposed --> draft: activate (revoke)
    proposed --> approved: activate
    approved --> active: activate
    approved --> proposed: activate (revoke)
    active --> retired: activate (retire previous on new active)
    retired --> [*]
    note right of active
        Only governance_state='active' rules fire.
        Activating a new 'active' rule set
        retires the previous active set of
        the same name.
    end note

Transitions are validated by yaml_parser.governance_transition(), which raises on illegal jumps (e.g. retired → active). The transition table:

From	Allowed to
`draft`	`proposed`
`proposed`	`approved`, `draft`
`approved`	`active`, `proposed`
`active`	`retired`
`retired`	— (terminal)

Rule evaluation algorithm¶

Load eligible rules: recommendation_type matches, is_active = TRUE, governance_state = 'active', rule_set_id matches (or NULL for legacy rules). Ordered priority DESC, id ASC.
Scope filter: scope_segments / scope_pipelines must include the recommendation's context (missing context value → out-of-scope if the rule restricts that dimension).
First match wins (priority order). The matched rule's explanation_template is rendered against the scope to produce the rationale. fire_count and last_evaluated_at are bumped (best-effort).
Planner routing runs as a special pre-pass: action='assign_planner' rules are evaluated first (locked decision #2); the first match sets planner_id before features are built.

No fork of historization.auto_approval

_evaluate_predicate lazily imports evaluate_predicate from historization.auto_approval. The rule engine only adds rule-set resolution, the governance gate, planner routing, mandatory-review handling, and template rendering on top.

10. Planner Imitation Model¶

The model is a planner-imitation classifier: it forecasts the probability a planner would approve / reject / modify a recommendation as-is. It is not an optimiser — it imitates observed planner behaviour.

flowchart LR
    HIST[Historical approvals<br/>interact_recommendation_approval] --> BF[backfill.build_training_dataset]
    BF --> X[X: feature_vector_dense rows]
    BF --> Y[y: approve/modify/reject labels]
    X --> SW[sample_weight<br/>inverse frequency]
    X --> SPLIT[time-based split<br/>last 20% = validation]
    Y --> SPLIT
    SPLIT --> TR[train split]
    SPLIT --> VAL[validation split]
    TR --> TRN[trainer.train]
    VAL --> MET[metrics: logloss, top1_acc, brier, recall]
    TRN --> ALG{algorithm}
    ALG -->|lightgbm| LGB[LGBMClassifier<br/>200 trees, 31 leaves]
    ALG -->|xgboost| XGB[XGBClassifier<br/>200 trees, depth 6]
    ALG -->|dummy| DUM[stratified prior baseline]
    LGB --> FIT[fitted model]
    XGB --> FIT
    DUM --> FIT
    FIT --> CAL[isotonic calibration<br/>on P(approve), train preds]
    CAL --> ART[artifact<br/>model.json + calib.json + manifest.json]
    ART --> RG[registry.save_artifact<br/>status=validated]
    RG --> CHAL[challenger]
    CHAL --> SHADOW[shadow score every decision<br/>2 planning weeks]
    SHADOW --> PROMO{agreement_on_auto_approve >= 0.99<br/>AND lower logloss<br/>AND lower brier?}
    PROMO -->|yes| ADM[Admin: POST /models/{id}/promote]
    ADM --> CHA[champion<br/>unique per feature_set_version]
    CHA --> RET[old champion → retired]
    CHA --> INF[inference singleton<br/>hot-reload on promote]

Property	Value
Task	multi-class softmax (3 classes)
Champion	LightGBM (`LGBMClassifier`, 200 trees, 31 leaves, lr=0.05)
Challenger	XGBoost (`XGBClassifier`, 200 trees, depth 6)
Fallback	`dummy` (stratified-prior baseline, no sklearn dep)
Imbalance	`sample_weight = inverse_frequency` (auditable, no SMOTE)
Split	time-based (last 20% by row order); configurable `GroupKFold` by `(planner_id, item_id)`
Calibration	isotonic on `P(approve)` (sklearn-free PAV in `models/calibration.py`)
Metrics	`logloss`, `top1_acc`, `brier`, `brier_calibrated`, per-class `recall`
Scope	one champion per tenant DB (locked decision #6)
Min rows	`model.min_train_rows` (default 200); retrain skipped below this
Lookback	`model.lookback_days` (default 180)

Promotion is not automatic

The nightly retrain only saves a validated/challenger artifact. Promotion to champion requires shadow_weeks=2 of agreement plus lower logloss and brier (config.yaml: promotion). An admin then calls POST /models/{id}/promote, which is transactional and enforced by the uq_model_champion_per_fset unique index.

11. Inference Service¶

models/inference.py is a singleton with hot-reload:

The champion is lazy-loaded on first score() and cached for 300s.
get_champion() reads interact_model_artifact WHERE status='champion' for the current feature_set_version.
_validate_model_uri() confines model loads to the models/artifacts/ directory (path-traversal guard).
invalidate_cache() is called by the /models/{id}/promote endpoint so the next score() re-loads the new champion.
score_many() supports batch scoring (used by bulk ingest paths).
Calibration is applied to the approve-class column, then probabilities are renormalised.

LightGBM, XGBoost, and the dummy baseline are each wrapped in a small _LGBWrapper / _XGBWrapper / _Dummy class exposing a uniform predict_proba(X) interface, so the orchestrator does not branch on algorithm.

12. Explainability Service¶

models/explain.py:explain returns (top_features, similar_history_ids):

SHAP top featureskNN similar historyRationale

shap.TreeExplainer on the approve class for LightGBM/XGBoost boosters.
Multi-class SHAP output is normalised (list-of-arrays or 3D array → sv[:, :, 0]).
Top-k by |contribution|; each FeatureContribution carries the signed SHAP value, the raw feature value, and a direction (positive / negative / neutral).
Fallback: if SHAP is unavailable, gain-based feature importance normalized to [0, 1] (sign unknown → positive).

Cosine similarity over feature_vector_dense, restricted to the same recommendation_type.
Pulls up to 500 candidates from PG (most recent first), ranks in-process.
Returns top-knn_k (default 5) historical recommendation ids.
Interface is swappable for a FAISS index at larger scale — no caller change required.

If a rule matched, its explanation_template is rendered with the scope variables (Jinja-style {{var|round(1)}}).
If the model decided, rules/templates.fallback_rationale() produces a templated explanation from the decision, confidence, top features, recommendation type, and planner id.

Explain is only invoked when decision_source == model or the final decision is review_approve / manual_review — i.e. when a human will actually read it. Rule short-circuits skip the (relatively expensive) SHAP computation.

13. Confidence Layer details¶

engine/confidence.py is deliberately small (75 lines) and pure — no DB, no IO — so it is trivially unit-testable. The decision logic:

def decide(rule_hit, model_score):
    thresholds = get_thresholds()           # config.yaml, hot-reloadable
    auto_thr = thresholds["auto_approve_threshold"]   # 0.99
    review_thr = thresholds["review_threshold"]       # 0.95

    # 1. Rule short-circuit (locked decision #1)
    if rule_hit is not None:
        action = rule_hit.action
        if action == "auto_approve":  return ("auto_approve",  "rule", 1.0, "high")
        if action == "auto_reject":   return ("auto_reject",   "rule", 0.0, "low")
        if action == "auto_modify":   return ("auto_modify",   "rule", None, None)
        if action == "mandatory_review":
            conf = model_score.confidence if model_score else None
            band = _band(conf, auto_thr, review_thr) if conf is not None else "low"
            return ("manual_review", "rule", conf, band)

    # 2. Model decides
    if model_score is None:
        return ("manual_review", "model", None, "low")
    p = model_score.confidence
    band = _band(p, auto_thr, review_thr)
    if p > auto_thr:    return ("auto_approve",   "model", p, band)
    if p > review_thr:  return ("review_approve", "model", p, band)
    return ("manual_review", "model", p, band)

mandatory_review beats a confident model

A mandatory_review rule returns manual_review even if p_approve > 0.99. This is intentional: governance can force human review for any class of recommendation (e.g. all EXPEDITE above a cost threshold) regardless of model confidence.

14. Execution & ERP Webhook¶

Locked decision #3: the default executor marks status; the ERP writes back via webhook. The Executor Protocol stays pluggable.

class Executor(Protocol):
    name: str
    def execute(self, pg_conn, schema, recommendation_db_id, recommendation) -> dict: ...

Executor	Behaviour	Selected via
`StatusFlagExecutor` (default)	`UPDATE ... SET status='executed'`; no ERP call	`config.yaml: executor.impl: status_flag`
Custom (future)	e.g. call ERP via `supply_router` write path	`register_executor(name, impl)` at process start

ERP webhook (idempotent)¶

POST /api/auto-approval/{rec_id}/execution-result
Content-Type: application/json

{
  "exec_status": "success",       // success|failed|skipped|rolled_back
  "erp_ref": "PO-12345",          // dedupe key (uq_exec_erp_ref)
  "error_msg": null,
  "executor": "external_erp"
}

Idempotent on erp_ref: a second POST with the same erp_ref returns {"id": null, "deduplicated": true} (no duplicate row).
On exec_status=success, the recommendation status is set to executed.
On failure, status is left for retry.

15. Outcome Measurement & RL Reward¶

Locked decision #5: split horizons.

Horizon	Weeks	Metrics
Short	4	`service_level_delta`, `shortage_qty_delta`, `expedite_count_delta`, `revenue_at_risk_delta`
Long	13	`working_capital_delta`, `excess_inventory_delta`

engine/outcome.py:measure_outcome:

Reads the recommendation's archive_date (= planning_monday(pid) at decision time) and item_id / site_id.
Snapshots baseline KPIs from PIPE_supply_inventory_projection (week 0), PIPE_service_actuals, PIPE_supply_orders.
For each horizon whose target = archive_date + N weeks ≤ planning_today(pid):
Snapshots actual KPIs at the target week.
Computes deltas (actual − baseline).
Computes reward_signal = weighted sum of relevant deltas (weights in config.yaml: reward).
Inserts one interact_recommendation_outcome row (idempotent — skips if a row already exists for (rec, horizon) unless force=True).
Mirrors to PIPE_recommendation_outcome (CH, best-effort) — this is the RL replay buffer.

Reward weights (default)¶

reward:
  service_level_delta: 1.0
  shortage_qty_delta: -2.0       # shortages heavily penalized
  expedite_count_delta: -0.5
  excess_inventory_delta: -0.3
  working_capital_delta: -0.4
  revenue_at_risk_delta: -1.5

RL extension¶

flowchart LR
    subgraph Today[Production policy]
        RULE[Rule short-circuit] --> POL[confidence.apply_policy]
        MOD[Model champion] --> POL
        POL --> DEC[Decision]
    end

    subgraph Shadow[Future RL shadow policy]
        RLP[Policy.act<br/>registered via register_shadow_policy] --> SH[shadow_score]
        SH --> LOG[decision_source='rl_shadow'<br/>no production effect]
    end

    DEC --> EXEC[execute]
    EXEC --> OUTC[outcome.measure_outcome]
    OUTC --> RW[reward_signal<br/>DefaultReward weighted sum]
    OUTC --> CH[(PIPE_recommendation_outcome<br/>RL replay buffer)]
    RW --> BUF[offline RL training<br/>future]
    BUF --> RLP

    LOG -.->|compare A/B| DEC

engine/rl_hooks.py defines the Policy and RewardFunction Protocols and a DefaultReward that computes the weighted reward persisted today — so the replay buffer (PIPE_recommendation_outcome) is populated before any RL agent exists. RuleHybridPolicy is the current production policy, listed here so the shadow path can be wired symmetrically. Shadow policies log as decision_source='rl_shadow' with no production effect until promoted.

16. Audit Trail Design¶

Three layers, all append-only or transition-logged:

flowchart TD
    subgraph L1[Layer 1 — current state fast query]
        S[interact_recommendation.status<br/>ingested → featured → ... → outcome_measured]
    end
    subgraph L2[Layer 2 — full transition chain]
        EV[interact_planning_event<br/>rec_decide_start, feature_built,<br/>rule_evaluated, model_scored,<br/>decision_emitted, ...]
    end
    subgraph L3[Layer 3 — longitudinal analytics + RL buffer]
        CHR[PIPE_recommendation]
        CHF[PIPE_recommendation_feature]
        CHO[PIPE_recommendation_outcome]
        CHA[PIPE_model_artifact_audit]
    end

    ORC[orchestrator.decide] -->|UPDATE status| S
    ORC -->|INSERT event per step| EV
    ORC -->|bulk_insert_ch| CHR
    ORC -->|persist_features → mirror| CHF
    OUTC[measure_outcome] -->|bulk_insert_ch| CHO
    PROMO[promote model] -->|audit| CHA

    QUERY[GET /{id}/audit] --> EV
    QUERY --> S

Layer	Store	Use
1. Status column	`interact_recommendation.status`	fast current-state query (`/pending`, `/{id}`)
2. Planning events	`interact_planning_event`	full transition chain with payload snapshot (`/{id}/audit`)
3. CH mirrors	`PIPE_recommendation*`, `PIPE_model_artifact_audit`	longitudinal analytics, cross-pipeline dashboards, RL replay buffer

Every decide() step emits a best-effort event via _emit_event (non-fatal: a failure logs a warning and rolls back the event insert, but the decision still completes).

17. REST API Reference¶

All 21 endpoints are mounted at /api/auto-approval in main.py.

#	Method	Path	Auth	Purpose
1	POST	`/ingest`	—	ingest one recommendation + run decision (sync or deferred)
2	POST	`/ingest/bulk`	—	bulk ingest (defaults to deferred)
3	GET	`/pending`	—	planner worklist (risk-first: `confidence ASC`)
4	GET	`/{id}`	—	full detail (rec + features + rationale)
5	POST	`/{id}/decide`	planner/admin	planner final decision (approve/reject/modify)
6	POST	`/{id}/execute`	—	manual execution trigger (idempotent on `erp_ref`)
7	POST	`/{id}/execution-result`	—	external ERP webhook (idempotent on `erp_ref`)
8	POST	`/rule-sets`	admin	upload YAML rule set (new version, `status='draft'`)
9	GET	`/rule-sets`	—	list rule sets (optional `status` filter)
10	GET	`/rule-sets/{id}`	—	rule set detail + child rules
11	POST	`/rule-sets/{id}/activate`	admin	governance transition (draft→proposed→approved→active→retired)
12	POST	`/rule-sets/{id}/simulate`	—	replay historical approvals through each child rule
13	GET	`/models`	—	list model artifacts (optional `status`, `feature_set_version`)
14	POST	`/models/train`	admin	trigger training (synchronous)
15	POST	`/models/{id}/promote`	admin	challenger → champion (transactional)
16	POST	`/explain`	—	SHAP top features + kNN similar history
17	GET	`/outcomes`	—	outcome deltas + reward signals
18	POST	`/{id}/measure-outcome`	—	trigger outcome measurement (idempotent)
19	GET	`/{id}/audit`	—	full event chain from `interact_planning_event`
20	POST	`/features/backfill`	—	rebuild features for historical approvals
21	POST	`/config/reload`	admin	hot-reload `config.yaml` (no redeploy)

Authorization

_require_admin() gates governance endpoints (rule-set upload/activate, model train/promote, config reload).
POST /{id}/decide verifies the caller is the assigned planner_id or an admin (role in ('admin','superadmin')); otherwise HTTP 403.

Every business endpoint guards pipeline_id

require_pipeline_id() is called at entry — pipeline_id or 0 is never allowed (per AGENTS.md). archive_date is always computed via planning_monday(pid), never date.today().

18. Module Layout¶

files/auto_approval/
    __init__.py
    domain.py              # StrEnum types + dataclasses (no DB/IO)
    config.yaml            # thresholds, horizons, reward weights
    config_loader.py       # single source of truth for config (hot-reload)
    features/
        manifest.py        # frozen feature definitions (feature_set_version='v1')
        builder.py         # build_features() — CH PIPE_* + PG master pulls
        backfill.py        # training-set builder from historical approvals
    rules/
        yaml_parser.py     # YAML ↔ JSON predicate AST round-trip + governance_transition
        engine.py          # evaluate_rules() + route_planner() (governance-gated)
        templates.py       # explanation_template renderer + fallback_rationale
    models/
        preprocess.py      # rows_to_matrix, safe_proba (NaN-safe)
        trainer.py         # LightGBM champion / XGBoost challenger / dummy
        calibration.py     # isotonic + Platt (sklearn-free PAV)
        registry.py        # champion/challenger promote, model_uri resolution
        inference.py       # singleton, hot-reload, batch scoring, path confinement
        explain.py         # SHAP TreeExplainer + kNN similar-history
        artifacts/         # m{id}.json + m{id}.calib.json + m{id}.manifest.json
    engine/
        orchestrator.py    # decide() pipeline (10-step state machine)
        confidence.py      # policy: rule short-circuit + mandatory_review + bands
        executor.py        # StatusFlagExecutor (pluggable Executor Protocol)
        outcome.py         # split-horizon KPI deltas + reward_signal
        rl_hooks.py        # Policy / RewardFunction Protocols + DefaultReward

files/api/auto_approval_router.py      # FastAPI router (mounted in main.py)
files/DDL/auto_approval_schema.sql     # PG schema (idempotent, additive)
files/DDL/ch_auto_approval_schema.sql  # CH mirror tables (MergeTree)
files/scripts/retrain_auto_approval.py # nightly retrain entry point
files/tests/auto_approval/             # 14 tests (smoke + train/infer)

19. Project Rules Enforced¶

These rules (from AGENTS.md) are enforced throughout the engine:

planning_today(pipeline_id) / planning_monday(pipeline_id) for all business dates — archive_date, as_of_date, outcome target weeks, retrain label windows. Never date.today() / datetime.now() for business logic. outcome.py uses planning_today(pipeline_id) to decide whether a horizon has elapsed.
require_pipeline_id() at every endpoint entry — no pipeline_id or 0. Recommendation.__post_init__ raises if pipeline_id <= 0.
get_conn() / get_schema() per request, conn.close() in finally — every router handler follows this pattern; the orchestrator receives an already-open connection.
NaN-safe JSON — _clean_nan in builder.py, router.py, and _NaNSafeJSONResponse convert NaN/Inf to null before serialization.
Cascade delete — all new tables (interact_recommendation, _feature, _execution, _outcome) carry pipeline_id, so utils/cascade_delete.wipe_pipeline_data (dynamic column discovery via information_schema.columns) wipes them automatically on pipeline delete. No registry update needed.
Authorization — _require_admin() on governance endpoints; planner verification on /decide.
No Docker — local dev stack only (localhost:5173 / localhost:8002).

20. Implementation Roadmap (Phases A–F)¶

Phase	Scope	Status
A — Foundations	`domain.py`, PG schema (`auto_approval_schema.sql`), CH mirrors, `config.yaml`/`config_loader.py`, cascade-delete wiring	✅ Done
B — Rule engine + planner routing	`rules/yaml_parser.py`, `rules/engine.py` (governance gate + `route_planner`), `rules/templates.py`, rule-set CRUD + activate + simulate endpoints	✅ Done
C — Feature store + orchestrator	`features/manifest.py`, `features/builder.py`, `features/backfill.py`, `engine/orchestrator.py`, `engine/confidence.py`, ingest + pending + decide + audit endpoints	✅ Done
D — Model + inference + explain	`models/{preprocess,trainer,calibration,registry,inference,explain}.py`, models endpoints, `/explain` endpoint	✅ Done
E — Execution + outcome + RL hooks	`engine/executor.py` (StatusFlagExecutor + webhook), `engine/outcome.py` (split horizons + reward), `engine/rl_hooks.py` (Protocols + DefaultReward), execution-result + measure-outcome + outcomes endpoints	✅ Done
F — Learning loop + hardening	`scripts/retrain_auto_approval.py` (nightly), `features/backfill.py` training-set builder, 14 tests, CH mirror inserts, governance state machine, config hot-reload	✅ Done

Future work (gated, not yet implemented)¶

RL agent — implement a Policy against the PIPE_recommendation_outcome replay buffer; run as rl_shadow and compare A/B against the rule+model policy before any production promotion.
Real ERP executor — implement an Executor that calls the ERP via the supply_router write path; select via config.yaml: executor.impl.
FAISS kNN index — swap the in-process cosine kNN in explain.py for a FAISS index once candidate volume exceeds ~10k per type.
Per-type champions — relax locked decision #6 if a tenant needs type-specialised models (requires a new feature_set_version per type).