Repository Docs & Knowledge Base

Consolidated Spensara documentation repository (722 Markdown & architecture documents in ../repository)

Built Word Documents (.docx): Spensara Overview (.docx) Capability Assessment (.docx) Brink POS Ingestion Analysis (.docx) The Borg Ingestion Architecture (.docx)

Implementation Specs (157)

Detailed engineering and feature implementation specifications.

implementation-specs/specs/AnomalyDetection/TASK4.2.md 11,290 bytes · Updated Oct 23, 2026 10:44 AM EDT

Raw File

TASK 4.2: Implement MonitorRunOrchestrator

Summary

Implement the main orchestrator that coordinates the complete anomaly detection flow: load spec, run agent, emit alerts, and record history.

Dependencies

TASK 2.1 (IAnomalyDetectorAgent)
TASK 2.2 (AnomalyDetectorAgent)
TASK 3.1 (IMonitorSpecRepository)
TASK 4.1 (IMonitorRunOrchestrator interface)
TASK 4.3 (IAlertEmitter)
TASK 4.4 (IMonitorRunRepository)

Files to Create

1. `backend/Kds.Ai.AnomalyDetector/Orchestration/MonitorRunOrchestrator.cs`

csharp

using System.Diagnostics;
using Kds.Ai.AnomalyDetector.Agents;
using Kds.Ai.AnomalyDetector.Alerts;
using Kds.Ai.Data.Entities;
using Kds.Ai.Data.Enums;
using Kds.Ai.Data.Repositories;
using Microsoft.Extensions.Logging;

namespace Kds.Ai.AnomalyDetector.Orchestration;

internal sealed class MonitorRunOrchestrator(
    IMonitorSpecRepository specRepository,
    IMonitorRunRepository runRepository,
    IAnomalyDetectorAgent agent,
    IAlertEmitter alertEmitter,
    ILogger<MonitorRunOrchestrator> logger) : IMonitorRunOrchestrator
{
    public async Task<MonitorRun> ExecuteAsync(
        Guid monitorSpecId,
        DateTimeOffset? analysisTime = null,
        RunOptions? options = null,
        CancellationToken ct = default)
    {
        options ??= RunOptions.Default;
        ReportProgress(options, RunPhase.LoadingSpec, "Loading monitor specification...");
        
        var spec = await specRepository.GetByIdAsync(monitorSpecId, ct);
        if (spec is null)
        {
            throw new InvalidOperationException($"MonitorSpec {monitorSpecId} not found");
        }
        
        return await ExecuteAsync(spec, analysisTime, options, ct);
    }
    
    public async Task<MonitorRun> ExecuteAsync(
        MonitorSpec spec,
        DateTimeOffset? analysisTime = null,
        RunOptions? options = null,
        CancellationToken ct = default)
    {
        options ??= RunOptions.Default;
        var time = analysisTime ?? DateTimeOffset.UtcNow;
        var stopwatch = Stopwatch.StartNew();
        
        logger.LogInformation(
            "Starting monitor run for '{MonitorName}' at {AnalysisTime}",
            spec.Name, time);
        
        // Create run record
        ReportProgress(options, RunPhase.Initializing, "Initializing monitor run...");
        
        var run = new MonitorRun
        {
            Id = Guid.NewGuid(),
            MonitorSpecId = spec.Id,
            Status = MonitorRunStatus.Running,
            StartedAt = DateTimeOffset.UtcNow,
            AnalysisWindowStart = options.WindowStart ?? time.AddHours(-24),
            AnalysisWindowEnd = options.WindowEnd ?? time
        };
        
        if (!options.DryRun)
        {
            run = await runRepository.CreateAsync(run, ct);
        }
        
        try
        {
            // Run detection (single generic agent, no factory)
            ReportProgress(options, RunPhase.RunningAgent, "Running anomaly detection agent...");
            
            using var timeoutCts = options.Timeout.HasValue
                ? new CancellationTokenSource(options.Timeout.Value)
                : new CancellationTokenSource();
            
            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ct, timeoutCts.Token);
            
            var result = await agent.RunDetectionAsync(spec, time, linkedCts.Token);
            
            // Update run with results
            ReportProgress(options, RunPhase.AnalyzingResults, "Processing detection results...");
            
            run.AnomaliesDetected = result.Findings.Count;
            run.ResultSummary = SerializeResultSummary(result);
            run.TokensUsed = result.TotalTokens;
            
            // Emit alerts if anomalies detected
            if (result.AnomalyDetected && !options.SuppressAlerts && !options.DryRun)
            {
                ReportProgress(options, RunPhase.EmittingAlerts, "Emitting alerts...");
                
                var alert = await alertEmitter.EmitAsync(spec, result, ct);
                if (alert.Id != Guid.Empty)
                {
                    run.AlertEvents.Add(alert);
                    
                    logger.LogInformation(
                        "Emitted {Severity} alert {AlertId} for monitor '{MonitorName}'",
                        result.Severity, alert.Id, spec.Name);
                }
            }
            
            // Mark completed
            run.Status = MonitorRunStatus.Completed;
            run.CompletedAt = DateTimeOffset.UtcNow;
            run.DurationMs = stopwatch.ElapsedMilliseconds;
            
            // Save results
            if (!options.DryRun)
            {
                ReportProgress(options, RunPhase.SavingResults, "Saving results...");
                await runRepository.UpdateAsync(run, ct);
                await specRepository.UpdateLastRunAsync(spec.Id, run.CompletedAt.Value, ct);
            }
            
            ReportProgress(options, RunPhase.Completed, 
                result.AnomalyDetected 
                    ? $"Completed: {result.Findings.Count} anomalies detected"
                    : "Completed: No anomalies detected");
            
            logger.LogInformation(
                "Monitor run {RunId} completed in {Duration}ms. Anomalies: {Count}",
                run.Id, run.DurationMs, run.AnomaliesDetected);
            
            return run;
        }
        catch (OperationCanceledException) when (options.Timeout.HasValue)
        {
            return await HandleFailure(run, "Execution timed out", options, ct);
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Monitor run failed for {MonitorSpecId}", spec.Id);
            return await HandleFailure(run, ex.Message, options, ct);
        }
    }
    
    public async Task<IReadOnlyList<MonitorRun>> ExecuteAllActiveAsync(
        DateTimeOffset? analysisTime = null,
        RunOptions? options = null,
        CancellationToken ct = default)
    {
        var specs = await specRepository.GetActiveAsync(ct);
        
        logger.LogInformation(
            "Executing {Count} active monitors",
            specs.Count);
        
        var results = new List<MonitorRun>();
        
        foreach (var spec in specs)
        {
            try
            {
                var run = await ExecuteAsync(spec, analysisTime, options, ct);
                results.Add(run);
            }
            catch (Exception ex)
            {
                logger.LogError(ex, 
                    "Failed to execute monitor '{MonitorName}' ({MonitorId})",
                    spec.Name, spec.Id);
                
                // Continue with other monitors
            }
        }
        
        return results;
    }
    
    private async Task<MonitorRun> HandleFailure(
        MonitorRun run, 
        string error, 
        RunOptions options,
        CancellationToken ct)
    {
        run.Status = MonitorRunStatus.Failed;
        run.CompletedAt = DateTimeOffset.UtcNow;
        run.ErrorMessage = error;
        
        if (!options.DryRun)
        {
            await runRepository.UpdateAsync(run, ct);
        }
        
        ReportProgress(options, RunPhase.Failed, $"Failed: {error}");
        
        return run;
    }
    
    private static void ReportProgress(RunOptions options, RunPhase phase, string message)
    {
        options.Progress?.Report(new RunProgress
        {
            Phase = phase,
            Message = message
        });
    }
    
    private static string SerializeResultSummary(AnomalyDetectionResult result)
    {
        return System.Text.Json.JsonSerializer.Serialize(new
        {
            anomalyDetected = result.AnomalyDetected,
            severity = result.Severity?.ToString(),
            summary = result.Summary,
            findingsCount = result.Findings.Count,
            findings = result.Findings.Take(10).Select(f => new
            {
                description = f.Description,
                severity = f.Severity.ToString(),
                period = f.AffectedPeriod,
                evidence = f.Evidence
            }),
            warnings = result.Warnings
        });
    }
}

Files to Modify

2. `backend/Kds.Ai.AnomalyDetector/DependencyInjectionExtensions.cs`

Add orchestrator registration:

csharp

// Register orchestrator
builder.Services.AddScoped<IMonitorRunOrchestrator, MonitorRunOrchestrator>();

Execution Flow

ExecuteAsync(specId)
    │
    ├── Load MonitorSpec from repository
    │
    ├── Create MonitorRun record (status: Running)
    │
    ├── Run agent.RunDetectionAsync(spec, time)
    │   │
    │   ├── Agent reads spec.SystemPrompt as instructions
    │   ├── Agent queries data via MCP tools (mysql_query)
    │   ├── Agent analyzes and reasons
    │   └── Agent calls emit_conclusion → AnomalyDetectionResult
    │
    ├── If anomaly detected:
    │   └── alertEmitter.EmitAsync() → AlertEvent
    │       └── run.AlertEvents.Add(alert)
    │
    ├── Update MonitorRun (status: Completed)
    │
    └── Update MonitorSpec.LastRunAt

Changes from Old Design

Uses single IAnomalyDetectorAgent directly (no IAnomalyDetectorAgentFactory, no .GetAgent(monitorType))
Class is internal sealed with primary constructor (matching codebase pattern for implementations)
Logs spec.Name instead of spec.MonitorType and spec.StoreGuid
Uses run.AlertEvents.Add(alert) instead of run.AlertEventId = alert.Id (collection, not single FK)
Uses result.TotalTokens instead of result.TokenUsage?.TotalTokens
SerializeResultSummary uses AnomalyFinding fields: Description, Severity, AffectedPeriod, Evidence
ExecuteAllActiveAsync replaces both ExecuteAllByTypeAsync and ExecuteForStoreAsync
Class is internal sealed with primary constructor

Acceptance Criteria

[ ] MonitorRunOrchestrator class created (internal sealed with primary constructor)
[ ] All interface methods implemented (no ExecuteAllByTypeAsync or ExecuteForStoreAsync)
[ ] Uses single IAnomalyDetectorAgent (no factory)
[ ] Uses IMonitorSpecRepository from Kds.Ai.Data.Repositories (not Kds.Ai.AnomalyDetector.Repositories)
[ ] Alert stored via run.AlertEvents collection (many-to-many)
[ ] Progress reporting integrated
[ ] Timeout handling with CancellationTokenSource
[ ] Error handling with proper run status updates
[ ] Dry-run mode skips persistence
[ ] Alert suppression option works
[ ] Batch execution continues on individual failures
[ ] Duration tracking accurate
[ ] Token usage recorded via result.TotalTokens
[ ] No references to MonitorType, StoreGuid, IsEnabled, or AlertEventId
[ ] Orchestrator registered in DI
[ ] Solution builds without errors

Notes

The orchestrator doesn't parallelize batch runs by default (could add option later)
Consider adding circuit breaker for repeated failures
May want to add retry logic for transient errors
Result summary is truncated (10 findings) to avoid huge DB entries