Role-Based Path Orchestration

The problem

Managing complex workflows is difficult because it is hard to control which specific steps a user or system is allowed to see or perform. It becomes messy to switch between different paths without breaking the whole process.

What it does

It creates a system that can turn specific parts of a workflow on or off based on the user's role. It tracks the progress of each step as it moves through these different paths.

Why it matters

It allows for precise control over how a process flows without having to rebuild the entire system for different users.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 orchestrator.py
Executing for role: admin

  -> Initialized
EXECUTED: init
  -> Processing Data
EXECUTED: process_data
  -> Report Generated
EXECUTED: generate_report
  -> Advanced Analysis Done
EXECUTED: advanced_analysis

Final States:
  init: COMPLETED
  process_data: COMPLETED
  generate_report: COMPLETED
  advanced_analysis: COMPLETED

Execution complete

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 250 lines, one file, standard library only.

import sys
from enum import Enum


class Status(Enum):
    PERMITTED = "permitted"
    RESTRICTED = "restricted"
    PENDING = "pending"
    SKIPPED = "skipped"


class FeatureToggle:
    def __init__(self):
        self.features = {
            'admin': {
                'init': True,
                'process_data': True,
                'generate_report': True,
                'advanced_analysis': True,
                'sentiment_analysis': True,
                'data_export': True,
                'ml_training': True,
            },
            'user': {
                'init': True,
                'process_data': True,
                'generate_report': True,
                'advanced_analysis': False,
                'sentiment_analysis': True,
                'data_export': True,
                'ml_training': False,
            },
            'guest': {
                'init': True,
                'process_data': False,
                'generate_report': False,
                'advanced_analysis': False,
                'sentiment_analysis': False,
                'data_export': False,
                'ml_training': False,
            }
        }

    def is_enabled(self, role, feature):
        return self.features.get(role, {}).get(feature, False)

    def evaluate_path_access(self, role, path):
        return self.is_enabled(role, path)


class TaskState:
    PENDING = "PENDING"
    IN_PROGRESS = "IN_PROGRESS"
    COMPLETED = "COMPLETED"
    SKIPPED = "SKIPPED"
    RESTRICTED = "RESTRICTED"
    BRANCHED = "BRANCHED"


class TaskResult:
    def __init__(self, success=True, branch=None, data=None):
        self.success = success
        self.branch = branch
        self.data = data or {}


class Task:
    def __init__(self, name, dependencies=None, action=None):
        self.name = name
        self.dependencies = dependencies or []
        self.action = action


class BranchMap:
    def __init__(self, output_key, routes):
        self.output_key = output_key
        self.routes = routes


class BranchingTask(Task):
    def __init__(self, name, dependencies=None, action=None, branch_map=None, fallback=None):
        super().__init__(name, dependencies, action)
        self.branch_map = branch_map
        self.fallback = fallback


class PathOrchestrator:
    def __init__(self, feature_toggle, tasks, roles_paths=None):
        self.feature_toggle = feature_toggle
        self.tasks = tasks
        self.roles_paths = roles_paths or {}
        self.states = {}
        self.results = {}

    def evaluate_access(self, role, task_name):
        access_status = self.feature_toggle.evaluate_path_access(role, task_name)
        if not access_status:
            return Status.RESTRICTED
        return Status.PERMITTED

    def check_path_restrictions(self, role, path):
        allowed_paths = self.roles_paths.get(role, set())
        if allowed_paths and path not in allowed_paths:
            return Status.RESTRICTED
        return Status.PERMITTED

    def _resolve_branch(self, task, result):
        if not isinstance(task, BranchingTask) or not task.branch_map:
            return None

        branch_key = task.branch_map.output_key
        branch_value = result.data.get(branch_key) if result.data else None

        if branch_value and branch_value in task.branch_map.routes:
            return task.branch_map.routes[branch_value]
        return task.fallback

    def execute_tasks(self, role):
        completed = set()
        task_order = list(self.tasks.keys())
        idx = 0

        while idx < len(task_order):
            task_name = task_order[idx]
            task = self.tasks[task_name]

            if task_name in self.states:
                idx += 1
                continue

            self.states[task_name] = TaskState.PENDING

            path_access = self.evaluate_access(role, task_name)
            role_path_access = self.check_path_restrictions(role, task_name)

            if path_access == Status.RESTRICTED or role_path_access == Status.RESTRICTED:
                self.states[task_name] = TaskState.RESTRICTED
                print(f"RESTRICTED: {task_name}")
                idx += 1
                continue

            deps_met = all(dep in completed for dep in task.dependencies)
            if not deps_met:
                self.states[task_name] = TaskState.SKIPPED
                print(f"SKIPPED: {task_name} (dependencies not met)")
                self.results[task_name] = TaskResult(success=False)
                idx += 1
                continue

            self.states[task_name] = TaskState.IN_PROGRESS
            result = TaskResult(success=True, data={})
            if task.action:
                result = task.action()

            if not isinstance(result, TaskResult):
                result = TaskResult(success=True, data=result if isinstance(result, dict) else {})

            self.results[task_name] = result

            if result.success:
                completed.add(task_name)
                self.states[task_name] = TaskState.COMPLETED
                print(f"EXECUTED: {task_name}")

                branch_target = self._resolve_branch(task, result)
                if branch_target and branch_target in self.tasks:
                    self.states[task_name] = TaskState.BRANCHED
                    print(f"  -> BRANCH to: {branch_target}")

                    for route_value, route_task_name in task.branch_map.routes.items():
                        if route_task_name != branch_target and route_task_name in self.tasks:
                            self.states[route_task_name] = TaskState.SKIPPED
                            self.results[route_task_name] = TaskResult(success=False, branch=None)
                            print(f"  -> SKIPPED branch: {route_task_name}")

                    insert_idx = idx + 1
                    if branch_target not in task_order[insert_idx:]:
                        task_order.insert(insert_idx, branch_target)
            else:
                fallback = getattr(task, 'fallback', None)
                if fallback and fallback in self.tasks:
                    print(f"  -> FALLBACK to: {fallback}")
                    task_order.insert(idx + 1, fallback)

            idx += 1

        return self.states


class RoleBasedPathOrchestrator:
    def __init__(self):
        self.feature_toggle = FeatureToggle()
        self.tasks = {
            'init': Task(name='init', action=lambda: print("  -> Initialized")),
            'process_data': Task(name='process_data', dependencies=['init'],
                                 action=lambda: (
                                     print("  -> Processing Data"),
                                     TaskResult(success=True, data={"sentiment": "positive"})
                                  )[1]),
            'sentiment_analysis': Task(name='sentiment_analysis', dependencies=['process_data'],
                action=lambda: TaskResult(success=True, data={})),
            'generate_report': BranchingTask(
                name='generate_report',
                dependencies=['process_data'],
                branch_map=BranchMap("sentiment", {"positive": "data_export", "negative": "ml_training"}),
                action=lambda: TaskResult(success=True, data={"sentiment": "positive"})
            ),
            'data_export': Task(name='data_export', dependencies=['generate_report'],
                action=lambda: print("  -> Data Exported")),
            'ml_training': Task(name='ml_training', dependencies=['generate_report'],
                action=lambda: print("  -> ML Training Run")),
            'advanced_analysis': Task(name='advanced_analysis', dependencies=['process_data'],
                action=lambda: print("  -> Advanced Analysis Done")),
        }
        self.roles_paths = {
            'admin': {'init', 'process_data', 'generate_report', 'advanced_analysis',
                      'sentiment_analysis', 'data_export', 'ml_training'},
            'user': {'init', 'process_data', 'generate_report',
                     'sentiment_analysis', 'data_export'},
            'guest': {'init'},
        }
        self.orchestrator = PathOrchestrator(self.feature_toggle, self.tasks, self.roles_paths)

    def run(self, role):
        print(f"Executing for role: {role}\n")
        states = self.orchestrator.execute_tasks(role)
        print('\nFinal States:')
        for name, state in states.items():
            print(f"  {name}: {state}")
        print('\nExecution complete')
        return states


def _resolve_role_from_args(args):
    roles = {'admin', 'user', 'guest'}
    for arg in args:
        if arg in roles:
            return arg
    return 'admin'


if __name__ == "__main__":
    orchestrator = RoleBasedPathOrchestrator()
    role = _resolve_role_from_args(sys.argv[1:]) if len(sys.argv) > 1 else 'admin'
    result = orchestrator.run(role)

    for name, state in result.items():
        if state == TaskState.RESTRICTED:
            sys.exit(1)
    sys.exit(0)