from __future__ import annotations from datetime import datetime from enum import Enum from pathlib import Path import re from pydantic import BaseModel, Field class ServiceName(str, Enum): PRODUCT = "product" ADMIN = "admin" class ToolLifecycleStatus(str, Enum): DRAFT = "draft" GENERATED = "generated" VALIDATED = "validated" APPROVED = "approved" ACTIVE = "active" FAILED = "failed" ARCHIVED = "archived" class ToolLifecycleStageContract(BaseModel): code: ToolLifecycleStatus label: str description: str order: int = Field(ge=1) terminal: bool = False TOOL_LIFECYCLE_STAGES: tuple[ToolLifecycleStageContract, ...] = ( ToolLifecycleStageContract( code=ToolLifecycleStatus.DRAFT, label="Draft", description="Estado inicial de uma tool ainda em definicao.", order=1, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.GENERATED, label="Generated", description="Implementacao gerada e pronta para analise tecnica.", order=2, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.VALIDATED, label="Validated", description="Tool validada automaticamente com verificacoes basicas.", order=3, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.APPROVED, label="Approved", description="Versao revisada e aprovada para publicacao controlada.", order=4, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.ACTIVE, label="Active", description="Tool publicada e apta a abastecer o runtime de produto.", order=5, terminal=False, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.FAILED, label="Failed", description="Falha registrada na geracao, validacao ou ativacao.", order=6, terminal=True, ), ToolLifecycleStageContract( code=ToolLifecycleStatus.ARCHIVED, label="Archived", description="Versao retirada de circulacao e mantida apenas para historico.", order=7, terminal=True, ), ) TOOL_LIFECYCLE_STATUS_SEQUENCE: tuple[ToolLifecycleStatus, ...] = tuple( stage.code for stage in TOOL_LIFECYCLE_STAGES ) _TOOL_LIFECYCLE_STAGE_BY_STATUS = { stage.code: stage for stage in TOOL_LIFECYCLE_STAGES } def get_tool_lifecycle_stage( status: ToolLifecycleStatus | str, ) -> ToolLifecycleStageContract: normalized_status = ( status if isinstance(status, ToolLifecycleStatus) else ToolLifecycleStatus(str(status or "").strip().lower()) ) return _TOOL_LIFECYCLE_STAGE_BY_STATUS[normalized_status] GENERATED_TOOLS_PACKAGE = "generated_tools" GENERATED_TOOL_ENTRYPOINT = "run" GENERATED_TOOL_PUBLICATION_MANIFEST = "published_runtime_tools.json" _GENERATED_TOOL_NAME_PATTERN = re.compile(r"^[a-z][a-z0-9_]{2,63}$") _PROJECT_ROOT = Path(__file__).resolve().parents[2] def _normalize_generated_tool_name(tool_name: str) -> str: normalized = str(tool_name or "").strip().lower() if not _GENERATED_TOOL_NAME_PATTERN.match(normalized): raise ValueError("tool_name must use lowercase snake_case to build the generated module path.") return normalized def build_generated_tool_module_name(tool_name: str) -> str: normalized = _normalize_generated_tool_name(tool_name) return f"{GENERATED_TOOLS_PACKAGE}.{normalized}" def build_generated_tool_module_path(tool_name: str) -> str: normalized = _normalize_generated_tool_name(tool_name) return f"{GENERATED_TOOLS_PACKAGE}/{normalized}.py" def get_generated_tools_runtime_dir(project_root: Path | None = None) -> Path: root = project_root or _PROJECT_ROOT return root / GENERATED_TOOLS_PACKAGE def build_generated_tool_file_path( tool_name: str, *, project_root: Path | None = None, ) -> Path: normalized = _normalize_generated_tool_name(tool_name) return get_generated_tools_runtime_dir(project_root) / f"{normalized}.py" def get_generated_tool_publication_manifest_path( project_root: Path | None = None, ) -> Path: return get_generated_tools_runtime_dir(project_root) / GENERATED_TOOL_PUBLICATION_MANIFEST class ToolParameterType(str, Enum): STRING = "string" INTEGER = "integer" NUMBER = "number" BOOLEAN = "boolean" OBJECT = "object" ARRAY = "array" class ToolParameterContract(BaseModel): name: str parameter_type: ToolParameterType description: str required: bool = True class PublishedToolContract(BaseModel): tool_name: str display_name: str description: str version: int = Field(ge=1) status: ToolLifecycleStatus parameters: tuple[ToolParameterContract, ...] = () implementation_module: str implementation_callable: str checksum: str | None = None published_at: datetime | None = None published_by: str | None = None class ToolPublicationEnvelope(BaseModel): source_service: ServiceName = ServiceName.ADMIN target_service: ServiceName = ServiceName.PRODUCT publication_id: str published_tool: PublishedToolContract emitted_at: datetime class ToolRuntimePublicationManifest(BaseModel): source_service: ServiceName = ServiceName.ADMIN target_service: ServiceName = ServiceName.PRODUCT emitted_at: datetime publications: tuple[ToolPublicationEnvelope, ...] = ()