# CodeUChain (Python) – Full LLM Reference (Canonical Implementation)

**Name:** CodeUChain (Python)

**Homepage:** https://github.com/codeuchain/codeuchain/tree/main/packages/python

**Docs:** https://codeuchain.github.io/codeuchain/python/

**Version:** 1.0.0

**License:** Apache 2.0

**Repository:** git+https://github.com/codeuchain/codeuchain.git

**Contact:** https://github.com/codeuchain/codeuchain/issues

**Authors:** CodeUChain contributors

**Language:** Python 3.8+ (async-first)

**Platform:** Cross-platform

**Paradigm Keywords:** Composable, Async, Generics, Immutable State, Type Evolution, Selfless Links

---

## 1. Purpose & Philosophy

Python is the *reference* implementation: every concept is demonstrated here first. Emphasis: clarity over cleverness, explicit transformations, compassionate error handling, and strong optional typing.

| Principle | Python Expression | Benefit |

|-----------|-------------------|---------|

| Selfless Links | `async def call(ctx)` | Pure async units |

| Immutable State | `ctx2 = ctx.insert(...)` | Predictable test state |

| Type Evolution | `ctx3 = ctx.insert_as("key", value)` | Safe shape widening |

| Mixed Typed/Untyped | `State[Any]` default | Gradual adoption |

| Async Everywhere | `await chain.call()` | Natural concurrency |

---

## 2. Architectural Overview

```

Raw Input --> State[T0]

│ then (validation_link)

▼

State[T1]

│ then (parse_link)

▼

State[T2]

│ then (enrich_link) + hook(before/after/error)

▼

State[T3] (final)

```

Advanced flows: branching, conditional execution, retry wrapping, error redirection.

---

## 3. Core Types (Conceptual Signatures)

```python

class State(Generic[T]):

def get(self, key: str, default: Any = None) -> Any: ...

def insert(self, key: str, value: Any) -> "State[T]": ...

def insert_as(self, key: str, value: Any) -> "State[Any]": ... # evolves type

def keys(self) -> list[str]: ...

def to_dict(self) -> dict[str, Any]: ...

class Link(Generic[TInput, TOutput]):

async def call(self, ctx: State[TInput]) -> State[TOutput]: ...

class Hook: # All optional

async def before(self, name: str, ctx: State[Any]) -> None: ...

async def after(self, name: str, ctx: State[Any]) -> None: ...

async def on_error(self, name: str, ctx: State[Any], err: Exception) -> None: ...

```

---

## 4. Creating Links

```python

from codeuchain import Link, State

class ValidateEmail(Link[Any, Any]):

async def call(self, ctx: State[Any]) -> State[Any]:

email = ctx.get("email")

if not email or "@" not in email:

raise ValueError("invalid_email")

return ctx.insert("validated", True)

```

### Type Evolution

```python

from dataclasses import dataclass

@dataclass

class RawInput:

text: str

@dataclass

class Parsed:

text: str

tokens: list[str]

class Parse(Link[RawInput, Parsed]):

async def call(self, ctx: State[RawInput]) -> State[Parsed]:

raw: RawInput = ctx.get("raw")

parsed = Parsed(text=raw.text, tokens=raw.text.split())

return ctx.insert_as("parsed", parsed)

```

---

## 5. Chain Composition & Error Handling

```python

from codeuchain import Chain

chain = (Chain()

.then(ValidateEmail())

.then(Parse())

.catch(lambda link, err, ctx: ctx.insert("error_tag", str(err))))

result = await chain.call(State[Any]({"email": "a@b.com", "raw": RawInput("hello world")}))

```

Branching strategies: implement conditional link wrappers or pre-insert flags used by downstream links.

Retry decorator pattern:

```python

def with_retry(link: Link[TInput, TOutput], attempts: int) -> Link[TInput, TOutput]:

class Retry(Link[TInput, TOutput]):

async def call(self, ctx: State[TInput]) -> State[TOutput]:

last = None

for i in range(attempts):

try:

return await link.call(ctx)

except Exception as e: # narrow if needed

last = e

await asyncio.sleep(0.01 * (i + 1))

raise last # surfaced after exhaustion

return Retry()

```

---

## 6. Hook Lifecycle

```python

class MetricsHook:

async def before(self, name: str, ctx: State[Any]) -> None:

ctx = ctx.insert("_t0", time.perf_counter())

async def after(self, name: str, ctx: State[Any]) -> None:

t0 = ctx.get("_t0")

if t0:

dt = time.perf_counter() - t0

print(f"{name} took {dt*1000:.2f}ms")

async def on_error(self, name: str, ctx: State[Any], err: Exception) -> None:

print(f"ERROR in {name}: {err}")

```

Guidelines:

- Side-effect work should be fast; offload heavy operations.

- Hook ordering = registration order.

---

## 7. Error Handling Patterns

| Pattern | Usage | Example |

|---------|-------|---------|

| Central catch | Uniform tagging | `.catch(handler)` |

| Retry wrapper | Transient failures | `with_retry(link, 3)` |

| Classification | Route by error type | branching inside catch |

| Enrichment | Attach state diagnostics | insert stack or counters |

Graceful classification snippet:

```python

def classify_catch(link_name: str, err: Exception, ctx: State[Any]) -> State[Any]:

tag = "transient" if isinstance(err, TimeoutError) else "fatal"

return ctx.insert("error_kind", tag).insert("error_msg", str(err))

chain = Chain().then(work_link).catch(classify_catch)

```

---

## 8. Testing & TDD

Why ideal:

- Pure async functions

- State = explicit contract

- Type evolution clarifies transitions

Recommended test style:

```python

import pytest

@pytest.mark.asyncio

async def test_validate_email_ok():

ctx = State[Any]({"email": "a@b.com"})

out = await ValidateEmail().call(ctx)

assert out.get("validated") is True

@pytest.mark.asyncio

async def test_validate_email_fail():

ctx = State[Any]({"email": "broken"})

with pytest.raises(ValueError):

await ValidateEmail().call(ctx)

```

Chain table-driven style:

```python

cases = [

("a@b.com", True),

("invalid", False),

]

for email, ok in cases:

ctx = State[Any]({"email": email, "raw": RawInput("hi all")})

try:

await chain.call(ctx)

assert ok

except Exception:

assert not ok

```

Coverage & typing:

```bash

pytest --cov=codeuchain --cov-report=term-missing

mypy codeuchain/

```

---

## 9. Observation & Debugging

Tools:

- Hook logging

- State key introspection

- Timing via perf_counter

- Assertion helpers in tests

Debug hook example:

```python

class Debug:

async def after(self, name: str, ctx: State[Any]) -> None:

print("DBG", name, "keys=", ctx.keys())

```

---

## 10. Performance Notes

| Concern | Strategy |

|---------|----------|

| Excess object churn | Reuse builders; limit deep copies |

| Serialization overhead | Defer (store raw payload) |

| Async fan-out | `asyncio.gather` with sub-chains |

| Logging cost | Structured logger + sampling |

| Type conversions | Narrow casts once; reuse typed vars |

Micro-bench idea:

```bash

pytest tests/perf/test_chain_perf.py -k bench --maxfail=1

```

---

## 11. Advanced Patterns

- Dynamic branching: insert a `route` key; have a dispatcher link

- Partial failure aggregation: collect errors and continue (`best-effort` mode)

- Saga compensation: pair forward links with compensators

- Streaming adaptation: wrap async generators as link outputs

---

## 12. Ecosystem Integrations

Examples:

- FastAPI endpoint: call chain inside request handler

- Celery task: each link is pure → easy unit test / idempotency

- Pydantic models: used as typed payload shapes evolving through `insert_as`

- Observability: integrate with OpenTelemetry in hook

---

## 13. Migration & Mixed Typing

Start with `State[Any]`. Once stable, replace hotspots with domain dataclasses + generics. Intermix freely—no rewrite required.

---

## 14. Anti-Patterns

| Issue | Why | Fix |

|-------|-----|-----|

| Storing huge blobs | Memory strain | External store + reference id |

| Overuse of `insert_as` without typing | Loses clarity | Introduce dataclasses |

| Catch-all `except` hiding bugs | Silent failures | Classify & rethrow critical |

| Hook doing business logic | Breaks separation | Move into a link |

---

## 15. FAQ

**Q: Can I use sync links?**

A: Wrap them: `async def call(): return sync_link(ctx)` inside an async link.

**Q: How to cancel?**

A: Pass an `asyncio.Task` cancellation upstream; chain surfaces errors naturally.

**Q: Is state thread-safe?**

A: It is immutable; each `insert` returns a new instance.

**Q: Where to validate types?**

A: Early links + optional Pydantic models.

**Q: Retry location?**

A: A decorator/wrapper link for clarity.

---

## 16. Glossary

- **Link**: Async transformer from State[TIn] → State[TOut].

- **Chain**: Ordered link composition.

- **State**: Immutable mapping with type evolution helpers.

- **Hook**: Observers for before/after/error phases.

- **Type Evolution**: Safe widening via `insert_as` returning new generic state.

---

## 17. TL;DR

```text

Install: pip install codeuchain

Model: Links (pure async) + Chain (composition) + State (immutable) + Hook (observability) + Type Evolution

Typing: Start Any → introduce dataclasses → use insert_as to evolve

Testing: Per-link async tests + chain table cases

Observability: Lightweight hook; avoid business logic there

Performance: Avoid deep copies; batch IO with asyncio.gather

Error Handling: Central catch + targeted retry decorators

Adoption: Gradual—mix typed/untyped seamlessly

Avoid: giant blobs, silent excepts, coupling in hook

```

---

### Support & Resources

- Issues: https://github.com/codeuchain/codeuchain/issues

- Discussions: https://github.com/codeuchain/codeuchain/discussions

- Examples: `packages/python/examples/`

- License: Apache 2.0

---

© 2025 Orchestrate LLC (Joshua @orchestrate.solutions) – Apache 2.0