Data Science, Enrichment & Tracking

State Transitions as Machine Learning Labels

Every entity that flows through a workflow produces a sequence of status transitions. That sequence is a behavioral label — a description of how the entity was processed.

These labels emerge naturally from the state record history. No separate labeling pipeline required. Query the Process_StateLogs table for an entity type, walk the status chains, and you have a labeled dataset ready for model training.

-- Generate labeled training data for failure prediction
SELECT
    sl.ItemEntityKey    AS entity_type,
    sl.ItemKey          AS entity_id,
    COUNT(*)            AS total_attempts,
    SUM(CASE WHEN sl.Status = 3 THEN 1 ELSE 0 END) AS failure_count,
    SUM(CASE WHEN sl.Status = 1 THEN 1 ELSE 0 END) AS success_count,
    MAX(sl.Created)     AS last_attempt_at,
    DATEDIFF(second,
        MIN(sl.Created),
        MAX(CASE WHEN sl.Status = 1 THEN sl.Created END)) AS time_to_success_seconds,
    CASE
        WHEN MAX(sl.Status) = 1 AND COUNT(*) = 1 THEN 'clean'
        WHEN MAX(sl.Status) = 1 AND COUNT(*) > 1 THEN 'retry_success'
        WHEN MAX(sl.Status) IN (2,3) THEN 'unresolved_failure'
    END AS processing_label
FROM Process_StateLogs sl
WHERE sl.ItemEntityKey = 'invoice'
  AND sl.Deleted = 0
GROUP BY sl.ItemEntityKey, sl.ItemKey
ORDER BY last_attempt_at DESC;

Failure Prediction

With labeled training data from state records, you can train a model to predict which entities are likely to fail before processing begins. The model watches incoming entity characteristics (size, type, source, time of day) and predicts the failure probability.

If the predicted failure probability exceeds a threshold, the node can:

• Route the entity to a slower, more careful processing path
• Trigger a pre-processing validation step
• Alert an operator before the entity enters the pipeline
• Apply a different handler with more conservative lock settings

// Model score injected via upstream AI node
var failureProbability = ctx.Input.GetDouble("inferenceScore");

if (failureProbability > 0.85)
{
    // High risk — acquire lock with extra caution, log to monitoring
    _logger.Warning("High failure probability {Score:P0} for {Entity}", failureProbability, entityId);
    // Use extended lock window for high-risk entities (passed via options)
}

bool locked = await DataStateMachine.TryAcquireLockAsync(entityId, entityKey, ct: ct);

Data Enrichment Pipeline

DataStateMachine provides native state management for enrichment workflows. Each enrichment operation is an independent tracked step. The SubItemKey distinguishes enrichment types. The Data field stores the enrichment result. The pipeline can be partially replayed — only failed enrichments need to re-run.

Customer Entity (customerId = ItemEntityKey)
    │
    ├── SubItemKey: "address"     → AddressValidationNode → Processed (normalized address in Data)
    ├── SubItemKey: "credit"      → CreditScoreNode       → Processed (score + bureau in Data)
    ├── SubItemKey: "social"      → SocialGraphNode       → Failed    (API timeout — retry pending)
    ├── SubItemKey: "compliance"  → SanctionScreenNode    → Processed (clear — no matches)
    └── SubItemKey: "industry"    → IndustryClassifier    → Processed (SIC code + confidence)

Query: which enrichments succeeded?
→ SELECT SubItemKey FROM Process_StateLogs
  WHERE ItemEntityKey = 'CUST-1001' AND Status = 1  -- Processed

Data Enhancement Workflows

Data enhancement differs from enrichment — enrichment adds new data (credit score, address validation), enhancement improves existing data (normalise a messy address, standardise a phone number, fix encoding). DataStateMachine handles both identically.

Enhancement nodes use UpsertAsync: each enhancement run creates a new record for the same entity. The SupersededByID chain shows the enhancement history — original value → first pass → second pass → current. You can reconstruct the pre-enhancement data at any point.

// Before: raw customer name "SMITH, JOHN J."
// Enhancement: normalize to "John J. Smith"
await DataStateMachine.UpsertAsync(
    itemKey:       customerId,
    itemEntityKey: "customer-name",
    subItemKey:    "name-normalisation",
    data: JsonSerializer.Serialize(new {
        original:   rawName,
        normalized: normalizedName,
        algorithm:  "Name.NET v2.1",
        confidence: 0.97
    }),
    ct: ct);
// → Creates new row, marks previous row Superseded
// → Full enhancement history preserved in append-only chain

Data Tracking Across Runs

Tracking asks: "What happened to entity X across ALL processing runs, ever?" The append-only state records answer this question directly.

Feature Store Integration

State records can feed a feature store for ML model serving. The processing history of an entity (how many times processed, average processing time, failure rate, last enrichment date) becomes a feature vector that is continuously updated as new records are written.

// Feature extraction service — runs asynchronously on state record writes
public class ProcessingHistoryFeatureExtractor
{
    public async Task<ProcessingFeatureVector> ExtractAsync(
        string entityId, string entityType, CancellationToken ct)
    {
        var records = await _stateLogRepo.QueryAsync(
            itemKey: entityId, itemEntityKey: entityType, ct: ct);

        return new ProcessingFeatureVector
        {
            TotalAttempts    = records.Count,
            SuccessCount     = records.Count(r => r.Status == IdempotencyStatus.Processed),
            FailureCount     = records.Count(r => r.Status == IdempotencyStatus.Failed),
            FailureRate      = records.Count > 0
                ? (double)records.Count(r => r.Status == IdempotencyStatus.Failed) / records.Count
                : 0,
            DaysSinceFirst   = (DateTime.UtcNow - records.Min(r => r.Created)).TotalDays,
            DaysSinceLast    = (DateTime.UtcNow - records.Max(r => r.Created)).TotalDays,
            EnrichmentCount  = records.Count(r => !string.IsNullOrEmpty(r.SubItemKey))
        };
    }
}

Summary — DataStateMachine as a Data Platform