M07C02: Effect Interfaces – Native Deferred IO vs Library Alternatives (`returns`, `effect`)¶

Module 07 – Optional Comparison Core

Main track: Cores 1, 3–10 (Ports & Adapters + Capability Protocols → Production).
This Core 2 is optional. It is a standalone comparison only.
You do not need any external effect library to implement anything in this series.
The canonical FuncPipe approach is the native IOPlan shown first – a tiny, zero-dependency, thunk-based deferred IO that composes perfectly with Result, Reader, Writer, and the ports from Core 1.
The returns and effect references are pure sidebars for ecosystem context.

Progression Note¶

Module 7 takes the lawful containers and pipelines from Module 6 and puts all effects behind explicit boundaries.

Module	Focus	Key Outcomes
6	Monadic Flows as Composable Pipelines	Lawful `and_then`, Reader/State patterns, error-typed flows
7	Effect Boundaries & Resource Safety	Ports & adapters, capability protocols, resource-safe IO, idempotency
8	Async / Concurrent Pipelines	Backpressure, timeouts, resumability, fairness (built on 6–7)

Core question
How do you represent effectful operations as pure, composable data – using either a minimal native ADT or popular libraries – while staying 100 % compatible with the ports & adapters architecture from Core 1?

What you now have after M07C01 + this core - Pure domain core (Module 5–6)
- Zero direct I/O anywhere in domain code
- All effects hidden behind ports (M07C01)
- Ports swappable between real and in-memory adapters with Hypothesis-checked equivalence
- A tiny, law-checked IOPlan that lets you describe any effectful computation as pure data and defer it to the shell

What the rest of Module 7 adds - Capability protocols (Clock, RNG, Logger as pure data)
- Resource-safe adapters with guaranteed cleanup
- Idempotent effect design and transaction patterns
- Incremental migration playbook
- Production story: CI, golden tests, shadow traffic

You are one small step away from a complete production-grade functional architecture.

1. Laws & Invariants (machine-checked in CI)¶

Law / Invariant	Description	Enforcement
No-Eagerness	Constructing an effect description performs zero side effects.	Mock tests
Monad Left Identity	`perform(io_bind(io_pure(v), f)) == perform(f(v))`	Hypothesis
Monad Right Identity	`perform(io_bind(m, io_pure)) == perform(m)`	Hypothesis
Monad Associativity	`perform(io_bind(io_bind(m, f), g)) == perform(io_bind(m, λx → io_bind(f(x), g)))`	Hypothesis
Isolation	Core produces only immutable descriptions; all effects happen exclusively in performers/adapters.	mypy --strict + review
Adapter Equivalence	Real vs in-memory implementations of the same port, given the same logical inputs, produce identical `Result` values.	Hypothesis (swappability)
Resource Safety (pattern)	When combined with Core 1 adapters and iterator discipline, cleanup is guaranteed even on partial consumption.	Integration tests

We standardise on one default interpreter (perform) for the main track.
Advanced shells or tests may define alternative interpreters (e.g. log-only), but all laws are stated and checked against perform.
The real power is adapter swappability, not interpreter proliferation.

2. Decision Table – Which Effect Representation?¶

Need	Complexity	Async?	Recommended Choice	Trade-offs
Continuity with FuncPipe stack	Zero dep	No	Native `IOPlan` (canonical)	Tiny, no HKT, perfect Reader/port integration
Full monadic IO + HKT typing	Medium	No	`returns`	Richer typing, heavier dependency
Intent-based effects	Medium	No	`effect` / `effects`	Dispatcher style, rarely needed in new code
Just thunks	Lowest	No	Stdlib `Callable`	No named type, easy to misuse

Verdict: Use IOPlan. It is <60 LOC, composes cleanly with Core 1, and gives you everything you need for production.

Global error policy: We fix E = ErrInfo everywhere. This is deliberate – it simplifies interop between Result, Reader, ports, and IOPlan. Per-subsystem error types are possible by parameterising IOPlan[A, E], but we don’t need that complexity here.

3. Public API – `IOPlan` (`src/funcpipe_rag/fp/effects/io_plan.py`)¶

# src/funcpipe_rag/fp/effects/io_plan.py – mypy --strict clean, zero external deps
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable, Generic, TypeVar

from funcpipe_rag.result.types import Err, ErrInfo, Ok, Result

A = TypeVar("A")
B = TypeVar("B")

__all__ = ["IOPlan", "io_pure", "io_delay", "io_bind", "io_map", "perform"]

@dataclass(frozen=True)
class IOPlan(Generic[A]):
    """Pure description of a (possibly effectful) computation returning Result[A, ErrInfo]."""
    thunk: Callable[[], Result[A, ErrInfo]]

def io_pure(value: A) -> IOPlan[A]:
    return IOPlan(lambda: Ok(value))

def io_delay(thunk: Callable[[], Result[A, ErrInfo]]) -> IOPlan[A]:
    """Lift an effectful thunk into a pure description."""
    return IOPlan(thunk)

def io_bind(plan: IOPlan[A], f: Callable[[A], IOPlan[B]]) -> IOPlan[B]:
    def thunk() -> Result[B, ErrInfo]:
        res = plan.thunk()                     # single evaluation – critical
        return res if isinstance(res, Err) else f(res.value).thunk()
    return IOPlan(thunk)

def io_map(plan: IOPlan[A], f: Callable[[A], B]) -> IOPlan[B]:
    return io_bind(plan, lambda x: io_pure(f(x)))

def perform(plan: IOPlan[A]) -> Result[A, ErrInfo]:
    """The default interpreter – runs the described effects."""
    return plan.thunk()

Guideline: keep perform in shells/boundaries; domain code should return IOPlan values.

4. Reference Implementations¶

4.1 Minimal Example – Copy via Port¶

# Illustrative example (not a repo file): describing time+logging as IOPlan.
from datetime import datetime

from funcpipe_rag.domain.capabilities import Clock, Logger
from funcpipe_rag.domain.logging import LogEntry
from funcpipe_rag.domain.effects.io_plan import IOPlan, io_bind, io_delay
from funcpipe_rag.result.types import ErrInfo, Ok, Result

def _log(logger: Logger, now: datetime) -> Result[None, ErrInfo]:
    logger.log(LogEntry("INFO", f"now={now.isoformat()}"))
    return Ok(None)

def log_time_plan(clock: Clock, logger: Logger) -> IOPlan[None]:
    return io_bind(
        io_delay(lambda: Ok(clock.now())),
        lambda now: io_delay(lambda: _log(logger, now)),
    )

Shell – the only place effects happen:

# Shell boundary – this is the only place we call `perform`
from funcpipe_rag.domain.effects.io_plan import perform
from funcpipe_rag.infra.adapters.clock import SystemClock
from funcpipe_rag.infra.adapters.logger import ConsoleLogger

plan = log_time_plan(SystemClock(), ConsoleLogger())
result = perform(plan)

4.2 Full RAG Integration¶

This repo’s end-of-Module-07 codebase provides the primitives (capability protocols, adapters, IOPlan, retry/tx wrappers), but does not fully migrate the RAG surface to IOPlan yet.

Current RAG pipeline entry points: src/funcpipe_rag/rag/rag_api.py and src/funcpipe_rag/rag/core.py
Capability protocols (ports): src/funcpipe_rag/domain/capabilities.py
IOPlan: src/funcpipe_rag/fp/effects/io_plan.py

Shell usage unchanged:

Property/law checks for IOPlan live in tests/unit/domain/test_io_plan_laws.py.

4.3 Stdlib-Only Thunk Variant (zero custom types)¶

from typing import Callable, TypeVar

A = TypeVar("A")
B = TypeVar("B")

IOThunk = Callable[[], Result[A, ErrInfo]]

def io_pure_thunk(v: A) -> IOThunk[A]:
    return lambda: Ok(v)

def io_bind_thunk(plan: IOThunk[A], f: Callable[[A], IOThunk[B]]) -> IOThunk[B]:
    def thunk() -> Result[B, ErrInfo]:
        res = plan()                       # single evaluation
        return res if isinstance(res, Err) else f(res.value)()
    return thunk

def perform_thunk(plan: IOThunk[A]) -> Result[A, ErrInfo]:
    return plan()

Same semantics, no dataclass – but weaker typing and no clear “effect type” in signatures.

We do not depend on these libraries anywhere in this series.

FuncPipe	`returns`	`effect` / `effects`
`Result`	`Result[A, E]` / `IOResult`	N/A (exceptions)
`IOPlan[A]`	`IO[A]` / `IOResult[A,E]`	`Effect[A]`
`io_bind`	`.bind`	Dispatcher folding
`perform`	`.unsafe_perform_io()`	`sync_perform`

Interop is trivial at boundaries.

5. Property-Based Proofs¶

We phrase the monad laws via perform because perform(IOPlan[A]) is the semantics of the plan – equality of perform results is equality of meaning.

@given(v=st.integers())
def test_monad_identity(v):
    p = io_pure(v)
    assert perform(io_bind(p, io_pure)) == perform(p)
    f = lambda x: io_pure(x * 7)
    assert perform(io_bind(io_pure(v), f)) == perform(f(v))

@given(v=st.integers(), k=st.integers(-10,10), m=st.integers(-10,10))
def test_monad_associativity(v, k, m):
    f = lambda x: io_pure(x + k)
    g = lambda x: io_pure(x * m)
    left  = perform(io_bind(io_bind(io_pure(v), f), g))
    right = perform(io_bind(io_pure(v), lambda x: io_bind(f(x), g)))
    assert left == right

@given(err_msg=st.text())
def test_io_bind_propagates_err(err_msg):
    bad = io_delay(lambda: Err(ErrInfo(code="TEST", msg=err_msg)))
    f   = lambda x: io_pure(x * 2)
    assert perform(io_bind(bad, f)) == perform(bad)

@given(src_path=st.text())
def test_no_eagerness(src_path):
    with mock.patch("builtins.open") as m:
        plan = copy_file_plan(src_path, "dst.txt", FileStorageAdapter())
        m.assert_not_called()
        perform(plan)
        m.assert_called()

6. Big-O & Allocation Guarantees¶

Operation	Time	Heap	Notes
`io_pure`	O(1)	O(1)	One closure + IOPlan
`io_delay`	O(1)	O(1)	Wraps existing thunk
`io_bind`	O(1)	O(1)	One additional closure; single inner evaluation
`io_map`	O(1)	O(1)	Via `io_bind`
`perform`	O(chain)	O(chain)	One stack frame per nested bind

No extra overhead beyond what any monadic IO style incurs in Python.

7. Anti-Patterns & Immediate Fixes¶

Anti-Pattern	Symptom	Fix
Performing in core	Untestable, eager effects	Return `IOPlan`, perform in shell
Raw exceptions in thunks	Crashes escape	Always return `Result` via ports
Multi-evaluation of plan	Effects run 2+N times	Single `plan.thunk()` call only
God “effect” object	Monolithic adapter	Keep ports narrow (Core 1)

8. Pre-Core Quiz¶

Effect description must be…? → Pure data, zero side effects on construction
Where do effects run? → Only in perform (shell)
Native effect type? → IOPlan with io_ combinators
Monad laws tested on…? → The interpreter (perform)
Real power comes from…? → Swapping adapters, not interpreters

9. Post-Core Exercise¶

Implement the full RAG rag_plan using IOPlan + Core 1 ports.
In a test or shell-only context, write a “log-only” interpreter variant that records intended operations but never touches the filesystem; prove via mock_open or tmpdir inspection that no real writes occur.
Add Hypothesis monad law tests for your own pipeline.

Next → M07C03: Capability Protocols (Clock, RNG, Logger as Pure Data)

You now have a minimal, law-checked effect ADT that slots perfectly into the Core 1 architecture: all logic stays pure, all effects are described as data, and the shell remains a thin, swappable layer around perform. The rest of Module 7 is just specialisations of this pattern.

M07C02: Effect Interfaces – Native Deferred IO vs Library Alternatives (returns, effect)¶