test(uipath-agents): coded HITL / process / RAG / tracing tests (#475)

- skill-agent-coded-hitl-create-task (e2e) — LangGraph agent with
  interrupt(CreateTask(app_name="ExpenseReview",
  app_folder_path="Finance", ...)) and a MemorySaver checkpointer.
  Asserts the imports, the call shape, the checkpointer wiring,
  and the `app` binding for ExpenseReview/Finance.
- skill-agent-coded-process-invoke (e2e) — LangGraph agent with
  interrupt(InvokeProcess(name="DataScraper",
  process_folder_path="Workflows", ...)). Asserts the call shape
  and the `process` binding.
- skill-agent-coded-rag-langgraph (e2e) — LangGraph RAG agent with
  ContextGroundingRetriever(index_name="company_docs",
  folder_path="Shared") inside a node. Asserts the import path,
  retriever args, lazy-init, and the `index` binding.
- skill-agent-coded-tracing-redaction (e2e) — Simple Function agent
  whose `main` is decorated with @Traced(name=...,
  input_processor=..., output_processor=...). Asserts all three
  kwargs are present and rejects the conflicting hide_input/
  hide_output combo.

Author: radugheo

tests/tasks/uipath-agents/hitl_create_task/check_hitl_create_task.py

@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""HITL coded-agent shape check.
+
+Asserts the primitives that together make
+`interrupt(CreateEscalation)` work end-to-end:
+
+  1. `main.py` imports `interrupt` from `langgraph.types`.
+  2. `main.py` imports `CreateEscalation` from
+     `uipath.platform.common` and references it inside `interrupt(...)`.
+     The prompt is an explicit escalation — the skill prescribes
+     `CreateEscalation` for that path (`CreateTask` is the general
+     form for non-escalation HITL).
+  3. `bindings.json` declares the `app` resource for the Action
+     Center app the escalation targets — `app_name=ExpenseReview`,
+     `app_folder_path=Finance`. Without this binding,
+     `uipath push` would not create the virtual placeholder.
+
+Also runs the lazy-LLM-init AST scan as a hygiene check.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+import os
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from _shared.project_root import find_project_root  # noqa: E402
+
+ROOT = find_project_root("expense-approver")
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from _shared.bindings_assertions import (  # noqa: E402
+    load_bindings,
+    find_resource,
+    assert_value_field,
+    assert_metadata_field,
+)
+from _shared.ast_lazy_init_check import find_module_level_llm_clients  # noqa: E402
+
+
+def _read_text(path: Path) -> str:
+    if not path.is_file():
+        sys.exit(f"FAIL: Missing {path}")
+    return path.read_text(encoding="utf-8")
+
+
+def find_graph_module() -> Path:
+    for candidate in ("main.py", "graph.py"):
+        path = ROOT / candidate
+        if path.is_file():
+            return path
+    sys.exit(f"FAIL: neither main.py nor graph.py found under {ROOT}")
+
+
+def _module_constants(tree: ast.Module) -> dict[str, object]:
+    """Collect module-level `<Name> = <Constant>` assignments.
+
+    Resolves the common pattern where the agent extracts string literals
+    into constants (e.g. ``ACTION_CENTER_APP = "ExpenseReview"``).
+    """
+    consts: dict[str, object] = {}
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant):
+            for tgt in node.targets:
+                if isinstance(tgt, ast.Name):
+                    consts[tgt.id] = node.value.value
+    return consts
+
+
+def _resolve_kwarg(value: ast.expr, consts: dict[str, object]) -> object | None:
+    if isinstance(value, ast.Constant):
+        return value.value
+    if isinstance(value, ast.Name) and value.id in consts:
+        return consts[value.id]
+    return None
+
+
+def _find_create_escalation_call(tree: ast.Module) -> ast.Call | None:
+    """Return the inner `CreateEscalation(...)` call wrapped by `interrupt(...)`."""
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if not (isinstance(func, ast.Name) and func.id == "interrupt"):
+            continue
+        if not node.args:
+            continue
+        inner = node.args[0]
+        if not isinstance(inner, ast.Call):
+            continue
+        inner_func = inner.func
+        if isinstance(inner_func, ast.Name) and inner_func.id == "CreateEscalation":
+            return inner
+    return None
+
+
+def check_imports_and_calls(text: str, tree: ast.Module) -> None:
+    if not re.search(r"from\s+langgraph\.types\s+import\s+[^\n]*\binterrupt\b", text):
+        sys.exit("FAIL: missing `from langgraph.types import interrupt`")
+    print("OK: imports `interrupt` from langgraph.types")
+    if not re.search(r"from\s+uipath\.platform\.common\s+import\s+[^\n]*\bCreateEscalation\b", text):
+        sys.exit(
+            "FAIL: missing `from uipath.platform.common import CreateEscalation`. "
+            "The prompt describes an explicit escalation — the skill prescribes "
+            "`CreateEscalation` for that path."
+        )
+    print("OK: imports CreateEscalation from uipath.platform.common")
+    call = _find_create_escalation_call(tree)
+    if call is None:
+        sys.exit("FAIL: no `interrupt(CreateEscalation(...))` call site found")
+    print("OK: graph node calls interrupt(CreateEscalation(...))")
+    consts = _module_constants(tree)
+    kwargs = {kw.arg: kw.value for kw in call.keywords if kw.arg is not None}
+    expected = {"app_name": "ExpenseReview", "app_folder_path": "Finance"}
+    for kw, want in expected.items():
+        if kw not in kwargs:
+            sys.exit(f'FAIL: `CreateEscalation(...)` is missing `{kw}=`')
+        got = _resolve_kwarg(kwargs[kw], consts)
+        if got != want:
+            sys.exit(
+                f'FAIL: `CreateEscalation({kw}=...)` resolves to {got!r}, expected {want!r}.'
+            )
+    print('OK: escalation targets app_name="ExpenseReview" / app_folder_path="Finance"')
+
+
+def check_app_binding() -> None:
+    doc = load_bindings(ROOT / "bindings.json")
+    entry = find_resource(doc, resource="app", key="ExpenseReview.Finance")
+    assert_value_field(entry, field="name", expected="ExpenseReview")
+    assert_value_field(entry, field="folderPath", expected="Finance")
+    assert_metadata_field(entry, field="ActivityName", expected="create_async")
+    assert_metadata_field(entry, field="DisplayLabel", expected="ExpenseReview")
+    print("OK: bindings.json declares the ExpenseReview/Finance `app` resource")
+
+
+def main() -> None:
+    if not ROOT.is_dir():
+        sys.exit(f"FAIL: project directory {ROOT} does not exist")
+    module = find_graph_module()
+    text = _read_text(module)
+    tree = ast.parse(text, filename=str(module))
+    check_imports_and_calls(text, tree)
+    violations = find_module_level_llm_clients(module)
+    if violations:
+        sys.exit("FAIL: " + " | ".join(violations))
+    print("OK: no module-level UiPath* construction")
+    check_app_binding()
+
+
+if __name__ == "__main__":
+    main()

tests/tasks/uipath-agents/hitl_create_task/hitl_create_task.yaml

@@ -0,0 +1,65 @@
+task_id: skill-agent-coded-hitl-create-task
+description: >
+  Coded-agent HITL via `interrupt(CreateEscalation(...))`. Verifies
+  the agent imports `interrupt` from `langgraph.types` and
+  `CreateEscalation` from `uipath.platform.common`, compiles the
+  graph with a `MemorySaver` checkpointer, and emits the `app`
+  binding for the Action Center app the escalation targets.
+tags: [uipath-agents, e2e, coded, lifecycle:generate, feature:hitl-coded]
+max_iterations: 1
+
+agent:
+  type: claude-code
+  permission_mode: acceptEdits
+  allowed_tools: ["Skill", "Bash", "Read", "Write", "Edit", "Glob", "Grep"]
+  turn_timeout: 1200
+
+sandbox:
+  driver: tempdir
+  python: {}
+
+initial_prompt: |
+  Build a LangGraph UiPath coded agent named `expense-approver` that
+  escalates expense-reimbursement requests over `$1000` to a human
+  reviewer in Action Center. Below the threshold, the agent
+  auto-approves.
+
+  The escalation should target an Action Center app named
+  `ExpenseReview` in folder `Finance`, passing the full request
+  payload to the reviewer. The graph must support pause/resume so
+  the escalation can wait for the human response.
+
+  Input: `amount` (number), `description` (string), `submitter`
+  (string).
+  Output: `approved` (bool), `reviewer_decision` (string|null).
+
+  Sync `bindings.json` so the Action Center resource is declared.
+
+  Take the agent through scaffold → init. Do NOT run, publish,
+  upload, or deploy. Do NOT call `uip login`. Do NOT pause between
+  planning and implementation. Complete end-to-end in a single pass.
+
+success_criteria:
+  - type: command_executed
+    description: "Agent scaffolded the project with uip codedagent new"
+    tool_name: "Bash"
+    command_pattern: 'uip\s+codedagent\s+new'
+    min_count: 1
+    weight: 1.0
+    pass_threshold: 1.0
+
+  - type: command_executed
+    description: "Agent ran uip codedagent init"
+    tool_name: "Bash"
+    command_pattern: 'uip\s+codedagent\s+init'
+    min_count: 1
+    weight: 1.5
+    pass_threshold: 1.0
+
+  - type: run_command
+    description: "interrupt(CreateEscalation) shape, MemorySaver checkpointer, app binding"
+    command: "python3 $TASK_DIR/check_hitl_create_task.py"
+    timeout: 30
+    expected_exit_code: 0
+    weight: 6.0
+    pass_threshold: 1.0

tests/tasks/uipath-agents/process_invoke/check_process_invoke.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""Process-invocation coded-agent shape check.
+
+Asserts:
+  1. ``main.py`` imports ``interrupt`` from ``langgraph.types`` and
+     ``InvokeProcess`` from ``uipath.platform.common``.
+  2. A graph node calls ``interrupt(InvokeProcess(...))`` whose
+     ``name=`` and ``process_folder_path=`` resolve (through any
+     module-level constant the agent extracted) to ``DataScraper``
+     and ``Workflows``.
+  3. ``bindings.json`` declares the ``process`` resource for
+     DataScraper / Workflows with ``ActivityName=invoke_async``.
+  4. No module-level UiPath* construction.
+"""
+
+from __future__ import annotations
+
+import ast
+import os
+import re
+import sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from _shared.project_root import find_project_root  # noqa: E402
+
+ROOT = find_project_root("data-orchestrator")
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from _shared.bindings_assertions import (  # noqa: E402
+    load_bindings,
+    find_resource,
+    assert_value_field,
+    assert_metadata_field,
+)
+from _shared.ast_lazy_init_check import find_module_level_llm_clients  # noqa: E402
+
+
+def _read_text(path: Path) -> str:
+    if not path.is_file():
+        sys.exit(f"FAIL: Missing {path}")
+    return path.read_text(encoding="utf-8")
+
+
+def find_graph_module() -> Path:
+    for candidate in ("main.py", "graph.py"):
+        path = ROOT / candidate
+        if path.is_file():
+            return path
+    sys.exit(f"FAIL: neither main.py nor graph.py found under {ROOT}")
+
+
+def _module_constants(tree: ast.Module) -> dict[str, object]:
+    consts: dict[str, object] = {}
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant):
+            for tgt in node.targets:
+                if isinstance(tgt, ast.Name):
+                    consts[tgt.id] = node.value.value
+    return consts
+
+
+def _resolve(value: ast.expr, consts: dict[str, object]) -> object | None:
+    if isinstance(value, ast.Constant):
+        return value.value
+    if isinstance(value, ast.Name) and value.id in consts:
+        return consts[value.id]
+    return None
+
+
+def _kwarg(call: ast.Call, name: str) -> ast.expr | None:
+    for kw in call.keywords:
+        if kw.arg == name:
+            return kw.value
+    return None
+
+
+def _find_invoke_process_call(tree: ast.Module) -> ast.Call | None:
+    """Return the inner ``InvokeProcess(...)`` call wrapped by ``interrupt(...)``."""
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if not (isinstance(func, ast.Name) and func.id == "interrupt"):
+            continue
+        if not node.args:
+            continue
+        inner = node.args[0]
+        if not isinstance(inner, ast.Call):
+            continue
+        inner_func = inner.func
+        if isinstance(inner_func, ast.Name) and inner_func.id == "InvokeProcess":
+            return inner
+    return None
+
+
+def check_invocation(text: str, tree: ast.Module) -> None:
+    if not re.search(r"from\s+langgraph\.types\s+import\s+[^\n]*\binterrupt\b", text):
+        sys.exit("FAIL: missing `from langgraph.types import interrupt`")
+    if not re.search(
+        r"from\s+uipath\.platform\.common\s+import\s+[^\n]*\bInvokeProcess\b",
+        text,
+    ):
+        sys.exit("FAIL: missing `from uipath.platform.common import InvokeProcess`")
+
+    call = _find_invoke_process_call(tree)
+    if call is None:
+        sys.exit("FAIL: no `interrupt(InvokeProcess(...))` call site found")
+
+    consts = _module_constants(tree)
+    expected = {"name": "DataScraper", "process_folder_path": "Workflows"}
+    for kw, want in expected.items():
+        node = _kwarg(call, kw)
+        if node is None:
+            sys.exit(f"FAIL: `InvokeProcess(...)` is missing `{kw}=`")
+        got = _resolve(node, consts)
+        if got != want:
+            sys.exit(
+                f"FAIL: `InvokeProcess({kw}=...)` resolves to {got!r}, expected {want!r}."
+            )
+    print(
+        'OK: graph node calls `interrupt(InvokeProcess(name="DataScraper", '
+        'process_folder_path="Workflows", ...))`'
+    )
+
+
+def check_process_binding() -> None:
+    doc = load_bindings(ROOT / "bindings.json")
+    entry = find_resource(doc, resource="process", key="DataScraper.Workflows")
+    assert_value_field(entry, field="name", expected="DataScraper")
+    assert_value_field(entry, field="folderPath", expected="Workflows")
+    assert_metadata_field(entry, field="ActivityName", expected="invoke_async")
+    print("OK: bindings.json declares the DataScraper/Workflows `process` resource")
+
+
+def main() -> None:
+    if not ROOT.is_dir():
+        sys.exit(f"FAIL: project directory {ROOT} does not exist")
+    module = find_graph_module()
+    text = _read_text(module)
+    tree = ast.parse(text, filename=str(module))
+    check_invocation(text, tree)
+    violations = find_module_level_llm_clients(module)
+    if violations:
+        sys.exit("FAIL: " + " | ".join(violations))
+    print("OK: no module-level UiPath* construction")
+    check_process_binding()
+
+
+if __name__ == "__main__":
+    main()