Enable executor recording mode

.github/workflows/promptflow-executor-e2e-test.yml

@@ -1,8 +1,8 @@
 name: promptflow-executor-e2e-test
 on:
   schedule:
-    - cron: "40 20 * * *" # Every day starting at 4:40 BJT
-  pull_request_target:
+    - cron: "35 17 * * *" # Every day starting at 1:35 BJT
+  pull_request:
     paths:
       - src/promptflow/*
       - src/promptflow/promptflow/*
@@ -26,28 +26,13 @@ env:
   PYTHONPATH: ${{ github.workspace }}/src/promptflow
   IS_IN_CI_PIPELINE: "true"
 jobs:
-  authorize:
-    environment:
-      # forked prs from pull_request_target will be run in external environment, domain prs will be run in internal environment
-      ${{ github.event_name == 'pull_request_target' &&
-      github.event.pull_request.head.repo.full_name != github.repository &&
-      'external' || 'internal' }}
-    runs-on: ubuntu-latest
-    steps:
-      - run: true
   build:
-    needs: authorize
     strategy:
       fail-fast: false
     runs-on: ubuntu-latest
     steps:
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Display and Set Environment Variables
       run: |
         env | sort >> $GITHUB_OUTPUT
@@ -78,13 +63,10 @@ jobs:
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
     steps:
+    - name: set test mode
+      run: echo "PROMPT_FLOW_TEST_MODE=$(if [[ "${{ github.event_name }}" == "pull_request" ]]; then echo replay; else echo live; fi)" >> $GITHUB_ENV
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Python Setup - ${{ matrix.os }} - Python Version 3.9
       uses: "./.github/actions/step_create_python_environment"
       with:
@@ -103,32 +85,36 @@ jobs:
         gci ./promptflow -Recurse | % {if ($_.Name.Contains('.whl')) {python -m pip install "$($_.FullName)[executor-service]"}}
         gci ./promptflow-tools -Recurse | % {if ($_.Name.Contains('.whl')) {python -m pip install $_.FullName}}
         pip freeze
-    - name: Azure Login
+    - name: Azure login (non pull_request workflow)
+      if: github.event_name != 'pull_request'
       uses: azure/login@v1
       with:
         creds: ${{ secrets.AZURE_CREDENTIALS }}
-    - name: Generate Configs
+    - name: generate live test resources (non pull_request workflow)
+      if: github.event_name != 'pull_request'
       uses: "./.github/actions/step_generate_configs"
       with:
         targetFolder: ${{ env.testWorkingDirectory }}
-    - name: Get number of CPU cores
-      uses: SimenB/github-actions-cpu-cores@v1
-      id: cpu-cores
+    - name: generate live test resources (pull_request workflow)
+      if: github.event_name == 'pull_request'
+      shell: pwsh
+      working-directory: ${{ env.testWorkingDirectory }}
+      run: |
+        cp ${{ github.workspace }}/src/promptflow/dev-connections.json.example ${{ github.workspace }}/src/promptflow/connections.json
     - name: Run Coverage Test
       shell: pwsh
       working-directory: ${{ github.workspace }}
       run: |
         gci env:* | sort-object name
-        az account show
         pip install langchain
+
         # numexpr is required by langchain in e2e tests.
         pip install numexpr
         python scripts/building/run_coverage_tests.py `
           -p ${{ env.testWorkingDirectory }}/promptflow `
           -t ${{ env.testWorkingDirectory }}/tests/executor/e2etests `
           -l eastus `
           -m "all" `
-          -n ${{ steps.cpu-cores.outputs.count }}`
           --coverage-config ${{ env.testWorkingDirectory }}/tests/executor/.coveragerc `
           --disable-cov-branch
     - name: Upload Test Results
@@ -139,6 +125,7 @@ jobs:
         path: |
           ${{ github.workspace }}/*.xml
           ${{ github.workspace }}/htmlcov/
+          ${{ env.testWorkingDirectory }}/tests/sdk_cli_test/count.json
   publish-test-results:
     name: "Publish Tests Results"
     needs: executor_e2e_tests
@@ -152,11 +139,6 @@ jobs:
     steps:
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Publish Test Results
       uses: "./.github/actions/step_publish_test_results"
       with:

.github/workflows/promptflow-executor-unit-test.yml

@@ -1,8 +1,8 @@
 name: promptflow-executor-unit-test
 on:
   schedule:
-    - cron: "40 19 * * *" # Every day starting at 3:40 BJT
-  pull_request_target:
+    - cron: "35 18 * * *" # Every day starting at 2:35 BJT
+  pull_request:
     paths:
       - src/promptflow/*
       - src/promptflow/promptflow/*
@@ -26,28 +26,13 @@ env:
   PYTHONPATH: ${{ github.workspace }}/src/promptflow
   IS_IN_CI_PIPELINE: "true"
 jobs:
-  authorize:
-    environment:
-      # forked prs from pull_request_target will be run in external environment, domain prs will be run in internal environment
-      ${{ github.event_name == 'pull_request_target' &&
-      github.event.pull_request.head.repo.full_name != github.repository &&
-      'external' || 'internal' }}
-    runs-on: ubuntu-latest
-    steps:
-      - run: true
   build:
-    needs: authorize
     strategy:
       fail-fast: false
     runs-on: ubuntu-latest
     steps:
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Display and Set Environment Variables
       run: |
         env | sort >> $GITHUB_OUTPUT
@@ -78,13 +63,10 @@ jobs:
         os: [ubuntu-latest]
     runs-on: ${{ matrix.os }}
     steps:
+    - name: set test mode
+      run: echo "PROMPT_FLOW_TEST_MODE=$(if [[ "${{ github.event_name }}" == "pull_request" ]]; then echo replay; else echo live; fi)" >> $GITHUB_ENV
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Display and Set Environment Variables
       run: |
         env | sort >> $GITHUB_OUTPUT
@@ -108,30 +90,32 @@ jobs:
         gci ./promptflow -Recurse | % {if ($_.Name.Contains('.whl')) {python -m pip install "$($_.FullName)[executor-service]"}}
         gci ./promptflow-tools -Recurse | % {if ($_.Name.Contains('.whl')) {python -m pip install $_.FullName}}
         pip freeze
-    - name: Azure Login
+    - name: Azure login (non pull_request workflow)
+      if: github.event_name != 'pull_request'
       uses: azure/login@v1
       with:
         creds: ${{ secrets.AZURE_CREDENTIALS }}
-    - name: Generate Configs
+    - name: generate live test resources (non pull_request workflow)
+      if: github.event_name != 'pull_request'
       uses: "./.github/actions/step_generate_configs"
       with:
         targetFolder: ${{ env.testWorkingDirectory }}
-    - name: Get number of CPU cores
-      uses: SimenB/github-actions-cpu-cores@v1
-      id: cpu-cores
+    - name: generate live test resources (pull_request workflow)
+      if: github.event_name == 'pull_request'
+      shell: pwsh
+      working-directory: ${{ env.testWorkingDirectory }}
+      run: |
+        cp ${{ github.workspace }}/src/promptflow/dev-connections.json.example ${{ github.workspace }}/src/promptflow/connections.json
     - name: Run Coverage Test
       shell: pwsh
       working-directory: ${{ github.workspace }}
       run: |
-        gci env:* | sort-object name
-        az account show
         pip install langchain
         python scripts/building/run_coverage_tests.py `
           -p ${{ env.testWorkingDirectory }}/promptflow `
           -t ${{ env.testWorkingDirectory }}/tests/executor/unittests `
           -l eastus `
           -m "all" `
-          -n ${{ steps.cpu-cores.outputs.count }} `
           --coverage-config ${{ env.testWorkingDirectory }}/tests/executor/.coveragerc `
           --disable-cov-branch
     - name: Upload Test Results
@@ -142,6 +126,7 @@ jobs:
         path: |
           ${{ github.workspace }}/*.xml
           ${{ github.workspace }}/htmlcov/
+          ${{ env.testWorkingDirectory }}/tests/sdk_cli_test/count.json
   publish-test-results:
     name: "Publish Tests Results"
     needs: executor_unit_tests
@@ -156,11 +141,6 @@ jobs:
     steps:
     - name: checkout
       uses: actions/checkout@v4
-      with:
-        ref: ${{ github.event.pull_request.head.sha || github.ref }}
-        fetch-depth: 0
-    - name: merge main to current branch
-      uses: "./.github/actions/step_merge_main"
     - name: Publish Test Results
       uses: "./.github/actions/step_publish_test_results"
       with:

src/promptflow/tests/executor/conftest.py

@@ -1,12 +1,136 @@
+import multiprocessing
+from pathlib import Path
+from unittest.mock import patch
+
 import pytest
+from executor.process_utils import (
+    MockForkServerProcess,
+    MockSpawnProcess,
+    current_process_manager_var,
+    current_process_wrapper_var,
+    override_process_class,
+)
+from sdk_cli_test.recording_utilities import (
+    RecordStorage,
+    delete_count_lock_file,
+    inject_async_with_recording,
+    inject_sync_with_recording,
+    is_live,
+    is_record,
+    is_replay,
+    mock_tool,
+    recording_array_extend,
+    recording_array_reset,
+)
+from sdk_cli_test.recording_utilities.record_storage import is_recording_enabled
 
 from promptflow._core.openai_injector import inject_openai_api
+from promptflow.executor._line_execution_process_pool import _process_wrapper
+from promptflow.executor._process_manager import create_spawned_fork_process_manager
+
+PROMPTFLOW_ROOT = Path(__file__) / "../../.."
+RECORDINGS_TEST_CONFIGS_ROOT = Path(PROMPTFLOW_ROOT / "tests/test_configs/node_recordings").resolve()
+
+
+@pytest.fixture
+def recording_setup():
+    patches = setup_recording()
+    try:
+        yield
+    finally:
+        for patcher in patches:
+            patcher.stop()
+
+
+def setup_recording():
+    patches = []
+
+    def start_patches(patch_targets):
+        for target, mock_func in patch_targets.items():
+            patcher = patch(target, mock_func)
+            patches.append(patcher)
+            patcher.start()
+
+    if is_replay() or is_record():
+        file_path = RECORDINGS_TEST_CONFIGS_ROOT / "executor_node_cache.shelve"
+        RecordStorage.get_instance(file_path)
+
+        from promptflow._core.tool import tool as original_tool
+
+        mocked_tool = mock_tool(original_tool)
+        patch_targets = {
+            "promptflow._core.tool.tool": mocked_tool,
+            "promptflow._internal.tool": mocked_tool,
+            "promptflow.tool": mocked_tool,
+            "promptflow._core.openai_injector.inject_sync": inject_sync_with_recording,
+            "promptflow._core.openai_injector.inject_async": inject_async_with_recording,
+        }
+        start_patches(patch_targets)
+        inject_openai_api()
+
+    if is_live():
+        patch_targets = {
+            "promptflow._core.openai_injector.inject_sync": inject_sync_with_recording,
+            "promptflow._core.openai_injector.inject_async": inject_async_with_recording,
+        }
+        start_patches(patch_targets)
+        inject_openai_api()
+
+    return patches
+
+
+def _default_mock_process_wrapper(*args, **kwargs):
+    # Default mock implementation of _process_wrapper in recording mode
+    setup_recording()
+    _process_wrapper(*args, **kwargs)
+
+
+def _default_mock_create_spawned_fork_process_manager(*args, **kwargs):
+    # Default mock implementation of create_spawned_fork_process_manager in recording mode
+    setup_recording()
+    create_spawned_fork_process_manager(*args, **kwargs)
+
+
+@pytest.fixture
+def process_override():
+    # This fixture is used to override the Process class to ensure the recording mode works
+
+    # Step I: set process pool targets placeholder with customized targets
+    current_process_wrapper_var.set(_default_mock_process_wrapper)
+    current_process_manager_var.set(_default_mock_create_spawned_fork_process_manager)
+
+    # Step II: override the process pool class
+    process_class_dict = {"spawn": MockSpawnProcess, "forkserver": MockForkServerProcess}
+    original_process_class = override_process_class(process_class_dict)
+
+    try:
+        yield
+    finally:
+        for start_method, MockProcessClass in process_class_dict.items():
+            if start_method in multiprocessing.get_all_start_methods():
+                multiprocessing.get_context(start_method).Process = original_process_class[start_method]
+                if start_method == multiprocessing.get_start_method():
+                    multiprocessing.Process = original_process_class[start_method]
+
+
+@pytest.fixture
+def recording_injection(recording_setup, process_override):
+    # This fixture is used to main entry point to inject recording mode into the test
+    try:
+        yield (is_replay() or is_record(), recording_array_extend)
+    finally:
+        if is_replay() or is_record():
+            RecordStorage.get_instance().delete_lock_file()
+        if is_live():
+            delete_count_lock_file()
+        recording_array_reset()
 
 
 @pytest.fixture(autouse=True, scope="session")
 def inject_api_executor():
-    """Inject OpenAI API during test session.
+    """Inject OpenAI API during test session when recording not enabled
 
     AOAI call in promptflow should involve trace logging and header injection. Inject
     function to API call in test scenario."""
-    inject_openai_api()
+    if not is_recording_enabled():
+        inject_openai_api()

src/promptflow/tests/executor/e2etests/test_activate.py

@@ -33,7 +33,7 @@
 ]
 
 
-@pytest.mark.usefixtures("dev_connections")
+@pytest.mark.usefixtures("dev_connections", "recording_injection")
 @pytest.mark.e2etest
 class TestExecutorActivate:
     @pytest.mark.parametrize("flow_folder", ACTIVATE_FLOW_TEST_CASES)