Artemis-Eapil/tests/unit/agents/test_native_receipts.py

"""Actual action-result/history/transcript rendering must keep native evidence."""

from copy import deepcopy
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
import pytest

from artemis.agents.validator.action_execution import exec_action
from artemis.mcp.action_types import ActionCode, ActionResult
from artemis.memory.transcript import EXECUTION_RESULT_MARKER, TranscriptLedger
from artemis.utils.task_tree import format_result_clean


NATIVE_ACTIONS = (
    "pinch",
    "touch_gesture",
    "read_control",
    "control_action",
    "wait_until",
    "scroll_to",
)
TOKEN = "a" * 64
SELECTOR = {"resource_id": "fixture:id/editor"}


def action_arguments(action):
    drag = {
        "start": [191, 501],
        "end": [192, 532],
        "coordinate_space": "pixel",
        "target_description": "observed canvas",
    }
    return {
        "pinch": {
            "center": [200, 400],
            "start_span": 100,
            "end_span": 200,
            "coordinate_space": "pixel",
            "target_description": "observed canvas",
        },
        "touch_gesture": {
            "frames": [
                {"at_ms": 0, "points": [[191, 501]]},
                {"at_ms": 400, "points": [[192, 532]]},
            ],
            "coordinate_space": "pixel",
            "target_description": "observed canvas",
        },
        "read_control": {"selector": SELECTOR},
        "control_action": {
            "selector": SELECTOR,
            "action": "set_selection",
            "selection_start": 3,
            "selection_end": 5,
        },
        "wait_until": {"selector": SELECTOR, "state": "value", "expected_value": 37.5},
        "scroll_to": {"container": {"resource_id": "fixture:id/list"}, "target": SELECTOR},
        "drag": drag,
        "action_sequence": {"steps": [{"action": "drag", **drag}]},
    }[action]


def control_observation():
    return {
        "resource_id": "fixture:id/editor",
        "text": "  A😀B \n",
        "selection_start": 3,
        "selection_end": 5,
        "range": {"min": 0, "max": 100, "current": 37.5, "type": 1},
        "token": TOKEN,
        "content_signature": "b" * 64,
        "editable": True,
        "password": False,
    }


def decode_receipt(text, label):
    marker = f"{label} receipt: "
    assert marker in text
    return json.loads(text.split(marker, 1)[1])


async def render_result_pipeline(action, metadata, failed):
    message = (
        "Native reply is uncertain; observe before continuing." if failed else "Input accepted."
    )
    result = (
        ActionResult.failure(action, message, code=ActionCode.TIMEOUT, metadata=metadata)
        if failed
        else ActionResult.success(action, message, metadata=metadata)
    )
    session = SimpleNamespace(call=AsyncMock(return_value=result))
    context = SimpleNamespace(
        data_engine=None, device=SimpleNamespace(device_width=400, device_height=800)
    )
    args = action_arguments(action)
    item = {"action": action, "args": args}
    ok, error = await exec_action(context, session, item)
    assert ok is not failed
    session.call.assert_awaited_once_with(action, args)
    assert item["result"] == result.agent_message
    report = {
        "status": "failed" if failed else "dispatched",
        "execution": [{**item, "attempts": [f"Error: {error}" if failed else "Dispatched"]}],
    }
    history = format_result_clean(report)
    assert isinstance(history, str)

    ledger = TranscriptLedger(clock=lambda: 0.0)
    ledger.stage_turn(
        [
            HumanMessage(content="Observed native controls"),
            AIMessage(content="", tool_calls=[{"name": action, "args": {}, "id": "native-call"}]),
            ToolMessage(content="Action Recorded", tool_call_id="native-call"),
        ]
    )
    ledger.commit_staged(step_key="native-step", validator_result=report)
    rendered = ledger.render([HumanMessage(content="Next observation")])
    execution_results = [
        message.content[0]["text"]
        for message in rendered
        if isinstance(message, HumanMessage)
        and isinstance(message.content, list)
        and message.content
        and EXECUTION_RESULT_MARKER in message.content[0].get("text", "")
    ]
    assert len(execution_results) == 1
    transcript = execution_results[0]
    assert f"Status: {report['status']}" in transcript
    if failed:
        assert "Error:" in history and "Error:" in transcript
    return result.agent_message, history, transcript


@pytest.mark.asyncio
@pytest.mark.parametrize("action", NATIVE_ACTIONS)
@pytest.mark.parametrize("failed", [False, True])
async def test_every_native_action_keeps_readings_selection_range_and_token_in_actual_result_rendering(
    action, failed
):
    node = control_observation()
    read_only = action in {"read_control", "wait_until"}
    execution = (
        ("not_started" if read_only else "unknown")
        if failed
        else ("observed" if read_only else "completed")
    )
    # The renderer must retain optional observed evidence even on gesture receipts.
    metadata = {
        "action": action,
        "execution": execution,
        "node": deepcopy(node),
        "native_receipt": {
            "success": not failed,
            "execution": execution,
            "code": "TIMEOUT" if failed else "OK",
            "node": deepcopy(node),
        },
        "verification": "pending_observation",
    }
    stages = await render_result_pipeline(action, metadata, failed)
    for rendered in stages:
        recovered = decode_receipt(rendered, "Native")
        assert recovered == metadata
        assert recovered["node"]["text"] == "  A😀B \n"
        assert recovered["node"]["range"]["current"] == 37.5
        assert (recovered["node"]["selection_start"], recovered["node"]["selection_end"]) == (3, 5)
        assert recovered["node"]["token"] == TOKEN
        assert recovered["native_receipt"]["success"] is not failed


@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["drag", "action_sequence"])
@pytest.mark.parametrize("failed", [False, True])
async def test_existing_drag_and_sequence_geometry_and_partial_receipts_still_render(
    action, failed
):
    metadata = {
        "execution": "partial" if failed else "completed",
        "coordinate_space": "pixel",
        "start": [191, 501],
        "end": [192, 532],
        "calibration": {"axis": "y", "target_value": 37.5},
    }
    if action == "action_sequence":
        metadata["sequence_results"] = [
            {"action": "click", "execution": "completed", "completed_taps": 1},
            {"action": "drag", "execution": "unknown" if failed else "completed"},
        ]
    for rendered in await render_result_pipeline(action, metadata, failed):
        assert decode_receipt(rendered, "Gesture") == metadata