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 in a workflow run. 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's details and status. This internal method is likely called by the public RecordAction method.

Parameters

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

Returns

Type	Description
*workflow.RecordActionResponse	A response containing the result of the action recording operation.

---

recordSingleAction()

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

Persists the details of a single action to the data store. This method handles the underlying storage logic for recording an action.

Parameters

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

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 a specific action within a workflow run. This method is used to reflect changes in an action's execution state.

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

Provides a bidirectional stream for updating 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 specific action. This internal method is likely called by the public UpdateActionStatus method.

Parameters

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

Returns

Type	Description
*workflow.UpdateActionStatusResponse	A response indicating the success or failure of the 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 handles the underlying storage logic for updating an action's status.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
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 an action. This method is used to log significant occurrences during an action's lifecycle.

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 the action events to be recorded.

Returns

Type	Description
*connect.Response[workflow.RecordActionEventsResponse], error	A response indicating the success or failure of the event recording, 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 for recording action events. This allows for continuous and efficient logging of events as they occur.

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 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 handles the underlying storage logic for recording events.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
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 handles the storage of run configuration and initial state.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
runId	*common.RunIdentifier	The unique identifier for the run.
taskID	*task.TaskIdentifier	The identifier of the task associated with this run.
taskSpec	*task.TaskSpec	The specification of the task being executed.
inputPrefix	string	The prefix for input data paths related to this run.
runOutputBase	string	The base path for output data generated by this run.
runSpec	*task.RunSpec	The specific runtime configuration for this run.
source	workflow.RunSource	The source or origin of the workflow run (e.g., manual, scheduled).
triggerName	string	The name of the trigger that initiated this run, if any.
triggerTaskName	string	The name of the task associated with the trigger, if applicable.
triggerRevision	int64	The revision number of the trigger, if applicable.
triggerType	string	The type of trigger that initiated this run (e.g., 'schedule', 'webhook').

Returns

Type	Description
*models.Run, error	The persisted run model, or an error if the model could not be saved.

---

AbortRun()

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

Aborts an active workflow run, stopping all ongoing actions and marking the run as failed or aborted. This is used to terminate a run 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 be aborted.

Returns

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

---

GetRunDetails()

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

Fetches comprehensive details for a specific workflow run. This allows clients to retrieve all metadata and status information about 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 whose details are to be retrieved.

Returns

Type	Description
*connect.Response[workflow.GetRunDetailsResponse], error	A response containing the detailed information of the requested run, or an error if the run is not found or the operation failed.

---

GetActionDetails()

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

Retrieves detailed information for a specific action within a workflow run. This provides granular insight into an action's execution.

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 whose details are to be retrieved.

Returns

Type	Description
*connect.Response[workflow.GetActionDetailsResponse], error	A response containing the detailed information of the requested action, or an error if the action is not found or the operation failed.

---

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 internal method retrieves the raw action data.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
actionId	*common.ActionIdentifier	The unique identifier for the action.

Returns

Type	Description
*workflow.ActionDetails, error	The detailed information of the requested action, 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 comprehensive ActionDetails object from a raw action model and its identifier. This method enriches the basic action data with additional context.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
model	*models.Action	The raw action model retrieved from the data store.
actionId	*common.ActionIdentifier	The unique identifier for the action.

Returns

Type	Description
*workflow.ActionDetails, error	A fully populated ActionDetails object, or an error if details cannot be built.

---

getAttempts()

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

Retrieves all attempts associated with a specific action. This is useful for understanding retries and execution history of an action.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
actionId	*common.ActionIdentifier	The unique identifier for 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 input and output data associated with a specific action. This allows clients to inspect the data processed by 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 whose data is to be retrieved.

Returns

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

---

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 method is used to browse and search for workflow executions.

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 workflow runs matching the criteria, or an error if the operation failed.

---

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 method allows for querying and viewing action history.

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 operation failed.

---

GetActionDataURIs()

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

Retrieves URIs for the input and output data of a specific action. This allows clients to directly access the data stored externally.

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 identifier of the action whose data URIs are to be retrieved.

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 generated or the operation failed.

---

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 links. This helps in locating and accessing 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 identifier of the action whose log context is to be retrieved.

Returns

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

---

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 is used to terminate an individual action prematurely.

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 be aborted.

Returns

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

---

SignalEvent()

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

Sends a signal event to the workflow system, which can trigger specific behaviors or state transitions. This allows for 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 the signal event, or an error if the operation failed.

---

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 workflow run's details. This allows clients to stay synchronized with run state 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 and any filtering criteria.
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 individual action progress.

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 and any filtering criteria.
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 continuous updates on run activity.

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 and other criteria 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 a workflow run or across multiple runs. This allows clients to receive continuous updates on action activity.

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 and other criteria 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 ID and sends them over the provided server stream. This is typically used to initialize a client's view of actions.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
runID	*common.RunIdentifier	The unique identifier of the run whose actions are to be listed and sent.
rsm	*runStateManager	The run state manager used to retrieve action information.
stream	*connect.ServerStream[workflow.WatchActionsResponse]	The server-side stream to which action updates will be sent.

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 (node updates) for a given run ID over the provided server stream. This optimizes updates by only sending deltas.

Parameters

Name	Type	Description
runID	*common.RunIdentifier	The unique identifier of the run for which action changes are to be sent.
updates	[]*nodeUpdate	A slice of node updates representing the changed actions.
stream	*connect.ServerStream[workflow.WatchActionsResponse]	The server-side stream to which action updates will be sent.

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-level events related to workflow execution. This provides insights into the underlying infrastructure's activity.

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, with optional time, offset, and limit filters. This is used to paginate and filter cluster event logs.

Parameters

Name	Type	Description
ctx	context.Context	The context for the request, used for cancellation and deadlines.
actionID	*common.ActionIdentifier	The unique identifier of the action.
attempt	uint32	The attempt number for the action.
since	time.Time	The starting timestamp for retrieving events.
offset	int	The number of events to skip from the beginning.
limit	int	The maximum number of events to retrieve.

Returns

Type	Description
clusterEventsInfo, error	A structure containing cluster event information, or an error if the events cannot be retrieved.

---

actionModelToDetails()

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

Converts an internal action model into a public-facing ActionDetails protobuf message. This method transforms raw data into a structured API response.

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 ActionDetails protobuf message.

---

convertRunToProto()

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

Converts an internal run model into a public-facing Run protobuf message. This method transforms raw run data into a structured API response.

Parameters

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

Returns

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

---

convertActionToEnrichedProto()

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

Converts an internal action model into an EnrichedAction protobuf message, adding additional context or derived information. This provides a more complete view of an action for clients.

Parameters

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

Returns

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

---

convertNodeUpdateToEnrichedProto()

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

Converts a node update (representing a change in an action) into an EnrichedAction protobuf message. This is used to send incremental updates to clients watching actions.

Parameters

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

Returns

Type	Description
*workflow.EnrichedAction	The converted EnrichedAction protobuf message representing the updated action.

---

buildTaskGroups()

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

Builds a list of task groups based on the provided watch request. This method aggregates actions into logical groups for display or management.

Parameters

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

Returns

Type	Description
[]*workflow.TaskGroup, error	A slice of TaskGroup protobuf messages, or an error if groups cannot be built.

---

convertWatchRequestToListInput()

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

Converts a WatchRunsRequest into a generic ListResourceInput interface. This abstracts the watch request for use with common listing logic.

Parameters

Name	Type	Description
req	*workflow.WatchRunsRequest	The WatchRunsRequest to convert.

Returns

Type	Description
interfaces.ListResourceInput	A generic ListResourceInput interface derived from the watch request.

---

runMatchesFilter()

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

Checks if a given run model matches the filters specified in a WatchRunsRequest. This is used internally to filter runs for streaming updates.

Parameters

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

Returns

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

---