Probabilistic Logic

AI Value Computation Pattern

This document describes how to implement probabilistic logic (AI-driven value computation) alongside deterministic rules in LogicBank.

Prerequisites

Read these foundation documents first:

1. docs/training/genai_logic_patterns.md - Import patterns, auto-discovery
2. docs/training/logic_bank_patterns.prompt - Event signatures, logging, Request Pattern
3. docs/training/logic_bank_api.prompt - Deterministic rule APIs

============================================================================= 🤖 AUTOMATIC DETECTION: When to Generate Probabilistic Logic =============================================================================

When you see these patterns in user's natural language prompt, AUTOMATICALLY create complete AI value computation infrastructure:

Trigger Phrases: - "use AI to select/choose..." - "optimize using AI..." - "AI should decide..." - "intelligently select based on..." - "considering [multiple factors] choose..." - "IF [condition] THEN use AI to [select/compute]..."

When triggered, automatically generate: - SysXxxReq model (if not exists) + Alembic migration - logic/logic_discovery/check_credit.py (business logic) - logic/logic_discovery/ai_requests/supplier_selection.py (AI handler) - config/ai_test_context.yaml (test context) - Update ui/admin/admin.yaml

============================================================================= 🚨 CRITICAL FIXES FROM PRODUCTION DEBUGGING (Nov 16-21, 2025) =============================================================================

These errors occurred during demo prep and production testing and MUST be avoided:

ERROR 1: Circular Import - "Session is already flushing"

Applies to main logic/declare_logic.py ONLY, not discovery files.

❌ Problem: Importing LogicBank at module level in MAIN logic/declare_logic.py

# In logic/declare_logic.py (main file)
from logic_bank.logic_bank import Rule  # ❌ At module level in main file

✅ Solution for main declare_logic.py: Import inside function

# In logic/declare_logic.py (main file)
from database import models  # ✅ At module level

def declare_logic():
    from logic_bank.logic_bank import Rule  # ✅ Inside function in main file

✅ Discovery files (logic_discovery/*.py) are SAFE with module-level imports:

# In logic/logic_discovery/check_credit.py or supplier_selection.py
from logic_bank.logic_bank import Rule  # ✅ Safe in discovery files
from logic_bank.exec_row_logic.logic_row import LogicRow  # ✅ Safe in discovery files  
from database import models  # ✅ Preferred pattern

def declare_logic():
    # Rules here

ERROR 2: Auto-Discovery Structure Requirements

⚠️ IMPORTANT: logic/logic_discovery/auto_discovery.py is AUTO-GENERATED by ApiLogicServer - It is ALREADY CORRECT in all new projects (handles recursion + skips init.py) - ❌ DO NOT modify auto_discovery.py - ✅ DO create logic files in proper structure that auto-discovery will find

✅ What auto_discovery.py does (already built-in): - Recursively scans logic_discovery/ and all subdirectories - Finds all .py files except auto_discovery.py and init.py - Imports each file and calls declare_logic() function - Works with nested directories like ai_requests/, validation/, etc.

✅ Your responsibility (what Copilot generates):

logic/logic_discovery/
  check_credit.py              # Has declare_logic() function
  ai_requests/                 # Subdirectory
    __init__.py                # Empty file (makes it a package)
    supplier_selection.py      # Has declare_logic() function

❌ Common mistake: Putting logic in init.py - auto_discovery.py skips init.py files (by design) - Always create separate .py files with declare_logic() functions

ERROR 3: Path Resolution for YAML Files

❌ Problem: Path(file).parent creates relative path

context_file = config_dir / 'ai_test_context.yaml'
if context_file.exists():  # ❌ May fail on relative paths

✅ Solution: Use .resolve() for absolute paths

current_file = Path(__file__).resolve()  # ✅ Absolute path
project_root = current_file.parent.parent.parent.parent
context_file = project_root / 'config' / 'ai_test_context.yaml'
if context_file.exists():
    with open(str(context_file), 'r') as f:  # ✅ Convert to string

ERROR 4: Missing is_deleted() Check in Early Events (Nov 21, 2025)

❌ Problem: Early events fire on delete, but old_row is None

def set_item_unit_price_from_supplier(row: models.Item, old_row: models.Item, logic_row):
    # Process on insert OR when product_id changes
    if not (logic_row.is_inserted() or row.product_id != old_row.product_id):  # ❌ CRASH on delete
        return

✅ Solution: Check is_deleted() FIRST, before accessing old_row

def set_item_unit_price_from_supplier(row: models.Item, old_row: models.Item, logic_row):
    from logic.logic_discovery.ai_requests.supplier_selection import get_supplier_selection_from_ai

    # Skip on delete (old_row is None) - CHECK THIS FIRST
    if logic_row.is_deleted():
        return

    # Now safe to access old_row
    if not (logic_row.is_inserted() or row.product_id != old_row.product_id):
        return

Rule: ALL early events that access old_row MUST check is_deleted() first.

ERROR 5: Incomplete Audit Trail - Empty request and Brief reason Fields (Nov 21, 2025)

❌ Problem: Request and reason fields not fully populated with actual data

# In wrapper function
supplier_req.request = f"Select optimal supplier for {product_name}"  # ❌ Generic, no context
# In AI handler  
row.reason = "Test context selection"  # ❌ Missing details

What happens: SysSupplierReq records lack actionable audit trail Impact: Cannot debug AI decisions, no visibility into what candidates were considered Business problem: Compliance, explainability, debugging impossible

✅ Solution: Populate request with FULL context in AI handler (where data exists)

# In AI handler (select_supplier_via_ai) - NOT in wrapper
def select_supplier_via_ai(row: models.SysSupplierReq, old_row, logic_row: LogicRow):
    """
    Populate request and reason fields with COMPLETE information:
    - request: Full context (product, candidates with prices, world conditions)
    - reason: Decision details (selected supplier name, price, full AI explanation)
    """
    product = row.product
    suppliers = product.ProductSupplierList if product else []

    # Build candidate summary for request field
    candidate_summary = ', '.join([
        f"{s.supplier.name if s.supplier else 'Unknown'}(${s.unit_cost})" 
        for s in suppliers
    ])

    # TEST CONTEXT case
    if test_context:
        world = test_context.get('world_conditions', 'normal conditions')
        row.request = f"Select supplier for {product.name}: Candidates=[{candidate_summary}], World={world}"
        row.reason = f"TEST MODE: Selected {selected_supplier.supplier.name} (${selected_supplier.unit_cost}) - world: {world}"

    # AI CALL case
    elif api_key:
        # Populate BEFORE calling AI
        row.request = f"AI Prompt: Product={product.name}, World={world_conditions}, Candidates={len(candidate_data)}: {candidate_summary}"

        # After AI responds
        supplier_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
        row.reason = f"AI: {supplier_name} (${selected_supplier.unit_cost}) - {ai_result.get('reason', 'No reason provided')}"

    # FALLBACK case
    else:
        row.request = f"Select supplier for {product.name}: Candidates=[{candidate_summary}] - NO API KEY"
        fallback_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
        row.reason = f"Fallback: {fallback_name} (${selected_supplier.unit_cost}) - minimum cost (no API key)"

Key Points: - ✅ request shows WHAT was asked (product, all candidates, world context) - ✅ reason shows WHY decision was made (selected supplier details, AI explanation) - ✅ Both fields populated in AI handler (has access to all data) - ✅ Includes supplier NAMES and PRICES (not just IDs) - ✅ Different patterns for test/AI/fallback modes - ❌ DO NOT populate in wrapper (doesn't have candidate data)

Wrapper function should:

def get_supplier_selection_from_ai(product_id: int, item_id: int, logic_row: LogicRow):
    supplier_req_logic_row = logic_row.new_logic_row(models.SysSupplierReq)
    supplier_req = supplier_req_logic_row.row

    # Set parent context (FK links)
    # Note: request/reason populated by AI event handler with actual data
    supplier_req.product_id = product_id
    supplier_req.item_id = item_id

    # Insert triggers AI handler which populates request/reason
    supplier_req_logic_row.insert(reason="AI supplier selection request")

    # Log results for visibility
    logic_row.log(f"AI Request: {supplier_req.request}")
    logic_row.log(f"AI Results: supplier_id={supplier_req.chosen_supplier_id}, price={supplier_req.chosen_unit_price}, reason={supplier_req.reason}")

    return supplier_req

See docs/training/genai_logic_patterns.md for complete patterns.

============================================================================= ⚡ PATTERN: Early Event with Wrapper Function =============================================================================

The Pattern

When user says "Use AI to Set field by finding optimal ":

1. Early event on receiver - Rule.early_row_event(on_class=models.Item, calling=set_item_unit_price_from_supplier)
2. Event calls wrapper - Wrapper hides Request Pattern complexity
3. Wrapper returns object - Returns populated request object (not scalar)
4. Event extracts values - row.unit_price = req.chosen_unit_price

Fallback Strategy

CRITICAL: AI rules need fallback logic for cases when AI shouldn't/can't run.

Strategy: Reasonable Default → Fail-Fast

1. Check for reasonable default: Copy from parent field with matching name
2. If no obvious default: Insert NotImplementedError with TODO_AI_FALLBACK marker
3. Never silently fail: Force developer decision at generation time, not runtime

Benefits: - ✅ Prevents silent production failures - ✅ Code won't run until developer addresses edge cases
- ✅ Clear markers for what needs attention - ✅ Works in dev/test, fails explicitly before production

For multi-value AI results: Apply per-field fallback strategy. Common: copy from parent matching field names. For fields with no obvious fallback, use TODO_AI_FALLBACK.

Complete Example

Natural Language

Use AI to Set Item field unit_price by finding the optimal Product Supplier 
based on cost, lead time, and world conditions

IF Product has no suppliers, THEN copy from Product.unit_price

Implementation

File: logic/logic_discovery/check_credit.py

"""
Check Credit Use Case - Business Logic Rules

Natural Language Requirements:
1. The Customer's balance is less than the credit limit
2. The Customer's balance is the sum of the Order amount_total where date_shipped is null
3. The Order's amount_total is the sum of the Item amount
4. The Item amount is the quantity * unit_price
5. The Product count suppliers is the sum of the Product Suppliers
6. Use AI to Set Item field unit_price by finding the optimal Product Supplier
   based on cost, lead time, and world conditions

version: 3.0
date: November 21, 2025
source: docs/training/probabilistic_logic.prompt
"""

from logic_bank.logic_bank import Rule
from database import models

def declare_logic():
    # Other deterministic rules...
    Rule.early_row_event(on_class=models.Item, calling=set_item_unit_price_from_supplier)

def set_item_unit_price_from_supplier(row: models.Item, old_row: models.Item, logic_row):
    """
    Early event: Sets unit_price using AI if suppliers exist, else uses fallback.

    Fires on insert AND when product_id changes (same semantics as copy rule).
    """
    from logic.logic_discovery.ai_requests.supplier_selection import get_supplier_selection_from_ai

    # Skip on delete (old_row is None) - CRITICAL: Check this FIRST
    if logic_row.is_deleted():
        return

    # Process on insert OR when product_id changes
    if not (logic_row.is_inserted() or row.product_id != old_row.product_id):
        return

    product = row.product

    # FALLBACK LOGIC when AI shouldn't/can't run:
    # Strategy: Try reasonable default (copy from parent matching field), else fail-fast
    if product.count_suppliers == 0:
        # Reasonable default: copy from parent.unit_price (matching field name)
        if hasattr(product, 'unit_price') and product.unit_price is not None:
            logic_row.log(f"No suppliers for {product.name}, using product default price")
            row.unit_price = product.unit_price
            return
        else:
            # No obvious fallback - fail-fast with explicit TODO
            raise NotImplementedError(
                "TODO_AI_FALLBACK: Define fallback for Item.unit_price when no suppliers exist. "
                "Options: (1) Use a default constant, (2) Leave NULL if optional, "
                "(3) Raise error if required field, (4) Copy from another source"
            )

    # Product has suppliers - call AI wrapper
    logic_row.log(f"Product {product.name} has {product.count_suppliers} suppliers, requesting AI selection")
    supplier_req = get_supplier_selection_from_ai(
        product_id=row.product_id,
        item_id=row.id,
        logic_row=logic_row
    )

    # Extract AI-selected value(s)
    row.unit_price = supplier_req.chosen_unit_price

File: logic/logic_discovery/ai_requests/init.py

# Empty file - makes this a Python package

File: logic/logic_discovery/ai_requests/supplier_selection.py

⚠️ IMPORTANT: There is NO pre-built populate_ai_values() utility in LogicBank. You must implement the AI selection logic directly as shown below.

"""
AI Supplier Selection - Probabilistic Logic Handler

This module implements AI-driven supplier selection based on cost, lead time,
and world conditions. It uses the Request Pattern for full audit trails.

See: https://apilogicserver.github.io/Docs/Logic-Using-AI/

version: 3.0
date: November 21, 2025
source: docs/training/probabilistic_logic.prompt
"""

from logic_bank.exec_row_logic.logic_row import LogicRow
from logic_bank.logic_bank import Rule
from database import models
from decimal import Decimal
import os

def declare_logic():
    """
    Register early event on SysSupplierReq to populate chosen_* fields via AI.

    This Request Pattern approach provides full audit trails and separation of concerns.
    See: https://apilogicserver.github.io/Docs/Logic/#rule-patterns
    """
    Rule.early_row_event(on_class=models.SysSupplierReq, calling=select_supplier_via_ai)

def select_supplier_via_ai(row: models.SysSupplierReq, old_row, logic_row: LogicRow):
    """
    Early event (called via insert from wrapper) to populate chosen_* fields via AI.

    This AI handler gets called automatically when SysSupplierReq is inserted,
    populating AI Results: chosen_supplier_id and chosen_unit_price.

    Strategy:
    1. Load test context for INPUT conditions (world conditions like "Suez Canal blocked")
    2. Always try AI with those conditions
    3. If no API key or API fails, use fallback (min cost)
    """
    if not logic_row.is_inserted():
        return

    # Get candidates (suppliers for this product)
    product = row.product
    suppliers = product.ProductSupplierList if product else []

    if not suppliers:
        row.request = f"Select supplier for {product.name if product else 'unknown product'} - No suppliers available"
        row.reason = "No suppliers exist for this product"
        logic_row.log("No suppliers available for AI selection")
        row.fallback_used = True
        return

    # Load test context for world conditions (not for predetermined supplier selection)
    from pathlib import Path
    import yaml

    current_file = Path(__file__).resolve()
    project_root = current_file.parent.parent.parent.parent
    context_file = project_root / 'config' / 'ai_test_context.yaml'

    test_context = {}
    if context_file.exists():
        with open(str(context_file), 'r') as f:
            test_context = yaml.safe_load(f) or {}

    world_conditions = test_context.get('world_conditions', 'normal conditions')

    selected_supplier = None

    # Try AI (check for API key)
    if True:  # Always try AI unless no key
        api_key = os.getenv("APILOGICSERVER_CHATGPT_APIKEY")
        if api_key:
            try:
                # Call OpenAI API with structured prompt
                from openai import OpenAI
                import json

                client = OpenAI(api_key=api_key)

                # Build candidate data for prompt - include ALL supplier fields for AI decision
                candidate_data = []
                for supplier in suppliers:
                    supplier_obj = supplier.supplier
                    candidate_data.append({
                        'supplier_id': supplier.supplier_id,
                        'supplier_name': supplier_obj.name if supplier_obj else 'Unknown',
                        'supplier_region': supplier_obj.region if supplier_obj else None,
                        'supplier_contact': supplier_obj.contact_name if supplier_obj else None,
                        'supplier_phone': supplier_obj.phone if supplier_obj else None,
                        'supplier_email': supplier_obj.email if supplier_obj else None,
                        'unit_cost': float(supplier.unit_cost) if supplier.unit_cost else 0.0,
                        'lead_time_days': supplier.lead_time_days if hasattr(supplier, 'lead_time_days') else None,
                        'supplier_part_number': supplier.supplier_part_number if hasattr(supplier, 'supplier_part_number') else None
                    })

                prompt = f"""
You are a supply chain optimization expert. Select the best supplier from the candidates below.

World Conditions: {world_conditions}

Optimization Goal: fastest reliable delivery while keeping costs reasonable

Candidates:
{yaml.dump(candidate_data, default_flow_style=False)}

Respond with ONLY valid JSON in this exact format (no markdown, no code blocks):
{{
    "chosen_supplier_id": <id>,
    "chosen_unit_price": <price>,
    "reason": "<brief explanation>"
}}
"""

                # Populate request field with actual prompt summary including key fields
                candidate_summary = ', '.join([
                    f"{c['supplier_name']}(${c['unit_cost']}, {c['supplier_region'] or 'unknown region'}, {c['lead_time_days'] or '?'}days)" 
                    for c in candidate_data
                ])
                row.request = f"Select supplier for {product.name}: Candidates=[{candidate_summary}], World={world_conditions}"

                logic_row.log(f"Calling OpenAI API with {len(candidate_data)} candidates, world conditions: {world_conditions}")

                response = client.chat.completions.create(
                    model="gpt-4o-2024-08-06",
                    messages=[
                        {"role": "system", "content": "You are a supply chain expert. Respond with valid JSON only."},
                        {"role": "user", "content": prompt}
                    ],
                    temperature=0.7
                )

                response_text = response.choices[0].message.content.strip()
                logic_row.log(f"OpenAI response: {response_text}")

                # Parse JSON response
                ai_result = json.loads(response_text)

                # Find the selected supplier
                selected_supplier = next((s for s in suppliers if s.supplier_id == ai_result['chosen_supplier_id']), None)
                if selected_supplier:
                    supplier_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
                    row.reason = f"Selected {supplier_name} (${selected_supplier.unit_cost}) - {ai_result.get('reason', 'No reason provided')}"
                    row.fallback_used = False
                else:
                    logic_row.log(f"AI selected invalid supplier_id {ai_result['chosen_supplier_id']}, using fallback")
                    selected_supplier = min(suppliers, key=lambda s: float(s.unit_cost) if s.unit_cost else 999999.0)
                    fallback_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
                    row.reason = f"Fallback: {fallback_name} (${selected_supplier.unit_cost}) - AI returned invalid supplier"
                    row.fallback_used = True

            except Exception as e:
                logic_row.log(f"OpenAI API error: {e}, using fallback")
                selected_supplier = min(suppliers, key=lambda s: float(s.unit_cost) if s.unit_cost else 999999.0)
                fallback_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
                candidate_summary = ', '.join([f"{s.supplier.name if s.supplier else 'Unknown'}(${s.unit_cost})" for s in suppliers])
                row.request = f"Select supplier for {product.name}: Candidates=[{candidate_summary}] - API ERROR"
                row.reason = f"Fallback: {fallback_name} (${selected_supplier.unit_cost}) - API error: {str(e)[:100]}"
                row.fallback_used = True
        else:
            # No API key - use fallback strategy (min cost)
            logic_row.log("No API key, using fallback: minimum cost")
            selected_supplier = min(suppliers, key=lambda s: float(s.unit_cost) if s.unit_cost else 999999.0)
            fallback_name = selected_supplier.supplier.name if selected_supplier.supplier else 'Unknown'
            candidate_summary = ', '.join([f"{s.supplier.name if s.supplier else 'Unknown'}(${s.unit_cost})" for s in suppliers])
            row.request = f"Select supplier for {product.name}: Candidates=[{candidate_summary}] - NO API KEY"
            row.reason = f"Fallback: {fallback_name} (${selected_supplier.unit_cost}) - minimum cost (no API key)"
            row.fallback_used = True

    # Populate AI results
    if selected_supplier:
        row.chosen_supplier_id = int(selected_supplier.supplier_id)  # Must be int for SQLite FK
        row.chosen_unit_price = selected_supplier.unit_cost
        logic_row.log(f"Selected supplier {selected_supplier.supplier_id} with price {selected_supplier.unit_cost}")

def get_supplier_selection_from_ai(product_id: int, item_id: int, logic_row: LogicRow) -> models.SysSupplierReq:
    """
    Wrapper function called from Item (Receiver) early event.

    See: https://apilogicserver.github.io/Docs/Logic-Using-AI/

    1. Creates SysSupplierReq and inserts it (triggering AI event that populates chosen_* fields)
    2. Returns populated object

    This wrapper hides Request Pattern implementation details.
    See https://apilogicserver.github.io/Docs/Logic/#rule-patterns.

    Returns populated SysSupplierReq object with:
    - Standard AI Audit: request, reason, created_on, fallback_used
    - Parent Context Links: item_id, product_id
    - AI Results: chosen_supplier_id, chosen_unit_price
    """
    # 1. Create request row using parent's logic_row
    supplier_req_logic_row = logic_row.new_logic_row(models.SysSupplierReq)
    supplier_req = supplier_req_logic_row.row

    # 2. Set parent context (FK links)
    # Note: request/reason fields populated by AI event handler with actual prompt/candidate data
    supplier_req.product_id = product_id
    supplier_req.item_id = item_id

    # 3. Insert triggers early event which populates AI values (chosen_* fields, request, reason)
    supplier_req_logic_row.insert(reason="AI supplier selection request")

    # 4. Log filled request object for visibility
    logic_row.log(f"AI Request: {supplier_req.request}")
    logic_row.log(f"AI Results: supplier_id={supplier_req.chosen_supplier_id}, price={supplier_req.chosen_unit_price}, reason={supplier_req.reason}")

    # 5. Return populated object (chosen_* fields now set by AI)
    return supplier_req

Key Patterns

Key Implementation Points

Test Context Usage: - Load test context for INPUT conditions (world_conditions like "Suez Canal blocked") - Test context provides CONDITIONS for AI, NOT predetermined outputs - File: config/ai_test_context.yaml - Example: world_conditions: "Suez Canal blocked, use alternate shipping routes"

AI Strategy: - Always try AI if API key exists - Pass world_conditions from test context to AI prompt - AI makes decision based on those conditions

Fallback Strategy: - When no suppliers: Set fallback_used = True, return early - When no API key: Use min cost fallback - When API call fails: Use min cost fallback

Type Handling: - Foreign keys (IDs): Must be int not Decimal - Prices: Can be Decimal - Use float() for comparisons: float(s.unit_cost)

Path Resolution: - Use Path(__file__).resolve() for absolute paths - Navigate up from logic/logic_discovery/ai_requests/ to project root - Then down to config/ai_test_context.yaml

Request Pattern

The wrapper function encapsulates LogicBank's Request Pattern:

# Create using new_logic_row (pass CLASS not instance)
req_logic_row = logic_row.new_logic_row(models.SysXxxReq)

# Access instance via .row property
req = req_logic_row.row

# Set context fields
req.context_id = some_value

# Insert triggers early event handler
req_logic_row.insert(reason="...")

# Return populated object
return req

Request Table Structure

Standard AI Audit (constant for all requests)

id = Column(Integer, primary_key=True)
request = Column(String(2000))      # AI prompt sent
reason = Column(String(500))        # AI reasoning
created_on = Column(DateTime)       # Timestamp
fallback_used = Column(Boolean)     # Did AI fail?

Parent Context Links (FKs to triggering entities)

item_id = Column(ForeignKey('item.id'))
product_id = Column(ForeignKey('product.id'))

AI Results (values selected by AI)

chosen_supplier_id = Column(ForeignKey('supplier.id'))
chosen_unit_price = Column(DECIMAL)

============================================================================= 🚨 REQUEST PATTERN FAILURE MODES (Learned from Production Debugging) =============================================================================

CONTEXT: These are REAL failures that occurred during implementation. Each pattern caused server crashes, test failures, or silent bugs.

FAILURE #1: Formula Returns AI Value Directly

# ❌ WRONG - AI handler never fires
Rule.formula(
    derive=models.Item.unit_price,
    as_expression=lambda row: get_ai_supplier_price(row)
)

What happens: Formula executes but AI handler never fires, no audit trail created. Error: Silent failure - no SysSupplierReq records, unit_price has wrong value Why it fails: Formula should PRESERVE value, not COMPUTE it via AI Fix: Use early event pattern (see above)

FAILURE #2: Pass Instance to new_logic_row()

# ❌ WRONG - Pass instance instead of class
supplier_req = models.SysSupplierReq()
supplier_req_logic_row = logic_row.new_logic_row(supplier_req)

What happens: Python tries to call the instance as a function Error: TypeError: 'SysSupplierReq' object is not callable Why it fails: new_logic_row() expects a CLASS, not an instance Fix: Pass the class: logic_row.new_logic_row(models.SysSupplierReq)

FAILURE #3: Access Attributes on LogicRow Instead of .row

# ❌ WRONG - LogicRow doesn't have business attributes
supplier_req_logic_row = logic_row.new_logic_row(models.SysSupplierReq)
item_row.unit_price = supplier_req_logic_row.chosen_unit_price

What happens: LogicRow is a wrapper, not the business object Error: AttributeError: 'LogicRow' object has no attribute 'chosen_unit_price' Why it fails: Business attributes are on .row property, not LogicRow wrapper Fix: Access via .row: supplier_req = supplier_req_logic_row.row

FAILURE #4: Use session.add/flush Directly

# ❌ WRONG - Bypasses LogicBank
supplier_req = models.SysSupplierReq()
logic_row.session.add(supplier_req)
logic_row.session.flush()

What happens: Object added to database but LogicBank events never fire Error: Silent failure - AI handler never executes, no AI selection Why it fails: Direct SQLAlchemy calls bypass LogicBank event chain Fix: Use logic_row.new_logic_row() + explicit .insert()

FAILURE #5: Forget to Copy Result Back

# ❌ WRONG - AI runs but result not propagated
supplier_req_logic_row.insert(reason="AI supplier selection")
# Missing copy: item_row.unit_price = supplier_req.chosen_unit_price

What happens: SysSupplierReq populated correctly but Item.unit_price unset Error: Silent failure - AI works but business logic breaks (unit_price = None) Why it fails: No automatic propagation between tables Fix: Explicitly copy: item_row.unit_price = supplier_req.chosen_unit_price

FAILURE #6: Test Context Checked After API Key

# ❌ WRONG - API key checked first
api_key = os.getenv("APILOGICSERVER_CHATGPT_APIKEY")
if not api_key:
    # Apply fallback - tests never reach test context!
    return

test_context = _load_test_context(logic_row)  # Never reached in tests

What happens: Tests use fallback logic instead of test context Error: Non-deterministic tests, "fallback_used" flag set incorrectly Why it fails: Test context should override API key check Fix: Check test context FIRST, then API key

✅ CORRECT ORDER (test context first):

# Check test context FIRST (for reproducible testing)
test_context = _load_test_context(logic_row)
if test_context and 'selected_supplier_id' in test_context:
    # Use test context
    return

# Then check API key
api_key = os.getenv("APILOGICSERVER_CHATGPT_APIKEY")
if not api_key:
    # Apply fallback
    return

Why: Tests should run consistently without requiring OpenAI API key. Test context is explicitly provided configuration that should override API calls.

============================================================================= 🚨 CRITICAL: Model Relationship Checklist =============================================================================

When adding SysXxxReq audit table, ONLY add relationships where FKs exist:

✅ DO add relationships: 1. Parent models referenced by FKs in SysXxxReq - Example: product_id FK → Add to Product class - Example: item_id FK → Add to Item class - Example: chosen_supplier_id FK → Add to Supplier class

2. SysXxxReq model itself (parent relationships) - Bidirectional: back_populates for standard FKs - Unidirectional: foreign_keys=[...] for non-standard FKs

❌ DO NOT add relationships: 1. Models with no FK to/from SysXxxReq - Example: ProductSupplier has no FK to SysSupplierReq - Example: Order has no FK to SysSupplierReq - Adding relationships without FKs causes NoForeignKeysError

Verification Before Adding Relationship:

# Before adding relationship to Model X, verify:
# 1. Does SysXxxReq have FK to Model X? OR
# 2. Does Model X have FK to SysXxxReq?
# If NO to both → DO NOT add relationship

Common Mistake:

# ❌ WRONG - ProductSupplier has no FK relationship to SysSupplierReq
class ProductSupplier(Base):
    SysSupplierReqList : Mapped[List["SysSupplierReq"]] = relationship(...)
# This will cause: NoForeignKeysError at server startup

Correct Pattern:

# ✅ CORRECT - Only add where FK exists
class Product(Base):  # Has FK from SysSupplierReq.product_id
    SysSupplierReqList : Mapped[List["SysSupplierReq"]] = relationship(back_populates="product")

class Item(Base):  # Has FK from SysSupplierReq.item_id
    SysSupplierReqList : Mapped[List["SysSupplierReq"]] = relationship(back_populates="item")

# ✅ DO NOT add relationship to Supplier for chosen_supplier_id
#    - This is an AI result field (not standard parent-child relationship)
#    - Access via SysSupplierReq.chosen_supplier (unidirectional) is sufficient
#    - Adding reverse relationship causes NoForeignKeysError

# ✅ DO NOT add relationship to ProductSupplier (no FK exists)

OpenAI API (v1.0.0+)

CRITICAL: Use modern OpenAI API

❌ OLD API (deprecated, will fail):

import openai
openai.api_key = api_key
response = openai.ChatCompletion.create(...)  # ❌ Not supported in openai>=1.0.0

✅ NEW API (correct pattern):

from openai import OpenAI

client = OpenAI(api_key=api_key)
response = client.chat.completions.create(
    model="gpt-4o-2024-08-06",
    messages=[...]
)

Common Pitfalls

Pass CLASS to new_logic_row, not instance:

# ❌ WRONG
req = models.SysXxxReq()
logic_row.new_logic_row(req)  # TypeError

# ✅ CORRECT
logic_row.new_logic_row(models.SysXxxReq)

Access attributes via .row property:

# ❌ WRONG
req_logic_row.product_id = 123  # AttributeError

# ✅ CORRECT
req = req_logic_row.row
req.product_id = 123

Use LogicBank insert, not SQLAlchemy:

# ❌ WRONG
session.add(req)
session.flush()  # Bypasses LogicBank

# ✅ CORRECT
req_logic_row.insert(reason="...")  # Triggers events

Decimal handling in AI scoring:

# ❌ WRONG - Decimal × float
cost = supplier.unit_cost  # Returns Decimal
score = cost * 0.5  # TypeError

# ✅ CORRECT - Convert to float first
cost = float(supplier.unit_cost) if supplier.unit_cost else 999999.0
score = cost * 0.5

File Structure

logic/
  logic_discovery/
    check_credit.py              # Business logic with deterministic rules + AI event
    ai_requests/                 # AI handlers directory
      __init__.py                # Python package marker
      supplier_selection.py      # AI handler + wrapper function
  system/
    populate_ai_values.py        # Reusable introspection utility

Multi-Value Pattern

For cases where multiple values are needed:

def assign_multiple_values(row: models.Order, old_row, logic_row):
    """Extract multiple values from AI request."""
    from logic.logic_discovery.ai_requests.supplier_selection import get_supplier_selection_from_ai

    if not logic_row.is_inserted():
        return

    # Call wrapper - returns object
    req = get_supplier_selection_from_ai(
        product_id=row.product_id,
        item_id=row.id,
        logic_row=logic_row
    )

    # Extract multiple values
    row.supplier_id = req.chosen_supplier_id
    row.unit_price = req.chosen_unit_price
    row.lead_time = req.chosen_lead_time

Test Context

Enable reproducible testing via config/ai_test_context.yaml:

world_conditions: 'ship aground in Suez Canal'
selected_supplier_id: 2

Database Model and Alembic Migration Workflow

CRITICAL: Request Pattern requires SysXxxReq audit table in database.

Table Structure Convention

# In database/models.py - Add import at top
import datetime

# Add model class
class SysSupplierReq(Base):
    __tablename__ = "sys_supplier_req"
    _s_collection_name = 'SysSupplierReq'

    # Standard AI Audit
    id = Column(Integer, primary_key=True)
    request = Column(String(2000))           # Full AI prompt
    reason = Column(String(500))             # AI explanation
    created_on = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
    fallback_used = Column(Boolean, default=False)

    # Parent Context Links (FKs to triggering entities)
    item_id = Column(Integer, ForeignKey("item.id"), index=True, nullable=True)
    product_id = Column(Integer, ForeignKey("product.id"), index=True, nullable=False)

    # AI Results (values selected by AI)
    chosen_supplier_id = Column(Integer, ForeignKey("supplier.id"))
    chosen_unit_price = Column(DECIMAL)

    # Relationships (bidirectional)
    item : Mapped["Item"] = relationship(back_populates="SysSupplierReqList")
    product : Mapped["Product"] = relationship(back_populates="SysSupplierReqList")
    chosen_supplier : Mapped["Supplier"] = relationship()

# Add to parent models (Product, Item):
SysSupplierReqList : Mapped[List["SysSupplierReq"]] = relationship(back_populates="product")

Alembic Migration Workflow (Production-Ready)

COPILOT MUST EXECUTE AUTOMATICALLY - No user intervention required:

Step 1: Add model to database/models.py

• Include import datetime at top
• Add SysXxxReq class with all fields
• Add bidirectional relationships to parent models

Step 2: Generate migration

cd database && alembic revision --autogenerate -m "Add sys_supplier_req table"

Creates: database/alembic/versions/xxxxx_add_sys_supplier_req_table.py

Step 3: Clean migration file (CRITICAL - COPILOT MUST DO THIS)

Alembic --autogenerate detects ALL differences between models.py and database.

COPILOT MUST AUTOMATICALLY: 1. Read generated migration file from database/alembic/versions/ 2. KEEP: CREATE TABLE sys_supplier_req statement 3. REMOVE: ANY ALTER TABLE operations on existing tables (unrelated changes) 4. SIMPLIFY: downgrade() to just DROP TABLE 5. Save cleaned migration file

Example cleaned migration:

def upgrade():
    op.create_table('sys_supplier_req',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('item_id', sa.Integer(), nullable=True),
        sa.Column('product_id', sa.Integer(), nullable=False),
        sa.Column('chosen_supplier_id', sa.Integer(), nullable=True),
        sa.Column('chosen_unit_price', sa.DECIMAL(), nullable=True),
        sa.Column('request', sa.String(length=2000), nullable=True),
        sa.Column('reason', sa.String(length=500), nullable=True),
        sa.Column('created_on', sa.DateTime(), nullable=False),
        sa.Column('fallback_used', sa.Boolean(), nullable=True),
        sa.ForeignKeyConstraint(['chosen_supplier_id'], ['supplier.id']),
        sa.ForeignKeyConstraint(['item_id'], ['item.id']),
        sa.ForeignKeyConstraint(['product_id'], ['product.id']),
        sa.PrimaryKeyConstraint('id')
    )

def downgrade():
    op.drop_table('sys_supplier_req')

Step 4: Apply migration

cd database && alembic upgrade head

Step 5: Update ui/admin/admin.yaml

Add SysSupplierReq resource with all fields and relationships.

Step 6: User restarts server

To load new model into SQLAlchemy session.

Complete Automated Workflow

When user provides natural language with AI requirement, COPILOT MUST:

1. ✅ Check if sys_xxx_req table exists in database (not just models.py)
2. ✅ Add model to database/models.py (include datetime import)
3. ✅ Add bidirectional relationships to parent models
4. ✅ Run: cd database && alembic revision --autogenerate -m "..."
5. ✅ AUTOMATICALLY find new migration file in database/alembic/versions/
6. ✅ AUTOMATICALLY read migration file
7. ✅ AUTOMATICALLY clean it (remove ALTER TABLE, keep CREATE TABLE)
8. ✅ AUTOMATICALLY save cleaned migration file
9. ✅ Run: cd database && alembic upgrade head
10. ✅ Update ui/admin/admin.yaml
11. ✅ Generate logic files (check_credit.py, supplier_selection.py)

NO USER INTERVENTION - Copilot handles entire migration workflow.

Common Alembic Issues

Issue: "Target database is not up to date" Solution: Run cd database && alembic stamp head

Issue: "Table already exists" Solution: Database has tables but Alembic history is empty. Use alembic stamp head.

Issue: "No config file 'alembic.ini' found" Solution: Must run alembic commands from database/ directory.

Why Alembic (not raw SQL): - ✅ Version controlled (migration files in git) - ✅ Reversible (alembic downgrade) - ✅ Team-friendly (others run same migration) - ✅ Staged deployment (dev → test → prod) - ✅ Audit trail (history of schema changes)

❌ Raw SQL is demo/testing only - NOT production-ready.

Benefits

• Separation of concerns - Event doesn't know Request Pattern details
• Early event - AI executes before other rules
• Wrapper hides complexity - Request Pattern encapsulated
• Returns object - Caller extracts needed values
• Reusable - Multiple events can call same wrapper
• Testable - Can mock wrapper independently

Complete Checklist

When implementing AI logic:

• [ ] Create request table (SysXxxReq) if needed
• [ ] Add fields: Standard AI Audit, Parent Context Links, AI Results
• [ ] Create logic/logic_discovery/ai_requests/ directory
• [ ] Create __init__.py in ai_requests/
• [ ] Implement AI handler in ai_requests/xxx.py
• [ ] Register early event on SysXxxReq
• [ ] Implement wrapper function (returns object)
• [ ] Call wrapper from receiver event
• [ ] Extract values from returned object
• [ ] Create config/ai_test_context.yaml
• [ ] Update ui/admin/admin.yaml

---

For detailed LogicBank patterns, see docs/training/logic_bank_patterns.prompt For deterministic rules, see docs/training/logic_bank_api.prompt