Wrapper gaps: recipe rate-limiter bypassable+worker-unsafe, multiedit corrupts files on crash, __main__ Typer/legacy split billed to LLM

# recipe/serve.py:90-123 — no lock; concurrent async requests race on _requests
class RateLimiter:
    def __init__(self, requests_per_minute: int = DEFAULT_RATE_LIMIT):
        self.requests_per_minute = requests_per_minute
        self.window_seconds = 60
        self._requests: Dict[str, List[float]] = defaultdict(list)

    def check(self, client_id: str) -> Tuple[bool, int]:
        ...
        self._requests[client_id] = [
            t for t in self._requests[client_id] if t > window_start
        ]
        if len(self._requests[client_id]) >= self.requests_per_minute:
            ...
        self._requests[client_id].append(now)
        return True, 0

# recipe/serve.py:779 — bucket id is untrusted client header
class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if rate_limiter is None:
            return await call_next(request)
        if request.url.path in rate_limit_exempt:
            return await call_next(request)

        # Attacker rotates X-API-Key each request → each value gets its own bucket
        # → 100 rpm limit becomes effectively unlimited.
        client_id = request.headers.get("X-API-Key") or request.client.host if request.client else "unknown"

        allowed, retry_after = rate_limiter.check(client_id)
        ...

# recipe/serve.py:836-873 — workers>1 silently doesn't work
def serve(host="127.0.0.1", port=8765, reload=False, config=None, workers: int = 1):
    ...
    app = create_app(config)
    if workers > 1 and reload:
        warnings.warn("Cannot use reload with multiple workers. Disabling reload.")
        reload = False
    # `workers > 1` requires an import string, not an app object.
    uvicorn.run(app, host=host, port=port, reload=reload, workers=workers if workers > 1 else None)

# recipe/serve.py — thread/async-safe limiter with pluggable backend
import asyncio
from typing import Protocol

class RateLimiterBackend(Protocol):
    async def check(self, client_id: str) -> Tuple[bool, int]: ...

class InMemoryRateLimiter(RateLimiterBackend):
    def __init__(self, requests_per_minute: int):
        self.requests_per_minute = requests_per_minute
        self.window_seconds = 60
        self._requests: Dict[str, List[float]] = defaultdict(list)
        self._lock = asyncio.Lock()   # <-- fixes the async race

    async def check(self, client_id: str) -> Tuple[bool, int]:
        if self.requests_per_minute <= 0:
            return True, 0
        now = time.time()
        window_start = now - self.window_seconds
        async with self._lock:
            bucket = [t for t in self._requests[client_id] if t > window_start]
            self._requests[client_id] = bucket
            if len(bucket) >= self.requests_per_minute:
                retry_after = int(min(bucket) + self.window_seconds - now) + 1
                return False, max(1, retry_after)
            bucket.append(now)
            return True, 0

# Optional: RedisRateLimiter(RateLimiterBackend) for workers > 1 — a shared
# atomic INCR + EXPIRE keyed by client_id is ~15 lines and closes the multi-
# worker gap without any new hard dependency (redis is already used elsewhere
# in the wrapper).

# recipe/serve.py — resolve bucket id after auth, from the validated identity
async def dispatch(self, request, call_next):
    if rate_limiter is None or request.url.path in rate_limit_exempt:
        return await call_next(request)

    # Precedence: validated JWT sub → validated API key → client IP → "anonymous".
    # Only a value the auth middleware has already validated ends up in client_id.
    user = getattr(request.state, "user", None)          # set by JWTAuthMiddleware
    if user and "sub" in user:
        client_id = f"jwt:{user['sub']}"
    elif getattr(request.state, "authenticated_api_key", None):
        client_id = f"apikey:{request.state.authenticated_api_key}"
    elif request.client:
        client_id = f"ip:{request.client.host}"
    else:
        client_id = "anonymous"

    allowed, retry_after = await rate_limiter.check(client_id)
    if not allowed:
        return JSONResponse(
            {"error": {"code": "rate_limited", "message": "Too many requests"}},
            status_code=429, headers={"Retry-After": str(retry_after)},
        )
    return await call_next(request)

# recipe/serve.py — honour workers>1 by handing uvicorn an import string
def serve(host="127.0.0.1", port=8765, reload=False, config=None, workers: int = 1):
    if workers > 1:
        # Uvicorn requires an import string for reload/workers. Expose an
        # app-factory module so the config still travels through the environment.
        os.environ["PRAISONAI_RECIPE_SERVE_CONFIG"] = json.dumps(config or {})
        uvicorn.run(
            "praisonai.recipe._serve_factory:app",   # calls create_app() from env
            host=host, port=port, reload=False, workers=workers,
            factory=False,
        )
        return
    uvicorn.run(create_app(config), host=host, port=port, reload=reload)

# tools/multiedit.py:104-166
try:
    with open(filepath, 'r') as f:
        original_content = f.read()

    content = original_content
    lines = content.split('\n')

    # Apply edits (find+replace, possibly with fuzzy match) ...

    # Write file if not dry run
    if not dry_run and result["edits_applied"] > 0:
        with open(filepath, 'w') as f:      # <-- truncates immediately;
            f.write(content)                #     interrupt here = corrupt file
except Exception as e:
    result["error"] = str(e)

# auto.py:70-105 — the correct pattern, already in the wrapper, not reused here
def _atomic_write_text(path, content_writer):
    target_dir = os.path.dirname(os.path.abspath(path)) or "."
    fd, tmp_path = tempfile.mkstemp(prefix=".tmp_", suffix=".yaml", dir=target_dir)
    try:
        # preserve existing mode so a replaced config stays readable
        try:
            existing_mode = os.stat(path).st_mode
        except OSError:
            existing_mode = None
        with os.fdopen(fd, "w") as f:
            content_writer(f)
            f.flush()
            os.fsync(f.fileno())            # durability before rename
        if existing_mode is not None:
            try:
                os.chmod(tmp_path, existing_mode)
            except OSError:
                pass
        os.replace(tmp_path, path)          # atomic on POSIX + Windows
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise

# praisonai/_io.py  (new — one owner for atomic writes across the wrapper)
import os, tempfile
from typing import Callable

def atomic_write_text(path: str, write: Callable[[object], None]) -> None:
    target_dir = os.path.dirname(os.path.abspath(path)) or "."
    fd, tmp_path = tempfile.mkstemp(prefix=".tmp_", dir=target_dir)
    try:
        try:
            existing_mode = os.stat(path).st_mode
        except OSError:
            existing_mode = None
        with os.fdopen(fd, "w") as f:
            write(f)
            f.flush()
            os.fsync(f.fileno())
        if existing_mode is not None:
            try:
                os.chmod(tmp_path, existing_mode)
            except OSError:
                pass
        os.replace(tmp_path, path)
    except BaseException:
        try:
            os.unlink(tmp_path)
        except OSError:
            pass
        raise

# tools/multiedit.py — use the atomic writer + a lockfile for the RMW window
import contextlib, fcntl
from praisonai._io import atomic_write_text

@contextlib.contextmanager
def _file_lock(path: str):
    lock_path = path + ".lock"
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)      # advisory; blocks concurrent editors
        yield
    finally:
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)

# in multiedit(...) — wrap the read-modify-write in the lock and swap the write
with _file_lock(filepath):
    with open(filepath, "r") as f:
        original_content = f.read()

    content = _apply_all_edits(original_content, edits, result)   # unchanged logic

    if not dry_run and result["edits_applied"] > 0:
        atomic_write_text(filepath, lambda f: f.write(content))

# __main__.py:42-55  — static mirror kept in sync by hand so a discovery
# failure doesn't reintroduce the LLM-billing regression from #4327
_LEGACY_VERBS_FALLBACK = frozenset({
    'chat', 'code', 'call', 'realtime', 'train', 'ui', 'context', 'research',
    'memory', 'rules', 'workflow', 'hooks', 'knowledge', 'session', 'tools',
    'todo', 'docs', 'mcp', 'commit', 'serve', 'schedule', 'skills', 'profile',
    ... (60 verbs)
})

# __main__.py:79-85 — same short flag has DIFFERENT meanings in the two engines
_LEGACY_COLLIDING_SHORT_OPTS = frozenset({"-s", "-f"})
# -s : legacy `--save` (bool) vs modern `--session` (takes a value)
# -f : legacy `--file` (input file) vs modern `--framework`

# __main__.py:210-265  — a difflib matcher exists here ONLY because a
# mistyped verb otherwise silently reaches the LLM as a prompt
def _mistyped_command_suggestions(argv, first_cmd):
    ...
    matches = difflib.get_close_matches(first_cmd, sorted(commands), n=3, cutoff=0.8)
    matches = [cmd for cmd in matches if not first_cmd.startswith(cmd)]
    return matches

# __main__.py:645-772 — main() is a hand-tuned decision tree over
# --version / --help / no-args / typer-command / legacy-verb / typo /
# bare-prompt / yaml-target / everything-else
def main():
    argv = sys.argv[1:]
    if "--version" in argv or "-V" in argv: ...
    if "--help" in argv or "-h" in argv: _run_typer(argv); return
    if not argv: _run_typer(argv); return
    ...
    if _is_implemented_legacy_verb(first_cmd): _run_legacy(argv); return
    suggestions = _mistyped_command_suggestions(argv, first_cmd)
    if suggestions: print(...); sys.exit(2)
    if _looks_like_bare_prompt(argv, first_cmd): _run_typer(_build_run_argv(...))
    elif _looks_like_yaml_run_target(argv, first_cmd): _run_typer([..., "run", *yaml_rest])
    else: _run_legacy(argv)

# cli/legacy/adapter.py  (new; the ONLY legacy-flag entry point)
# Each formerly-legacy flag becomes a Typer command that emits a
# deprecation warning and calls the same handler the legacy module
# already implemented, keeping every behaviour intact for one release.

@app.command("run")
def run_cmd(
    target: st