Session Factory & Pool

from ufo.module.session_pool import SessionFactory

factory = SessionFactory()

# Automatically detects platform and creates appropriate session
sessions = factory.create_session(
    task="email_task",
    mode="normal",
    plan="",
    request="Send an email to John"
)

def create_session(
    self,
    task: str,
    mode: str,
    plan: str,
    request: str = "",
    platform_override: Optional[str] = None,
    **kwargs,
) -> List[BaseSession]

factory = SessionFactory()

sessions = factory.create_session(
    task="browse_web",
    mode="normal",
    plan="",
    request="Open Chrome and navigate to google.com"
)

# Returns: [Session(task="browse_web", ...)]
session = sessions[0]
await session.run()

from aip.protocol.task_execution import TaskExecutionProtocol

protocol = TaskExecutionProtocol(websocket_connection)

sessions = factory.create_session(
    task="remote_control",
    mode="service",
    plan="",
    request="Click the Start button",
    task_protocol=protocol
)

# On Windows: Returns [ServiceSession(...)]
# On Linux: Returns [LinuxServiceSession(...)]

sessions = factory.create_session(
    task="batch_email",
    mode="follower",
    plan="/path/to/plan_folder",  # Folder with multiple .json plan files
    request=""
)

# Returns: [
#   FollowerSession(task="batch_email/plan1", ...),
#   FollowerSession(task="batch_email/plan2", ...),
#   FollowerSession(task="batch_email/plan3", ...)
# ]

# Execute with SessionPool
pool = SessionPool(sessions)
await pool.run_all()

sessions = factory.create_session(
    task="edit_document",
    mode="normal",
    plan="",
    request="Type 'Hello World'",
    platform_override="linux",
    application_name="gedit"
)

# Returns: [LinuxSession(task="edit_document", application_name="gedit")]

sessions = factory.create_session(
    task="complex_workflow",
    mode="operator",
    plan="",
    request="Organize my desktop files by date"
)

# Returns: [OpenAIOperatorSession(task="complex_workflow", ...)]

def create_service_session(
    self,
    task: str,
    should_evaluate: bool,
    id: str,
    request: str,
    task_protocol: Optional["TaskExecutionProtocol"] = None,
    platform_override: Optional[str] = None,
) -> BaseSession

factory = SessionFactory()
protocol = TaskExecutionProtocol(websocket)

session = factory.create_service_session(
    task="remote_task",
    should_evaluate=True,
    id="session_001",
    request="Open Notepad",
    task_protocol=protocol
)

# Type varies by platform
if isinstance(session, ServiceSession):
    print("Windows service session")
elif isinstance(session, LinuxServiceSession):
    print("Linux service session")

await session.run()

def create_follower_session_in_batch(
    self, 
    task: str, 
    plan: str
) -> List[BaseSession]

# Folder structure:
# /plans/
#   ├── email_john.json
#   ├── email_jane.json
#   └── email_bob.json

sessions = factory.create_follower_session_in_batch(
    task="send_emails",
    plan="/plans/"
)

# Returns:
# [
#   FollowerSession(task="send_emails/email_john", plan="/plans/email_john.json", id=0),
#   FollowerSession(task="send_emails/email_jane", plan="/plans/email_jane.json", id=1),
#   FollowerSession(task="send_emails/email_bob", plan="/plans/email_bob.json", id=2)
# ]

def create_sessions_in_batch(
    self, 
    task: str, 
    plan: str
) -> List[BaseSession]

{
  "email_john": true,
  "email_jane": false,
  "email_bob": false
}

# First run
sessions = factory.create_sessions_in_batch(
    task="batch_emails",
    plan="/requests/"
)
# Returns 3 sessions: email_john, email_jane, email_bob

# email_john completes successfully
# tasks_status.json updated: {"email_john": true, "email_jane": false, "email_bob": false}

# Second run (after restart)
sessions = factory.create_sessions_in_batch(
    task="batch_emails",
    plan="/requests/"
)
# Returns 2 sessions: email_jane, email_bob (skips completed email_john)

# Enable task status tracking
ufo_config.system.task_status = True

# Custom status file location
ufo_config.system.task_status_file = "/path/to/status.json"

from ufo.module.session_pool import SessionPool

# Create sessions
sessions = factory.create_session(
    task="batch_task",
    mode="follower",
    plan="/plans_folder/"
)

# Create pool
pool = SessionPool(session_list=sessions)

# Execute all
await pool.run_all()

def __init__(self, session_list: List[BaseSession]) -> None

async def run_all(self) -> None

pool = SessionPool(sessions)

# Execute all sequentially
await pool.run_all()

# All sessions have completed
print("Batch execution complete")

def add_session(self, session: BaseSession) -> None

pool = SessionPool([session1, session2])

# Add another session
pool.add_session(session3)

# Now pool has 3 sessions

def next_session(self) -> BaseSession

pool = SessionPool([session1, session2, session3])

# Get next session (FIFO)
next_sess = pool.next_session()
# next_sess == session1
# Pool now has [session2, session3]

await next_sess.run()

@property
def session_list(self) -> List[BaseSession]

pool = SessionPool(sessions)

print(f"Pool has {len(pool.session_list)} sessions")

for session in pool.session_list:
    print(f"Task: {session.task}")

factory = SessionFactory()

sessions = factory.create_session(
    task="user_task",
    mode="normal",
    plan="",
    request="Open Word and create a document"
)

session = sessions[0]
await session.run()

from aip.protocol.task_execution import TaskExecutionProtocol

# WebSocket connection established
protocol = TaskExecutionProtocol(websocket)

factory = SessionFactory()

session = factory.create_service_session(
    task="remote_automation",
    should_evaluate=True,
    id="session_123",
    request="Click the Submit button",
    task_protocol=protocol
)

await session.run()

# Create batch sessions
factory = SessionFactory()

sessions = factory.create_session(
    task="daily_reports",
    mode="batch_normal",
    plan="/request_files/",  # Folder with .json request files
    request=""
)

# Execute with pool
pool = SessionPool(sessions)
await pool.run_all()

print(f"Completed {len(sessions)} tasks")

import platform

factory = SessionFactory()

# Detect current platform
current_os = platform.system().lower()

sessions = factory.create_session(
    task="cross_platform_task",
    mode="normal",
    plan="",
    request="Open text editor",
    application_name="gedit" if current_os == "linux" else None
)

# Correct session type automatically created
await sessions[0].run()

pool = SessionPool([])

# Add sessions dynamically
for user_request in user_requests:
    sessions = factory.create_session(
        task=f"request_{len(pool.session_list)}",
        mode="normal",
        plan="",
        request=user_request
    )
    pool.add_session(sessions[0])

# Execute all
await pool.run_all()

# Enable task status tracking
ufo_config.system.task_status = True
ufo_config.system.task_status_file = "progress.json"

factory = SessionFactory()

# First run
sessions = factory.create_sessions_in_batch(
    task="large_batch",
    plan="/tasks/"
)

pool = SessionPool(sessions)

try:
    await pool.run_all()
except KeyboardInterrupt:
    print("Interrupted - progress saved")

# Second run (resumes from last incomplete)
sessions = factory.create_sessions_in_batch(
    task="large_batch",
    plan="/tasks/"
)
# Only uncompleted tasks loaded

pool = SessionPool(sessions)
await pool.run_all()

# config/ufo/config.yaml
system:
  eva_session: true
  task_status: true
  task_status_file: "./logs/task_status.json"

from config.config_loader import get_ufo_config

ufo_config = get_ufo_config()

# These settings affect SessionFactory behavior
factory = SessionFactory()

# Uses ufo_config.system.eva_session for should_evaluate
sessions = factory.create_session(
    task="configured_task",
    mode="normal",
    plan="",
    request="Do something"
)

import platform

current_platform = platform.system().lower()
# Returns: "windows", "linux", "darwin" (macOS)

# Force Windows session on Linux machine (for testing)
sessions = factory.create_session(
    task="test",
    mode="normal",
    plan="",
    request="Test request",
    platform_override="windows"
)

# Creates Session instead of LinuxSession

try:
    sessions = factory.create_session(
        task="task",
        mode="follower",
        plan="",
        request="",
        platform_override="darwin"  # macOS not supported
    )
except NotImplementedError as e:
    print(f"Error: {e}")
    # Error: Platform darwin is not supported yet.

try:
    sessions = factory.create_session(
        task="task",
        mode="follower",
        plan="",
        request="",
        platform_override="linux"
    )
except ValueError as e:
    print(f"Error: {e}")
    # Error: The follower mode is not supported on Linux yet.
    #        Supported modes: normal, normal_operator, service

def create_session_safely(task, mode, plan, request):
    """Create session with error handling."""
    factory = SessionFactory()

    try:
        sessions = factory.create_session(
            task=task,
            mode=mode,
            plan=plan,
            request=request
        )
        return sessions

    except NotImplementedError as e:
        logger.error(f"Platform not supported: {e}")
        return []

    except ValueError as e:
        logger.error(f"Invalid mode: {e}")
        # Fallback to normal mode
        return factory.create_session(
            task=task,
            mode="normal",
            plan="",
            request=request
        )

session = sessions[0]
print(f"Session type: {type(session).__name__}")
print(f"Platform: {platform.system().lower()}")

plan_files = factory.get_plan_files("/path/to/folder")
print(f"Found {len(plan_files)} plan files")
print(f"Files: {plan_files}")

protocol = kwargs.get("task_protocol")
print(f"Protocol: {protocol}")
print(f"Type: {type(protocol)}")

from aip.protocol.task_execution import TaskExecutionProtocol

protocol = TaskExecutionProtocol(websocket)

session = factory.create_service_session(
    task="service_task",
    should_evaluate=True,
    id="sess_001",
    request="Do something",
    task_protocol=protocol  # ← Required!
)