Python in cwkPippa — A Walking Tour

# Sketch of cwkPippa's Claude adapter (simplified)
from abc import ABC, abstractmethod
from typing import AsyncIterator
from dataclasses import dataclass

@dataclass
class StreamEvent:
    """What an adapter yields back to the caller."""
    kind: str            # 'token' | 'tool_use' | 'thinking' | 'done'
    content: str | None = None

class Adapter(ABC):
    @abstractmethod
    async def stream(self, prompt: str) -> AsyncIterator[StreamEvent]:
        ...

class ClaudeAdapter(Adapter):
    async def stream(self, prompt: str) -> AsyncIterator[StreamEvent]:
        # 1. Connect to the SDK (real version uses claude_agent_sdk)
        # 2. Send the prompt
        # 3. As chunks arrive, yield them
        # 4. Persist each chunk to JSONL before yielding (write-before-show)
        # 5. On done, write the final marker and return
        async for chunk in self._sdk_stream(prompt):
            yield StreamEvent(kind="token", content=chunk.text)
        yield StreamEvent(kind="done")

    async def _sdk_stream(self, prompt):
        # ... actual SDK integration ...
        yield  # placeholder for the type checker

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter()

class ChatRequest(BaseModel):
    conversation_id: str
    message: str
    brain: str = "claude"

class ChatChunk(BaseModel):
    kind: str
    content: str | None = None

@router.post("/api/chat")
async def chat(req: ChatRequest):
    """Stream chat events as Server-Sent Events."""
    # 1. Heal any broken turns from the previous session (idempotent)
    # 2. Get the right adapter (Claude / Codex / Gemini / Ollama)
    # 3. Stream events out, persisting each one to JSONL first
    # 4. On done, derive the final SQLite row from JSONL
    # ...
    # The shape — Pydantic in, async stream out, JSONL ground truth in the middle
    pass

import json
from pathlib import Path
from datetime import datetime, timezone

class SessionLogger:
    """Append-only JSONL writer for one conversation."""

    def __init__(self, conversation_id: str, root: Path):
        self.path = root / f"{conversation_id}.jsonl"

    def append(self, event: dict) -> None:
        # Write before show — durability before user sees the chunk
        event["ts"] = datetime.now(timezone.utc).isoformat()
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False) + "\n")

    def replay(self):
        # Recovery — read the JSONL line by line to rebuild SQLite
        with self.path.open(encoding="utf-8") as f:
            for line in f:
                yield json.loads(line)

# Why JSONL? Append-only writes, line-by-line streaming reads, durable per-event.
# The ground truth. SQLite + ChromaDB are derived mirrors.

Foundations          → variables, types, strings, basic I/O
Data                 → list, dict, tuple, set, comprehensions, bytes
Flow                 → if/match, loops, functions, closures, walrus
Iterators            → iter/next, generators, itertools, async iter
Decorators           → @, wraps, factories, class decorators, @property
OOP                  → class, dunders, dataclass, Protocol, ABC
OOP Advanced         → MRO, mixins, dispatch, metaclasses, descriptors
Errors               → try/except, EAFP, custom exceptions, contextmanagers
Files & I/O          → open, pathlib, JSON, CSV, encoding, mmap
Standard Library     → collections, functools, datetime, re, logging
Modules              → import, packages, venv, pyproject.toml
Typing               → hints, Literal, TypedDict, generics, Pydantic
Concurrency          → asyncio, threading, multiprocessing, GIL
Tooling              → pytest, mock, pdb, ruff, mypy, profilers
CLI                  → argparse, click, typer, rich
Pythonic             → EAFP, duck typing, when not to class, the Zen

# Every line of cwkPippa code uses something from this list.
# You can now read all of it.