ComfyUI_Skills_OpenClaw/scripts/shared/execution_history.py

from __future__ import annotations

import hashlib
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from shared.config import get_server_history_dir, get_server_history_entry_path
from shared.json_utils import load_json, save_json


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as file:
        for chunk in iter(lambda: file.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def build_run_record(
    server_id: str,
    workflow_id: str,
    run_id: str,
    raw_args: dict[str, Any],
    workflow_path: Path,
    schema_path: Path,
) -> dict[str, Any]:
    timestamp = utc_now_iso()
    return {
        "run_id": run_id,
        "server_id": server_id,
        "workflow_id": workflow_id,
        "status": "queued",
        "created_at": timestamp,
        "started_at": None,
        "finished_at": None,
        "duration_ms": None,
        "prompt_id": None,
        "raw_args": raw_args,
        "resolved_args": {},
        "workflow_snapshot": {
            "path": str(workflow_path),
            "sha256": file_sha256(workflow_path),
        },
        "schema_snapshot": {
            "path": str(schema_path),
            "sha256": file_sha256(schema_path),
        },
        "result": {
            "images": [],
            "image_count": 0,
        },
        "error": None,
    }


def save_run_record(server_id: str, workflow_id: str, record: dict[str, Any]) -> None:
    save_json(get_server_history_entry_path(server_id, workflow_id, str(record["run_id"])), record)


def get_run_record(server_id: str, workflow_id: str, run_id: str) -> dict[str, Any]:
    path = get_server_history_entry_path(server_id, workflow_id, run_id)
    if not path.exists():
        raise FileNotFoundError(run_id)
    payload = load_json(path)
    if not isinstance(payload, dict):
        raise ValueError(f"Run history record is invalid: {run_id}")
    return payload


def list_run_records(server_id: str, workflow_id: str) -> list[dict[str, Any]]:
    history_dir = get_server_history_dir(server_id, workflow_id)
    if not history_dir.exists():
        return []

    records: list[dict[str, Any]] = []
    for path in sorted(history_dir.glob("*.json")):
        try:
            payload = load_json(path)
        except Exception:
            continue
        if isinstance(payload, dict):
            records.append(payload)

    records.sort(
        key=lambda item: (
            str(item.get("created_at") or ""),
            str(item.get("run_id") or ""),
        ),
        reverse=True,
    )
    return records


def delete_run_record(server_id: str, workflow_id: str, run_id: str) -> None:
    path = get_server_history_entry_path(server_id, workflow_id, run_id)
    if not path.exists():
        raise FileNotFoundError(run_id)
    path.unlink()


def clear_run_records(server_id: str, workflow_id: str) -> int:
    history_dir = get_server_history_dir(server_id, workflow_id)
    if not history_dir.exists():
        return 0

    deleted = 0
    for path in history_dir.glob("*.json"):
        if path.is_file():
            path.unlink()
            deleted += 1

    try:
        history_dir.rmdir()
    except OSError:
        pass

    return deleted


def summarize_run_record(record: dict[str, Any]) -> dict[str, Any]:
    result = record.get("result") if isinstance(record.get("result"), dict) else {}
    error = record.get("error") if isinstance(record.get("error"), dict) else {}
    return {
        "run_id": record.get("run_id"),
        "server_id": record.get("server_id"),
        "workflow_id": record.get("workflow_id"),
        "status": record.get("status"),
        "created_at": record.get("created_at"),
        "started_at": record.get("started_at"),
        "finished_at": record.get("finished_at"),
        "duration_ms": record.get("duration_ms"),
        "prompt_id": record.get("prompt_id"),
        "raw_args": record.get("raw_args") if isinstance(record.get("raw_args"), dict) else {},
        "resolved_args": record.get("resolved_args") if isinstance(record.get("resolved_args"), dict) else {},
        "image_count": result.get("image_count", 0),
        "images": result.get("images", []) if isinstance(result.get("images"), list) else [],
        "error_message": error.get("message") if isinstance(error.get("message"), str) else "",
    }