from __future__ import annotations import json from datetime import datetime, timezone from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session from app.modules.services.models import ClientServiceTaskInstance, ServiceTaskToolRun from app.modules.services.task_tools import get_task_tool, normalize_task_tool_code MAX_TOOL_OUTPUT_BYTES = 25 * 1024 * 1024 VALID_RESULT_STATUSES = {"completed", "completed_with_exceptions", "failed", "cancelled"} def start_task_tool_run( db: Session, *, task: ClientServiceTaskInstance, user_id: int | None, ) -> ServiceTaskToolRun: tool_code = normalize_task_tool_code(getattr(task, "task_tool_code", "NONE")) if tool_code == "NONE": raise ValueError("This task does not have an ERP tool linked.") get_task_tool(tool_code) run = ServiceTaskToolRun( tenant_id=int(task.tenant_id), task_instance_id=int(task.id), client_id=int(task.client_id), tool_code=tool_code, status="started", started_by_user_id=user_id, ) db.add(run) db.flush() db.commit() db.refresh(run) return run def list_task_tool_runs(db: Session, *, tenant_id: int, task_id: int) -> list[ServiceTaskToolRun]: return list( db.scalars( select(ServiceTaskToolRun) .where( ServiceTaskToolRun.tenant_id == int(tenant_id), ServiceTaskToolRun.task_instance_id == int(task_id), ) .order_by(ServiceTaskToolRun.started_at_utc.desc(), ServiceTaskToolRun.id.desc()) ).all() ) def get_task_tool_run(db: Session, *, tenant_id: int, run_id: int) -> ServiceTaskToolRun | None: return db.scalar( select(ServiceTaskToolRun).where( ServiceTaskToolRun.id == int(run_id), ServiceTaskToolRun.tenant_id == int(tenant_id), ) ) def _json_text(payload: Any) -> str | None: if payload is None or payload == "": return None if isinstance(payload, str): text = payload.strip() if not text: return None try: parsed = json.loads(text) return json.dumps(parsed, ensure_ascii=False, separators=(",", ":")) except Exception: return json.dumps({"value": text}, ensure_ascii=False, separators=(",", ":")) return json.dumps(payload, ensure_ascii=False, default=str, separators=(",", ":")) def record_task_tool_result( db: Session, *, run: ServiceTaskToolRun, status: str, summary: str | None = None, result_payload: Any = None, clean_pass: bool | None = None, exception_count: int | None = None, output_filename: str | None = None, output_content_type: str | None = None, output_bytes: bytes | None = None, completed_by_user_id: int | None = None, ) -> ServiceTaskToolRun: normalized_status = (status or "completed").strip().lower() if normalized_status not in VALID_RESULT_STATUSES: raise ValueError(f"Unsupported tool result status: {normalized_status}") if output_bytes is not None and len(output_bytes) > MAX_TOOL_OUTPUT_BYTES: raise ValueError("Tool output exceeds the 25 MB task-result attachment limit.") run.status = normalized_status run.summary = (summary or "").strip() or None run.result_json = _json_text(result_payload) run.clean_pass = clean_pass run.exception_count = exception_count run.output_filename = (output_filename or "").strip() or None run.output_content_type = (output_content_type or "").strip() or None run.output_bytes = output_bytes run.completed_by_user_id = completed_by_user_id run.completed_at_utc = datetime.now(timezone.utc) db.add(run) db.commit() db.refresh(run) return run def record_task_tool_result_by_id( db: Session, *, tenant_id: int, run_id: int, status: str, summary: str | None = None, result_payload: Any = None, clean_pass: bool | None = None, exception_count: int | None = None, output_filename: str | None = None, output_content_type: str | None = None, output_bytes: bytes | None = None, completed_by_user_id: int | None = None, ) -> ServiceTaskToolRun: run = get_task_tool_run(db, tenant_id=tenant_id, run_id=run_id) if run is None: raise ValueError("Task tool run not found.") return record_task_tool_result( db, run=run, status=status, summary=summary, result_payload=result_payload, clean_pass=clean_pass, exception_count=exception_count, output_filename=output_filename, output_content_type=output_content_type, output_bytes=output_bytes, completed_by_user_id=completed_by_user_id, )