agenticraft_foundation.protocols.workflow¶
Protocol workflow model \(W = (T, \prec, \rho)\) — task definitions, precedence ordering, protocol assignment, validation, and optimal assignment.
Protocol Workflow Model: W = (T, ≺, ρ).
Formal model for protocol-aware workflow composition.
- T: Set of tasks (nodes in the workflow DAG)
- ≺: Precedence relation (directed edges, forming a DAG)
- ρ: T → P assigns each task to a protocol
Implements: - Workflow executability verification (Theorem 9 conditions) - Optimal protocol assignment via DP (tree-structured) and greedy (general DAGs) - Translation cost minimization across task boundaries
WorkflowTaskStatus
¶
Bases: str, Enum
Status of a workflow task.
WorkflowTask
dataclass
¶
A task in the protocol workflow.
Each task has an ID, required capabilities, and an assigned protocol.
task_id
instance-attribute
¶
Unique task identifier
capabilities = field(default_factory=list)
class-attribute
instance-attribute
¶
Capabilities required to execute this task
assigned_protocol = None
class-attribute
instance-attribute
¶
Protocol assigned to this task (ρ(t))
compatible_protocols = field(default_factory=set)
class-attribute
instance-attribute
¶
Protocols that can execute this task
assigned_agent = None
class-attribute
instance-attribute
¶
Agent assigned to execute this task
metadata = field(default_factory=dict)
class-attribute
instance-attribute
¶
Additional task metadata
status = WorkflowTaskStatus.PENDING
class-attribute
instance-attribute
¶
Current task status
WorkflowEdge
dataclass
¶
ProtocolWorkflow
dataclass
¶
Protocol workflow model: W = (T, ≺, ρ).
Represents a directed acyclic graph of tasks with protocol assignments.
workflow_id
instance-attribute
¶
Unique workflow identifier
tasks = field(default_factory=dict)
class-attribute
instance-attribute
¶
Task set T, keyed by task_id
edges = field(default_factory=list)
class-attribute
instance-attribute
¶
Precedence relation ≺
add_task(task_id, capabilities=None, compatible_protocols=None, assigned_protocol=None, metadata=None)
¶
Add a task to the workflow.
add_precedence(source, target, data_dependency=True)
¶
Add a precedence edge: source ≺ target (source must complete before target).
predecessors(task_id)
¶
Get predecessor task IDs.
successors(task_id)
¶
Get successor task IDs.
roots()
¶
Get tasks with no predecessors (entry points).
leaves()
¶
Get tasks with no successors (exit points).
topological_sort()
¶
Return tasks in topological order. Raises ValueError if cycle detected.
is_tree()
¶
Check if the workflow DAG is a tree (each node has at most one predecessor).
translation_points()
¶
Find edges where protocol translation is needed.
Returns list of (source_task, target_task, source_protocol, target_protocol).
total_translation_cost(base_cost_per_translation=0.1)
¶
Calculate total protocol translation cost across the workflow.
WorkflowValidationResult
dataclass
¶
Result of workflow validation.
valid
instance-attribute
¶
Whether the workflow is executable
errors = field(default_factory=list)
class-attribute
instance-attribute
¶
Validation errors
warnings = field(default_factory=list)
class-attribute
instance-attribute
¶
Validation warnings
translation_count = 0
class-attribute
instance-attribute
¶
Number of protocol translations needed
estimated_cost = 0.0
class-attribute
instance-attribute
¶
Estimated total execution cost
WorkflowValidator
¶
Validate workflow executability.
Checks Theorem 9 conditions: 1. The workflow forms a valid DAG (no cycles) 2. Every task has an assigned protocol 3. Every task's assigned protocol is in its compatible set 4. For each precedence edge, the protocols are compatible (directly or via translation)
validate(workflow, graph=None)
¶
Validate workflow executability.
| PARAMETER | DESCRIPTION |
|---|---|
workflow
|
The protocol workflow to validate
TYPE:
|
graph
|
Optional protocol graph for agent-level validation
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
WorkflowValidationResult
|
WorkflowValidationResult with validation outcome |
OptimalProtocolAssigner
¶
Assign protocols to tasks minimizing total translation cost.
For tree-structured workflows: DP in O(|T| * |P|^2). For general DAGs: greedy heuristic.
assign(workflow)
¶
Assign protocols to minimize translation cost.
Uses DP for trees, greedy for general DAGs.
| PARAMETER | DESCRIPTION |
|---|---|
workflow
|
The workflow to assign protocols for
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, ProtocolName]
|
Mapping from task_id to assigned protocol |
apply_assignment(workflow, assignment)
¶
Apply a protocol assignment to the workflow tasks.