diff --git a/README.md b/README.md index e9369be..2fb635c 100644 --- a/README.md +++ b/README.md @@ -155,8 +155,11 @@ Useful control commands: ```bash loopy status +loopy status --json loopy update Prioritize the failing integration test loopy stop +loopy stop --force +loopy reload loopy traces list ``` @@ -406,6 +409,21 @@ Important rules: COORDINATOR model only — agent-CLI subprocesses (codex, claude, gemini) bill through their own accounts and are not measurable here. +An explicit `loopy reload` refreshes the running coordinator at the next task +boundary. It reloads workflow `prompt.txt` contents and these non-frozen, +coordinator-operational root settings: `recovery_policy`, +`recovery_drain_timeout_s`, `workflow_consecutive_failures_cap`, +`max_cost_usd`, and `model_prices`. The update is atomic across the +coordinator's cached workflow sets. + +Reload deliberately does **not** change a live session's frozen goal or +`config_snapshot` (including harness provider/model/families, retry settings, +system prompt extension, criteria, and turn limit), its workflow membership or +`config.yaml` cadence, its workflow contract/roster, or its capability roster. +Those changes require a coordinator restart and a new session. A later child +workflow set that has never been loaded still freezes its on-disk definition +when that child session is created. + Workflow config lives beside each workflow prompt: ```yaml @@ -651,13 +669,16 @@ loopy worker --coordinator http://127.0.0.1:8080 Runs a blocking worker until the coordinator returns `action: "stop"`. ```bash -loopy status # session stack, usage totals, estimated cost +loopy status # stack, liveness, family health, usage, estimated cost +loopy status --json # the same status as machine-readable JSON loopy status --watch # re-render every 2 seconds loopy events # the active session's event stream loopy events --follow # tail it live (--json for raw lines) loopy update TEXT... # append to the deepest active layer loopy update --session SESSION_ID TEXT... loopy stop # tree-wide stop at the next safe boundary +loopy stop --force # also reap the active iteration's tracked agent CLIs +loopy reload # refresh prompts/operational config at the next boundary loopy traces list loopy traces inspect MANIFEST_OR_ID ``` @@ -665,11 +686,30 @@ loopy traces inspect MANIFEST_OR_ID `status` prints the latest session state — the whole session stack while a child runs (the live child is shown under its suspended parent), each session's subtree token usage, and (with `model_prices` configured) estimated -cost. `update` appends the input record exactly as supplied; without `--session` it is -routed to the deepest active layer and later assignments append delivery and -acknowledgement records rather than editing history. `stop` projects root stop -intent through the whole active path and takes effect at the next register or -finish boundary; it does not invent a mid-harness interruption mechanism. +cost. For an active task it also reports the newest mtime anywhere in that +attempt's raw/output tree as `last activity: Ns ago`, plus unexpired +`rate_limited_families` found in the current team-harness `run.json` as +`model families rate-limited: FAMILY until TIME`. Missing, incomplete, or +older run records are tolerated. `status --json` exposes per-session +`last_activity_at`, `last_activity_age_s`, `rate_limit_data_available`, and +`rate_limited_families` fields. + +`update` appends the input record exactly as supplied; without `--session` it +is routed to the deepest active layer and later assignments append delivery +and acknowledgement records rather than editing history. `stop` projects root +stop intent through the whole active path and takes effect at the next +register or finish boundary. `stop --force` first records that same durable +tree-wide intent, then invokes the existing recovery reaper with immediate +`reap` policy and reports harness-run, settled-agent, and unsettled-agent +counts. Force reaping is limited to tracked process groups on the local host; +a remote active worker is reported as unreachable rather than pretending its +agents were stopped. + +`reload` writes an ignored, atomic reload generation under the sessions root. +At the next task boundary, the coordinator reruns preflight and atomically +refreshes only the prompt/operational fields described in Configuration above. +It does not mutate an already-frozen attempt; the following attempt receives +the refreshed prompt in its own immutable workflow snapshot. Trace commands accept a manifest ID, a trace root, or a manifest path confined to this repository's `.loopy_loop/traces/`. `inspect` prints the manifest plus diff --git a/design/decisions.md b/design/decisions.md index 8c0b40a..5e96782 100644 --- a/design/decisions.md +++ b/design/decisions.md @@ -234,8 +234,10 @@ own: periodic heartbeat; liveness is checked directly against the stored identity. **Default: bounded drain** — let an in-flight agent finish within a timeout (fits loopy's git-is-truth, cost-conscious profile; no concurrent-writer problem because it runs during - recovery), with **reap** as the escape for hung-past-timeout or unsafe-to-finish work. A - future force-stop command must reuse this cleanup path rather than inventing another one. + recovery), with **reap** as the escape for hung-past-timeout or unsafe-to-finish work. + `loopy stop --force` reuses this exact cleanup path with team-harness's explicit + live-parent override after recording tree-wide stop intent; it does not implement a + second process killer. A drained iteration is **never accepted or synthesized from drained outputs**: its `result.json` never existed, and fabricating one would trigger the false-closure trap D3 prevents. If stop conditions still allow the session to continue, the scheduler dispatches @@ -267,6 +269,9 @@ drain, reap as escape). For identity-tracked same-host runs this makes D6's stat verify-dead-before-reclaim rather than optimistic; legacy and remote identities retain the documented limitation above. The implemented protocol and salvage boundary are described in `designs/long-running-loop-reliability.md`; the process-group mechanism is team-harness TH-D5. +Force-stop has the same host and durable-run-record boundary: it reports an unreachable remote +worker or zero discovered runs honestly instead of claiming that untracked processes were +terminated. ## D8. Semantic constraints are visible detection with accountable repair or disposition, never hard prevention diff --git a/design/designs/long-running-loop-reliability.md b/design/designs/long-running-loop-reliability.md index c9341d9..18e3c0f 100644 --- a/design/designs/long-running-loop-reliability.md +++ b/design/designs/long-running-loop-reliability.md @@ -222,6 +222,16 @@ workflow is selected next. dead-worker reclaim, stale-owner completions, bounded recovery, unsettled-writer refusal, and salvage records. +`loopy stop --force` is the operator-triggered form of the same mechanism. It +first projects durable stop intent through the active session stack, then calls +`recover_interrupted_iteration()` with immediate `reap` policy and +team-harness's explicit live-parent override. This terminates the current +iteration's tracked same-host agent process groups without teaching Loopy a +second process-killing implementation. The command prints processed-run and +settled/unsettled-agent counts. A remote worker cannot be signalled from the +coordinator host and is reported as unreachable; absent durable run records are +reported as zero processed runs rather than a false guarantee. + --- ## The planner/dispatcher template is executable from a clean init @@ -276,6 +286,34 @@ The CLI makes that projection useful without adding a dashboard: These commands live in `src/loopy_loop/cli.py`; stream and CLI behavior are covered by `src/tests/test_events_and_usage.py`. +For the deepest active task, `loopy status` also scans the current folded raw +directory (or the legacy mirror/output location) for the newest mtime. The age +is an observation, not a heartbeat guarantee: a quiet but computing process can +have an old timestamp, while a recently written file proves only recent I/O. +The same best-effort scan reads unexpired `rate_limited_families` entries from +caller-owned team-harness `run.json` files. Missing or partially written data is +treated as unavailable, never as a coordinator failure. `status --json` +publishes the underlying timestamp, age, data-availability flag, and family +records. + +### Explicit preflight-cache reload keeps frozen session policy intact + +`loopy reload` atomically writes a generation marker below the ignored sessions +root. `CoordinatorService._reload_preflights_if_requested()` notices a new +generation at the next task boundary, validates fresh preflight data for every +cached workflow set, and swaps all refreshed caches together. The next attempt +therefore freezes the current workflow prompt instead of the startup copy. + +The reload boundary follows the existing durable contracts. It may refresh +workflow prompt text and the root settings used only by the coordinator: +recovery policy/timeout, workflow failure cap, cost budget, and model prices. +It preserves workflow membership and `config.yaml` cadence because +`workflow_roster.json` is session-frozen. It also preserves the session's goal, +workflow contract, worker `config_snapshot`, model/retry/system-prompt policy, +and tree capability roster. Those values change only after restart/new-session +creation; descendants continue inheriting the parent's frozen execution +snapshot. + ### Unknown usage stays unknown The worker's `_read_harness_usage()` reads coordinator-model turn usage from diff --git a/skills/loopy-loop/SKILL.md b/skills/loopy-loop/SKILL.md index 5b97990..30ce110 100644 --- a/skills/loopy-loop/SKILL.md +++ b/skills/loopy-loop/SKILL.md @@ -418,17 +418,27 @@ Useful operations: ```bash loopy status +loopy status --json loopy status --watch loopy events --follow loopy update TEXT... loopy stop +loopy stop --force +loopy reload loopy traces list loopy traces inspect MANIFEST_OR_ID loopy coordinator --resume ``` -`status` and `events` walk the active session stack. `loopy stop` is the -tree-wide stop request and applies at a safe register/finish boundary. +`status` and `events` walk the active session stack. Status includes the active +attempt's last raw/output write and any unexpired model-family rate-limit +circuits; `--json` exposes the same observations as structured fields. +`loopy stop` is the tree-wide stop request and applies at a safe +register/finish boundary. `loopy stop --force` also invokes the existing +same-host process-group reaper for the active iteration. `loopy reload` +refreshes workflow prompts and coordinator-operational recovery/failure/cost +settings at the next task boundary; it never changes a session-frozen config +snapshot, workflow roster/contract, or capability roster. `--resume` reconstructs the active path and continues the deepest live session. Exactly one worker is deliberate; a second verifiably live worker is refused. diff --git a/src/loopy_loop/cli.py b/src/loopy_loop/cli.py index 656dd56..aaa03f8 100644 --- a/src/loopy_loop/cli.py +++ b/src/loopy_loop/cli.py @@ -1,10 +1,13 @@ from __future__ import annotations +from datetime import datetime +from datetime import UTC from importlib.resources import files from importlib.resources.abc import Traversable import json from pathlib import Path import shutil +import socket import time import uuid @@ -26,10 +29,20 @@ from loopy_loop.git_evidence import capture_git_evidence from loopy_loop.models import LoopState from loopy_loop.models import utc_now +from loopy_loop.recovery import recover_interrupted_iteration +from loopy_loop.recovery import RecoveryIncompleteError +from loopy_loop.recovery import RecoveryOutcome +from loopy_loop.recovery import RecoveryRefusedError from loopy_loop.sessions import append_jsonl_record +from loopy_loop.sessions import attempt_trace_dir_path +from loopy_loop.sessions import iteration_harness_output_root +from loopy_loop.sessions import preflight_reload_request_path +from loopy_loop.sessions import raw_attempt_dir_path from loopy_loop.sessions import raw_dir_path from loopy_loop.sessions import RAW_DIRNAME from loopy_loop.sessions import session_dir_path +from loopy_loop.sessions import session_layout +from loopy_loop.sessions import SESSION_LAYOUT_FOLDED from loopy_loop.sessions import SESSION_METADATA_FILENAME from loopy_loop.sessions import sessions_root_path from loopy_loop.sessions import state_path @@ -391,11 +404,23 @@ def worker(coordinator_url: str) -> None: default=False, help="Re-render every 2 seconds until interrupted.", ) -def status(watch: bool) -> None: +@click.option( + "--json", + "as_json", + is_flag=True, + default=False, + help="Print machine-readable JSON.", +) +def status(watch: bool, as_json: bool) -> None: """Show loop status (the whole session stack, with usage totals).""" repo_root = Path.cwd() + if watch and as_json: + raise click.ClickException("--watch and --json cannot be used together") if not watch: try: + if as_json: + click.echo(json.dumps(_status_payload(repo_root=repo_root), indent=2)) + return lines = _status_lines(repo_root=repo_root) except FileLockTimeout: raise click.ClickException( @@ -463,6 +488,7 @@ def _status_lines(*, repo_root: Path, tolerate_lock: bool = False) -> list[str]: def _session_status_lines( *, repo_root: Path, state: LoopState, indent: str, prices: ModelPrices | None ) -> list[str]: + observability = _active_iteration_observability(repo_root=repo_root, state=state) lines = [ f"{indent}status: {state.status}", f"{indent}session: {state.active_session_id}", @@ -477,6 +503,21 @@ def _session_status_lines( f"session {state.current_task.session_id}, " f"started {state.current_task.started_at})" ) + activity_age = observability["last_activity_age_s"] + if activity_age is None: + lines.append(f"{indent}last activity: unavailable") + else: + lines.append(f"{indent}last activity: {activity_age}s ago") + rate_limits = observability["rate_limited_families"] + if rate_limits: + rendered = ", ".join( + f"{item['family']} until {item['resets_at']}" for item in rate_limits + ) + lines.append(f"{indent}model families rate-limited: {rendered}") + elif not observability["rate_limit_data_available"]: + lines.append(f"{indent}model families rate-limited: unavailable") + else: + lines.append(f"{indent}model families rate-limited: none") lines.append(f"{indent}stop_reason: {state.stop_reason or 'none'}") # Subtree totals: this session's own iterations plus finalized children. totals = session_tree_usage_totals(repo_root=repo_root, state=state) @@ -497,6 +538,220 @@ def _session_status_lines( return lines +def _status_payload(*, repo_root: Path) -> dict[str, object]: + """Return the same session-stack status as a stable JSON object.""" + + state = StateStore(repo_root=repo_root).read_state() + if state is None: + return {"schema_version": 1, "sessions": []} + prices = _configured_model_prices(repo_root=repo_root) + sessions: list[dict[str, object]] = [] + warnings: list[str] = [] + seen: set[str] = set() + while state.active_session_id not in seen: + seen.add(state.active_session_id) + sessions.append( + _session_status_payload(repo_root=repo_root, state=state, prices=prices) + ) + child_id = state.active_child_session_id + if child_id is None: + break + try: + child_state = StateStore( + repo_root=repo_root, + state_path=state_path(repo_root=repo_root, session_id=child_id), + ).read_state() + except FileLockTimeout: + warnings.append(f"active child {child_id}: state locked; retry shortly") + break + if child_state is None: + warnings.append( + f"active_child_session_id points at {child_id}, but its state " + "is missing (stale pointer)" + ) + break + state = child_state + payload: dict[str, object] = {"schema_version": 1, "sessions": sessions} + if warnings: + payload["warnings"] = warnings + return payload + + +def _session_status_payload( + *, repo_root: Path, state: LoopState, prices: ModelPrices | None +) -> dict[str, object]: + totals = session_tree_usage_totals(repo_root=repo_root, state=state) + cost = estimate_cost_usd( + prompt_tokens=totals.prompt_tokens, + completion_tokens=totals.completion_tokens, + prices=prices, + ) + current_task: dict[str, object] | None = None + if state.current_task is not None: + current_task = { + "workflow_id": state.current_task.workflow_id, + "iteration": state.current_task.iteration, + "session_id": state.current_task.session_id, + "attempt_id": state.current_task.attempt_id, + "started_at": state.current_task.started_at.isoformat(), + } + return { + "status": state.status, + "session_id": state.active_session_id, + "iteration_count": state.iteration_count, + "current_task": current_task, + "stop_reason": state.stop_reason, + "subtree_usage": totals.model_dump(mode="json"), + "subtree_estimated_cost_usd": cost, + **_active_iteration_observability(repo_root=repo_root, state=state), + } + + +def _active_iteration_observability( + *, repo_root: Path, state: LoopState +) -> dict[str, object]: + """Best-effort liveness and family health for one session's active task.""" + + roots = _active_iteration_output_roots(repo_root=repo_root, state=state) + latest_mtime = _latest_tree_mtime(roots=roots) + now = time.time() + last_activity_at: str | None = None + last_activity_age_s: int | None = None + if latest_mtime is not None: + last_activity_at = ( + datetime.fromtimestamp(latest_mtime, tz=UTC) + .isoformat() + .replace("+00:00", "Z") + ) + last_activity_age_s = max(0, int(now - latest_mtime)) + rate_limits, rate_limit_data_available = _active_rate_limited_families( + roots=roots, now=datetime.fromtimestamp(now, tz=UTC) + ) + return { + "last_activity_at": last_activity_at, + "last_activity_age_s": last_activity_age_s, + "rate_limit_data_available": rate_limit_data_available, + "rate_limited_families": rate_limits, + } + + +def _active_iteration_output_roots(*, repo_root: Path, state: LoopState) -> list[Path]: + task = state.current_task + if task is None: + return [] + roots: list[Path] = [] + if session_layout(repo_root=repo_root, session_id=task.session_id) == ( + SESSION_LAYOUT_FOLDED + ): + roots.append( + raw_attempt_dir_path( + repo_root=repo_root, + session_id=task.session_id, + iteration=task.iteration, + workflow_id=task.workflow_id, + ) + ) + elif task.attempt_id is not None: + roots.append( + attempt_trace_dir_path( + repo_root=repo_root, + root_session_id=state.root_session_id or state.active_session_id, + session_id=task.session_id, + attempt_id=task.attempt_id, + ) + ) + # Historical runs wrote below harness_outputs instead of the caller-owned + # raw trace. Include that location so resumed legacy sessions remain useful. + roots.append( + iteration_harness_output_root( + repo_root=repo_root, + session_id=task.session_id, + iteration=task.iteration, + workflow_id=task.workflow_id, + ) + ) + return list(dict.fromkeys(roots)) + + +def _latest_tree_mtime(*, roots: list[Path]) -> float | None: + latest: float | None = None + for root in roots: + try: + candidates = [root, *root.rglob("*")] if root.exists() else [] + for path in candidates: + if path.is_symlink(): + continue + modified = path.stat().st_mtime + latest = modified if latest is None else max(latest, modified) + except OSError: + continue + return latest + + +def _active_rate_limited_families( + *, roots: list[Path], now: datetime +) -> tuple[list[dict[str, str | None]], bool]: + """Read active family circuit intervals from caller-owned run.json files.""" + + active: dict[str, tuple[datetime, dict[str, str | None]]] = {} + data_available = False + for root in roots: + harness_roots = [root / "harness", root] + for path in ( + candidate + for harness_root in harness_roots + if harness_root.is_dir() and not harness_root.is_symlink() + for candidate in harness_root.glob("*/run.json") + ): + if path.is_symlink(): + continue + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + continue + if not isinstance(payload, dict): + continue + records = payload.get("rate_limited_families") + if not isinstance(records, list): + continue + data_available = True + for record in records: + if not isinstance(record, dict): + continue + family = record.get("family") + resets_at = record.get("resets_at") + if not isinstance(family, str) or not family.strip(): + continue + if not isinstance(resets_at, str): + continue + try: + reset = datetime.fromisoformat(resets_at.replace("Z", "+00:00")) + except ValueError: + continue + if reset.tzinfo is None: + reset = reset.replace(tzinfo=UTC) + if reset <= now: + continue + item = { + "family": family, + "model": ( + record.get("model") + if isinstance(record.get("model"), str) + else None + ), + "resets_at": resets_at, + "reason": ( + record.get("reason") + if isinstance(record.get("reason"), str) + else None + ), + } + previous = active.get(family) + if previous is None or reset > previous[0]: + active[family] = (reset, item) + return [active[family][1] for family in sorted(active)], data_available + + def _configured_model_prices(*, repo_root: Path) -> ModelPrices | None: try: return load_root_config(repo_root=repo_root).model_prices @@ -707,8 +962,14 @@ def _remove_prunable(*, path: Path) -> None: @main.command() -def stop() -> None: - """Request a tree-wide stop at the next safe assignment boundary.""" +@click.option( + "--force", + is_flag=True, + default=False, + help="Also immediately reap the active iteration's tracked agent processes.", +) +def stop(force: bool) -> None: + """Request a tree-wide stop, optionally reaping active agent processes.""" repo_root = Path.cwd() store = StateStore(repo_root=repo_root) @@ -738,6 +999,78 @@ def mutator(state: LoopState | None) -> tuple[LoopState, None]: "coordinator state is locked (likely mid-request); retry shortly" ) from None click.echo("stop requested") + if not force: + return + if state is None or state.current_task is None: + click.echo("force reap: no active iteration") + return + task = state.current_task + if task.worker is not None and task.worker.hostname != socket.gethostname(): + raise click.ClickException( + "stop was requested, but force reap cannot reach the active worker " + f"on remote host {task.worker.hostname}" + ) + try: + outcome = recover_interrupted_iteration( + repo_root=repo_root, + session_id=task.session_id, + iteration=task.iteration, + workflow_id=task.workflow_id, + policy="reap", + drain_timeout_s=0.0, + attempt_id=task.attempt_id, + force=True, + ) + except RecoveryIncompleteError as exc: + if exc.outcome is not None: + click.echo(_force_reap_summary(outcome=exc.outcome)) + raise click.ClickException( + f"stop was requested, but force reap was incomplete: {exc}" + ) from exc + except RecoveryRefusedError as exc: + raise click.ClickException( + f"stop was requested, but force reap was incomplete: {exc}" + ) from exc + click.echo(_force_reap_summary(outcome=outcome)) + + +def _force_reap_summary(*, outcome: RecoveryOutcome) -> str: + """Render recovery counters without exposing team-harness report internals.""" + + return ( + "force reap: " + f"harness_runs={outcome.reaped_runs} " + f"settled_agents={outcome.settled_workers} " + f"unsettled_agents={outcome.unsettled_workers}" + ) + + +@main.command() +def reload() -> None: + """Reload prompts and coordinator-only config at the next task boundary.""" + + repo_root = Path.cwd() + try: + state = StateStore(repo_root=repo_root).read_state() + except FileLockTimeout: + raise click.ClickException( + "coordinator state is locked (likely mid-request); retry shortly" + ) from None + if state is None: + raise click.ClickException("No loopy-loop state found.") + request_id = f"reload-{uuid.uuid4().hex}" + write_json_atomic( + path=preflight_reload_request_path(repo_root=repo_root), + payload={ + "schema_version": 1, + "request_id": request_id, + "requested_at": utc_now().isoformat().replace("+00:00", "Z"), + }, + ) + click.echo( + "reload requested; workflow prompts and coordinator-operational config " + "will refresh at the next task boundary" + ) @main.command() diff --git a/src/loopy_loop/coordinator_app.py b/src/loopy_loop/coordinator_app.py index dd3993d..736aa17 100644 --- a/src/loopy_loop/coordinator_app.py +++ b/src/loopy_loop/coordinator_app.py @@ -100,6 +100,7 @@ from loopy_loop.sessions import goal_contract_path from loopy_loop.sessions import handoff_path from loopy_loop.sessions import pending_finished_request_path +from loopy_loop.sessions import preflight_reload_request_path from loopy_loop.sessions import protocol_failures_dir_path from loopy_loop.sessions import raw_attempt_dir_path from loopy_loop.sessions import result_path @@ -146,6 +147,17 @@ "default_tier", } +# These root settings are operational coordinator policy, not part of a +# session's frozen worker config or model/capability roster. An explicit +# `loopy reload` may refresh only these fields plus workflow prompt text. +_HOT_RELOADABLE_ROOT_FIELDS = { + "recovery_policy", + "recovery_drain_timeout_s", + "workflow_consecutive_failures_cap", + "max_cost_usd", + "model_prices", +} + _LIVE_CHILD_STATUSES = frozenset({"dispatching", "running"}) _TERMINAL_CHILD_STATUSES = frozenset({"stopped", "goal_met", "failed", "max_turns"}) _VALID_CHILD_STATUSES = ( @@ -483,6 +495,9 @@ def __init__( self.preflights: dict[str, PreflightResult] = { preflight.workflow_set: preflight } + # A request already present at startup is already reflected by the + # startup preflight. Only a later request invalidates this cache. + self._preflight_reload_request_id = self._read_preflight_reload_request_id() self.state_store = state_store # Serializes cross-store transitions (parent<->child handoff and the # phase-B commits): FastAPI runs sync endpoints in a threadpool, so @@ -930,6 +945,7 @@ def _advance( so the three former copies (register, finished no-task, finished matched) cannot drift apart. """ + self._reload_preflights_if_requested() suspended = self._suspended_parent_response(state=state) if suspended is not None: return suspended @@ -3459,6 +3475,7 @@ def _preflight_for(self, *, workflow_set: str) -> PreflightResult: frozen config_snapshot (see _dispatch_child_session_if_requested), so a mid-session edit of loopy_loop_config.yaml cannot split the session tree across different models or policies.""" + self._reload_preflights_if_requested() preflight = self.preflights.get(workflow_set) if preflight is None: preflight = run_preflight( @@ -3467,6 +3484,67 @@ def _preflight_for(self, *, workflow_set: str) -> PreflightResult: self.preflights[workflow_set] = preflight return preflight + def _read_preflight_reload_request_id(self) -> str | None: + """Read the explicit operator reload generation, best-effort.""" + + path = preflight_reload_request_path(repo_root=self.repo_root) + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + if not isinstance(payload, dict): + return None + request_id = payload.get("request_id") + return request_id if isinstance(request_id, str) and request_id else None + + def _reload_preflights_if_requested(self) -> None: + """Atomically refresh mutable preflight inputs after `loopy reload`. + + Workflow prompts and coordinator-operational root settings are mutable. + Workflow membership/config/cadence, role contracts, the session config + snapshot, and model/capability rosters remain frozen for live sessions. + """ + + request_id = self._read_preflight_reload_request_id() + if request_id is None or request_id == self._preflight_reload_request_id: + return + refreshed: dict[str, PreflightResult] = {} + for workflow_set, cached in self.preflights.items(): + loaded = run_preflight(repo_root=self.repo_root, workflow_set=workflow_set) + cached_by_id = {workflow.id: workflow for workflow in cached.workflows} + loaded_by_id = {workflow.id: workflow for workflow in loaded.workflows} + if set(cached_by_id) != set(loaded_by_id): + raise ConfigError( + "hot reload cannot change the session-frozen workflow roster " + f"for {workflow_set!r}; restart with a new session" + ) + workflows = [ + workflow.model_copy( + update={ + "prompt_path": loaded_by_id[workflow.id].prompt_path, + "prompt_text": loaded_by_id[workflow.id].prompt_text, + "prompt_sha256": loaded_by_id[workflow.id].prompt_sha256, + } + ) + for workflow in cached.workflows + ] + root_config = cached.root_config.model_copy( + update={ + field: getattr(loaded.root_config, field) + for field in _HOT_RELOADABLE_ROOT_FIELDS + } + ) + refreshed[workflow_set] = cached.model_copy( + update={"root_config": root_config, "workflows": workflows} + ) + self.preflights = refreshed + self.preflight = refreshed[self.preflight.workflow_set] + self._preflight_reload_request_id = request_id + logger.info( + "reloaded workflow prompts and coordinator-operational config (%s)", + request_id, + ) + def _workflows_for(self, *, workflow_set: str) -> list[WorkflowDefinition]: return self._preflight_for(workflow_set=workflow_set).workflows diff --git a/src/loopy_loop/recovery.py b/src/loopy_loop/recovery.py index 71d1038..111361d 100644 --- a/src/loopy_loop/recovery.py +++ b/src/loopy_loop/recovery.py @@ -41,6 +41,7 @@ from loopy_loop.models import utc_now from loopy_loop.sessions import iteration_dir_path from loopy_loop.sessions import iteration_harness_output_root +from loopy_loop.sessions import raw_attempt_dir_path from loopy_loop.sessions import traces_root_path from loopy_loop.sessions import write_json_atomic @@ -78,6 +79,10 @@ class RecoveryIncompleteError(RuntimeError): operator can kill them (or wait for them to exit) and register again. """ + def __init__(self, message: str, *, outcome: RecoveryOutcome | None = None) -> None: + super().__init__(message) + self.outcome = outcome + class RecoveryRefusedError(RuntimeError): """team-harness refused to reap: the run's owning process is still alive. @@ -125,12 +130,14 @@ def recover_interrupted_iteration( policy: str, drain_timeout_s: float, attempt_id: str | None = None, + force: bool = False, ) -> RecoveryOutcome: """Drain/reap the interrupted iteration's orphaned agents; write salvage.json. Raises ``RecoveryRefusedError`` when team-harness's parent-liveness guard finds the run's owning process still alive — the caller should treat that - as "the previous worker is still running". + as "the previous worker is still running". Operator force-stop passes + ``force=True`` to reuse this path against the current live iteration. """ outcome = RecoveryOutcome(policy=policy) loaded = _load_reaper() @@ -149,6 +156,15 @@ def recover_interrupted_iteration( iteration=iteration, workflow_id=workflow_id, ) + folded_harness_root = ( + raw_attempt_dir_path( + repo_root=repo_root, + session_id=session_id, + iteration=iteration, + workflow_id=workflow_id, + ) + / "harness" + ) # ONE deadline shared across every discovered run — the advertised # timeout bounds the whole recovery, not each run separately. deadline = time.monotonic() + drain_timeout_s @@ -158,6 +174,7 @@ def recover_interrupted_iteration( traces_root=traces_root_path(repo_root=repo_root), session_id=session_id, attempt_id=attempt_id, + folded_harness_root=folded_harness_root, legacy_runs_root=Path(th_config.RUNS_DIR), ): if not run_json.exists(): @@ -182,6 +199,7 @@ def recover_interrupted_iteration( run_json, policy=policy, drain_timeout_s=max(0.0, deadline - time.monotonic()), + force=force, ) except reaper.ReapRefusedError as exc: raise RecoveryRefusedError(str(exc)) from exc @@ -212,7 +230,8 @@ def recover_interrupted_iteration( "(unverifiable identity, probe failure, or a kill that did not " "land); refusing to dispatch replacement work. See salvage.json " "in the iteration directory, resolve the leftover processes, " - "and register again." + "and register again.", + outcome=outcome, ) return outcome @@ -266,6 +285,7 @@ def _discover_run_records( traces_root: Path, session_id: str, attempt_id: str | None, + folded_harness_root: Path, legacy_runs_root: Path, ) -> list[tuple[str, Path, str]]: """Return explicit caller records first, then non-duplicate legacy records.""" @@ -279,6 +299,13 @@ def _discover_run_records( continue seen.add(resolved) records.append((path.parent.name, path, "caller")) + if attempt_id and folded_harness_root.is_dir(): + for path in sorted(folded_harness_root.glob("*/run.json")): + resolved = path.resolve() + if resolved in seen: + continue + seen.add(resolved) + records.append((path.parent.name, path, "caller")) for run_id in _discover_run_ids(output_root=output_root): path = legacy_runs_root / run_id / "run.json" resolved = path.resolve() diff --git a/src/loopy_loop/sessions.py b/src/loopy_loop/sessions.py index 2abed20..786b156 100644 --- a/src/loopy_loop/sessions.py +++ b/src/loopy_loop/sessions.py @@ -52,6 +52,7 @@ HARNESS_OUTPUTS_DIRNAME = "harness_outputs" TRACES_DIRNAME = "traces" TRACE_FINALIZATION_OUTBOX_DIRNAME = "trace_finalization_outbox" +PREFLIGHT_RELOAD_REQUEST_FILENAME = ".preflight_reload.json" UPDATES_FROM_USER_FILENAME = "updates_from_user.md" FINISHED_FILENAME = "finished.md" PROMPT_FILENAME = "prompt.txt" @@ -641,6 +642,12 @@ def sessions_root_path(*, repo_root: Path) -> Path: return repo_root / LOOPY_DIRNAME / SESSIONS_DIRNAME +def preflight_reload_request_path(*, repo_root: Path) -> Path: + """Return the ignored operator trigger for coordinator preflight reloads.""" + + return sessions_root_path(repo_root=repo_root) / PREFLIGHT_RELOAD_REQUEST_FILENAME + + def session_dir_path(*, repo_root: Path, session_id: str) -> Path: """Resolve a session ID through the validated recursive topology.""" diff --git a/src/tests/test_cli.py b/src/tests/test_cli.py index 949ce1e..f74d618 100644 --- a/src/tests/test_cli.py +++ b/src/tests/test_cli.py @@ -7,7 +7,9 @@ from click.testing import CliRunner from loopy_loop.cli import main +from loopy_loop.coordinator_app import create_coordinator_app from loopy_loop.models import REQUIRED_V3_WORKER_CAPABILITIES +from loopy_loop.recovery import RecoveryOutcome from loopy_loop.sessions import create_session_dir from loopy_loop.state_store import StateStore from tests.protocol_helpers import v2_finished_body @@ -269,6 +271,102 @@ def test_status_and_stop_commands( assert updated.stop_requested is True +def test_stop_force_reaps_active_iteration( + repo_builder: Any, monkeypatch: Any, state_factory: Any, current_task_factory: Any +) -> None: + repo_root = repo_builder() + monkeypatch.chdir(repo_root) + store = StateStore(repo_root=repo_root) + state = state_factory( + current_task=current_task_factory( + attempt_id="attempt-force-stop", + session_id="20260419_143022_cdbf6975e8a3_ab12cd34", + ) + ) + create_session_dir( + repo_root=repo_root, + session_id=state.active_session_id, + goal_hash=state.goal_hash, + workflow_set=state.workflow_set, + ) + store.write_state(state=state) + calls: list[dict[str, Any]] = [] + + def fake_recover(**kwargs: Any) -> RecoveryOutcome: + calls.append(kwargs) + return RecoveryOutcome( + policy="reap", reaped_runs=1, settled_workers=2, unsettled_workers=0 + ) + + monkeypatch.setattr("loopy_loop.cli.recover_interrupted_iteration", fake_recover) + + result = CliRunner().invoke(main, ["stop", "--force"]) + + updated = store.read_state() + assert result.exit_code == 0, result.output + assert updated is not None and updated.stop_requested is True + assert len(calls) == 1 + assert calls[0]["policy"] == "reap" + assert calls[0]["force"] is True + assert calls[0]["attempt_id"] == "attempt-force-stop" + assert "harness_runs=1 settled_agents=2 unsettled_agents=0" in result.output + + +def test_reload_refreshes_prompt_and_operational_config_but_not_frozen_snapshot( + repo_builder: Any, monkeypatch: Any +) -> None: + from fastapi.testclient import TestClient + + monkeypatch.setenv("OPENROUTER_API_KEY", "secret") + repo_root = repo_builder() + monkeypatch.chdir(repo_root) + app = create_coordinator_app(repo_root=repo_root, resume=False) + service = app.state.service + original_state = StateStore(repo_root=repo_root).read_state() + assert original_state is not None + original_snapshot = original_state.config_snapshot.model_dump() + prompt_path = repo_root.joinpath( + ".loopy_loop/workflow_sets/main/workflows/planner/prompt.txt" + ) + prompt_path.write_text("Reloaded planner prompt.\n", encoding="utf-8") + workflow_config_path = prompt_path.with_name("config.yaml") + workflow_config_path.write_text( + workflow_config_path.read_text(encoding="utf-8").replace( + "description: Plan work.", "description: Changed on disk." + ), + encoding="utf-8", + ) + config_path = repo_root / "loopy_loop_config.yaml" + config_text = config_path.read_text(encoding="utf-8") + config_path.write_text( + config_text.replace( + "team_harness_model: gpt-5.5", "team_harness_model: changed-model" + ) + + "\nrecovery_policy: reap\n", + encoding="utf-8", + ) + + requested = CliRunner().invoke(main, ["reload"]) + task = TestClient(app).post("/register", json=v2_register_body(repo_root)).json() + refreshed = service.preflight + persisted = StateStore(repo_root=repo_root).read_state() + + assert requested.exit_code == 0, requested.output + planner = next(item for item in refreshed.workflows if item.id == "planner") + assert planner.prompt_text == "Reloaded planner prompt.\n" + assert planner.description == "Plan work." + assert service.preflight.root_config.recovery_policy == "reap" + assert service.preflight.root_config.team_harness_model == "gpt-5.5" + assert ( + Path(task["workflow_snapshot"]["workflow_prompt_path"]).read_text( + encoding="utf-8" + ) + == "Reloaded planner prompt.\n" + ) + assert persisted is not None + assert persisted.config_snapshot.model_dump() == original_snapshot + + def test_coordinator_requires_resume_for_running_state( repo_builder: Any, monkeypatch: Any, state_factory: Any ) -> None: diff --git a/src/tests/test_operational_cli.py b/src/tests/test_operational_cli.py index 90c3537..6a1eaea 100644 --- a/src/tests/test_operational_cli.py +++ b/src/tests/test_operational_cli.py @@ -123,6 +123,108 @@ def test_update_rejects_missing_state_or_unknown_explicit_session( assert "session not found: missing" in missing_session.output +def test_status_json_includes_liveness_and_tolerates_missing_rate_limit_data( + tmp_path: Path, monkeypatch: Any, state_factory: Any, current_task_factory: Any +) -> None: + session_id = "active-session" + attempt_id = "attempt-live" + _create_session(repo_root=tmp_path, session_id=session_id) + state = state_factory( + active_session_id=session_id, + root_session_id=session_id, + current_task=current_task_factory(session_id=session_id, attempt_id=attempt_id), + ) + StateStore(repo_root=tmp_path).write_state(state=state) + trace_root, _ = create_attempt_trace( + repo_root=tmp_path, + root_session_id=session_id, + session_id=session_id, + request_id=None, + work_item_id=None, + workflow_set="main", + workflow_id="planner", + iteration=1, + attempt_id=attempt_id, + ) + run_dir = trace_root / "harness" / "run-live" + run_dir.mkdir(parents=True) + run_dir.joinpath("run.json").write_text( + json.dumps({"run_id": "run-live", "turns": []}), encoding="utf-8" + ) + monkeypatch.chdir(tmp_path) + + result = CliRunner().invoke(main, ["status", "--json"]) + + assert result.exit_code == 0, result.output + session = json.loads(result.output)["sessions"][0] + assert session["last_activity_at"].endswith("Z") + assert isinstance(session["last_activity_age_s"], int) + assert session["last_activity_age_s"] >= 0 + assert session["rate_limit_data_available"] is False + assert session["rate_limited_families"] == [] + text_result = CliRunner().invoke(main, ["status"]) + assert "last activity:" in text_result.output + assert "model families rate-limited: unavailable" in text_result.output + + +def test_status_surfaces_active_rate_limited_model_families( + tmp_path: Path, monkeypatch: Any, state_factory: Any, current_task_factory: Any +) -> None: + session_id = "rate-limited-session" + attempt_id = "attempt-rate-limit" + _create_session(repo_root=tmp_path, session_id=session_id) + state = state_factory( + active_session_id=session_id, + root_session_id=session_id, + current_task=current_task_factory(session_id=session_id, attempt_id=attempt_id), + ) + StateStore(repo_root=tmp_path).write_state(state=state) + trace_root, _ = create_attempt_trace( + repo_root=tmp_path, + root_session_id=session_id, + session_id=session_id, + request_id=None, + work_item_id=None, + workflow_set="main", + workflow_id="planner", + iteration=1, + attempt_id=attempt_id, + ) + run_dir = trace_root / "harness" / "run-limited" + run_dir.mkdir(parents=True) + run_dir.joinpath("run.json").write_text( + json.dumps( + { + "run_id": "run-limited", + "rate_limited_families": [ + { + "family": "claude", + "model": "claude-opus", + "tripped_at": "2098-01-01T00:00:00Z", + "resets_at": "2099-01-01T00:00:00Z", + "reason": "provider returned 429", + } + ], + } + ), + encoding="utf-8", + ) + monkeypatch.chdir(tmp_path) + + text_status = CliRunner().invoke(main, ["status"]) + json_status = CliRunner().invoke(main, ["status", "--json"]) + + assert text_status.exit_code == 0, text_status.output + assert "model families rate-limited: claude until 2099-01-01T00:00:00Z" in ( + text_status.output + ) + json_session = json.loads(json_status.output)["sessions"][0] + assert json_session["rate_limit_data_available"] is True + family = json_session["rate_limited_families"][0] + assert family["family"] == "claude" + assert family["resets_at"] == "2099-01-01T00:00:00Z" + + def test_trace_commands_list_and_inspect_lifecycle_and_integrity( tmp_path: Path, monkeypatch: Any ) -> None: diff --git a/src/tests/test_worker_liveness_recovery.py b/src/tests/test_worker_liveness_recovery.py index eb5fd38..c99eb82 100644 --- a/src/tests/test_worker_liveness_recovery.py +++ b/src/tests/test_worker_liveness_recovery.py @@ -457,6 +457,60 @@ def test_recover_interrupted_iteration_reaps_and_writes_salvage( assert len(payload["reports"]) == 2 +def test_force_recovery_discovers_protocol_v3_folded_run( + tmp_path: Path, monkeypatch: Any +) -> None: + session_id = "folded-session" + attempt_id = "folded-attempt" + run_json = ( + tmp_path + / ".loopy_loop" + / "sessions" + / session_id + / "raw" + / "0001_implement" + / "harness" + / "run-folded" + / "run.json" + ) + run_json.parent.mkdir(parents=True) + run_json.write_text( + json.dumps( + { + "run_id": "run-folded", + "caller_context": { + "session_id": session_id, + "parent_attempt_id": attempt_id, + }, + } + ), + encoding="utf-8", + ) + legacy_runs = tmp_path / "legacy-runs" + legacy_runs.mkdir() + reaper = _FakeReaperModule(["reaped"]) + monkeypatch.setattr( + recovery_module, "_load_reaper", lambda: (reaper, _FakeThConfig(legacy_runs)) + ) + + outcome = recover_interrupted_iteration( + repo_root=tmp_path, + session_id=session_id, + iteration=1, + workflow_id="implement", + policy="reap", + drain_timeout_s=0, + attempt_id=attempt_id, + force=True, + ) + + assert outcome.reaped_runs == 1 + assert outcome.settled_workers == 1 + assert reaper.calls[0]["run_json"] == run_json + assert reaper.calls[0]["policy"] == "reap" + assert reaper.calls[0]["force"] is True + + def test_recover_interrupted_iteration_refusal_propagates( tmp_path: Path, monkeypatch: Any ) -> None: diff --git a/website/src/app/docs/cli-reference/page.mdx b/website/src/app/docs/cli-reference/page.mdx index 6c01696..f87ca99 100644 --- a/website/src/app/docs/cli-reference/page.mdx +++ b/website/src/app/docs/cli-reference/page.mdx @@ -1,12 +1,12 @@ export const metadata = { title: "CLI Reference", description: - "Every loopy-loop command: init, coordinator, worker, status, events, and stop, with their flags and usage examples.", + "Every loopy-loop command: init, coordinator, worker, status, events, stop, and reload, with their flags and usage examples.", }; # CLI Reference -loopy-loop is driven by a single command-line tool with six subcommands: `init` scaffolds a repo, `coordinator` and `worker` run the loop, `status` and `events` inspect it, and `stop` controls it. Every command operates on the current working directory, so run them from the root of your target repository. +loopy-loop is driven by a single command-line tool: `init` scaffolds a repo, `coordinator` and `worker` run the loop, `status` and `events` inspect it, and `stop` and `reload` control it. Every command operates on the current working directory, so run them from the root of your target repository. ## Command aliases @@ -85,9 +85,16 @@ current_task: inner (iteration 12, session 20260712_193000_a1b2c3d4e5f6_7a8b9c0d stop_reason: none subtree_usage: prompt_tokens=12000 completion_tokens=3000 (iterations fully measured: 11, unknown: 0) subtree_harness_duration_s: 840 +last activity: 12s ago +model families rate-limited: none ``` -`loopy status --watch` clears and re-renders the view every two seconds until interrupted. +For the active task it also reports two best-effort health signals: + +- **`last activity`** — how long ago the running iteration last wrote to its output directory, so you can tell a progressing loop from a wedged one at a glance. Shows `unavailable` when there is no active task or no recent write. +- **`model families rate-limited`** — any agent model families the harness has recorded as rate-limited (with the reset time), read from the harness run log. Shows `none` when all families are available, or `unavailable` when the running harness version does not report this. + +`loopy status --watch` clears and re-renders the view every two seconds until interrupted. `loopy status --json` prints the same information as a machine-readable JSON payload instead of the text layout — use it for scripts and dashboards rather than parsing the text lines. ## loopy events @@ -107,9 +114,29 @@ Requests a graceful stop by setting `stop_requested=true` in the latest top-leve loopy stop ``` -`stop` takes no options. For how a *workflow* stops the loop on its own — versus this operator-initiated stop — see [Success & Control](/docs/success-and-control). +For how a *workflow* stops the loop on its own — versus this operator-initiated stop — see [Success & Control](/docs/success-and-control). + +`stop` targets the latest top-level session. If a child is active, the child does not see the flag; it terminates normally, then the resumed parent honors the request. + +### `--force` + +```bash +loopy stop --force +``` + +The cooperative stop above takes effect only at the next assignment boundary, so a long-running iteration keeps going until it finishes. `--force` additionally reaps the active iteration's tracked agent subprocesses right away, so a hard stop does not leave orphaned agent CLIs running after the coordinator exits. It reuses the same recovery/reaper path the engine uses for crash cleanup, and the session remains resumable afterward. Use plain `stop` to let the current work finish; use `--force` when you need the agents killed now. + +## loopy reload + +```bash +loopy reload +``` + +Applies a fixed workflow prompt or a changed coordinator-operational setting to a **running** program without restarting it. It records a reload request; at the next task boundary the coordinator re-reads the workflow prompts and its reloadable, coordinator-side configuration. + +Reload is deliberately narrow. It refreshes workflow prompts and coordinator-operational config (for example recovery and failure-cap settings). It does **not** touch anything frozen per session — the goal, completion/stop criteria, `max_turns`, the harness provider/model/families, or the workflow contracts and rosters — because a session's model and policy are intentionally immutable for its whole lifetime. To change any of those, stop and start a fresh program. -`stop` currently targets the latest top-level session. If a child is active, the child does not see the flag; it terminates normally, then the resumed parent honors the request. Child-aware and force-stop behavior remains proposed work. +It prints a confirmation on success, reports `No loopy-loop state found.` when nothing is running, and asks you to retry if the coordinator state is momentarily locked mid-request. ## Where to go next diff --git a/website/src/app/docs/troubleshooting/page.mdx b/website/src/app/docs/troubleshooting/page.mdx index bc47764..3a6fcfb 100644 --- a/website/src/app/docs/troubleshooting/page.mdx +++ b/website/src/app/docs/troubleshooting/page.mdx @@ -134,7 +134,55 @@ the session as running. stays `running` on disk. **Fix.** Either pass `--resume` the next time you start the coordinator to reattach, -or run `loopy stop` first to reach a terminal state before a fresh start. +or run `loopy stop` first to reach a terminal state before a fresh start. If you kill +the coordinator while an iteration is mid-flight, its spawned agent CLIs can be left +running with no parent to reap them — prefer `loopy stop --force`, which reaps the +active iteration's tracked agent processes as part of stopping. + +## Is the loop still making progress? + +**Symptom.** `loopy status` shows `running`, but you cannot tell whether the current +iteration is actually working or wedged. + +**Cause.** A running status only means no terminal state was written; it says nothing +about live activity. + +**Fix.** Read the `last activity: Ns ago` line in `loopy status`. It reports how long +ago the active iteration last wrote to its output directory. A small, steadily +updating value means the harness is working; a value that stops advancing (or +`unavailable`) alongside a long-running iteration points at a wedged agent — inspect +its output, then `loopy stop --force` and `--resume` if needed. + +## Agents keep failing and a whole model family is unavailable + +**Symptom.** Iterations run slowly and a particular model family (for example Claude) +seems to keep failing, or `loopy status` lists it under `model families rate-limited`. + +**Cause.** That family hit a hard provider rate limit (a weekly cap or an exhausted +overage). The harness records the tripped family and its reset time, and its circuit +breaker short-circuits further spawns of that family until the window passes, failing +over to other families rather than re-launching a subprocess that fails every time. + +**Fix.** Usually nothing — the loop keeps delivering on the remaining families until +the reset time shown in status. Note that this degrades cross-family review, so treat +the results on security-sensitive work with extra care until the family recovers, or +raise the limit with your provider. This surfacing needs a harness new enough to +report it; older versions show `unavailable`. + +## A prompt or config fix is not taking effect on a running loop + +**Symptom.** You edited a stock workflow prompt (or a coordinator-operational setting) +on disk, but the running program keeps using the old one. + +**Cause.** The coordinator caches its preflight at startup, so on-disk edits are not +re-read mid-run by default. + +**Fix.** Run `loopy reload`. At the next task boundary the coordinator re-reads the +workflow prompts and its reloadable, coordinator-side config. Reload deliberately does +**not** change anything frozen per session — goal, criteria, `max_turns`, harness +model/family, or workflow contracts/rosters — so those still require a fresh program. +See the [CLI Reference](/docs/cli-reference) for exactly what reload does and does not +refresh. ## A worker crashed mid-iteration