Workflow Visualization & Debugging

Overview

Workflow Visualization & Debugging transforms the existing DAG-based workflow engine into a fully observable, debuggable system. This enhancement provides real-time visualization of workflow execution, step-level log streaming, time-travel debugging, and “what-if” simulation for workflow testing.

This is a best-in-class implementation inspired by modern workflow tools (Temporal, Argo, Dagster) but tailored for release orchestration with security-gated deployments.

Design Principles

Real-Time Observability: Every state transition visible within milliseconds
Time-Travel Debugging: Replay any past workflow execution step-by-step
Deterministic Simulation: Test workflows without side effects
Minimal Performance Impact: Visualization doesn’t slow execution
Offline-Compatible: Core visualization works without external services
Security-Aware: Sensitive data masked in logs and visualizations

Architecture

Component Overview

┌────────────────────────────────────────────────────────────────────────┐
│                  Workflow Visualization System                         │
├────────────────────────────────────────────────────────────────────────┤
│                                                                        │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ WorkflowEngine   │───▶│ EventBroadcaster  │───▶│ WebSocket Hub   │ │
│  │ (existing)       │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│           │                       │                        │          │
│           ▼                       ▼                        ▼          │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ ExecutionRecorder│    │ LogAggregator     │    │ React DAG UI    │ │
│  │                  │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│           │                       │                        │          │
│           ▼                       ▼                        ▼          │
│  ┌──────────────────┐    ┌───────────────────┐    ┌─────────────────┐ │
│  │ TimeTravel       │    │ SimulationEngine  │    │ DebugInspector  │ │
│  │ Debugger         │    │                   │    │                 │ │
│  └──────────────────┘    └───────────────────┘    └─────────────────┘ │
│                                                                        │
└────────────────────────────────────────────────────────────────────────┘

Key Components

1. EventBroadcaster

Captures and broadcasts all workflow events in real-time:

public sealed class EventBroadcaster : IWorkflowEventSink
{
    private readonly IHubContext<WorkflowHub> _hubContext;
    private readonly IExecutionRecorder _recorder;
    private readonly Channel<WorkflowEvent> _eventChannel;

    public async Task BroadcastAsync(WorkflowEvent @event, CancellationToken ct)
    {
        // Record for time-travel debugging
        await _recorder.RecordAsync(@event, ct);

        // Broadcast to connected clients
        var group = _hubContext.Clients.Group($"workflow:{@event.RunId}");
        await group.SendAsync("WorkflowEvent", @event, ct);

        // Also publish to general channel for dashboard
        await _hubContext.Clients.Group("workflows:all")
            .SendAsync("WorkflowEvent", @event.ToSummary(), ct);
    }
}

public abstract record WorkflowEvent
{
    public Guid RunId { get; init; }
    public Guid EventId { get; init; }
    public long SequenceNumber { get; init; }
    public DateTimeOffset Timestamp { get; init; }
    public string EventType { get; init; }
}

public sealed record StepStateChangedEvent : WorkflowEvent
{
    public Guid StepId { get; init; }
    public string StepName { get; init; }
    public StepStatus PreviousStatus { get; init; }
    public StepStatus NewStatus { get; init; }
    public TimeSpan? Duration { get; init; }
    public ImmutableDictionary<string, object>? Outputs { get; init; }
    public string? Error { get; init; }
}

public sealed record StepLogEvent : WorkflowEvent
{
    public Guid StepId { get; init; }
    public LogLevel Level { get; init; }
    public string Message { get; init; }
    public ImmutableDictionary<string, string>? Properties { get; init; }
}

2. ExecutionRecorder

Records full execution history for time-travel debugging:

public sealed class ExecutionRecorder : IExecutionRecorder
{
    private readonly IExecutionSnapshotStore _snapshotStore;

    public async Task RecordAsync(WorkflowEvent @event, CancellationToken ct)
    {
        var snapshot = new ExecutionSnapshot
        {
            RunId = @event.RunId,
            SequenceNumber = @event.SequenceNumber,
            Timestamp = @event.Timestamp,
            Event = @event,
            WorkflowState = await CaptureCurrentStateAsync(@event.RunId, ct)
        };

        await _snapshotStore.SaveAsync(snapshot, ct);
    }

    private async Task<WorkflowStateSnapshot> CaptureCurrentStateAsync(
        Guid runId, CancellationToken ct)
    {
        var run = await _workflowRunStore.GetAsync(runId, ct);
        return new WorkflowStateSnapshot
        {
            Status = run.Status,
            Steps = run.Steps.Select(s => new StepSnapshot
            {
                Id = s.Id,
                Name = s.Name,
                Status = s.Status,
                Inputs = s.Inputs,
                Outputs = s.Outputs,
                StartedAt = s.StartedAt,
                CompletedAt = s.CompletedAt,
                RetryCount = s.RetryCount
            }).ToImmutableArray(),
            Variables = run.Variables
        };
    }
}

public sealed record ExecutionSnapshot
{
    public Guid RunId { get; init; }
    public long SequenceNumber { get; init; }
    public DateTimeOffset Timestamp { get; init; }
    public WorkflowEvent Event { get; init; }
    public WorkflowStateSnapshot WorkflowState { get; init; }
}

3. TimeTravelDebugger

Enables step-by-step replay of past executions:

public sealed class TimeTravelDebugger
{
    private readonly IExecutionSnapshotStore _snapshotStore;

    public async Task<TimeTravelSession> CreateSessionAsync(
        Guid runId, CancellationToken ct)
    {
        var snapshots = await _snapshotStore.GetSnapshotsAsync(runId, ct);
        return new TimeTravelSession
        {
            RunId = runId,
            TotalSnapshots = snapshots.Count,
            Snapshots = snapshots,
            CurrentPosition = 0
        };
    }

    public ExecutionSnapshot StepForward(TimeTravelSession session)
    {
        if (session.CurrentPosition >= session.TotalSnapshots - 1)
            throw new EndOfExecutionException();

        session.CurrentPosition++;
        return session.Snapshots[session.CurrentPosition];
    }

    public ExecutionSnapshot StepBackward(TimeTravelSession session)
    {
        if (session.CurrentPosition <= 0)
            throw new BeginningOfExecutionException();

        session.CurrentPosition--;
        return session.Snapshots[session.CurrentPosition];
    }

    public ExecutionSnapshot JumpToSnapshot(TimeTravelSession session, long sequenceNumber)
    {
        var snapshot = session.Snapshots
            .FirstOrDefault(s => s.SequenceNumber == sequenceNumber)
            ?? throw new SnapshotNotFoundException(sequenceNumber);

        session.CurrentPosition = session.Snapshots.IndexOf(snapshot);
        return snapshot;
    }

    public ExecutionSnapshot JumpToStep(TimeTravelSession session, Guid stepId, StepStatus status)
    {
        var snapshot = session.Snapshots
            .FirstOrDefault(s =>
                s.Event is StepStateChangedEvent e &&
                e.StepId == stepId &&
                e.NewStatus == status)
            ?? throw new StepSnapshotNotFoundException(stepId, status);

        session.CurrentPosition = session.Snapshots.IndexOf(snapshot);
        return snapshot;
    }
}

4. SimulationEngine

Executes workflows in simulation mode without side effects:

public sealed class SimulationEngine
{
    private readonly IWorkflowTemplateStore _templateStore;
    private readonly IDagScheduler _dagScheduler;

    public async Task<SimulationResult> SimulateAsync(
        SimulationRequest request,
        CancellationToken ct)
    {
        var template = await _templateStore.GetAsync(request.TemplateId, ct);

        // Create simulation context with mocked dependencies
        var context = new SimulationContext
        {
            Template = template,
            Variables = request.Variables,
            MockedGateResults = request.MockedGateResults,
            MockedStepDurations = request.MockedStepDurations,
            FailureScenarios = request.FailureScenarios
        };

        var result = new SimulationResult
        {
            SimulationId = Guid.NewGuid(),
            TemplateId = template.Id,
            StartedAt = _timeProvider.GetUtcNow()
        };

        // Execute simulation
        var steps = template.Steps.ToList();
        var completed = new HashSet<Guid>();
        var stepResults = new List<SimulatedStepResult>();

        while (completed.Count < steps.Count)
        {
            var ready = _dagScheduler.GetReadyNodes(steps, completed);
            if (!ready.Any())
            {
                result.DeadlockDetected = true;
                break;
            }

            foreach (var step in ready)
            {
                var stepResult = await SimulateStepAsync(step, context, ct);
                stepResults.Add(stepResult);

                if (stepResult.Status == StepStatus.Succeeded ||
                    stepResult.Status == StepStatus.Skipped)
                {
                    completed.Add(step.Id);
                }
                else if (stepResult.Status == StepStatus.Failed)
                {
                    // Handle failure based on step config
                    if (step.OnFailure == FailureAction.Fail)
                    {
                        result.FailedAtStep = step.Id;
                        break;
                    }
                }
            }
        }

        result.StepResults = stepResults.ToImmutableArray();
        result.CompletedAt = _timeProvider.GetUtcNow();
        result.Status = DetermineOutcome(result);
        result.CriticalPath = CalculateCriticalPath(stepResults);
        result.EstimatedDuration = CalculateEstimatedDuration(stepResults);

        return result;
    }

    private async Task<SimulatedStepResult> SimulateStepAsync(
        WorkflowStep step,
        SimulationContext context,
        CancellationToken ct)
    {
        var result = new SimulatedStepResult
        {
            StepId = step.Id,
            StepName = step.Name,
            StepType = step.Type
        };

        // Check for injected failure scenarios
        if (context.FailureScenarios.TryGetValue(step.Id, out var failure))
        {
            result.Status = StepStatus.Failed;
            result.Error = failure.ErrorMessage;
            result.SimulatedDuration = failure.FailAfter;
            return result;
        }

        // Check for mocked gate results
        if (step.Type == StepType.Gate &&
            context.MockedGateResults.TryGetValue(step.Id, out var gateResult))
        {
            result.Status = gateResult ? StepStatus.Succeeded : StepStatus.Failed;
            result.Outputs = new Dictionary<string, object>
            {
                ["gate_passed"] = gateResult
            }.ToImmutableDictionary();
        }
        else
        {
            // Default: step succeeds
            result.Status = StepStatus.Succeeded;
        }

        // Apply simulated duration
        result.SimulatedDuration = context.MockedStepDurations
            .GetValueOrDefault(step.Id, TimeSpan.FromSeconds(1));

        return result;
    }
}

public sealed record SimulationRequest
{
    public Guid TemplateId { get; init; }
    public ImmutableDictionary<string, object> Variables { get; init; }
    public ImmutableDictionary<Guid, bool> MockedGateResults { get; init; }
    public ImmutableDictionary<Guid, TimeSpan> MockedStepDurations { get; init; }
    public ImmutableDictionary<Guid, FailureScenario> FailureScenarios { get; init; }
}

public sealed record SimulationResult
{
    public Guid SimulationId { get; init; }
    public Guid TemplateId { get; init; }
    public SimulationStatus Status { get; init; }
    public ImmutableArray<SimulatedStepResult> StepResults { get; init; }
    public ImmutableArray<Guid> CriticalPath { get; init; }
    public TimeSpan EstimatedDuration { get; init; }
    public bool DeadlockDetected { get; init; }
    public Guid? FailedAtStep { get; init; }
    public DateTimeOffset StartedAt { get; init; }
    public DateTimeOffset CompletedAt { get; init; }
}

5. LogAggregator

Aggregates and streams step logs in real-time:

public sealed class LogAggregator
{
    private readonly ILogStore _logStore;
    private readonly IHubContext<WorkflowHub> _hubContext;
    private readonly ConcurrentDictionary<Guid, LogBuffer> _buffers = new();

    public async Task AppendLogAsync(Guid runId, Guid stepId, LogEntry entry, CancellationToken ct)
    {
        // Mask sensitive data
        var maskedEntry = _sensitiveDataMasker.Mask(entry);

        // Store for retrieval
        await _logStore.AppendAsync(runId, stepId, maskedEntry, ct);

        // Buffer for batched broadcast
        var buffer = _buffers.GetOrAdd(runId, _ => new LogBuffer());
        buffer.Add(stepId, maskedEntry);

        // Broadcast immediately for active viewers
        await _hubContext.Clients
            .Group($"workflow:{runId}:logs")
            .SendAsync("StepLog", new { stepId, entry = maskedEntry }, ct);
    }

    public async IAsyncEnumerable<LogEntry> StreamLogsAsync(
        Guid runId,
        Guid? stepId,
        [EnumeratorCancellation] CancellationToken ct)
    {
        // First, return historical logs
        await foreach (var entry in _logStore.GetLogsAsync(runId, stepId, ct))
        {
            yield return entry;
        }

        // Then stream live logs
        var channel = Channel.CreateUnbounded<LogEntry>();
        var subscription = SubscribeToLiveLogs(runId, stepId, channel.Writer);

        try
        {
            await foreach (var entry in channel.Reader.ReadAllAsync(ct))
            {
                yield return entry;
            }
        }
        finally
        {
            subscription.Dispose();
        }
    }
}

public sealed record LogEntry
{
    public Guid StepId { get; init; }
    public DateTimeOffset Timestamp { get; init; }
    public LogLevel Level { get; init; }
    public string Message { get; init; }
    public string? Source { get; init; }
    public ImmutableDictionary<string, string>? Properties { get; init; }
}

DAG Visualization Model

Graph Data Structure

interface WorkflowGraph {
  id: string;
  name: string;
  status: WorkflowStatus;
  nodes: WorkflowNode[];
  edges: WorkflowEdge[];
  metadata: WorkflowMetadata;
}

interface WorkflowNode {
  id: string;
  name: string;
  type: StepType;
  status: StepStatus;
  position: { x: number; y: number };  // Auto-calculated or manual

  // Timing
  startedAt?: string;
  completedAt?: string;
  duration?: number;
  estimatedDuration?: number;

  // State
  inputs?: Record<string, unknown>;
  outputs?: Record<string, unknown>;
  error?: string;
  retryCount?: number;

  // Visual
  icon?: string;
  color?: string;
  highlight?: boolean;
}

interface WorkflowEdge {
  id: string;
  source: string;
  target: string;
  type: 'dependency' | 'data-flow' | 'conditional';
  label?: string;
  animated?: boolean;  // For in-progress transitions
}

interface WorkflowMetadata {
  totalSteps: number;
  completedSteps: number;
  failedSteps: number;
  duration?: number;
  estimatedRemaining?: number;
  criticalPath: string[];
}

Layout Algorithm

// Dagre-based automatic layout with customizations
function calculateLayout(graph: WorkflowGraph): LayoutResult {
  const g = new dagre.graphlib.Graph();
  g.setGraph({
    rankdir: 'LR',           // Left to right
    nodesep: 50,             // Horizontal spacing
    ranksep: 100,            // Vertical spacing between ranks
    marginx: 20,
    marginy: 20
  });

  // Add nodes
  graph.nodes.forEach(node => {
    g.setNode(node.id, {
      width: getNodeWidth(node),
      height: getNodeHeight(node)
    });
  });

  // Add edges
  graph.edges.forEach(edge => {
    g.setEdge(edge.source, edge.target);
  });

  // Calculate layout
  dagre.layout(g);

  // Extract positions
  return {
    nodes: graph.nodes.map(node => ({
      ...node,
      position: {
        x: g.node(node.id).x,
        y: g.node(node.id).y
      }
    })),
    edges: graph.edges.map(edge => {
      const points = g.edge(edge.source, edge.target).points;
      return { ...edge, points };
    })
  };
}

Real-Time Updates

WebSocket Protocol

// Client subscription
socket.emit('subscribe', {
  type: 'workflow',
  runId: 'abc-123',
  channels: ['state', 'logs', 'metrics']
});

// Server events
interface WorkflowStateUpdate {
  type: 'state';
  runId: string;
  event: WorkflowEvent;
  graph: Partial<WorkflowGraph>;  // Delta update
}

interface WorkflowLogUpdate {
  type: 'log';
  runId: string;
  stepId: string;
  entry: LogEntry;
}

interface WorkflowMetricsUpdate {
  type: 'metrics';
  runId: string;
  metrics: {
    cpuUsage?: number;
    memoryUsage?: number;
    networkIO?: number;
    activeConnections?: number;
  };
}

Delta Updates

public sealed class DeltaCalculator
{
    public GraphDelta CalculateDelta(WorkflowGraph previous, WorkflowGraph current)
    {
        var delta = new GraphDelta();

        // Node changes
        foreach (var node in current.Nodes)
        {
            var prev = previous.Nodes.FirstOrDefault(n => n.Id == node.Id);
            if (prev == null)
            {
                delta.AddedNodes.Add(node);
            }
            else if (!NodeEquals(prev, node))
            {
                delta.UpdatedNodes.Add(node);
            }
        }

        // Edge changes
        foreach (var edge in current.Edges)
        {
            var prev = previous.Edges.FirstOrDefault(e => e.Id == edge.Id);
            if (prev == null)
            {
                delta.AddedEdges.Add(edge);
            }
            else if (!EdgeEquals(prev, edge))
            {
                delta.UpdatedEdges.Add(edge);
            }
        }

        // Metadata changes
        if (!MetadataEquals(previous.Metadata, current.Metadata))
        {
            delta.MetadataUpdate = current.Metadata;
        }

        return delta;
    }
}

Debug Inspector

Step Inspection

public sealed class DebugInspector
{
    public async Task<StepInspection> InspectStepAsync(
        Guid runId, Guid stepId, CancellationToken ct)
    {
        var run = await _workflowRunStore.GetAsync(runId, ct);
        var step = run.Steps.First(s => s.Id == stepId);
        var logs = await _logStore.GetLogsAsync(runId, stepId, ct);

        return new StepInspection
        {
            Step = step,

            // Input/Output analysis
            Inputs = step.Inputs,
            ResolvedInputs = await ResolveInputSourcesAsync(step.Inputs, run, ct),
            Outputs = step.Outputs,
            OutputConsumers = FindOutputConsumers(stepId, run),

            // Timing analysis
            QueuedAt = step.CreatedAt,
            StartedAt = step.StartedAt,
            CompletedAt = step.CompletedAt,
            QueueTime = step.StartedAt - step.CreatedAt,
            ExecutionTime = step.CompletedAt - step.StartedAt,

            // Dependencies
            WaitedFor = GetWaitedForSteps(stepId, run),
            BlockedBy = GetBlockingSteps(stepId, run),

            // Retry history
            RetryAttempts = step.RetryHistory,

            // Logs summary
            LogSummary = new LogSummary
            {
                TotalLines = logs.Count,
                ErrorCount = logs.Count(l => l.Level == LogLevel.Error),
                WarningCount = logs.Count(l => l.Level == LogLevel.Warning),
                LastError = logs.LastOrDefault(l => l.Level == LogLevel.Error),
                FirstTimestamp = logs.FirstOrDefault()?.Timestamp,
                LastTimestamp = logs.LastOrDefault()?.Timestamp
            },

            // Environment
            ExecutionEnvironment = new ExecutionEnvironment
            {
                AgentId = step.AgentId,
                TargetId = step.TargetId,
                Variables = MaskSensitiveVariables(run.Variables)
            }
        };
    }
}

Diff View

interface StepDiff {
  stepId: string;
  changes: {
    field: string;
    previous: unknown;
    current: unknown;
    changeType: 'added' | 'removed' | 'modified';
  }[];
}

function calculateStepDiff(
  previousSnapshot: ExecutionSnapshot,
  currentSnapshot: ExecutionSnapshot,
  stepId: string
): StepDiff {
  const prev = previousSnapshot.workflowState.steps.find(s => s.id === stepId);
  const curr = currentSnapshot.workflowState.steps.find(s => s.id === stepId);

  const changes: StepDiff['changes'] = [];

  // Compare status
  if (prev?.status !== curr?.status) {
    changes.push({
      field: 'status',
      previous: prev?.status,
      current: curr?.status,
      changeType: 'modified'
    });
  }

  // Compare outputs (deep diff)
  const outputDiff = deepDiff(prev?.outputs, curr?.outputs);
  outputDiff.forEach(d => changes.push({ ...d, field: `outputs.${d.path}` }));

  return { stepId, changes };
}

API Design

REST Endpoints

# Workflow Visualization
GET    /api/v1/workflows/{runId}/graph              # Get full graph state
GET    /api/v1/workflows/{runId}/graph/layout       # Get calculated layout
GET    /api/v1/workflows/{runId}/steps/{stepId}     # Get step details
GET    /api/v1/workflows/{runId}/steps/{stepId}/logs # Get step logs
GET    /api/v1/workflows/{runId}/critical-path      # Get critical path analysis

# Time-Travel Debugging
POST   /api/v1/workflows/{runId}/debug/session      # Create debug session
GET    /api/v1/workflows/{runId}/debug/snapshots    # List all snapshots
GET    /api/v1/workflows/{runId}/debug/snapshots/{seq} # Get specific snapshot
POST   /api/v1/workflows/{runId}/debug/step-forward # Step forward
POST   /api/v1/workflows/{runId}/debug/step-backward # Step backward
POST   /api/v1/workflows/{runId}/debug/jump         # Jump to snapshot

# Simulation
POST   /api/v1/workflows/templates/{templateId}/simulate # Run simulation
GET    /api/v1/workflows/simulations/{simId}        # Get simulation result
POST   /api/v1/workflows/templates/{templateId}/validate # Validate template

# Comparison
GET    /api/v1/workflows/compare?runIds=a,b,c       # Compare multiple runs

WebSocket Endpoints

/ws/workflows/{runId}                # Subscribe to workflow updates
/ws/workflows/{runId}/logs           # Subscribe to log stream
/ws/workflows/all                    # Subscribe to all workflow summaries

UI Components

React Component Architecture

// Main visualization component
interface WorkflowVisualizerProps {
  runId: string;
  mode: 'live' | 'replay' | 'simulation';
  onStepSelect?: (stepId: string) => void;
}

// DAG renderer
interface DAGRendererProps {
  graph: WorkflowGraph;
  layout: LayoutResult;
  selectedStep?: string;
  highlightPath?: string[];
  onNodeClick: (nodeId: string) => void;
  onNodeHover: (nodeId: string | null) => void;
}

// Step detail panel
interface StepDetailPanelProps {
  inspection: StepInspection;
  logs: LogEntry[];
  onLogSearch: (query: string) => void;
  onRetry?: () => void;
}

// Time-travel controls
interface TimeTravelControlsProps {
  session: TimeTravelSession;
  onStepForward: () => void;
  onStepBackward: () => void;
  onJumpTo: (seq: number) => void;
  onPlay: () => void;
  onPause: () => void;
  playbackSpeed: number;
  onSpeedChange: (speed: number) => void;
}

// Log viewer
interface LogViewerProps {
  logs: LogEntry[];
  streaming: boolean;
  filter?: LogFilter;
  onFilterChange: (filter: LogFilter) => void;
  highlightPattern?: string;
}

Visual States

// Node status colors
.node {
  &--pending { background: #6b7280; }      // Gray
  &--running {
    background: #3b82f6;                    // Blue
    animation: pulse 2s infinite;
  }
  &--succeeded { background: #22c55e; }    // Green
  &--failed { background: #ef4444; }       // Red
  &--skipped { background: #a855f7; }      // Purple
  &--retrying {
    background: #f59e0b;                    // Amber
    animation: pulse 1s infinite;
  }
}

// Edge animations
.edge {
  &--active {
    stroke-dasharray: 5;
    animation: dash 1s linear infinite;
  }
  &--data-flow {
    stroke: #60a5fa;
    stroke-dasharray: 3 3;
  }
}

// Critical path highlight
.critical-path {
  .node { box-shadow: 0 0 10px #f59e0b; }
  .edge { stroke: #f59e0b; stroke-width: 3; }
}

Performance Optimizations

Snapshot Compression

public sealed class SnapshotCompressor
{
    // Store deltas instead of full snapshots after initial
    public CompressedSnapshot Compress(
        ExecutionSnapshot current,
        ExecutionSnapshot? previous)
    {
        if (previous == null)
        {
            return new CompressedSnapshot
            {
                Type = SnapshotType.Full,
                Data = Serialize(current)
            };
        }

        var delta = CalculateDelta(previous, current);
        var deltaSize = Serialize(delta).Length;
        var fullSize = Serialize(current).Length;

        // Use delta if significantly smaller
        if (deltaSize < fullSize * 0.7)
        {
            return new CompressedSnapshot
            {
                Type = SnapshotType.Delta,
                BaseSequence = previous.SequenceNumber,
                Data = Serialize(delta)
            };
        }

        return new CompressedSnapshot
        {
            Type = SnapshotType.Full,
            Data = Serialize(current)
        };
    }
}

Log Pagination

public sealed class LogPaginator
{
    private const int DefaultPageSize = 100;
    private const int MaxPageSize = 1000;

    public async Task<PagedLogs> GetPagedLogsAsync(
        Guid runId,
        Guid stepId,
        int page,
        int pageSize,
        LogFilter? filter,
        CancellationToken ct)
    {
        pageSize = Math.Min(pageSize, MaxPageSize);

        var query = _logStore.Query(runId, stepId);

        if (filter?.Level != null)
            query = query.Where(l => l.Level >= filter.Level);

        if (!string.IsNullOrEmpty(filter?.Search))
            query = query.Where(l => l.Message.Contains(filter.Search));

        var total = await query.CountAsync(ct);
        var logs = await query
            .Skip((page - 1) * pageSize)
            .Take(pageSize)
            .ToListAsync(ct);

        return new PagedLogs
        {
            Logs = logs.ToImmutableArray(),
            Page = page,
            PageSize = pageSize,
            TotalCount = total,
            TotalPages = (int)Math.Ceiling(total / (double)pageSize)
        };
    }
}

Metrics & Observability

Prometheus Metrics

# Visualization
stella_workflow_visualization_connections{run_id}
stella_workflow_visualization_events_broadcast_total{event_type}
stella_workflow_visualization_snapshot_size_bytes{compression_type}

# Time-travel
stella_workflow_debug_sessions_active
stella_workflow_debug_snapshots_total{run_id}
stella_workflow_debug_replay_operations_total{operation}

# Simulation
stella_workflow_simulations_total{template_id, status}
stella_workflow_simulation_duration_seconds{template_id}
stella_workflow_simulation_steps_evaluated_total{template_id}

# Log streaming
stella_workflow_logs_streamed_total{run_id, step_id}
stella_workflow_log_buffer_size{run_id}

Security Considerations

Sensitive Data Masking

public sealed class SensitiveDataMasker
{
    private readonly ImmutableArray<Regex> _patterns;

    public SensitiveDataMasker()
    {
        _patterns = new[]
        {
            new Regex(@"password[""']?\s*[:=]\s*[""']?([^""'\s]+)", RegexOptions.IgnoreCase),
            new Regex(@"secret[""']?\s*[:=]\s*[""']?([^""'\s]+)", RegexOptions.IgnoreCase),
            new Regex(@"token[""']?\s*[:=]\s*[""']?([^""'\s]+)", RegexOptions.IgnoreCase),
            new Regex(@"api[_-]?key[""']?\s*[:=]\s*[""']?([^""'\s]+)", RegexOptions.IgnoreCase),
            new Regex(@"bearer\s+([a-zA-Z0-9_\-\.]+)", RegexOptions.IgnoreCase),
        }.ToImmutableArray();
    }

    public string Mask(string input)
    {
        var result = input;
        foreach (var pattern in _patterns)
        {
            result = pattern.Replace(result, m =>
            {
                var prefix = m.Value[..^m.Groups[1].Length];
                return prefix + "***MASKED***";
            });
        }
        return result;
    }
}

Access Control

// Only allow debug access to users with appropriate permissions
[Authorize(Policy = "WorkflowDebug")]
public class WorkflowDebugController : ControllerBase
{
    // Debug operations require elevated permissions
}

// Log access restricted by environment sensitivity
public async Task<bool> CanAccessLogsAsync(
    ClaimsPrincipal user, Guid runId, CancellationToken ct)
{
    var run = await _workflowRunStore.GetAsync(runId, ct);
    var environment = await _environmentStore.GetAsync(run.EnvironmentId, ct);

    if (environment.Sensitivity == EnvironmentSensitivity.Production)
    {
        return user.HasClaim("workflow:logs:production", "read");
    }

    return user.HasClaim("workflow:logs", "read");
}