Combining Effects and Coeffects via Grading

# Graded Hoare triple: {P} c : e {Q}
# The grade e tracks side effects.

# Example: grades = number of writes to a log
def logged_double(x):
    print(f"  [LOG] doubling {x}")  # 1 write
    return x * 2

def logged_add1(x):
    print(f"  [LOG] adding 1 to {x}")  # 1 write
    return x + 1

# Composed: grade = 1 + 1 = 2 writes
def pipeline(x):
    return logged_add1(logged_double(x))

print("Pipeline grade: 2 (two log writes)")
result = pipeline(5)
print(f"  result = {result}")
print()
print("Grades compose: 1 write + 1 write = 2 writes.")
print("The type system tracks this statically.")

# Different grade monoids for different effects

# (ℕ, +, 0) — count operations
count_pure = 0
count_compose = lambda e1, e2: e1 + e2
print(f"Count: 3 ops + 2 ops = {count_compose(3, 2)} ops")

# (ℝ≥0, ×, 1) — noise multiplication
noise_pure = 1.0
noise_compose = lambda e1, e2: e1 * e2
print(f"Noise: 0.9 × 0.8 = {noise_compose(0.9, 0.8)} fidelity")

# ({R,W,RW}, max, ∅) — read/write permissions
perms = {"": 0, "R": 1, "W": 2, "RW": 3}
perm_names = {0: "∅", 1: "R", 2: "W", 3: "RW"}
perm_compose = lambda e1, e2: max(e1, e2)
print(f"Perms: R + W = {perm_names[perm_compose(perms['R'], perms['W'])]}")
print()
print("Each monoid is a different grading.")
print("Composition rule changes, structure stays.")

# Graded COMP: grades compose through the handshake
# {P} c1 : e1 {R}  and  {R} c2 : e2 {Q}
#   => {P} c1;c2 : e1*e2 {Q}

def graded_comp(c1, e1, c2, e2, grade_op):
    """Compose graded computations."""
    def composed(x):
        mid = c1(x)
        out = c2(mid)
        return out
    return composed, grade_op(e1, e2)

# Stage 1: filter (0.95 fidelity)
def stage1(x): return [v for v in x if v > 0]

# Stage 2: normalize (0.90 fidelity)
def stage2(x):
    total = sum(x) if x else 1
    return [v/total for v in x]

pipeline, total_fidelity = graded_comp(
    stage1, 0.95,
    stage2, 0.90,
    lambda e1, e2: round(e1 * e2, 4)
)

data = [-1, 3, -2, 5, 1]
print(f"Input: {data}")
print(f"Output: {pipeline(data)}")
print(f"Fidelity: 0.95 × 0.90 = {total_fidelity}")
print()
print("The grade tracks cumulative effect.")
print("No stage inspects another's grade —")
print("composition handles the accounting.")

# Graded Freyd category in miniature
# Pure values embed with grade "none"
# Effectful computations carry their grade

class GradedComp:
    def __init__(self, fn, grade):
        self.fn = fn
        self.grade = grade

    def then(self, other, grade_op):
        """Sequential composition — grades compose."""
        def composed(x):
            return other.fn(self.fn(x))
        return GradedComp(composed, grade_op(self.grade, other.grade))

# Pure embedding: grade = 0 (additive identity)
pure_inc = GradedComp(lambda x: x + 1, 0)

# Effectful: grade = 1 (one side effect)
effectful_log = GradedComp(lambda x: x * 2, 1)

# Compose
pipeline = pure_inc.then(effectful_log, lambda a, b: a + b)
print(f"pure(+1) then log(*2): grade = {pipeline.grade}")
print(f"pipeline(5) = {pipeline.fn(5)}")

# Pure;Pure stays pure
pure2 = pure_inc.then(GradedComp(lambda x: x * 3, 0), lambda a, b: a + b)
print(f"pure;pure: grade = {pure2.grade} (still pure)")

# Effects (what you produce) + Coeffects (what you consume)
# Unified in one grade: (reads, writes)

def graded_fn(fn, reads, writes):
    return {"fn": fn, "reads": reads, "writes": writes}

def compose_graded(g1, g2):
    return graded_fn(
        lambda x: g2["fn"](g1["fn"](x)),
        g1["reads"] + g2["reads"],    # coeffects add
        g1["writes"] + g2["writes"]   # effects add
    )

# Stage 1 reads config (1 read, 0 writes)
s1 = graded_fn(lambda x: x.upper(), 1, 0)

# Stage 2 logs result (0 reads, 1 write)
s2 = graded_fn(lambda x: x + "!", 0, 1)

pipeline = compose_graded(s1, s2)
print(f"Result: {pipeline['fn']('hello')}")
print(f"Total reads:  {pipeline['reads']}")
print(f"Total writes: {pipeline['writes']}")
print()
print("Both tracked in one grade pair.")
print("The semiring handles the accounting.")