Metadata-Version: 2.4
Name: flowstepper
Version: 0.1.0
Summary: DAG-based pipeline visualizer — log from any language, explore in the browser
Author-email: Pavel Elizarov <pavel.elizarov.an@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/pavelthei/flowstepper
Project-URL: Bug Tracker, https://github.com/pavelthei/flowstepper/issues
Project-URL: Source, https://github.com/pavelthei/flowstepper
Keywords: pipeline,dag,visualization,ml
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Visualization
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastapi>=0.110.0
Requires-Dist: uvicorn[standard]>=0.29.0
Requires-Dist: click>=8.1.0
Requires-Dist: watchfiles>=0.21.0
Requires-Dist: orjson>=3.9.0
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.0; extra == "dev"
Requires-Dist: httpx>=0.27.0; extra == "dev"
Requires-Dist: ruff>=0.4.0; extra == "dev"
Requires-Dist: mypy>=1.9.0; extra == "dev"
Dynamic: license-file

# Flowstepper

**Visualize pipelines as interactive DAGs — from log files.**

Works like TensorBoard: instrument your code with the SDK, run `flowstepper`, and explore your pipeline in a browser.

[![PyPI version](https://img.shields.io/pypi/v/flowstepper.svg)](https://pypi.org/project/flowstepper/)
[![Python](https://img.shields.io/pypi/pyversions/flowstepper.svg)](https://pypi.org/project/flowstepper/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

---

## What is Flowstepper?

Flowstepper is an **observability tool for pipelines**. It lets you:

- **Instrument** your pipeline stages with a lightweight SDK (zero-blocking, production-safe)
- **Visualize** the DAG structure, node timings, and runtime metadata in a browser
- **Debug** slow stages, data drops, and execution errors across runs
- **Compare** different runs side-by-side to track regressions

The SDK writes structured `.flowlog` files. The viewer reads them — no database, no external service, no setup beyond `pip install`.

---

## Quick Start

### Install

```bash
pip install flowstepper
```

### Instrument your pipeline

```python
from flowstepper.sdk import BaseNode, Pipeline

# Optional: pick a colour per kind shown in the UI
COLORS = {
    "source": "#3b82f6",
    "ranker": "#22c55e",
    "filter": "#f59e0b",
}

with Pipeline("recommendation", log_dir="./logs", node_kind_colors=COLORS) as p:
    src  = p.add_node(BaseNode("catalog", kind="source"))
    pre  = p.add_node(BaseNode("scorer", kind="ranker"))
    rank = p.add_node(BaseNode("ltr_model", kind="ranker",
                               tags={"model_name": "xgb", "model_version": "v3"}))
    filt = p.add_node(BaseNode("dedup", kind="filter"))

    p.connect(src, pre)
    p.connect(pre, rank)
    p.connect(rank, filt)

    with p.node_scope(src):
        items = fetch_catalog()
        p.record(src, item_count=len(items))

    with p.node_scope(pre):
        scored = score(items)
        p.record(pre, scored_count=len(scored))

    with p.node_scope(rank):
        ranked = ltr_model.rank(scored)

    with p.node_scope(filt):
        results = dedup(ranked)
        p.record(filt, output_count=len(results))
```

### Launch the viewer

```bash
flowstepper --logdir ./logs
# Open http://localhost:6006
```

---

## Features

| Feature | Description |
|---------|-------------|
| **Interactive DAG** | Auto-laid-out graph using dagre; click nodes to inspect metadata |
| **Timeline trace** | Gantt-style execution timeline, similar to Chrome DevTools |
| **Live reload** | `--reload` flag tails new runs as they land |
| **Zero-overhead sampling** | `sample_rate=0.0` makes all SDK calls pure no-ops |
| **Production-safe** | Logging failures never propagate; configurable drop policies |
| **No external deps** | Just files — no database, no broker, no cloud account |
| **Language-agnostic format** | Any language can write `.flowlog` files |

---

## SDK

### Nodes

There is a single `BaseNode` class. Every node has a `name`, a free-form
`kind` string (any label you choose — e.g. `"retrieval"`, `"ranker"`,
`"tool_call"`), and optional `tags`:

```python
from flowstepper.sdk import BaseNode

BaseNode("catalog", kind="source")
BaseNode("ltr", kind="ranker", tags={"model_name": "xgb", "model_version": "v3"})
```

Colours for each `kind` are supplied once at pipeline init via
`node_kind_colors={"source": "#3b82f6", ...}`.

### Pipeline API

```python
# Add nodes
node = p.add_node(BaseNode("ranker", kind="ranker",
                           tags={"model_name": "xgb"}))

# Declare edges
p.connect(source_node, target_node)                        # stream (default)
p.connect(source_node, target_node, kind="batch")          # batch transfer
p.connect(signal_node, target_node, kind="signal")         # control signal

# Track execution
with p.node_scope(node) as scope:
    result = run_stage()
    p.record(node, latency_ms=12.4, output_size=len(result))
    if not result:
        scope.is_error = True  # soft-error flag

# Inspect
print(p.run_id)          # unique run identifier
print(p.is_sampled)      # whether this run is being logged
print(p.dropped_events)  # events dropped due to queue pressure
```

### Logging Configuration

```python
from flowstepper.sdk import Pipeline, LoggingConfig, DropPolicy

# Development — log everything
with Pipeline("my_pipeline", log_dir="./logs") as p:
    ...

# Production — 1% sample rate, bounded queue
cfg = LoggingConfig(
    sample_rate=0.01,
    queue_maxsize=10_000,
    drop_policy=DropPolicy.DROP_NEWEST,
    flush_interval_ms=200,
    flush_batch_size=256,
    on_drop=lambda n: metrics.increment("flowstepper.drops", n),
    on_error=lambda e: logger.warning("flowstepper error", exc_info=e),
)

with Pipeline("my_pipeline", log_dir="./logs", config=cfg) as p:
    ...
```

**Built-in presets:**

```python
LoggingConfig.for_development()   # ALWAYS sample, sync flush
LoggingConfig.for_production()    # RATE sample, async flush, DROP_NEWEST
LoggingConfig.disabled()          # NEVER sample — all calls are no-ops
```

---

## Architecture

```
User Code ──► SDK ──► async queue ──► Logger (background thread) ──► .flowlog files
                                                                             │
                                                                    LogParser (reader/)
                                                                             │
                                                                    FastAPI Server
                                                                             │
                                                                    React SPA (browser)
```

**Key design principles:**

- **Hot-path zero-blocking** — sampling decision made once at pipeline entry; all logging calls enqueue asynchronously
- **Bounded memory** — configurable queue size with drop policies (`DROP_NEWEST`, `DROP_OLDEST`, `BLOCK`)
- **Error isolation** — logging failures are never propagated to caller code
- **Stateless reader** — `LogParser` is a pure function: `Path → PipelineRun`
- **Language-agnostic** — log format is newline-delimited JSON; any language can produce logs

### Log Format

Logs are stored as newline-delimited JSON at `<log_dir>/<pipeline_name>/<run_id>.flowlog`.

Eight event types: `pipeline_start`, `pipeline_end`, `node_declared`, `edge_declared`, `node_start`, `node_end`, `node_error`, `metadata`.

See [Log Format Specification](https://github.com/pavelthei/flowstepper/blob/main/docs/log-format.md) for the full schema.

---

## Installation from Source

```bash
git clone https://github.com/pavelthei/flowstepper
cd flowstepper
pip install -e ".[dev]"
```

The editable install automatically builds the frontend bundle. To rebuild manually after frontend changes:

```bash
./flowstepper/scripts/build_frontend.sh
```

### Running Tests

```bash
pytest tests/                              # all tests
pytest tests/sdk/test_logger.py -v         # single file
ruff check flowstepper tests               # lint
mypy flowstepper                           # type check
```

---

## SDK Language Support

| Language | Status    |
|----------|-----------|
| Python 3 | Supported |
| Java     | Planned   |
| Go       | Planned   |

---

## Documentation

- [Architecture](https://github.com/pavelthei/flowstepper/blob/main/docs/architecture.md) — design rationale and component overview
- [Log Format](https://github.com/pavelthei/flowstepper/blob/main/docs/log-format.md) — event schema specification
- [Python SDK](https://github.com/pavelthei/flowstepper/blob/main/docs/sdk/python.md) — full SDK reference

---

## License

[MIT](https://github.com/pavelthei/flowstepper/blob/main/LICENSE)
