### Install mysorf-flow with mysorf-base runtime

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Recommended installation that includes the mysorf-base runtime along with mysorf-flow.

```bash
pip install \
  git+https://github.com/mysorf-9239/mysorf-base.git@v0.1.0 \
  git+https://github.com/mysorf-9239/mysorf-flow.git@v0.1.0
```

--------------------------------

### Define and run a simple flow

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Example demonstrating how to register a plugin, define a flow, and then build and run it.

```python
from mysorf_flow.api import (
    register_plugin,
    register_flow,
    build_runner,
)
from mysorf_flow.core.contracts import (
    FlowDefinition,
    StagePlan,
    StageSpec,
    ModuleSpec,
    PluginSpec,
    CapabilityProfile,
)

# 1. Register a plugin
@register_plugin("normalize")
def build_normalizer(spec):
    def run(batch, state):
        return {"normalized": [x / 255.0 for x in batch["pixels"]]}
    return run

# 2. Define a flow
flow_def = FlowDefinition(
    name="preprocess",
    stage_plan=StagePlan(stages=[
        StageSpec(
            name="encode",
            modules=[
                ModuleSpec(
                    id="norm",
                    plugin=PluginSpec(id="normalize"),
                    declared_outputs=["normalized"],
                    required_inputs=["pixels"],
                    capability_profile=CapabilityProfile(),
                ),
            ],
        ),
    ]),
)

# 3. Build and run
runner = build_runner(flow_def)
result = runner.run({"pixels": [128, 64, 255]}, mode="infer")
```

--------------------------------

### Install mysorf-flow from GitHub

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Use this command to install the mysorf-flow package directly from its GitHub repository.

```bash
pip install git+https://github.com/mysorf-9239/mysorf-flow.git@v0.1.0
```

--------------------------------

### Commit Message Examples

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/CONTRIBUTING.md

Provides examples of valid commit messages following the specified format.

```text
feat: add train_only flag to ModuleSpec and V9 validator
fix: validator V5 now checks same-stage outputs correctly
docs: add structural validator table to README
```

--------------------------------

### Local editable install of mysorf-flow

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Install the project locally in editable mode, including development dependencies and pre-commit hooks.

```bash
git clone https://github.com/mysorf-9239/mysorf-flow.git
cd mysorf-flow
pip install -e ".[dev]"
pre-commit install
```

--------------------------------

### Repository Layout

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Illustrates the directory structure of the mysorf-flow project, detailing the organization of source code, examples, tests, and configuration files.

```text
mysorf-flow/
├── src/mysorf_flow/      # Python package (src layout)
│   ├── api/              # Public API re-exports
│   │   ├── flows.py      # register_flow, build_runner
│   │   ├── plugins.py    # register_plugin, build_plugin
│   │   ├── runner.py     # run_registered_validators
│   │   ├── taskpacks.py  # register_task_pack, load_task_pack
│   │   └── validators.py # flow validator registry
│   ├── core/             # Engine internals
│   │   ├── contracts.py  # FlowDefinition, StageSpec, ModuleSpec, ...
│   │   ├── runner.py     # Runner with training API
│   │   ├── validators.py # V1–V9 structural validators
│   │   ├── pipeline.py   # Stage execution
│   │   └── state.py      # Write-once execution state
│   ├── flows/            # Flow registry
│   ├── taskpacks/        # Task-pack loader and registry
│   └── protocols.py      # RuntimeProtocol, FlowValidatorProtocol
├── examples/             # Annotated usage examples
│   └── toy_taskpack/     # End-to-end tutorial task pack
├── tests/                # Test suite (pytest)
├── pyproject.toml        # Packaging and tool configuration
└── Makefile              # Local workflow commands
```

--------------------------------

### Iterate Modules by ID

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Iterate through modules using `iter_modules()` for tasks like optimizer setup. This is useful for accessing module IDs and their instances.

```python
for module_id, instance in runner.iter_modules():
    print(module_id, type(instance).__name__)
```

--------------------------------

### Task pack registration and loading

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Shows how to register a custom task pack with the engine and then load it.

```python
from mysorf_flow.api import register_task_pack, load_task_pack

register_task_pack("my_pack", MyTaskPack())
pack = load_task_pack("my_pack")
```

--------------------------------

### Register and Run a Task Pack

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Registers a task pack with its metadata, loads it, builds a flow definition, constructs a runner, and executes the flow with sample input. This demonstrates the typical lifecycle of using a task pack.

```python
metadata = TaskPackMetadata(
    name="toy-pack",
    version="0.1.0",
    engine_compat=">=0.1,<0.2",
    description="Minimal toy task pack.",
    task_types=("toy_binary",),
    default_task_type="toy_binary",
)

register_task_pack(metadata, register_toy_pack)
handle = load_task_pack("toy-pack")
flow_def = build_flow_definition(handle, "toy_baseline", config=None)
runner = build_runner(flow_def, runtime=None)

result = runner.run({"features": [1.0, 2.0, 3.0]}, mode="infer")
print(result)
```

--------------------------------

### Run a Mysorf Flow Definition

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/examples/toy_taskpack/README.md

This snippet demonstrates the typical usage pattern for loading a task pack, building a flow definition, and running it with a given configuration. Ensure the task pack is registered before loading.

```python
from types import SimpleNamespace

from examples.toy_taskpack import METADATA, register
from mysorf_flow.api import build_flow_definition, build_runner, load_task_pack, register_task_pack

register_task_pack(METADATA, register)
handle = load_task_pack("toy-taskpack")
flow_def = build_flow_definition(handle, "toy_baseline", config={})
runner = build_runner(flow_def, SimpleNamespace())

outputs = runner.run({"features": 3, "labels": 5}, mode="train")
```

--------------------------------

### Training API usage

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Demonstrates setting training mode, iterating through modules, collecting parameters, and managing checkpoints.

```python
# Set mode
runner.set_train()   # or runner.set_eval()

# Iterate modules (for optimiser setup)
for module_id, module in runner.iter_modules():
    print(module_id)

# Collect parameters (exclude frozen layers)
params = runner.collect_parameters(exclude_ids={"frozen_encoder"})

# Checkpoint
state = runner.state_dict()
runner.load_state_dict(state)
```

--------------------------------

### Development Commands

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Provides essential commands for local development, including code checking, formatting, testing, and building.

```bash
make check        # ruff + mypy + pytest
make format       # ruff format --fix
make test         # run test suite
make smoke-wheel  # build wheel + isolated import test
pre-commit run --all-files
```

--------------------------------

### Module Layout Convention

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/CONTRIBUTING.md

Illustrates the standard directory structure for the mysorf-flow project.

```text
src/mysorf_flow/
├── api/              # Thin re-export wrappers — the public surface
├── core/             # Engine internals (contracts, runner, validators, pipeline, state)
├── flows/            # Flow registry
├── taskpacks/        # Task-pack loader and registry
└── protocols.py      # Structural protocols (RuntimeProtocol, FlowValidatorProtocol)
```

--------------------------------

### Register and Load Task Packs

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Bundle plugins, flows, and validators into a task pack. Use `register_task_pack` for metadata and a hook, and `load_task_pack` to validate compatibility and activate the hook.

```python
from mysorf_flow.api import register_task_pack, load_task_pack, list_task_packs
from mysorf_flow.taskpacks.schemas import TaskPackMetadata
from mysorf_flow.core import CompatibilityError

METADATA = TaskPackMetadata(
    name="vision-pack",
    version="1.0.0",
    engine_compat=">=0.1,<0.2",
    description="Vision classification task pack.",
    task_types=("binary_classification", "multiclass"),
    default_task_type="binary_classification",
)

def registration_hook() -> None:
    from mysorf_flow.api import register_plugin
    from mysorf_flow.flows import register_flow
    register_plugin("representer", "cnn_repr", CNNRepresenter)
    register_flow("cnn_baseline", build_my_flow)

register_task_pack(METADATA, registration_hook)

# List what is registered (not yet loaded)
for summary in list_task_packs():
    print(summary.name, summary.version, summary.task_types)
# vision-pack 1.0.0 ('binary_classification', 'multiclass')

# Load activates the registration hook and validates engine version
try:
    handle = load_task_pack("vision-pack")
    print(handle.loaded)                          # True
    print(handle.context["registered_flows"])
    print(handle.context["registered_plugins"])
except CompatibilityError as e:
    print(e)  # engine version mismatch
```

--------------------------------

### register_task_pack / load_task_pack

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Bundles and loads domain logic into a self-contained unit. `register_task_pack` declares metadata and a registration hook, while `load_task_pack` validates compatibility, runs the hook, and returns a handle.

```APIDOC
## Task Pack API

### `register_task_pack` / `load_task_pack` — bundle and load domain logic

#### Description
A task pack bundles plugins, flows, and semantic validators into one self-contained unit. `register_task_pack` declares metadata and a registration hook. `load_task_pack` validates engine compatibility, runs the hook once, and returns a `TaskPackHandle` containing registered flow/plugin details.

#### Parameters for `register_task_pack`
- `metadata` (TaskPackMetadata): Metadata describing the task pack.
- `registration_hook` (callable): A function to be called when the task pack is loaded, responsible for registering plugins and flows.

#### Parameters for `load_task_pack`
- `name` (string): The name of the task pack to load.

#### Example Usage
```python
from mysorf_flow.api import register_task_pack, load_task_pack, list_task_packs
from mysorf_flow.taskpacks.schemas import TaskPackMetadata
from mysorf_flow.core import CompatibilityError

METADATA = TaskPackMetadata(
    name="vision-pack",
    version="1.0.0",
    engine_compat=">=0.1,<0.2",
    description="Vision classification task pack.",
    task_types=("binary_classification", "multiclass"),
    default_task_type="binary_classification",
)

def registration_hook() -> None:
    from mysorf_flow.api import register_plugin
    from mysorf_flow.flows import register_flow
    register_plugin("representer", "cnn_repr", CNNRepresenter) # Assuming CNNRepresenter is defined elsewhere
    register_flow("cnn_baseline", build_my_flow) # Assuming build_my_flow is defined elsewhere

register_task_pack(METADATA, registration_hook)

# List what is registered (not yet loaded)
for summary in list_task_packs():
    print(summary.name, summary.version, summary.task_types)
# vision-pack 1.0.0 ('binary_classification', 'multiclass')

# Load activates the registration hook and validates engine version
try:
    handle = load_task_pack("vision-pack")
    print(handle.loaded)                          # True
    print(handle.context["registered_flows"])
    print(handle.context["registered_plugins"])
except CompatibilityError as e:
    print(e)  # engine version mismatch
```
```

--------------------------------

### Runtime Protocol Mocks with SimpleNamespace

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Create mock implementations of runtime protocols like `EventBusProtocol` and `RuntimeProtocol` using `types.SimpleNamespace`. These mocks are useful for testing and can be passed to `build_runner`.

```python
from types import SimpleNamespace
from mysorf_flow.protocols import (
    RuntimeProtocol, EventBusProtocol, LoggerProtocol,
    TrackerProtocol, ArtifactManagerProtocol,
)

# isinstance checks work for duck-typed mocks
event_bus = SimpleNamespace(
    publish=lambda name, payload: None,
    subscribe=lambda name, handler: None,
)
print(isinstance(event_bus, EventBusProtocol))  # True

runtime = SimpleNamespace(
    event_bus=event_bus,
    logger=SimpleNamespace(info=print, warning=print, error=print, debug=print),
    tracker=SimpleNamespace(
        log_metrics=lambda metrics, step=None: None,
        log_params=lambda params: None,
    ),
    artifact_manager=SimpleNamespace(
        save=lambda name, src, artifact_type="generic": None,
        load=lambda name, artifact_type="generic": None,
    ),
)
print(isinstance(runtime, RuntimeProtocol))  # True

# Use with build_runner
runner = build_runner(flow_def, runtime=runtime)
```

--------------------------------

### Register a Plugin Class

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Register a Python class as a plugin. The class must have a `forward` method. Registration is scoped to a task pack origin when called within `task_pack_registration_context`.

```python
from mysorf_flow.api import register_plugin
from mysorf_flow.core import DataContract

class CNNRepresenter:
    PLUGIN_FAMILY = "representer"

    def forward(self, inputs, state, context):
        # inputs["pixels"] is available because declared_inputs=("pixels",)
        pixels = inputs["pixels"]
        embedding = [sum(row) / len(row) for row in pixels]  # mock encoding
        return {"repr.embedding": embedding}

register_plugin(
    "representer",
    "cnn_repr",
    CNNRepresenter,
    version="1.0.0",
    tags=("vision",),
    input_contracts={"pixels": DataContract(name="pixels", kind="tensor")},
    output_contracts={"repr.embedding": DataContract(name="repr.embedding", kind="tensor")},
)

# Verify registration
from mysorf_flow.api import list_plugin_types, get_plugin_metadata
print(list_plugin_types("representer"))      # ['cnn_repr']
meta = get_plugin_metadata("representer", "cnn_repr")
print(meta.version)                          # 1.0.0
print(meta.tags)                             # ('vision',)
```

--------------------------------

### Run Project Checks

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/CONTRIBUTING.md

Execute various checks including linting, type checking, testing, and full pre-commit pipeline.

```bash
make lint        # ruff check + format check
make typecheck   # mypy
make test        # pytest
make check       # all of the above
make smoke-wheel # build wheel + import from an isolated target dir
pre-commit run --all-files  # full pre-commit pipeline
```

--------------------------------

### Run flow with mysorf-base runtime

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Integrate mysorf-flow with mysorf-base by passing a RuntimeContext to the build_runner function.

```python
from mysorf_base.runtime import bootstrap
from mysorf_flow.api import build_runner

with bootstrap(["logging=rich"]) as ctx:
    # ctx is RuntimeContext, which satisfies RuntimeProtocol
    runner = build_runner(flow_def, runtime=ctx)
    result = runner.run(batch, mode="train")
```

--------------------------------

### Manage Plugin Families

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Register new plugin families beyond the built-ins or list existing families. Duplicate registration raises an `InvalidRegistrationError`.

```python
from mysorf_flow.api import register_plugin_family, list_plugin_families
from mysorf_flow.core import InvalidRegistrationError

# Built-in families are always present
print(list_plugin_families())
# ['binder', 'interactor', 'objective', 'predictor', 'representer']

# Register a custom family
register_plugin_family("tokenizer")
print(list_plugin_families())
# ['binder', 'interactor', 'objective', 'predictor', 'representer', 'tokenizer']

# Duplicate raises an error
try:
    register_plugin_family("tokenizer")
except InvalidRegistrationError as e:
    print(e)  # duplicate plugin family registration is not allowed
```

--------------------------------

### Toy Task Pack Plugin Implementations

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Define plugin classes like `ToyRepresenter` and `ToyPredictor` for a task pack. These classes specify the `PLUGIN_FAMILY` and implement the `forward` method for their respective functionalities.

```python
from mysorf_flow.api import register_task_pack, load_task_pack, build_flow_definition, build_runner
from mysorf_flow.taskpacks.schemas import TaskPackMetadata
from mysorf_flow.api import register_plugin
from mysorf_flow.flows import register_flow
from mysorf_flow.core import (
    CapabilityProfile, DiagnosticsPlan, FallbackPolicy,
    FlowDefinition, InputSpec, ModuleSpec, ObjectivePlan, StagePlan, StageSpec,
)

# --- Plugin implementations ---
class ToyRepresenter:
    PLUGIN_FAMILY = "representer"
    def forward(self, inputs, state, context):
        return {"repr.primary": [x * 2 for x in inputs["features"]]}

class ToyPredictor:
    PLUGIN_FAMILY = "predictor"
    def forward(self, inputs, state, context):
        return {"predict.logits": [sum(inputs["repr.primary")]}
```

--------------------------------

### Checkpointing and Restoring State

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Save and load the runner's state using `state_dict()` and `load_state_dict()`. Partial loading is supported, ignoring missing or extra keys in the checkpoint.

```python
state = runner.state_dict()
# state == {"encoder": {...}, "head": {...}}
torch.save(state, "checkpoint.pt")

# Restore from checkpoint (partial loading is safe — missing/extra keys are ignored)
checkpoint = torch.load("checkpoint.pt")
runner.load_state_dict(checkpoint)
```

--------------------------------

### Runner training lifecycle methods

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

The Runner supports a duck-typed training API compatible with PyTorch modules or any object implementing standard PyTorch training methods.

```APIDOC
## Runner training lifecycle methods

### Description
The `Runner` exposes a duck-typed training API that works with PyTorch modules or any object implementing `.train()` / `.eval()` / `.parameters()` / `.state_dict()` / `.load_state_dict()`.

### Usage
These methods are part of the standard PyTorch training lifecycle and can be called directly on objects that conform to this interface.

### Example
```python
import torch

# Assuming 'runner' is an instance of Runner and contains modules
# that implement the training lifecycle methods.

# Example of accessing parameters (if the runner or its components expose them)
# for param in runner.parameters():
#     print(param.shape)

# Example of saving/loading state dict (if applicable)
# state = runner.state_dict()
# runner.load_state_dict(state)
```
```

--------------------------------

### Runner Training Lifecycle Methods

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

The Runner exposes a duck-typed training API compatible with PyTorch modules or any object implementing standard training methods like .train(), .eval(), .parameters(), .state_dict(), and .load_state_dict().

```python
import torch
```

--------------------------------

### register_plugin_family / list_plugin_families

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Manages plugin families, allowing for the addition of new families beyond the built-in ones and listing all available families.

```APIDOC
## register_plugin_family / list_plugin_families

### Description
Manages plugin families. New plugin families beyond the five built-ins (`representer`, `binder`, `interactor`, `predictor`, `objective`) can be added at runtime. Duplicate registration raises `InvalidRegistrationError`.

### Parameters
- **family_name** (string) - Required - The name of the plugin family to register.

### Returns
- `list_plugin_families`: Returns a list of all registered plugin family names.

### Raises
- `InvalidRegistrationError`: If attempting to register a duplicate plugin family.

### Example
```python
from mysorf_flow.api import register_plugin_family, list_plugin_families
from mysorf_flow.core import InvalidRegistrationError

# List built-in families
print(list_plugin_families())

# Register a custom family
register_plugin_family("tokenizer")
print(list_plugin_families())

# Duplicate registration raises an error
try:
    register_plugin_family("tokenizer")
except InvalidRegistrationError as e:
    print(e)
```
```

--------------------------------

### Define a Toy Flow Definition

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Defines a basic 'toy_baseline' flow with 'represent' and 'predict' stages. This is useful for creating simple, end-to-end ML pipelines for testing or demonstration.

```python
def build_toy_flow(config) -> FlowDefinition:
    return FlowDefinition(
        flow_type="toy_baseline",
        task_type="toy_binary",
        version="0.1.0",
        description="Toy end-to-end flow.",
        capability_profile=CapabilityProfile(supports=("toy-example",)),
        input_spec=InputSpec(required_keys=("features",)),
        stage_plan=StagePlan(
            represent=StageSpec(modules=(
                ModuleSpec("repr", "representer", "toy_repr", None, ("features",), ("repr.primary",)),
            )),
            predict=StageSpec(modules=(
                ModuleSpec("pred", "predictor", "toy_predict", None, ("repr.primary",), ("predict.logits",)),
            )),
        ),
        objective_plan=ObjectivePlan(),
        diagnostics_plan=DiagnosticsPlan(),
        fallback_policy=FallbackPolicy(strategy="fail"),
    )
```

--------------------------------

### Build Executable Runner

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Constructs an executable Runner by running structural and semantic validators, resolving plugin classes, and instantiating them. Raises InvalidFlowError for missing methods or unused plugin configurations.

```python
from mysorf_flow.api import build_runner, register_plugin
from mysorf_flow.core import InvalidFlowError
from types import SimpleNamespace

# Register required plugins first
register_plugin("representer", "cnn_repr", CNNRepresenter)
register_plugin("predictor", "linear_head", LinearHead)

# Optional: runtime object satisfying RuntimeProtocol (or None/SimpleNamespace)
runtime = SimpleNamespace(
    event_bus=SimpleNamespace(publish=lambda name, payload: None, subscribe=lambda *a: None),
    logger=SimpleNamespace(info=print, warning=print, error=print, debug=print),
    tracker=SimpleNamespace(log_metrics=lambda *a, **kw: None, log_params=lambda *a: None),
    artifact_manager=SimpleNamespace(save=lambda *a, **kw: None, load=lambda *a, **kw: None),
)

try:
    runner = build_runner(
        flow_def,
        runtime=runtime,
        plugin_configs={"encoder_cfg": {"hidden_dim": 256}},
    )
    print(type(runner))  # <class 'mysorf_flow.core.runner.Runner'>
except InvalidFlowError as e:
    print(f"Runner construction failed: {e}")
```

--------------------------------

### Runner.run

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Executes all six stages of a flow in canonical order. It skips 'train_only' modules when mode is not 'train' and returns a detached snapshot of the execution state.

```APIDOC
## Runner.run

### Description
Executes a full staged pass through the flow. It runs all six stages in canonical order (`input → represent → bind → interact → predict → objective`), skipping `train_only` modules when `mode != "train"`. Returns a detached snapshot of the write-once execution state.

### Method
`runner.run(inputs, mode="infer")`

### Parameters
- **inputs** (dict) - A dictionary containing the input data for the flow.
- **mode** (str, optional) - The execution mode. Can be "infer", "train", or "evaluate". Defaults to "infer".

### Returns
A dictionary representing the execution state snapshot.

### Request Example
```python
# Inference pass — train_only modules are automatically skipped
result = runner.run({"pixels": [[128, 64, 255], [32, 16, 48]]}, mode="infer")
print(result.keys())          # dict_keys(['repr.embedding', 'predict.logits'])
print(result["predict.logits"])

# Training pass — all modules run including train_only ones
result_train = runner.run(
    {"pixels": [[128, 64, 255]], "labels": [1]},
    mode="train",
)
print("objective.loss" in result_train)  # True
```
```

--------------------------------

### Snapshot and Restore Registries for Test Isolation

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Use snapshot and restore helpers to isolate tests from global state changes in plugin, flow, and task-pack registries.

```python
from mysorf_flow.api import (
    snapshot_plugin_registry,
    restore_plugin_registry,
    snapshot_flow_registry,
    restore_flow_registry,
    snapshot_task_pack_registry,
    restore_task_pack_registry,
    register_plugin,
)

# Capture state before test
plugin_snap = snapshot_plugin_registry()
flow_snap = snapshot_flow_registry()
pack_snap = snapshot_task_pack_registry()

# Register something in the test
register_plugin("representer", "test_repr", type("TestRepr", (), {
    "forward": lambda self, inputs, state, ctx: {"repr.out": inputs.get("x")}
}))

# ... run test code ...

# Restore clean state after test
restore_plugin_registry(plugin_snap)
restore_flow_registry(flow_snap)
restore_task_pack_registry(pack_snap)
```

--------------------------------

### Registry Snapshot and Restore

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Provides helpers for safe test isolation by snapshotting and restoring the state of plugin, flow, and task-pack registries.

```APIDOC
## Registry Snapshot/Restore Helpers

### Description
These functions allow for safe test isolation by capturing the current state of registries and restoring them later, preventing global state leakage during tests.

### Functions
- `snapshot_plugin_registry()`: Captures the current state of the plugin registry.
- `restore_plugin_registry(snapshot)`: Restores the plugin registry to a previously captured state.
- `snapshot_flow_registry()`: Captures the current state of the flow registry.
- `restore_flow_registry(snapshot)`: Restores the flow registry to a previously captured state.
- `snapshot_task_pack_registry()`: Captures the current state of the task-pack registry.
- `restore_task_pack_registry(snapshot)`: Restores the task-pack registry to a previously captured state.

### Example Usage
```python
from mysorf_flow.api import (
    snapshot_plugin_registry,
    restore_plugin_registry,
    snapshot_flow_registry,
    restore_flow_registry,
    snapshot_task_pack_registry,
    restore_task_pack_registry,
    register_plugin,
)

# Capture state before test
plugin_snap = snapshot_plugin_registry()
flow_snap = snapshot_flow_registry()
pack_snap = snapshot_task_pack_registry()

# Register something in the test
register_plugin("representer", "test_repr", type("TestRepr", (), {
    "forward": lambda self, inputs, state, ctx: {"repr.out": inputs.get("x")}
}))

# ... run test code ...

# Restore clean state after test
restore_plugin_registry(plugin_snap)
restore_flow_registry(flow_snap)
restore_task_pack_registry(pack_snap)
```
```

--------------------------------

### build_runner

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Constructs an executable Runner by performing validation, resolving plugins, and instantiating them with optional configurations. It raises InvalidFlowError if plugin classes are missing methods or if plugin_configs contains unused keys.

```APIDOC
## build_runner

### Description
Constructs an executable `Runner`. This process includes running `validate_flow_definition`, all registered flow validators, resolving plugin classes, and instantiating them with optional per-module configurations. It raises `InvalidFlowError` if any plugin class is missing a `forward` method or if `plugin_configs` contains unused keys.

### Method
`build_runner(flow_def, runtime=None, plugin_configs=None)`

### Parameters
- **flow_def** (FlowDefinition) - The flow definition blueprint.
- **runtime** (object, optional) - A runtime object satisfying `RuntimeProtocol`, or `None`/`SimpleNamespace`. Defaults to `None`.
- **plugin_configs** (dict, optional) - A dictionary mapping plugin names to their configurations. Defaults to `None`.

### Exceptions
- `InvalidFlowError`: If any plugin class is missing a `forward` method or if `plugin_configs` contains unused keys.

### Request Example
```python
from mysorf_flow.api import build_runner, register_plugin
from mysorf_flow.core import InvalidFlowError
from types import SimpleNamespace

# Assume CNNRepresenter and LinearHead are defined elsewhere
# class CNNRepresenter: ...
# class LinearHead: ...

# Register required plugins first
register_plugin("representer", "cnn_repr", CNNRepresenter)
register_plugin("predictor", "linear_head", LinearHead)

# Optional: runtime object satisfying RuntimeProtocol (or None/SimpleNamespace)
runtime = SimpleNamespace(
    event_bus=SimpleNamespace(publish=lambda name, payload: None, subscribe=lambda *a: None),
    logger=SimpleNamespace(info=print, warning=print, error=print, debug=print),
    tracker=SimpleNamespace(log_metrics=lambda *a, **kw: None, log_params=lambda *a: None),
    artifact_manager=SimpleNamespace(save=lambda *a, **kw: None, load=lambda *a, **kw: None),
)

try:
    runner = build_runner(
        flow_def,
        runtime=runtime,
        plugin_configs={"encoder_cfg": {"hidden_dim": 256}},
    )
    print(type(runner))  # <class 'mysorf_flow.core.runner.Runner'>
except InvalidFlowError as e:
    print(f"Runner construction failed: {e}")
```
```

--------------------------------

### Register Toy Plugins and Flow

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Registers the 'toy_repr' representer and 'toy_predict' predictor plugins, along with the 'toy_baseline' flow builder. This function is typically called during task pack registration.

```python
def register_toy_pack() -> None:
    register_plugin("representer", "toy_repr", ToyRepresenter)
    register_plugin("predictor", "toy_predict", ToyPredictor)
    register_flow("toy_baseline", build_toy_flow)
```

--------------------------------

### Define Data Contracts and Plugin Metadata

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Use DataContract to describe data values and PluginMetadata to package plugin information. This is useful for defining the inputs and outputs of your plugins.

```python
from mysorf_flow.core import DataContract, PluginMetadata

pixels_contract = DataContract(
    name="pixels",
    kind="tensor",
    dtype="float32",
    rank=4,
    shape=("N", "C", "H", "W"),
    semantics="normalized image batch",
    optional=False,
)

embedding_contract = DataContract(
    name="repr.embedding",
    kind="tensor",
    dtype="float32",
    rank=2,
    shape=("N", "D"),
)

meta = PluginMetadata(
    plugin_id="my_pack:representer:cnn_repr",
    family="representer",
    type_name="cnn_repr",
    version="1.0.0",
    input_contracts={"pixels": pixels_contract},
    output_contracts={"repr.embedding": embedding_contract},
    tags=("vision", "cnn"),
    task_pack_origin="my_pack",
)

print(meta.plugin_id)  # my_pack:representer:cnn_repr
print(meta.tags)       # ('vision', 'cnn')
```

--------------------------------

### Add mysorf-flow to pyproject.toml

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

Specify mysorf-flow as a project dependency in your pyproject.toml file.

```toml
[project]
dependencies = [
    "mysorf-flow @ git+https://github.com/mysorf-9239/mysorf-flow.git@v0.1.0",
]
```

--------------------------------

### Collect Trainable Parameters

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Collect trainable parameters using `collect_parameters()`, with the option to exclude specific module IDs. This is useful for setting up optimizers.

```python
params = runner.collect_parameters(exclude_ids={"encoder"})
optimizer = torch.optim.Adam(params, lr=1e-3)
```

--------------------------------

### register_plugin

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Registers a Python class as a plugin under a specified family and type name. It allows for versioning, tagging, and defining input/output contracts.

```APIDOC
## register_plugin

### Description
Registers a Python class as a plugin under a `family` (e.g., `"representer"`, `"predictor"`) and `type_name`. The class must define a `forward(inputs, state, context)` method and optionally declare `PLUGIN_FAMILY`. Registration is scoped to a `task_pack_origin` when called inside `task_pack_registration_context`.

### Parameters
- **family** (string) - Required - The family name for the plugin.
- **type_name** (string) - Required - The specific type name of the plugin.
- **plugin_class** (class) - Required - The Python class to register.
- **version** (string) - Optional - The version of the plugin.
- **tags** (tuple of strings) - Optional - Tags associated with the plugin.
- **input_contracts** (dict) - Optional - A dictionary mapping input names to `DataContract` objects.
- **output_contracts** (dict) - Optional - A dictionary mapping output names to `DataContract` objects.

### Example
```python
from mysorf_flow.api import register_plugin
from mysorf_flow.core import DataContract

class CNNRepresenter:
    PLUGIN_FAMILY = "representer"

    def forward(self, inputs, state, context):
        pixels = inputs["pixels"]
        embedding = [sum(row) / len(row) for row in pixels]  # mock encoding
        return {"repr.embedding": embedding}

register_plugin(
    "representer",
    "cnn_repr",
    CNNRepresenter,
    version="1.0.0",
    tags=("vision",),
    input_contracts={"pixels": DataContract(name="pixels", kind="tensor")},
    output_contracts={"repr.embedding": DataContract(name="repr.embedding", kind="tensor")},
)
```
```

--------------------------------

### Execute Runner Stages

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Executes all six stages of a flow in canonical order, skipping train_only modules when mode is not 'train'. Returns a detached snapshot of the execution state.

```python
# Inference pass — train_only modules are automatically skipped
result = runner.run({"pixels": [[128, 64, 255], [32, 16, 48]]}, mode="infer")
print(result.keys())          # dict_keys(['repr.embedding', 'predict.logits'])
print(result["predict.logits"])

# Training pass — all modules run including train_only ones
result_train = runner.run(
    {"pixels": [[128, 64, 255]], "labels": [1]},
    mode="train",
)
print("objective.loss" in result_train)  # True

# Predict shortcut — filters state to predict.* / prediction.* keys only
preds = runner.predict({"pixels": [[0, 128, 255]]})
print(list(preds.keys()))  # ['predict.logits']

# Evaluate shortcut — runs in mode="evaluate"
eval_out = runner.evaluate({"pixels": [[0, 128, 255]], "labels": [0]})
```

--------------------------------

### Citation Metadata (BibTeX)

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/README.md

BibTeX formatted citation for the mysorf-flow software, suitable for academic publications.

```bibtex
@software{mysorfflow2026,
    author = {Nguyen, Duc Danh},
    title = {mysorf-flow: Generic Flow-Defined Research Engine},
    year = {2026},
    version = {0.1.0},
    url = {https://github.com/mysorf-9239/mysorf-flow},
    license = {MIT}
}
```

--------------------------------

### Set Training/Evaluation Mode

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Use `set_train()` and `set_eval()` to switch the runner's mode. These methods call `.train()` or `.eval()` on all bound module instances.

```python
runner.set_train()   # calls .train() on all bound module instances
runner.set_eval()    # calls .eval() on all bound module instances
```

--------------------------------

### Define a FlowDefinition Blueprint

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Use FlowDefinition to declare a complete ML pipeline blueprint, including stages, inputs, objectives, and fallback policies. Access modules in stage order after definition.

```python
from mysorf_flow.core import (
    CapabilityProfile,
    DiagnosticsPlan,
    FallbackPolicy,
    FlowDefinition,
    InputSpec,
    ModuleSpec,
    ObjectivePlan,
    StagePlan,
    StageSpec,
)

flow_def = FlowDefinition(
    flow_type="image_classify",
    task_type="binary_classification",
    version="1.0.0",
    description="Binary image classifier pipeline.",
    capability_profile=CapabilityProfile(
        supports=("image-input",),
        output_style="scalar_logit",
    ),
    input_spec=InputSpec(
        required_keys=("pixels",),
        optional_keys=("labels",),
    ),
    stage_plan=StagePlan(
        represent=StageSpec(
            modules=(
                ModuleSpec(
                    module_id="encoder",
                    family="representer",
                    type_name="cnn_repr",
                    config_ref=None,
                    declared_inputs=("pixels",),
                    declared_outputs=("repr.embedding",),
                ),
            )
        ),
        predict=StageSpec(
            modules=(
                ModuleSpec(
                    module_id="head",
                    family="predictor",
                    type_name="linear_head",
                    config_ref=None,
                    declared_inputs=("repr.embedding",),
                    declared_outputs=("predict.logits",),
                ),
            )
        ),
        objective=StageSpec(
            modules=(
                ModuleSpec(
                    module_id="loss",
                    family="objective",
                    type_name="bce_loss",
                    config_ref=None,
                    declared_inputs=("predict.logits", "labels"),
                    declared_outputs=("objective.loss",),
                    train_only=True,
                ),
            )
        ),
    ),
    objective_plan=ObjectivePlan(
        module_ids=("loss",),
        required_keys=("predict.logits", "labels",),
    ),
    diagnostics_plan=DiagnosticsPlan(),
    fallback_policy=FallbackPolicy(strategy="fail"),
)

# Access all modules in stage order
for m in flow_def.modules():
    print(m.module_id, m.family, m.type_name)
# encoder representer cnn_repr
# head predictor linear_head
# loss objective bce_loss

```

--------------------------------

### Build and Validate Flow Definition from Task Pack

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Use `build_flow_definition` with a loaded `TaskPackHandle` to combine flow resolution, task-type validation, and structural validation in one call.

```python
from mysorf_flow.api import load_task_pack, build_flow_definition
from mysorf_flow.core import InvalidFlowError

handle = load_task_pack("vision-pack")

try:
    flow_def = build_flow_definition(
        task_pack=handle,
        flow_type="cnn_baseline",
        config={"hidden_dim": 256},
    )
    print(flow_def.flow_type)         # cnn_baseline
    print(flow_def.task_pack_origin)  # vision-pack
except InvalidFlowError as e:
    print(f"Flow invalid: {e}")
```

--------------------------------

### Commit Message Format

Source: https://github.com/mysorf-9239/mysorf-flow/blob/main/CONTRIBUTING.md

Defines the standard format for commit messages, including type and summary.

```text
<type>: <short summary>

<optional body>
```

--------------------------------

### Register a Flow Builder Callable

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Register a flow builder callable that accepts a config object and returns a FlowDefinition. Registration is scoped to a task-pack origin when called within its context.

```python
from mysorf_flow.flows import register_flow
from mysorf_flow.core import (
    CapabilityProfile, DiagnosticsPlan, FallbackPolicy,
    FlowDefinition, InputSpec, ModuleSpec,
    ObjectivePlan, StagePlan, StageSpec,
)

def build_my_flow(config) -> FlowDefinition:
    return FlowDefinition(
        flow_type="my_flow",
        task_type="binary_classification",
        version="1.0.0",
        description="Custom binary classification flow.",
        capability_profile=CapabilityProfile(),
        input_spec=InputSpec(required_keys=("features",), optional_keys=("labels",)),
        stage_plan=StagePlan(
            represent=StageSpec(modules=(
                ModuleSpec("repr", "representer", "cnn_repr", None, ("features",), ("repr.out",)),
            )),
            predict=StageSpec(modules=(
                ModuleSpec("pred", "predictor", "linear_head", None, ("repr.out",), ("predict.logits",)),
            )),
        ),
        objective_plan=ObjectivePlan(),
        diagnostics_plan=DiagnosticsPlan(),
        fallback_policy=FallbackPolicy(strategy="fail"),
    )

register_flow("my_flow", build_my_flow)

from mysorf_flow.flows.registry import list_flow_types
print(list_flow_types())  # ['my_flow']
```

--------------------------------

### Runner.evaluate

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

A shortcut method to execute the flow in evaluation mode. This runs the flow with mode set to 'evaluate'.

```APIDOC
## Runner.evaluate

### Description
Executes the flow in evaluation mode. This is a shortcut for running the flow with `mode="evaluate"`.

### Method
`runner.evaluate(inputs)`

### Parameters
- **inputs** (dict) - A dictionary containing the input data for the flow.

### Returns
A dictionary representing the execution state snapshot when run in evaluation mode.

### Request Example
```python
eval_out = runner.evaluate({"pixels": [[0, 128, 255]], "labels": [0]})
```
```

--------------------------------

### build_flow_definition

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Builds and validates a flow definition from a loaded task pack, combining flow resolution, task-type validation, and structural validation.

```APIDOC
## Task Pack API (Continued)

### `build_flow_definition` — build and validate a flow from a task pack

#### Description
Combines flow resolution, task-type validation, and structural validation into a single call that requires a loaded `TaskPackHandle`.

#### Parameters
- `task_pack` (TaskPackHandle): The handle of the loaded task pack.
- `flow_type` (string): The type of the flow to build.
- `config` (dict): Configuration parameters for the flow.

#### Example Usage
```python
from mysorf_flow.api import load_task_pack, build_flow_definition
from mysorf_flow.core import InvalidFlowError

handle = load_task_pack("vision-pack") # Assuming 'vision-pack' is loaded

try:
    flow_def = build_flow_definition(
        task_pack=handle,
        flow_type="cnn_baseline",
        config={"hidden_dim": 256},
    )
    print(flow_def.flow_type)         # cnn_baseline
    print(flow_def.task_pack_origin)  # vision-pack
except InvalidFlowError as e:
    print(f"Flow invalid: {e}")
```
```

--------------------------------

### Resolve Registered Plugin

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Look up a registered plugin class or its full registration record by family and type name. Scoped resolution avoids ambiguity when multiple task packs register the same type name.

```python
from mysorf_flow.api import resolve_plugin, resolve_plugin_registration, get_plugin_metadata
from mysorf_flow.core import UnknownPluginError

# Resolve just the class
cls = resolve_plugin("representer", "cnn_repr")
instance = cls()
print(cls.__name__)  # CNNRepresenter

# Resolve the full registration (includes metadata)
reg = resolve_plugin_registration("representer", "cnn_repr")
print(reg.metadata.plugin_id)     # global:representer:cnn_repr
print(reg.metadata.task_pack_origin)  # None (global)

# Scoped lookup (avoids cross-task-pack ambiguity)
try:
    reg_scoped = resolve_plugin_registration(
        "representer", "cnn_repr", task_pack_origin="my_pack"
    )
except UnknownPluginError as e:
    print(e)  # if not registered under that task pack

# Error handling
try:
    resolve_plugin("representer", "nonexistent")
except UnknownPluginError as e:
    print(e)
    # Unknown plugin type 'nonexistent' for family 'representer'. Available types: ['cnn_repr'].
```

--------------------------------

### register_flow

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Registers a flow builder callable with the flow registry. Registration is scoped to a task-pack origin when called within a task-pack context.

```APIDOC
## Flow Registry API

### `register_flow` — register a flow builder callable

#### Description
A flow builder is any callable that takes a config object and returns a validated `FlowDefinition`. Registration is scoped to a task-pack origin when called inside the task-pack context.

#### Parameters
- `name` (string): The name to register the flow under.
- `builder` (callable): The function that builds the `FlowDefinition`.

#### Example Usage
```python
from mysorf_flow.flows import register_flow
from mysorf_flow.core import (
    CapabilityProfile, DiagnosticsPlan, FallbackPolicy,
    FlowDefinition, InputSpec, ModuleSpec,
    ObjectivePlan, StagePlan, StageSpec,
)

def build_my_flow(config) -> FlowDefinition:
    return FlowDefinition(
        flow_type="my_flow",
        task_type="binary_classification",
        version="1.0.0",
        description="Custom binary classification flow.",
        capability_profile=CapabilityProfile(),
        input_spec=InputSpec(required_keys=("features",), optional_keys=("labels",)),
        stage_plan=StagePlan(
            represent=StageSpec(modules=(
                ModuleSpec("repr", "representer", "cnn_repr", None, ("features",), ("repr.out",)),
            )),
            predict=StageSpec(modules=(
                ModuleSpec("pred", "predictor", "linear_head", None, ("repr.out",), ("predict.logits",)),
            )),
        ),
        objective_plan=ObjectivePlan(),
        diagnostics_plan=DiagnosticsPlan(),
        fallback_policy=FallbackPolicy(strategy="fail"),
    )

register_flow("my_flow", build_my_flow)

from mysorf_flow.flows.registry import list_flow_types
print(list_flow_types())  # ['my_flow']
```
```

--------------------------------

### resolve_plugin / resolve_plugin_registration

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Looks up a registered plugin class or its full registration record by family and type name. Scoped resolution avoids ambiguity when multiple task packs register the same type name.

```APIDOC
## resolve_plugin / resolve_plugin_registration

### Description
Resolves a plugin class or its full `RegisteredPlugin` record by family and type name. Scoped resolution (with `task_pack_origin`) avoids ambiguity when multiple task packs register the same type name.

### Parameters
- **family** (string) - Required - The family name of the plugin.
- **type_name** (string) - Required - The type name of the plugin.
- **task_pack_origin** (string) - Optional - The origin of the task pack for scoped resolution.

### Returns
- `resolve_plugin`: Returns the plugin class.
- `resolve_plugin_registration`: Returns the full `RegisteredPlugin` record.

### Raises
- `UnknownPluginError`: If the plugin is not found.

### Example
```python
from mysorf_flow.api import resolve_plugin, resolve_plugin_registration
from mysorf_flow.core import UnknownPluginError

# Resolve just the class
cls = resolve_plugin("representer", "cnn_repr")
instance = cls()

# Resolve the full registration (includes metadata)
reg = resolve_plugin_registration("representer", "cnn_repr")

# Scoped lookup
try:
    reg_scoped = resolve_plugin_registration(
        "representer", "cnn_repr", task_pack_origin="my_pack"
    )
except UnknownPluginError as e:
    print(e)

# Error handling
try:
    resolve_plugin("representer", "nonexistent")
except UnknownPluginError as e:
    print(e)
```
```

--------------------------------

### Register Custom Flow Validators

Source: https://context7.com/mysorf-9239/mysorf-flow/llms.txt

Extends engine validation with custom semantic rules by registering FlowValidatorProtocol instances. These run after structural validators during build_runner.

```python
from mysorf_flow.api import register_flow_validator, run_registered_validators
from mysorf_flow.protocols import FlowValidatorProtocol
from mysorf_flow.core import InvalidFlowError

class RequiresLabelsValidator:
    """Enforce that flows declare 'labels' as an optional or required input."""
    def validate(self, flow_def) -> None:
        all_keys = (
            set(flow_def.input_spec.required_keys)
            | set(flow_def.input_spec.optional_keys)
        )
        if "labels" not in all_keys:
            raise InvalidFlowError(
                flow_type=flow_def.flow_type,
                reason="flow must declare 'labels' as an input key",
            )

register_flow_validator(RequiresLabelsValidator())

# Manually run all validators (build_runner does this automatically)
try:
    run_registered_validators(flow_def)
except InvalidFlowError as e:
    print(e)
```