Metadata-Version: 2.4
Name: loopgraph
Version: 0.5.1
Summary: Event-driven graph workflow engine with native loop support.
Author: LoopGraph Team
License: MIT
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Provides-Extra: test
Requires-Dist: pytest>=7.4.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "test"
Provides-Extra: lint
Requires-Dist: ruff>=0.6.4; extra == "lint"
Requires-Dist: mypy>=1.11.0; extra == "lint"

# LoopGraph

**The agent workflow engine that treats loops as a first-class citizen.**

Build graph-based AI agent workflows where cycles, re-entry, and iterative reasoning are native — not hacks. More intuitive and lightweight than LangGraph, with loops as a first-class primitive.

- **Zero dependencies** — pure Python 3.10+, nothing to install beyond your own agents
- **Native loop support** — cycles in your graph are validated, tracked, and safe by design
- **Event-driven** — every state transition emits events; hook in logging, metrics, or triggers anywhere
- **Async-first** — built on `asyncio`; ready nodes in a single run execute concurrently, bounded by your concurrency policy
- **Recoverable** — snapshot and replay any run from any point

```bash
pip install loopgraph
```

---

## Quickstart

```python
import asyncio
from loopgraph.core.graph import Graph, Node, Edge, NodeKind
from loopgraph.bus.eventbus import EventBus
from loopgraph.registry.function_registry import FunctionRegistry
from loopgraph.scheduler.scheduler import Scheduler

async def my_agent(payload):
    # your agent logic here
    return {"result": "done", "loop_again": False}

async def router(payload):
    # return the next node id
    return "end" if not payload.get("loop_again") else "agent"

graph = Graph(
    nodes=[
        Node(id="agent", kind=NodeKind.TASK),
        Node(id="router", kind=NodeKind.SWITCH),
        Node(id="end", kind=NodeKind.TASK),
    ],
    edges=[
        Edge(source="agent", target="router"),
        Edge(source="router", target="agent"),   # the loop back-edge
        Edge(source="router", target="end"),
    ],
    entry="agent",
)

registry = FunctionRegistry()
registry.register("agent", my_agent)
registry.register("router", router)
registry.register("end", lambda p: p)

bus = EventBus()
scheduler = Scheduler(graph=graph, registry=registry, bus=bus)

asyncio.run(scheduler.run(payload={"input": "hello"}))
```

---

## Why LoopGraph?

Most workflow engines assume a DAG — a graph with no cycles. That works for linear pipelines, but agent workflows are inherently iterative: an agent reasons, reflects, decides to try again, and loops back. Forcing that into a DAG requires awkward workarounds.

LoopGraph makes loops explicit and safe:

- **Back-edges are first-class** — declare a cycle in your graph and the engine handles reset, visit tracking, and state management automatically
- **Loop safety** — the engine validates your graph at construction time; overlapping loops that share nodes are rejected before anything runs
- **Full observability** — every loop iteration emits events (`NODE_SCHEDULED`, `NODE_COMPLETED`, `NODE_FAILED`) so you always know where you are

---

## Event Hooks

Subscribe to any workflow event to add logging, metrics, or side effects:

```python
from loopgraph.core.types import EventType

async def on_completed(event):
    print(f"{event.node_id} finished → {event.payload}")

bus.subscribe(EventType.NODE_COMPLETED, on_completed)
```

Available events: `NODE_SCHEDULED`, `NODE_STARTED`, `NODE_COMPLETED`, `NODE_FAILED`.

---

## Custom Events from Handlers

Wrap your handler in a closure to emit custom events mid-execution:

```python
def make_handler(bus, base_handler):
    async def wrapper(payload):
        await bus.emit(Event(id="pre", graph_id="g", node_id="n",
                             type=EventType.NODE_SCHEDULED, payload={"stage": "pre"}))
        result = await base_handler(payload)
        await bus.emit(Event(id="post", graph_id="g", node_id="n",
                             type=EventType.NODE_COMPLETED, payload={"stage": "post"}))
        return result
    return wrapper

registry.register("my_node", make_handler(bus, my_agent))
```

---

## Loop Re-entry Rules

- Re-entry is triggered by a `SWITCH` node selecting a back-edge
- Only `COMPLETED` nodes can be reset for re-entry
- Reset clears upstream-completion tracking and preserves cumulative `visit_count`
- Overlapping loops sharing any node are rejected at graph construction time

## Scheduler Semantics

- The scheduler seeds its internal pending set from graph entry nodes only. A
  node enters pending later only when an upstream edge actually activates it.
- Unselected `SWITCH` branches never enter pending, so leaf branches that were
  not chosen cannot deadlock the workflow.
- A graph with nodes but no entry nodes now fails fast with `ValueError`
  instead of entering a deadlocked run loop.
- If a `SWITCH` returns a route that matches no downstream edge and no
  `exit` fallback edge exists, the scheduler raises `ValueError`.
- `NodeKind.TERMINAL` keeps the same runtime scheduling semantics as `TASK`.

## Concurrency Within a Run (0.4)

Since 0.4, the scheduler dispatches every ready node the moment it becomes
ready (eager dispatch), so independent branches of a fan-out genuinely run
in parallel — bounded by your `ConcurrencyManager`. Nothing new to
configure: give the policy a capacity above 1 and a diamond fan-out runs in
roughly one branch's wall time instead of the sum.

- **Capacity is the only concurrency knob.** `SemaphorePolicy(limit=1)` keeps
  execution sequential (at most one node in flight, ever). The
  decision-sensitive recovery checkpoint described below is a narrow
  persistence-cadence correction, independent of capacity.
- **Deterministic dispatch order.** Simultaneously-ready nodes dispatch in
  graph-definition order, replacing the previous seed-dependent set order.
- **Merges are "first-k".** An `AGGREGATE` with `config={"required": k}`
  fires at its k-th input completion, exactly once per activation, with the
  k earliest results in graph-definition order. `k` defaults to the full
  fan-in — i.e. wait-for-all, unchanged. Inputs finishing after the merge
  fired are recorded silently; they never re-fire it and never error. With a
  `SnapshotStore` configured, the chosen input is frozen in a private,
  visit-scoped envelope inside the existing version-2 `last_payload`, so a
  queued/running crash resumes with the same winners. Every aggregate with
  `k < fan-in` immediately persists an isolated marker before task creation;
  fresh execution keeps the original input object and replay receives a new
  isolated marker copy, so handler mutation cannot alter the durable decision.
  On loop re-entry, only
  upstreams recorded for the current activation can be winners — stale
  run-level results never displace the input that reopened the quorum.
- **Partial-input nodes** (`allow_partial_upstream=True`) receive the
  result of the input whose completion made them ready. Multi-input partial
  nodes use the same immediate isolated-marker/replay rule.
- **Error tolerance stays handler-owned.** A branch that may fail catches
  its own exception and returns it in the payload (`{"ok": False, ...}`);
  a wait-for-all merge handler then enforces "at least k successes". The
  engine itself remains fail-fast: the first uncaught handler exception
  cancels in-flight siblings and surfaces unchanged from `run()`.

**Upgrading from 0.3**: pipelines, switch chains, loops, and wait-for-all
merges retain their ordinary results, events, and final snapshots with no
migration. Decision-sensitive first-k/partial active snapshots have the
recovery-only envelope and extra save described above.
For fan-out graphs, note: lifecycle events of concurrent nodes interleave
(per-node `SCHEDULED`→`COMPLETED` ordering per visit is still guaranteed);
`NODE_SCHEDULED` now means execution actually started, not queued;
mid-run snapshots may show several `running` nodes (resume handles this), and
a pending/running decision-sensitive node's existing `last_payload` may
temporarily contain the private fire-decision envelope described above;
quorum merges (`required` < fan-in) now fire deterministically exactly
once — previously they could double-fire depending on process hash seed;
and handlers in parallel branches can interleave at their `await` points,
so shared mutable state needs handler-level coordination.

## Sub-Workflows (0.5)

Since 0.5, a workflow can be packaged as a single node: `NodeKind.SUBGRAPH`
carries a complete child graph in `config={"graph": child.to_dict()}`. The
child runs to its single `TERMINAL` node, whose payload becomes the node's
result — from the parent's perspective a sub-workflow is just another task
(all downstream edges activate; it never routes). This also unlocks the
fan-out+merge-inside-a-loop topology the flat validator rejects: pack the
diamond as a child and the parent loop is a single clean cycle.

- **Child shape.** A child needs exactly one entry node and exactly one
  `TERMINAL` node, statically reachable, with no outgoing edges. Terminal
  execution *ends the child run*: in-flight child work is cancelled and
  awaited, pending work never dispatches. Top-level `TERMINAL` nodes keep
  their 0.4 behavior — nothing stops early outside child runs. Embedded
  `nodes`/`edges` may be lists, tuples, or other repeatable iterables; one-shot
  iterators/generators are rejected before consumption because validation,
  serialization, and execution must see the same definition.
- **Canonical child serialization.** `Graph.to_dict()` reparses every embedded
  SUBGRAPH body and emits the canonical graph schema recursively. An
  unparseable body or a composition cycle raises `ValueError` during
  serialization. Unsupported schema-level keys on raw graph, node, or edge
  records are not emitted; arbitrary values inside the supported `Node.config`
  and `Edge.metadata` extension containers are preserved.
- **Deadlock-free by construction.** The sub-workflow node holds one unit
  of parent capacity like any task, while the child's nodes draw only on
  the child's own budget: unlimited by default, `config={"concurrency_limit": N}`
  for a serialized cap, or an explicit manager instance via
  `Scheduler(..., child_concurrency={"node/path": manager})`. Share one
  instance between *siblings* or across runs for a global ceiling — never
  an ancestor's manager (the engine rejects that wiring; it is exactly the
  old hand-rolled deadlock). The unlimited default means a wide child fans
  out fully — and nested children multiply (branching b at depth d can
  reach b^d concurrent handlers) — so cap resource-hungry children
  explicitly. A queued `PrioritySemaphorePolicy` acquisition removed by
  cancellation cannot poison later sibling/cross-run users of that manager.
- **Namespaced run identifiers (public contract).** A child run's state and
  events live under `{parent}/{node}@{visit}` (visits numbered from 1;
  nests as e.g. `orders/etl@2/load@1`). Split on `/`, then `@` — the
  segments give nesting level, parent node, and visit for every event; no
  new event fields exist. Root run ids and node ids in composed workflows
  must not contain `/` or `@`; composed root runs enforce that grammar
  iteratively before dispatch even if the caller omitted `Graph.validate()`.
- **Loop re-entry.** Every permitted visit (bounded by `max_visits` as
  usual) runs the child fresh under a new `@{visit}` namespace; resume
  after a crash targets the latest visit. A back-edge reaching a
  sub-workflow node while its child is still running hard-stops the run,
  like any non-terminal re-entry target.
- **Failure is black-box node-equivalent.** A failed child run fails the
  node exactly like a failed handler (same marking, event, snapshot,
  unwrapped exception); a failure the child's own composition tolerates is
  not a node failure. The child's failure-time state stays under its
  namespace for post-mortem. If recovery finds that failure in the child
  snapshot before the parent recorded it, the failed handler is not retried.
  Snapshot format 2 stores the failure detail but not its Python exception
  type, so recovery raises a descriptive `RuntimeError` naming the child run,
  failed node, and recorded detail; the parent then records that reconstructed
  failure through its ordinary node-failure path.
- **Cancellation and listener failures.** Cancellation containment applies to
  child runs and roots containing sub-workflows: descendant tasks are cancelled
  and awaited before the composed run surfaces cancellation. Plain legacy roots
  retain 0.4 cancellation timing: a cancelled flat `run()` may surface before
  its in-flight handler tasks finish, so those tasks may still emit events or
  write snapshots afterwards. `EventBus` accounts for every ordinary listener
  task. If its caller is cancelled while listeners are still running, unfinished
  listeners are cancelled and fully drained; failures completed before
  cancellation or raised during listener cleanup still reach configured
  `on_error` (or a warning). Repeated `cancel()` calls cannot replace the first
  caller-cancellation arguments. The first callback failure propagates only
  after routing unless caller cancellation has precedence. Awaited cleanup
  drains absorb repeated cancel requests, so a listener or `on_error` callback
  that never returns blocks cancellation indefinitely — write them to
  terminate. If `on_error` itself raises
  `CancelledError`, that is a callback-owned failure and surfaces as
  `RuntimeError` with the cancellation as its cause; only cancellation of the
  `emit()` caller propagates as the original `CancelledError`. This distinction,
  listener/callback warnings, and warning visibility for suppressed sibling
  failures enforce Constitution Principle XV
  rather than silently mistaking callback failure for caller cancellation.
  Lifecycle callback failures leave deterministic node state: scheduled failure
  persists FAILED even though the handler never ran, failure-event callbacks
  never replace the original node error, and completed-event callbacks propagate
  only after completion/downstream readiness are durable. If downstream
  finalization itself hard-fails, that engine error wins and the callback failure
  is logged.
- **State retention & cleanup.** Every visit's child state is retained; the
  engine never deletes persisted state and `SnapshotStore` gains no
  list/delete operations. Clean up through your backend's own tooling
  using the `{root}/{node}@{visit}` prefix — never delete state belonging
  to an active or resumable root run (removing an unfinished child
  snapshot makes resume re-run that work). The in-memory store is cleaned
  by discarding the store instance once the run is no longer needed.
- **Runaway-nesting backstop.** Composition cycles are rejected at
  `validate()` with the containment chain named; a runtime backstop
  (ancestor identity + `Scheduler(..., max_subworkflow_depth=64)`,
  inclusive, positive integers only) converts unvalidated or mutated
  pathological compositions into a clean error instead of unbounded
  recursion.

### Handler namespaces

Handler names may use the qualified form `publisher.package/function`
(e.g. `acme.etl/clean`) — plain registry keys, so two publishers' `clean`
functions coexist with zero collisions. Bare names remain valid forever as
local names. Authoring convenience: a definition may carry a transient
top-level `"namespace"` key (or `Graph(..., namespace="acme.etl")`) that
qualifies its bare names at load; canonical `to_dict()` output always
contains the fully qualified names and never the directive. The
`name@version` suffix is **reserved** for a future installation story:
referencing it in a workflow that uses sub-workflows or declares a
namespace fails with a reservation error today.

## Recovery Boundaries

- Persisted scheduler snapshots now include `snapshot_format_version`.
- Resume is supported only for snapshots with the current supported snapshot
  format version.
- If a snapshot is missing `snapshot_format_version` or carries an unsupported
  version, resume fails fast with a `ValueError` that reports the actual
  version, the supported version, and discard-or-migrate guidance.
- On resume, pending is rebuilt from uncompleted entry nodes plus nodes already
  persisted as `PENDING` or `RUNNING`. Persisted `RUNNING` nodes are reset to
  `PENDING` before scheduling.
- With a `SnapshotStore` configured, every decision-sensitive aggregate
  (`required < fan-in`) and multi-input partial activation persists an isolated
  copy of its chosen input in a private visit envelope inside existing
  `last_payload` immediately before task creation; no snapshot key or format
  version is added. Fresh execution keeps
  the original input identity, while resume receives a new isolated marker copy,
  so mutable handler input cannot alter the durable decision, including in a
  reference-retaining snapshot. Resume therefore replays the same input at every
  sub-workflow depth. Re-entered aggregates use current-activation upstream
  marks rather than stale results from prior visits.
- Since 0.5, every public root run consults its configured event log at start and
  continues event-id numbering above the highest identifier already recorded
  for that `graph_id`, even when no snapshot remains. A private child consults
  the log only when its own snapshot loaded or the parent identifies its derived
  visit as a recovery candidate. In a non-recovered parent run, a fresh child
  visit does not inherit stale log-only history. A recovered parent
  conservatively supplies that hint on each SUBGRAPH node's first dispatch in
  the current run process, so the hinted child may consult the log even when its
  derived visit is new.
  On actual resume, snapshot `last_event_id` values remain the fallback when the
  log has no parseable ids. Event-log backends may report an unknown `graph_id`
  with `LookupError`; the scheduler treats that as empty history only if lookup
  yielded no events. A `LookupError` after any yielded event propagates because
  the history is truncated and its maximum is unknowable. Thus public-root and
  recovery-seeded-child `(graph_id, event_id)` pairs stay unique across
  crash/resume. Any composed invocation that reuses a root `graph_id` without
  loading its parent snapshot — including crash recovery with an event log but
  no snapshot store, where log-only recovery keeps the root's own ids
  collision-free but not child namespaces — may re-derive old child ids while
  starting those engine-fresh child counters at 1. Use a new root id to
  preserve lifetime pair uniqueness. Clearing or partitioning retained child logs instead begins a new
  consumer epoch, so reset child cursors/deduplication state and include that
  epoch or partition outside the pair. Event consumers must not assume
  recovery-seeded counters restart and should retain
  cursors/deduplication state across those recoveries; resumed snapshots carry
  continued `last_event_id` values.
- A `NODE_SCHEDULED` listener failure is persisted as a terminal FAILED visit
  before the handler runs. Snapshot format 2 records failure detail but has no
  origin marker, so resume cannot distinguish this case from a handler failure
  and will not retry it. To deliberately rerun, an operator must discard,
  migrate, or reset the affected child and ancestor snapshots using backend
  administration after the run is inactive, accepting the normal cleanup and
  re-execution consequences; the format is not extended with a lifecycle-
  failure marker.
- Narrow exceptional-path correctness changes also ship in 0.5: all gathered
  ordinary listener errors reach configured `on_error`, lifecycle-dispatch
  errors preserve coherent snapshots/original node errors, and cancelled
  queued priority acquisitions are removed. These may add warning records or
  change state observed after a failing listener; ordinary successful runs and
  handler failures without failing listeners keep their prior surface.

---

## Installation

```bash
pip install loopgraph
```

Requires Python 3.10+. No runtime dependencies.

---

## Known Limitations

### Cancellation reason strings on Python 3.10

Cancellation itself works identically on every supported Python: cancelling a
run (or a subgraph, or an in-flight event dispatch) still stops it, drains
in-flight siblings and nested children, and preserves snapshot/resume
invariants.

What is **not** reliable on Python 3.10 is the *reason string* attached with
`task.cancel("reason")`. loopgraph propagates that label back to whoever awaits
the run so callers can tell *why* it stopped (e.g. `"deadline-exceeded"` vs a
parent failure vs shutdown), and it keeps the first/initiating reason when
composed teardown triggers further cancellations. That relies on cancellation
messages surviving propagation through `asyncio.wait`/task-await, which only
holds on **Python 3.11+**. On 3.10 the message is dropped and the propagated
`CancelledError` arrives with empty `args` — the control flow is unaffected,
only the diagnostic label is lost.

The tests that assert the reason label are skipped below Python 3.11. If your
orchestration depends on cancellation reason strings, run on Python 3.11 or
newer.

---

## Development

```bash
git clone https://github.com/your-org/loopgraph
cd loopgraph
python3 -m venv .venv && source .venv/bin/activate
pip install -e ".[test,lint]"
pytest
```

---

## Design Principles

- **Keep the core compact.** Nodes stay stateless and the scheduler stays simple, with minimum opinionated design and maximum freedom for users to compose their own workflow patterns. Handlers capture their own context (event bus, metrics, side effects) so the framework never grows special cases for custom behaviour.
- **Push heavy lifting to the edge.** Long-running work should run via remote APIs, threads, or separate nodes/clusters. We avoid building a distributed fan-out scheduler; within one process the engine executes ready nodes concurrently under an explicit capacity policy, and users orchestrate anything beyond the process boundary.
- **Flexible aggregation semantics.** Aggregator nodes may proceed when only a subset of upstream nodes finish — as long as those nodes reach a terminal state. Fail-fast and error-tolerance are user-level workflow patterns, and the engine stays policy-light so users can implement either.
- **Retries live with handlers.** The framework doesn't implement automatic retries. Each handler decides whether to retry, abort, or compensate, keeping recovery logic close to the business code.
- **Pluggable concurrency.** A shared ConcurrencyManager (semaphore or priority-aware) controls global slots. Multiple schedulers can share one manager, but there's no hidden magic — users choose the policy, preserving clarity and control.
- **Recovery through snapshots.** The engine snapshots execution state and event logs so users can resume or replay runs without re-executing nodes. Payloads flow naturally between nodes, satisfying replay needs without extra APIs.
