Skip to content

Effect Interfaces

Page Maps

graph LR
  family["Python Programming"]
  program["Python Functional Programming"]
  section["Effect Boundaries Resource Safety"]
  page["Effect Interfaces"]
  capstone["Capstone evidence"]

  family --> program --> section --> page
  page -.applies in.-> capstone
flowchart LR
  orient["Orient on the page map"] --> read["Read the main claim and examples"]
  read --> inspect["Inspect the related code, proof, or capstone surface"]
  inspect --> verify["Run or review the verification path"]
  verify --> apply["Apply the idea back to the module and capstone"]

Read the first diagram as a placement map: this page is one concept inside its parent module, not a detached essay, and the capstone is the pressure test for whether the idea holds. Read the second diagram as the working rhythm for the page: name the problem, study the example, identify the boundary, then carry one review question forward.

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.

If you are following the main track, read this page in two passes:

  1. read the native IOPlan sections (Core question, Public API, Minimal Example, and Property-Based Proofs)
  2. treat the external-library comparison as optional context, not as a second required implementation path

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 with a minimal native ADT, and how does that compare with popular library approaches while staying 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: In this capstone, IOPlan is the default path. It stays small, composes cleanly with the earlier cores, and covers the boundary pressures this repository is designed to teach without adding a library dependency.

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 (capstone/src/funcpipe_rag/fp/effects/io_plan.py)

# capstone/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: capstone/src/funcpipe_rag/rag/rag_api.py and capstone/src/funcpipe_rag/rag/core.py
  • Capability protocols (ports): capstone/src/funcpipe_rag/domain/capabilities.py
  • IOPlan: capstone/src/funcpipe_rag/fp/effects/io_plan.py

Shell usage unchanged:

Property/law checks for IOPlan live in capstone/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.

4.4 Mapping to External Libraries (sidebar)

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

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

9. Post-Core Exercise

  1. Implement the full RAG rag_plan using IOPlan + Core 1 ports.
  2. 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.
  3. Add Hypothesis monad law tests for your own pipeline.

Continue with: Capability Protocols

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.