Artemis-Eapil/artemis/agents/checker/receipts.py

"""Small delivery receipts from actual tool records, never executor prose."""

import json


_FILE_FIELDS = (
    "ok",
    "action",
    "path",
    "trace_path",
    "device_path",
    "name",
    "bytes",
    "sha256",
    "verified",
    "media_type",
    "error",
)


def _fields(value: dict, keys) -> dict:
    # Exclude bodies, tool arguments and embedded images; they are not receipts.
    return {
        key: item[:1000] if isinstance(item, str) else item
        for key in keys
        if isinstance(item := value.get(key), (str, int, float, bool))
    }


def _result_object(value) -> dict:
    for _ in range(3):
        if isinstance(value, str):
            try:
                value = json.loads(value)
            except ValueError:
                return {}
        if not isinstance(value, dict):
            return {}
        if "outcome" not in value:
            return value
        value = value["outcome"]
    return {}


def _file_receipt(value) -> dict:
    result = _result_object(value)
    if result:
        return _fields(result, _FILE_FIELDS)
    # Legacy notes and ToolFailure both return plain text. Preserve it without
    # inventing a success/failure or read-back verification flag.
    return {"recorded_result": str(value)[:500]} if value else {"receipt": "not recorded"}


def _plugin_receipt(result: dict) -> dict | None:
    attachments = result.get("attachments")
    if not isinstance(attachments, list) or not attachments:
        return None
    receipt = _fields(
        result, ("plugin_id", "call_id", "status", "plugin_status", "execution", "error_code")
    )
    source = result.get("source")
    if isinstance(source, dict):
        receipt["source"] = _fields(source, ("kind", "target_id", "label", "captured_at"))
    # These are runtime-validated attachments, not structured_content supplied by a plugin.
    receipt["attachments"] = [
        _fields(item, _FILE_FIELDS) for item in attachments[:8] if isinstance(item, dict)
    ]
    if len(attachments) > 8:
        receipt["attachments_omitted"] = len(attachments) - 8
    return receipt


def _step_receipts(step: dict):
    seen = set()
    android_seen = False
    for event in step.get("tool_calls") or []:
        if not isinstance(event, dict):
            continue
        name = str(event.get("name") or "")
        if name in {"android_file", "save_note"}:
            android_seen = android_seen or name == "android_file"
            yield {"tool": name, **_file_receipt(event.get("result"))}
        elif name.startswith("plug_"):
            receipt = _plugin_receipt(_result_object(event.get("result")))
            if receipt:
                seen.add(receipt.get("call_id"))
                yield {"tool": name, **receipt}
    # Older/helper records may lack tool traces. Execution metadata is still factual.
    result = step.get("last_execution_result") or {}
    result = result if isinstance(result, dict) else {}
    actions = result.get("execution")
    if not isinstance(actions, list):
        actions = step.get("action_taken")
    actions = actions if isinstance(actions, list) else [actions]
    for action in actions:
        if not isinstance(action, dict):
            continue
        name = action.get("action")
        if name == "android_file" and not android_seen:
            yield {"tool": name, **_file_receipt(result.get("result"))}
        elif action.get("plugin_id") and action.get("call_id") not in seen:
            receipt = _plugin_receipt(action)
            if receipt:
                yield {"tool": name, **receipt}


def file_receipt_summary(steps, *, max_receipts: int = 24, max_chars: int = 16000) -> str:
    """Bounded preview of the complete run, including receipts before the history tail."""
    if not isinstance(steps, list):
        return "File receipt history unavailable; inspect recorded history before judging delivery."
    rows = [
        json.dumps({"step": step.get("step_number"), **receipt}, ensure_ascii=False)
        for step in steps
        for receipt in _step_receipts(step)
    ]
    selected, size = [], 0
    for row in reversed(rows):
        if len(selected) >= max_receipts or size + len(row) + 1 > max_chars:
            break
        selected.append(row)
        size += len(row) + 1
    text = "\n".join(reversed(selected)) or "(no file receipts in this preview)"
    if len(rows) > len(selected):
        text += f"\n{len(rows) - len(selected)} receipt(s) omitted from bounded preview."
    return (
        text
        + "\nReceipts establish only the recorded file operation, destination and verification at "
        "that time, not correctness/completeness of the file's contents or delivery elsewhere. "
        "A missing preview entry is not proof of a missing file. Use search_history / replay_steps "
        "to resolve omissions, errors or contradictions before judging delivery."
    )