Enforce (node_id, artifact_name) uniqueness in published artifacts

.github/workflows/test.yaml

@@ -25,6 +25,7 @@ jobs:
       backend_worker: ${{ steps.filter.outputs.backend_worker }}
       backend_frame: ${{ steps.filter.outputs.backend_frame }}
       backend_flowfile: ${{ steps.filter.outputs.backend_flowfile }}
+      kernel: ${{ steps.filter.outputs.kernel }}
       frontend: ${{ steps.filter.outputs.frontend }}
       docs: ${{ steps.filter.outputs.docs }}
       shared: ${{ steps.filter.outputs.shared }}
@@ -46,6 +47,11 @@ jobs:
               - 'flowfile_frame/**'
             backend_flowfile:
               - 'flowfile/**'
+            kernel:
+              - 'kernel_runtime/**'
+              - 'flowfile_core/flowfile_core/kernel/**'
+              - 'flowfile_core/tests/flowfile/test_kernel_integration.py'
+              - 'flowfile_core/tests/kernel_fixtures.py'
             frontend:
               - 'flowfile_frontend/**'
             docs:
@@ -145,7 +151,7 @@ jobs:
           needs.detect-changes.outputs.shared == 'true' ||
           needs.detect-changes.outputs.test_workflow == 'true' ||
           github.event.inputs.run_all_tests == 'true'
-        run: poetry run pytest flowfile_core/tests --disable-warnings $COV_ARGS
+        run: poetry run pytest flowfile_core/tests -m "not kernel" --disable-warnings $COV_ARGS
         env:
           COV_ARGS: ${{ (matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12') && '--cov --cov-append --cov-report=' || '' }}
 
@@ -271,7 +277,7 @@ jobs:
           needs.detect-changes.outputs.test_workflow == 'true' ||
           github.event.inputs.run_all_tests == 'true'
         shell: pwsh
-        run: poetry run pytest flowfile_core/tests --disable-warnings
+        run: poetry run pytest flowfile_core/tests -m "not kernel" --disable-warnings
 
       - name: Run pytest for flowfile_worker
         if: |
@@ -299,6 +305,48 @@ jobs:
         shell: pwsh
         run: poetry run pytest flowfile/tests --disable-warnings
 
+  # Kernel integration tests - runs in parallel on a separate worker
+  kernel-tests:
+    needs: detect-changes
+    if: |
+      needs.detect-changes.outputs.kernel == 'true' ||
+      needs.detect-changes.outputs.backend_core == 'true' ||
+      needs.detect-changes.outputs.shared == 'true' ||
+      needs.detect-changes.outputs.test_workflow == 'true' ||
+      github.event.inputs.run_all_tests == 'true'
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: 'pip'
+
+      - name: Install Poetry
+        run: |
+          curl -sSL https://install.python-poetry.org | python -
+          echo "$HOME/.poetry/bin" >> $GITHUB_PATH
+
+      - name: Install Dependencies
+        run: |
+          poetry install --no-interaction --no-ansi --with dev
+
+      - name: Build kernel Docker image
+        run: |
+          docker build -t flowfile-kernel -f kernel_runtime/Dockerfile kernel_runtime/
+
+      - name: Run kernel_runtime unit tests
+        run: |
+          pip install -e "kernel_runtime/[test]"
+          python -m pytest kernel_runtime/tests -v --disable-warnings
+
+      - name: Run kernel integration tests
+        run: |
+          poetry run pytest flowfile_core/tests -m kernel -v --disable-warnings
+
   # Frontend web build test - runs when frontend changes or test workflow changes
   test-web:
     needs: detect-changes
@@ -472,7 +520,7 @@ jobs:
 
   # Summary job - always runs to provide status
   test-summary:
-    needs: [detect-changes, backend-tests, backend-tests-windows, test-web, electron-tests-macos, electron-tests-windows, docs-test]
+    needs: [detect-changes, backend-tests, backend-tests-windows, kernel-tests, test-web, electron-tests-macos, electron-tests-windows, docs-test]
     if: always()
     runs-on: ubuntu-latest
     steps:
@@ -485,6 +533,7 @@ jobs:
           echo "  - Backend Worker: ${{ needs.detect-changes.outputs.backend_worker }}"
           echo "  - Backend Frame: ${{ needs.detect-changes.outputs.backend_frame }}"
           echo "  - Backend Flowfile: ${{ needs.detect-changes.outputs.backend_flowfile }}"
+          echo "  - Kernel: ${{ needs.detect-changes.outputs.kernel }}"
           echo "  - Frontend: ${{ needs.detect-changes.outputs.frontend }}"
           echo "  - Docs: ${{ needs.detect-changes.outputs.docs }}"
           echo "  - Shared/Dependencies: ${{ needs.detect-changes.outputs.shared }}"
@@ -493,6 +542,7 @@ jobs:
           echo "Job results:"
           echo "  - Backend Tests: ${{ needs.backend-tests.result }}"
           echo "  - Backend Tests (Windows): ${{ needs.backend-tests-windows.result }}"
+          echo "  - Kernel Tests: ${{ needs.kernel-tests.result }}"
           echo "  - Web Tests: ${{ needs.test-web.result }}"
           echo "  - Electron Tests (macOS): ${{ needs.electron-tests-macos.result }}"
           echo "  - Electron Tests (Windows): ${{ needs.electron-tests-windows.result }}"
@@ -501,6 +551,7 @@ jobs:
           # Fail if any non-skipped job failed
           if [[ "${{ needs.backend-tests.result }}" == "failure" ]] || \
              [[ "${{ needs.backend-tests-windows.result }}" == "failure" ]] || \
+             [[ "${{ needs.kernel-tests.result }}" == "failure" ]] || \
              [[ "${{ needs.test-web.result }}" == "failure" ]] || \
              [[ "${{ needs.electron-tests-macos.result }}" == "failure" ]] || \
              [[ "${{ needs.electron-tests-windows.result }}" == "failure" ]] || \

.gitignore

@@ -67,6 +67,9 @@ htmlcov/
 # Docker
 flowfile_data/
 
+# Egg info
+*.egg-info/
+
 # Secrets and keys - NEVER commit these
 master_key.txt
 *.key

flowfile_core/flowfile_core/catalog/__init__.py

@@ -0,0 +1,44 @@
+"""Flow Catalog service layer.
+
+Public interface:
+
+* ``CatalogService`` — business-logic orchestrator
+* ``CatalogRepository`` — data-access protocol (for type-hints / mocking)
+* ``SQLAlchemyCatalogRepository`` — concrete SQLAlchemy implementation
+* Domain exceptions (``CatalogError`` hierarchy)
+"""
+
+from .exceptions import (
+    CatalogError,
+    FavoriteNotFoundError,
+    FlowExistsError,
+    FlowNotFoundError,
+    FollowNotFoundError,
+    NamespaceExistsError,
+    NamespaceNotEmptyError,
+    NamespaceNotFoundError,
+    NestingLimitError,
+    NoSnapshotError,
+    NotAuthorizedError,
+    RunNotFoundError,
+)
+from .repository import CatalogRepository, SQLAlchemyCatalogRepository
+from .service import CatalogService
+
+__all__ = [
+    "CatalogService",
+    "CatalogRepository",
+    "SQLAlchemyCatalogRepository",
+    "CatalogError",
+    "NamespaceNotFoundError",
+    "NamespaceExistsError",
+    "NestingLimitError",
+    "NamespaceNotEmptyError",
+    "FlowNotFoundError",
+    "FlowExistsError",
+    "RunNotFoundError",
+    "NotAuthorizedError",
+    "FavoriteNotFoundError",
+    "FollowNotFoundError",
+    "NoSnapshotError",
+]

flowfile_core/flowfile_core/catalog/exceptions.py

@@ -0,0 +1,121 @@
+"""Domain-specific exceptions for the Flow Catalog system.
+
+These exceptions represent business-rule violations and are raised by the
+service layer.  Route handlers catch them and translate to appropriate
+HTTP responses.
+"""
+
+
+class CatalogError(Exception):
+    """Base exception for all catalog domain errors."""
+
+
+class NamespaceNotFoundError(CatalogError):
+    """Raised when a namespace lookup fails."""
+
+    def __init__(self, namespace_id: int | None = None, name: str | None = None):
+        self.namespace_id = namespace_id
+        self.name = name
+        detail = "Namespace not found"
+        if namespace_id is not None:
+            detail = f"Namespace with id={namespace_id} not found"
+        elif name is not None:
+            detail = f"Namespace '{name}' not found"
+        super().__init__(detail)
+
+
+class NamespaceExistsError(CatalogError):
+    """Raised when attempting to create a duplicate namespace."""
+
+    def __init__(self, name: str, parent_id: int | None = None):
+        self.name = name
+        self.parent_id = parent_id
+        super().__init__(
+            f"Namespace '{name}' already exists"
+            + (f" under parent_id={parent_id}" if parent_id is not None else " at root level")
+        )
+
+
+class NestingLimitError(CatalogError):
+    """Raised when attempting to nest namespaces deeper than catalog -> schema."""
+
+    def __init__(self, parent_id: int, parent_level: int):
+        self.parent_id = parent_id
+        self.parent_level = parent_level
+        super().__init__("Cannot nest deeper than catalog -> schema")
+
+
+class NamespaceNotEmptyError(CatalogError):
+    """Raised when trying to delete a namespace that still has children or flows."""
+
+    def __init__(self, namespace_id: int, children: int = 0, flows: int = 0):
+        self.namespace_id = namespace_id
+        self.children = children
+        self.flows = flows
+        super().__init__("Cannot delete namespace with children or flows")
+
+
+class FlowNotFoundError(CatalogError):
+    """Raised when a flow registration lookup fails."""
+
+    def __init__(self, registration_id: int | None = None, name: str | None = None):
+        self.registration_id = registration_id
+        self.name = name
+        detail = "Flow not found"
+        if registration_id is not None:
+            detail = f"Flow with id={registration_id} not found"
+        elif name is not None:
+            detail = f"Flow '{name}' not found"
+        super().__init__(detail)
+
+
+class FlowExistsError(CatalogError):
+    """Raised when attempting to create a duplicate flow registration."""
+
+    def __init__(self, name: str, namespace_id: int | None = None):
+        self.name = name
+        self.namespace_id = namespace_id
+        super().__init__(f"Flow '{name}' already exists in namespace_id={namespace_id}")
+
+
+class RunNotFoundError(CatalogError):
+    """Raised when a flow run lookup fails."""
+
+    def __init__(self, run_id: int):
+        self.run_id = run_id
+        super().__init__(f"Run with id={run_id} not found")
+
+
+class NotAuthorizedError(CatalogError):
+    """Raised when a user attempts an action they are not permitted to perform."""
+
+    def __init__(self, user_id: int, action: str = "perform this action"):
+        self.user_id = user_id
+        self.action = action
+        super().__init__(f"User {user_id} is not authorized to {action}")
+
+
+class FavoriteNotFoundError(CatalogError):
+    """Raised when a favorite record is not found."""
+
+    def __init__(self, user_id: int, registration_id: int):
+        self.user_id = user_id
+        self.registration_id = registration_id
+        super().__init__(f"Favorite not found for user={user_id}, flow={registration_id}")
+
+
+class FollowNotFoundError(CatalogError):
+    """Raised when a follow record is not found."""
+
+    def __init__(self, user_id: int, registration_id: int):
+        self.user_id = user_id
+        self.registration_id = registration_id
+        super().__init__(f"Follow not found for user={user_id}, flow={registration_id}")
+
+
+class NoSnapshotError(CatalogError):
+    """Raised when a run has no flow snapshot available."""
+
+    def __init__(self, run_id: int):
+        self.run_id = run_id
+        super().__init__(f"No flow snapshot available for run id={run_id}")