Create DeterministicStepComponent for Appointed Tool Execution

Summary

Create a DeterministicStepComponent that implements the "Deterministic step" from the design document, enabling execution of appointed tools with provided arguments while adhering to the new componentization framework with fixed output conventions.

Description

This issue implements part of the design document from gitlab-com/content-sites/handbook!14034 (merged) and follows the componentization architecture established in #1216 (closed). The DeterministicStepComponent should provide a standardized way to execute specific tools deterministically using inputs to extract necessary data and following fixed output conventions.

Reference Implementation: The existing RunToolNode at duo_workflow_service/agents/run_tool_node.py provides the foundation for this component, demonstrating:

Tool execution with input and output parsing
Security validation and monitoring integration
Error handling

Key Design Principles:

Input-Driven Execution: Use component inputs to extract necessary data from state for tool execution
Fixed Output Convention: Adhere to standardized output patterns using IOKeyTemplate
Dynamic Naming: Use IOKeyTemplate with component name replacement to avoid output collisions
Deterministic Behavior: Execute appointed tools with predetermined logic

Requirements

Core Functionality:

Create DeterministicStepComponent class extending BaseComponent
Execute appointed tools using data extracted from component inputs
Follow fixed output convention with IOKeyTemplate patterns
Support dynamic component name replacement in output keys
Enable deterministic tool execution without AI agent involvement

Input Management:

Use component inputs to extract tool execution parameters from state
Support flexible input extraction from various state targets
Handle input validation and parameter preparation
Enable complex parameter extraction for tool arguments

Output Management:

Implement fixed output convention using IOKeyTemplate
Support dynamic component name replacement to avoid collisions
Follow standardized output patterns similar to AgentComponent
Enable predictable output routing for downstream components

Integration Points:

Follow BaseComponent interface patterns from #1216 (closed)
Integrate with existing tool security validation
Support workflow monitoring and metrics
Maintain compatibility with existing toolsets

Technical Specifications

Fixed Output Convention with IOKeyTemplate:

class DeterministicStepComponent(BaseComponent):
    _tool_result_key: ClassVar[IOKeyTemplate] = IOKeyTemplate(
        target="context",
        subkeys=[IOKeyTemplate.COMPONENT_NAME_TEMPLATE, "tool_result"],
    )
    
    _outputs: ClassVar[tuple[IOKeyTemplate, ...]] = (
        IOKeyTemplate(target="ui_chat_log"),
        _tool_result_key,
    )
    
    tool_name: str
    toolset: Toolset
    
    _allowed_input_targets: list[str] = ["context", "conversation_history"]
    
    def __entry_hook__(self) -> str:
        return f"{self.name}#deterministic_step"
    
    def attach(self, graph: StateGraph, router: BaseRouter) -> str:
        graph.add_node(self.__entry_hook__(), self._execute_tool)
        graph.add_conditional_edges(self.__entry_hook__(), router.route)
        return self.__entry_hook__()

Input-Driven Tool Execution:

async def _execute_tool(self, state) -> dict[str, Any]:
    # Extract tool parameters from component inputs
    variables = get_vars_from_state(self.inputs, state)
    
    # Get appointed tool from toolset
    tool = self.toolset[self.tool_name]
    
    # Execute tool with extracted parameters
    tool_result = await tool.arun(**variables)
    
    # Apply security validation
    secure_result = PromptSecurity.apply_security(
        response=tool_result, tool_name=self.tool_name
    )
    
    # Return using fixed output convention
    return {
        "ui_chat_log": [self._create_ui_log(variables, secure_result)],
        "context": {
            self.name: {
                "tool_result": secure_result
            }
        }
    }

Dynamic Output Key Generation:

# Component outputs follow fixed pattern with dynamic naming
# Example: context:read_config.tool_result, context:scan_files.tool_result
def get_output_key(self) -> IOKey:
    return self._tool_result_key.to_iokey({
        IOKeyTemplate.COMPONENT_NAME_TEMPLATE: self.name
    })

Use Cases

File Reading with Fixed Output:

# Reads file and outputs to context:read_config.tool_result
read_component = DeterministicStepComponent(
    name="read_config",
    tool_name="read_file",
    inputs=[IOKey(target="context", subkeys=["config_path"])],
    toolset=file_toolset,
    workflow_id="workflow_123",
    workflow_type=CategoryEnum.DUO_WORKFLOW,
)
# Output: state["context"]["read_config"]["tool_result"] = file_content

Directory Scanning with Dynamic Naming:

# Scans directory and outputs to context:scan_files.tool_result
scan_component = DeterministicStepComponent(
    name="scan_files",
    tool_name="find_files",
    inputs=[IOKey(target="context", subkeys=["search_pattern"])],
    toolset=file_toolset,
    workflow_id="workflow_123",
    workflow_type=CategoryEnum.DUO_WORKFLOW,
)
# Output: state["context"]["scan_files"]["tool_result"] = found_files_list

Multiple Parameter Extraction:

# Extracts multiple inputs for tool execution
grep_component = DeterministicStepComponent(
    name="search_content",
    tool_name="grep",
    inputs=[
        IOKey(target="context", subkeys=["search_pattern"]),
        IOKey(target="context", subkeys=["target_directory"]),
    ],
    toolset=search_toolset,
    workflow_id="workflow_123",
    workflow_type=CategoryEnum.DUO_WORKFLOW,
)
# Tool called with: grep(pattern=search_pattern, search_directory=target_directory)
# Output: state["context"]["search_content"]["tool_result"] = grep_results

Chaining Components:

# Component A outputs to context:read_config.tool_result
# Component B reads from context:read_config.tool_result
process_component = DeterministicStepComponent(
    name="process_config",
    tool_name="edit_file",
    inputs=[
        IOKey(target="context", subkeys=["read_config", "tool_result"]),  # From previous component
        IOKey(target="context", subkeys=["edit_instructions"]),
    ],
    toolset=file_toolset,
    workflow_id="workflow_123",
    workflow_type=CategoryEnum.DUO_WORKFLOW,
)

Acceptance Criteria

DeterministicStepComponent follows BaseComponent interface patterns
Component uses inputs to extract tool execution parameters from state
Tool execution works with extracted parameters deterministically
Fixed output convention using IOKeyTemplate is implemented
Dynamic component name replacement prevents output collisions
Security validation and monitoring integration works
UI logging captures tool execution with proper formatting
Component outputs follow predictable patterns for downstream use

Technical Notes

Build on RunToolNode patterns from duo_workflow_service/agents/run_tool_node.py
Follow the new BaseComponent framework from #1216 (closed)
Use IOKeyTemplate patterns similar to AgentComponent._final_answer_key
Support create_nested_dict() for output structure
Integrate with existing security and monitoring systems
Enable deterministic tool execution without AI involvement

Related Issues

Builds on #1216 (closed) (Base Component Class Architecture)
Part of epic gitlab-org/duo-workflow&1 (Componentize Duo Workflow Graph)
Related to #1303 (Human-in-the-Loop Tool Approval)
Related to #1304 (PlannerComponent for Plan Generation)
Related to #1305 (HumanInputComponent for User Input)

Definition of Done

DeterministicStepComponent implementation complete
Input-driven tool execution working correctly
Fixed output convention with IOKeyTemplate implemented
Dynamic component name replacement preventing collisions
Security validation and monitoring integrated
error handling working
Integration tests with various tools and input scenarios
Documentation and usage examples provided
Component follows established framework patterns

Edited Jul 23, 2025 by Mikołaj Wawrzyniak