# Estandar de Zonas — AroFlo Connector App **Archivo:** `apps/aroflo_connector_app/docs/AROFLO_ZONE_STANDARD.md` ## 0. Propósito y no‑objetivos ### Propósito Establecer un estándar **repetible y determinista** para implementar y mantener zonas (`zones//`) que consumen la API de AroFlo, usando una arquitectura por **zonas + secciones** y operaciones tipadas (`ZoneOperation`). El estándar está diseñado para: - Escalar a **30+ zonas** sin refactor masivo por zona. - Facilitar que, en futuros chats por zona, se implemente código **solo a partir de la documentación** (sin inventar campos o parámetros). - Permitir pruebas manuales consistentes (CLI determinista y agente barato con confirmación). ### No‑objetivos - No añadir features nuevas. - No cambiar la arquitectura general. - No romper compatibilidad con comportamientos ya validados en `tasks`. --- ## 1. Definiciones (términos del estándar) - **Zona (Zone):** Unidad de integración alineada a un `zone` de AroFlo (ej: `tasks`, `users`, `clients`). Implementa un conjunto de operaciones (`operations`) y un método `execute()` que traduce operaciones a requests AroFlo. - **Sección (Section):** Submódulo dentro de una zona que implementa un subconjunto de operaciones. Ejemplos típicos: - `base.py` (lecturas principales) - `join_*.py` (lecturas relacionadas / joins) - `mutations.py` (escrituras / POST/PUT) - **Operación (ZoneOperation):** Descriptor declarativo de una acción soportada (ej: `get_tasks`, `create_task`). Incluye metadata para: - ejecución determinista (método HTTP, side_effect) - manifest de capacidades para agentes - políticas de seguridad (riesgo, confirmación) - **Manifest de capacidades:** JSON serializable que lista zonas y operaciones sin llamar a AroFlo. Lo consumen agentes o herramientas. - **Despacho (Dispatcher):** Lógica central que recibe `(zone_code, operation, params)` y ejecuta la zona. - **dry_run:** Modo seguro (especialmente para escrituras) que retorna previsualización sin ejecutar mutación real. - **raw / raw_wrap:** Modalidad de debug (cuando se implemente por zona) que puede devolver request meta (params/var_string). --- ## 2. Contratos “normativos” del Core Esta sección define el contrato base que todas las zonas deben respetar. ### 2.1. ParamSpec **Archivo:** `zones/base.py` `ParamSpec` describe un parámetro de operación. Campos normativos: - `name` (str): nombre del parámetro. - `type` (str): tipo lógico (`string`, `integer`, `boolean`, `array`, etc.). - `required` (bool): obligatorio o no. - `description` (str): descripción humana. - `enum` (list[str] | None): valores permitidos. Campo opcional retrocompatible: - `items_schema` (dict | None): **solo** si `type == "array"`. Se usa para describir el esquema del ítem. **Regla del estándar:** `ParamSpec` debe representar **solo lo que la documentación de AroFlo declara**. Si no está documentado, no se define. ### 2.2. ZoneOperation **Archivo:** `zones/base.py` `ZoneOperation` es el descriptor único por operación. Sirve tanto para ejecución (zonas/dispatcher) como para describir capacidades (manifest). Campos normativos: - `code` (str): identificador interno estable (ej: `get_tasks`, `create_task`). - `label` (str): nombre legible. - `description` (str): descripción. - `http_method` (str): `GET`, `POST`, `PUT`, etc. - `default_params` (dict): parámetros base por defecto. - `params` (list[ParamSpec]): especificación de parámetros. - `side_effect` (`read` | `write`): indica si modifica datos. - `idempotent` (bool): si repetir la operación es seguro. Metadata para agentes (normativa para el manifest): - `category` (str | None): categoría lógica. - `use_cases` (list[str]): casos típicos en lenguaje humano. - `examples` (list[dict]): ejemplos de prompts → params. - `risk_level` (`low` | `medium` | `high`): nivel de riesgo. - `requires_confirmation` (bool): si el agente debe solicitar confirmación explícita. **Reglas del estándar:** 1) `side_effect="write"` implica que el CLI/agent pueden forzar `dry_run=True` por política. 2) `requires_confirmation=True` se usa para que el agente barato aplique `confirm` antes de ejecutar. 3) `idempotent=False` se debe usar en mutaciones no repetibles (crear, insertar) según doc. ### 2.3. Zone (clase base) **Archivo:** `zones/base.py` Contrato mínimo: - `code`, `label`, `description` (atributos de clase o instancia). - `operations -> List[ZoneOperation]` (property abstracta). - `get_operation(op_code)` (resolver operación por código). - `execute(operation_code, params)` (abstracta): ejecuta una operación concreta. **Regla del estándar:** `execute()` recibe `operation_code: str` y `params: dict` (del dispatcher o CLIs). Las zonas no reciben objetos `ZoneOperation` para ejecutar. --- ## 3. Manifest de capacidades (Capabilities) **Archivo:** `services/capabilities.py` ### 3.1. Objetivo Construir un manifest de capacidades para todas las zonas registradas: - **No llama a AroFlo**. - Solo instancia zonas desde el registry y serializa sus operaciones. - Debe ser seguro en cualquier entorno. ### 3.2. Serialización - `_paramspec_to_dict(p: ParamSpec)` serializa `ParamSpec`. - Incluye `items_schema` solo si no es `None`. - `_operation_to_dict(op: ZoneOperation, fallback_category: str)` serializa `ZoneOperation`. - `category` usa `op.category` o `fallback_category=zone.code`. - Exporta `risk_level` y `requires_confirmation`. ### 3.3. `build_capabilities_manifest(client)` Itera `registry = build_registry(client)` y produce: ```json { "schema_version": "1.0", "app": "aroflo_connector_app", "zones": [ { "code": "tasks", "label": "Tasks", "description": "...", "operations": [ ... ] } ] } ``` **Regla del estándar:** El manifest refleja exactamente lo que el código declara en `zone.operations`. No debe inventar params ni operaciones. --- ## 4. Dispatcher (ejecución central) **Archivo:** `services/dispatcher.py` ### 4.1. `get_zones_overview(client)` Devuelve metadata para `GET /zones` usando `list_zones_metadata(client)`. ### 4.2. `dispatch_zone_query(client, zone_code, payload)` Contrato de payload: ```json { "operation": "get_users", "params": { "...": "..." } } ``` Reglas: - `operation` es obligatorio. - `params` por defecto `{}`. - Resuelve `zone = get_zone(client, zone_code)`. - Ejecuta `zone.execute(operation, params=params)`. Respuesta estándar del dispatcher: ```json { "zone": "tasks", "operation": "get_tasks", "params": { ... }, "result": { ... } } ``` --- ## 5. AroFloClient (contrato de requests) **Archivo:** `client.py` ### 5.1. Objetivo Cliente HTTP para la API de AroFlo: - Carga settings (`AroFloSettings.from_env()`). - Construye headers HMAC. - Ejecuta request y parsea JSON. ### 5.2. Parámetros repetibles (where/join/order) AroFlo permite parámetros repetidos. Se soporta `ParamsType`: - `dict` (casos simples) - `list[tuple[str,str]]` (repetidos) Helper normativo: - `_inject_multi_param(params, key, value)` - Si `params is None` → crea lista de tuplas. - Si `params` es lista de tuplas → append. - Si `params` es dict: - 1 valor → `params[key]=value` - múltiples valores → convierte dict a lista de tuplas y agrega repetidos. **Regla del estándar:** joins / where / order deben inyectarse sin perder repetidos. ### 5.3. Normalización de `order` AroFlo usa `order=`. El client normaliza alias: - `order_by`, `orderby` → `order`. - Consume `order_by/orderby` de `kwargs` para evitar que rompa `requests`. ### 5.4. `var_string` Si no se pasa, se deriva de: - `params` si existe - `data` si es `dict` (form) Nota: - Si `data` es `str` (XML), **no** se usa para var_string. ### 5.5. Errores - Lanza `AroFloError` si: - request falla (red) - respuesta no es JSON - status HTTP >= 400 o `status` del payload no es OK --- # Parte II — CLI Determinista (Capa 2) ## 6. Objetivo del CLI determinista **Archivos:** - `agent/det_cli.py` - `agent/det_policy.py` - `agent/det_runner.py` - `agent/parse_kv.py` Propósito: - Ejecutar una operación específica (`zone + op`) de forma controlada. - Aplicar políticas seguras (writes → `dry_run` por defecto). - Facilitar pruebas repetibles por zona. --- ## 7. parse_kv (coerción básica de parámetros CLI) **Archivo:** `agent/parse_kv.py` `parse_kv_items(items, lowercase_keys=True)`: - Acepta múltiples `key=value`. - Divide por el primer `=` (preserva valores con `==`, ej base64). - Intenta `json.loads(v)`; si falla, deja string. **Regla del estándar:** - Para soportar arrays tipo `taskids=["a","b"]`, el usuario debe pasar JSON válido (y parse_kv lo convierte a list). - `lowercase_keys` debe definirse por CLI según necesidad de compatibilidad. --- ## 8. DeterministicPolicy (política de seguridad) **Archivo:** `agent/det_policy.py` Campos: - `force_dry_run_for_writes` (default `True`) - `default_page_size` (default `10`) - `max_page_size` (default `50`) `apply_policy(op, params, policy)`: - Si `op.side_effect == "write"` y `force_dry_run_for_writes=True` → `dry_run=True` si no existe. - Manejo de paging: - Lee `pageSize` o `pagesize`. - Si no existe y `default_page_size` definido → set `pageSize`. - Si existe y excede `max_page_size` → clamp a `max_page_size`. **Regla del estándar:** - `pageSize` es la forma canónica preferida. - `pagesize` es aceptado como compatibilidad, pero debe converger a `pageSize` en parámetros efectivos. --- ## 9. det_runner (ejecución determinista) **Archivo:** `agent/det_runner.py` Responsabilidades: 1) Construir `AroFloClient` cargando `.env`. 2) `build_registry(client)`. 3) Validar `zone_code`. 4) Resolver operación por `op_code` (solo para meta/policy). 5) Aplicar `apply_policy(...)` → `effective_params`. 6) Ejecutar `zone.execute(op_code, params=effective_params)`. 7) Opcional: envolver en meta (`include_meta=True`). ### 9.1. Meta opcional Si `include_meta=True` retorna: ```json { "data": , "meta": { "zone": "...", "operation": "...", "side_effect": "read|write", "http_method": "GET|POST|...", "idempotent": true, "policy": { ... }, "effective_params": { ... } }, "request": { ... } // solo si raw_wrap } ``` `raw_wrap` (convención): si la operación devuelve `{"data":..., "meta":{"params":..., "var_string":...}}`, se extrae a `payload.request`. --- ## 10. det_cli (interfaz CLI) **Archivo:** `agent/det_cli.py` Comandos: - `ops --zone `: lista operaciones de la zona. - `run --zone --op --param key=value ...`: - Construye `DeterministicPolicy` (con defaults y límites). - Ejecuta `execute_deterministic(...)`. - Soporta `--meta`. - Para writes: - Por defecto fuerza `dry_run=True`. - `--force-no-dryrun` permite ejecutar sin `dry_run` (riesgo). **Regla del estándar:** - `--param` debe aceptar JSON válido para listas/objetos. - No se deben inventar claves; todo param debe venir de doc o pruebas del API. --- # Parte III — Agente Barato (Capa 3) ## 11. Objetivo del agente barato **Archivos:** - `agent/cheap/cheap_cli.py` - `agent/cheap/router.py` - `agent/cheap/plan.py` - `agent/cheap/zone_resolvers/.py` - `agent/pending_store.py` Propósito: - Convertir una pregunta en lenguaje natural (MVP) en un **Plan determinista**: `(zone_code, op_code, params)`. - **No inventar operaciones**: toda operación debe existir en `list_operations(zone)`. - Ejecutar el plan usando el mismo motor determinista (`execute_deterministic`). - Para escrituras (`side_effect=write`): - Ejecutar primero con `dry_run=True` forzado por policy. - Generar token de confirmación (`pending_store`). - Aplicar el cambio solo cuando el usuario ejecute `confirm `. --- ## 12. Modelo de plan (Plan) **Archivo:** `agent/cheap/plan.py` `Plan` es la estructura mínima que el router produce: - `zone_code` (str) - `op_code` (str) - `params` (dict) - `side_effect` (`read` | `write`) - `needs_confirmation` (bool) - `summary` (str) - `missing` (list[MissingParam]) Reglas: - `is_ready()` retorna `True` cuando `missing` está vacío. - `needs_confirmation=True` se debe activar en writes. --- ## 13. Router (selección de zona y operación) **Archivo:** `agent/cheap/router.py` Responsabilidades: 1) Elegir zona (hoy: heurística simple `_guess_zone`). 2) Detectar intent (`detect_intent(question)`), que retorna un `op_code` propuesto. 3) Consultar operaciones disponibles en runtime: - `available = _ops_set(zone)` donde internamente usa `list_operations(zone)`. 4) Delegar a un resolver por zona: - `zone_resolvers/tasks.py` produce un `Plan`. **Regla del estándar (no-inventar):** - Antes de retornar un plan final, el resolver debe validar `op_code in available_ops`; si no, debe caer a un read seguro (ej. `list_tasks`). --- ## 14. CLI del agente barato (cheap_cli) **Archivo:** `agent/cheap/cheap_cli.py` ### 14.1. Comando `ask` Uso: ```bash python -m apps.aroflo_connector_app.agent.cheap.cheap_cli ask "..." ``` Flujo normativo: 1) `build_plan(q)` produce `Plan`. 2) Extrae `key=value` incrustados en el texto (`extract_kv_params(q)`). 3) Combina params: - primero `plan.params` - luego KV del usuario sobreescribe 4) Aplica coerciones UX (sin alterar arquitectura de zonas): - `_coerce_params` (lista bracket, int/float básicos) - `_normalize_pagesize` (convierte `pagesize` → `pageSize`) 5) Si `plan.missing` no está vacío, muestra faltantes y aborta con exit code 2. 6) Ejecuta con política segura: - `policy = DeterministicPolicy(force_dry_run_for_writes=True, ...)` - `execute_deterministic(zone_code, op_code, params, policy, include_meta=--meta)` 7) Si el resultado contiene lista de tasks, intenta guardarlas como contexto (`remember_tasks_if_present`). 8) Si el plan es escritura (`plan.needs_confirmation=True`): - Guarda payload en `pending_store.save_pending(payload)`. - Imprime instrucción de confirmación. **Regla del estándar (writes):** `ask` nunca debe ejecutar write real. Debe forzar `dry_run=True` por policy. ### 14.2. Comando `confirm` Uso: ```bash python -m apps.aroflo_connector_app.agent.cheap.cheap_cli confirm "" ``` Flujo normativo: 1) `load_pending()`. 2) `match_token(token, pending)`. 3) Si no existe o expiró → exit code 2. 4) Reconstruye params y normaliza: - `_coerce_params` - `_normalize_pagesize` 5) Ejecuta con policy que permite writes: - `DeterministicPolicy(force_dry_run_for_writes=False, ...)` 6) En `finally`, limpia el pending con `clear_pending()`. --- ## 15. Pending Store (confirmaciones) **Archivo:** `agent/pending_store.py` Objetivo: - Persistir una sola acción pendiente en archivo JSON (por defecto `/tmp/aroflo_agent_pending.json`). Contrato: - `save_pending(payload, ttl_seconds=DEFAULT_TTL_SECONDS)` retorna: - `token` - `created_at` - `expires_at` - `payload` - `load_pending()`: - Si expiró, borra el archivo y retorna `None`. - `match_token(token, pending)`: - Coincidencia exacta. - `clear_pending()`: - Borra el archivo. Reglas: - TTL por defecto: `AROFLO_AGENT_PENDING_TTL_SECONDS` (default 900s). - Path configurable: `AROFLO_AGENT_PENDING_PATH`. --- ## 16. Resolvers por zona **Archivo ejemplo:** `agent/cheap/zone_resolvers/tasks.py` Cada zona soportada por el agente barato debe tener un resolver dedicado, con contrato: ```py def build_plan(*, question: str, intent: str, available_ops: Set[str]) -> Plan: ... ``` Reglas normativas del resolver: 1) **No inventar operaciones:** - Usa `available_ops` como fuente de verdad. - Si `intent` no está disponible, fallback a `list_`. 2) Debe construir `missing: List[MissingParam]` para parámetros obligatorios. 3) Debe definir `side_effect` y `needs_confirmation` con base en un conjunto `WRITE_OPS`. 4) Debe permitir `key=value` solo para claves permitidas por operación (allowlist). 5) Debe normalizar `pageSize/pagesize`. 6) Debe implementar UX opcional pero segura: - `task #N` → resuelve `taskid` desde un snapshot previo guardado por `context_bridge`. - joins: si `join` está presente, elegir la operación `get__with_` si existe. Ejemplo en `tasks.py`: - `JOINABLE`: conjunto de joins soportados. - `_choose_join_op(join, available_ops)` retorna op joinable si existe. - Allowlists por operación: `READ_KEYS`, `CREATE_KEYS`, `UPDATE_KEYS`, etc. - Required fields de `create_task`: `orgid`, `clientid`, `tasktypeid`, `taskname`, `duedate`. --- # Parte IV — Ejecución de Tools y Schema (Capa 3.5) ## 17. tool_schema: nombres de tools y mapeo estable **Archivo:** `agent/tool_schema.py` Objetivo: - Generar herramientas tipo function-calling a partir del manifest. - Mantener nombres de tool <= 64 chars. - Mantener un mapa global: - `TOOL_NAME_MAP[tool_name] = (zone_code, op_code)` Reglas: - `build_tools_from_manifest(manifest)`: - Limpia `TOOL_NAME_MAP` al inicio. - Para cada `op` crea un `tool_name` con `_short_tool_name(zone, op)`. - Si el op es largo, trunca y agrega hash SHA1 corto. - Arrays: - Si `ParamSpec.type == "array"`, se exporta `items` desde `items_schema` si existe; si no, `{}`. - `additionalProperties=False` para evitar args inventados por el modelo. ## 18. executor: ejecución directa de tools **Archivo:** `agent/executor.py` Objetivo: - Ejecutar una tool resolviendo `(zone, op)` y delegando a `zone.execute(op_code, params)`. Reglas: 1) Coerción defensiva de args: - `_coerce_args` acepta dict o JSON string. 2) Resolución de tool_name: - Primero usa `TOOL_NAME_MAP`. - Si no existe, intenta parsers legacy: - `aroflo____` - `__` - `aroflo..` - `.` - `_` 3) Normalización defensiva de `where`: - `_normalize_where_clause` intenta convertir variaciones a un formato consistente con `;` y prefijos `and|`. 4) `execute_tool_call` retorna un wrapper: ```json { "ok": true, "tool_name": "...", "zone": "tasks", "operation": "list_tasks", "requires_confirmation": false, "data": { ... } } ``` 5) Confirmaciones: - `executor.py` no aplica confirmación. Solo expone `requires_confirmation` a nivel informativo. - La confirmación se controla por el flujo del agente (pendings/tokens). --- # Estándar de Arquitectura AroFlo Connector — Capas 3 y 4 Versión: 2026-01-03 Ámbito: `apps/aroflo_connector_app/` (Flask Server) Este documento describe el **estándar** para: - **Capa 3 (Agent / Cheap Agent):** enrutamiento determinista “barato” (sin LLM) para construir un `Plan` y ejecutar operaciones de zona con política segura (dry-run forzado en escrituras, confirmación con token). - **Capa 4 (Zones):** implementación por zona (ej. `tasks`) con secciones (`base`, `join_*`, `mutations`) y contratos estables (ops + params + ejecución). > Nota: Los **anexos** contienen el código completo de `tasks/mutations.py` y joins. > Archivo de anexos: `aroflo_standard_capa3_capa4_anexos.md`. --- ## 1) Objetivo y principios ### 1.1 Objetivo Estandarizar cómo se agregan nuevas zonas (≈30+) sin refactorizar cada vez: - La lógica determinista se queda “arriba” (Agent). - La lógica “AroFlo-specífica” se encapsula en Zones. - El Agent **no inventa** operaciones: valida contra `list_operations()`. ### 1.2 Principios de diseño 1. **No inventar operaciones:** el Agent valida op contra `available_ops`. 2. **Separación de responsabilidades:** - Zone: define ops, params, execute (GET/POST + postxml). - Cheap Agent: entiende intención mínima, arma `Plan`, coerciona params, ejecuta determinísticamente. 3. **Seguridad por defecto:** - Escrituras => `dry_run=True` forzado + `confirm` por token. 4. **UX tolerante:** - `pagesize` (minúscula) se normaliza a `pageSize`. - `taskids=[a,b]` o `taskids=a,b` se coerciona a lista. - `notes="texto"` se vuelve `[{content: "..."}]` (MVP). 5. **Extensibilidad por zona:** - Cada zona debe tener su `zone_resolvers/.py` que traduzca texto->Plan, sin tocar las otras zonas. --- ## 2) Capa 3 — Cheap Deterministic Agent (visión general) ### 2.1 Flujo (ask -> execute -> confirm) 1. `cheap_cli ask ""`: - `build_plan()` decide `zone_code`, `op_code` y `params` (mínimos). - Extrae `key=value` del texto para sobrescribir/inyectar params. - Coerciona params (ids, listas, numbers, bool). - Aplica política: si es write => `dry_run=True` forzado. - Ejecuta `execute_deterministic(...)`. - Si es write => guarda payload en `pending_store` y muestra `confirm ""`. 2. `cheap_cli confirm ""`: - Carga payload y ejecuta con `force_dry_run_for_writes=False`. - Limpia el pending store siempre (finally). ### 2.2 Archivos y responsabilidades #### 2.2.1 `agent/cheap/cheap_cli.py` - CLI determinista: - `ask` => produce plan, ejecuta, y si es escritura genera token. - `confirm` => aplica payload “real” (dry_run desactivado). - Coerciones clave: - `_normalize_pagesize()` - `_coerce_params()` + `_coerce_param_by_name()` (ids, lists, notes/materials JSON-like) #### 2.2.2 `agent/cheap/plan.py` - Dataclasses: - `Plan`: `zone_code`, `op_code`, `params`, `side_effect`, `needs_confirmation`, `missing`. #### 2.2.3 `agent/cheap/router.py` - `build_plan(question)`: - `_guess_zone()` (por ahora tasks por keyword/prefix). - `detect_intent()` (extractor simple). - `available_ops = list_operations(zone)` para evitar op inventada. - Delegación a `zone_resolvers/.py`. #### 2.2.4 `agent/pending_store.py` - Guarda acción pendiente con TTL (default 15 min): - `save_pending(payload)` => `{token, created_at, expires_at, payload}` - `match_token(token, pending)` y `clear_pending()` #### 2.2.5 `agent/executor.py` - Ejecuta una tool (zona+op) resolviendo nombre largo/corto con `TOOL_NAME_MAP`. - Normaliza `where` pipe-syntax defensivamente. - Ojo: la confirmación NO se maneja acá, se maneja en `cheap_cli.py`. #### 2.2.6 `agent/tool_schema.py` - Genera tools desde el manifest y crea `TOOL_NAME_MAP` para nombres <= 64 chars. --- ## 3) Capa 4 — Zones (contrato estándar) ### 3.1 Estructura recomendada por zona Ejemplo `zones/tasks/`: - `__init__.py` (Zone principal + SECTIONS) - `base.py` (ops de lectura base: list/get/due_range) - `mutations.py` (ops de escritura: create/update/notes/materials/etc.) - `join_*.py` (una sección por join, solo lectura) - `_join_utils.py` (helpers compartidos: request/raw_wrap/coerce/build_list_params) - `cli.py` (opcional pero recomendado) CLI “humana” para pruebas manuales > Cada archivo de sección debe implementar: - `get_operations() -> List[ZoneOperation]` - `supports(operation_code: str) -> bool` - `execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any` ### 3.2 Dispatcher de zona: `zones//__init__.py` Patrón: - Mantener un arreglo `SECTIONS = [base_section, join_x, ..., mutations_section]` - `operations`: concatena `get_operations()` de todas las secciones. - `execute(op, params)`: busca `supports(op)` y delega `execute`. ### 3.3 READ base: `base.py` Convenciones: - `OP_LIST`, `OP_GET`, `OP_DUE_RANGE` - `default_params`: incluir `raw`, `page`, `pageSize`, `where`, `order` - `request(client, "GET", params_list)` siempre con `var_string` alineado. ### 3.4 Mutations (WRITE): `mutations.py` Convenciones: - `requires_confirmation=True` en `ZoneOperation` de escrituras. - Soportar `dry_run`: - Si `dry_run=True`, devolver preview `{dry_run, http_method, params, join, postxml}` sin ejecutar POST. - POST siempre debe respetar HMAC: - body x-www-form-urlencoded debe coincidir con `var_string`. - “Zone fallback”: - Algunos tenants usan `zone=Tasks` vs `zone=tasks` => reintento alterno si el error sugiere “zone”. ### 3.5 Joins: `join_*.py` Convenciones: - `JOIN_NAME` + `OP_CODE = f"get_tasks_with_{JOIN_NAME}"` - Reusar `_join_utils.build_list_params(...)` - Params consistentes: `where, order, page, pageSize, raw`. --- ## 4) ¿Es necesario `zones/tasks/cli.py`? ### 4.1 Recomendación Sí, **es altamente recomendado** y en la práctica “clave” para: - Pruebas manuales deterministas sin Agent. - Soportar debugging cuando el Agent falle o aún no soporte un intent. - Validar rápidamente que: - el `postxml` está bien, - `zone` fallback funciona, - el `join` correcto devuelve lo esperado. ### 4.2 Relación con el Cheap Agent - El Cheap Agent no depende del CLI de zona. - Pero el CLI de zona **reduce fricción** en QA: primero validas zona+op en CLI, luego agregas soporte al resolver del Agent. **Conclusión:** Mantener `zones//cli.py` como parte del estándar (aunque sea “opcional”). --- ## 5) Registro de zonas: `zones/registry.py` (clave) Sí: este archivo es **obligatorio** para que una zona exista en runtime. ### 5.1 Regla - Toda nueva zona debe: 1) tener `Zones//` implementada, 2) exportar una `FooZone(Zone)`, 3) registrarse en `build_registry()`. Ejemplo (patrón actual): ```py from .tasks import TasksZone def build_registry(client): return { ... "tasks": TasksZone(client), } ``` ### 5.2 Checklist al agregar una zona nueva 1. Crear carpeta `zones//` con: - `__init__.py` (Zone + SECTIONS) - `base.py` (al menos list/get) - `_join_utils.py` (o reutilizar util común si lo centralizas después) - `cli.py` (recomendado) 2. Registrar `NewZone` en `zones/registry.py`. 3. Agregar resolver: `agent/cheap/zone_resolvers/.py`. 4. Extender `agent/cheap/router.py`: - `_guess_zone()` detectar keywords/prefix. - `if zone == "": return _resolver.build_plan(...)` --- ## 6) Anexos (código completo) Ver: `aroflo_standard_capa3_capa4_anexos.md` --- # Anexos — Código completo (Tasks Zone) Versión: 2026-01-03 Este documento incluye el **código completo** de las piezas que suelen ser largas, para que puedas empalmarlo en `docs/` sin placeholders. --- ## A1) `zones/tasks/mutations.py` (completo) ```py from __future__ import annotations from typing import Any, Dict, List, Tuple, Optional from ..base import ZoneOperation, ParamSpec from ._join_utils import request, raw_wrap import json import re # ------------------------- # Operation codes # ------------------------- OP_CREATE = "create_task" OP_UPDATE = "update_task" OP_MARK_LINKPROCESSED = "mark_task_linkprocessed" OP_INSERT_NOTES = "insert_task_notes" OP_INSERT_MATERIALS = "insert_task_adhoc_materials" OP_UPDATE_SUBSTATUS = "update_task_substatus" OP_UPDATE_PROJECT_STAGE = "update_task_project_stage" # ------------------------- # Zone names # ------------------------- # Nota: en ejemplos/documentación AroFlo aparece a veces "Tasks" y a veces "tasks". # Algunos tenants pueden ser sensibles. Implementamos fallback seguro por error de zona. AF_ZONE_TASKS_TITLE = "Tasks" AF_ZONE_TASKS_LOWER = "tasks" # ------------------------- # Join names # ------------------------- # GET joins (lectura) suelen usar "material" (singular) en tasks join=material AF_JOIN_MATERIALS_GET = "material" # POST para insertar AdHoc material, la doc suele indicar join=materials (plural) AF_JOIN_MATERIALS_POST = "materials" # ------------------------- # Helpers # ------------------------- def _cdata(s: Optional[str]) -> str: if s is None: return "" return f"" def _normalize_taskname(params: Dict[str, Any]) -> str: """ Permite que el CLI use --summary y nosotros lo traduzcamos a taskname. También soporta taskname directamente si lo envían. """ taskname = params.get("taskname") if taskname is None: taskname = params.get("summary") # alias CLI if taskname is None or not str(taskname).strip(): raise ValueError("taskname/summary no puede estar vacío.") return str(taskname) _LIST_TOKEN_RE = re.compile(r"[,\s]+") def _coerce_str_list(value: Any, *, field_name: str) -> List[str]: """ Acepta: - list[str] - "id1" (single) - "[id1]" (string que parece lista) - '["id1","id2"]' (json) - "id1,id2" (csv) - "[id1,id2]" (brackets sin comillas) """ if value is None: raise ValueError(f"{field_name} no puede ser vacío.") # Ya es lista if isinstance(value, list): out = [str(x).strip() for x in value if str(x).strip()] if not out: raise ValueError(f"{field_name} no puede ser una lista vacía.") return out # Si viene como string if isinstance(value, str): s = value.strip() # JSON list real: '["a","b"]' if s.startswith("[") and s.endswith("]"): # 1) intenta json.loads try: parsed = json.loads(s) if isinstance(parsed, list): out = [str(x).strip() for x in parsed if str(x).strip()] if not out: raise ValueError(f"{field_name} no puede ser una lista vacía.") return out except Exception: # 2) fallback: brackets sin comillas: "[a,b]" o "[a]" inner = s[1:-1].strip() if not inner: raise ValueError(f"{field_name} no puede ser una lista vacía.") tokens = [t.strip().strip('"').strip("'") for t in _LIST_TOKEN_RE.split(inner) if t.strip()] if not tokens: raise ValueError(f"{field_name} no puede ser una lista vacía.") return tokens # CSV "a,b" if "," in s: tokens = [t.strip() for t in s.split(",") if t.strip()] if not tokens: raise ValueError(f"{field_name} no puede ser vacío.") return tokens # single "a" if s: return [s] # Cualquier otro tipo raise ValueError(f"{field_name} debe ser una lista o un string convertible a lista.") def _build_postxml_create(p: Dict[str, Any]) -> str: clientid = p["clientid"] tasktypeid = p["tasktypeid"] taskname = _normalize_taskname(p) taskname = str(taskname).strip() if len(taskname) > 50: taskname = taskname[:50] orgid = p.get("orgid") if not orgid: # En tu tenant ya vimos que es requerido raise ValueError("orgid es requerido para crear tasks en este tenant.") duedate = p.get("duedate") # doc: YYYY-MM-DD o YYYY/MM/DD según tenant description = p.get("description") contactname = p.get("contactname") contactphone = p.get("contactphone") custon = p.get("custon") parts: List[str] = [""] # --- Forma 100% alineada con la doc (anidada) --- parts.append(f"{orgid}") parts.append(f"{clientid}") parts.append(f"{tasktypeid}") parts.append(f"{_cdata(taskname)}") if duedate: parts.append(f"{duedate}") if description: parts.append(f"{_cdata(str(description))}") if contactname: parts.append(f"{_cdata(str(contactname))}") if contactphone: parts.append(f"{_cdata(str(contactphone))}") if custon: parts.append(f"{_cdata(str(custon))}") parts.append("") return "".join(parts) def _build_postxml_update(p: Dict[str, Any]) -> str: taskid = p["taskid"] taskname = p.get("taskname") if taskname is None: taskname = p.get("summary") # alias CLI status = p.get("status") if taskname is None and status is None: raise ValueError("Debes enviar al menos taskname/summary o status para actualizar.") parts: List[str] = ["", f"{taskid}"] if taskname is not None: if not str(taskname).strip(): raise ValueError("taskname/summary no puede ser vacío si se envía.") parts.append(f"{_cdata(str(taskname))}") if status is not None: if not str(status).strip(): raise ValueError("status no puede ser vacío si se envía.") parts.append(f"{_cdata(str(status))}") parts.append("") return "".join(parts) def _build_postxml_linkprocessed(taskids: List[str]) -> str: if not taskids: raise ValueError("taskids no puede estar vacío.") parts: List[str] = [""] for tid in taskids: parts.append(f"{tid}true") parts.append("") return "".join(parts) def _build_postxml_notes(taskid: str, notes: List[Dict[str, Any]]) -> str: if not notes: raise ValueError("notes no puede estar vacío.") parts: List[str] = [f"{taskid}"] for n in notes: content = n.get("content") if not content or not str(content).strip(): raise ValueError("Cada note debe tener content.") flt = n.get("filter") # ej: "internal only" sticky = n.get("sticky", False) parts.append("") parts.append(f"{_cdata(str(content))}") if flt is not None: parts.append(f"{_cdata(str(flt))}") parts.append(f"{'true' if bool(sticky) else 'false'}") parts.append("") parts.append("") return "".join(parts) def _build_postxml_materials(taskid: str, materials: List[Dict[str, Any]]) -> str: if not materials: raise ValueError("materials no puede estar vacío.") parts: List[str] = [f"{taskid}"] for m in materials: item = m.get("item") if not item or not str(item).strip(): raise ValueError("Cada material debe incluir item.") parts.append("") if m.get("partnumber"): parts.append(f"{_cdata(str(m['partnumber']))}") parts.append(f"{_cdata(str(item))}") if m.get("cost") is not None: parts.append(f"{m['cost']}") if m.get("sell") is not None: parts.append(f"{m['sell']}") if m.get("dateused") is not None: parts.append(f"{m['dateused']}") # doc: YYYY/MM/DD if m.get("quantity") is not None: parts.append(f"{m['quantity']}") parts.append("") parts.append("") return "".join(parts) def _build_postxml_update_substatus(taskid: str, status: str, substatusid: str) -> str: return ( "" f"{taskid}" f"{_cdata(status)}" f"{substatusid}" "" ) def _build_postxml_update_project_stage(taskid: str, projectid: Optional[str], stageid: Optional[str]) -> str: if not projectid and not stageid: raise ValueError("Debes enviar projectid y/o stageid.") parts = ["", f"{taskid}"] if projectid: parts.append(f"{projectid}") if stageid: parts.append(f"{stageid}") parts.append("") return "".join(parts) def _is_zone_error(resp: Any) -> bool: """ Detecta error de "zone" de forma defensiva, sin asumir formato exacto. """ try: if isinstance(resp, dict): # Campos comunes vistos: statusmessage, error, message for k in ("statusmessage", "error", "message"): v = resp.get(k) if isinstance(v, str) and "zone" in v.lower(): return True if isinstance(resp, str) and "zone" in resp.lower(): return True except Exception: return False return False def _post_with_zone_fallback( client: Any, *, primary_zone: str, alternate_zone: str, postxml: str, join: Optional[str] = None, ) -> Any: """ Ejecuta POST y si falla por error de zona, reintenta una sola vez alternando el zone. """ params_list: List[Tuple[str, str]] = [("zone", primary_zone)] if join: params_list.append(("join", join)) params_list.append(("postxml", postxml)) resp = request(client, "POST", params_list) if _is_zone_error(resp) and alternate_zone and alternate_zone != primary_zone: params_list2: List[Tuple[str, str]] = [("zone", alternate_zone)] if join: params_list2.append(("join", join)) params_list2.append(("postxml", postxml)) return request(client, "POST", params_list2) return resp # ------------------------- # Operations registry # ------------------------- def get_operations() -> List[ZoneOperation]: ops: List[ZoneOperation] = [ ZoneOperation( code=OP_CREATE, label="Create Task", description="Crea una nueva task (POST zone=Tasks|tasks) usando postxml (fallback automático por error de zone).", http_method="POST", side_effect="write", idempotent=False, default_params={ "raw": False, "dry_run": False, "orgid": None, "clientid": None, "tasktypeid": None, "taskname": None, # o summary (alias) "summary": None, # alias "duedate": None, # YYYY/MM/DD "description": None, "contactname": None, "contactphone": None, "custon": None, }, params=[ ParamSpec("orgid", "string", False, "OrgID (opcional pero recomendado)."), ParamSpec("clientid", "string", True, "ClientID requerido por AroFlo."), ParamSpec("tasktypeid", "string", True, "TaskTypeID requerido."), ParamSpec("taskname", "string", False, "Nombre de la task (taskname)."), ParamSpec("summary", "string", False, "Alias de taskname (para CLI)."), ParamSpec("duedate", "string", False, "Fecha vencimiento (YYYY/MM/DD)."), ParamSpec("description", "string", False, "Descripción."), ParamSpec("contactname", "string", False, "Nombre contacto."), ParamSpec("contactphone", "string", False, "Tel contacto."), ParamSpec("custon", "string", False, "Campo custon (según doc)."), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=["Crear task de prueba", "Crear task desde automatización"], risk_level="high", requires_confirmation=True, ), ZoneOperation( code=OP_UPDATE, label="Update Task", description="Actualiza una task existente por taskid (POST zone=Tasks|tasks) usando postxml (fallback por zone).", http_method="POST", side_effect="write", idempotent=False, default_params={ "raw": False, "dry_run": False, "taskid": None, "taskname": None, "summary": None, "status": None, }, params=[ ParamSpec("taskid", "string", True, "TaskID a actualizar."), ParamSpec("taskname", "string", False, "Nuevo taskname."), ParamSpec("summary", "string", False, "Alias de taskname (para CLI)."), ParamSpec("status", "string", False, "Nuevo status (ej: Pending)."), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=["Cambiar status", "Actualizar nombre"], risk_level="high", requires_confirmation=True, ), ZoneOperation( code=OP_MARK_LINKPROCESSED, label="Mark Task as linkprocessed", description="Marca linkprocessed=true en una o múltiples tasks (POST zone=Tasks|tasks) con fallback por zone.", http_method="POST", side_effect="write", idempotent=False, default_params={"raw": False, "dry_run": False}, params=[ ParamSpec("taskids", "array", True, "Lista de taskid a marcar linkprocessed=true.", items_schema={"type": "string"}), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=["Marcar tasks procesadas por sistema externo"], risk_level="medium", requires_confirmation=True, ), ZoneOperation( code=OP_INSERT_NOTES, label="Insert Notes to Task", description="Inserta una o múltiples notas en una task (POST zone=tasks|Tasks) con fallback por zone.", http_method="POST", side_effect="write", idempotent=False, default_params={"raw": False, "dry_run": False}, params=[ ParamSpec("taskid", "string", True, "TaskID destino."), ParamSpec( "notes", "array", True, "Lista de notas: [{content, filter, sticky}]", items_schema={ "type": "object", "properties": { "content": {"type": "string", "description": "Contenido de la nota."}, "filter": {"type": "string", "description": "Filtro (ej: internal only)."}, "sticky": {"type": "boolean", "description": "Si la nota es sticky."}, }, "required": ["content"], "additionalProperties": False, }, ), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Debug raw."), ], category="tasks", risk_level="medium", requires_confirmation=True, ), ZoneOperation( code=OP_INSERT_MATERIALS, label="Insert AdHoc material to Task", description=( "Inserta materiales AdHoc a una task (POST zone=tasks|Tasks join=materials). " "Nota: GET suele usar join=material, pero para este POST la doc suele pedir join=materials." ), http_method="POST", side_effect="write", idempotent=False, default_params={"raw": False, "dry_run": False}, params=[ ParamSpec("taskid", "string", True, "TaskID destino."), ParamSpec( "materials", "array", True, "Lista de materials AdHoc.", items_schema={ "type": "object", "properties": { "partnumber": {"type": "string", "description": "Part number (opcional)."}, "item": {"type": "string", "description": "Nombre/ítem del material."}, "cost": {"type": "number", "description": "Costo (opcional)."}, "sell": {"type": "number", "description": "Precio de venta (opcional)."}, "dateused": {"type": "string", "description": "Fecha uso (YYYY/MM/DD)."}, "quantity": {"type": "number", "description": "Cantidad (opcional)."}, }, "required": ["item"], "additionalProperties": False, }, ), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Debug raw."), ], category="tasks", risk_level="high", requires_confirmation=True, ), ZoneOperation( code=OP_UPDATE_SUBSTATUS, label="Update Task Substatus", description="Actualiza status y substatusid de una task (POST zone=tasks|Tasks) con fallback por zone.", http_method="POST", side_effect="write", idempotent=False, default_params={"raw": False, "dry_run": False}, params=[ ParamSpec("taskid", "string", True, "TaskID destino."), ParamSpec("status", "string", True, "Status (ej: pending)."), ParamSpec("substatusid", "string", True, "SubstatusID."), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Debug raw."), ], category="tasks", risk_level="high", requires_confirmation=True, ), ZoneOperation( code=OP_UPDATE_PROJECT_STAGE, label="Update Task Project/Stage", description="Actualiza projectid y/o stageid de una task (POST zone=Tasks|tasks) con fallback por zone.", http_method="POST", side_effect="write", idempotent=False, default_params={"raw": False, "dry_run": False}, params=[ ParamSpec("taskid", "string", True, "TaskID destino."), ParamSpec("projectid", "string", False, "ProjectID."), ParamSpec("stageid", "string", False, "StageID."), ParamSpec("dry_run", "boolean", False, "Si true, no ejecuta POST; devuelve preview (postxml/params)."), ParamSpec("raw", "boolean", False, "Debug raw."), ], category="tasks", risk_level="high", requires_confirmation=True, ), ] return ops def supports(operation_code: str) -> bool: return operation_code in { OP_CREATE, OP_UPDATE, OP_MARK_LINKPROCESSED, OP_INSERT_NOTES, OP_INSERT_MATERIALS, OP_UPDATE_SUBSTATUS, OP_UPDATE_PROJECT_STAGE, } # ------------------------- # Execute # ------------------------- def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: raw = bool(params.get("raw", False)) dry_run = bool(params.get("dry_run", False)) def _preview(params_list: List[Tuple[str, str]], postxml: str, join: Optional[str] = None) -> Dict[str, Any]: return { "dry_run": True, "http_method": "POST", "params": params_list, "join": join, "postxml": postxml, } if operation_code == OP_CREATE: postxml = _build_postxml_create(params) params_list: List[Tuple[str, str]] = [("zone", AF_ZONE_TASKS_TITLE), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_TITLE, alternate_zone=AF_ZONE_TASKS_LOWER, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_UPDATE: postxml = _build_postxml_update(params) params_list = [("zone", AF_ZONE_TASKS_TITLE), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_TITLE, alternate_zone=AF_ZONE_TASKS_LOWER, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_MARK_LINKPROCESSED: taskids = _coerce_str_list(params.get("taskids"), field_name="taskids") postxml = _build_postxml_linkprocessed(taskids) params_list = [("zone", AF_ZONE_TASKS_TITLE), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_TITLE, alternate_zone=AF_ZONE_TASKS_LOWER, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_INSERT_NOTES: taskid = params["taskid"] notes = params["notes"] if not isinstance(notes, list): raise ValueError("notes debe ser una lista.") postxml = _build_postxml_notes(taskid, notes) params_list = [("zone", AF_ZONE_TASKS_LOWER), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_LOWER, alternate_zone=AF_ZONE_TASKS_TITLE, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_INSERT_MATERIALS: taskid = params["taskid"] materials = params["materials"] if not isinstance(materials, list): raise ValueError("materials debe ser una lista.") postxml = _build_postxml_materials(taskid, materials) params_list = [("zone", AF_ZONE_TASKS_LOWER), ("join", AF_JOIN_MATERIALS_POST), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml, join=AF_JOIN_MATERIALS_POST) resp = _post_with_zone_fallback( client, primary_zone=AF_ZONE_TASKS_LOWER, alternate_zone=AF_ZONE_TASKS_TITLE, postxml=postxml, join=AF_JOIN_MATERIALS_POST, ) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_UPDATE_SUBSTATUS: postxml = _build_postxml_update_substatus(params["taskid"], params["status"], params["substatusid"]) params_list = [("zone", AF_ZONE_TASKS_LOWER), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_LOWER, alternate_zone=AF_ZONE_TASKS_TITLE, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp if operation_code == OP_UPDATE_PROJECT_STAGE: postxml = _build_postxml_update_project_stage(params["taskid"], params.get("projectid"), params.get("stageid")) params_list = [("zone", AF_ZONE_TASKS_TITLE), ("postxml", postxml)] if dry_run: return _preview(params_list, postxml) resp = _post_with_zone_fallback(client, primary_zone=AF_ZONE_TASKS_TITLE, alternate_zone=AF_ZONE_TASKS_LOWER, postxml=postxml) return raw_wrap(resp, params_list) if raw else resp raise ValueError(f"[Tasks.mutations] Operación no soportada: {operation_code}") ``` --- ## A2) `zones/tasks/join_notes.py` (completo) ```py from __future__ import annotations from typing import Any, Dict, List from ..base import ZoneOperation, ParamSpec from ._join_utils import request, raw_wrap, coerce_page_size, coerce_order, build_list_params JOIN_NAME = "notes" OP_CODE = "get_tasks_with_notes" def get_operations() -> List[ZoneOperation]: return [ ZoneOperation( code=OP_CODE, label="Get Tasks with Notes", description=( "Devuelve tasks (zona=tasks) incluyendo el join 'notes'. " "Útil para consultar notas asociadas a cada task." ), http_method="GET", side_effect="read", idempotent=True, default_params={ "where": "and|createdutc|>|2001-01-01", "order": None, "page": 1, "pageSize": None, "raw": False, }, params=[ ParamSpec("where", "string", False, "Cláusula WHERE estilo AroFlo."), ParamSpec("order", "string", False, "ORDER estilo AroFlo. Ej: createdutc|desc"), ParamSpec("page", "integer", False, "Número de página (1..N)."), ParamSpec("pageSize", "integer", False, "Tamaño de página (AroFlo 'pageSize')."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=[ "Consultar tasks con notas", "Auditar historial de comentarios/observaciones", ], risk_level="low", requires_confirmation=False, ) ] def supports(operation_code: str) -> bool: return operation_code == OP_CODE def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: if operation_code != OP_CODE: raise ValueError(f"[Tasks.join_notes] Operación no soportada: {operation_code}") raw = bool(params.get("raw", False)) where = params.get("where", "and|createdutc|>|2001-01-01") order = coerce_order(params) page = int(params.get("page", 1)) if page < 1: raise ValueError("page debe ser >= 1.") pageSize = coerce_page_size(params) params_list = build_list_params( zone="tasks", where=where, join=JOIN_NAME, order=order, page=page, pageSize=pageSize, ) resp = request(client, "GET", params_list) return raw_wrap(resp, params_list) if raw else resp ``` --- ## A3) `zones/tasks/join_material.py` (completo) ```py from __future__ import annotations from typing import Any, Dict, List from ..base import ZoneOperation, ParamSpec from ._join_utils import request, raw_wrap, coerce_page_size, coerce_order, build_list_params JOIN_NAME = "material" OP_CODE = "get_tasks_with_material" def get_operations() -> List[ZoneOperation]: return [ ZoneOperation( code=OP_CODE, label="Get Tasks with Material", description=( "Devuelve tasks (zona=tasks) incluyendo el join 'material'. " "Útil para ver materiales/consumibles asociados al task." ), http_method="GET", side_effect="read", idempotent=True, default_params={ "where": "and|createdutc|>|2001-01-01", "order": None, "page": 1, "pageSize": None, "raw": False, }, params=[ ParamSpec("where", "string", False, "Cláusula WHERE estilo AroFlo."), ParamSpec("order", "string", False, "ORDER estilo AroFlo. Ej: createdutc|desc"), ParamSpec("page", "integer", False, "Número de página (1..N)."), ParamSpec("pageSize", "integer", False, "Tamaño de página (AroFlo 'pageSize')."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=[ "Listar tasks con materiales", "Auditar materiales por rango/cliente (con WHERE)", ], risk_level="low", requires_confirmation=False, ) ] def supports(operation_code: str) -> bool: return operation_code == OP_CODE def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: if operation_code != OP_CODE: raise ValueError(f"[Tasks.join_material] Operación no soportada: {operation_code}") raw = bool(params.get("raw", False)) where = params.get("where", "and|createdutc|>|2001-01-01") order = coerce_order(params) page = int(params.get("page", 1)) if page < 1: raise ValueError("page debe ser >= 1.") pageSize = coerce_page_size(params) params_list = build_list_params( zone="tasks", where=where, join=JOIN_NAME, order=order, page=page, pageSize=pageSize, ) resp = request(client, "GET", params_list) return raw_wrap(resp, params_list) if raw else resp ``` --- ## A4) `zones/tasks/join_documentsandphotos.py` (completo) ```py from __future__ import annotations from typing import Any, Dict, List from ..base import ZoneOperation, ParamSpec from ._join_utils import request, raw_wrap, coerce_page_size, coerce_order, build_list_params JOIN_NAME = "documentsandphotos" OP_CODE = "get_tasks_with_documentsandphotos" def get_operations() -> List[ZoneOperation]: return [ ZoneOperation( code=OP_CODE, label="Get Tasks with Documents and Photos", description=( "Devuelve tasks (zona=tasks) incluyendo el join 'documentsandphotos'. " "Útil para listar adjuntos/fotos asociados a cada task." ), http_method="GET", side_effect="read", idempotent=True, default_params={ "where": "and|createdutc|>|2001-01-01", "order": None, "page": 1, "pageSize": None, "raw": False, }, params=[ ParamSpec("where", "string", False, "Cláusula WHERE estilo AroFlo."), ParamSpec("order", "string", False, "ORDER estilo AroFlo. Ej: createdutc|desc"), ParamSpec("page", "integer", False, "Número de página (1..N)."), ParamSpec("pageSize", "integer", False, "Tamaño de página (AroFlo 'pageSize')."), ParamSpec("raw", "boolean", False, "Si true, devuelve respuesta cruda + meta debug."), ], category="tasks", use_cases=[ "Consultar tasks con documentos/fotos", "Auditar evidencia adjunta", "Revisar adjuntos por rango de fechas", ], risk_level="low", requires_confirmation=False, ) ] def supports(operation_code: str) -> bool: return operation_code == OP_CODE def execute(operation_code: str, client: Any, params: Dict[str, Any]) -> Any: if operation_code != OP_CODE: raise ValueError(f"[Tasks.join_documentsandphotos] Operación no soportada: {operation_code}") raw = bool(params.get("raw", False)) where = params.get("where", "and|createdutc|>|2001-01-01") order = coerce_order(params) page = int(params.get("page", 1)) if page < 1: raise ValueError("page debe ser >= 1.") pageSize = coerce_page_size(params) params_list = build_list_params( zone="tasks", where=where, join=JOIN_NAME, order=order, page=page, pageSize=pageSize, ) resp = request(client, "GET", params_list) return raw_wrap(resp, params_list) if raw else resp ``` --- ## A5) `zones/tasks/cli.py` (completo, recomendado) ```py from __future__ import annotations import json from typing import Any, Dict, List, Optional, Callable import click DEFAULT_WHERE = "and|createdutc|>|2001-01-01" def _echo(result: Any) -> None: if isinstance(result, (dict, list)): click.echo(json.dumps(result, indent=2, ensure_ascii=False)) else: click.echo(str(result)) def _available_ops(zone: Any) -> List[str]: ops = getattr(zone, "operations", []) return sorted([o.code for o in ops]) def _run(zone: Any, op_code: str, params: Dict[str, Any]) -> Any: available = set(_available_ops(zone)) if op_code not in available: click.echo(f"❌ Operación '{op_code}' no existe en zona '{zone.code}'.") click.echo("Operaciones disponibles:") for c in _available_ops(zone): click.echo(f" - {c}") raise SystemExit(2) return zone.execute(op_code, params=params) def _add_optional( params: Dict[str, Any], *, pagesize: Optional[int], order: Optional[str], ) -> Dict[str, Any]: """ AroFlo usa pageSize como parámetro de entrada para tamaño de página. """ if pagesize is not None: params["pageSize"] = pagesize if order: params["order"] = order return params def _common_list_options(fn: Callable[..., Any]) -> Callable[..., Any]: """ Decorador para opciones comunes de list/join: - page, where, order, pageSize (AroFlo) y raw """ fn = click.option("--page", default=1, type=int, show_default=True)(fn) fn = click.option("--where", default=DEFAULT_WHERE, show_default=True, help="Cláusula WHERE estilo AroFlo.")(fn) fn = click.option("--order", default=None, show_default=True, help="ORDER estilo AroFlo. Ej: createdutc|desc")(fn) fn = click.option( "--pagesize", type=int, default=None, show_default=True, help="Cantidad de registros por página (AroFlo pageSize). Ej: 5, 20, 50.", )(fn) fn = click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.")(fn) return fn def _require_confirm_if_needed(*, confirm: bool, dry_run: bool, op_label: str) -> None: if dry_run: return if not confirm: raise SystemExit(f"❌ Falta --confirm. Operación de escritura cancelada ({op_label}).") def register_cli(root: click.Group, zone: Any) -> None: @root.group(name=zone.code) def tasks_group(): """Operaciones de la zona tasks.""" # ------------------------- # BASE # ------------------------- @tasks_group.command("list") @_common_list_options def list_cmd( page: int, where: str, order: Optional[str], pagesize: Optional[int], raw: bool, ): params = _add_optional( {"page": page, "where": where, "raw": raw}, pagesize=pagesize, order=order, ) _echo(_run(zone, "list_tasks", params)) @tasks_group.command("get") @click.option("--taskid", required=True, help="TaskID codificado (AroFlo).") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") def get_cmd(taskid: str, raw: bool): _echo(_run(zone, "get_task", {"taskid": taskid, "raw": raw})) @tasks_group.command("due-range") @click.option("--daterangetype", required=True, default="DueDate", show_default=True) @click.option("--fromdate", required=True, help="YYYY-MM-DD") @click.option("--todate", required=True, help="YYYY-MM-DD") @click.option("--where", default=None, help="WHERE adicional (opcional).") @click.option("--order", default=None, show_default=True, help="ORDER estilo AroFlo. Ej: createdutc|desc") @click.option("--page", default=1, type=int, show_default=True) @click.option("--pagesize", type=int, default=None, show_default=True, help="Cantidad de registros por página (AroFlo pageSize).") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") def due_range_cmd( daterangetype: str, fromdate: str, todate: str, where: Optional[str], order: Optional[str], page: int, pagesize: Optional[int], raw: bool, ): params: Dict[str, Any] = { "daterangetype": daterangetype, "fromdate": fromdate, "todate": todate, "where": where, "order": order, "page": page, "raw": raw, } params = _add_optional(params, pagesize=pagesize, order=order) _echo(_run(zone, "get_tasks_due_for_date_range", params)) # ------------------------- # MUTATIONS (write) # ------------------------- @tasks_group.command("create") @click.option("--orgid", required=False, default=None, help="OrgID (si aplica para tu tenant).") @click.option("--clientid", required=True, help="ClientID (AroFlo).") @click.option("--tasktypeid", required=True, help="TaskTypeID (AroFlo).") @click.option("--summary", required=True, help="Resumen / título de la task.") @click.option("--duedate", required=False, default=None, help="YYYY/MM/DD o YYYY-MM-DD (según AroFlo).") @click.option("--description", required=False, default=None, help="Descripción (opcional).") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo CREAR una task (escritura).") def create_cmd( orgid: Optional[str], clientid: str, tasktypeid: str, summary: str, duedate: Optional[str], description: Optional[str], raw: bool, dry_run: bool, confirm: bool, ): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="create") params: Dict[str, Any] = { "clientid": clientid, "tasktypeid": tasktypeid, "summary": summary, "raw": raw, "dry_run": dry_run, } if orgid: params["orgid"] = orgid if duedate: params["duedate"] = duedate if description: params["description"] = description _echo(_run(zone, "create_task", params)) @tasks_group.command("update") @click.option("--taskid", required=True, help="TaskID (AroFlo).") @click.option("--status", required=False, default=None, help="Nuevo status (quote, notstarted, inprogress, etc.).") @click.option("--summary", required=False, default=None, help="Nuevo resumen / título.") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo ACTUALIZAR una task (escritura).") def update_cmd(taskid: str, status: Optional[str], summary: Optional[str], raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="update") params: Dict[str, Any] = {"taskid": taskid, "raw": raw, "dry_run": dry_run} if status is not None: params["status"] = status if summary is not None: params["summary"] = summary _echo(_run(zone, "update_task", params)) @tasks_group.command("add-note") @click.option("--taskid", required=True, help="TaskID destino (AroFlo).") @click.option("--content", required=True, help="Contenido de la nota (HTML o texto).") @click.option("--filter", "filter_", default=None, help="Filtro (ej: internal admin only, internal only).") @click.option("--sticky", is_flag=True, help="Marca la nota como sticky.") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo INSERTAR una nota (escritura).") def add_note_cmd(taskid: str, content: str, filter_: Optional[str], sticky: bool, raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="add-note") params: Dict[str, Any] = { "taskid": taskid, "notes": [{"content": content, "filter": filter_, "sticky": sticky}], "raw": raw, "dry_run": dry_run, } _echo(_run(zone, "insert_task_notes", params)) @tasks_group.command("add-material") @click.option("--taskid", required=True, help="TaskID destino (AroFlo).") @click.option("--item", required=True, help="Nombre del item/material.") @click.option("--partnumber", default=None, help="Partnumber (opcional).") @click.option("--cost", type=float, default=None, help="Costo unitario (opcional).") @click.option("--sell", type=float, default=None, help="Precio venta unitario (opcional).") @click.option("--dateused", default=None, help="Fecha usada (YYYY/MM/DD).") @click.option("--quantity", type=float, default=None, help="Cantidad (opcional).") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo INSERTAR material (escritura).") def add_material_cmd( taskid: str, item: str, partnumber: Optional[str], cost: Optional[float], sell: Optional[float], dateused: Optional[str], quantity: Optional[float], raw: bool, dry_run: bool, confirm: bool, ): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="add-material") material: Dict[str, Any] = {"item": item} if partnumber: material["partnumber"] = partnumber if cost is not None: material["cost"] = cost if sell is not None: material["sell"] = sell if dateused: material["dateused"] = dateused if quantity is not None: material["quantity"] = quantity params: Dict[str, Any] = {"taskid": taskid, "materials": [material], "raw": raw, "dry_run": dry_run} _echo(_run(zone, "insert_task_adhoc_materials", params)) @tasks_group.command("mark-processed") @click.option("--taskid", multiple=True, required=True, help="Repetible. Ej: --taskid X --taskid Y") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo marcar linkprocessed=true (escritura).") def mark_processed_cmd(taskid: tuple[str, ...], raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="mark-processed") params = {"taskids": list(taskid), "raw": raw, "dry_run": dry_run} _echo(_run(zone, "mark_task_linkprocessed", params)) @tasks_group.command("update-substatus") @click.option("--taskid", required=True, help="TaskID destino (AroFlo).") @click.option("--status", required=True, help="Status (ej: pending).") @click.option("--substatusid", required=True, help="SubstatusID.") @click.option("--raw", is_flag=True, help="Devuelve respuesta cruda + meta debug.") @click.option("--dry-run", is_flag=True, help="No ejecuta POST; muestra preview (postxml/params).") @click.option("--confirm", is_flag=True, help="Confirmo que deseo actualizar substatus (escritura).") def update_substatus_cmd(taskid: str, status: str, substatusid: str, raw: bool, dry_run: bool, confirm: bool): _require_confirm_if_needed(confirm=confirm, dry_run=dry_run, op_label="update-substatus") params = {"taskid": taskid, "status": status, "substatusid": substatusid, "raw": raw, "dry_run": dry_run} _echo(_run(zone, "update_task_substatus", params)) # ------------------------- # JOINS (registración compacta) # ------------------------- def add_join_command(cmd_name: str, op_code: str, help_text: str) -> None: @tasks_group.command(cmd_name, help=help_text) @_common_list_options def _cmd( page: int, where: str, order: Optional[str], pagesize: Optional[int], raw: bool, ): params = _add_optional({"page": page, "where": where, "raw": raw}, pagesize=pagesize, order=order) _echo(_run(zone, op_code, params)) add_join_command("assets", "get_tasks_with_assets", "Tasks + join=assets") add_join_command("assignedhistory", "get_tasks_with_assignedhistory", "Tasks + join=assignedhistory") add_join_command("customfields", "get_tasks_with_customfields", "Tasks + join=customfields") add_join_command("documentsandphotos", "get_tasks_with_documentsandphotos", "Tasks + join=documentsandphotos") add_join_command("expense", "get_tasks_with_expense", "Tasks + join=expense") add_join_command("labour", "get_tasks_with_labour", "Tasks + join=labour") add_join_command("material", "get_tasks_with_material", "Tasks + join=material") add_join_command("notes", "get_tasks_with_notes", "Tasks + join=notes") add_join_command("purchaseorders", "get_tasks_with_purchaseorders", "Tasks + join=purchaseorders") add_join_command("quote", "get_tasks_with_quote", "Tasks + join=quote") add_join_command("tasktotals", "get_tasks_with_tasktotals", "Tasks + join=tasktotals") add_join_command("location", "get_tasks_with_location", "Tasks + join=location") add_join_command("locationcustomfields", "get_tasks_with_locationcustomfields", "Tasks + join=locationcustomfields") add_join_command("project", "get_tasks_with_project", "Tasks + join=project") add_join_command("salesperson", "get_tasks_with_salesperson", "Tasks + join=salesperson") add_join_command("substatus", "get_tasks_with_substatus", "Tasks + join=substatus") ``` --- ## A6) `zones/registry.py` (completo, clave para nuevas zonas) ```py from __future__ import annotations from typing import Dict, List, Any from .lastupdate import LastUpdateZone from .users import UsersZone from .invoices import InvoicesZone from .clients import ClientsZone from .timesheets import TimesheetsZone from .tasks import TasksZone def build_registry(client: Any) -> Dict[str, Any]: """ Crea el registro de zonas para un cliente AroFlo concreto. """ zones: Dict[str, Any] = { "lastupdate": LastUpdateZone(client), "users": UsersZone(client), "invoices": InvoicesZone(client), "clients": ClientsZone(client), "timesheets": TimesheetsZone(client), "tasks": TasksZone(client), } return zones def list_zones_metadata(client: Any) -> List[dict]: registry = build_registry(client) return [zone.to_dict() for zone in registry.values()] def get_zone(client: Any, code: str): registry = build_registry(client) try: return registry[code] except KeyError: raise KeyError(f"Zona '{code}' no está registrada") ``` --- ## A7) Patrón para el resto de joins (rápido) Para cada join adicional: - Archivo: `zones/tasks/join_.py` - Constantes: - `JOIN_NAME = ""` - `OP_CODE = "get_tasks_with_"` - `get_operations()` con params estándar: - `where`, `order`, `page`, `pageSize`, `raw` - `execute()` reusa: - `build_list_params(zone="tasks", join=JOIN_NAME, ...)` - `request(client, "GET", params_list)` Esto garantiza consistencia con el `cheap agent` y con el CLI.