Fix __main__ module resolution for RQ workers and improve logging

- Add automatic resolution of __main__ module paths to fully qualified names
- Functions defined in scripts run as __main__ are now properly resolved
- Update examples/with_rq.py to use logging with timestamps
- Improve error messages for unresolvable module paths

This fixes the issue where functions were referenced as __main__.add instead
of the full path (e.g., examples.with_rq.add), making them importable by RQ workers.

Author: koenvo

durable_monty/__init__.py

@@ -5,11 +5,10 @@
 execute tasks in parallel, and resume when results are ready.
 """
 
-__version__ = "0.1.3"
+__version__ = "0.1.4"
 
 from durable_monty.models import init_db, Execution, Call, ExecutionStatus, CallStatus
 from durable_monty.service import OrchestratorService
-from durable_monty.functions import register_function, FUNCTION_REGISTRY
 from durable_monty.worker import Worker
 from durable_monty.executor import Executor, LocalExecutor
 
@@ -24,6 +23,4 @@
     "Worker",
     "Executor",
     "LocalExecutor",
-    "register_function",
-    "FUNCTION_REGISTRY",
 ]

durable_monty/service.py

@@ -1,14 +1,81 @@
 """Orchestrator service that manages executions."""
 
 import uuid
-from typing import Any
+import inspect
+import os
+import sys
+from typing import Any, Callable
+from pathlib import Path
 import pydantic_monty
 from sqlalchemy.orm import Session
 from sqlalchemy import Engine
 
 from durable_monty.models import Execution, Call, ExecutionStatus, CallStatus, to_json, from_json
 
 
+def _resolve_function_path(func: Callable) -> str:
+    """
+    Resolve the full import path for a function, handling __main__ modules.
+
+    When a script is run directly (python script.py), functions defined in it
+    have __module__ = "__main__". This converts it to the actual module path
+    that can be imported by workers.
+
+    Args:
+        func: Function object
+
+    Returns:
+        Full import path like "examples.with_rq.add"
+
+    Raises:
+        ValueError: If module path cannot be resolved
+    """
+    module_name = func.__module__
+    func_name = func.__name__
+
+    # If not __main__, use as-is
+    if module_name != "__main__":
+        return f"{module_name}.{func_name}"
+
+    # Try to resolve __main__ to actual module path
+    try:
+        # Get the file where the function is defined
+        source_file = inspect.getfile(func)
+        source_path = Path(source_file).resolve()
+
+        # Find which sys.path entry contains this file
+        for path_entry in sys.path:
+            try:
+                path_entry = Path(path_entry).resolve()
+                if source_path.is_relative_to(path_entry):
+                    # Convert file path to module path
+                    rel_path = source_path.relative_to(path_entry)
+
+                    # Remove .py extension and convert to module notation
+                    module_parts = list(rel_path.parts[:-1]) + [rel_path.stem]
+                    resolved_module = ".".join(module_parts)
+
+                    return f"{resolved_module}.{func_name}"
+            except (ValueError, OSError):
+                continue
+
+        # Could not resolve - provide helpful error
+        raise ValueError(
+            f"Cannot resolve module path for function '{func_name}' defined in __main__.\n"
+            f"The function is in '{source_file}' which is not in any Python path.\n"
+            f"Either:\n"
+            f"  1. Import the function from its module instead of running the script directly\n"
+            f"  2. Ensure the script's directory is in PYTHONPATH\n"
+            f"  3. Pass the full import path as a string instead of the function object"
+        )
+
+    except (TypeError, OSError) as e:
+        raise ValueError(
+            f"Cannot determine source file for function '{func_name}': {e}\n"
+            f"Pass the full import path as a string instead of the function object."
+        )
+
+
 class OrchestratorService:
     """Service for managing durable executions."""
 
@@ -18,18 +85,46 @@ def __init__(self, engine: Engine):
     def start_execution(
         self,
         code: str,
-        external_functions: list[str],
+        external_functions: list[str | Callable],  # Accept strings or callable objects
         inputs: dict | None = None,
     ) -> str:
-        """Schedule a new workflow execution. Returns execution_id."""
+        """
+        Schedule a new workflow execution. Returns execution_id.
+
+        Args:
+            code: Python code to execute
+            external_functions: List of function names (as full paths or callable objects)
+            inputs: Optional input variables for the code
+
+        Example:
+            # Pass actual function objects (recommended)
+            exec_id = service.start_execution(code, [add, multiply])
+
+            # Or pass full import paths as strings
+            exec_id = service.start_execution(code, ["myapp.tasks.add"])
+        """
         execution_id = str(uuid.uuid4())
 
+        # Convert callable objects to {short_name: full_path} mapping
+        function_mapping = {}
+        for func in external_functions:
+            if callable(func):
+                # Extract full path and short name from function object
+                # This handles __main__ module resolution
+                full_path = _resolve_function_path(func)
+                short_name = func.__name__
+                function_mapping[short_name] = full_path
+            else:
+                # String path - extract short name
+                short_name = func.rsplit(".", 1)[-1] if "." in func else func
+                function_mapping[short_name] = func
+
         with Session(self.engine) as session:
-            # Just save to DB - worker will pick it up
+            # Save mapping to DB - worker will use it
             execution = Execution(
                 id=execution_id,
                 code=code,
-                external_functions=to_json(external_functions),
+                external_functions=to_json(function_mapping),
                 status=ExecutionStatus.SCHEDULED,
                 inputs=to_json(inputs),
             )
@@ -53,14 +148,16 @@ def process_execution(
             if not execution:
                 return
 
+            # Load function mapping {short_name: full_path} for converting names
+            function_mapping = from_json(execution.external_functions)
+
             # Get Monty progress based on execution status
             if execution.status == ExecutionStatus.SCHEDULED:
                 # First time - start fresh
-                external_functions = from_json(execution.external_functions)
                 inputs = from_json(execution.inputs)
                 m = pydantic_monty.Monty(
                     execution.code,
-                    external_functions=external_functions,
+                    external_functions=list(function_mapping.keys()),  # Pass short names to Monty
                     inputs=list(inputs.keys()) if inputs else None,
                 )
                 progress = m.start(inputs=inputs) if inputs else m.start()
@@ -117,12 +214,16 @@ def process_execution(
                 # Save all calls in this group
                 for call_id in progress.pending_call_ids:
                     call_info = pending_calls[call_id]
+                    # Convert short name to full path for execution
+                    short_name = call_info["function"]
+                    full_path = function_mapping.get(short_name, short_name)
                     call = Call(
                         execution_id=execution_id,
                         resume_group_id=new_resume_group_id,
                         call_id=call_id,
-                        function_name=call_info["function"],
+                        function_name=full_path,  # Store full path for RQ workers
                         args=to_json(call_info["args"]),
+                        kwargs=to_json(call_info["kwargs"]),
                         status=CallStatus.PENDING,
                     )
                     session.add(call)
@@ -183,6 +284,7 @@ def poll(self, execution_id: str | None = None) -> dict[str, Any] | list[dict[st
                     "call_id": c.call_id,
                     "function_name": c.function_name,
                     "args": from_json(c.args),
+                    "kwargs": from_json(c.kwargs),
                     "status": c.status,
                 }
                 for c in calls
@@ -245,6 +347,7 @@ def get_pending_calls(self, execution_id: str) -> list[dict]:
                     "call_id": c.call_id,
                     "function_name": c.function_name,
                     "args": from_json(c.args),
+                    "kwargs": from_json(c.kwargs),
                 }
                 for c in calls
             ]

examples/with_rq.py

@@ -4,27 +4,41 @@
 1. Start Redis: redis-server
 2. Start RQ worker: rq worker durable-monty
 3. Run: uv sync --extra rq && uv run python examples/with_rq.py
-"""
 
+Note: Functions must be importable by RQ workers. We import them from the module
+(not __main__) so they have the correct __module__ attribute for RQ to find them.
+"""
+import logging
 import time
 import threading
-from durable_monty import init_db, OrchestratorService, Worker, register_function
+from durable_monty import init_db, OrchestratorService, Worker
 from durable_monty.executors.rq import RQExecutor
 
+# Configure logging with timestamps
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(levelname)s - %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+logger = logging.getLogger(__name__)
 
-@register_function("add")
+
+# Define functions that RQ workers can import
 def add(a, b):
+    time.sleep(2)
     return a + b
 
 
-@register_function("multiply")
 def multiply(a, b):
+    time.sleep(2)
     return a * b
 
 
 code = """
 from asyncio import gather
 results = await gather(add(1, 2), add(3, 4), multiply(5, 6))
+results += [await add(5, 7)]
+results += await gather(add(1, 2), add(3, 4), multiply(5, 6))
 sum(results)
 """
 
@@ -34,28 +48,22 @@ def multiply(a, b):
     try:
         executor = RQExecutor()
     except Exception as e:
-        print(f"Error: {e}")
-        print("Make sure Redis is running: redis-server")
+        logger.error(f"Failed to connect to Redis: {e}")
+        logger.error("Make sure Redis is running: redis-server")
         exit(1)
 
-    # Schedule execution
-    exec_id = service.start_execution(code, ["add", "multiply"])
-    print(f"Scheduled: {exec_id[:8]}...")
+    # Schedule execution - pass function objects
+    exec_id = service.start_execution(code, [add, multiply])
+    logger.info(f"Scheduled execution: {exec_id[:8]}...")
 
     # Run worker
-    worker = Worker(service, executor)
-
-    def run_worker():
-        for _ in range(20):
-            worker.run(once=True)
-            time.sleep(0.5)
+    worker = Worker(service, executor, poll_interval=0.1)
+    logger.info("Starting worker (will run until execution completes)...")
 
-    thread = threading.Thread(target=run_worker)
-    thread.start()
-    thread.join()
+    worker.run(until_complete=True)
 
     # Check result
     result = service.poll(exec_id)
-    print(f"Status: {result['status']}, Output: {result['output']}")
+    logger.info(f"Execution {result['status']}: output = {result['output']}")
     if result["status"] != "completed":
-        print("Note: Start RQ workers with: rq worker durable-monty")
+        logger.warning("Note: Start RQ workers with: rq worker durable-monty")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "durable-monty"
-version = "0.1.3"
+version = "0.1.4"
 description = "Durable functions using monty-python"
 requires-python = ">=3.10"
 dependencies = [