Workflow Timeouts

import signal
import time

class TimeoutError(Exception):
    pass

def run_with_timeout(fn, args: dict, timeout_seconds: int) -> dict:
    """
    Run a function with a timeout. Returns error dict if timeout exceeded.
    Uses signal.alarm on Unix systems.
    """
    def _timeout_handler(signum, frame):
        raise TimeoutError(f"Exceeded {timeout_seconds}s timeout")

    # Set the timeout alarm
    signal.signal(signal.SIGALRM, _timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        result = fn(**args)
        signal.alarm(0)  # Cancel the alarm
        return {"status": "success", "result": result}
    except TimeoutError as e:
        return {"status": "error", "error_type": "timeout",
                "error": str(e)}
    except Exception as e:
        signal.alarm(0)
        return {"status": "error", "error_type": "execution", "error": str(e)}

# Example usage in a workflow node
def slow_api_call(query: str) -> str:
    time.sleep(0.5)  # Simulates a fast call (use 10 for a slow one)
    return f"Results for: {query}"

result = run_with_timeout(slow_api_call, {"query": "Python frameworks"}, timeout_seconds=5)
print(f"Status: {result['status']}")
if result["status"] == "success":
    print(f"Result: {result['result']}")