python-context-manager-writer

1. INVENTORY -> list every resource/state change: what acquires it, who owns it, how it is released
2. SCAN      -> python scripts/ctx_tool.py leaks <file|dir> (static; never runs code)
3. CHOOSE    -> with-statement / @contextmanager / class / ExitStack / lifespan (table below)
4. WRITE     -> setup before try, one yield inside try/finally, falsy __exit__, typed
                (scripts/ctx_tool.py scaffold ... gives a verified skeleton)
5. EDGE      -> partial setup failure, body raises, early return, cancellation, reuse, threads
6. TEST      -> teardown on success + on exception + exception still propagates
7. RE-SCAN   -> ctx_tool.py leaks --strict exits 0

from contextlib import contextmanager
from typing import Iterator


@contextmanager
def leased_gpu(pool: "DevicePool") -> Iterator[int]:
    device = pool.take()            # 1. setup BEFORE try (nothing to undo if it fails)
    try:
        yield device                # 2. exactly one yield, inside try
    finally:
        pool.give_back(device)      # 3. teardown in finally (runs on error too)

# BEFORE
def ingest(paths, db_path):
    conn = sqlite3.connect(db_path)                 # never closed
    pool = ThreadPoolExecutor(8)                    # never shut down
    for p in paths:
        text = open(p).read()                       # closed only by GC
        chunks = list(pool.map(embed, split(text)))
        conn.executemany("INSERT INTO chunks VALUES (?, ?)", chunks)
    conn.commit()

# AFTER
from contextlib import closing
from pathlib import Path


def ingest(paths: list[Path], db_path: str) -> None:
    with closing(sqlite3.connect(db_path)) as conn, ThreadPoolExecutor(8) as pool:
        with conn:                                  # sqlite3: transaction, not close
            for p in paths:
                text = p.read_text(encoding="utf-8")
                chunks = list(pool.map(embed, split(text)))
                conn.executemany("INSERT INTO chunks VALUES (?, ?)", chunks)

# BEFORE                                  # AFTER
class EvalMode:                           class EvalMode:
    def __enter__(self):                      def __enter__(self) -> "EvalMode":
        self.m.eval()                             self.was_training = self.m.training
                                                  self.m.eval()
                                                  return self
    def __exit__(self, *a):                   def __exit__(self, exc_type, exc, tb) -> None:
        self.m.train()                            self.m.train(self.was_training)
        return True   # hides every bug           # falsy: exceptions propagate

# BEFORE                                  # AFTER
lock.acquire()                            with lock:
cache[key] = compute()  # raises ->           cache[key] = compute()
lock.release()          # deadlock

# BEFORE: new pool + TLS handshake per call, never closed
async def complete(prompt):
    client = httpx.AsyncClient()
    r = await client.post(URL, json={"prompt": prompt})
    return r.json()

# AFTER
from contextlib import asynccontextmanager
from typing import AsyncIterator


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(60, connect=5))
    try:
        yield
    finally:
        await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

from contextlib import ExitStack


def open_shards(paths):
    with ExitStack() as stack:
        files = [stack.enter_context(open(p, encoding="utf-8")) for p in paths]
        stack.callback(log.info, "closed %d shards", len(files))
        return merge(files)


class Index:
    def __init__(self, vec_path, meta_path):
        with ExitStack() as stack:                       # releases on partial failure
            self.vec = stack.enter_context(open(vec_path, "rb"))
            self.meta = stack.enter_context(closing(sqlite3.connect(meta_path)))
            self._stack = stack.pop_all()                # success: keep them open

    def close(self) -> None:
        self._stack.close()

# Static leak and protocol-bug scan (files, directories, or - for stdin)
python scripts/ctx_tool.py leaks app/ingest.py
python scripts/ctx_tool.py leaks src/ --json
cat agent.py | python scripts/ctx_tool.py leaks - --strict             # exit 1 on medium/high findings
python scripts/ctx_tool.py leaks src/ --strict --fail-on high          # CI gate on high only
python scripts/ctx_tool.py leaks src/ --min-severity medium            # hide low-severity notes
python scripts/ctx_tool.py rules                                       # CTX001..CTX012

# Scaffold a typed, documented manager (verified with ast.parse; exit 2 if it does not parse)
python scripts/ctx_tool.py scaffold vector_tx --kind generator \
    --setup "conn = pool.getconn()" --teardown "pool.putconn(conn)" --yields conn
python scripts/ctx_tool.py scaffold ModelSession --kind async-class \
    --setup "self.client = httpx.AsyncClient()" --teardown "await self.client.aclose()" --import "import httpx"
python scripts/ctx_tool.py scaffold lease --kind async-generator \
    --setup "conn = await pool.acquire()" --teardown "await pool.release(conn)" --yields conn
python scripts/ctx_tool.py scaffold --list-recipes
python scripts/ctx_tool.py scaffold --recipe gpu-inference              # timer|temp-env|cwd|gpu-inference|
python scripts/ctx_tool.py scaffold eval_seed --recipe seed --json      # atomic-write|semaphore|span|seed