RunService

No overview available.

Methods

---

RecordAction()

@classmethod
def RecordAction(
    ctx: context.Context,
    req: *connect.Request[workflow.RecordActionRequest]
) - > *connect.Response[workflow.RecordActionResponse], error

Records a single action, including its details and status, within a workflow run. This method is used to persist the state of an action.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.RecordActionRequest]	The request containing the action details to be recorded.

Returns

Type	Description
*connect.Response[workflow.RecordActionResponse], error	A response containing the result of the action recording operation, or an error if the operation failed.

---

RecordActionStream()

@classmethod
def RecordActionStream(
    ctx: context.Context,
    stream: *connect.BidiStream[workflow.RecordActionStreamRequest, workflow.RecordActionStreamResponse]
) - > error

Establishes a bidirectional stream to record multiple actions. This allows for efficient, continuous updates of action states.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
stream	*connect.BidiStream[workflow.RecordActionStreamRequest, workflow.RecordActionStreamResponse]	The bidirectional stream for sending action recording requests and receiving responses.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

recordAction()

@classmethod
def recordAction(
    ctx: context.Context,
    req: *workflow.RecordActionRequest
) - > *workflow.RecordActionResponse

Records a single action internally, processing the provided action request. This is a helper method for the public API.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
req	*workflow.RecordActionRequest	The request containing the action details to record.

Returns

Type	Description
*workflow.RecordActionResponse	The response indicating the outcome of the action recording.

---

recordSingleAction()

@classmethod
def recordSingleAction(
    ctx: context.Context,
    req: *workflow.RecordActionRequest
) - > error

Persists a single action's details to the data store. This method is an internal utility for recording individual actions.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
req	*workflow.RecordActionRequest	The request containing the action details to record.

Returns

Type	Description
error	An error if the action could not be recorded.

---

UpdateActionStatus()

@classmethod
def UpdateActionStatus(
    ctx: context.Context,
    req: *connect.Request[workflow.UpdateActionStatusRequest]
) - > *connect.Response[workflow.UpdateActionStatusResponse], error

Updates the status of an existing action within a workflow run. This allows tracking the progress and outcome of actions.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.UpdateActionStatusRequest]	The request containing the action identifier and the new status.

Returns

Type	Description
*connect.Response[workflow.UpdateActionStatusResponse], error	A response indicating the success or failure of the status update, or an error if the operation failed.

---

UpdateActionStatusStream()

@classmethod
def UpdateActionStatusStream(
    ctx: context.Context,
    stream: *connect.BidiStream[workflow.UpdateActionStatusStreamRequest, workflow.UpdateActionStatusStreamResponse]
) - > error

Establishes a bidirectional stream to update the status of multiple actions. This enables real-time status synchronization for actions.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
stream	*connect.BidiStream[workflow.UpdateActionStatusStreamRequest, workflow.UpdateActionStatusStreamResponse]	The bidirectional stream for sending status update requests and receiving responses.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

updateActionStatus()

@classmethod
def updateActionStatus(
    ctx: context.Context,
    req: *workflow.UpdateActionStatusRequest
) - > *workflow.UpdateActionStatusResponse

Updates the status of a single action internally, processing the provided status update request. This is a helper method for the public API.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
req	*workflow.UpdateActionStatusRequest	The request containing the action identifier and the new status.

Returns

Type	Description
*workflow.UpdateActionStatusResponse	The response indicating the outcome of the action status update.

---

updateSingleActionStatus()

@classmethod
def updateSingleActionStatus(
    ctx: context.Context,
    req: *workflow.UpdateActionStatusRequest
) - > error

Persists the status update for a single action to the data store. This method is an internal utility for updating individual action statuses.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
req	*workflow.UpdateActionStatusRequest	The request containing the action identifier and the new status.

Returns

Type	Description
error	An error if the action status could not be updated.

---

RecordActionEvents()

@classmethod
def RecordActionEvents(
    ctx: context.Context,
    req: *connect.Request[workflow.RecordActionEventsRequest]
) - > *connect.Response[workflow.RecordActionEventsResponse], error

Records a batch of events associated with actions. This allows for capturing detailed logs and state changes for actions.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.RecordActionEventsRequest]	The request containing a list of action events to be recorded.

Returns

Type	Description
*connect.Response[workflow.RecordActionEventsResponse], error	A response indicating the success or failure of recording the events, or an error if the operation failed.

---

RecordActionEventStream()

@classmethod
def RecordActionEventStream(
    ctx: context.Context,
    stream: *connect.BidiStream[workflow.RecordActionEventStreamRequest, workflow.RecordActionEventStreamResponse]
) - > error

Establishes a bidirectional stream to record multiple action events. This enables efficient, continuous logging of action-related occurrences.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
stream	*connect.BidiStream[workflow.RecordActionEventStreamRequest, workflow.RecordActionEventStreamResponse]	The bidirectional stream for sending action event recording requests and receiving responses.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

recordEvents()

@classmethod
def recordEvents(
    ctx: context.Context,
    events: []*workflow.ActionEvent
) - > error

Persists a list of action events to the data store. This method is an internal utility for recording event batches.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
events	[]*workflow.ActionEvent	A slice of action events to be recorded.

Returns

Type	Description
error	An error if the events could not be recorded.

---

WatchGroups()

@classmethod
def WatchGroups(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchGroupsRequest],
    stream: *connect.ServerStream[workflow.WatchGroupsResponse]
) - > error

Establishes a server-side stream to watch for changes in task groups. This allows clients to receive real-time updates on group status and composition.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchGroupsRequest]	The request specifying criteria for watching task groups.
stream	*connect.ServerStream[workflow.WatchGroupsResponse]	The server-side stream to send group updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

CreateRun()

@classmethod
def CreateRun(
    ctx: context.Context,
    req: *connect.Request[workflow.CreateRunRequest]
) - > *connect.Response[workflow.CreateRunResponse], error

Initiates a new workflow run based on the provided specifications. This method is the entry point for starting a workflow execution.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.CreateRunRequest]	The request containing the specifications for the new workflow run.

Returns

Type	Description
*connect.Response[workflow.CreateRunResponse], error	A response containing details of the newly created run, or an error if the run could not be created.

---

persistRunModel()

@classmethod
def persistRunModel(
    ctx: context.Context,
    runId: *common.RunIdentifier,
    taskID: *task.TaskIdentifier,
    taskSpec: *task.TaskSpec,
    inputPrefix: string,
    runOutputBase: string,
    runSpec: *task.RunSpec,
    source: workflow.RunSource,
    triggerName: string,
    triggerTaskName: string,
    triggerRevision: int64,
    triggerType: string
) - > *models.Run, error

Persists a new run model to the data store with all its associated metadata. This internal method is crucial for establishing a run's initial state.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
runId	*common.RunIdentifier	The unique identifier for the run.
taskID	*task.TaskIdentifier	The identifier for the task associated with the run.
taskSpec	*task.TaskSpec	The specification of the task being executed.
inputPrefix	string	The prefix for input data paths.
runOutputBase	string	The base path for run output data.
runSpec	*task.RunSpec	The specification for how the run should be executed.
source	workflow.RunSource	The source or origin of the run.
triggerName	string	The name of the trigger that initiated the run.
triggerTaskName	string	The name of the task associated with the trigger.
triggerRevision	int64	The revision number of the trigger.
triggerType	string	The type of the trigger (e.g., 'manual', 'schedule').

Returns

Type	Description
*models.Run, error	The created run model, or an error if persistence failed.

---

AbortRun()

@classmethod
def AbortRun(
    ctx: context.Context,
    req: *connect.Request[workflow.AbortRunRequest]
) - > *connect.Response[workflow.AbortRunResponse], error

Aborts an active workflow run, stopping its execution and marking it as failed or cancelled. This is used to terminate runs prematurely.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.AbortRunRequest]	The request containing the identifier of the run to abort.

Returns

Type	Description
*connect.Response[workflow.AbortRunResponse], error	A response indicating the success or failure of the abort operation, or an error if the run could not be aborted.

---

GetRunDetails()

@classmethod
def GetRunDetails(
    ctx: context.Context,
    req: *connect.Request[workflow.GetRunDetailsRequest]
) - > *connect.Response[workflow.GetRunDetailsResponse], error

Fetches detailed information about a specific workflow run. This allows clients to inspect the state and metadata of a run.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.GetRunDetailsRequest]	The request containing the identifier of the run to retrieve details for.

Returns

Type	Description
*connect.Response[workflow.GetRunDetailsResponse], error	A response containing the run's details, or an error if the run is not found or the operation fails.

---

GetActionDetails()

@classmethod
def GetActionDetails(
    ctx: context.Context,
    req: *connect.Request[workflow.GetActionDetailsRequest]
) - > *connect.Response[workflow.GetActionDetailsResponse], error

Retrieves comprehensive details for a specific action within a workflow run. This provides insight into an action's execution and status.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.GetActionDetailsRequest]	The request containing the identifier of the action to retrieve details for.

Returns

Type	Description
*connect.Response[workflow.GetActionDetailsResponse], error	A response containing the action's details, or an error if the action is not found or the operation fails.

---

getActionDetails()

@classmethod
def getActionDetails(
    ctx: context.Context,
    actionId: *common.ActionIdentifier
) - > *workflow.ActionDetails, error

Fetches the detailed information for a specific action from the data store. This is an internal method used to construct action details.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
actionId	*common.ActionIdentifier	The unique identifier of the action.

Returns

Type	Description
*workflow.ActionDetails, error	The detailed action information, or an error if the action is not found.

---

buildActionDetails()

@classmethod
def buildActionDetails(
    ctx: context.Context,
    model: *models.Action,
    actionId: *common.ActionIdentifier
) - > *workflow.ActionDetails, error

Constructs a workflow.ActionDetails object from a models.Action and an ActionIdentifier. This helper method formats raw data into a structured response.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
model	*models.Action	The action model containing the raw data.
actionId	*common.ActionIdentifier	The unique identifier for the action.

Returns

Type	Description
*workflow.ActionDetails, error	The constructed action details, or an error if building fails.

---

getAttempts()

@classmethod
def getAttempts(
    ctx: context.Context,
    actionId: *common.ActionIdentifier
) - > []*workflow.ActionAttempt, error

Retrieves all attempts associated with a given action. This is used to track retries and multiple executions of an action.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
actionId	*common.ActionIdentifier	The unique identifier of the action.

Returns

Type	Description
[]*workflow.ActionAttempt, error	A slice of action attempts, or an error if attempts cannot be retrieved.

---

GetActionData()

@classmethod
def GetActionData(
    ctx: context.Context,
    req: *connect.Request[workflow.GetActionDataRequest]
) - > *connect.Response[workflow.GetActionDataResponse], error

Fetches the data associated with a specific action, such as inputs, outputs, or logs. This allows clients to access the operational data of an action.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.GetActionDataRequest]	The request containing the identifier of the action and the type of data to retrieve.

Returns

Type	Description
*connect.Response[workflow.GetActionDataResponse], error	A response containing the action's data, or an error if the data is not found or the operation fails.

---

ListRuns()

@classmethod
def ListRuns(
    ctx: context.Context,
    req: *connect.Request[workflow.ListRunsRequest]
) - > *connect.Response[workflow.ListRunsResponse], error

Lists workflow runs based on specified filters and pagination criteria. This allows clients to query and browse available runs.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.ListRunsRequest]	The request containing filters, sorting, and pagination parameters for listing runs.

Returns

Type	Description
*connect.Response[workflow.ListRunsResponse], error	A response containing a list of runs matching the criteria, or an error if the listing fails.

---

ListActions()

@classmethod
def ListActions(
    ctx: context.Context,
    req: *connect.Request[workflow.ListActionsRequest]
) - > *connect.Response[workflow.ListActionsResponse], error

Lists actions within a specific workflow run or across runs, based on provided filters. This enables clients to inspect the actions that comprise a workflow.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.ListActionsRequest]	The request containing filters, sorting, and pagination parameters for listing actions.

Returns

Type	Description
*connect.Response[workflow.ListActionsResponse], error	A response containing a list of actions matching the criteria, or an error if the listing fails.

---

GetActionDataURIs()

@classmethod
def GetActionDataURIs(
    ctx: context.Context,
    req: *connect.Request[workflow.GetActionDataURIsRequest]
) - > *connect.Response[workflow.GetActionDataURIsResponse], error

Retrieves URIs for data associated with a specific action, such as input, output, or log locations. This allows clients to directly access action-related data.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.GetActionDataURIsRequest]	The request containing the action identifier and the type of data URIs to retrieve.

Returns

Type	Description
*connect.Response[workflow.GetActionDataURIsResponse], error	A response containing the URIs for the action's data, or an error if the URIs cannot be retrieved.

---

GetActionLogContext()

@classmethod
def GetActionLogContext(
    ctx: context.Context,
    req: *connect.Request[workflow.GetActionLogContextRequest]
) - > *connect.Response[workflow.GetActionLogContextResponse], error

Retrieves contextual information for an action's logs, such as log stream identifiers or access credentials. This is essential for accessing detailed action logs.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.GetActionLogContextRequest]	The request containing the action identifier for which to retrieve log context.

Returns

Type	Description
*connect.Response[workflow.GetActionLogContextResponse], error	A response containing the log context, or an error if the context cannot be retrieved.

---

AbortAction()

@classmethod
def AbortAction(
    ctx: context.Context,
    req: *connect.Request[workflow.AbortActionRequest]
) - > *connect.Response[workflow.AbortActionResponse], error

Aborts a specific action within a workflow run, stopping its execution. This allows for granular control over individual action lifecycles.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.AbortActionRequest]	The request containing the identifier of the action to abort.

Returns

Type	Description
*connect.Response[workflow.AbortActionResponse], error	A response indicating the success or failure of the abort operation, or an error if the action could not be aborted.

---

SignalEvent()

@classmethod
def SignalEvent(
    ctx: context.Context,
    _: *connect.Request[workflow.SignalEventRequest]
) - > *connect.Response[workflow.SignalEventResponse], error

Sends a signal event to a workflow run, potentially triggering specific actions or state transitions. This enables external interaction with running workflows.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
_	*connect.Request[workflow.SignalEventRequest]	The request containing the details of the signal event to be sent.

Returns

Type	Description
*connect.Response[workflow.SignalEventResponse], error	A response indicating the success or failure of sending the signal, or an error if the operation fails.

---

WatchRunDetails()

@classmethod
def WatchRunDetails(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchRunDetailsRequest],
    stream: *connect.ServerStream[workflow.WatchRunDetailsResponse]
) - > error

Establishes a server-side stream to watch for real-time updates to a specific run's details. This allows clients to monitor run progress and status changes.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchRunDetailsRequest]	The request specifying the run to watch.
stream	*connect.ServerStream[workflow.WatchRunDetailsResponse]	The server-side stream to send run detail updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

WatchActionDetails()

@classmethod
def WatchActionDetails(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchActionDetailsRequest],
    stream: *connect.ServerStream[workflow.WatchActionDetailsResponse]
) - > error

Establishes a server-side stream to watch for real-time updates to a specific action's details. This allows clients to monitor action progress and status changes.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchActionDetailsRequest]	The request specifying the action to watch.
stream	*connect.ServerStream[workflow.WatchActionDetailsResponse]	The server-side stream to send action detail updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

WatchRuns()

@classmethod
def WatchRuns(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchRunsRequest],
    stream: *connect.ServerStream[workflow.WatchRunsResponse]
) - > error

Establishes a server-side stream to watch for changes and new workflow runs based on specified criteria. This allows clients to receive real-time notifications about run lifecycle events.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchRunsRequest]	The request specifying filters for watching runs.
stream	*connect.ServerStream[workflow.WatchRunsResponse]	The server-side stream to send run updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

WatchActions()

@classmethod
def WatchActions(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchActionsRequest],
    stream: *connect.ServerStream[workflow.WatchActionsResponse]
) - > error

Establishes a server-side stream to watch for changes and new actions within workflow runs based on specified criteria. This allows clients to receive real-time notifications about action lifecycle events.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchActionsRequest]	The request specifying filters for watching actions.
stream	*connect.ServerStream[workflow.WatchActionsResponse]	The server-side stream to send action updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

listAndSendAllActions()

@classmethod
def listAndSendAllActions(
    ctx: context.Context,
    runID: *common.RunIdentifier,
    rsm: *runStateManager,
    stream: *connect.ServerStream[workflow.WatchActionsResponse]
) - > error

Lists all actions for a given run and sends them over a server stream. This is typically used to initialize a client's view of actions for a run.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
runID	*common.RunIdentifier	The unique identifier of the run.
rsm	*runStateManager	The run state manager used to access run data.
stream	*connect.ServerStream[workflow.WatchActionsResponse]	The server-side stream to send action updates to the client.

Returns

Type	Description
error	An error if listing or sending actions fails.

---

sendChangedActions()

@classmethod
def sendChangedActions(
    runID: *common.RunIdentifier,
    updates: []*nodeUpdate,
    stream: *connect.ServerStream[workflow.WatchActionsResponse]
) - > error

Sends only the changed actions from a list of node updates over a server stream. This optimizes updates by only transmitting modified action states.

Parameters

Name	Type	Description
runID	*common.RunIdentifier	The unique identifier of the run.
updates	[]*nodeUpdate	A slice of node updates containing information about changed actions.
stream	*connect.ServerStream[workflow.WatchActionsResponse]	The server-side stream to send action updates to the client.

Returns

Type	Description
error	An error if sending changed actions fails.

---

WatchClusterEvents()

@classmethod
def WatchClusterEvents(
    ctx: context.Context,
    req: *connect.Request[workflow.WatchClusterEventsRequest],
    stream: *connect.ServerStream[workflow.WatchClusterEventsResponse]
) - > error

Establishes a server-side stream to watch for cluster-related events. This allows clients to monitor infrastructure events relevant to workflow execution.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
req	*connect.Request[workflow.WatchClusterEventsRequest]	The request specifying criteria for watching cluster events.
stream	*connect.ServerStream[workflow.WatchClusterEventsResponse]	The server-side stream to send cluster event updates to the client.

Returns

Type	Description
error	An error if the stream operation encounters an issue.

---

getClusterEventsInfo()

@classmethod
def getClusterEventsInfo(
    ctx: context.Context,
    actionID: *common.ActionIdentifier,
    attempt: uint32,
    since: time.Time,
    offset: int,
    limit: int
) - > (clusterEventsInfo, error)

Retrieves information about cluster events for a specific action attempt, optionally filtered by time, offset, and limit. This is used to gather diagnostic data.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
actionID	*common.ActionIdentifier	The unique identifier of the action.
attempt	uint32	The specific attempt number of the action.
since	time.Time	The timestamp from which to retrieve events.
offset	int	The starting offset for event retrieval.
limit	int	The maximum number of events to retrieve.

Returns

Type	Description
(clusterEventsInfo, error)	A structure containing cluster event information, or an error if retrieval fails.

---

actionModelToDetails()

@classmethod
def actionModelToDetails(
    action: *models.Action,
    actionID: *common.ActionIdentifier
) - > *workflow.ActionDetails

Converts an internal models.Action object into a workflow.ActionDetails protobuf message. This facilitates consistent data representation for API responses.

Parameters

Name	Type	Description
action	*models.Action	The internal action model to convert.
actionID	*common.ActionIdentifier	The unique identifier for the action.

Returns

Type	Description
*workflow.ActionDetails	The converted workflow.ActionDetails message.

---

convertRunToProto()

@classmethod
def convertRunToProto(
    run: *models.Run
) - > *workflow.Run

Converts an internal models.Run object into a workflow.Run protobuf message. This standardizes the run data for external consumption.

Parameters

Name	Type	Description
run	*models.Run	The internal run model to convert.

Returns

Type	Description
*workflow.Run	The converted workflow.Run protobuf message.

---

convertActionToEnrichedProto()

@classmethod
def convertActionToEnrichedProto(
    action: *models.Action
) - > *workflow.EnrichedAction

Converts an internal models.Action object into an workflow.EnrichedAction protobuf message. This provides a more detailed view of an action for API responses.

Parameters

Name	Type	Description
action	*models.Action	The internal action model to convert.

Returns

Type	Description
*workflow.EnrichedAction	The converted workflow.EnrichedAction protobuf message.

---

convertNodeUpdateToEnrichedProto()

@classmethod
def convertNodeUpdateToEnrichedProto(
    runID: *common.RunIdentifier,
    update: *nodeUpdate
) - > *workflow.EnrichedAction

Converts a nodeUpdate object into an workflow.EnrichedAction protobuf message, associating it with a specific run. This is used to format action updates for streaming.

Parameters

Name	Type	Description
runID	*common.RunIdentifier	The unique identifier of the run to which the action belongs.
update	*nodeUpdate	The node update containing information about the action change.

Returns

Type	Description
*workflow.EnrichedAction	The converted workflow.EnrichedAction protobuf message.

---

buildTaskGroups()

@classmethod
def buildTaskGroups(
    ctx: context.Context,
    req: *workflow.WatchGroupsRequest
) - > []*workflow.TaskGroup, error

Builds a list of task groups based on a watch request. This organizes tasks into logical groups for display or processing.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request.
req	*workflow.WatchGroupsRequest	The request containing criteria for building task groups.

Returns

Type	Description
[]*workflow.TaskGroup, error	A slice of task groups, or an error if group building fails.

---

convertWatchRequestToListInput()

@classmethod
def convertWatchRequestToListInput(
    req: *workflow.WatchRunsRequest
) - > interfaces.ListResourceInput

Converts a workflow.WatchRunsRequest into an interfaces.ListResourceInput for internal data store querying. This adapts watch requests to a generic listing interface.

Parameters

Name	Type	Description
req	*workflow.WatchRunsRequest	The watch runs request to convert.

Returns

Type	Description
interfaces.ListResourceInput	The converted list resource input.

---

runMatchesFilter()

@classmethod
def runMatchesFilter(
    run: *models.Run,
    req: *workflow.WatchRunsRequest
) - > bool

Checks if a given run matches the filters specified in a workflow.WatchRunsRequest. This is used to determine which runs should be included in watch streams.

Parameters

Name	Type	Description
run	*models.Run	The run model to check against the filter.
req	*workflow.WatchRunsRequest	The watch runs request containing the filter criteria.

Returns

Type	Description
bool	True if the run matches the filter, false otherwise.

---