API reference

claudeloop.domain.errors

Domain-level error hierarchy. Pure — carries no I/O state.

AuthenticationFailedError

Bases: AutoclaudeError

Raised when the agent gateway reports a terminal authentication failure.

Never retryable — the run loop must abort rather than wait.

Source code in src/claudeloop/domain/errors.py

class AuthenticationFailedError(AutoclaudeError):
    """Raised when the agent gateway reports a terminal authentication failure.

    Never retryable — the run loop must abort rather than wait.
    """

AutoclaudeError

Bases: Exception

Base class for every error raised by claudeloop's own logic.

Source code in src/claudeloop/domain/errors.py

class AutoclaudeError(Exception):
    """Base class for every error raised by claudeloop's own logic."""

BudgetExceededError

Bases: AutoclaudeError

Raised when a run exceeds its configured turn, dollar, or wall-clock budget.

Source code in src/claudeloop/domain/errors.py

class BudgetExceededError(AutoclaudeError):
    """Raised when a run exceeds its configured turn, dollar, or wall-clock budget."""

InvalidPlanError

Bases: AutoclaudeError

Raised when a work plan file cannot be parsed into work items.

Source code in src/claudeloop/domain/errors.py

class InvalidPlanError(AutoclaudeError):
    """Raised when a work plan file cannot be parsed into work items."""

InvalidSessionSelectorError

Bases: AutoclaudeError

Raised when a session selector is malformed or ambiguous.

Source code in src/claudeloop/domain/errors.py

class InvalidSessionSelectorError(AutoclaudeError):
    """Raised when a session selector is malformed or ambiguous."""

claudeloop.domain.plan

Work plan value objects — parsing a handoff markdown file into discrete items.

PlanItem dataclass

One unit of work parsed from a plan file's checkbox list.

Source code in src/claudeloop/domain/plan.py

@dataclass(frozen=True, slots=True)
class PlanItem:
    """One unit of work parsed from a plan file's checkbox list."""

    text: str
    done: bool = False

    def __post_init__(self) -> None:
        if not self.text.strip():
            raise InvalidPlanError("Plan item text must not be blank")

WorkPlan dataclass

The full body of a handoff plan, plus any checkbox items found in it.

Source code in src/claudeloop/domain/plan.py

@dataclass(frozen=True, slots=True)
class WorkPlan:
    """The full body of a handoff plan, plus any checkbox items found in it."""

    raw_text: str
    items: tuple[PlanItem, ...] = field(default_factory=tuple)

    def __post_init__(self) -> None:
        if not self.raw_text.strip():
            raise InvalidPlanError("Plan text must not be blank")

    @property
    def has_items(self) -> bool:
        return len(self.items) > 0

    @property
    def remaining_items(self) -> tuple[PlanItem, ...]:
        return tuple(item for item in self.items if not item.done)

    @property
    def is_fully_done(self) -> bool:
        return self.has_items and len(self.remaining_items) == 0

    @staticmethod
    def parse(raw_text: str) -> WorkPlan:
        """Parse a markdown plan. Checkbox lines (`- [ ] ...` / `- [x] ...`) become
        tracked items; a plan with no checkboxes is still valid (bare instructions),
        just with an empty items tuple."""
        if not raw_text.strip():
            raise InvalidPlanError("Plan text must not be blank")

        items: list[PlanItem] = []
        for line in raw_text.splitlines():
            match = _CHECKBOX_RE.match(line)
            if match:
                done = match.group(1).lower() == "x"
                items.append(PlanItem(text=match.group(2), done=done))

        return WorkPlan(raw_text=raw_text, items=tuple(items))

    def with_items_marked_done(self, done_texts: frozenset[str]) -> WorkPlan:
        """Return a new WorkPlan with any item whose text is in `done_texts` marked done.
        Used to reconcile a structured-output verdict's `remaining_work` against the
        plan's own checklist between turns."""
        new_items = tuple(
            item if item.text not in done_texts else PlanItem(text=item.text, done=True)
            for item in self.items
        )
        return WorkPlan(raw_text=self.raw_text, items=new_items)

parse(raw_text) staticmethod

Parse a markdown plan. Checkbox lines (- [ ] ... / - [x] ...) become tracked items; a plan with no checkboxes is still valid (bare instructions), just with an empty items tuple.

Source code in src/claudeloop/domain/plan.py

@staticmethod
def parse(raw_text: str) -> WorkPlan:
    """Parse a markdown plan. Checkbox lines (`- [ ] ...` / `- [x] ...`) become
    tracked items; a plan with no checkboxes is still valid (bare instructions),
    just with an empty items tuple."""
    if not raw_text.strip():
        raise InvalidPlanError("Plan text must not be blank")

    items: list[PlanItem] = []
    for line in raw_text.splitlines():
        match = _CHECKBOX_RE.match(line)
        if match:
            done = match.group(1).lower() == "x"
            items.append(PlanItem(text=match.group(2), done=done))

    return WorkPlan(raw_text=raw_text, items=tuple(items))

with_items_marked_done(done_texts)

Return a new WorkPlan with any item whose text is in done_texts marked done. Used to reconcile a structured-output verdict's remaining_work against the plan's own checklist between turns.

Source code in src/claudeloop/domain/plan.py

def with_items_marked_done(self, done_texts: frozenset[str]) -> WorkPlan:
    """Return a new WorkPlan with any item whose text is in `done_texts` marked done.
    Used to reconcile a structured-output verdict's `remaining_work` against the
    plan's own checklist between turns."""
    new_items = tuple(
        item if item.text not in done_texts else PlanItem(text=item.text, done=True)
        for item in self.items
    )
    return WorkPlan(raw_text=self.raw_text, items=new_items)

claudeloop.domain.session

Session reference and selection value objects.

ExplicitSessionSelector dataclass

Resume a specific, caller-known session id.

Source code in src/claudeloop/domain/session.py

@dataclass(frozen=True, slots=True)
class ExplicitSessionSelector:
    """Resume a specific, caller-known session id."""

    session_id: str

    def __post_init__(self) -> None:
        if not self.session_id.strip():
            raise InvalidSessionSelectorError("session_id must not be blank")

MostRecentSessionSelector dataclass

Auto-select the most recently modified session for a working directory.

Source code in src/claudeloop/domain/session.py

@dataclass(frozen=True, slots=True)
class MostRecentSessionSelector:
    """Auto-select the most recently modified session for a working directory."""

    cwd: str

    def __post_init__(self) -> None:
        if not self.cwd.strip():
            raise InvalidSessionSelectorError("cwd must not be blank")

PlanFileSelector dataclass

Start a brand-new session seeded from the contents of a plan file.

Source code in src/claudeloop/domain/session.py

@dataclass(frozen=True, slots=True)
class PlanFileSelector:
    """Start a brand-new session seeded from the contents of a plan file."""

    plan_path: str

    def __post_init__(self) -> None:
        if not self.plan_path.strip():
            raise InvalidSessionSelectorError("plan_path must not be blank")

SessionRef dataclass

A resolved reference to a Claude Code session.

Source code in src/claudeloop/domain/session.py

@dataclass(frozen=True, slots=True)
class SessionRef:
    """A resolved reference to a Claude Code session."""

    session_id: str
    cwd: str
    last_modified: datetime | None = None
    git_branch: str | None = None
    first_prompt_preview: str | None = None

    def __post_init__(self) -> None:
        if not self.session_id.strip():
            raise InvalidSessionSelectorError("session_id must not be blank")
        if not self.cwd.strip():
            raise InvalidSessionSelectorError("cwd must not be blank")

claudeloop.domain.capacity

Capacity state — whether the account can currently spend a real turn, and why not if it can't. This is the typed replacement for regex-scraping stream-json for limit language.

AuthenticationFailed dataclass

Terminal — credentials are invalid or revoked. Never retryable.

Source code in src/claudeloop/domain/capacity.py

@dataclass(frozen=True, slots=True)
class AuthenticationFailed:
    """Terminal — credentials are invalid or revoked. Never retryable."""

    detail: str = ""

Available dataclass

Capacity exists; a real turn may be spent. utilization is informational — it reflects an allowed_warning signal and must never itself block a turn.

Source code in src/claudeloop/domain/capacity.py

@dataclass(frozen=True, slots=True)
class Available:
    """Capacity exists; a real turn may be spent. `utilization` is informational —
    it reflects an `allowed_warning` signal and must never itself block a turn."""

    utilization: float | None = None

CreditsExhausted dataclass

No token/time budget will fix this — the account is out of usage credits and requires a human to purchase more. There is no reset time by construction: waiting for a clock to advance can never resolve this state, only a probe that notices a top-up can.

Source code in src/claudeloop/domain/capacity.py

@dataclass(frozen=True, slots=True)
class CreditsExhausted:
    """No token/time budget will fix this — the account is out of usage credits and
    requires a human to purchase more. There is no reset time by construction: waiting
    for a clock to advance can never resolve this state, only a probe that notices a
    top-up can."""

    can_purchase: bool = True

WindowExhausted dataclass

A rate-limit window (five_hour / seven_day / seven_day_opus / seven_day_sonnet / overage) has been rejected. resets_at is the trusted reset instant when known; when None, the caller must fall back to a configured wait interval rather than assuming any particular reset time.

Source code in src/claudeloop/domain/capacity.py

@dataclass(frozen=True, slots=True)
class WindowExhausted:
    """A rate-limit window (five_hour / seven_day / seven_day_opus / seven_day_sonnet /
    overage) has been rejected. `resets_at` is the trusted reset instant when known;
    when None, the caller must fall back to a configured wait interval rather than
    assuming any particular reset time."""

    rate_limit_type: str
    resets_at: datetime | None = None

is_waitable(state)

Whether the run loop should ever schedule a wait/probe cycle for this state. AuthenticationFailed is the only capacity state that must abort outright.

Source code in src/claudeloop/domain/capacity.py

def is_waitable(state: CapacityState) -> bool:
    """Whether the run loop should ever schedule a wait/probe cycle for this state.
    AuthenticationFailed is the only capacity state that must abort outright."""
    return not isinstance(state, AuthenticationFailed)

claudeloop.domain.classify

Pure classification of raw turn signals into a CapacityState.

This is the direct replacement for extract_limit_signals() in the legacy script (legacy/claude_autoresume.py:290-333), except it operates on typed fields the Agent SDK already parsed, instead of regexing a raw JSON stream. rate_limit_status == "allowed_warning" is deliberately NOT checked as a rejection signal — it falls through the rejected computation below to Available, so it can never be mistaken for a hard limit. Once rejected, credit signals are checked before falling back to WindowExhausted, so a credits rejection can never be mistaken for a waitable window even if a stray resets_at rides along with it.

TurnSignals dataclass

Everything the classifier needs from one turn, gathered from the Agent SDK's RateLimitEvent, ResultMessage, and AssistantMessage — deliberately not a single source, because RateLimitEvent is reportedly dropped on some adapter paths.

Source code in src/claudeloop/domain/classify.py

@dataclass(frozen=True, slots=True)
class TurnSignals:
    """Everything the classifier needs from one turn, gathered from the Agent SDK's
    RateLimitEvent, ResultMessage, and AssistantMessage — deliberately not a single
    source, because RateLimitEvent is reportedly dropped on some adapter paths."""

    rate_limit_status: str | None = None  # "allowed" | "allowed_warning" | "rejected"
    rate_limit_type: str | None = None
    resets_at: datetime | None = None
    utilization: float | None = None
    overage_status: str | None = None
    overage_resets_at: datetime | None = None
    overage_disabled_reason: str | None = None
    api_error_status: int | None = None
    assistant_error: str | None = None
    error_code: str | None = None
    disabled_reason: str | None = None

claudeloop.domain.completion

Completion verdicts — was the whole task finished, or just this turn?

Primary source is the structured-output verdict the model returns per turn (ClaudeAgentOptions.output_format). A legacy substring marker is retained as a fallback for when structured output isn't available on a given model/config.

StructuredVerdict dataclass

Mirrors the JSON schema handed to the model via output_format: {"complete": bool, "remaining_work": [str], "blocked_on": str|null, "summary": str}

Source code in src/claudeloop/domain/completion.py

@dataclass(frozen=True, slots=True)
class StructuredVerdict:
    """Mirrors the JSON schema handed to the model via output_format:
    {"complete": bool, "remaining_work": [str], "blocked_on": str|null, "summary": str}
    """

    complete: bool
    remaining_work: tuple[str, ...] = ()
    blocked_on: str | None = None
    summary: str = ""

evaluate(*, structured, output_text, done_marker=DEFAULT_DONE_MARKER)

Decide what a single turn's outcome means for the overall task.

Precedence: a structured verdict is authoritative when present. Only when it is absent do we fall back to substring-matching the legacy marker in raw text.

Source code in src/claudeloop/domain/completion.py

def evaluate(
    *,
    structured: StructuredVerdict | None,
    output_text: str,
    done_marker: str = DEFAULT_DONE_MARKER,
) -> CompletionVerdict:
    """Decide what a single turn's outcome means for the overall task.

    Precedence: a structured verdict is authoritative when present. Only when it is
    absent do we fall back to substring-matching the legacy marker in raw text.
    """
    if structured is not None:
        if structured.blocked_on:
            return Blocked(reason=structured.blocked_on)
        if structured.complete:
            return Done(summary=structured.summary)
        return Continue(remaining_work=structured.remaining_work)

    if done_marker in output_text:
        return Done(summary="")
    return Continue(remaining_work=())

claudeloop.domain.waiting

Adaptive wait policy — decides the next probe instant, never a blind sleep.

This replaces the time.sleep(wait_seconds) calls in the legacy script (legacy/claude_autoresume.py:505,667) with a policy that can notice a mid-wait credit top-up or an overage lift instead of blocking until a fixed deadline. See docs/architecture/decisions/0004-adaptive-waiting-with-probes-not-sleep.md.

next_probe_instant(state, *, now, started_waiting_at, probe_count, config=DEFAULT_WAIT_POLICY_CONFIG)

Compute the next instant a probe should run. Never returns an instant in the past relative to now, and — when config.max_wait is set — never proposes an instant beyond started_waiting_at + config.max_wait (callers must treat that as "give up", not "wait longer").

Source code in src/claudeloop/domain/waiting.py

def next_probe_instant(
    state: CapacityState,
    *,
    now: datetime,
    started_waiting_at: datetime,
    probe_count: int,
    config: WaitPolicyConfig = DEFAULT_WAIT_POLICY_CONFIG,
) -> datetime:
    """Compute the next instant a probe should run. Never returns an instant in the
    past relative to `now`, and — when `config.max_wait` is set — never proposes an
    instant beyond `started_waiting_at + config.max_wait` (callers must treat that as
    "give up", not "wait longer")."""
    if isinstance(state, CreditsExhausted):
        # Compute the exponent in float seconds and clamp to the ceiling *before*
        # constructing a timedelta — an unclamped exponential can overflow
        # timedelta's max magnitude (~2.7e6 years) well within realistic probe counts.
        ceiling_seconds = config.credits_probe_ceiling.total_seconds()
        interval_seconds = config.credits_probe_interval.total_seconds()
        backoff_seconds = min(
            interval_seconds * (config.credits_backoff_factor**probe_count), ceiling_seconds
        )
        candidate = now + timedelta(seconds=backoff_seconds)
    elif isinstance(state, WindowExhausted) and state.resets_at is not None:
        by_reset = state.resets_at + config.reset_grace
        by_interval = now + config.window_probe_interval
        candidate = min(by_reset, by_interval)
    else:
        candidate = now + config.window_probe_interval

    if candidate < now:  # pragma: no cover — unreachable: all config intervals are
        candidate = now  # validated positive in __post_init__, so every branch above
        # already yields candidate >= now. Kept as a defensive invariant guard.

    if config.max_wait is not None:
        deadline = started_waiting_at + config.max_wait
        if candidate > deadline:
            candidate = deadline

    return candidate

wait_exceeded(*, started_waiting_at, now, config)

Whether the configured max_wait budget has been consumed and the run loop should give up rather than schedule another probe.

Source code in src/claudeloop/domain/waiting.py

def wait_exceeded(*, started_waiting_at: datetime, now: datetime, config: WaitPolicyConfig) -> bool:
    """Whether the configured max_wait budget has been consumed and the run loop
    should give up rather than schedule another probe."""
    if config.max_wait is None:
        return False
    return now - started_waiting_at >= config.max_wait

claudeloop.domain.budget

Budget guardrails for an unattended, potentially multi-hour/multi-day run.

BudgetLedger dataclass

Tracks consumption against a Budget. Immutable — every spend returns a new ledger, so the run loop's state transitions stay pure and testable.

Source code in src/claudeloop/domain/budget.py

@dataclass(frozen=True, slots=True)
class BudgetLedger:
    """Tracks consumption against a Budget. Immutable — every spend returns a new
    ledger, so the run loop's state transitions stay pure and testable."""

    budget: Budget
    turns_spent: int = 0
    dollars_spent: float = 0.0
    attempts_spent: int = 0

    def spend_turn(self, *, dollars: float = 0.0) -> BudgetLedger:
        return replace(
            self,
            turns_spent=self.turns_spent + 1,
            dollars_spent=self.dollars_spent + dollars,
        )

    def spend_attempt(self) -> BudgetLedger:
        return replace(self, attempts_spent=self.attempts_spent + 1)

    @property
    def turns_exhausted(self) -> bool:
        return self.budget.max_turns is not None and self.turns_spent >= self.budget.max_turns

    @property
    def dollars_exhausted(self) -> bool:
        return self.budget.max_dollars is not None and self.dollars_spent >= self.budget.max_dollars

    @property
    def attempts_exhausted(self) -> bool:
        return (
            self.budget.max_attempts is not None and self.attempts_spent >= self.budget.max_attempts
        )

    @property
    def any_exhausted(self) -> bool:
        return self.turns_exhausted or self.dollars_exhausted or self.attempts_exhausted

claudeloop.domain.loop

The autonomous run loop's pure state machine.

application.runner.AutonomousRunner executes the Decisions this module produces against real ports (agent gateway, clock, sleeper, ...). Nothing in this module performs I/O; every transition is a function of (RunState, an event, now).

RunProbe dataclass

Spend a cheap, throwaway turn purely to re-check capacity.

Source code in src/claudeloop/domain/loop.py

@dataclass(frozen=True, slots=True)
class RunProbe:
    """Spend a cheap, throwaway turn purely to re-check capacity."""

SendTurn dataclass

Spend a real turn against the live session.

Source code in src/claudeloop/domain/loop.py

@dataclass(frozen=True, slots=True)
class SendTurn:
    """Spend a real turn against the live session."""

decide_after_probe(state, capacity, *, now, config=DEFAULT_WAIT_POLICY_CONFIG)

Called once a throwaway probe turn has completed while waiting.

Source code in src/claudeloop/domain/loop.py

def decide_after_probe(
    state: RunState,
    capacity: CapacityState,
    *,
    now: datetime,
    config: WaitPolicyConfig = DEFAULT_WAIT_POLICY_CONFIG,
) -> tuple[RunState, Decision]:
    """Called once a throwaway probe turn has completed while waiting."""
    if isinstance(capacity, AuthenticationFailed):
        return _fail(state, "authentication failed"), Finish(
            success=False, reason="authentication failed"
        )
    if isinstance(capacity, Available):
        return RunState(phase=Phase.RUNNING, ledger=state.ledger), SendTurn()
    return _enter_waiting(state, capacity, now=now, config=config, is_reprobe=True)

decide_after_turn(state, *, capacity, verdict, now, config=DEFAULT_WAIT_POLICY_CONFIG)

Called once a real turn has completed. A capacity rejection always outranks a completion claim — a limit message truncating mid-response could coincidentally contain marker-like text, but hitting a real limit is never "done".

Source code in src/claudeloop/domain/loop.py

def decide_after_turn(
    state: RunState,
    *,
    capacity: CapacityState,
    verdict: CompletionVerdict,
    now: datetime,
    config: WaitPolicyConfig = DEFAULT_WAIT_POLICY_CONFIG,
) -> tuple[RunState, Decision]:
    """Called once a real turn has completed. A capacity rejection always outranks a
    completion claim — a limit message truncating mid-response could coincidentally
    contain marker-like text, but hitting a real limit is never "done"."""
    new_ledger = state.ledger.spend_turn()

    if isinstance(capacity, AuthenticationFailed):
        return _fail(state, "authentication failed"), Finish(
            success=False, reason="authentication failed"
        )

    if not isinstance(capacity, Available):
        return _enter_waiting(
            RunState(phase=state.phase, ledger=new_ledger), capacity, now=now, config=config
        )

    if isinstance(verdict, Done):
        return (
            RunState(phase=Phase.COMPLETE, ledger=new_ledger),
            Finish(success=True, reason=verdict.summary),
        )
    if isinstance(verdict, Blocked):
        return (
            RunState(phase=Phase.FAILED, ledger=new_ledger, failure_reason=verdict.reason),
            Finish(success=False, reason=verdict.reason),
        )
    # Precondition, not a security gate: CompletionVerdict is the closed union
    # {Done, Blocked, Continue} and both other members are handled above, so this
    # is exhaustive by construction — asserted here to fail loudly if a future
    # variant is added to the union without a matching branch here.
    assert isinstance(verdict, Continue)  # nosec B101

    running = RunState(phase=Phase.RUNNING, ledger=new_ledger)
    if new_ledger.any_exhausted:
        return _fail(running, "budget exhausted"), Finish(success=False, reason="budget exhausted")
    return running, SendTurn()

decide_preflight(state, capacity, *, now, config=DEFAULT_WAIT_POLICY_CONFIG)

The very first thing a run does: check whether we're already mid-cooldown before spending a real attempt (mirrors preflight_wait() in the legacy script).

Source code in src/claudeloop/domain/loop.py

def decide_preflight(
    state: RunState,
    capacity: CapacityState,
    *,
    now: datetime,
    config: WaitPolicyConfig = DEFAULT_WAIT_POLICY_CONFIG,
) -> tuple[RunState, Decision]:
    """The very first thing a run does: check whether we're already mid-cooldown
    before spending a real attempt (mirrors preflight_wait() in the legacy script)."""
    if isinstance(capacity, AuthenticationFailed):
        return _fail(state, "authentication failed"), Finish(
            success=False, reason="authentication failed"
        )
    if isinstance(capacity, Available):
        return RunState(phase=Phase.RUNNING, ledger=state.ledger), SendTurn()
    return _enter_waiting(state, capacity, now=now, config=config)