Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/uipath_langchain/agent/exceptions/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ class AgentRuntimeErrorCode(str, Enum):

# Input / output validation
INVALID_ATTACHMENT_ID = "INVALID_ATTACHMENT_ID"
INVALID_ATTACHMENT = "INVALID_ATTACHMENT"
OUTPUT_VALIDATION_ERROR = "OUTPUT_VALIDATION_ERROR"
INVALID_STATIC_ARGUMENT = "INVALID_STATIC_ARGUMENT"

Expand Down
64 changes: 57 additions & 7 deletions src/uipath_langchain/agent/react/job_attachments.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@

from jsonpath_ng import parse # type: ignore[import-untyped]
from langchain_core.messages import BaseMessage, HumanMessage
from pydantic import BaseModel
from pydantic import BaseModel, ValidationError
from uipath.platform.attachments import Attachment
from uipath.runtime.errors import UiPathErrorCategory

from ..exceptions import AgentRuntimeError, AgentRuntimeErrorCode
from .json_utils import extract_values_by_paths, get_json_paths_by_type


Expand All @@ -23,20 +25,68 @@ def get_job_attachments(
data: The data object (dict or Pydantic model) to extract attachments from

Returns:
List of Attachment objects
List of Attachment objects.

Raises:
AgentRuntimeError: If a tool-output attachment fails validation (e.g. its
ID is not a valid UUID). This is unrecoverable invalid data and is
surfaced as a SYSTEM failure rather than silently skipped.
"""
job_attachment_paths = get_job_attachment_paths(schema)
job_attachments = extract_values_by_paths(data, job_attachment_paths)

result = [
Attachment.model_validate(att, from_attributes=True)
for att in job_attachments
if att
]
result = []
for att in job_attachments:
if not att:
continue
attachment_id = (
att.get("ID") if isinstance(att, dict) else getattr(att, "ID", None)
)
try:
uuid.UUID(str(attachment_id))
except (ValueError, TypeError) as e:
raise AgentRuntimeError(
code=AgentRuntimeErrorCode.INVALID_ATTACHMENT_ID,
title="Invalid attachment id",
detail=(
f"A tool returned a job attachment with id {attachment_id!r}, "
f"which is not a valid UUID. The agent cannot proceed with an "
f"invalid attachment."
),
category=UiPathErrorCategory.SYSTEM,
) from e
try:
result.append(Attachment.model_validate(att, from_attributes=True))
except ValidationError as e:
raise AgentRuntimeError(
code=AgentRuntimeErrorCode.INVALID_ATTACHMENT,
title="Invalid job attachment",
detail=(
f"A tool returned a job attachment (id {attachment_id!r}) that "
f"does not match the expected shape — {_describe_validation_errors(e)}. "
f"Verify the tool's output provides valid attachment fields; the "
f"agent cannot proceed with an invalid attachment."
),
category=UiPathErrorCategory.SYSTEM,
) from e

return result


def _describe_validation_errors(exc: ValidationError) -> str:
"""Render a pydantic ValidationError as a short, human-readable field list.

Reports each failing field path and reason (e.g. ``'MimeType': Field required``)
without echoing the offending input values, so the message is actionable and
safe to surface.
"""
issues = []
for err in exc.errors():
field = ".".join(str(part) for part in err.get("loc", ())) or "attachment"
issues.append(f"'{field}': {err.get('msg', 'invalid value')}")
return "; ".join(issues)


def get_job_attachment_paths(model: type[BaseModel]) -> list[str]:
"""Get JSONPath expressions for all job attachment fields in a Pydantic model.

Expand Down
87 changes: 87 additions & 0 deletions tests/agent/react/test_job_attachments.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,17 @@
import uuid
from typing import Any

import pytest
from jsonschema_pydantic_converter import transform_with_modules
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from pydantic import BaseModel
from uipath.platform.attachments import Attachment
from uipath.runtime.errors import UiPathErrorCategory

from uipath_langchain.agent.exceptions import (
AgentRuntimeError,
AgentRuntimeErrorCode,
)
from uipath_langchain.agent.react.job_attachments import (
get_job_attachments,
parse_attachments_from_conversation_messages,
Expand Down Expand Up @@ -446,6 +452,87 @@ def test_deeply_nested_and_array_structures(self):
ids = {str(att.id) for att in result}
assert ids == {uuid1, uuid2, uuid3}

def test_raises_system_error_on_non_uuid_attachment_id(self):
"""A tool-output attachment with a non-UUID ID must fail loud as SYSTEM."""
schema = {
"type": "object",
"properties": {
"attachments": {
"type": "array",
"items": {"$ref": "#/definitions/job-attachment"},
}
},
"definitions": {
"job-attachment": {
"type": "object",
"properties": {
"ID": {"type": "string"},
"FullName": {"type": "string"},
"MimeType": {"type": "string"},
},
"required": ["ID"],
}
},
}
model = create_model(schema)
data = {
"attachments": [
{"ID": "att_x", "FullName": "bad.pdf", "MimeType": "application/pdf"},
]
}

with pytest.raises(AgentRuntimeError) as exc_info:
get_job_attachments(model, data)

error_info = exc_info.value.error_info
assert error_info.category == UiPathErrorCategory.SYSTEM
assert error_info.code == AgentRuntimeError.full_code(
AgentRuntimeErrorCode.INVALID_ATTACHMENT_ID
)
assert "att_x" in error_info.detail

def test_raises_system_error_with_field_on_invalid_attachment_shape(self):
"""A valid-UUID attachment missing a required field fails loud as SYSTEM,
naming the offending field in a human-readable message."""
schema = {
"type": "object",
"properties": {
"attachments": {
"type": "array",
"items": {"$ref": "#/definitions/job-attachment"},
}
},
"definitions": {
"job-attachment": {
"type": "object",
"properties": {
"ID": {"type": "string"},
"FullName": {"type": "string"},
"MimeType": {"type": "string"},
},
"required": ["ID"],
}
},
}
model = create_model(schema)
valid_uuid = "550e8400-e29b-41d4-a716-446655440400"
data = {
"attachments": [
{"ID": valid_uuid, "FullName": "doc.pdf"}, # MimeType missing
]
}

with pytest.raises(AgentRuntimeError) as exc_info:
get_job_attachments(model, data)

error_info = exc_info.value.error_info
assert error_info.category == UiPathErrorCategory.SYSTEM
assert error_info.code == AgentRuntimeError.full_code(
AgentRuntimeErrorCode.INVALID_ATTACHMENT
)
assert "MimeType" in error_info.detail or "mime_type" in error_info.detail
assert valid_uuid in error_info.detail

def test_filters_out_none_attachments_in_array(self):
"""Should filter out None items from attachment arrays."""
schema = {
Expand Down
Loading