Skip to content

agent

View module diagram

Agent abstractions for live coding-agent execution.

This package defines how orchestrator/runtime code interacts with running agents: addressing types for locating an agent process and agent interfaces for sending kickoff instructions in a concrete execution environment.

Subpackages

core: Abstract interfaces (Agent ABC, AgentAddress ABC). implementations: Concrete implementations (TmuxAgent, TmuxAddress).

AgentAddress

Bases: ABC

Abstract base for addressing a running agent instance.

Defines the interface for locating a live agent regardless of execution environment. Concrete subclasses carry the environment-specific connection details needed to communicate with the agent (pane IDs, endpoints, etc.).

Attributes:

Name Type Description
label str

A human-readable string identifier for the agent's location.
Source code in src/agentrelay/agent/core/addressing.py 
17
18
19
20
21
22
23
24
25
26
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
64
65
66
67
68
69
class AgentAddress(ABC):
    """Abstract base for addressing a running agent instance.

    Defines the interface for locating a live agent regardless of execution
    environment. Concrete subclasses carry the environment-specific connection
    details needed to communicate with the agent (pane IDs, endpoints, etc.).

    Attributes:
        label: A human-readable string identifier for the agent's location.
    """

    @property
    @abstractmethod
    def label(self) -> str:
        """Return a human-readable identifier for this agent's location.

        Returns:
            String representation of the agent's address/location.
        """
        ...

    def capture_log(self, signal_dir: Optional[Path] = None) -> None:
        """Capture agent execution log to the signal directory.

        Called unconditionally when a task reaches a terminal status, regardless
        of teardown mode. The default implementation is a no-op; subclasses
        override for environment-specific log capture (e.g. tmux scrollback).

        Args:
            signal_dir: Directory for writing the captured log. None if no
                signal directory is available.
        """

    def teardown(
        self,
        signal_dir: Optional[Path] = None,
        keep_panes: bool = False,
    ) -> None:
        """Release environment-specific resources for this agent address.

        Called during task teardown to clean up execution resources. The default
        implementation is a no-op; subclasses override for environment-specific
        cleanup (e.g. killing tmux windows, deleting branches).

        If ``capture_log()`` has already written ``agent.log``, implementations
        should skip redundant capture (idempotent).

        Args:
            signal_dir: Directory for writing teardown artifacts (e.g. agent logs).
                None if no signal directory is available.
            keep_panes: If True, preserve the agent's execution environment
                (e.g. tmux window) for debugging rather than destroying it.
        """

label abstractmethod property

Return a human-readable identifier for this agent's location.

Returns:

Type Description
str

String representation of the agent's address/location.

capture_log(signal_dir=None)

Capture agent execution log to the signal directory.

Called unconditionally when a task reaches a terminal status, regardless of teardown mode. The default implementation is a no-op; subclasses override for environment-specific log capture (e.g. tmux scrollback).

Parameters:

Name Type Description Default
signal_dir Optional[Path]

Directory for writing the captured log. None if no signal directory is available.

 None
Source code in src/agentrelay/agent/core/addressing.py 
38
39
40
41
42
43
44
45
46
47
48
def capture_log(self, signal_dir: Optional[Path] = None) -> None:
    """Capture agent execution log to the signal directory.

    Called unconditionally when a task reaches a terminal status, regardless
    of teardown mode. The default implementation is a no-op; subclasses
    override for environment-specific log capture (e.g. tmux scrollback).

    Args:
        signal_dir: Directory for writing the captured log. None if no
            signal directory is available.
    """

teardown(signal_dir=None, keep_panes=False)

Release environment-specific resources for this agent address.

Called during task teardown to clean up execution resources. The default implementation is a no-op; subclasses override for environment-specific cleanup (e.g. killing tmux windows, deleting branches).

If capture_log() has already written agent.log, implementations should skip redundant capture (idempotent).

Parameters:

Name Type Description Default
signal_dir Optional[Path]

Directory for writing teardown artifacts (e.g. agent logs). None if no signal directory is available.

 None
keep_panes bool

If True, preserve the agent's execution environment (e.g. tmux window) for debugging rather than destroying it.

 False
Source code in src/agentrelay/agent/core/addressing.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def teardown(
    self,
    signal_dir: Optional[Path] = None,
    keep_panes: bool = False,
) -> None:
    """Release environment-specific resources for this agent address.

    Called during task teardown to clean up execution resources. The default
    implementation is a no-op; subclasses override for environment-specific
    cleanup (e.g. killing tmux windows, deleting branches).

    If ``capture_log()`` has already written ``agent.log``, implementations
    should skip redundant capture (idempotent).

    Args:
        signal_dir: Directory for writing teardown artifacts (e.g. agent logs).
            None if no signal directory is available.
        keep_panes: If True, preserve the agent's execution environment
            (e.g. tmux window) for debugging rather than destroying it.
    """

Agent

Bases: ABC

Abstract base for a live running agent instance.

An Agent represents a coding assistant that has been spawned in a specific execution environment and can be instructed to begin work. Different execution environments (tmux, cloud, etc.) are represented by concrete subclasses.

The Agent does not store configuration or role information — these are accessible via runtime.task.primary_agent and runtime.task.role, ensuring a single source of truth.

Source code in src/agentrelay/agent/core/agent.py 
18
19
20
21
22
23
24
25
26
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
class Agent(ABC):
    """Abstract base for a live running agent instance.

    An Agent represents a coding assistant that has been spawned in a specific
    execution environment and can be instructed to begin work. Different execution
    environments (tmux, cloud, etc.) are represented by concrete subclasses.

    The Agent does not store configuration or role information — these are
    accessible via runtime.task.primary_agent and runtime.task.role, ensuring
    a single source of truth.
    """

    @abstractmethod
    def send_kickoff(self, instructions_path: str) -> None:
        """Send kickoff instructions to the running agent.

        Activates the agent to begin autonomous work.

        Args:
            instructions_path: Path to the instructions file the agent should read.
                Typically the signal_dir/instructions.md file written by the orchestrator.
        """
        ...

    @property
    @abstractmethod
    def address(self) -> AgentAddress:
        """Return the address of this running agent.

        Returns:
            An AgentAddress indicating where the agent is running (type depends
            on the concrete Agent subclass).
        """
        ...

address abstractmethod property

Return the address of this running agent.

Returns:

Type Description
AgentAddress

An AgentAddress indicating where the agent is running (type depends
AgentAddress

on the concrete Agent subclass).

send_kickoff(instructions_path) abstractmethod

Send kickoff instructions to the running agent.

Activates the agent to begin autonomous work.

Parameters:

Name Type Description Default
instructions_path str

Path to the instructions file the agent should read. Typically the signal_dir/instructions.md file written by the orchestrator.

 required
Source code in src/agentrelay/agent/core/agent.py 
30
31
32
33
34
35
36
37
38
39
40
@abstractmethod
def send_kickoff(self, instructions_path: str) -> None:
    """Send kickoff instructions to the running agent.

    Activates the agent to begin autonomous work.

    Args:
        instructions_path: Path to the instructions file the agent should read.
            Typically the signal_dir/instructions.md file written by the orchestrator.
    """
    ...

TmuxAddress dataclass

Bases: AgentAddress

Address of an agent running in a tmux pane.

Attributes:

Name Type Description
session str

The name of the tmux session.
pane_id str

The identifier of the tmux pane (e.g., "%1", "%2").
Source code in src/agentrelay/agent/implementations/tmux_address.py 
18
19
20
21
22
23
24
25
26
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
@dataclass(frozen=True)
class TmuxAddress(AgentAddress):
    """Address of an agent running in a tmux pane.

    Attributes:
        session: The name of the tmux session.
        pane_id: The identifier of the tmux pane (e.g., "%1", "%2").
    """

    session: str
    pane_id: str

    @property
    def label(self) -> str:
        """Return a human-readable identifier combining session and pane.

        Returns:
            String in format "session:pane_id" (e.g., "agentrelay:%1").
        """
        return f"{self.session}:{self.pane_id}"

    def capture_log(self, signal_dir: Optional[Path] = None) -> None:
        """Capture pane scrollback and write ``agent.log``.

        Best-effort: errors are swallowed since the pane may already be gone.

        Args:
            signal_dir: If provided, pane scrollback is written as ``agent.log``.
        """
        try:
            log = tmux.capture_pane(self.pane_id, full_history=True)
            if signal_dir is not None:
                signals.write_text(signal_dir, "agent.log", log)
        except subprocess.CalledProcessError:
            pass  # Best-effort: pane may already be gone

    def teardown(
        self,
        signal_dir: Optional[Path] = None,
        keep_panes: bool = False,
    ) -> None:
        """Optionally capture scrollback (if not already done) and kill the tmux window.

        Scrollback capture is skipped if ``agent.log`` already exists in the
        signal directory (idempotent with prior ``capture_log()`` call).

        Best-effort: errors are swallowed since the pane may already be gone.

        Args:
            signal_dir: If provided and ``agent.log`` is missing, pane
                scrollback is captured before cleanup.
            keep_panes: If True, keep the tmux window open for debugging.
        """
        if signal_dir is not None and not (signal_dir / "agent.log").exists():
            self.capture_log(signal_dir)

        if not keep_panes:
            try:
                tmux.kill_window(self.pane_id)
            except subprocess.CalledProcessError:
                pass  # Best-effort

label property

Return a human-readable identifier combining session and pane.

Returns:

Type Description
str

String in format "session:pane_id" (e.g., "agentrelay:%1").

capture_log(signal_dir=None)

Capture pane scrollback and write agent.log.

Best-effort: errors are swallowed since the pane may already be gone.

Parameters:

Name Type Description Default
signal_dir Optional[Path]

If provided, pane scrollback is written as agent.log.

 None
Source code in src/agentrelay/agent/implementations/tmux_address.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def capture_log(self, signal_dir: Optional[Path] = None) -> None:
    """Capture pane scrollback and write ``agent.log``.

    Best-effort: errors are swallowed since the pane may already be gone.

    Args:
        signal_dir: If provided, pane scrollback is written as ``agent.log``.
    """
    try:
        log = tmux.capture_pane(self.pane_id, full_history=True)
        if signal_dir is not None:
            signals.write_text(signal_dir, "agent.log", log)
    except subprocess.CalledProcessError:
        pass  # Best-effort: pane may already be gone

teardown(signal_dir=None, keep_panes=False)

Optionally capture scrollback (if not already done) and kill the tmux window.

Scrollback capture is skipped if agent.log already exists in the signal directory (idempotent with prior capture_log() call).

Best-effort: errors are swallowed since the pane may already be gone.

Parameters:

Name Type Description Default
signal_dir Optional[Path]

If provided and agent.log is missing, pane scrollback is captured before cleanup.

 None
keep_panes bool

If True, keep the tmux window open for debugging.

 False
Source code in src/agentrelay/agent/implementations/tmux_address.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def teardown(
    self,
    signal_dir: Optional[Path] = None,
    keep_panes: bool = False,
) -> None:
    """Optionally capture scrollback (if not already done) and kill the tmux window.

    Scrollback capture is skipped if ``agent.log`` already exists in the
    signal directory (idempotent with prior ``capture_log()`` call).

    Best-effort: errors are swallowed since the pane may already be gone.

    Args:
        signal_dir: If provided and ``agent.log`` is missing, pane
            scrollback is captured before cleanup.
        keep_panes: If True, keep the tmux window open for debugging.
    """
    if signal_dir is not None and not (signal_dir / "agent.log").exists():
        self.capture_log(signal_dir)

    if not keep_panes:
        try:
            tmux.kill_window(self.pane_id)
        except subprocess.CalledProcessError:
            pass  # Best-effort

TmuxAgent dataclass

Bases: Agent

A live agent running in a tmux pane.

Represents a Claude Code instance launched in a tmux pane. The agent holds the TmuxAddress (session + pane_id) of its running process.

Attributes:

Name Type Description
_address TmuxAddress

The tmux address (session and pane_id) of the running agent.
Source code in src/agentrelay/agent/implementations/tmux_agent.py 
22
23
24
25
26
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
64
65
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
@dataclass
class TmuxAgent(Agent):
    """A live agent running in a tmux pane.

    Represents a Claude Code instance launched in a tmux pane. The agent holds
    the TmuxAddress (session + pane_id) of its running process.

    Attributes:
        _address: The tmux address (session and pane_id) of the running agent.
    """

    _address: TmuxAddress

    @property
    def address(self) -> TmuxAddress:
        """Return the tmux address of this agent."""
        return self._address

    @classmethod
    def from_config(
        cls,
        config: AgentConfig,
        task_id: str,
        worktree_path: Path,
        cmd: str,
    ) -> TmuxAgent:
        """Create a tmux pane, send a pre-built command, and return the TmuxAgent handle.

        Creates a new tmux window in the session specified by config.environment,
        sends the provided command string, and returns a TmuxAgent object that
        can be used to communicate with the running process.

        Args:
            config: AgentConfig specifying environment.
                config.environment must be TmuxEnvironment.
            task_id: Window name for the tmux window (typically ``{graph_name}-{task_id}``).
            worktree_path: Path to the git worktree where the agent will work.
            cmd: Pre-built command string to send to the tmux pane. Typically
                constructed by an :class:`AgentFrameworkAdapter` and wrapped
                by an :class:`AgentSandbox`.

        Returns:
            A TmuxAgent instance with its address set to the created tmux pane.
        """
        session = config.environment.session
        pane_id = tmux.new_window(session, task_id, worktree_path)
        tmux.send_keys(pane_id, cmd)

        return cls(_address=TmuxAddress(session=session, pane_id=pane_id))

    def send_kickoff(self, instructions_path: str) -> None:
        """Wait for the agent's TUI to be ready, then send the kickoff message.

        Sends an initial prompt to the agent pointing it to its task instructions.
        This activates the agent to begin autonomous work.

        The message sent is: "Read {instructions_path} and follow the steps exactly."

        Args:
            instructions_path: Path to the instructions file the agent should read.
                Typically the signal_dir/instructions.md file written by the orchestrator.
        """
        pane_id = self._address.pane_id
        tmux.wait_for_tui_ready(pane_id)
        tmux.send_keys(
            pane_id,
            f"Read {instructions_path} and follow the steps exactly.",
        )
        # Claude Code >= 2.1.105 interprets the first Enter as a newline
        # in the TUI input box. A second Enter after a brief delay submits.
        time.sleep(0.3)
        tmux.send_keys(pane_id, "", press_enter=True)

address property

Return the tmux address of this agent.

from_config(config, task_id, worktree_path, cmd) classmethod

Create a tmux pane, send a pre-built command, and return the TmuxAgent handle.

Creates a new tmux window in the session specified by config.environment, sends the provided command string, and returns a TmuxAgent object that can be used to communicate with the running process.

Parameters:

Name Type Description Default
config AgentConfig

AgentConfig specifying environment. config.environment must be TmuxEnvironment.

 required
task_id str

Window name for the tmux window (typically {graph_name}-{task_id}).

 required
worktree_path Path

Path to the git worktree where the agent will work.

 required
cmd str

Pre-built command string to send to the tmux pane. Typically constructed by an :class:AgentFrameworkAdapter and wrapped by an :class:AgentSandbox.

 required

Returns:

Type Description
TmuxAgent

A TmuxAgent instance with its address set to the created tmux pane.
Source code in src/agentrelay/agent/implementations/tmux_agent.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@classmethod
def from_config(
    cls,
    config: AgentConfig,
    task_id: str,
    worktree_path: Path,
    cmd: str,
) -> TmuxAgent:
    """Create a tmux pane, send a pre-built command, and return the TmuxAgent handle.

    Creates a new tmux window in the session specified by config.environment,
    sends the provided command string, and returns a TmuxAgent object that
    can be used to communicate with the running process.

    Args:
        config: AgentConfig specifying environment.
            config.environment must be TmuxEnvironment.
        task_id: Window name for the tmux window (typically ``{graph_name}-{task_id}``).
        worktree_path: Path to the git worktree where the agent will work.
        cmd: Pre-built command string to send to the tmux pane. Typically
            constructed by an :class:`AgentFrameworkAdapter` and wrapped
            by an :class:`AgentSandbox`.

    Returns:
        A TmuxAgent instance with its address set to the created tmux pane.
    """
    session = config.environment.session
    pane_id = tmux.new_window(session, task_id, worktree_path)
    tmux.send_keys(pane_id, cmd)

    return cls(_address=TmuxAddress(session=session, pane_id=pane_id))

send_kickoff(instructions_path)

Wait for the agent's TUI to be ready, then send the kickoff message.

Sends an initial prompt to the agent pointing it to its task instructions. This activates the agent to begin autonomous work.

The message sent is: "Read {instructions_path} and follow the steps exactly."

Parameters:

Name Type Description Default
instructions_path str

Path to the instructions file the agent should read. Typically the signal_dir/instructions.md file written by the orchestrator.

 required
Source code in src/agentrelay/agent/implementations/tmux_agent.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def send_kickoff(self, instructions_path: str) -> None:
    """Wait for the agent's TUI to be ready, then send the kickoff message.

    Sends an initial prompt to the agent pointing it to its task instructions.
    This activates the agent to begin autonomous work.

    The message sent is: "Read {instructions_path} and follow the steps exactly."

    Args:
        instructions_path: Path to the instructions file the agent should read.
            Typically the signal_dir/instructions.md file written by the orchestrator.
    """
    pane_id = self._address.pane_id
    tmux.wait_for_tui_ready(pane_id)
    tmux.send_keys(
        pane_id,
        f"Read {instructions_path} and follow the steps exactly.",
    )
    # Claude Code >= 2.1.105 interprets the first Enter as a newline
    # in the TUI input box. A second Enter after a brief delay submits.
    time.sleep(0.3)
    tmux.send_keys(pane_id, "", press_enter=True)