diff --git a/python/functions/infra/mark_claude_parent.md b/python/functions/infra/mark_claude_parent.md new file mode 100644 index 00000000..70c1f0dc --- /dev/null +++ b/python/functions/infra/mark_claude_parent.md @@ -0,0 +1,96 @@ +--- +name: mark_claude_parent +kind: function +lang: py +domain: infra +version: "1.0.0" +purity: impure +signature: "def mark_claude_parent(pid: int, parent_orchestrator: str, wait_s: float = 10.0, sessions_dir: str | None = None, goals_dir: str | None = None) -> dict" +description: "Marca el orquestador padre de una sesion de Claude Code resolviendo PID -> sessionId. Sondea ~/.claude/sessions/.json (escrito por Claude Code unos segundos despues de arrancar) hasta wait_s segundos con deadline time.monotonic, extrae el sessionId y escribe SOLO la clave `parent_orchestrator` en ~/.claude/goals/.json preservando el resto del goal (goal, phase, dod, dod_contract, role...). Escritura atomica tmp + os.replace. Si el sessions JSON no aparece a tiempo devuelve ok=False timeout sin lanzar. Equivalente de mark_claude_role para la clave parent_orchestrator: lo invoca spawn_fleet_agent (--parent) para que el watcher de fleetview rutee los avisos del ejecutor al orquestador que lo lanzo cuando hay mas de uno activo." +tags: [fleet, claude-fleet, orchestration, parent, session, goal, orchestrator, launcher, pid] +uses_functions: [] +uses_types: [] +returns: [] +returns_optional: false +error_type: "error_go_core" +imports: ["os", "json", "time"] +params: + - name: pid + desc: "PID del proceso Claude Code recien arrancado (el ejecutor). Se usa para localizar ~/.claude/sessions/.json" + - name: parent_orchestrator + desc: "sessionId del orquestador que lanzo el ejecutor. No puede ser vacio (lanza ValueError). Cualquier string no vacio es valido: es un sessionId arbitrario, no un enum" + - name: wait_s + desc: "segundos maximos a esperar (sondeo cada ~0.25s) a que aparezca sessions/.json con un sessionId no vacio. Default 10.0" + - name: sessions_dir + desc: "directorio de los sessions JSON. Default ~/.claude/sessions" + - name: goals_dir + desc: "directorio de los goal JSON. Default ~/.claude/goals (se crea si falta)" +output: "dict. En exito: ok=True, pid (int), session_id (str), parent_orchestrator (str), path (ruta del goal escrito). En timeout del poll: ok=False, error (str con 'timeout esperando sessions/.json'), pid (int). NO lanza en timeout; el launcher decide" +tested: true +tests: + - "sessions presente resuelve y escribe parent_orchestrator preservando otros campos (incl role)" + - "parent vacio lanza ValueError sin escribir" + - "sessions ausente devuelve timeout sin crash" + - "goal inexistente se crea con solo parent_orchestrator" +test_file_path: "python/functions/infra/mark_claude_parent_test.py" +file_path: "python/functions/infra/mark_claude_parent.py" +--- + +## Ejemplo + +```python +from infra.mark_claude_parent import mark_claude_parent + +# El orquestador acaba de lanzar un ejecutor (Claude Code) y conoce su PID, pero +# todavia NO su sessionId. Deja constancia de quien lo lanzo para que el watcher +# de fleetview rutee al orquestador correcto el aviso de cierre del ejecutor. +res = mark_claude_parent(executor_pid, my_session_id, wait_s=15.0) + +if res["ok"]: + print(f"padre marcado: {res['session_id']} -> {res['parent_orchestrator']}") +else: + # Timeout: el sessions/.json no aparecio a tiempo. El launcher decide. + print("no se pudo marcar el padre:", res["error"]) +``` + +Invocacion canonica via `fn run` (asi lo llama `spawn_fleet_agent --parent`): + +```bash +./fn run mark_claude_parent +``` + +## Cuando usarla + +Usala desde el launcher del orquestador de flota (`spawn_fleet_agent --parent`) +justo despues de arrancar un ejecutor, cuando conoces su PID pero todavia NO su +sessionId. Resuelve PID -> sessionId esperando el archivo que Claude Code escribe +unos segundos despues, y deja `parent_orchestrator` en el goal.json del ejecutor. +El watcher de fleetview consume esa clave para rutear el aviso de `DICE_TERMINADO` +/ estancamiento de cada ejecutor al pane tmux del orquestador que lo lanzo, en vez +de a toda la flota — necesario cuando hay mas de un orquestador activo. Es el par +de `mark_claude_role` (misma mecanica de poll, distinta clave escrita). + +## Gotchas + +- **Impura**: lee `~/.claude/sessions/.json` y escribe + `~/.claude/goals/.json` (crea `goals_dir` si falta). +- **Timing del sessions JSON**: Claude Code NO escribe `sessions/.json` al + instante de arrancar — tarda unos segundos. Por eso la funcion sondea hasta + `wait_s` (deadline con `time.monotonic`, no un unico `sleep` largo) y reacciona + en cuanto el archivo aparece con un `sessionId` no vacio. +- **Timeout NO lanza**: si el archivo no aparece a tiempo devuelve + `{"ok": False, "error": "timeout esperando sessions/.json", "pid": pid}`. + Solo `parent_orchestrator` vacio lanza (ValueError), y lo hace antes de tocar + disco. +- **NO pisa otros campos del goal**: abre el goal existente (dict vacio si no + existe o es JSON invalido / no-dict), setea UNICAMENTE la clave + `parent_orchestrator` y conserva todo lo demas (`goal`, `phase`, `dod`, + `dod_contract`, `role`, ...). Si el goal.json no existia, lo crea con solo + `{"parent_orchestrator": ...}`. +- **Escritura atomica**: escribe a un archivo temporal y hace `os.replace`, asi un + lector concurrente (la TUI / watcher fleetview) nunca ve un goal.json a medio + escribir. +- **Orden con mark_claude_role**: cuando `spawn_fleet_agent` aplica `--role` y + `--parent` a la vez, los encadena secuencialmente en el mismo subshell (primero + role, luego parent) para que el segundo lea el goal ya con la primera clave + escrita y no haya carrera de lectura-modificacion-escritura entre ambos. diff --git a/python/functions/infra/mark_claude_parent.py b/python/functions/infra/mark_claude_parent.py new file mode 100644 index 00000000..13bb8eb6 --- /dev/null +++ b/python/functions/infra/mark_claude_parent.py @@ -0,0 +1,135 @@ +"""Marca el orquestador padre de una sesion de Claude Code resolviendo PID -> sessionId. + +El meta-orquestador de flota arranca un ejecutor (Claude Code) y necesita dejar +constancia de QUIEN lo lanzo, para que el watcher de fleetview rutee los avisos de +ese ejecutor (cierre, estancamiento) al orquestador correcto cuando hay mas de uno +activo. El problema es de timing: el `sessionId` del nuevo Claude no se conoce al +arrancar el proceso; Claude Code escribe el archivo `~/.claude/sessions/.json` +(que contiene el `sessionId`) unos segundos despues. Esta funcion espera ese +archivo sondeando, resuelve el sessionId y escribe SOLO la clave +`parent_orchestrator` en `~/.claude/goals/.json` preservando todos los +demas campos del goal (goal, phase, dod, dod_contract, role, ...). + +Es el equivalente de mark_claude_role para la clave `parent_orchestrator`: misma +mecanica de poll PID -> sessionId, distinta clave escrita. El orquestador lo invoca +desde spawn_fleet_agent (flag --parent) justo despues de lanzar el ejecutor. +""" + +import json +import os +import time + + +def mark_claude_parent( + pid: int, + parent_orchestrator: str, + wait_s: float = 10.0, + sessions_dir: str | None = None, + goals_dir: str | None = None, +) -> dict: + """Resuelve PID -> sessionId esperando el sessions/.json y marca el padre. + + Sondea `/.json` cada ~0.25s hasta `wait_s` segundos (medido + con time.monotonic) esperando que exista y contenga un `sessionId` no vacio. + Si lo encuentra, abre `/.json` (dict vacio si no existe), + setea SOLO la clave `parent_orchestrator` preservando el resto (goal, phase, + dod, dod_contract, role, ...), y lo escribe atomicamente (tmp + os.replace). Si + expira, devuelve un dict de error sin lanzar excepcion (el launcher decide). + + Args: + pid: PID del proceso Claude Code recien arrancado (el ejecutor). + parent_orchestrator: sessionId del orquestador que lanzo el ejecutor. No + puede ser vacio (un padre vacio no aporta routing). Cualquier string no + vacio es valido (es un sessionId arbitrario, no un enum). + wait_s: segundos maximos a esperar el sessions/.json. Default 10.0. + sessions_dir: directorio de los sessions JSON. Default ~/.claude/sessions. + goals_dir: directorio de los goal JSON. Default ~/.claude/goals. + + Returns: + Si tuvo exito: dict con ok=True, pid, session_id, parent_orchestrator y + path (ruta del goal escrito). Si expiro el poll: dict con ok=False, error + (mensaje) y pid. + + Raises: + ValueError: si `parent_orchestrator` es vacio o solo espacios. No escribe + nada (se valida antes de tocar disco). + """ + if not parent_orchestrator or not parent_orchestrator.strip(): + raise ValueError( + "parent_orchestrator no puede ser vacio: un padre vacio no aporta routing" + ) + + home = os.path.expanduser("~") + if sessions_dir is None: + sessions_dir = os.path.join(home, ".claude", "sessions") + if goals_dir is None: + goals_dir = os.path.join(home, ".claude", "goals") + + session_path = os.path.join(sessions_dir, f"{pid}.json") + + # POLL: espera hasta wait_s segundos a que exista sessions/.json con un + # sessionId no vacio. Bucle corto con time.monotonic como deadline (no un + # unico sleep largo) para reaccionar en cuanto aparezca el archivo. + deadline = time.monotonic() + wait_s + sid = "" + while True: + sid = _read_session_id(session_path) + if sid: + break + if time.monotonic() >= deadline: + return { + "ok": False, + "error": f"timeout esperando sessions/{pid}.json", + "pid": pid, + } + time.sleep(0.25) + + goal_path = os.path.join(goals_dir, f"{sid}.json") + + # Lee el goal existente (dict vacio si no existe o es ilegible/no-dict). + goal: dict = {} + if os.path.exists(goal_path): + try: + with open(goal_path, "r", encoding="utf-8") as fh: + loaded = json.load(fh) + if isinstance(loaded, dict): + goal = loaded + except (json.JSONDecodeError, ValueError, OSError): + goal = {} + + # Setea SOLO la clave parent_orchestrator, preservando todo lo demas. + goal["parent_orchestrator"] = parent_orchestrator + + # Escritura atomica: tmp + os.replace en el mismo directorio. + os.makedirs(goals_dir, exist_ok=True) + tmp_path = goal_path + f".tmp.{os.getpid()}" + with open(tmp_path, "w", encoding="utf-8") as fh: + json.dump(goal, fh, ensure_ascii=False, indent=2) + fh.flush() + os.fsync(fh.fileno()) + os.replace(tmp_path, goal_path) + + return { + "ok": True, + "pid": pid, + "session_id": sid, + "parent_orchestrator": parent_orchestrator, + "path": goal_path, + } + + +def _read_session_id(session_path: str) -> str: + """Lee el sessionId del sessions/.json. Devuelve "" si falta/ilegible/vacio.""" + if not os.path.exists(session_path): + return "" + try: + with open(session_path, "r", encoding="utf-8") as fh: + data = json.load(fh) + except (json.JSONDecodeError, ValueError, OSError): + return "" + if not isinstance(data, dict): + return "" + sid = data.get("sessionId") + if isinstance(sid, str) and sid: + return sid + return "" diff --git a/python/functions/infra/mark_claude_parent_test.py b/python/functions/infra/mark_claude_parent_test.py new file mode 100644 index 00000000..2b0437be --- /dev/null +++ b/python/functions/infra/mark_claude_parent_test.py @@ -0,0 +1,121 @@ +"""Tests para mark_claude_parent.""" + +import json +import os + +import pytest + +from mark_claude_parent import mark_claude_parent + + +def _write_session(sessions_dir, pid, session_id): + """Escribe un sessions/.json con el sessionId dado.""" + os.makedirs(sessions_dir, exist_ok=True) + path = os.path.join(sessions_dir, f"{pid}.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump({"sessionId": session_id, "cwd": "/tmp/whatever"}, fh) + return path + + +def _write_goal(goals_dir, session_id, goal): + """Escribe un goal/.json con el dict dado.""" + os.makedirs(goals_dir, exist_ok=True) + path = os.path.join(goals_dir, f"{session_id}.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(goal, fh) + return path + + +def _read_goal(goals_dir, session_id): + path = os.path.join(goals_dir, f"{session_id}.json") + with open(path, "r", encoding="utf-8") as fh: + return json.load(fh) + + +def test_sessions_presente_resuelve_y_escribe_parent_preservando_otros_campos(tmp_path): + sessions_dir = str(tmp_path / "sessions") + goals_dir = str(tmp_path / "goals") + pid = 4242 + sid = "executor-abc-123" + parent = "orchestrator-xyz-789" + + _write_session(sessions_dir, pid, sid) + # Goal preexistente con campos que NO deben perderse (incluido role). + _write_goal( + goals_dir, + sid, + { + "goal": "implementar fase 3", + "phase": "trabajando", + "role": "executor", + "dod_contract": {"capa1": "mecanica"}, + }, + ) + + res = mark_claude_parent(pid, parent, wait_s=2.0, + sessions_dir=sessions_dir, goals_dir=goals_dir) + + assert res["ok"] is True + assert res["pid"] == pid + assert res["session_id"] == sid + assert res["parent_orchestrator"] == parent + assert res["path"] == os.path.join(goals_dir, f"{sid}.json") + + goal = _read_goal(goals_dir, sid) + assert goal["parent_orchestrator"] == parent + # Todos los demas campos del goal se preservan intactos. + assert goal["goal"] == "implementar fase 3" + assert goal["phase"] == "trabajando" + assert goal["role"] == "executor" + assert goal["dod_contract"] == {"capa1": "mecanica"} + + +def test_parent_vacio_lanza_value_error_sin_escribir(tmp_path): + sessions_dir = str(tmp_path / "sessions") + goals_dir = str(tmp_path / "goals") + pid = 4242 + sid = "executor-abc-123" + _write_session(sessions_dir, pid, sid) + + with pytest.raises(ValueError): + mark_claude_parent(pid, " ", wait_s=2.0, + sessions_dir=sessions_dir, goals_dir=goals_dir) + + # No escribio nada: el goals_dir ni siquiera deberia existir. + assert not os.path.exists(goals_dir) + + +def test_sessions_ausente_devuelve_timeout_sin_crash(tmp_path): + sessions_dir = str(tmp_path / "sessions") + goals_dir = str(tmp_path / "goals") + pid = 9999 # sin sessions/.json escrito + + res = mark_claude_parent(pid, "orchestrator-xyz-789", wait_s=0.5, + sessions_dir=sessions_dir, goals_dir=goals_dir) + + assert res["ok"] is False + assert res["pid"] == pid + assert "timeout" in res["error"] + assert f"{pid}.json" in res["error"] + # No se escribio ningun goal. + assert not os.path.exists(goals_dir) + + +def test_goal_inexistente_se_crea_con_solo_parent(tmp_path): + sessions_dir = str(tmp_path / "sessions") + goals_dir = str(tmp_path / "goals") + pid = 7 + sid = "fresh-executor-uuid" + parent = "orchestrator-xyz-789" + _write_session(sessions_dir, pid, sid) + # No existe goal previo para esta sesion. + + res = mark_claude_parent(pid, parent, wait_s=2.0, + sessions_dir=sessions_dir, goals_dir=goals_dir) + + assert res["ok"] is True + assert res["session_id"] == sid + assert res["parent_orchestrator"] == parent + + goal = _read_goal(goals_dir, sid) + assert goal == {"parent_orchestrator": parent}