Skip to content

session

View module diagram

Session resolution and validation for agent execution environments.

This module defines the SessionResolver protocol and its TmuxSessionResolver implementation for resolving and validating tmux sessions used by agent tasks.

Protocols

SessionResolver: Resolve and validate agent execution sessions.

Classes:

Name Description
TmuxSessionResolver

Tmux-based session resolution.
SessionError

Raised when session resolution or validation fails.

SessionError

Bases: RuntimeError

Raised when the tmux session is not specified or doesn't exist.

Source code in src/agentrelay/session.py 
23
24
class SessionError(RuntimeError):
    """Raised when the tmux session is not specified or doesn't exist."""

SessionResolver

Bases: Protocol

Protocol for resolving and validating agent execution sessions.

Implementations handle auto-detection and validation of the execution environment (e.g., tmux sessions) where agent tasks will run.

Methods:

Name Description
resolve

Resolve CLI session override to a concrete session name.
validate

Validate that all tasks' sessions exist.
Source code in src/agentrelay/session.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@runtime_checkable
class SessionResolver(Protocol):
    """Protocol for resolving and validating agent execution sessions.

    Implementations handle auto-detection and validation of the execution
    environment (e.g., tmux sessions) where agent tasks will run.

    Methods:
        resolve: Resolve CLI session override to a concrete session name.
        validate: Validate that all tasks' sessions exist.
    """

    def resolve(self, cli_session: Optional[str]) -> str:
        """Resolve a session name from CLI override or auto-detection.

        Args:
            cli_session: Explicit session name from CLI, or ``None``
                for auto-detection.

        Returns:
            Resolved session name.

        Raises:
            SessionError: If no session can be resolved.
        """
        ...

    def validate(self, graph: TaskGraph) -> None:
        """Validate that all tasks reference existing sessions.

        Args:
            graph: Validated task graph whose tasks' sessions to check.

        Raises:
            SessionError: If any session is missing or does not exist.
        """
        ...

resolve(cli_session)

Resolve a session name from CLI override or auto-detection.

Parameters:

Name Type Description Default
cli_session Optional[str]

Explicit session name from CLI, or None for auto-detection.

 required

Returns:

Type Description
str

Resolved session name.

Raises:

Type Description
SessionError

If no session can be resolved.
Source code in src/agentrelay/session.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def resolve(self, cli_session: Optional[str]) -> str:
    """Resolve a session name from CLI override or auto-detection.

    Args:
        cli_session: Explicit session name from CLI, or ``None``
            for auto-detection.

    Returns:
        Resolved session name.

    Raises:
        SessionError: If no session can be resolved.
    """
    ...

validate(graph)

Validate that all tasks reference existing sessions.

Parameters:

Name Type Description Default
graph TaskGraph

Validated task graph whose tasks' sessions to check.

 required

Raises:

Type Description
SessionError

If any session is missing or does not exist.
Source code in src/agentrelay/session.py 
54
55
56
57
58
59
60
61
62
63
def validate(self, graph: TaskGraph) -> None:
    """Validate that all tasks reference existing sessions.

    Args:
        graph: Validated task graph whose tasks' sessions to check.

    Raises:
        SessionError: If any session is missing or does not exist.
    """
    ...

TmuxSessionResolver

Tmux-based session resolver.

Resolves the session by checking the CLI override, then auto-detecting from the current tmux session. Validates by checking tmux has-session for each unique session in the graph.

Source code in src/agentrelay/session.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class TmuxSessionResolver:
    """Tmux-based session resolver.

    Resolves the session by checking the CLI override, then
    auto-detecting from the current tmux session.  Validates by
    checking ``tmux has-session`` for each unique session in the graph.
    """

    def resolve(self, cli_session: Optional[str]) -> str:
        """Resolve tmux session: CLI flag > auto-detect > error.

        Args:
            cli_session: Explicit session name, or ``None``.

        Returns:
            Resolved tmux session name.

        Raises:
            SessionError: If not in tmux and no CLI override given.
        """
        if cli_session is not None:
            return cli_session
        detected = tmux.current_session()
        if detected is None:
            raise SessionError(
                "No tmux session specified and not running inside tmux.\n"
                "Either run from inside a tmux session or use --tmux-session."
            )
        return detected

    def validate(self, graph: TaskGraph) -> None:
        """Validate that all tasks' tmux sessions exist.

        Checks two things for every task in the graph:

        1. The task has a non-empty session name.
        2. The tmux session actually exists (``tmux has-session``).

        Args:
            graph: Task graph to validate.

        Raises:
            SessionError: If any task's session is empty or the tmux
                session does not exist.
        """
        sessions_seen: set[str] = set()
        for task_id in graph.task_ids():
            task = graph.task(task_id)
            session = task.primary_agent.environment.session
            if not session:
                raise SessionError(
                    f"Task '{task_id}' has no tmux session specified.\n"
                    "Use --tmux-session on the CLI or run from inside a tmux session."
                )
            sessions_seen.add(session)

        for session in sorted(sessions_seen):
            if not tmux.has_session(session):
                raise SessionError(
                    f"Tmux session '{session}' does not exist.\n"
                    f"Create it first: tmux new-session -d -s {session}"
                )

resolve(cli_session)

Resolve tmux session: CLI flag > auto-detect > error.

Parameters:

Name Type Description Default
cli_session Optional[str]

Explicit session name, or None.

 required

Returns:

Type Description
str

Resolved tmux session name.

Raises:

Type Description
SessionError

If not in tmux and no CLI override given.
Source code in src/agentrelay/session.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def resolve(self, cli_session: Optional[str]) -> str:
    """Resolve tmux session: CLI flag > auto-detect > error.

    Args:
        cli_session: Explicit session name, or ``None``.

    Returns:
        Resolved tmux session name.

    Raises:
        SessionError: If not in tmux and no CLI override given.
    """
    if cli_session is not None:
        return cli_session
    detected = tmux.current_session()
    if detected is None:
        raise SessionError(
            "No tmux session specified and not running inside tmux.\n"
            "Either run from inside a tmux session or use --tmux-session."
        )
    return detected

validate(graph)

Validate that all tasks' tmux sessions exist.

Checks two things for every task in the graph:

  1. The task has a non-empty session name.
  2. The tmux session actually exists (tmux has-session).

Parameters:

Name Type Description Default
graph TaskGraph

Task graph to validate.

 required

Raises:

Type Description
SessionError

If any task's session is empty or the tmux session does not exist.
Source code in src/agentrelay/session.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def validate(self, graph: TaskGraph) -> None:
    """Validate that all tasks' tmux sessions exist.

    Checks two things for every task in the graph:

    1. The task has a non-empty session name.
    2. The tmux session actually exists (``tmux has-session``).

    Args:
        graph: Task graph to validate.

    Raises:
        SessionError: If any task's session is empty or the tmux
            session does not exist.
    """
    sessions_seen: set[str] = set()
    for task_id in graph.task_ids():
        task = graph.task(task_id)
        session = task.primary_agent.environment.session
        if not session:
            raise SessionError(
                f"Task '{task_id}' has no tmux session specified.\n"
                "Use --tmux-session on the CLI or run from inside a tmux session."
            )
        sessions_seen.add(session)

    for session in sorted(sessions_seen):
        if not tmux.has_session(session):
            raise SessionError(
                f"Tmux session '{session}' does not exist.\n"
                f"Create it first: tmux new-session -d -s {session}"
            )