> ## Documentation Index
> Fetch the complete documentation index at: https://uom-demo.vercel.app/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# graph

> [<span className="flex items-center gap-1"><Icon icon="external-link"/>View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/tree/main/services/orchestrator/src/react_agent/graph.py)

Define the Universal Object Mapping orchestrator graph.

## react\_agent.graph.ExtractionOutput Objects

```python theme={null}
class ExtractionOutput(BaseModel)
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L80)

Structured output for identifying user intent from messages.

This model is used by the initial extraction agent to parse unstructured conversation
history into a structured representation. It identifies the origin/target frameworks,
versions, and the raw code blocks intended for translation.

#### ExtractionOutput.join\_lists

```python theme={null}
@model_validator(mode="after")
def join_lists()
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L115)

Clean and normalize the fields after validation by joining list inputs into single strings.

**Returns**:

* `BaseExtractionOutput` - The validated and normalized model instance.

## react\_agent.graph.BaseTranslationOutput Objects

```python theme={null}
class BaseTranslationOutput(BaseModel)
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L128)

Structured output for the translated schema and/or queries.

This acts as the base Pydantic schema for the primary translation LLM node. It dynamically
expands based on whether the `translation_type` is SCHEMA, QUERY, or BOTH. It mandates that
the agent provide both the translated raw code and fully functional execution harnesses
(with explicitly declared entry points) for downstream sandbox validation.

#### BaseTranslationOutput.check\_entrypoint\_names

```python theme={null}
@model_validator(mode="after")
def check_entrypoint_names()
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L183)

Clean, normalize, and validate that the entrypoint class names exist in the generated harness code.

**Returns**:

* `BaseTranslationOutput` - The validated and normalized model instance.

**Raises**:

* `ValueError` - If the entrypoint type name is missing from the harness/schema code.

#### react\_agent.graph.is\_input\_extracted

```python theme={null}
def is_input_extracted(state: State | ExtractionOutput) -> bool
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L382)

Check if the necessary structured input has been successfully extracted from conversation.

Validates that based on the `translation_type`, the corresponding source code fields
(schema, query, or both) and the framework targets are present.

**Arguments**:

* `state` *State | ExtractionOutput* - The current graph state or extraction output model.

**Returns**:

* `bool` - True if all required fields are present, False otherwise.

#### react\_agent.graph.extract\_input

```python theme={null}
async def extract_input(state: State, config: RunnableConfig,
                        runtime: Runtime[Context])
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L411)

Extract raw source code and targets from recent messages if missing from structured input.

This node uses a specialized ReAct agent to analyze the conversation history and extract
the necessary parameters (origin framework, destination framework, and the raw source code
snippets) needed to begin the translation process. If it fails, it updates the extraction
loop counter and may terminate the graph if the maximum retry limit is reached.

**Arguments**:

* `state` *State* - The current state of the graph.
* `config` *RunnableConfig* - Configuration parameters for the run.
* `runtime` *Runtime\[Context]* - The execution runtime containing context.

**Returns**:

dict\[str, Any] | Command: State updates with extracted parameters or a Command to
terminate the graph on failure.

#### react\_agent.graph.schema\_inspection

```python theme={null}
async def schema_inspection(state: State, config: RunnableConfig,
                            runtime: Runtime[Context]) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L548)

Inspect source and target database schemas using database tools.

Runs a lightweight ReAct agent equipped with database inspection tools (MCP) to examine
the relevant database schemas before translation begins. This gathers contextual data
that is then injected into the main translation prompt to ensure accuracy.

**Arguments**:

* `state` *State* - The current state of the graph.
* `config` *RunnableConfig* - Configuration parameters for the run.
* `runtime` *Runtime\[Context]* - The execution runtime containing context.

**Returns**:

dict\[str, Any]: State updates containing the extracted `schema_context` string.

#### react\_agent.graph.translation\_agent

```python theme={null}
async def translation_agent(state: State, config: RunnableConfig,
                            runtime: Runtime[Context]) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L675)

Use a ReAct agent to perform translation and validation loops natively.

Combines static tools (validators, fallback docs) with dynamically loaded
database and documentation MCP tools.

**Warnings**:

This node is DEPRECATED in favor of `generate_translation_node` coupled with
explicit state machine nodes for validation and evaluation, which provides
better determinism and observability.

**Arguments**:

* `state` *State* - The current state of the graph.
* `config` *RunnableConfig* - Configuration parameters for the run.
* `runtime` *Runtime\[Context]* - The execution runtime containing context.

**Returns**:

dict\[str, Any]: State updates containing the translation output.

#### react\_agent.graph.generate\_translation\_node

```python theme={null}
async def generate_translation_node(
        state: State, config: RunnableConfig,
        runtime: Runtime[Context]) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L802)

Deterministically generate the translation using structured LLM output via a React Agent without tools.

This node acts as the core "Generation" step in the iterative translation loop.
It takes the extracted source code, the schema context, and the previous translation
attempts/feedback (if any) and generates the translated code and execution harnesses
using a strongly-typed Pydantic model (`TranslationOutput`).

**Arguments**:

* `state` *State* - The current state of the graph.
* `config` *RunnableConfig* - Configuration parameters for the run.
* `runtime` *Runtime\[Context]* - The execution runtime containing context.

**Returns**:

dict\[str, Any]: State updates containing the generated translation outputs.

## react\_agent.graph.HumanInterventionResponse Objects

```python theme={null}
class HumanInterventionResponse(BaseModel)
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L892)

Pydantic model representing the feedback and decision from a human-in-the-loop intervention.

**Attributes**:

* `decision` - The logical decision, either "accept" to commit the translation or "reject" to loop back with feedback.
* `feedback` - Text description or critique describing necessary adjustments.

#### react\_agent.graph.human\_intervention\_node

```python theme={null}
async def human_intervention_node(state: State)
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L904)

Pause the graph execution to request human-in-the-loop (HITL) feedback.

This node interrupts the state machine, surfacing the current translation code and validation
results (including deep diffs) to the user via the `interrupt` LangGraph API. The user can
either 'accept' the translation to terminate successfully, or 'reject' it with feedback to
trigger another generation loop.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

dict\[str, Any] | Command: State updates appending the user's feedback to messages,
or a Command to end the graph if accepted.

#### react\_agent.graph.prep\_schema\_validation

```python theme={null}
def prep_schema_validation(state: State) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L956)

Inject ToolCalls into the message history to trigger schema compilation validation.

Prepares the state for the `validate_schema_node` by appending an AIMessage with explicitly
defined tool calls (`validate_dotnet_code` or `validate_java_code`) containing the target
schema harness code.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

dict\[str, Any]: State updates with the injected validation tool calls.

#### react\_agent.graph.prep\_query\_validation

```python theme={null}
def prep_query_validation(state: State) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1011)

Inject ToolCalls into the message history for parallel source and target query validation.

Prepares the state for the `validate_query_node` by appending an AIMessage with multiple
tool calls to run both the source validation harness and the target validation harness in
sandbox environments concurrently.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

dict\[str, Any]: State updates with the injected validation tool calls.

#### react\_agent.graph.prep\_query\_equivalence

```python theme={null}
def prep_query_equivalence(state: State) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1102)

Inject a ToolCall into the message history to run the query equivalence checker.

Prepares the state for the `check_query_equivalence_node`. It parses the JSON validation
outputs from the previous query validation step and issues a tool call for `DeepDiff`
equivalence testing.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

dict\[str, Any]: State updates with the injected equivalence tool call.

#### react\_agent.graph.custom\_tool\_node\_wrapper

```python theme={null}
async def custom_tool_node_wrapper(
    request: ToolCallRequest, execute: Callable[[ToolCallRequest],
                                                Awaitable[Union[ToolMessage,
                                                                Command]]]
) -> Union[ToolMessage, Command]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1139)

Wrap tool execution to provide robust retries against transient infrastructure errors.

Used by `ToolNode` to intercept and retry tool calls (like Daytona sandbox provisioning)
with an exponential backoff. If max retries are exceeded, it gracefully injects an error
ToolMessage into the state rather than crashing the graph.

**Arguments**:

* `request` *ToolCallRequest* - The requested tool call payload.
* `execute` *Callable* - The underlying ToolNode execution function.

**Returns**:

Union\[ToolMessage, Command]: The result of the tool execution or an error ToolMessage.

#### react\_agent.graph.route\_post\_query\_validation

```python theme={null}
def route_post_query_validation(
        state: State) -> Literal["prep_query_equivalence", "evaluation_node"]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1248)

Determine the next state transition after parallel query validation.

If both the source and target query validations completed successfully (indicated by
'\[Validation Passed]' in their tool output), it routes to `prep_query_equivalence`.
Otherwise, it skips equivalence checking and routes directly to the `evaluation_node`
to analyze the validation failures.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

Literal\["prep\_query\_equivalence", "evaluation\_node"]: The name of the next node.

## react\_agent.graph.EvaluationOutput Objects

```python theme={null}
class EvaluationOutput(BaseModel)
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1282)

Pydantic model representing the LLM evaluation outcome for translation acceptance.

**Attributes**:

* `decision` - The logical decision, either ACCEPT to complete the process or REJECT to loop back for correction.
* `explanation` - Detailed textual reasoning explaining the decision, citing specific equivalence or compiler errors.

#### react\_agent.graph.evaluation\_node

```python theme={null}
async def evaluation_node(state: State, config: RunnableConfig,
                          runtime: Runtime[Context]) -> dict[str, Any]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1296)

Evaluate validation and equivalence testing results to decide on translation acceptance.

This node uses a specialized evaluation LLM to act as a judge. It reviews the compiler
outputs from the sandboxes and the JSON DeepDiff equivalence results. Based on this, it
generates a structured `EvaluationOutput` deciding whether to 'ACCEPT' the translation
or 'REJECT' it (which triggers another generation iteration with feedback).

**Arguments**:

* `state` *State* - The current state of the graph.
* `config` *RunnableConfig* - Configuration parameters for the run.
* `runtime` *Runtime\[Context]* - The execution runtime containing context.

**Returns**:

dict\[str, Any]: State updates containing the evaluation decision and explanation.

#### react\_agent.graph.route\_post\_evaluation

```python theme={null}
def route_post_evaluation(
    state: State
) -> Literal["generate_translation_node", "human_intervention_node",
             "__end__"]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1386)

Determine the next state transition after evaluation.

If the evaluation was rejected or failed, it routes back to `generate_translation_node`
to retry. If the maximum translation loop count is reached, it routes to
`human_intervention_node` instead. If accepted, it routes to `__end__`.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

Literal\["generate\_translation\_node", "human\_intervention\_node", "**end**"]: The next node.

#### react\_agent.graph.should\_extract\_input

```python theme={null}
def should_extract_input(
        state: State
) -> Literal["schema_inspection", "extract_input", "__end__"]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1419)

Determine the next state transition during the initial extraction phase.

Checks if all required structured inputs have been successfully parsed. If so, it routes
to `schema_inspection`. If not, it routes back to `extract_input` up to a maximum
retry limit, after which it routes to `__end__` to terminate the graph gracefully.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

Literal\["schema\_inspection", "extract\_input", "**end**"]: The next node.

#### react\_agent.graph.route\_post\_translation

```python theme={null}
def route_post_translation(
    state: State
) -> Literal["prep_schema_validation", "prep_query_validation"]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1441)

Determine the next validation state transition after code generation.

Routes to `prep_schema_validation` if the translation type is SCHEMA. For QUERY or BOTH
translation types, it routes to `prep_query_validation`.

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

Literal\["prep\_schema\_validation", "prep\_query\_validation"]: The next node.

#### react\_agent.graph.route\_post\_schema\_validation

```python theme={null}
def route_post_schema_validation(
    state: State
) -> Literal[
        "prep_query_validation",
        "generate_translation_node",
        "human_intervention_node",
        "__end__",
]
```

[<span className="flex items-center gap-1"><Icon icon="external-link" />View Source</span>](https://github.com/corovcam/Universal-Object-Mapping/blob/main/services/orchestrator/src/react_agent/graph.py#L1460)

Determine the next state transition after schema validation.

If schema compilation failed, it routes back to `generate_translation_node` (or
`human_intervention_node` if max retries exceeded) without proceeding further.
If it passed and the translation type is BOTH, it routes to `prep_query_validation`.
Otherwise, it terminates execution (`__end__`).

**Arguments**:

* `state` *State* - The current state of the graph.

**Returns**:

Literal\["prep\_query\_validation", "generate\_translation\_node", "human\_intervention\_node", "**end**"]: The next node.