Skip to main content

Recovered Analysis Workflow

The recoveredAnalysisWorkflow transforms raw wellbeing check-in data into actionable insights through a three-phase pipeline. This guide explains the technical architecture, data flow, and key design decisions for developers.

Quick Reference​

// Workflow endpoint
http://localhost:4111/workflows/recoveredAnalysisWorkflow/graph

// Programmatic invocation
const result = await mastra.runWorkflow('recoveredAnalysisWorkflow', {
  jsonData: '[{"Created At":"2025-12-13", "Label":"stress-checkin", ...}]',
  userPrompt: 'How have my stress levels been trending?',
  enableStepOrchestration: false,
});

Architecture Overview​

Transform raw JSON exports into canonical reasoning engine format

Execute 6-step analysis pipeline via sub-workflow call

Format results for dashboard and generate conversational response

Phase 1: Normalize (JSON Processing Layer)​

Responsibility​

Convert any input format into the canonical schema that the reasoning engine expects.

Input Schema​

{
  jsonData: string;              // Raw JSON string (any format)
  userPrompt: string;            // User's question
  enableStepOrchestration?: boolean; // Optional: enable smart step skipping
}

Output Schema​

{
  objective: string;             // User's question
  series: Series[];              // Time-series data per metric
  datasets: NumericDataset[];    // Aggregated daily values
  compactedRows: string[];       // Human-readable summaries
  enableStepOrchestration: boolean;
  _intakeMetadata: {             // Preserved for Phase 3
    detectedFormat: 'small' | 'large' | 'canonical';
    recordCount: number;
    dayCount: number;
    dateRange: { start: string; end: string };
    metrics: string[];
  };
}

Format Detection Logic​

Step 1: Check if Already Canonical
const isAlreadyCanonical = inputData.rawData && 
  typeof inputData.rawData === 'object' &&
  ('series' in inputData.rawData || 'datasets' in inputData.rawData);

if (isAlreadyCanonical) {
  // Skip transformation, validate schema, pass through
  return normalizedInputSchema.parse({
    ...raw,
    objective: inputData.prompt,
    _intakeMetadata: { detectedFormat: 'canonical', skippedNormalization: true },
  });
}

When this happens:

  • Data already in canonical format (pre-processed)
  • Contains series, datasets, or compactedRows keys
  • Common when using processed examples from examples/recovered/
Step 2: Detect Raw Export Format
if (!recoveredIntakeAdapter.canHandle(inputData.rawData)) {
  throw new Error('Input format not recognized');
}

const result = await recoveredIntakeAdapter.transform(inputData);

Adapter Detection Logic:

Large Format (from Recovered_Large.json):

{
  "created_at": "2025-12-13T08:48:11.387Z",
  "label": "stress-checkin",
  "value": "{\"options\":[\"very-low\"],\"notes\":\"\"}",
  "user_id": "6114",
  "extra": "",
  "age": 25,
  "gender": "female",
  "goals": "[\"reduce-stress\"]"
}

Small Format (from Recovered_Small.json):

{
  "Created At": "December 13, 2025, 8:48 AM",
  "Label": "stress-checkin",
  "Value": "{\"options\":[\"very-low\"],\"notes\":\"\"}",
  "User ID": "6114",
  "ID": "18410",
  "Kind": "stress-checkin",
  "Extra → Breakfast": "",
  "Extra → Lunch": ""
}
Step 3: Transform to Canonical

Transformation Process:

  1. Parse Value Field (JSON string with options, notes, date)
  2. Group by Metric Type (stress, emotion, energy, sleep, cycle)
  3. Build Series (one per metric with timestamps)
  4. Build Datasets (aggregated by day with numeric values)
  5. Build CompactedRows (human-readable daily summaries)
  6. Extract Metadata (record count, date range, detected metrics)

Example Output:

{
  objective: "Analyze my stress patterns",
  series: [
    {
      name: "Stress Check-ins",
      category: "stress-checkin",
      data: [
        { timestamp: "2025-12-12T08:36", value: "low", notes: "" },
        { timestamp: "2025-12-13T08:48", value: "very-low", notes: "" }
      ]
    }
  ],
  datasets: [
    {
      label: "Stress Level",
      values: [2, 1], // Numeric scale: very-low=1, low=2, etc.
      dates: ["2025-12-12", "2025-12-13"]
    }
  ],
  compactedRows: [
    "2025-12-12: Stress (low), Sleep (yes, disturbed but decent hours)",
    "2025-12-13: Stress (very-low), Emotion (apathetic), Sleep (yes)"
  ],
  _intakeMetadata: {
    detectedFormat: "small",
    recordCount: 8,
    dayCount: 2,
    dateRange: { start: "2025-12-12", end: "2025-12-13" },
    metrics: ["stress", "emotion", "sleep", "energy", "cycle"]
  }
}

Key Design Decisions​

Format Agnostic

Accepts any valid JSON structure—detection logic determines the path.

Single Normalization

All format handling happens ONCE at the start—no format logic in later phases.

Metadata Preservation

Intake metadata flows through to Phase 3 for confidence scoring and context.

Fail Fast

Invalid formats throw immediately with clear error messages.

Phase 2: Reasoning Engine (Analysis Layer)​

Responsibility​

Execute multi-agent analysis pipeline on normalized data using the reasoning engine as a sub-workflow.

Input Schema​

Uses the canonical format from Phase 1 (without the _intakeMetadata wrapper).

Output Schema​

{
  taskId: string;                // Unique workflow execution ID
  objective: string;             // User's question
  outputProfile?: string;        // Optional output format profile
  
  // Step outputs from reasoning engine
  temporal: {
    summary: string;
    insights: string[];
    patterns: Array<{
      type: string;
      description: string;
      significance: number;
    }>;
  };
  
  pattern: {
    matches: Array<{
      theme: string;
      occurrences: number;
      examples: string[];
    }>;
  };
  
  correlation: {
    pairs: Array<{
      metrics: [string, string];
      strength: number;
      direction: 'positive' | 'negative';
      insight: string;
    }>;
  };
  
  domain: {
    documents: string[];
    context: string[];
  };
  
  explanation: {
    narrative: string;
    keyPoints: string[];
    recommendations: string[];
  };
  
  validation: {
    isValid: boolean;
    warnings?: string[];
    errors?: string[];
  };
  
  result: {
    summary: string;
    insights: string[];
    recommendations: string[];
    confidence: number;
  };
  
  executedSteps?: string[];      // If step orchestration enabled
  _intakeMetadata?: Record<string, any>; // Passed through for Phase 3
}

Sub-Workflow Pattern​

info

Why a Sub-Workflow?

The reasoning engine is invoked as a separate workflow rather than duplicating its code:

  • Single Source of Truth: Changes to reasoning engine automatically propagate
  • Reusability: Same engine used by K12, TAP, Recovered, etc.
  • Separation of Concerns: Normalization logic stays separate from analysis
  • Testing: Can test reasoning engine independently
// Execute reasoning engine as sub-workflow
const run = await reasoningEngineWorkflow.createRun();
const result = await run.start({ inputData: reasoningEngineInput });

if (result.status !== 'success') {
  throw new Error(`Reasoning engine workflow failed: ${errorMessage}`);
}

Reasoning Engine Pipeline (6 Steps)​

Purpose: Identify time-based patterns (daily trends, weekly cycles)

Process:

  • Analyze series data for temporal patterns
  • Detect rising/falling trends
  • Identify cyclical behavior (e.g., weekly stress spikes)

Output:

{
  "summary": "Stress levels decreased by 30% over the past week",
  "insights": [
    "Stress lowest on weekends",
    "Energy peaks on Wednesday and Thursday"
  ],
  "patterns": [
    {
      "type": "weekly-cycle",
      "description": "Stress increases Monday-Wednesday, decreases Thursday-Friday",
      "significance": 0.85
    }
  ]
}

Purpose: Find recurring themes across metrics

Process:

  • Group related check-ins by theme
  • Count occurrences of patterns
  • Extract representative examples

Output:

{
  "matches": [
    {
      "theme": "sleep-disruption",
      "occurrences": 5,
      "examples": [
        "2025-12-12: 'My sleep was disturbed but I slept for a decent amount of hours'",
        "2025-12-10: 'Woke up multiple times during the night'"
      ]
    }
  ]
}

Purpose: Discover relationships between metrics

Process:

  • Calculate correlation coefficients between metric pairs
  • Determine direction (positive/negative)
  • Generate insight descriptions

Output:

{
  "pairs": [
    {
      "metrics": ["stress", "sleep"],
      "strength": 0.72,
      "direction": "negative",
      "insight": "Higher stress correlates with poorer sleep quality"
    }
  ]
}

Purpose: Fetch relevant health knowledge from K12Safety corpus

Process:

  • Query Azure AI Search with user's objective + detected patterns
  • Retrieve top-k relevant documents
  • Extract contextual knowledge

Output:

{
  "documents": [
    "Stress Management Techniques",
    "Sleep Hygiene Guidelines"
  ],
  "context": [
    "Progressive muscle relaxation can reduce stress by 40%",
    "Consistent sleep schedule improves sleep quality"
  ]
}

Purpose: Synthesize findings into narrative explanation

Process:

  • Combine temporal, pattern, and correlation findings
  • Incorporate domain knowledge
  • Generate cohesive narrative addressing user's question

Output:

{
  "narrative": "Your stress levels have been decreasing over the past week, with the most significant drops on weekends. There's a strong connection between your stress and sleep quality...",
  "keyPoints": [
    "Stress decreased by 30% week-over-week",
    "Sleep quality improved on days with lower stress",
    "Energy levels highest mid-week"
  ],
  "recommendations": [
    "Maintain weekend relaxation practices during weekdays",
    "Establish consistent sleep schedule",
    "Practice stress-reduction techniques before bed"
  ]
}

Purpose: Review results for logical consistency and flag low-confidence findings

Process:

  • Check for contradictory insights
  • Verify correlation strengths are statistically meaningful
  • Flag insufficient data scenarios

Output:

{
  "isValid": true,
  "warnings": [
    "Limited data for cycle analysis (only 1 entry)",
    "Energy trend confidence low due to missing data points"
  ],
  "errors": []
}

Database-Backed State​

info

When REASONING_DB_STEP_DATA=true:

  • Each reasoning step persists output to REASONING.data_WorkflowStepOutputs
  • Final workflow state contains only {taskId, objective, validation}
  • Phase 2 enriches result by reading step outputs from SQL

Benefits:

  • Handles large payloads without memory issues
  • Enables workflow resume after failures
  • Supports audit trails and debugging

Enrichment Logic:

async function enrichFromDb(result) {
  const taskId = result.taskId;
  if (!taskId) return result;
  
  // Check if step outputs already present (DB mode disabled)
  if (result.temporal && result.pattern) {
    return result; // Already have full data
  }
  
  // Read step outputs from DB
  const [temporal, pattern, correlation, domain, explanation, validation] = 
    await Promise.all([
      stepDataService.get(taskId, 'temporal-analysis'),
      stepDataService.get(taskId, 'pattern-recognition'),
      // ... other steps
    ]);
  
  return {
    ...result,
    temporal,
    pattern,
    correlation,
    domain,
    explanation,
    validation,
  };
}

Phase 3: Business Logic (Dashboard Layer)​

Responsibility​

Transform reasoning engine output into dashboard-ready format and generate conversational response.

Input Schema​

Reasoning engine output from Phase 2 (with _intakeMetadata preserved).

Output Schema​

{
  // Dashboard sections
  summary: {
    headline: string;              // e.g., "Your stress is improving"
    subHeadline: string;           // e.g., "Down 30% from last week"
    confidenceLevel: 'high' | 'medium' | 'low';
    periodSummary: string;         // e.g., "Based on 8 check-ins over 2 days"
  };
  
  insights: Array<{
    category: string;              // e.g., "stress", "sleep", "energy"
    description: string;
    icon: string;                  // UI icon identifier
    severity: 'info' | 'warning' | 'positive';
  }>;
  
  recommendations: Array<{
    title: string;
    description: string;
    priority: 'high' | 'medium' | 'low';
    icon: string;
  }>;
  
  trends: Array<{
    metric: string;
    direction: 'up' | 'down' | 'stable';
    change: string;                // e.g., "-30%"
    significance: 'high' | 'medium' | 'low';
  }>;
  
  correlations: Array<{
    metrics: [string, string];
    relationship: string;
    strength: number;
    insight: string;
  }>;
  
  meta: {
    checkInCount: number;
    dayCount: number;
    dateRange: { start: string; end: string };
    lastUpdated: string;
  };
  
  // Conversational response
  answer: string;                  // Clean text (no JSON block)
  
  // Analytics metadata
  backendLogging?: {
    session_type: 'reflection' | 'concern' | 'celebration';
    mood_tone: 'positive' | 'neutral' | 'struggling';
    themes_active: string[];
    personalization_flags: Record<string, any>;
  };
}

Transformation Process​

Purpose: Convert reasoning output to dashboard format

const result = await recoveredOutputAdapter.transform(inputData, context);

Transformations Applied:

  • Extract headline from explanation summary
  • Map confidence from validation warnings
  • Format insights with icons and severity
  • Generate trend descriptions from temporal analysis
  • Structure correlations for UI display
  • Build metadata from intake context

Purpose: Generate conversational response

Agent Configuration:

  • Agent ID: recovered-business-logic-agent
  • System Prompt: Configured in Core.AgentProfiles (swappable)
  • Invocation: Uses agent's configured instructions (NO override)

Context Provided:

const reasoningContext = {
  objective: inputData.objective,
  temporal: inputData.temporal,
  pattern: inputData.pattern,
  correlation: inputData.correlation,
  domain: inputData.domain,
  explanation: inputData.explanation,
  validation: inputData.validation,
  result: inputData.result,
  dashboardSummary: result.output.summary,
  dashboardInsights: result.output.insights?.slice(0, 5),
  dashboardRecommendations: result.output.recommendations?.slice(0, 3),
  meta: result.output.meta,
};

Agent Task:

  • Address user's question directly
  • Reference specific data points from check-ins
  • Provide actionable guidance
  • Maintain supportive, non-clinical tone
  • Include backend logging JSON for analytics

Purpose: Parse analytics metadata from agent response

Expected Formats (Supported):

Standard Format (with markdown delimiters):

[Conversational answer text here...]
{
  "session_type": "reflection",
  "mood_tone": "positive",
  "themes_active": ["stress", "sleep"],
  "personalization_flags": { "example_flag": true }
}

Edge Case: Raw JSON (no markdown delimiters):

[Conversational answer text here...]
{"timestamp":"2026-01-08T16:20:00Z","session_type":"standard",...}

Extraction Logic (Multiple Regex Patterns):

const jsonBlockPatterns = [
  /\n\n```json\s*([\s\S]*?)```\s*$/,      // Standard: double newline + ```json
  /\n\n```\s*([\s\S]*?)```\s*$/,           // Double newline + ``` (no language tag)
  /\n```json\s*([\s\S]*?)```\s*$/,         // Single newline + ```json
  /\n```\s*\n?([\s\S]*?)\n?```\s*$/,       // Single newline + ``` (flexible whitespace)
  /```json\s*([\s\S]*?)```\s*$/,           // Just ```json anywhere at end
  /```\s*\n?(\{[\s\S]*?\})\n?```\s*$/,     // ``` with JSON object inside (flexible)
  /\n\s*(\{[\s\S]*?\})\s*$/,                // **Raw JSON object at end (no markdown)**
];

let jsonBlockMatch = null;
for (const pattern of jsonBlockPatterns) {
  jsonBlockMatch = rawAnswerText.match(pattern);
  if (jsonBlockMatch) break;
}

if (jsonBlockMatch) {
  // Clean answer (remove JSON block)
  answerText = rawAnswerText.slice(0, jsonBlockMatch.index).trim();
  
  // Parse backend logging
  try {
    backendLogging = JSON.parse(jsonBlockMatch[1].trim());
  } catch (parseErr) {
    logger.warn('Failed to parse backend logging JSON', { error: parseErr.message });
  }
}

Why Multiple Patterns?

  • LLM output format varies by model/prompt
  • Some models omit markdown delimiters
  • Some include extra whitespace
  • Some use single vs double newlines
  • Last pattern catches raw JSON without ``` delimiters (edge case fix Jan 2026)

Use Cases:

  • Session type tracking (reflection, concern, celebration)
  • Mood tone analysis for personalization
  • Active themes for follow-up questions
  • Personalization flags for UI customization
  • Timestamp for session logging

Purpose: Save output for dashboard queries

Table: RECOVERED.data_WorkflowOutputs

Schema:

CREATE TABLE RECOVERED.data_WorkflowOutputs (
  id UNIQUEIDENTIFIER PRIMARY KEY DEFAULT NEWID(),
  reasoning_task_id UNIQUEIDENTIFIER NOT NULL,
  agent_id VARCHAR(255) NOT NULL,
  agent_profile_id UNIQUEIDENTIFIER NULL,
  output_json NVARCHAR(MAX) NOT NULL,
  duration_ms INT NOT NULL,
  created_at DATETIME2 DEFAULT GETUTCDATE(),
  
  FOREIGN KEY (reasoning_task_id) 
    REFERENCES REASONING.data_WorkflowTasks(id)
);

Persisted Data:

await recoveredDataService.save({
  reasoningTaskId,
  agentId: 'recovered-business-logic-agent',
  agentProfileId: undefined, // TODO: Track when A/B testing prompts
  output: outputWithAnswer,
  durationMs: Date.now() - startTime,
});

Error Handling:

  • Logs warning if save fails
  • Does NOT fail workflow (user still gets output)
  • Useful for queries like "show recent analysis results"

Agent Prompt Strategy​

warning

DO NOT override system prompt in code

The agent is invoked using its configured prompt from Core.AgentProfiles:

// ✅ Correct: Uses agent's configured prompt
const agentResponse = await agent.generate([
  { role: 'user', content: contextString }
]);

// ❌ Wrong: Overrides prompt (prevents A/B testing)
const agentResponse = await agent.generate([
  { role: 'user', content: contextString }
], {
  systemPrompt: 'Generate empathetic response...'
});

Benefits of Prompt Separation:

  • A/B Testing: Swap prompts without code changes
  • Iteration: Update tone/style via admin tools
  • Tracking: Link outputs to specific prompt versions
  • Rollback: Revert prompt if quality degrades

Data Flow Diagram​

Usage Examples​

Basic Usage (Raw Export)​

// Paste raw Recovered export data
const result = await mastra.runWorkflow('recoveredAnalysisWorkflow', {
  jsonData: JSON.stringify([
    {
      "Created At": "December 13, 2025, 8:48 AM",
      "Label": "stress-checkin",
      "Value": "{\"options\":[\"very-low\"],\"notes\":\"\"}",
      "User ID": "6114",
      "Kind": "stress-checkin"
    },
    // ... more check-ins
  ]),
  userPrompt: 'How have my stress levels been trending?',
});

console.log(result.summary.headline);
// "Your stress is improving"

console.log(result.answer);
// "Based on your check-ins over the past 2 days, I can see that your stress 
//  levels have decreased significantly. On December 12th you reported feeling 
//  'low' stress, and by December 13th you were feeling 'very-low' stress..."

Using Pre-Normalized Data​

// Use processed canonical format (skips normalization)
const result = await mastra.runWorkflow('recoveredAnalysisWorkflow', {
  jsonData: JSON.stringify({
    objective: "Analyze my stress patterns",
    series: [
      {
        name: "Stress Check-ins",
        category: "stress-checkin",
        data: [
          { timestamp: "2025-12-12T08:36", value: "low" },
          { timestamp: "2025-12-13T08:48", value: "very-low" }
        ]
      }
    ],
    datasets: [
      {
        label: "Stress Level",
        values: [2, 1],
        dates: ["2025-12-12", "2025-12-13"]
      }
    ]
  }),
  userPrompt: 'Show me stress trends',
});

Enable Step Orchestration​

// Let reasoning engine skip unnecessary steps (saves time & cost)
const result = await mastra.runWorkflow('recoveredAnalysisWorkflow', {
  jsonData: myJsonData,
  userPrompt: 'Just analyze my sleep quality',
  enableStepOrchestration: true, // Skips irrelevant steps
});

console.log(result.executedSteps);
// ["temporal-analysis", "explanation-agent", "validation-agent"]
// (Skipped pattern, correlation, domain since not needed for simple query)

Error Handling​

Invalid JSON Format

Error: Failed to parse JSON: Unexpected token

Cause: User provided malformed JSON string

Solution:

  • Validate JSON syntax before submitting
  • Use JSON.stringify() if working with JS objects
  • Check for trailing commas, unescaped quotes
Unrecognized Data Format

Error: Input format not recognized as Recovered export

Cause: Data structure doesn't match any known format

Solution:

  • Ensure data has either:
    • Large format fields: created_at, label, value, user_id
    • Small format fields: Created At, Label, Value, User ID
    • OR canonical fields: series, datasets, objective
  • Check field names (case-sensitive)
Reasoning Engine Failure

Error: Reasoning engine workflow failed: [reason]

Cause: Sub-workflow execution error

Solution:

  • Check reasoning engine logs for details
  • Verify Azure OpenAI endpoint is accessible
  • Ensure sufficient data (at least 2 days, 3+ check-ins)
  • Review validation warnings in output
Agent Not Found

Error: Recovered Phase 3 requires agent "recovered-business-logic-agent" to be registered

Cause: Business-logic agent not available in Mastra instance

Solution:

  • Verify agent is registered in apps/runtime/src/mastra-runtime.ts
  • Check agent creation wasn't skipped due to missing dependencies
  • Ensure agent profile exists in Core.AgentProfiles
Database Persistence Failure

Warning: Failed to persist output to DB: [reason]

Cause: SQL connection or schema issue

Impact: Workflow succeeds, user gets output, but no DB record

Solution:

  • Check Azure SQL connection settings
  • Verify RECOVERED schema exists
  • Ensure data_WorkflowOutputs table is created
  • Review foreign key constraints (reasoning_task_id must exist)

Performance Considerations​

Memory vs Database Trade-offs​

ModeConfigurationUse WhenProsCons
In-MemoryREASONING_DB_STEP_DATA=falseSmall datasets (<100 records)Faster (no DB I/O)Memory limits, no audit trail
DB-BackedREASONING_DB_STEP_DATA=trueLarge datasets (100+ records)Handles large data, audit trailSlower (DB reads), SQL required

Optimization Tips​

Enable Step Orchestration

Set enableStepOrchestration: true for simple queries to skip unnecessary steps

Pre-Normalize Large Exports

Process large exports offline, pass canonical format to workflow

Batch Check-Ins

Group check-ins by day/week before analysis to reduce series size

Cache Domain Retrieval

Azure AI Search results are cached by query, reuse objectives when possible

Debugging Tools​

View Workflow Graph​

# Navigate to workflow visualizer
http://localhost:4111/workflows/recoveredAnalysisWorkflow/graph

Check Execution Logs​

// Logs are tagged with [recovered-workflow]
// Search for:
// - "Phase 1: Starting normalization"
// - "Phase 2: Reasoning engine completed"
// - "Phase 3: Agent response received"

Inspect Step Data (DB Mode)​

-- Get step outputs for a specific run
SELECT step_name, output_json
FROM REASONING.data_WorkflowStepOutputs
WHERE task_id = '...';

-- Get all recovered outputs
SELECT reasoning_task_id, agent_id, output_json, created_at
FROM RECOVERED.data_WorkflowOutputs
ORDER BY created_at DESC;

Test with Example Data​

# Small dataset (8 check-ins, 2 days)
cat examples/recovered/Recovered_Small.json

# Large dataset (100+ check-ins, 30 days)
cat examples/recovered/Recovered_Large.json

# Pre-processed canonical format
cat examples/recovered/Recovered_Small.processed.dropIdentifiers-false.json

Next Steps​

Try the Workflow

Paste sample data and run analysis

Customize Agent Prompt

Adjust conversational tone and style

View Workflow Code

Learn how formats are detected and transformed

Review Schema

Understand data persistence and queries