"""Autonomous iteration loop for a fractal node."""
from __future__ import annotations
import dataclasses
import functools
import logging
import os
import pathlib
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
import traceback
import typing
from typing import Any, Optional
import fractal.util
from fractal.constants import PAUSE_ABORT_FILE, PGID_FILE, SOCKET_FILE, STEP_PGID_FILE
from fractal.exceptions import AgentStreamError, _Abort
from . import commit, pricing, render, worktree
from .config import DEFAULT_RESERVE_FRACTION
from .event import FailureEvent, InitialEvent, TerminalEvent
if typing.TYPE_CHECKING:
from collections.abc import Callable
from .agent import Agent, StreamEvent
from .node import Node
__all__ = ['Loop']
# the hook pairings log event text through this module's logger hierarchy
# with no handlers configured -- a NullHandler keeps the stdlib lastResort
# fallback from spilling raw event text onto the pane's stderr when no host
# attaches a handler (the transcript carries only the loop's own prints);
# scoped here, not package-wide: other fractal warnings ride lastResort
# to stderr deliberately
logging.getLogger(__name__).addHandler(logging.NullHandler())
# strict flat-scalar frontmatter grammar: a lowercase key, a colon, and a
# non-empty scalar on one line -- nothing nested, nothing quoted specially
_FRONTMATTER_LINE = re.compile(r'^([a-z_]+):\s*(.+)$')
# NN- step filename: a digit prefix, a dash, and a name
_STEP_PREFIX = re.compile(r'^([0-9]+)-.*\.md$')
# grace between the deadline's TERM and the follow-up KILL on the step group
_KILL_GRACE_SECONDS = 10.0
# cap consecutive setup failures so a deterministically broken setup.sh ends
# the run exited with the honest reason instead of crash-looping into a
# healthy-looking completed max-iters end; the counter resets on any success
_SETUP_FAIL_CAP = 3
# cap a failure reason or log-tail snippet to a readable one-liner: the raw
# log line or message behind it can be arbitrarily long
_REASON_CHARS = 250
# figure-free heads of the loop's budget-abort reasons; a finish reason
# bearing one is a budget abort -- bare when self-sent, carrying the fan-out
# attribution when cascaded from an ancestor
_BUDGET_REASON_STEMS = (
'cost budget reserve reached',
'subtree cost budget reached',
'cost budget exceeded in finish wind-down',
)
# the loop-authored reason shape: stem + figures -- classifiers match this,
# never the bare stem, so agent free text opening with the vocabulary is
# still a deliberate finish
_BUDGET_REASON_HEADS = tuple(f'{stem} (spent $' for stem in _BUDGET_REASON_STEMS)
[docs]
class Step:
"""One discovered step file.
Carries the step's number, name, path, and frontmatter overrides
(requires_approval, agent, provider, model, effort, timeout, detached).
"""
def __init__(
self: Step,
path: pathlib.Path,
*,
number: int = 0,
frontmatter: Optional[dict[str, str]] = None,
) -> None:
"""Initialize ``Step``."""
self.path = path
self.number = number
self.frontmatter = frontmatter or {}
[docs]
@classmethod
def load(cls: type[Step], path: pathlib.Path, *, number: int = 0) -> Step:
"""Parse the strict flat-scalar frontmatter grammar.
The block opens with ``---`` on the first line and closes at the
next ``---``; each line inside matching ``key: value`` (lowercase
key, non-empty scalar) contributes one override, first occurrence
winning. Anything else -- including a file with no opening fence
-- contributes nothing.
Args:
path: Step markdown file.
number: 1-based step number within the iteration (``0`` for
the SYNC pseudo-step).
Returns:
The parsed step.
"""
frontmatter: dict[str, str] = {}
lines = path.read_text(encoding='utf-8').split('\n')
if lines and lines[0].lstrip() == '---':
for line in lines[1:]:
trimmed = line.lstrip()
if trimmed == '---':
break
if match := _FRONTMATTER_LINE.match(trimmed):
frontmatter.setdefault(match.group(1), match.group(2).rstrip())
return cls(path, number=number, frontmatter=frontmatter)
@property
def name(self: Step) -> str:
"""Return the step name: the filename stem past its ``NN-`` prefix."""
stem = self.path.name.removesuffix('.md')
_, _, name = stem.partition('-')
return name or stem
@property
def requires_approval(self: Step) -> bool:
"""Return whether the ``requires_approval: true`` frontmatter is set."""
return self.frontmatter.get('requires_approval') == 'true'
@property
def agent(self: Step) -> Optional[str]:
"""Return the ``agent:`` frontmatter override, if any."""
return self.frontmatter.get('agent')
@property
def provider(self: Step) -> Optional[str]:
"""Return the ``provider:`` frontmatter override, if any."""
return self.frontmatter.get('provider')
@property
def model(self: Step) -> Optional[str]:
"""Return the ``model:`` frontmatter override, if any."""
return self.frontmatter.get('model')
@property
def effort(self: Step) -> Optional[str]:
"""Return the ``effort:`` frontmatter override, if any."""
return self.frontmatter.get('effort')
@property
def timeout(self: Step) -> Optional[str]:
"""Return the ``timeout:`` frontmatter override, if any."""
return self.frontmatter.get('timeout')
@property
def detached_raw(self: Step) -> Optional[str]:
"""Return the raw ``detached:`` frontmatter value, if any."""
return self.frontmatter.get('detached')
[docs]
@dataclasses.dataclass(frozen=True)
class StepResult:
"""Outcome of one step launch: status, reason, session, cost.
The in-process attribution record for a launch -- deadline overruns,
budget skips, pause aborts, and failures land here as statuses, never
as process exit sentinels or marker files.
``status`` is one of ``'completed' | 'timed out' | 'skipped' |
'paused' | 'failed'``; ``reason`` carries the fractal-owned label
that lands in step-row metadata (``'agent error (exit N)'`` /
``'stream error'``), suffixed with the one-line cause when the
launch captured one (the stderr tail or the stream exception) -- the
iter/run rollups keep the bare label (a budget skip's ``'over
budget'`` label is applied at step close, not carried here). A
budget-cap stop is a clean ``completed`` with ``budget_stopped``
set, never a failure.
"""
status: str
exit_code: int = 0
reason: Optional[str] = None
session: Optional[str] = None
cost: Optional[float] = None
budget_stopped: bool = False
[docs]
class Loop:
"""One node's agent iteration loop, run inside the node's tmux pane.
Owns preflight, run adoption, step execution, budget/deadline
policy, and the terminal cascade. All state flows through ``Node``
(rows, signals, config) and the Agent seam (invocation, streaming,
cost); shell remains only at the process boundary (setup/lint
hooks, the agent binary itself).
Observability rides the on_iteration/on_step event pairings --
override them (calling super and returning the event) to ship
structured loop progress to a host application; stdout stays the
byte-exact operator transcript.
"""
def __init__(
self: Loop,
node: Node,
*,
agent_command: Optional[str] = None,
continue_: bool = False,
resume: bool = False,
render: Optional[Callable[[StreamEvent], None]] = None,
) -> None:
"""Initialize ``Loop``.
Args:
node: The node whose loop to run.
agent_command: Agent invocation (e.g. ``'claude'``); defaults
to the node's configured agent.
continue_: Continue a stopped/exited node (clean worktree,
further iterations).
resume: Resume a paused node (adopt its open run where the
pause left it).
render: Presentation callback for parsed agent stream events
(the pane rendering); ``None`` renders nothing.
Raises:
ValueError: If no agent is configured or the agent command
names no registered backend, or a run parameter is
invalid (a malformed duration, conflicting schedules).
"""
# bind the node and the pane renderer
self.node = node
self._render = render
self._continue = continue_
self._resume = resume
# the agent comes from the argument or the node config (--agent at init)
config = node.config
self._agent_command = agent_command or config.get('agent')
if not self._agent_command:
raise ValueError('No agent configured; set --agent at node init.')
# resolve the backend now so an unsupported agent fails before any row
# is opened (the registry's error names the supported commands)
self._agent: Agent = node.agent(self._agent_command)
# read the budget caps (re-read at each iteration top for retunes)
self._read_cost_caps()
# label the per-step cost cap honestly: enforced for agents with a
# budget flag (claude, passed as --max-budget-usd), warn-only for
# agents without one (codex)
self._step_cost_verb = 'max' if self._agent.enforces_budget else 'warn'
# read the iteration cap and the wall-clock/schedule parameters
self._max_iters = int(config.get('max_iters') or -1)
self._timeout = config.get('timeout') or ''
self._iter_timeout = config.get('iter_timeout') or ''
self._step_timeout = config.get('step_timeout') or ''
self._interval = config.get('interval') or ''
self._sleep = config.get('sleep') or ''
self._wait = config.get('wait') or ''
self._run_timeout_seconds = _parse_duration(self._timeout, 'timeout')
self._iter_timeout_seconds = _parse_duration(self._iter_timeout, 'iter_timeout')
self._step_timeout_seconds = _parse_duration(self._step_timeout, 'step_timeout')
self._interval_seconds = _parse_duration(self._interval, 'interval')
self._sleep_seconds = _parse_duration(self._sleep, 'sleep')
self._wait_seconds = _parse_duration(self._wait, 'wait')
if self._wait_seconds <= 0:
self._wait_seconds = 60
# failed-step retries: extra attempts a failed launch gets (0
# disables) and the backoff before each retry; clamp a hand-edited
# negative to 0 -- an empty attempt loop would crash the iteration
step_retries = config.get('step_retries')
self._step_retries = 1 if step_retries is None else max(0, int(step_retries))
self._step_retry_backoff = config.get('step_retry_backoff') or '10s'
self._step_retry_backoff_seconds = _parse_duration(
value=self._step_retry_backoff,
label='step_retry_backoff',
)
# --interval and --sleep are mutually exclusive (checked after
# durations are parsed, so the *_SECONDS values are real)
if self._interval_seconds > 0 and self._sleep_seconds > 0:
raise ValueError('--interval and --sleep are mutually exclusive.')
# interval defaults the per-iteration deadline to the slot (an
# iteration cannot run past its cadence) -- but only when none was
# set: an explicit tighter iter_timeout is honored, never loosened to
# the full interval; a looser one is rejected at the config boundary
# (Config.validate); this raise backstops direct loop construction
if self._interval_seconds > 0:
if self._iter_timeout_seconds > self._interval_seconds:
raise ValueError(
f'--iter-timeout ({self._iter_timeout}) exceeds'
f' --interval ({self._interval}).'
)
if self._iter_timeout_seconds <= 0:
self._iter_timeout = self._interval
self._iter_timeout_seconds = self._interval_seconds
# SYNC_MODE gates the SYNC step run before each step; unlike the
# prompt-injection modes (DETACHED_MODE/CONTINUE_MODE/RESUME_MODE/
# RESERVE_MODE/META_MODE) SYNC.md is run as its own step, not appended
# to step prompts (the prompt builder skips it)
sync = config.get('sync')
self._sync_mode = True if sync is None else bool(sync)
# pin the boot-time modes so a mid-run config edit cannot flip a run's
# mode composition (detached drives the run mode, meta the injection)
self._detached = bool(config.get('detached', False))
self._node_model = config.get('model') or ''
self._node_effort = config.get('effort') or ''
meta = config.get('meta') or ''
self._meta_mode = bool(meta)
# the boot-static env/prompt vocabulary, derived once (the single
# var-map derivation) and pinned for the run
self._env_base = _boot_env(node, detached=self._detached, meta=meta)
# run/iteration lifecycle state
self._iter = 0
self._iter_timestamp = ''
self._iter_ref = ''
self._iter_label = ''
self._timed_out = False
self._paused = False
self._last_iter_failed = False
self._run_end_epoch = 0
self._iter_end_epoch = 0
self._run_id: Optional[int] = None
self._iter_id: Optional[int] = None
self._iteration_event: Optional[LoopIterationEvent] = None
self._adopt_iter_id: Optional[int] = None
self._resume_step: Optional[int] = None
self._resume_mode = resume
self._reserve = False
self._budget_hit = False
self._budget_reason = ''
self._budget_stem = ''
self._cap_overshoot = ''
self._untracked_warned = False
self._unreadable_warned = False
self._soft_cap_warned = False
# last good ledger readings -- a failed read falls back to these,
# never to the full cap
self._last_run_spent: Optional[float] = None
self._last_iter_headroom: Optional[float] = None
self._setup_fails = 0
self._setup_abort = False
self._fail_reason: Optional[str] = None
self._time_budget = 'no limit'
self._cost_budget = 'no limit'
# per-launch state (the exported step vocabulary)
self._step_label = ''
self._step_id: Optional[int] = None
self._step_model = ''
self._step_effort = ''
self._step_detached = False
self._step_launched = False
self._step_limit_seconds = 0
self._step_timeout_effective = self._step_timeout_seconds
self._step_budget_value: Optional[float] = None
# the immutable mode docs run from the installed package,
# never a per-node copy
self._modes_dir = node._package_dir / '_node' / 'modes'
self._sync_file = self._modes_dir / 'SYNC.md'
# whether any step will need the pricing cache (resolved in preflight)
self._needs_pricing = False
# armed caps with a non-enforcing agent and no timeout have no
# in-step brake -- disclose the unbounded-overshoot exposure once
self._warn_soft_cap()
[docs]
@functools.cached_property
def logger(self: Loop) -> logging.Logger:
"""Return the stdlib logger named for the concrete class."""
name = f'{type(self).__module__}.{type(self).__name__}'
return logging.getLogger(name)
[docs]
def log(self: Loop, message: str, level: int = logging.INFO) -> None:
"""Log a message to this loop's logger."""
self.logger.log(level, message)
[docs]
def run(self: Loop) -> int:
"""Execute the loop end to end and return the process exit code.
Body is guarded by try/finally -> ``_on_exit`` (the EXIT-trap
mirror); no signal handlers are installed -- SIGTERM kill
classification belongs to kill.sh's Python and out-of-band
deaths to ``Node._reconcile_status``.
"""
node = self.node
# the pane transcript must stream live (operators and the e2e harness
# tail it mid-run): line-buffer stdout even when it is a pipe/file
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(line_buffering=True)
# owning node for every fractal call this loop spawns (caller
# resolution + the self-finish reconcile guard); start.sh sets it
# before the re-entry exec, but set it here too so a
# direct launch self-identifies
os.environ['_NODE'] = f'{node.node_dir}'
if self._continue:
self._clean_worktree()
self._labels()
try:
self._preflight()
self._adopt()
except _Abort:
return 1
# record the tmux socket the session lives on ($TMUX is
# "socket_path,server_pid,session_index" inside the pane), before
# the active stamp so no reconcile ever probes an active node
# without it: Node._reconcile_status asks this server, not
# whichever socket the probing shell resolves, so a shell with a
# different TMUX_TMPDIR never misreads the live session as gone; a
# tmux-less launch (the test harness) drops any stale record so
# the ambient probe rules
try:
tmux_env = os.environ.get('TMUX', '')
if tmux_env:
socket_path, *_ = tmux_env.split(',')
(node.node_dir / SOCKET_FILE).write_text(
f'{socket_path}\n',
encoding='utf-8',
)
else:
(node.node_dir / SOCKET_FILE).unlink(missing_ok=True)
except OSError:
pass
# stamp the node active -- under the .worktrees flock, so a retire
# that won the boot window (its flock-guarded read legally saw
# 'idle' after start's checks but before this stamp) is honored
# instead of clobbered; guarded so a transient failure (a DB
# contended under wide fan-out) never aborts the run and strands
# .status at 'idle' (a missed stamp self-corrects as the loop runs)
retired = False
try:
with worktree.lock(node.repo_dir):
if node.status() == 'retired':
retired = True
else:
node.status_set('active')
except Exception:
pass
if retired:
print('=== Stood down at boot: node was retired ===')
try:
node.record.run_end(
run_id=self._run_id,
status='exited',
exit_code=1,
metadata='retired before boot',
)
except Exception:
pass
return 1
# record the run's process group beside .status -- the handle kill.sh
# and Node._reconcile_status fall back to when an out-of-band pane
# death leaves the agent group running headless; removed by _on_exit
# below, so a surviving file marks a death no cleanup could catch
# (SIGKILL / host crash)
try:
pgid = os.getpgid(0)
(node.node_dir / PGID_FILE).write_text(f'{pgid}\n', encoding='utf-8')
except OSError:
pass
# an in-loop abort before the terminal cascade (an unhandled error)
# would otherwise strand .status at 'active'; on exit, stamp the
# honest terminal -- but only when the cascade never recorded one
# (status still 'active'), so neither a clean exit nor a transient
# failure of the cascade's own status write (a contended DB) gets
# relabeled; (a SIGTERM kill is handled by kill.sh's Python; a
# SIGKILL / host crash can't run this -- those are the job of
# Node._reconcile_status at the next reject-active op)
# NOTE: the finally also runs on KeyboardInterrupt -- a deliberate parity
# delta from the untrapped shell fatal-signal path, so an attached-pane
# Ctrl-C stamps `exited` instead of waiting for reconcile
try:
if self._park_if_latched():
return 0
# compute the whole-run deadline once -- the per-iteration
# deadline resets each pass, but the run wall clock is fixed for
# this invocation; an adopted run anchors on the credited
# reading, which credits the paused spans (wall clock alone
# would charge the frozen time against the deadline)
if self._run_timeout_seconds > 0:
remaining = None
if self._resume:
try:
remaining = node.time.remaining(scope='run')
except Exception:
remaining = None
if remaining is not None:
self._run_end_epoch = int(time.time()) + int(remaining)
else:
self._run_end_epoch = int(time.time()) + self._run_timeout_seconds
else:
self._run_end_epoch = 0
self._main_loop()
return self._finalize()
except _Abort:
return 1
except BaseException as error:
# an unhandled loop error escaping mid-iteration leaves the
# initial event dangling -- fire the failure pairing for it
# before the crash path (the finally below) settles the run
if self._iteration_event is not None:
self.on_iteration_failure(
initial_event=self._iteration_event,
error=error,
)
raise
finally:
self._on_exit()
def _preflight(self: Loop) -> None:
"""Fail fast (step files, binary, pricing, provider probe via seam hooks).
Aborts stamp ``exited`` + reason, never strand ``idle``.
"""
node = self.node
# pricing.json is needed when a step will run an agent that needs
# pricing with a model to price -- the step's own model, the node
# default, or a routed backend's pinned default; an agent that
# reports its own cost natively, or a token-priced agent with no
# model, gives pricing nothing to do
self._needs_pricing = False
for path in sorted((node.node_dir / 'steps').glob('*.md')):
# step files are hand-edited: a bad byte must abort through the
# machinery below -- a bare UnicodeDecodeError names a byte
# offset but no file, and would escape before the run row or
# the active stamp exists, stranding .status at idle
try:
step = Step.load(path)
except UnicodeDecodeError as error:
reason = f'Step file {path.name} is not valid UTF-8.'
print(f'Error: {reason} ({error})', file=sys.stderr)
self._abort_preflight(reason)
command = step.agent or self._agent_command
try:
backend = node.agent(command, step.provider)
except ValueError:
continue
if not backend.tracks_cost():
if step.model or self._node_model or backend.provider is not None:
self._needs_pricing = True
break
# refresh model pricing before the run; a fetch failure with no cache
# is fatal -- the cost/cap pipeline cannot price token usage without it
if self._needs_pricing:
status = pricing.update()
if status == 'missing':
reason = 'Could not fetch pricing and no cached pricing.json exists.'
print(f'Error: {reason}', file=sys.stderr)
self._abort_preflight(reason)
if status == 'stale':
print(
'Warning: could not refresh pricing; using cached pricing.json.',
file=sys.stderr,
)
# warn when the agent's cost is priced from a model but none is set
# (and no cap forces the issue) -- its spend would silently go untracked
caps = (self._max_cost, self._max_iter_cost, self._max_step_cost)
no_caps = all(cap is None for cap in caps)
if self._agent.needs_pricing and not self._node_model and no_caps:
print(
f'Warning: no model set for {self._agent.name}; its cost'
' cannot be priced and will not be tracked',
file=sys.stderr,
)
# provider preflight via the seam: the binary must be on PATH, and a
# token-priced agent with an explicit model is probed once (some codex
# accounts reject some explicit models; the pricing check only proves
# the model priceable, not that the account accepts it) -- the abort
# persists the probe's first line as the reason and relays the full
# diagnosis on stderr
try:
self._agent.preflight(self._node_model or None)
except RuntimeError as error:
print(f'Error: {error}', file=sys.stderr)
reason, *_ = f'{error}'.split('\n')
self._abort_preflight(reason)
def _abort_preflight(
self: Loop,
reason: str,
*,
force_record: bool = False,
) -> None:
"""Abort a failed preflight loudly and recoverably.
The probe runs before the run row and the ``active`` stamp, so a
bare exit would strand ``.status`` at ``idle`` (indistinguishable
from a never-started node) and lose the diagnosis in the dying
tmux pane -- instead persist a short reason on a closed run row
(surfaced by ``node activity``) and stamp the honest terminal
``exited``, which both names the failure and unwedges recovery
(``--continue`` accepts ``exited``; a plain start refuses with a
restart hint rather than silently re-failing).
``force_record`` overrides the resume-preserve guard for a resume
boot that has no paused run to protect (``_adopt`` found none open):
there the durable exited record must land, or the node wedges
``paused`` with no reason and only ``kill`` recovers.
"""
# a resume boot preserves its paused run: the open run and iteration
# rows are exactly what `resume` re-adopts, and the frozen worktree is
# what pause preserves -- recording a fresh exited run would reconcile
# them shut, leaving `--continue` (which discards the frozen work) as
# the only recovery; the caller already surfaced the reason, so leave
# the node paused and adoptable for a fixed-environment resume; the
# exception is a resume with no open run to adopt (force_record): there
# is nothing to preserve, so record and stamp exited like a fresh boot
if self._resume and not force_record:
# re-open the pause credit span: `Node._resume` stamps the resume
# event completed when tmux comes up (closing the span), before this
# preflight runs, so the wait until a fixed-environment resume would
# otherwise charge the run/iter deadlines -- timing the run out on
# the very recovery this guard preserves; best-effort like the
# boot-path writes below: a transient DB error must not turn the
# clean park into a crash (the node stays paused either way)
try:
event_id = self.node.record.event_start(
'pause',
metadata=f'resume preflight failed: {reason}',
)
self.node.record.event_end(event_id=event_id, status='completed')
except Exception:
pass
raise _Abort
# record the reason on a closed run row -- the durable home
# `node status`/`activity` surface it from
try:
run_id = self.node.record.run_start()
self.node.record.run_end(
run_id=run_id,
status='exited',
exit_code=1,
metadata=reason,
)
except Exception:
pass
try:
self.node.status_set('exited')
except Exception:
pass
raise _Abort
def _park_if_latched(self: Loop) -> bool:
"""Park a boot into a pausing/paused subtree.
Resume relaunches are exempt from the ancestor walk but not a
new tree brake.
"""
node = self.node
# a loop that boots into a pausing/paused subtree parks immediately:
# the pause fan-out cannot have signaled a node whose start was still
# in flight when it swept; a resume relaunch is exempt from the
# ancestor walk (its fan-out is leaf-first, so ancestors legitimately
# still read paused while a child boots) but not from a NEW tree-wide
# brake landing during the relaunch window
try:
latched = node.pause_latched(tree_only=self._resume)
except Exception:
latched = None
if latched is None:
return False
print(f'=== Parked at boot: {latched} is paused ===')
try:
node.record.signal_set('pause', f'via pause latch ({latched})')
except Exception:
pass
# open the pause span -- without the instant, the parked time would
# burn against this run's deadline at resume
try:
event_id = node.record.event_start(
'pause',
metadata=f'via pause latch ({latched})',
run_id=self._run_id,
)
if event_id is not None:
node.record.event_end(event_id=event_id, status='completed')
except Exception:
pass
try:
node.status_set('paused')
except Exception:
pass
return True
def _adopt(self: Loop) -> None:
"""Adopt a paused run (resume) or open a fresh run row."""
node = self.node
if not self._resume:
self._run_id = node.record.run_start()
return
# adopt the paused run on a resume relaunch: pause parks with the run
# and iteration rows open, and a fresh run start would close them as
# orphaned and re-arm the full budget -- the resume boot must reuse
# them instead; the adoption context (open run, newest iteration,
# re-entry step) is resolved by the record surface
try:
adopt = node.record.run_open()
except Exception:
adopt = None
if adopt is None:
# one retry insulates a transient read failure (a contended DB)
# from the honest no-open-run abort below
time.sleep(2)
try:
adopt = node.record.run_open()
except Exception:
adopt = None
if adopt is None:
print(
'Error: no open run to adopt -- was the node paused?',
file=sys.stderr,
)
# no paused run to preserve here, so record the honest exited row
# (a fixed-environment resume can't help -- there is nothing to
# adopt -- but --continue can recover)
self._abort_preflight('no open run to adopt', force_record=True)
self._run_id = adopt['run_id']
if adopt['iter_id'] is not None:
# the pause landed inside this iteration: reuse its open row (the
# loop's increment lands back on its number)
self._iter = adopt['iter'] - 1
self._adopt_iter_id = adopt['iter_id']
self._resume_step = adopt['resume_step']
elif adopt['iter'] is not None:
# boundary pause: the iteration closed before the park -- continue
# the count from it (the increment opens the next one)
self._iter = adopt['iter']
# withdraw the pause signals that parked this run (the finish_cancel
# precedent) -- the first checkpoint would otherwise re-park on them;
# done here, not by the resume CLI, so a bare --resume launch
# (e.g. after a filesystem transplant) self-clears too
try:
node.record.signal_clear('pause', run_id=self._run_id)
except Exception:
pass
# close the pause span for the deadline credit even when no resume CLI
# ran (a bare --resume launch) -- a resume event with no open pause
# is inert to the credit walk, so the CLI-then-boot double write is safe
try:
event_id = node.record.event_start('resume', run_id=self._run_id)
if event_id is not None:
node.record.event_end(event_id=event_id, status='completed')
except Exception:
pass
print(f'Resuming run {self._run_id} where the pause left it')
def _clean_worktree(self: Loop) -> None:
"""Restore the worktree for a continue launch.
Continue-mode cleanup: commit operator seed edits --no-verify,
clean, re-arm budgets, preserve config.json.
"""
node = self.node
worktree = node.worktree
node_dir = node.node_dir
# preserve operator edits: the documented steering flow edits node-dir
# files between runs without committing, and the checkout/clean below
# would revert them -- commit them first; --no-verify like every
# backstop save (a host hook must not veto it); unguarded: failing
# loud beats destroying the edits
cmd = ['status', '--porcelain', '--', f'{node_dir}']
if fractal.util.git.run(cmd, cwd=worktree, check=False):
print(
f'Continuing: committing operator edits under .fractal/{node.branch}...'
)
fractal.util.git.run(['add', '-A', '--', f'{node_dir}'], cwd=worktree)
cmd = [
'commit',
'--no-verify',
'-m',
f'{node.branch}: operator edits (committed at continue)',
]
fractal.util.git.run(cmd, cwd=worktree)
print('Continuing: cleaning uncommitted changes...')
# preserve config.json across the clean -- the documented way to
# re-tune before continuing (e.g. adjust max_iters), and the agent
# never owns it; the iteration count is not reseeded, so each run
# counts from 1 (max_iters caps iterations per run)
config_path = node_dir / 'config.json'
try:
backup = config_path.read_bytes()
except OSError:
backup = None
fractal.util.git.run(['checkout', '--', '.'], cwd=worktree, check=False)
fractal.util.git.run(['clean', '-fd'], cwd=worktree, check=False)
if backup is not None:
try:
# atomic, so a kill mid-restore cannot tear the file a later
# node load would refuse to parse
fractal.util.filesystem.write_atomic(config_path, backup)
except OSError:
pass
def _labels(self: Loop) -> None:
"""Print the startup banner from the boot-pinned run parameters."""
max_label = f'{self._max_iters}' if self._max_iters > 0 else 'unlimited'
timeout_label = self._timeout if self._run_timeout_seconds > 0 else 'none'
if self._iter_timeout_seconds > 0:
iter_timeout_label = self._iter_timeout
else:
iter_timeout_label = 'none'
if self._step_timeout_seconds > 0:
step_timeout_label = self._step_timeout
else:
step_timeout_label = 'none'
interval_label = self._interval if self._interval_seconds > 0 else 'none'
sleep_label = self._sleep if self._sleep_seconds > 0 else 'none'
if self._max_cost is not None:
cost_label = f'${self._max_cost_label}'
else:
cost_label = 'none'
if self._max_iter_cost is not None:
iter_cost_label = f'${self._max_iter_cost_label}/iter'
else:
iter_cost_label = 'none'
if self._max_step_cost is not None:
step_cost_label = f'${self._max_step_cost_label}/step'
else:
step_cost_label = 'none'
run_mode_label = 'detached' if self._detached else 'continuous'
continue_label = 'yes' if self._continue else 'no'
sync_label = 'yes' if self._sync_mode else 'no'
wait_label = self._wait if self._wait else '1m'
print(f'Starting node on {self.node.node_dir} with {self._agent_command}')
print(
f' iterations: {max_label} | timeout: {timeout_label}'
f' | iter-timeout: {iter_timeout_label}'
f' | step-timeout: {step_timeout_label}'
f' | interval: {interval_label} | sleep: {sleep_label}'
f' | max-cost: {cost_label} | max-iter-cost: {iter_cost_label}'
f' | max-step-cost: {step_cost_label}'
f' | mode: {run_mode_label} | continue: {continue_label}'
f' | sync: {sync_label} | wait: {wait_label}'
)
def _main_loop(self: Loop) -> None:
"""Run iterations until a stop gate, signal, or park breaks out."""
node = self.node
while True:
# re-read max_iters here so a mid-run retune reaches the stop gate
# below (the cost-cap re-read sits after that gate, so it cannot
# cover this); skip the first pass -- run start read the same
# config moments earlier; config is a live-edited steering
# surface, so a malformed edit warns and keeps the prior value
# instead of crashing the run
if self._iter > 0:
try:
self._max_iters = int(node.config.get('max_iters') or -1)
except (TypeError, ValueError, OverflowError):
print(
'Warning: max_iters must be an integer;'
' keeping the previous value',
file=sys.stderr,
)
if self._max_iters > 0 and self._iter >= self._max_iters:
print(f'Reached max iterations ({self._max_iters}). Stopping.')
break
# whole-run wall clock: stop before starting another iteration
# past the deadline
if self._run_end_epoch > 0 and time.time() >= self._run_end_epoch:
print(f'Reached run timeout ({self._timeout}). Stopping.')
self._timed_out = True
break
self._timed_out = False
self._iter += 1
self._iter_timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
self._iter_ref = f'{self._run_id}.{self._iter}'
self._iter_label = self._iter_label_text()
# reset the per-iteration deadline (the run deadline is fixed for
# the run); an adopted iteration anchors on the credited reading,
# like the run
if self._iter_timeout_seconds > 0:
remaining = None
if self._adopt_iter_id is not None:
try:
remaining = node.time.remaining(scope='iter')
except Exception:
remaining = None
if remaining is not None:
self._iter_end_epoch = int(time.time()) + int(remaining)
else:
self._iter_end_epoch = int(time.time()) + self._iter_timeout_seconds
else:
self._iter_end_epoch = 0
# re-read the retunable pacing knobs -- ahead of the time-budget
# label build so a mid-run wait/step_timeout retune (config edit
# or node update) reaches this iteration's label, launches, and
# drains; iteration 1 skips it like the cap re-read below -- run
# start read the same config moments earlier; a malformed edit
# warns and keeps the prior value (mirrors the frontmatter
# timeout fallback), never crashing the run
if self._iter > 1:
try:
step_timeout = node.config.get('step_timeout') or ''
self._step_timeout_seconds = _parse_duration(
value=step_timeout,
label='step_timeout',
)
self._step_timeout = step_timeout
except ValueError as error:
print(
f'Warning: {error}; keeping the previous step_timeout',
file=sys.stderr,
)
try:
wait = node.config.get('wait') or ''
self._wait_seconds = _parse_duration(wait, 'wait')
self._wait = wait
except ValueError as error:
print(
f'Warning: {error}; keeping the previous wait',
file=sys.stderr,
)
if self._wait_seconds <= 0:
self._wait_seconds = 60
# build the time-budget label from whichever of the
# run/iter/step timeouts are set
time_budget = ''
if self._timeout:
time_budget = f'{self._timeout} total'
if self._iter_timeout:
if time_budget:
time_budget += ', '
time_budget += f'{self._iter_timeout}/iter'
if self._step_timeout:
if time_budget:
time_budget += ', '
time_budget += f'{self._step_timeout}/step'
self._time_budget = time_budget or 'no limit'
# refresh the budget caps and heal registry drift: the boundary
# checks take their own fresh reads, so this read feeds the
# iteration's budget labels and per-step reserve derivation, and
# a config-vs-registry mismatch must warn loudly, not linger;
# iteration 1 skips both -- run start read the same config
# moments earlier; a malformed edit warns and keeps the prior
# caps, like the retunes above
if self._iter > 1:
try:
self._read_cost_caps()
except ValueError as error:
print(
f'Warning: {error}; keeping the previous cost caps',
file=sys.stderr,
)
self._reconcile_caps()
# a cap granted mid-run can arm the soft-cap exposure late
self._warn_soft_cap()
self._build_cost_budget()
print()
print(f'=== Iteration {self._iter_label} at {self._iter_timestamp} ===')
# check signals before starting (pause outranks stop/finish: it
# parks the run with the other signals intact, to fire after
# resume) -- except on an adopted iteration: it IS the current
# iteration a pending finish or stop lets run out, so only pause
# can pre-empt its re-entry (the post-iteration checks below
# still fire once it closes)
if self._check_pause():
self._paused = True
break
if self._adopt_iter_id is None:
if self._check_finish():
# the run-end drain is bounded by the run wall alone --
# clear the iteration deadline armed above so it cannot
# time out a drain over still-active children (mirrors the
# post-iteration clear); the iteration that would bound it
# never runs (the break precedes iter_start)
self._iter_end_epoch = 0
self._wait_for_children('run end')
break
if self._check_stop():
break
# reset the per-iteration session map (sessions are per-agent,
# per-iteration) -- except when adopting a paused iteration, whose
# map holds the interrupted step's resumable session
if not self._detached and self._adopt_iter_id is None:
try:
node.sessions.clear()
except Exception:
pass
# refresh stale pricing for long-running token-reporting nodes
if self._needs_pricing:
status = pricing.update(max_age='24h')
if status == 'missing':
raise RuntimeError(
'Could not fetch pricing and no cached pricing.json exists.'
)
if status == 'stale':
print(
'Warning: could not refresh pricing;'
' using cached pricing.json.',
file=sys.stderr,
)
# reuse the adopted iteration's open row; a fresh iteration
# opens its own
if self._adopt_iter_id is not None:
self._iter_id = self._adopt_iter_id
else:
try:
self._iter_id = node.record.iter_start(
run_id=self._run_id,
iter=self._iter,
)
except Exception:
self._iter_id = None
if self._iter_id is None:
print(
f'Error: failed to start iteration {self._iter}',
file=sys.stderr,
)
break
iteration_event = self.on_iteration(
iteration=self._iter,
run_id=self._run_id,
iter_id=self._iter_id,
)
# the open initial event backs the crash pairing: an unhandled
# error escaping this body leaves it set, and run()'s abnormal
# path fires the failure pairing for it
self._iteration_event = iteration_event
# run setup -- setup.sh is the mutable, agent-editable node copy,
# so a bad edit must record a clean iteration failure (and run the
# commit backstop + iteration close below), not abort the whole
# loop and strand the open iter/step rows for reconcile to settle
# at the next start
setup_failed = not self._run_setup()
iter_start = time.monotonic()
if setup_failed:
print(
'Error: setup.sh failed; skipping this iteration',
file=sys.stderr,
)
iter_failed = True
self._setup_fails += 1
else:
self._setup_fails = 0
iter_failed = not self._iterate()
iter_duration = int(time.monotonic() - iter_start)
# park mid-iteration: skip the commit backstop (the dirty worktree
# is the frozen mid-step state resume continues from) and leave
# the iteration row open for resume to adopt
if self._paused:
self.on_iteration_success(
initial_event=iteration_event,
outcome='paused',
)
self._iteration_event = None
break
# the re-entry marker and resume framing apply to the adopted
# pass only; later iterations run normally, with normal prompts
self._resume_step = None
self._resume_mode = False
self._adopt_iter_id = None
# ensure iteration committed
if not self._commit_check():
self._force_commit('auto')
iter_reason = None
if iter_failed:
iter_status = 'failed'
iter_exit_code = 1
# a short fractal-owned reason, mirroring the step close (no
# provider blob); prefer the recorded launch/discovery reason
# so it names the real source (agent error / stream error /
# no step files), not a guess
if self._timed_out:
# the step/SYNC timeout branches record the enriched
# '<step> timed out (<limit>)' reason; a wait-deadline
# overrun has no launch to name
iter_reason = self._fail_reason or 'timed out'
elif setup_failed:
# carry the tail of the captured setup output so the
# actual error is durable in the iteration metadata, not
# just the (mortal) tmux tty
tail = _log_tail(node.node_dir / 'setup.log')
iter_reason = 'setup failed'
if tail:
iter_reason = f'setup failed: {tail}'
elif self._fail_reason:
iter_reason = self._fail_reason
else:
iter_reason = 'agent error'
elif self._signal('stop'):
iter_status = 'stopped'
iter_exit_code = 0
else:
iter_status = 'completed'
iter_exit_code = 0
# carried to the finalize matrix: a failed final iteration must
# not be laundered into a clean max-iters completion
self._last_iter_failed = iter_failed
try:
node.record.iter_end(
iter_id=self._iter_id,
status=iter_status,
exit_code=iter_exit_code,
metadata=iter_reason,
)
except Exception:
pass
# the iteration is closed: clear its id so the run-end child
# drains below book no zombie SYNC step rows (the drain-wait SYNC
# gate keys on the open iteration), and its deadline so those
# drains are bounded by the run wall alone -- a stale iteration
# deadline would time a drain out over still-active children
self._iter_id = None
self._iter_end_epoch = 0
print()
if iter_failed:
print(f'=== Iteration {self._iter_label} failed ({iter_duration}s) ===')
error = RuntimeError(iter_reason or 'iteration failed')
self.on_iteration_failure(
initial_event=iteration_event,
error=error,
)
else:
print(
f'=== Iteration {self._iter_label} completed ({iter_duration}s) ==='
)
self.on_iteration_success(
initial_event=iteration_event,
outcome=iter_status,
)
self._iteration_event = None
# consecutive-setup-failure cap: end the run rather than grind the
# remaining iterations through the same broken setup
if self._setup_fails >= _SETUP_FAIL_CAP:
print(
f'=== Setup failed {self._setup_fails} consecutive times,'
f' ending run ==='
)
self._setup_abort = True
break
# total-cost reserve stop: the just-finished iteration ran the
# RESERVE wind-down, so if the run has entered the reserve window
# end it here rather than start another iteration that would only
# re-enter reserve (this subsumes the hard ceiling at the
# boundary); the finish it sends is caught by the finish check
# just below, which drains children and breaks the loop
self._check_reserve_boundary()
# check signals after iteration (pause first, mirroring
# the pre-iteration order)
if self._check_pause():
self._paused = True
break
if self._check_finish():
self._wait_for_children('run end')
break
if self._check_stop():
break
# sleep between iterations
if self._max_iters <= 0 or self._iter < self._max_iters:
sleep_amount = 0
sleep_label_text = ''
if self._interval_seconds > 0:
interval_sleep = self._interval_seconds - iter_duration
if interval_sleep > 0:
sleep_amount = interval_sleep
sleep_label_text = (
f'{interval_sleep}s (next iteration in {self._interval})'
)
elif self._sleep_seconds > 0:
sleep_amount = self._sleep_seconds
sleep_label_text = self._sleep
# never sleep past the run deadline (keeps --timeout
# a true wall)
if sleep_amount > 0 and self._run_end_epoch > 0:
run_remaining = self._run_end_epoch - int(time.time())
if run_remaining <= 0:
sleep_amount = 0
elif sleep_amount > run_remaining:
sleep_amount = run_remaining
if sleep_amount > 0:
print(f'Sleeping {sleep_label_text}...')
# sleep in chunks, polling every signal -- an interval node
# would otherwise not react (park, stop, or drain-to-finish)
# until its next wake, hours away
slept = 0
interrupted = False
while slept < sleep_amount:
chunk = min(30, sleep_amount - slept)
time.sleep(chunk)
slept += chunk
signals = ('pause', 'finish', 'stop')
interrupted = any(self._signal(name) for name in signals)
if interrupted:
break
# handle a signal that landed during the sleep exactly like
# the post-iteration checks (the iteration deadline is
# already cleared, so a finish drain uses the run wall)
if interrupted:
if self._check_pause():
self._paused = True
break
if self._check_finish():
self._wait_for_children('run end')
break
if self._check_stop():
break
def _discover_steps(self: Loop) -> Optional[list[Step]]:
"""Re-discover the NN- step files (lexicographic order).
Violations -- an empty dir, a missing prefix, mixed prefix
widths -- fail the iteration loudly.
"""
node_dir = self.node.node_dir
paths = sorted((node_dir / 'steps').glob('*.md'))
if not paths:
print(f'Error: no step files found in {node_dir}/steps/', file=sys.stderr)
self._fail_reason = 'no step files'
return None
steps = []
prefix_width = None
for index, path in enumerate(paths):
match = _STEP_PREFIX.match(path.name)
if match is None:
print(
f'Error: step file without NN- prefix: {path.name}',
file=sys.stderr,
)
self._fail_reason = 'invalid step files'
return None
width = len(match.group(1))
if prefix_width is None:
prefix_width = width
elif width != prefix_width:
print(
f'Error: inconsistent digit prefix widths in {node_dir}/steps/',
file=sys.stderr,
)
self._fail_reason = 'invalid step files'
return None
steps.append(Step.load(path, number=index + 1))
return steps
def _iterate(self: Loop) -> bool:
"""Run one iteration's steps.
SYNC-before-step, signal checkpoints, approval gates,
failed-launch retries, and honest attribution; the on_iteration
pairing brackets this body from the main loop.
Returns:
Whether the iteration succeeded (``False`` marks it failed).
"""
node = self.node
# start each iteration with a clean failure reason, so a stale reason
# from a prior iteration/run is never misattributed to this one
self._fail_reason = None
# discover steps
steps = self._discover_steps()
if steps is None:
return False
step_count = len(steps)
started = False
self._reserve = False
# a fresh iteration voids the cached headroom
self._last_iter_headroom = None
for step in steps:
step_num = step.number
step_name = step.name
self._step_label = f'step {step_num} of {step_count} ({step_name})'
# re-enter an adopted iteration at its interrupted step -- the
# steps before it already ran before the pause; a re-entry past
# the last step (an approved final step) means the iteration's
# work is done, so run none of it and let the
# iteration close normally
if self._resume_step is not None:
if self._resume_step > step_count:
self._resume_step = None
return True
if step_num < self._resume_step:
continue
self._resume_step = None
if not self._reserve and self._max_iter_cost is not None:
iter_spent = self._cost_spent(iter_id=self._iter_id)
if iter_spent is not None and iter_spent >= float(self._max_iter_cost):
self._reserve = True
# enter RESERVE when total cost drains into the reserve window --
# the buffer below max_cost that steers cleanup before the ceiling
# (may drain mid-iteration); the boundary then ends the run,
# never self-stop here
if not self._reserve and self._max_cost is not None:
remaining = self._run_remaining()
if remaining <= self._reserve_budget:
self._reserve = True
# enter RESERVE when an ancestor's budget abort left a pending
# finish -- the current iteration is the run's last (the
# post-iteration gate ends it), so the remaining steps run the
# wind-down; a deliberate finish leaves the steps plain
if not self._reserve:
try:
finish_reason = (
node.record.signal_get('finish', run_id=self._run_id) or ''
)
except Exception:
finish_reason = ''
if _is_cascaded_budget_reason(finish_reason):
self._reserve = True
# park before the step (step 1 included -- a pause landing during
# setup must not buy a whole agent turn); no commit: the dirty
# worktree is the frozen mid-iteration state resume continues from
if self._check_pause():
self._paused = True
return True
if step_num > 1:
if self._check_stop():
break
# hard subtree ceiling: a spawn-heavy iteration can blow the
# budget before the iteration-boundary check -- trip here too
# so a long iteration stops queuing steps sooner
if self._check_subtree_ceiling():
break
# --- SYNC before each step ---
if self._sync_mode and self._sync_file.is_file():
started = True
# the sync belongs to the step it precedes (the drain-wait
# sync, which precedes nothing, stays step 0)
result, duration, sync_step_id = self._sync(
step_num=step_num,
label=f'before {step_name}',
strict=True,
)
if result.status == 'timed out':
self._timed_out = True
# a deadline that expired before the launch resolved no
# ceiling, so the label skips the parenthetical
limit = ''
if self._step_limit_seconds > 0:
limit = f' ({self._step_limit_seconds}s)'
self._fail_reason = f'SYNC timed out{limit}'
print(f'--- SYNC (before {step_name}): timed out ---')
elif result.status == 'paused':
# the abort was pause's kill landing on the SYNC
# invocation -- park before launching the step's own agent
self._paused = True
print(f'--- SYNC (before {step_name}): paused ---')
elif result.status in ('failed', 'skipped'):
# a budget-skipped SYNC (exit 125) lands in the failure
# vocabulary here -- the skip banner belongs to the step
# it precedes, which prints its own just below
print(
f'--- SYNC (before {step_name}):'
f' exit {result.exit_code} ({duration}s) ---'
)
else:
print(f'--- SYNC (before {step_name}): done ({duration}s) ---')
if self._paused:
return True
# SYNC failure is non-fatal; timeout is fatal
if result.status == 'timed out':
print('--- Committing directly (SYNC timed out) ---')
self._force_commit(
'timed out during SYNC',
body=self._fail_reason,
step_id=sync_step_id,
)
return False
if result.status in ('failed', 'skipped'):
# the absorbed failure's launch reason must not relabel a
# later wait-deadline overrun in this iteration
self._fail_reason = None
print(f'--- SYNC failed (non-fatal), continuing to {step_name} ---')
# --- end SYNC ---
# if finishing, wait for children before last step
if step_num == step_count and self._signal('finish'):
# the drain is bounded by the run wall alone -- clear the
# iteration deadline so it cannot time out a wait over
# still-active children (mirrors the pre/post-iteration
# clears); the final step that follows runs off the step
# and run walls
self._iter_end_epoch = 0
if not self._wait_for_children(f'before {step_name}'):
# a pause parks the drain as-is (no commit); the finish
# signal survives on the adopted run and re-arms the
# wait after resume
if self._paused:
return True
if started:
print(
'--- Committing directly'
' (wait-for-children interrupted) ---'
)
self._force_commit('interrupted waiting for children')
# a stop interrupt is a requested end, not a failure --
# return success so the close books 'stopped'/0 off the
# still-set signal; a timeout keeps the failure path and
# its 'timed out' attribution
if self._signal('stop') and not self._timed_out:
return True
return False
started = True
# attempt loop: only a failed launch retries -- timed out, paused,
# and skipped are deliberate outcomes -- and every attempt books
# its own step row, so cost and duration attribute honestly
for attempt in range(self._step_retries + 1):
print()
print(f'--- Step {step_num}/{step_count} ({step_name}) ---')
try:
self._step_id = node.record.step_start(
iter_id=self._iter_id,
run_id=self._run_id,
step=step_num,
step_name=step_name,
)
except Exception:
self._step_id = None
# every attempt's row arms its own approval gate -- a retry
# only follows a failed attempt, whose wait never ran, so no
# granted approval is ever re-demanded; the fresh pend voids
# the failed row's stale gate
if step.requires_approval and self._step_id is not None:
try:
node.record.step_pending(step_id=self._step_id)
except Exception:
pass
step_start = time.monotonic()
step_budget_skip = False
result = self._run_step(step)
duration = int(time.monotonic() - step_start)
failed = False
if result.status == 'timed out':
self._timed_out = True
# name the step and its effective ceiling for the
# iteration close (the step row alone says 'timed out');
# a deadline that expired before the launch resolved no
# ceiling, so the label skips the parenthetical
limit = ''
if self._step_limit_seconds > 0:
limit = f' ({self._step_limit_seconds}s)'
self._fail_reason = f'{step_name} timed out{limit}'
step_status = 'exited'
step_exit_code = 1
failed = True
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' timed out ({duration}s) ---'
)
elif result.status == 'skipped':
# over budget in the finish wind-down: the step was
# skipped, not run -- record it stopped (flagging the
# reason) so the iteration winds down via the normal
# finish path (the force-commit backstop saves prior
# work), never the failure return below
step_status = 'stopped'
step_exit_code = 0
step_budget_skip = True
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' skipped (over budget) ---'
)
elif result.status == 'paused':
# the abort was pause's kill, not an agent failure: record
# the step paused (resume re-enters here on a fresh row)
# -- the failure path below, whose force-commit would bury
# the frozen mid-step worktree, never fires
step_status = 'paused'
step_exit_code = 0
self._paused = True
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' paused ({duration}s) ---'
)
elif result.status == 'failed':
step_status = 'failed'
step_exit_code = 1
failed = True
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' exit {result.exit_code} ({duration}s) ---'
)
else:
step_status = 'completed'
step_exit_code = 0
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' done ({duration}s) ---'
)
# --- approval wait loop ---
approval_interrupted = False
step_succeeded = not failed and not self._paused
approval_gated = step.requires_approval and self._step_id is not None
if step_succeeded and approval_gated and not self._reserve:
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' awaiting approval (step_id={self._step_id}) ---'
)
outcome = self._wait_for_approval(step)
approval_interrupted = outcome != 'approved'
if outcome == 'paused':
# a pause parks the wait as-is: the step's work is
# done but unapproved -- resume reads the recorded
# approval and either skips past this step or re-runs
# it (re-arming the wait)
step_status = 'paused'
step_exit_code = 0
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' paused awaiting approval ---'
)
elif outcome == 'timed out':
step_status = 'exited'
step_exit_code = 1
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' timed out waiting for approval ---'
)
failed = True
elif outcome == 'stopped':
# a stop/finish during approval wait is a clean
# interruption, not a failure: record the step stopped
# and let the iteration end stopped (the next stop
# check breaks) or completed (finish runs out via the
# finish check), never failed
step_status = 'stopped'
step_exit_code = 0
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' stopped waiting for approval ---'
)
else:
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' approved ---'
)
# --- end approval wait loop ---
# recompute duration (includes approval wait)
duration = int(time.monotonic() - step_start)
if self._step_id is not None:
# a one-line fractal-owned reason for failure visibility in
# `node activity` (the full stderr/traceback blob lives in
# the tmp/err/ snapshot): exited == hit a deadline,
# failed == an error whose source the launch recorded --
# the agent itself (with its exit code and captured cause)
# vs the downstream stream consumer (default "agent error"
# if unmarked); stopped == skipped because the run is
# out of budget
step_reason = None
if step_status == 'exited':
step_reason = 'timed out'
elif step_status == 'stopped':
if step_budget_skip:
step_reason = 'over budget'
elif step_status == 'paused':
# the awaiting-approval marker drives resume's
# re-entry: an approved step is skipped
# past, not re-run
if approval_interrupted:
step_reason = 'awaiting approval'
elif step_status == 'failed':
step_reason = result.reason or 'agent error'
# a retry attempt's row is marked so the row-per-attempt
# shape reads in `node activity`; the failed first
# attempt keeps its plain reason
if attempt > 0:
if step_reason:
step_reason = f'{step_reason}; retry'
else:
step_reason = 'retry'
try:
node.record.step_end(
step_id=self._step_id,
status=step_status,
exit_code=step_exit_code,
metadata=step_reason,
)
# a step that never reached its launch spent nothing:
# stamp the knowable zero so NULL cost stays reserved
# for unknowable spend
if not self._step_launched:
node.record.step_cost(step_id=self._step_id, cost=0.0)
except Exception:
pass
# warn (do not enforce) when a step's recorded
# cost exceeds --max-step-cost
if self._max_step_cost is not None and self._step_id is not None:
step_spent = self._cost_spent(step_id=self._step_id)
max_step_cost = float(self._max_step_cost)
if step_spent is not None and step_spent >= max_step_cost:
print(
f'Warning: step {step_num}/{step_count} ({step_name})'
f' cost ${step_spent:.4f} exceeded --max-step-cost'
f' ${self._max_step_cost_label}',
file=sys.stderr,
)
# park after recording the paused step -- no commit; the dirty
# worktree is the frozen mid-step state resume continues from
if self._paused:
return True
# only a failed launch buys a retry, and only
# while attempts remain
if result.status != 'failed' or attempt >= self._step_retries:
break
print(
f'--- Step {step_num}/{step_count} ({step_name}):'
f' retrying in {self._step_retry_backoff} ---'
)
if not self._retry_backoff():
# a pause landing during the backoff parks; stop and the
# spent subtree ceiling abandon the retry to the
# failure path below
if self._paused:
return True
break
if failed:
# book the never-run tail as real rows (start + immediate
# stopped close) so `node activity` answers which steps never
# ran -- the stopped status carries the not-run semantics
remaining = steps[step_num:]
for missed in remaining:
try:
missed_id = node.record.step_start(
iter_id=self._iter_id,
run_id=self._run_id,
step=missed.number,
step_name=missed.name,
)
node.record.step_end(
step_id=missed_id,
status='stopped',
exit_code=0,
metadata=f'failed on {step_name}',
)
# never launched, so the spend is a knowable zero
node.record.step_cost(step_id=missed_id, cost=0.0)
except Exception:
pass
if started:
print('--- Committing directly (step failed/timed out) ---')
# the backstop commit's body carries the failure reason
# and the never-run tail, so the save describes itself
if self._timed_out:
reason = self._fail_reason or 'timed out'
else:
reason = result.reason or 'agent error'
if remaining:
names = ', '.join(missed.name for missed in remaining)
reason = f'{reason}\nsteps not run: {names}'
self._force_commit(f'failed on {step_name}', body=reason)
return False
return True
def _run_step(self: Loop, step: Step) -> StepResult:
"""Resolve one step's launch parameters, then launch it.
Per-step agent/model/detached overrides, the deadline, and the
budget resolve here; a non-positive budget skips the step. Fires
the on_step pairing around the launch (SYNC launches fire it
too, via ``_sync``).
"""
node = self.node
step_event = self.on_step(step=self._step_label)
# tracks whether this step reached its agent launch: a pre-launch return
# spent nothing, so its row's cost is a knowable zero -- NULL cost stays
# reserved for unknowable spend (a kill before the first stream flush)
self._step_launched = False
def close(result: StepResult) -> StepResult:
"""Fire the terminal pairing for this step's attribution."""
if result.status in ('failed', 'timed out'):
error = RuntimeError(result.reason or result.status)
self.on_step_failure(
initial_event=step_event,
result=result,
error=error,
)
else:
self.on_step_success(initial_event=step_event, result=result)
return result
# resolve this step's effective agent + provider + model (frontmatter
# overrides node defaults)
step_agent_command = step.agent or self._agent_command
# validate a per-step agent override (the node agent is
# validated at startup)
if step_agent_command != self._agent_command:
try:
agent = node.agent(step_agent_command, step.provider)
except ValueError as error:
msg = f'{error} (required by {step.path.name})'
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
base, *_ = step_agent_command.split()
if shutil.which(base) is None:
msg = f'{base} is not installed (required by {step.path.name})'
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
print(f' (using agent: {step_agent_command})')
elif step.provider is not None:
# a provider-only override rebinds the boot agent on the step's
# route (the boot binding pinned the node's own)
agent = node.agent(step_agent_command, step.provider)
else:
agent = self._agent
# a per-step route must be one the step's agent supports -- an
# invocation-time ValueError would crash the loop at the launch site
if agent.provider is not None and agent.provider not in agent.providers:
msg = (
f'unsupported provider {agent.provider!r} for'
f' {agent.name} (required by {step.path.name})'
)
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
step_model = step.model or self._node_model
step_effort = step.effort or self._node_effort
# validate detached: in a detached node any per-step detached: key is
# invalid (already detached); in a continuous node detached: true
# detaches this step -- detached: false restates the default (like
# requires_approval: false), a no-op; an empty value counts as absent
if self._detached and step.detached_raw:
msg = (
f'detached: in {step.path.name} is invalid'
f' in detached mode (already detached)'
)
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
# a failed build (builder error, missing charter/step file) must fail
# the step -- launching the agent on an empty prompt would burn
# a full turn
try:
prompt = self._build_step_prompt(step)
except Exception as error:
print(f'Error: {error}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(f'{error}'))
)
# the step's own ceiling: timeout: frontmatter substitutes for the
# node-global step_timeout; step files are live-edited steering
# surfaces, so a malformed scalar warns and falls back to the global
# instead of crashing the loop
self._step_timeout_effective = self._step_timeout_seconds
if step.timeout is not None:
try:
self._step_timeout_effective = _parse_duration(step.timeout, 'timeout')
except ValueError as error:
print(
f'Warning: {error} (timeout: in {step.path.name});'
f' falling back to the node step_timeout',
file=sys.stderr,
)
# compute step time limit: min(run remaining, iteration
# remaining, step timeout)
self._step_limit_seconds = 0
deadline = self._soonest_deadline()
if deadline > 0:
remaining = deadline - int(time.time())
if remaining <= 0:
return close(StepResult(status='timed out', exit_code=124))
self._step_limit_seconds = remaining
if self._step_timeout_effective > 0:
no_limit = self._step_limit_seconds == 0
tighter = self._step_timeout_effective < self._step_limit_seconds
if no_limit or tighter:
self._step_limit_seconds = self._step_timeout_effective
# a cost cap requires spend the loop can track: token-priced and
# routed agents price from token counts, so a cap needs a priceable
# model (cost-reporting natives always track)
caps = (self._max_cost, self._max_iter_cost, self._max_step_cost)
any_cap = any(cap is not None for cap in caps)
if not agent.tracks_cost(step_model or None) and any_cap:
if not step_model:
msg = (
f'a cost cap requires a model for {agent.name} in'
f' {step.path.name}; set --model or model: to a priced'
f' model'
)
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
msg = (
f'cost cap set but model {step_model!r} has no'
f' pricing entry; set a priced model or remove the cost'
f' cap'
)
print(f'Error: {msg}', file=sys.stderr)
return close(
StepResult(status='failed', exit_code=1, reason=_one_line(msg))
)
# this turn runs detached if the node is detached or
# the step requests it
self._step_detached = self._detached or step.detached_raw == 'true'
# the configured budget is exhausted (a computed but non-positive
# leash -- the run is at its ceiling) -- skip the launch so a
# wind-down step never runs uncapped; the skip lets the iteration
# record a clean budget stop (with the reason for `node activity`)
# and the force-commit backstop save prior work (mirrors the deadline
# early-return above); 125 is the budget-skip sentinel the SYNC
# failure banner prints
budget = self._step_budget()
if budget is not None and budget <= 0:
self._step_budget_value = None
return close(StepResult(status='skipped', exit_code=125))
self._step_budget_value = budget
self._step_model = step_model
self._step_effort = step_effort
self._step_launched = True
return close(self._launch(step, prompt, agent=agent, budget=budget))
def _build_step_prompt(self: Loop, step: Step) -> str:
"""Assemble and render a step's full prompt.
The charter, the step (frontmatter stripped), and any active
modes (SYNC.md runs as its own step, so it is skipped there):
static vars (paths, limits) come from the node's config/git, and
the run-scoped vars pass through as overrides -- the boot-pinned
DETACHED/META flags included, so a mid-run config edit cannot
flip a run's mode composition.
"""
# refresh the cost budget so each step's context reflects spend so far
self._build_cost_budget()
overrides = {
'STEP_LABEL': self._step_label,
'ITER_LABEL': self._iter_label,
'ITER_TIMESTAMP': self._iter_timestamp,
'ITER_REF': self._iter_ref,
'TIME_BUDGET': self._time_budget,
'COST_BUDGET': self._cost_budget,
'CONTINUE_MODE': 'true' if self._continue else 'false',
'RESUME_MODE': 'true' if self._resume_mode else 'false',
# reserve mode is per-iteration (budget can exhaust mid-run)
'RESERVE_MODE': 'true' if self._reserve else 'false',
'DETACHED_MODE': self._env_base['DETACHED_MODE'],
'META_MODE': self._env_base['META_MODE'],
}
return self.node.build_prompt(f'{step.path}', overrides=overrides)
def _build_cost_budget(self: Loop) -> None:
"""Refresh the cost-budget label from the CURRENT remaining.
Cost is recorded at each step's end, so this must be recomputed
per step -- a once-per-iteration value shows the iteration-start
budget on every step (stale, and it appears to never decrement
within an iteration, contradicting reserve mode which keys on
real spend). The remaining derives from the loop's own
``max_cost`` global (fresh spend against the same read set the
labels are stamped from), NOT the live remaining reader, whose
independent config read could mix new-cap figures into old-cap
labels mid-retune.
"""
if self._max_cost is None:
self._cost_budget = 'no limit'
return
spent = self._run_spent() or 0.0
remaining = max(0.0, float(self._max_cost) - spent)
cost_budget = f'${remaining:.4f} remaining of ${self._max_cost_label}'
if self._max_iter_cost is not None:
cost_budget += f' (max ${self._max_iter_cost_label}/iter)'
if self._max_step_cost is not None:
cost_budget += (
f' ({self._step_cost_verb} ${self._max_step_cost_label}/step)'
)
self._cost_budget = cost_budget
def _agent_env(self: Loop, step_label: str) -> dict[str, str]:
"""Build the per-launch agent environment.
Rebuilt at every step/SYNC launch, never boot-static: the boot
vocabulary (paths, caps, modes, ``AGENT_COMMAND``, timeouts --
``STEP_TIMEOUT_SECONDS`` carrying the step's own effective
ceiling, ``_NODE``, ``NODE_BRANCH``) plus live lineage
(``RUN_ID``, ``ITER_ID``, ``STEP_ID``, ``STEP_MODEL``), the
per-iteration quartet
(``ITER``, ``ITER_TIMESTAMP``, ``ITER_REF``, ``ITER_LABEL``),
and the current ``STEP_LABEL``; ``$ITER_REF`` feeds
``fractal plan init``'s envvar fallback. Passed to the seam as
``invocation(env=...)``'s caller overlay (merge order ruled at
the seam: ``os.environ``, then this overlay, then provider keys
-- provider wins).
"""
step_budget = ''
if self._step_budget_value is not None:
step_budget = f'{self._step_budget_value:.6g}'
env = dict(self._env_base)
overlay = {
'MAX_COST': self._max_cost_label,
'MAX_ITER_COST': self._max_iter_cost_label,
'MAX_STEP_COST': self._max_step_cost_label,
'AGENT_COMMAND': self._agent_command,
'DETACHED': 'true' if self._detached else 'false',
'RUN_TIMEOUT_SECONDS': f'{self._run_timeout_seconds}',
'ITER_TIMEOUT_SECONDS': f'{self._iter_timeout_seconds}',
'STEP_TIMEOUT_SECONDS': f'{self._step_timeout_effective}',
'INTERVAL_SECONDS': f'{self._interval_seconds}',
'ITER': f'{self._iter}',
'CONTINUE_MODE': 'true' if self._continue else 'false',
'RESUME_MODE': 'true' if self._resume_mode else 'false',
'_NODE': f'{self.node.node_dir}',
# external env consumers key node identity off this branch,
# not _NODE's basename (an incidental dir-layout fact)
'NODE_BRANCH': self.node.branch,
'RUN_ID': f'{self._run_id}',
'RUN_END_EPOCH': f'{self._run_end_epoch}',
'ITER_END_EPOCH': f'{self._iter_end_epoch}',
'ITER_TIMESTAMP': self._iter_timestamp,
'ITER_REF': self._iter_ref,
'ITER_LABEL': self._iter_label,
'TIME_BUDGET': self._time_budget,
'COST_BUDGET': self._cost_budget,
'STEP_LABEL': step_label,
'STEP_MODEL': self._step_model,
'STEP_LIMIT_SECONDS': f'{self._step_limit_seconds}',
'STEP_DETACHED': 'true' if self._step_detached else 'false',
'STEP_BUDGET': step_budget,
}
env.update(overlay)
# always exported, blank when unset: the overlay must mask any
# inherited ITER_ID/STEP_ID (a nested launch keeps the caller's
# env), or the agent's CLI calls would attribute rows to
# a foreign step
env['ITER_ID'] = f'{self._iter_id}' if self._iter_id is not None else ''
env['STEP_ID'] = f'{self._step_id}' if self._step_id is not None else ''
return env
def _launch(
self: Loop,
step: Step,
prompt: str,
*,
agent: Agent,
budget: Optional[float],
) -> StepResult:
"""Spawn the seam-built invocation as a process-group leader.
Records ``.step_pgid``, redirects stderr to ``<agent>.err``
(copied to a durable ``tmp/err/`` snapshot on failure/timeout),
parses the stream in-process, enforces the deadline
(TERM-grace-KILL on the group), and attributes the outcome
(completed / timed out / paused / failed / budget stop).
"""
node = self.node
node_dir = node.node_dir
# clear any stale step group handle and abort marker; the spawn below
# re-records the group, pause.sh the marker
(node_dir / STEP_PGID_FILE).unlink(missing_ok=True)
(node_dir / PAUSE_ABORT_FILE).unlink(missing_ok=True)
# a stale launch reason must never label this invocation's outcome
self._fail_reason = None
# resolve this agent's session (empty unless a
# continuous session exists)
session = None
if not self._step_detached:
session = node.sessions.get(agent.name)
# the model recorded with the woven session: the explicit step model,
# else the agent's own configured default -- the launch passes only
# the explicit one (the agent already applies its own config itself)
record_model = self._step_model or agent.config_model()
# only an enforcing agent takes the hard per-step budget flag; the
# cap stays soft (warn-only) for the rest
budget_arg = budget if agent.enforces_budget else None
model = self._step_model or None
effort = self._step_effort or None
invocation = agent.invocation(
prompt,
session=session,
model=model,
effort=effort,
budget=budget_arg,
env=self._agent_env(self._step_label),
)
# stamp the launch's session before spawning, fresh mints and resumes
# alike -- the stream re-stamps it from the first frame, but a killed
# or timed-out launch (or an agent whose stream names the session only
# on its terminal frame) must still leave a resumable,
# attributable step row
if invocation.session is not None and self._step_id is not None:
try:
agent.record_session(
step_id=self._step_id,
session=invocation.session,
model=record_model,
detached=self._step_detached,
)
except Exception:
pass
# run in the worktree (the project) so a bare relative write lands in
# the deliverable tree, never inside .fractal/; capture stderr
# straight to a log so auth/startup failures are not hidden -- a
# direct redirect (never an un-awaited tee) can't be truncated when a
# fast-failing launch exits before a tee flushes
with open(agent.err_path, 'wb') as err_file:
# a launch that cannot exec (binary gone mid-run, fork failure)
# is a failed step, never a loop death -- 127 is the cannot-exec
# convention, and the failed close lets the force-commit
# backstop save prior work
try:
process = agent.spawn(
invocation,
start_new_session=True,
stderr=err_file,
)
except OSError as error:
print(f'Error: {error}', file=sys.stderr)
return StepResult(
status='failed',
exit_code=127,
reason=_one_line(f'agent launch failed: {error}'),
)
# record the launch as the leader of its own process group -- the
# handle pause.sh/kill.sh use to abort the agent subtree without
# touching this loop
try:
(node_dir / STEP_PGID_FILE).write_text(f'{process.pid}\n', encoding='utf-8')
except OSError:
pass
# enforce the deadline in-process: TERM the whole invocation group at
# the limit, then KILL any survivor after a short grace
timed_out = threading.Event()
def _deadline() -> None:
# a launch that finished in the hair between wait() returning
# and cancel() keeps its real outcome -- a fired timer over an
# already-gone invocation group must mark nothing; a live group
# is a real timeout even after the leader is reaped (a
# TERM-trapping survivor can outlive it holding the stdout
# pipe), so probe the group, not the leader (poll() first
# clears the leader's zombie so the probe reads true)
process.poll()
try:
os.killpg(process.pid, 0)
except ProcessLookupError:
return
except PermissionError:
pass
timed_out.set()
try:
os.killpg(process.pid, signal.SIGTERM)
except (ProcessLookupError, PermissionError):
return
deadline = time.monotonic() + _KILL_GRACE_SECONDS
while time.monotonic() < deadline:
# reap the leader (clears its zombie so the group probe below
# reads true), then check the WHOLE group: a survivor that traps
# TERM -- e.g. a grandchild holding the stdout pipe -- must still
# draw the KILL, which a leader-only poll() would skip, leaving
# the reader blocked on that pipe for the rest of the run
process.poll()
try:
os.killpg(process.pid, 0)
except ProcessLookupError:
return
except PermissionError:
pass
time.sleep(0.1)
try:
os.killpg(process.pid, signal.SIGKILL)
except (ProcessLookupError, PermissionError):
pass
timer: Optional[threading.Timer] = None
if self._step_limit_seconds > 0:
timer = threading.Timer(self._step_limit_seconds, _deadline)
# daemon: a BaseException mid-stream skips the cancel below, and
# a non-daemon survivor would hold the interpreter open to fire a
# late kill on a possibly-recycled pgid (the cancel must NOT move
# into the finally -- it would drop the deadline bounding wait())
timer.daemon = True
timer.start()
# drive the stream in-process: recording (session, per-frame cost
# flush) and the pane rendering ride the seam's driver
stream_error: Optional[Exception] = None
result = None
try:
result = agent.stream(
process.stdout,
step_id=self._step_id,
model=record_model,
detached=self._step_detached,
render=self._render,
)
except Exception as error:
stream_error = error
# the reader died with the agent possibly still writing -- drop
# the group so a blocked writer cannot outlive its step
if process.poll() is None:
try:
os.killpg(process.pid, signal.SIGKILL)
except (ProcessLookupError, PermissionError):
pass
finally:
# a truncated stream (no result frame) ends mid-line with no summary
close = getattr(self._render, 'close', None)
if close is not None:
close()
agent_status = process.wait()
if timer is not None:
timer.cancel()
# the invocation is over -- drop the group handle so a later
# pause/kill can never signal a recycled pgid
(node_dir / STEP_PGID_FILE).unlink(missing_ok=True)
session_out = result.session if result is not None else None
cost_out = result.cost if result is not None else None
# a failed/timed-out launch keeps its stderr durably: <agent>.err is
# the per-invocation live capture the next launch truncates, so the
# diagnosis is copied to a per-launch snapshot under git-ignored tmp/;
# a stream-consumer failure appends its traceback -- the .err capture
# alone is blank for a parser-side death
def _snapshot_err(trace: str = '') -> None:
snapshot = (
node_dir
/ 'tmp'
/ 'err'
/ f'{self._run_id}-{self._iter}-{step.name}.err'
)
# one snapshot per failing launch: a retry attempt or a later
# same-iteration SYNC serializes beside the first instead of
# overwriting its diagnosis
serial = 1
while snapshot.exists():
serial += 1
snapshot = snapshot.with_name(
f'{self._run_id}-{self._iter}-{step.name}-{serial}.err'
)
try:
snapshot.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(agent.err_path, snapshot)
if trace:
with snapshot.open('a', encoding='utf-8') as handle:
handle.write(trace)
except OSError:
pass
# attribute the outcome
if agent_status != 0 and result is not None and result.budget_stopped:
# the agent stopped because it reached its per-step budget cap
# (claude exits non-zero + emits result subtype
# error_max_budget_usd) -- a clean budget stop, not a failure; the
# loop's boundary cost checks decide whether to wind the run down
print(
'reached per-step budget (--max-budget-usd); stopping this'
' step cleanly',
file=sys.stderr,
)
return StepResult(
status='completed',
session=session_out,
cost=cost_out,
budget_stopped=True,
)
if agent_status != 0:
# the agent itself failed -- surface the captured stderr so the
# diagnosis is not hidden in the log file
try:
# lenient decode: agent stderr is external output, and a
# stray byte must degrade, not crash the run
captured = agent.err_path.read_text(encoding='utf-8', errors='replace')
except OSError:
captured = ''
if captured:
sys.stderr.write(captured)
# the abort was pause's kill, not an agent failure; the durable
# abort marker backs the signal, which a racing resume may
# already have withdrawn -- checked ahead of the deadline so a
# pause racing the timeout parks the worktree uncommitted
# instead of booking a timed-out failure that force-commits it
if self._consume_pause_abort():
return StepResult(status='paused', session=session_out, cost=cost_out)
if timed_out.is_set():
_snapshot_err()
return StepResult(
status='timed out',
exit_code=124,
session=session_out,
cost=cost_out,
)
# a stream-consumer failure the handler above SIGKILLed the agent
# for surfaces here as a signal death -- defer to the stream_error
# branch below so it owns the attribution (right side + traceback),
# rather than bury the real fault under a generic 'agent error'; safe
# because stream_error is set only on a fractal-side stream failure:
# a genuine agent crash closes stdout, so the reader hits EOF and
# returns without raising (stream_error stays None -> here)
if stream_error is None:
# a signal death surfaces as a negative returncode -- report it
# in the 128+N convention the exit banner prints
exit_code = agent_status if agent_status > 0 else 128 - agent_status
_snapshot_err()
# the stderr tail rides the durable reason -- the pane is mortal,
# and agents put the cause on the last line
reason = f'agent error (exit {exit_code})'
tail = _log_tail(agent.err_path)
if tail:
reason = f'{reason}: {tail}'
return StepResult(
status='failed',
exit_code=exit_code,
reason=reason,
session=session_out,
cost=cost_out,
)
if timed_out.is_set():
# the deadline fired but the agent shut down cleanly within the
# TERM grace -- the limit was still reached, so the step is timed
# out, never completed
_snapshot_err()
return StepResult(
status='timed out',
exit_code=124,
session=session_out,
cost=cost_out,
)
if stream_error is not None:
# an agent that named a failure on its own stream (drained clean,
# exit 0) is an agent error; a fractal-side stream-consumer failure
# is a "stream error"; both surface the detail on the pane and .err,
# but node activity must attribute the failure to the right side
print(f'Error: {stream_error}', file=sys.stderr)
if self._consume_pause_abort():
return StepResult(status='paused', session=session_out, cost=cost_out)
agent_borne = isinstance(stream_error, AgentStreamError)
self._fail_reason = 'agent error' if agent_borne else 'stream error'
_snapshot_err(trace=''.join(traceback.format_exception(stream_error)))
# the step row carries the exception one-liner (the iter/run
# rollup keeps the short label); an agent-borne message is
# self-describing, a fractal-side one leads with its type
if agent_borne:
message = f'{stream_error}'
else:
message = f'{type(stream_error).__name__}: {stream_error}'
detail = _one_line(message)
return StepResult(
status='failed',
exit_code=1,
reason=f'{self._fail_reason}: {detail}',
session=session_out,
cost=cost_out,
)
# a budget stop reported on the stream keeps its attribution even
# when the agent exits 0
budget_stopped = result is not None and result.budget_stopped
return StepResult(
status='completed',
session=session_out,
cost=cost_out,
budget_stopped=budget_stopped,
)
def _consume_pause_abort(self: Loop) -> bool:
"""Return whether a failed launch was pause's kill (consuming the marker)."""
marker = self.node.node_dir / PAUSE_ABORT_FILE
if marker.exists() or self._signal('pause'):
marker.unlink(missing_ok=True)
return True
return False
def _sync(
self: Loop,
step_num: int,
label: str,
*,
close: Optional[str] = None,
strict: bool = False,
) -> tuple[StepResult, int, Optional[int]]:
"""Run the SYNC step with its own billed step row.
The single implementation of the SYNC bookkeeping every waiting
and before-step site shares.
Args:
step_num: The step number the sync belongs to (the step it
precedes; the drain-wait sync, which precedes nothing,
stays ``0``).
label: The banner/context label (e.g. ``before EXECUTE``).
close: Closing banner label when it differs from ``label``
(the approval-wait sync).
strict: Record the full status vocabulary (timeout ->
``exited``, failure -> ``failed``, budget skip ->
``stopped`` flagging ``'over budget'``) and
leave the closing banner to the caller -- the before-step
sync; the lenient waits record ``completed`` for anything
but a pause (a genuinely failed launch keeps ``completed``
but its row's metadata names the failure) and close
themselves.
Returns:
The launch result, its duration in whole seconds, and the
sync's own step row id (the fatal-timeout force-commit's
lineage -- the caller's live step id is restored on return).
"""
node = self.node
saved_label = self._step_label
saved_step_id = self._step_id
# an approval-wait sync runs between a work step's launch and its
# close: the sync's _run_step resets _step_launched, and a pre-launch
# sync return (budget skip) would leave it False -- the work step's
# close would then stamp 0.0 over its real flushed cost, so restore
# the caller's flag after the sync's own zero-stamp check
saved_launched = self._step_launched
print()
print(f'--- SYNC ({label}) ---')
self._step_label = f'SYNC ({label})'
try:
self._step_id = node.record.step_start(
iter_id=self._iter_id,
run_id=self._run_id,
step=step_num,
step_name='SYNC',
)
except Exception:
self._step_id = None
step_start = time.monotonic()
sync_step = Step.load(self._sync_file)
result = self._run_step(sync_step)
duration = int(time.monotonic() - step_start)
# a pause abort landing on a lenient (waiting) SYNC is a park, not a
# completion; the wait's next poll parks the loop
sync_reason = None
if strict:
if result.status == 'timed out':
sync_status, sync_exit_code = 'exited', 1
elif result.status == 'paused':
sync_status, sync_exit_code = 'paused', 0
elif result.status == 'skipped':
# over budget: the sync was skipped, not run -- book it like
# the work step it precedes (stopped, reason flagged), so the
# activity log reads one cleanly-labeled budget stop, never
# an unexplained failure beside it
sync_status, sync_exit_code = 'stopped', 0
sync_reason = 'over budget'
elif result.status == 'failed':
sync_status, sync_exit_code = 'failed', 1
else:
sync_status, sync_exit_code = 'completed', 0
else:
sync_status = 'paused' if result.status == 'paused' else 'completed'
sync_exit_code = 0
# a waiting sync is best-effort, so a failed launch must not read
# as a work failure -- but silently swallowing it would hide a
# rotting node, so the completed row's metadata names it; the
# absorbed launch reason must not relabel a later wait overrun
if result.status == 'failed':
sync_reason = f'sync failed (exit {result.exit_code})'
self._fail_reason = None
# a budget-skipped waiting sync books completed like any other,
# but the skip must leave a ledger trace (the strict path labels
# the identical skip 'over budget')
elif result.status == 'skipped':
sync_reason = 'sync skipped (over budget)'
if self._step_id is not None:
try:
node.record.step_end(
step_id=self._step_id,
status=sync_status,
exit_code=sync_exit_code,
metadata=sync_reason,
)
# a sync that never reached its launch spent nothing: stamp the
# knowable zero so NULL cost stays reserved for unknowable spend
if not self._step_launched:
node.record.step_cost(step_id=self._step_id, cost=0.0)
except Exception:
pass
if not strict:
print(f'--- SYNC ({close or label}): done ---')
sync_step_id = self._step_id
self._step_label = saved_label
self._step_id = saved_step_id
self._step_launched = saved_launched
return result, duration, sync_step_id
def _wait_for_children(self: Loop, context: str) -> bool:
"""Drain active+paused descendants, SYNCing while waiting.
Returns ``False`` if interrupted (pause/stop/timeout, with the
pause flag set on a pause), ``True`` otherwise -- the interrupts
are mandatory: a crashed-but-active child would otherwise hang
the wait forever.
"""
poll_interval = min(self._wait_seconds, 5)
if poll_interval < 1:
poll_interval = 1
if not self._descendants_active():
return True
print()
print(f'--- Finishing: waiting for child nodes to finish ({context}) ---')
while self._descendants_active():
waited = 0
drained = False
while waited < self._wait_seconds:
time.sleep(poll_interval)
waited += poll_interval
if self._check_pause():
self._paused = True
return False
if self._check_stop():
return False
deadline = self._soonest_deadline()
if deadline > 0 and time.time() >= deadline:
self._timed_out = True
print('--- Waiting for children: timed out ---')
return False
if not self._descendants_active():
drained = True
break
if drained:
break
# SYNC while waiting (gated on sync mode + a live iteration)
live_iter = self._iter_id is not None
if self._sync_mode and self._sync_file.is_file() and live_iter:
self._sync(0, 'waiting for children')
print('--- Finishing: all child nodes finished ---')
return True
def _wait_for_approval(self: Loop, step: Step) -> str:
"""Poll the approval gate until released or interrupted.
Pause, timeout, and stop/finish interruptions map to distinct
outcomes.
Returns:
``'approved'``, ``'paused'``, ``'timed out'``, or
``'stopped'`` (a stop/finish interruption).
"""
step_id = self._step_id
step_name = step.name
poll_interval = min(self._wait_seconds, 5)
if poll_interval < 1:
poll_interval = 1
while not self._approved(step_id):
waited = 0
approved = False
while waited < self._wait_seconds:
time.sleep(poll_interval)
waited += poll_interval
if self._check_pause():
self._paused = True
return 'paused'
if self._check_stop() or self._check_finish():
return 'stopped'
deadline = self._soonest_deadline()
if deadline > 0 and time.time() >= deadline:
self._timed_out = True
return 'timed out'
if self._approved(step_id):
approved = True
break
if approved:
break
# SYNC while waiting for approval -- the sync gets its own row
# (charged to the awaited step's number), and the awaited step's
# id is restored after so the wait keeps polling and ending the
# awaited step, not the SYNC
if self._sync_mode and self._sync_file.is_file():
self._sync(
step_num=step.number,
label=f'approval wait for {step_name}',
close='approval wait',
)
return 'approved'
def _approved(self: Loop, step_id: Optional[int]) -> bool:
"""Return whether a step's approval gate has been released."""
if step_id is None:
return True
try:
return self.node.record.step_approved(step_id=step_id)
except Exception:
return False
def _retry_backoff(self: Loop) -> bool:
"""Sleep out the step-retry backoff, polling the abort signals.
One-second increments so a pause (parks -- the flag is set for
the caller), a stop, or a spent subtree ceiling lands within
seconds -- a failed step must never buy another attempt once
the cap is spent.
Returns:
Whether the retry may proceed (``False`` abandons it).
"""
waited = 0
while waited < self._step_retry_backoff_seconds:
time.sleep(1)
waited += 1
if self._check_pause():
self._paused = True
return False
if self._check_stop():
return False
if self._check_subtree_ceiling():
return False
return True
def _run_setup(self: Loop) -> bool:
"""Run setup.sh from the worktree root, teeing to setup.log.
Setup runs unbounded and unpausable: the run/iter deadlines gate
step launches only, and pause's abort targets the recorded step
group, which does not exist here.
Returns:
Whether setup succeeded.
"""
node = self.node
setup_script = node.node_dir / 'scripts' / 'setup.sh'
# tee the output to the node dir so the last run's errors survive the
# tmux tty; pin the CWD to the worktree root so relative setup lines
# land beside the work, not in the ambient launch dir
process = subprocess.Popen(
['bash', f'{setup_script}'],
cwd=node.worktree,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
# setup output is for logging; a non-UTF-8 byte must not crash the
# run (the exit code, not the text, decides success)
errors='replace',
)
with open(node.node_dir / 'setup.log', 'w', encoding='utf-8') as log:
for line in process.stdout:
sys.stdout.write(line)
sys.stdout.flush()
log.write(line)
return process.wait() == 0
def _read_cost_caps(self: Loop) -> None:
"""Load the budget-cap keys.
Called at run start, at each iteration top, and by the two
boundary checks themselves, so a mid-run retune (config edit or
node update) reaches the very next probe instead of staying
pinned to the run-start values.
Raises:
ValueError: When an edited cap violates the launch
invariants (:meth:`Config.validate`), before any cap is
reassigned -- boot propagates the refusal; the mid-run
readers warn and keep the previous caps.
"""
config = self.node.config
max_cost = config.get('max_cost')
max_iter_cost = config.get('max_iter_cost')
max_step_cost = config.get('max_step_cost')
reserve = config.get('reserve_budget')
# re-check the launch invariants before assigning anything: the
# documented steering path edits config.json directly, bypassing the
# setters' validation, and an unchecked cap would crash the next
# float() or degenerate the boundary probes
config_values = {
'max_cost': max_cost,
'max_iter_cost': max_iter_cost,
'max_step_cost': max_step_cost,
'reserve_budget': reserve,
}
config.validate(config_values)
self._max_cost = max_cost
self._max_iter_cost = max_iter_cost
self._max_step_cost = max_step_cost
# a budget set out-of-band (config set max_cost) bypasses node init's
# 10% reserve default, leaving no cleanup buffer; mirror that default
# here so the RESERVE nudge fires before the ceiling however
# max_cost was set
if reserve is None and self._max_cost is not None:
reserve = DEFAULT_RESERVE_FRACTION * float(self._max_cost)
self._reserve_label = f'{reserve:.6g}'
else:
self._reserve_label = f'{reserve}' if reserve is not None else '0'
self._reserve_budget = float(reserve) if reserve is not None else 0.0
# the label strings mirror the config values verbatim (an int cap
# reads `5`, a float `5.0`)
self._max_cost_label = f'{self._max_cost}' if self._max_cost is not None else ''
if self._max_iter_cost is not None:
self._max_iter_cost_label = f'{self._max_iter_cost}'
else:
self._max_iter_cost_label = ''
if self._max_step_cost is not None:
self._max_step_cost_label = f'{self._max_step_cost}'
else:
self._max_step_cost_label = ''
def _reconcile_caps(self: Loop) -> None:
"""Heal config/registry cap drift, one loud warning per drift."""
try:
drifted = self.node.config.reconcile()
except Exception:
return
# scream per drifted key -- a silent heal would hide that enforcement
# and the registry ever disagreed
for key, values in drifted.items():
config_value, registry_value = values
print(
f'WARNING: config {key}={config_value} != registry'
f' {registry_value} -- registry updated to match config'
f" (retune with 'fractal node update')"
)
def _soonest_deadline(self: Loop) -> int:
"""Return the soonest active wall-clock deadline, 0 if none.
The single source for "how long may the agent run right now"
across the run and iteration walls.
"""
deadline = 0
if self._run_end_epoch > 0:
deadline = self._run_end_epoch
if self._iter_end_epoch > 0:
if deadline == 0 or self._iter_end_epoch < deadline:
deadline = self._iter_end_epoch
return deadline
def _run_spent(self: Loop) -> Optional[float]:
"""Return the run's subtree spend at the ledger's display precision.
Budgets are per-run (runs are isolated); ``None`` when the
spend is untracked. A failed read warns and returns the last
good reading -- never ``None``, which reads as untracked and
would disarm the budget probes for a mere contention window.
Cost is recorded, never estimated: an ended step with a NULL
cost (killed before its usage flush) counts as zero, not an
imputed per-step figure -- NULL is the ledger's honest signal
that no cost was recorded.
"""
try:
spent = self.node.cost.spent(run_id=self._run_id)
if spent == 0.0 and self.node.cost.untracked(run_id=self._run_id):
return None
except Exception:
self._warn_unreadable_spend()
return self._last_run_spent
self._last_run_spent = round(spent, 4)
return self._last_run_spent
def _run_remaining(self: Loop) -> float:
"""Return the run's remaining budget, clamped at 0 (display precision).
A failed read derives from the last good spend reading -- never
the full cap, which would size the per-step leash as if nothing
were spent while the ledger is merely contended. With no last good
reading either (a failed read before the run's first successful
probe -- the resume path meets prior spend on a cold in-memory
cache), the leash floors at 0 and the step skips to retry next
probe, rather than handing out the untouched cap over real spend.
"""
try:
remaining = self.node.cost.remaining(run_id=self._run_id)
except Exception:
self._warn_unreadable_spend()
if self._last_run_spent is None:
return 0.0
remaining = float(self._max_cost or 0) - self._last_run_spent
if remaining is None:
return 0.0
return round(max(0.0, remaining), 4)
def _step_budget(self: Loop) -> Optional[float]:
"""Return the per-step budget leash, ``None`` when uncapped.
min(run remaining - reserve, iter headroom, step cap) with the
reserve-window floor.
The per-step USD budget for agents that accept one -- a hard
per-step ceiling that bounds the in-step overshoot the boundary
cost checks can't catch (a non-enforcing agent's cap stays
soft). A non-positive result means the configured budget is
exhausted; the caller skips the launch.
"""
caps = (self._max_cost, self._max_iter_cost, self._max_step_cost)
no_caps = all(cap is None for cap in caps)
if no_caps:
return None
result: Optional[float] = None
if self._max_cost is not None:
run_remaining = self._run_remaining()
result = run_remaining - self._reserve_budget
# in the reserve window the leash would go non-positive and drop
# off entirely -- floor it at the full remaining instead, so
# wind-down steps spend the reserve but never past the ceiling
if result <= 0:
result = run_remaining
# the iteration's live headroom (its cap minus recorded spend), not
# the static cap -- a later step must not get the full per-iter budget
# again; skip a drained (non-positive) headroom rather than zero the
# leash: reserve mode and the run-level bound govern
# past iter exhaustion
if self._max_iter_cost is not None and self._iter_id is not None:
try:
iter_remaining = self.node.cost.remaining(iter_id=self._iter_id)
self._last_iter_headroom = iter_remaining
except Exception:
# a failed read holds the last good headroom, not the full
# cap the unset-cap fallback below grants
self._warn_unreadable_spend()
iter_remaining = self._last_iter_headroom
if iter_remaining is None:
iter_remaining = float(self._max_iter_cost)
iter_remaining = round(max(0.0, iter_remaining), 4)
if iter_remaining > 0 and (result is None or iter_remaining < result):
result = iter_remaining
if self._max_step_cost is not None:
step_cap = float(self._max_step_cost)
if result is None or step_cap < result:
result = step_cap
return result
def _check_subtree_ceiling(self: Loop) -> bool:
"""Mid-iteration hard ceiling; signals finish once.
Finish recursively when the run's subtree budget is fully spent,
returning whether it tripped (the caller stops queuing steps);
soft cap -- the parent mitigates over-spend; a spawn-heavy
iteration can blow the budget before the boundary, so this stops
queuing more steps within the iteration.
"""
# read the caps live: a cap granted (or retuned) mid-iteration must
# reach this very probe, not wait for the iteration top; a malformed
# edit warns and keeps the prior caps, never crashing the run
try:
self._read_cost_caps()
except ValueError as error:
print(
f'Warning: {error}; keeping the previous cost caps',
file=sys.stderr,
)
if self._max_cost is None:
return False
if self._signal('finish'):
return False
spent = self._run_spent()
if spent is None:
self._warn_untracked_spend()
return False
if spent >= float(self._max_cost):
self._send_budget_finish(
f'Subtree cost budget reached'
f' (${spent:.4f} of ${self._max_cost_label}), finishing',
reason=f'subtree cost budget reached'
f' (spent ${spent:.4f} >= ${self._max_cost_label} max)',
stem='subtree cost budget reached',
)
return True
return False
def _check_reserve_boundary(self: Loop) -> bool:
"""Iteration-boundary reserve stop; triggers the run's wind-down.
End the run once spend enters the total-cost reserve window, so
the just-finished wind-down iteration is the last (a new one
would only re-enter reserve); keyed on subtree spend (>=
``max_cost`` minus reserve), not the clamped remaining reader,
so it also catches an over-ceiling overshoot; subsumes the
boundary hard ceiling; soft, recursive.
"""
# read the caps live: a cap granted (or retuned) mid-iteration must
# reach this very probe, not wait for the iteration top; a malformed
# edit warns and keeps the prior caps, never crashing the run
try:
self._read_cost_caps()
except ValueError as error:
print(
f'Warning: {error}; keeping the previous cost caps',
file=sys.stderr,
)
if self._max_cost is None:
return False
if self._signal('finish'):
return False
spent = self._run_spent()
if spent is None:
self._warn_untracked_spend()
return False
if spent >= float(self._max_cost) - self._reserve_budget:
self._send_budget_finish(
f'Total cost budget reserve reached'
f' (${spent:.4f} of ${self._max_cost_label} spent), ending run',
reason=f'cost budget reserve reached (spent ${spent:.4f} >='
f' ${self._max_cost_label} max'
f' - ${self._reserve_label} reserve)',
stem='cost budget reserve reached',
)
return True
return False
def _send_budget_finish(
self: Loop,
notice: str,
*,
reason: str,
stem: str,
) -> None:
"""Send the recursive finish + record the budget abort.
Shared by the hard subtree ceiling (mid-iteration) and the
reserve-threshold stop (boundary); ``reason`` names which bound
tripped so ``node activity`` reads clearly, and ``stem`` is its
figure-free head for the exit notice, whose only figure is the
fresh final read.
"""
print(f'=== {notice} ===')
try:
self.node.finish(reason)
except Exception:
pass
# mark the abort so the terminal status records it as exited, not the
# goal-met `completed` a plain finish signal would otherwise produce
self._budget_hit = True
self._budget_reason = reason
self._budget_stem = stem
def _warn_soft_cap(self: Loop) -> None:
"""Scream once per run when armed caps have no in-step brake.
A non-enforcing agent takes no per-step budget flag, so its cap
is checked only between steps -- with no run/iter/step timeout
either, one runaway step can overshoot every cap unbounded while
the loop waits on the process. Advisory only; the remedy it
names (``--step-timeout``) is the existing in-step brake.
"""
if self._soft_cap_warned:
return
caps = (self._max_cost, self._max_iter_cost, self._max_step_cost)
if all(cap is None for cap in caps) or self._agent.enforces_budget:
return
timeouts = (
self._run_timeout_seconds,
self._iter_timeout_seconds,
self._step_timeout_seconds,
)
if any(seconds > 0 for seconds in timeouts):
return
self._soft_cap_warned = True
print(
f'WARNING: cost caps are set but {self._agent.name} does not'
' enforce a per-step budget and no timeout is configured; one'
' runaway step can overshoot the caps unbounded -- set'
' --step-timeout to bound it.'
)
def _warn_untracked_spend(self: Loop) -> None:
"""Scream once per run when armed caps read untracked spend.
Shared by both budget probes: an all-NULL run counts $0 in the
guards, so neither can ever trip -- advisory only, never a
block (the edge only arises from runs whose real un-metered
spend is near zero).
"""
if self._untracked_warned:
return
self._untracked_warned = True
print(
"WARNING: cost caps are set but this run's spend is untracked"
' (unpriced steps); budget guards cannot trip.'
)
def _warn_unreadable_spend(self: Loop) -> None:
"""Scream once per run when a cost-ledger read fails.
Shared by every budget reader: the failed read falls back to
the last good reading rather than the full cap, so the guards
stay armed (if stale) until the ledger reads again -- advisory
only; a contended read never re-inflates spend authority.
"""
if self._unreadable_warned:
return
self._unreadable_warned = True
print(
'WARNING: cost read failed; budget guards hold the last good'
' reading until the ledger reads again.'
)
def _send_exit_notice(self: Loop, status: str, reason: str) -> None:
"""Post a radio notice for an abnormal run end.
The death shows in a parent's feed; ``reason`` mirrors the
recorded reason's stem (the fresh spend read below is the
notice's only figure), and the send is guarded -- radio must
never break the exit path.
"""
# the run-qualified iteration label already names the run
notice = (
f'Run ended {status} at iteration {self._run_id}.{self._iter}: {reason}.'
)
# a cost-capped node reports the figures alongside the reason
if self._max_cost is not None:
spent = self._run_spent()
spent_label = 'untracked' if spent is None else f'${spent:.4f}'
notice = f'{notice} Spend {spent_label} of ${self._max_cost_label}.'
try:
self.node.radio.send(
channel='outbox',
subject=f'run {status}: {reason}',
data=notice,
priority=7,
)
except Exception:
pass
def _signal(self: Loop, name: str) -> bool:
"""Return whether a signal row is set on this run."""
try:
return self.node.record.signal_get(name, run_id=self._run_id) is not None
except Exception:
return False
def _print_signal_reason(self: Loop, name: str) -> None:
"""Echo a set signal's reason to the pane, ahead of its banner.
The transcript shows why, not just that, the loop is parking or
ending (the reason also survives on the signal row).
"""
try:
reason = self.node.record.signal_get(name, run_id=self._run_id)
except Exception:
return
if reason:
print(reason)
def _check_stop(self: Loop) -> bool:
"""Check the stop signal, printing its reason and banner when set."""
if self._signal('stop'):
self._print_signal_reason('stop')
print()
print('=== Stop requested ===')
print()
return True
return False
def _check_finish(self: Loop) -> bool:
"""Check the finish signal, printing its reason and banner when set."""
if self._signal('finish'):
self._print_signal_reason('finish')
print()
print('=== Finish requirements met ===')
print()
return True
return False
def _check_pause(self: Loop) -> bool:
"""Check the pause signal, printing its reason and banner when set."""
if self._signal('pause'):
self._print_signal_reason('pause')
print()
print('=== Pause requested ===')
print()
return True
return False
def _descendants_active(self: Loop) -> bool:
"""Return whether any descendant is live and active-or-paused.
Checked live (not the cached registry); a paused descendant
still counts -- it is frozen mid-work and drains only after
resume, so a finishing parent must wait for it, never complete
over it; one query for both statuses, so a child flipping
between them mid-poll is never missed by two separate snapshots.
A booting child counts too -- the live view reads an idle node
with a live tmux session as ``active``, so a child started by a
late step (its loop still in preflight, the active stamp pending)
is never read as drained.
"""
try:
rows = self.node.list(status='active,paused', live=True, decorated=False)
except Exception:
# an inconclusive probe (DB contention, a transient git
# failure) counts as active -- the drain must prove the
# subtree empty, never infer it from a failed read; the
# wait's stop/timeout interrupts bound a persistent fault
return True
return len(rows) > 0
def _finalize(self: Loop) -> int:
"""Run the ordered terminal cascade and return the exit code.
Drain backstop, pause park (rows left open), final commit sweep,
over-cap sweep, cascaded- and parked-budget sweeps, status
matrix, run end, abnormal-exit radio notice.
"""
node = self.node
# drain backstop: a resume can re-enter at the loop's max-iters or
# timeout gate without reaching an in-loop drain -- a pending finish
# must still wait for the subtree before the cascade below stamps
# completed; idempotent for normal exits (a drained finish returns
# immediately), and a pause landing here parks like any other; a
# stop or timeout that interrupts the drain abandons the finish:
# the cascade below must not claim a completed finish over a
# subtree it never drained
drained = True
if not self._paused and self._signal('finish'):
drained = self._wait_for_children('run end')
# park: the pause terminal -- stamp paused and leave the run and
# iteration rows open for resume to adopt; no run end, no exit
# signal, and no commit sweep (the dirty worktree is the frozen
# mid-step state resume continues from); _on_exit reads the paused
# stamp and leaves it alone
if self._paused:
# (re)open the pause span at the park instant -- a duplicate
# pause event collapses in the credit walk, and a park after a
# withdrawn pause (whose resume event closed the span) would
# otherwise burn its parked wall-clock against
# the adopted deadlines
try:
event_id = node.record.event_start(
'pause',
metadata='parked',
run_id=self._run_id,
)
if event_id is not None:
node.record.event_end(event_id=event_id, status='completed')
except Exception:
pass
try:
node.status_set('paused')
except Exception:
pass
print()
print('=== Paused (resume with: fractal node resume) ===')
return 0
# final safety sweep: a trailing step (e.g. a wind-down after a
# finish/ceiling signal) can write to the worktree after the last
# per-iteration commit -- commit it so the node never exits unclean,
# which a later --continue (`git clean -fd`) would otherwise discard;
# mirrors the in-loop backstop: check (porcelain) then --force;
# `kill` bypasses this by design (abrupt stop)
if not self._commit_check():
self._force_commit('final')
# over-cap sweep: the in-loop budget checks disarm once a finish
# signal is set (they exist to send one), so a finish that crossed
# the cap reaches here with the budget flag clear -- classify it by
# its reason: a deliberate (non-budget-stemmed) finish keeps its
# goal-met completed landing with the overshoot recorded on the run
# row, while a budget-stemmed finish falls through to the sweeps
# below for reclassification with its own figured attribution
# fetch the run's finish rows once (newest first), shared by every
# sweep below: finish rows can land in any order (a cascade arriving
# during a goal-met drain), so classification prefers a deliberate
# row wherever one exists instead of trusting arrival order -- the
# same inputs must book the same terminal status either way round;
# an unreadable store reads as unreadable, never as deliberate
finish_deliberate: Optional[str] = None
finish_budget = ''
finish_readable = True
if self._signal('finish'):
try:
finish_rows = node.record.signals(
run_id=self._run_id,
signal='finish',
)
except Exception:
finish_readable = False
else:
for finish_row in finish_rows:
reason = finish_row['metadata'] or ''
if _is_budget_reason(reason):
if not finish_budget:
finish_budget = reason
elif finish_deliberate is None:
finish_deliberate = reason
cap_unflagged = not self._budget_hit and self._max_cost is not None
if cap_unflagged and self._signal('finish'):
spent = self._run_spent()
if spent is not None and spent >= float(self._max_cost):
if not finish_readable:
# unreadable signal store: deterministic budget
# reclassification -- an over-cap finish must never
# fail open into a goal-met completed
print(
f'=== Cost budget exceeded in finish wind-down'
f' (${spent:.4f} of ${self._max_cost_label} spent) ==='
)
self._budget_hit = True
self._budget_reason = 'cost budget exceeded in finish wind-down'
self._budget_stem = self._budget_reason
elif finish_deliberate is not None:
print(
f'=== Cost budget exceeded in finish wind-down'
f' (${spent:.4f} of ${self._max_cost_label} spent);'
f' deliberate finish -- completing ==='
)
self._cap_overshoot = (
f'cost budget exceeded in finish wind-down'
f' (spent ${spent:.4f} >= ${self._max_cost_label} max)'
)
# cascaded-budget sweep: an ancestor's budget abort propagates finish
# to every active descendant, but the budget flag stays local to the
# tripping loop -- without this the killed descendant closes as a
# goal-met completed; the propagated reason carries the budget prefix
# plus the `(via finish of <branch>)` attribution, so reclassify
# exactly those, relabeling with this run's own child-scope figures
# -- the ancestor's figures name its scope, not this run's; the
# signal row keeps the propagated reason as the honest record of
# what was sent. A deliberate row outranks every budget row: a node
# that met its goal keeps completed whichever order the rows landed
if not self._budget_hit and finish_deliberate is None and finish_budget:
if _is_cascaded_budget_reason(finish_budget):
print(
f'=== Budget abort cascaded from an ancestor ({finish_budget}) ==='
)
self._budget_hit = True
spent = self._run_spent()
figures = 'untracked' if spent is None else f'${spent:.4f}'
if self._max_cost is not None:
figures += f' of ${self._max_cost_label}'
self._budget_reason = (
f'ancestor budget abort: {finish_budget}; this run spent {figures}'
)
attribution = finish_budget[finish_budget.index('(via finish of ') :]
self._budget_stem = f'ancestor budget abort {attribution}'
else:
# parked-abort sweep: a pause can park the run between a
# self-sent budget finish and its landing, and the abort
# flags die with the parked loop process -- the resumed
# probes stay disarmed on the pending finish and a reserve
# stop sits below the over-cap threshold, so re-adopt the
# abort from the signal row (its persisted record) to keep
# the exited/0 landing
print(f'=== Budget abort resumed from park ({finish_budget}) ===')
self._budget_hit = True
self._budget_reason = finish_budget
# the stem is the reason's figure-free head
self._budget_stem = finish_budget.split(' (', 1)[0]
exit_reason = None
if self._budget_hit:
# a budget finish is a cost abort -- record the reason the
# tripping check set (reserve stop vs hard ceiling) so
# `node activity` explains the early stop (the node is
# marked exited below)
exit_reason = self._budget_reason
elif self._setup_abort:
# a setup crash-loop abort mirrors the budget abort: record the
# honest reason and the exit signal (matching the non-finish
# exits below) for `node activity`
exit_reason = f'setup failed x{self._setup_fails}'
try:
node.record.signal_set('exit', exit_reason)
except Exception:
pass
elif not (self._signal('finish') and drained):
# iteration labels are run-qualified ({run}.{iter}), composed
# directly -- _iter_ref is empty before the first iteration, so
# a pre-iteration end still reads 'iteration <run>.0'
if self._timed_out:
exit_reason = (
f'Timed out at iteration {self._run_id}.{self._iter}'
f' ({self._time_budget})'
)
elif self._signal('stop'):
exit_reason = 'Stopped by request'
elif self._max_iters > 0 and self._iter >= self._max_iters:
exit_reason = f'Reached max iterations ({self._max_iters})'
if self._last_iter_failed:
exit_reason += '; final iteration failed'
else:
exit_reason = f'Exited at iteration {self._run_id}.{self._iter}'
try:
node.record.signal_set('exit', exit_reason)
except Exception:
pass
elif self._cap_overshoot:
# the goal-met finish keeps its completed landing; the overshoot
# figures ride the run row so `node activity` explains the spend
exit_reason = self._cap_overshoot
if self._budget_hit:
# budget landing: a budget stop is not a goal-met completion --
# mark it exited so a parent (and `node merge`) can tell
# unfinished work from done, even though the finish signal it set
# is checked below -- but it is a designed stop, not an abnormal
# death, so it lands exit 0: exited/0 is the DB-level
# budget-landing discriminator (every other exited path keeps 1)
node_status, run_status, run_exit_code = 'exited', 'exited', 0
elif self._setup_abort:
# setup crash-loop abort: never a goal-met completion, and never
# shadowed by the max-iters clause below (which would read a
# crash-loop as a full run)
node_status, run_status, run_exit_code = 'exited', 'exited', 1
elif self._signal('finish') and drained:
node_status, run_status, run_exit_code = 'completed', 'completed', 0
elif self._timed_out:
# a timeout is abnormal even when it lands on the final iteration
# -- it must never be shadowed by the max-iters clause below
# (mirrors the reason block)
node_status, run_status, run_exit_code = 'exited', 'exited', 1
elif self._signal('stop'):
# a stop on the final iteration must not be shadowed
# by the max-iters clause
node_status, run_status, run_exit_code = 'stopped', 'stopped', 0
elif self._max_iters > 0 and self._iter >= self._max_iters:
if self._last_iter_failed:
# a failed final iteration is abnormal even at the max-iters
# boundary (mirrors the timeout leg) -- completed would
# launder a dead iteration into a clean run
node_status, run_status, run_exit_code = 'exited', 'exited', 1
else:
node_status, run_status, run_exit_code = 'completed', 'completed', 0
else:
# unexpected exit -- abnormal, exit 1
node_status, run_status, run_exit_code = 'exited', 'exited', 1
# guarded like every status write: status_set stamps .status before
# the registry row, so a transient DB failure here would leave a
# terminal .status that _on_exit refuses to relabel -- the run_end
# below must still close the row
try:
node.status_set(node_status)
except Exception:
pass
# carry the run's exit reason (set above for a non-finish exit or
# a cap-overshot finish) so `node activity` shows why the run ended;
# a clean finish leaves the reason unset -> no metadata
try:
node.record.run_end(
run_id=self._run_id,
status=run_status,
exit_code=run_exit_code,
metadata=exit_reason,
)
except Exception:
pass
# surface an abnormal end on radio; clean finishes, max-iters
# completions, and requested stops stay quiet -- a budget landing's
# notice carries the figure-free stem, so the fresh final read inside
# the notice is its only figure (the run row keeps the full reason)
if run_status == 'exited':
notice_reason = self._budget_stem if self._budget_hit else exit_reason or ''
self._send_exit_notice(run_status, notice_reason)
return 0
def _on_exit(self: Loop) -> None:
"""Settle the loop's exit (the EXIT-trap mirror).
Stamps ``exited`` and closes the still-open run/iter/step rows
only when status is still ``active``; heals caps; drops ``.pgid``
and ``.socket``; never relabels a recorded terminal.
"""
node = self.node
# drop the pgid and socket handles -- the loop is ending in-band,
# nothing to reap and no session left to probe
try:
(node.node_dir / PGID_FILE).unlink(missing_ok=True)
(node.node_dir / SOCKET_FILE).unlink(missing_ok=True)
except OSError:
pass
# heal config/registry cap drift before the row goes dark -- a
# pre-boundary death never reaches the next boundary reconcile and
# would strand it forever
self._reconcile_caps()
try:
still_active = node.status() == 'active'
except Exception:
still_active = False
if still_active:
try:
node.status_set('exited')
except Exception:
pass
# close the whole open lifecycle (the shared cascade), not just
# the run row: an in-band crash strands the iteration and step
# rows open too, and nothing heals them later --
# _reconcile_status no-ops once .status reads exited -- so a
# row left open here would read active forever
try:
node.record.close_open(
'exited',
metadata='Loop exited abnormally',
)
except Exception:
pass
# the crash-path twin of the terminal cascade's notice, mirroring
# the reason this branch just recorded
self._send_exit_notice('exited', 'Loop exited abnormally')
# consolidate the WAL at this fleet-quiet boundary: close-time
# checkpoints are disabled (db.py keeps the sidecars for write-denied
# readers), so without this the log only ever grows and the bare .db
# is incomplete at rest -- best-effort, a live sibling fleet defers
# it to the next exiting loop
try:
node.db.checkpoint()
except Exception:
pass
def _commit_check(self: Loop) -> bool:
"""Return whether the worktree is clean (the porcelain backstop check)."""
try:
commit.commit(self.node, '', check=True)
except Exception:
return False
return True
def _force_commit(
self: Loop,
message: str,
*,
body: Optional[str] = None,
step_id: Optional[int] = None,
) -> None:
"""Force-commit the worktree with the live iteration + lineage.
Every backstop save passes the loop's own lineage explicitly --
the cascade's ``final`` sweep runs after the iteration row
closes, and an open-row lookup alone would regress its message
to ``iteration 0``. ``body`` carries a caller paragraph (the
step-failure site's reason and never-run tail) below the
subject; ``step_id`` overrides the live step id (the fatal SYNC
timeout commits against the SYNC's own row). Guarded: a failing
backstop must never abort the loop.
"""
if step_id is None:
step_id = self._step_id if self._iter_id is not None else None
try:
output = commit.commit(
node=self.node,
message=message,
force=True,
body=body,
iteration=self._iter,
run_id=self._run_id,
iter_id=self._iter_id,
step_id=step_id,
)
if output:
print(output)
except Exception as error:
print(f'Error: {error}', file=sys.stderr)
def _iter_label_text(self: Loop) -> str:
"""Return the iteration progress label (``N of M`` / ``N (no limit)``)."""
if self._max_iters > 0:
return f'{self._iter} of {self._max_iters}'
return f'{self._iter} (no limit)'
def _cost_spent(
self: Loop,
*,
iter_id: Optional[int] = None,
step_id: Optional[int] = None,
) -> Optional[float]:
"""Return a scoped spend reading at display precision; ``None`` on error."""
try:
spent = self.node.cost.spent(iter_id=iter_id, step_id=step_id)
except Exception:
return None
return round(spent, 4)
[docs]
def on_iteration(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.INFO,
**kwargs: Any,
) -> LoopIterationEvent:
"""Handle iteration event.
Fired when an iteration begins.
"""
event = LoopIterationEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
[docs]
def on_iteration_success(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.DEBUG,
**kwargs: Any,
) -> LoopIterationSuccessEvent:
"""Handle iteration success event.
Fired when an iteration ends in-band with an honest attribution.
"""
event = LoopIterationSuccessEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
[docs]
def on_iteration_failure(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.WARNING,
**kwargs: Any,
) -> LoopIterationFailureEvent:
"""Handle iteration failure event.
Fired when an iteration fails.
"""
event = LoopIterationFailureEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
[docs]
def on_step(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.INFO,
**kwargs: Any,
) -> LoopStepEvent:
"""Handle step event.
Fired when a step (or SYNC) launch begins.
"""
event = LoopStepEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
[docs]
def on_step_success(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.DEBUG,
**kwargs: Any,
) -> LoopStepSuccessEvent:
"""Handle step success event.
Fired when a step ends with a non-failure attribution.
"""
event = LoopStepSuccessEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
[docs]
def on_step_failure(
self: Loop,
message: Optional[str | list[str]] = None,
*,
logging_level: int = logging.WARNING,
**kwargs: Any,
) -> LoopStepFailureEvent:
"""Handle step failure event.
Fired when a step fails.
"""
event = LoopStepFailureEvent(
source=self,
message=message,
**kwargs,
)
self.log(event.description, logging_level)
return event
# ------ events
[docs]
class LoopIterationEvent(InitialEvent):
"""Emitted when an iteration begins."""
iteration: int
run_id: int
iter_id: int
[docs]
class LoopIterationSuccessEvent(TerminalEvent):
"""Emitted when an iteration ends in-band.
Any honest attribution: completed, paused, stopped.
"""
outcome: str
[docs]
class LoopIterationFailureEvent(FailureEvent):
"""Emitted when an iteration fails.
A setup crash-loop, a fatal SYNC timeout, or an unhandled loop
error.
"""
[docs]
class LoopStepEvent(InitialEvent):
"""Emitted when a step (or SYNC) launch begins."""
step: str
[docs]
class LoopStepSuccessEvent(TerminalEvent):
"""Emitted when a step ends with a non-failure attribution.
Completed, skipped, paused, or a budget stop.
"""
result: StepResult
[docs]
class LoopStepFailureEvent(FailureEvent):
"""Emitted when a step fails (agent error / stream error / timeout)."""
result: StepResult
# ------ helper functions
def _is_budget_reason(reason: str, /) -> bool:
"""Return whether a finish reason is a loop-authored budget abort.
Matches the loop's own reason shape -- a budget stem followed by its
figures -- so an agent's free-text reason that merely opens with the
vocabulary never classifies as a budget abort.
"""
return reason.startswith(_BUDGET_REASON_HEADS)
def _is_cascaded_budget_reason(reason: str, /) -> bool:
"""Return whether a finish reason is an ancestor's cascaded budget abort.
Fan-out always carries the tripping ancestor's raw reason, so the
match is the loop-authored budget shape plus the fan-out attribution.
"""
return '(via finish of ' in reason and _is_budget_reason(reason)
def _boot_env(node: Node, *, detached: bool, meta: str) -> dict[str, str]:
"""Derive the boot-static env/prompt vocabulary, pinned for the run.
The single var-map derivation (``render.variables``) with the mode
flags pinned to their boot values, so a mid-run config edit cannot
flip a run's mode composition or its exported environment.
"""
base = render.variables(node)
base['DETACHED_MODE'] = 'true' if detached else 'false'
base['META_MODE'] = 'true' if meta else 'false'
base['META_TARGET'] = meta
return base
def _parse_duration(value: str, label: str) -> int:
"""Parse a duration string into whole seconds, ``-1`` when unset.
Args:
value: Raw duration string, or ``''`` for unset.
label: Config key named in the error messages.
Returns:
Whole (truncated) seconds, or ``-1`` when ``value`` is empty.
Raises:
ValueError: When ``value`` is malformed or truncates to zero.
"""
if not value:
return -1
seconds = fractal.util.parse_duration_seconds(f'{value}')
if seconds is None:
raise ValueError(
f'{label} must be a duration with a unit suffix (e.g. 30s, 10m, 1.5h).'
)
# truncate to whole seconds -- deadlines are integral, and a sub-second
# duration (0.5s) must still error rather than round up
result = int(seconds)
if result <= 0:
raise ValueError(f'{label} must be greater than zero.')
return result
def _log_tail(path: pathlib.Path) -> str:
"""Return the last non-blank line of a captured log, capped at ``_REASON_CHARS``."""
try:
# lenient decode: captured logs are external output (agent stderr,
# setup.sh), and the tail feeds durable reasons -- degrade, never raise
lines = path.read_text(encoding='utf-8', errors='replace').splitlines()
except OSError:
return ''
for line in reversed(lines[-5:]):
if line.strip():
return line[:_REASON_CHARS]
return ''
def _one_line(text: str) -> str:
"""Collapse text to one whitespace-normalized line, capped at ``_REASON_CHARS``."""
return ' '.join(text.split())[:_REASON_CHARS]
