Skip to content

task_runtime

Task runtime model for mutable per-task execution state.

This package contains runtime envelopes and lifecycle state for tasks derived from immutable task specs.

TaskArtifacts dataclass

Outputs and observations produced by a task's agent.

These are accumulated as the agent works, representing the external artifacts and notable observations from task execution.

Attributes:

Name Type Description
pr_url Optional[str]

URL of the pull request created by the agent, or None if no PR has been created yet.
concerns list[str]

List of design concerns or observations noted by the agent during execution. Defaults to empty list.
agent_address Optional[AgentAddress]

Address of the agent that executed this task (e.g. tmux session + pane), or None if no agent has been launched yet. Stored as an immutable audit trail after agent teardown.
Source code in src/agentrelay/task_runtime/runtime.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@dataclass
class TaskArtifacts:
    """Outputs and observations produced by a task's agent.

    These are accumulated as the agent works, representing the external
    artifacts and notable observations from task execution.

    Attributes:
        pr_url: URL of the pull request created by the agent,
            or None if no PR has been created yet.
        concerns: List of design concerns or observations noted by the agent
            during execution. Defaults to empty list.
        agent_address: Address of the agent that executed this task (e.g. tmux
            session + pane), or None if no agent has been launched yet. Stored
            as an immutable audit trail after agent teardown.
    """

    pr_url: Optional[str] = None
    concerns: list[str] = field(default_factory=list)
    agent_address: Optional[AgentAddress] = None

TaskRuntime dataclass

Mutable runtime envelope for a Task.

Groups a frozen Task specification with all mutable state accumulated during execution. The Task spec itself never changes; TaskRuntime tracks how that spec is being executed.

Attributes:

Name Type Description
task Task

The immutable Task specification being executed.
state TaskState

Operational state (status, worktree, branch, error, attempts). Defaults to a new TaskState (PENDING, no paths, no error, attempt 0).
artifacts TaskArtifacts

Work outputs and observations (PR URL, concerns, agent address). Defaults to a new TaskArtifacts (no PR, no concerns, no agent address).
Source code in src/agentrelay/task_runtime/runtime.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@dataclass
class TaskRuntime:
    """Mutable runtime envelope for a Task.

    Groups a frozen Task specification with all mutable state accumulated
    during execution. The Task spec itself never changes; TaskRuntime tracks
    how that spec is being executed.

    Attributes:
        task: The immutable Task specification being executed.
        state: Operational state (status, worktree, branch, error, attempts).
            Defaults to a new TaskState (PENDING, no paths, no error, attempt 0).
        artifacts: Work outputs and observations (PR URL, concerns, agent address).
            Defaults to a new TaskArtifacts (no PR, no concerns, no agent address).
    """

    task: Task
    state: TaskState = field(default_factory=TaskState)
    artifacts: TaskArtifacts = field(default_factory=TaskArtifacts)

    def prepare_for_attempt(self, attempt_num: int) -> None:
        """Reset error and set attempt number before a task attempt."""
        self.state.attempt_num = attempt_num
        self.state.error = None

    def mark_failed(self, error: str) -> None:
        """Transition to FAILED with an error message."""
        self.state.status = TaskStatus.FAILED
        self.state.error = error

    def reset_for_retry(self) -> None:
        """Archive current error to concerns and reset to PENDING for retry."""
        if self.state.error:
            self.artifacts.concerns.append(
                f"attempt_{self.state.attempt_num}_error: {self.state.error}"
            )
        self.state.status = TaskStatus.PENDING
        self.state.error = None

    def mark_pending(self) -> None:
        """Set status to PENDING (used for retry normalization)."""
        self.state.status = TaskStatus.PENDING

prepare_for_attempt(attempt_num)

Reset error and set attempt number before a task attempt.

Source code in src/agentrelay/task_runtime/runtime.py 
125
126
127
128
def prepare_for_attempt(self, attempt_num: int) -> None:
    """Reset error and set attempt number before a task attempt."""
    self.state.attempt_num = attempt_num
    self.state.error = None

mark_failed(error)

Transition to FAILED with an error message.

Source code in src/agentrelay/task_runtime/runtime.py 
130
131
132
133
def mark_failed(self, error: str) -> None:
    """Transition to FAILED with an error message."""
    self.state.status = TaskStatus.FAILED
    self.state.error = error

reset_for_retry()

Archive current error to concerns and reset to PENDING for retry.

Source code in src/agentrelay/task_runtime/runtime.py 
135
136
137
138
139
140
141
142
def reset_for_retry(self) -> None:
    """Archive current error to concerns and reset to PENDING for retry."""
    if self.state.error:
        self.artifacts.concerns.append(
            f"attempt_{self.state.attempt_num}_error: {self.state.error}"
        )
    self.state.status = TaskStatus.PENDING
    self.state.error = None

mark_pending()

Set status to PENDING (used for retry normalization).

Source code in src/agentrelay/task_runtime/runtime.py 
144
145
146
def mark_pending(self) -> None:
    """Set status to PENDING (used for retry normalization)."""
    self.state.status = TaskStatus.PENDING

TaskState dataclass

Mutable operational state of a running task.

This tracks the execution progress and infrastructure of a task as managed by the orchestrator.

Attributes:

Name Type Description
status TaskStatus

Current execution state (TaskStatus enum).
worktree_path Optional[Path]

Filesystem path to the git worktree where the agent works, or None if not yet created.
branch_name Optional[str]

Name of the feature branch in the worktree, or None if not yet created.
error Optional[str]

Error message if the task failed, or None if no error.
attempt_num int

The current attempt number (0-indexed). Used to track retries and conditional logic like when to start self-review.
integration_branch Optional[str]

Name of the workstream integration branch this task targets. Set by the orchestrator before dispatch from the workstream runtime. None until set.
workstream_worktree_path Optional[Path]

Filesystem path to the shared workstream worktree. Set by the orchestrator before dispatch from the workstream runtime. None until set.
Source code in src/agentrelay/task_runtime/runtime.py 
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
@dataclass
class TaskState:
    """Mutable operational state of a running task.

    This tracks the execution progress and infrastructure of a task
    as managed by the orchestrator.

    Attributes:
        status: Current execution state (TaskStatus enum).
        worktree_path: Filesystem path to the git worktree where the agent works,
            or None if not yet created.
        branch_name: Name of the feature branch in the worktree,
            or None if not yet created.
        error: Error message if the task failed, or None if no error.
        attempt_num: The current attempt number (0-indexed). Used to track
            retries and conditional logic like when to start self-review.
        integration_branch: Name of the workstream integration branch this task
            targets. Set by the orchestrator before dispatch from the workstream
            runtime. None until set.
        workstream_worktree_path: Filesystem path to the shared workstream worktree.
            Set by the orchestrator before dispatch from the workstream runtime.
            None until set.
    """

    status: TaskStatus = TaskStatus.PENDING
    worktree_path: Optional[Path] = None
    branch_name: Optional[str] = None
    signal_dir: Optional[Path] = None
    error: Optional[str] = None
    attempt_num: int = 0
    integration_branch: Optional[str] = None
    workstream_worktree_path: Optional[Path] = None

TaskStatus

Bases: str, Enum

Execution state of a task during orchestration.

Attributes:

Name Type Description
PENDING

Task is waiting to be executed.
RUNNING

Task is currently being executed by an agent.
PR_CREATED

Agent completed work; pull request exists against worktree branch.
PR_MERGED

Pull request has been merged into the worktree primary branch.
FAILED

Task execution failed.
Source code in src/agentrelay/task_runtime/runtime.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class TaskStatus(str, Enum):
    """Execution state of a task during orchestration.

    Attributes:
        PENDING: Task is waiting to be executed.
        RUNNING: Task is currently being executed by an agent.
        PR_CREATED: Agent completed work; pull request exists against worktree branch.
        PR_MERGED: Pull request has been merged into the worktree primary branch.
        FAILED: Task execution failed.
    """

    PENDING = "pending"
    RUNNING = "running"
    PR_CREATED = "pr_created"  # Agent done; PR exists against worktree branch
    PR_MERGED = "pr_merged"  # PR merged into worktree primary branch
    FAILED = "failed"