"""skills/_lib/contract.py — Skill interface + Registry + helpers. Cualquier skill nueva crea un módulo `skills/.py` con UNA función: def run(inputs: dict, context: SkillContext) -> SkillResult: ... Donde: SkillContext = ejecuta-time helpers (caja, shared_state, run_dir, claude CLI helper). SkillResult = {ok: bool, output: dict, errors: list[str], metadata: dict} El registry auto-descubre todas las skills bajo `skills/*.py` (skip _lib y __init__). Patrón canónico: cada skill ELIGE su backend internamente: - Python puro (yt-dlp / faster-whisper / Trello API / file ops) - claude CLI subprocess (LLM via plan Max — sin API key) - HTTP / API externa (Apify / GetHookd / etc) - Browser CDP (Playwright) El orchestrator NO conoce los backends. Solo invoca la skill por nombre. """ from __future__ import annotations import importlib import importlib.util import json import re import subprocess import sys from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Any, Callable ROOT = Path(__file__).resolve().parents[2] # _factory_3d_demo/ SKILLS_DIR = ROOT / 'skills' # ============================================================ # RESULT TYPES — contrato I/O uniforme de las skills # ============================================================ @dataclass class SkillResult: ok: bool output: dict = field(default_factory=dict) errors: list[str] = field(default_factory=list) metadata: dict = field(default_factory=dict) def to_dict(self) -> dict: return asdict(self) # ============================================================ # CONTEXT — pasado a cada skill # ============================================================ class SkillContext: """Helpers que cada skill puede usar para tocar caja, shared_state, claude CLI, etc. NO permite que la skill toque estado de OTRAS skills directamente — solo escribir su sección en la caja vía `context.write_caja(section, value, mode)`. """ def __init__(self, flow_id: str, run_dir: Path, station_id: str, caja, shared_state_path: Path | None = None): self.flow_id = flow_id self.run_dir = run_dir self.station_id = station_id self.caja = caja # objeto Caja (de orchestrator) self.shared_state_path = shared_state_path # opcional, para clusters paralelos def read_caja(self) -> dict: return self.caja.read_full() def read_section(self, section: str) -> Any: return self.caja.read_section(section) def write_section(self, section: str, value: Any, mode: str = 'replace', actor: str = ''): actor = actor or self.station_id self.caja.write_section(section, value, mode=mode, actor=actor) def claude_cli(self, prompt: str, model: str = 'claude-opus-4-7', system: str | None = None, output_format: str = 'text', timeout_s: int = 300) -> dict: """Invoca el binario `claude` CLI vía subprocess. Usa el plan Max — sin API key. Args: prompt: el contenido a procesar. model: claude-opus-4-7 default (REGLA msg 4658 Fer 2026-05-09). system: system prompt opcional. output_format: 'text' | 'json' (claude CLI flag). timeout_s: máximo en segundos. Returns: {ok, stdout, stderr, returncode, parsed_json (si output_format=json)} """ cmd = ['claude', '--print', '--model', model] if output_format == 'json': cmd += ['--output-format', 'json'] if system: cmd += ['--system-prompt', system] cmd.append(prompt) try: r = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', timeout=timeout_s) result = { 'ok': r.returncode == 0, 'returncode': r.returncode, 'stdout': r.stdout, 'stderr': r.stderr, } if output_format == 'json' and r.returncode == 0: try: result['parsed_json'] = json.loads(r.stdout) except Exception as e: result['parse_error'] = str(e) return result except subprocess.TimeoutExpired: return {'ok': False, 'error': f'timeout {timeout_s}s', 'stdout': '', 'stderr': ''} except FileNotFoundError: return {'ok': False, 'error': 'claude CLI no encontrado en PATH', 'stdout': '', 'stderr': ''} def shared_state_take(self, claim_id: str, agent_id: str) -> bool: """Intenta tomar un claim del shared_state. Útil para parallel clusters. Returns True si pudo tomar, False si ya estaba tomado. """ if not self.shared_state_path: return True # sin cluster, siempre puede try: from orchestrator_v2 import SharedState # type: ignore except ImportError: ssp = self.shared_state_path ssp.parent.mkdir(parents=True, exist_ok=True) if not ssp.exists(): ssp.write_text(json.dumps({'claims': {}, 'queries_in_progress': [], 'queries_done': []}, ensure_ascii=False, indent=2), encoding='utf-8') d = json.loads(ssp.read_text(encoding='utf-8')) if claim_id in d.get('queries_in_progress', []) or claim_id in d.get('queries_done', []): return False d.setdefault('queries_in_progress', []).append(claim_id) d.setdefault('claims', {}).setdefault(agent_id, []).append(claim_id) ssp.write_text(json.dumps(d, ensure_ascii=False, indent=2), encoding='utf-8') return True ss = SharedState(self.shared_state_path) return ss.take_query(agent_id, claim_id) # ============================================================ # REGISTRY — auto-descubre skills bajo skills/*.py # ============================================================ class SkillRegistry: """Carga y cachea skills bajo skills/.py. Cada skill DEBE exponer `run`.""" def __init__(self, skills_dir: Path = SKILLS_DIR): self.skills_dir = skills_dir self._cache: dict[str, Callable] = {} def discover(self) -> list[str]: """Devuelve lista de skill names disponibles.""" names = [] for p in self.skills_dir.glob('*.py'): if p.name.startswith('_') or p.stem == '__init__': continue names.append(p.stem) return sorted(names) def load(self, skill_name: str) -> Callable: """Carga una skill por nombre. Cachea.""" if skill_name in self._cache: return self._cache[skill_name] py_path = self.skills_dir / f'{skill_name}.py' if not py_path.exists(): raise SkillNotFound(f'Skill no encontrada: {skill_name} (esperaba {py_path})') spec = importlib.util.spec_from_file_location(f'skills.{skill_name}', py_path) if not spec or not spec.loader: raise SkillLoadError(f'No se pudo cargar spec de {py_path}') mod = importlib.util.module_from_spec(spec) sys.modules[f'skills.{skill_name}'] = mod try: spec.loader.exec_module(mod) except Exception as e: raise SkillLoadError(f'Error cargando {skill_name}: {e}') from e run_fn = getattr(mod, 'run', None) if not callable(run_fn): raise SkillLoadError(f'Skill {skill_name} no expone función run(inputs, context)') self._cache[skill_name] = run_fn return run_fn def invoke(self, skill_name: str, inputs: dict, context: SkillContext) -> SkillResult: """Invoca una skill. Convierte excepción en SkillResult con ok=False.""" try: run_fn = self.load(skill_name) except (SkillNotFound, SkillLoadError) as e: return SkillResult(ok=False, errors=[str(e)]) try: res = run_fn(inputs, context) except Exception as e: return SkillResult(ok=False, errors=[f'Skill {skill_name} crashed: {type(e).__name__}: {e}']) # Normalizar if isinstance(res, SkillResult): return res if isinstance(res, dict): return SkillResult( ok=bool(res.get('ok', False)), output=res.get('output', {}), errors=res.get('errors', []), metadata=res.get('metadata', {}), ) return SkillResult(ok=False, errors=[f'Skill {skill_name} devolvió tipo no esperado: {type(res).__name__}']) # ============================================================ # Excepciones # ============================================================ class SkillNotFound(Exception): pass class SkillLoadError(Exception): pass