Phase 4.2-4.4: require-auth + API key docs, rate limit, access logs, Parquet export

- RUNSTREAM_REQUIRE_AUTH: POST requires Bearer; 503 if key unset when required
- .env.example + compose defaults; Docker POST auth documented in README
- AccessLogMiddleware + optional RateLimitMiddleware (RUNSTREAM_ENABLE_RATE_LIMIT)
- runstream serve configures runstream.access logging
- runstream export-parquet + optional [parquet] extra; pyarrow in [dev] for CI
- Tests: auth misconfig, rate limit 429, Parquet roundtrip + empty DB

Made-with: Cursor

Author: MemoryWorld

.env.example

@@ -0,0 +1,14 @@
+# Copy to .env and adjust. Docker Compose reads RUNSTREAM_API_KEY from the environment.
+
+# Required when RUNSTREAM_REQUIRE_AUTH=1 (recommended for any exposed deployment)
+RUNSTREAM_API_KEY=generate-a-long-random-secret
+
+# Set to 1 in production / Docker to force POST /runs to require Bearer token
+RUNSTREAM_REQUIRE_AUTH=1
+
+# Optional: per-IP sliding window (requests per minute, default 120)
+# RUNSTREAM_ENABLE_RATE_LIMIT=1
+# RUNSTREAM_RATE_LIMIT_RPM=120
+
+# Optional: disable one-line app access logs (default: logs enabled when using runstream serve)
+# RUNSTREAM_DISABLE_ACCESS_LOG=1

README.md

@@ -32,7 +32,8 @@ Same body shape as `meta.json` / `RunRecord`. Idempotent via SHA-256 of canonica
 curl -s -X POST http://127.0.0.1:8000/runs -H "Content-Type: application/json" -d @fixtures/example_run/meta.json
 ```
 
-If `RUNSTREAM_API_KEY` is set, send `Authorization: Bearer <key>`.
+- **Dev (default):** if only `RUNSTREAM_API_KEY` is set, send `Authorization: Bearer <key>` on POST.
+- **Production (recommended):** set `RUNSTREAM_REQUIRE_AUTH=1` and **`RUNSTREAM_API_KEY`** — POST without a valid Bearer returns 401/503. Docker Compose enables this by default; override the key via env (see `.env.example`).
 
 ### LLM + tools (optional)
 
@@ -51,6 +52,21 @@ runstream watch fixtures/example_run --db runstream.db --debounce 2
 
 On `meta.json` create/modify under the watched tree, ingest re-runs after **debounce** seconds (default 2). Ctrl+C stops.
 
+### Rate limit & access logs (API)
+
+When **`RUNSTREAM_ENABLE_RATE_LIMIT=1`**, each client IP is limited to **`RUNSTREAM_RATE_LIMIT_RPM`** requests per sliding 60s window (default **120**). `/health`, `/docs`, `/openapi.json`, and `/redoc` are exempt.
+
+With **`runstream serve`**, one-line access logs go to stderr under the logger **`runstream.access`**. Set **`RUNSTREAM_DISABLE_ACCESS_LOG=1`** to turn them off.
+
+### Parquet export
+
+```bash
+pip install 'runstream[parquet]'   # or dev extra already includes pyarrow
+runstream export-parquet --db runstream.db --out runs.parquet
+```
+
+Writes the SQLite `runs` table as Parquet (JSON fields remain JSON strings).
+
 ### Cron (scheduled ingest)
 
 `ingest-once` is idempotent; suitable for cron:
@@ -66,10 +82,13 @@ On `meta.json` create/modify under the watched tree, ingest re-runs after **debo
 ```bash
 mkdir data
 runstream ingest-once fixtures/example_run --db data/runstream.db
+export RUNSTREAM_API_KEY="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
 docker compose build
 docker compose up
 ```
 
+Compose sets **`RUNSTREAM_REQUIRE_AUTH=1`** by default; POST `/runs` needs `Authorization: Bearer $RUNSTREAM_API_KEY`. Copy [`.env.example`](.env.example) to tune variables.
+
 API listens on port **8000**; the DB file is `./data/runstream.db` mounted at `/data/runstream.db`.
 
 ---
@@ -94,16 +113,16 @@ API listens on port **8000**; the DB file is `./data/runstream.db` mounted at `/
 | FastAPI: `GET /health`, `/runs`, `/runs/{id}`, **`POST /runs`** | Done |
 | OpenAI tools: `search_runs`, `get_run` → `tools.py` / `execute_tool` | Done |
 | `ask_with_llm` + mock regression test | Done |
-| Pytest (18 tests) | Done |
+| Pytest (CI) | Done |
 | GitHub Actions CI | Done |
 | Dockerfile + docker-compose | Done |
 
-### Phase 4 (in progress)
+### Phase 4
 
 1. ~~Watch / cron~~ — **done** (`runstream watch`, cron uses `ingest-once`).
-2. **Auth default-on** — next: `RUNSTREAM_REQUIRE_AUTH` + compose defaults.
-3. **Rate limit + access logs** — then.
-4. **Parquet export** — then.
+2. ~~Auth default-on~~ — **done**: `RUNSTREAM_REQUIRE_AUTH` + Docker Compose + `.env.example`.
+3. ~~Rate limit + access logs~~ — **done**: `RUNSTREAM_ENABLE_RATE_LIMIT`, `RUNSTREAM_RATE_LIMIT_RPM` (default 120), `runstream.access` logs (disable with `RUNSTREAM_DISABLE_ACCESS_LOG=1`).
+4. ~~Parquet export~~ — **done**: `runstream export-parquet --db … --out runs.parquet` (`pip install 'runstream[parquet]'`).
 
 ---
 
@@ -117,6 +136,8 @@ src/runstream/
   tools.py      # OpenAI tool schemas + execute_tool (no shell)
   ask.py        # optional LLM loop
   api.py
+  http_middleware.py  # access logs + optional rate limit
+  parquet_export.py
   cli.py
   watch_ingest.py
 schemas/

docker-compose.yml

@@ -8,3 +8,9 @@ services:
       - ./data:/data
     environment:
       RUNSTREAM_DB: /data/runstream.db
+      # Production-style defaults: override RUNSTREAM_API_KEY in shell or .env
+      RUNSTREAM_REQUIRE_AUTH: "1"
+      RUNSTREAM_API_KEY: ${RUNSTREAM_API_KEY:-change-me-before-exposing-this-port}
+      # Optional hardening (uncomment in .env)
+      # RUNSTREAM_ENABLE_RATE_LIMIT: "1"
+      # RUNSTREAM_RATE_LIMIT_RPM: "120"

pyproject.toml

@@ -18,7 +18,8 @@ dependencies = [
 
 [project.optional-dependencies]
 llm = ["openai>=1.40"]
-dev = ["pytest>=8.0", "httpx>=0.27", "openai>=1.40"]
+parquet = ["pyarrow>=14.0"]
+dev = ["pytest>=8.0", "httpx>=0.27", "openai>=1.40", "pyarrow>=14.0"]
 
 [project.scripts]
 runstream = "runstream.cli:main"

src/runstream/api.py

@@ -6,19 +6,39 @@
 from fastapi import FastAPI, HTTPException, Header, Query
 from fastapi.responses import JSONResponse
 
+from .http_middleware import AccessLogMiddleware, RateLimitMiddleware
 from .ingest import ingest_record
 from .models import RunRecord
 from .store import connect, get_run, list_runs
 
 app = FastAPI(title="Runstream", version="0.2.0")
+app.add_middleware(RateLimitMiddleware)
+app.add_middleware(AccessLogMiddleware)
 
 
 def _db_path() -> Path:
     return Path(os.environ.get("RUNSTREAM_DB", "runstream.db")).resolve()
 
 
+def _env_truthy(name: str) -> bool:
+    return os.getenv(name, "").strip().lower() in ("1", "true", "yes", "on")
+
+
 def _require_write_auth(authorization: str | None) -> None:
     expected = os.environ.get("RUNSTREAM_API_KEY")
+    require = _env_truthy("RUNSTREAM_REQUIRE_AUTH")
+    if require:
+        if not expected:
+            raise HTTPException(
+                status_code=503,
+                detail="RUNSTREAM_API_KEY must be set when RUNSTREAM_REQUIRE_AUTH=1",
+            )
+        if not authorization or not authorization.startswith("Bearer "):
+            raise HTTPException(status_code=401, detail="Authorization Bearer token required")
+        token = authorization[7:].strip()
+        if token != expected:
+            raise HTTPException(status_code=403, detail="invalid token")
+        return
     if not expected:
         return
     if not authorization or not authorization.startswith("Bearer "):
@@ -72,7 +92,8 @@ def create_run(
 ) -> dict:
     """
     Upsert a run (same validation + idempotency as file ingest).
-    If env RUNSTREAM_API_KEY is set, require `Authorization: Bearer <key>`.
+    If `RUNSTREAM_REQUIRE_AUTH=1`, Bearer token is **always** required and `RUNSTREAM_API_KEY` must be set.
+    Otherwise, if only `RUNSTREAM_API_KEY` is set, Bearer is required for POST.
     """
     _require_write_auth(authorization)
     action = ingest_record(record, _db_path(), source_path=None)

src/runstream/cli.py

@@ -7,6 +7,7 @@
 
 from .ask import ask_with_llm
 from .ingest import ingest_path
+from .parquet_export import export_runs_parquet
 from .tools import openai_tools_json
 from .watch_ingest import watch_and_ingest
 
@@ -23,6 +24,16 @@ def ingest_once(
     typer.echo(f"ingest complete -> {db}: {stats}")
 
 
+@app.command("export-parquet")
+def export_parquet_cmd(
+    out: Path = typer.Option(..., "--out", help="Output .parquet file path"),
+    db: Path = typer.Option(Path("runstream.db"), "--db", help="SQLite database path"),
+) -> None:
+    """Export all rows from the runs table to Parquet (requires pip install 'runstream[parquet]')."""
+    n = export_runs_parquet(db, out)
+    typer.echo(f"exported {n} rows -> {out}")
+
+
 @app.command("watch")
 def watch_cmd(
     path: Path = typer.Argument(..., exists=True, help="Directory (or file under a directory) to watch"),
@@ -40,9 +51,16 @@ def serve(
     port: int = typer.Option(8000, "--port"),
 ) -> None:
     """Run HTTP API (FastAPI + uvicorn)."""
+    import logging
     import os
 
     os.environ["RUNSTREAM_DB"] = str(db.resolve())
+    access = logging.getLogger("runstream.access")
+    if not access.handlers:
+        _h = logging.StreamHandler()
+        _h.setFormatter(logging.Formatter("%(levelname)s [access] %(message)s"))
+        access.addHandler(_h)
+    access.setLevel(logging.INFO)
     uvicorn.run(
         "runstream.api:app",
         host=host,

src/runstream/http_middleware.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import logging
+import os
+import time
+from collections import defaultdict
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+logger = logging.getLogger("runstream.access")
+
+# Shared bucket for rate limiting (cleared in tests via clear_rate_limit_state_for_tests).
+_rate_limit_hits: dict[str, list[float]] = defaultdict(list)
+
+
+def clear_rate_limit_state_for_tests() -> None:
+    _rate_limit_hits.clear()
+
+
+def _env_truthy(name: str) -> bool:
+    return os.getenv(name, "").strip().lower() in ("1", "true", "yes", "on")
+
+
+class AccessLogMiddleware(BaseHTTPMiddleware):
+    """Structured-ish one-line access log (disable with RUNSTREAM_DISABLE_ACCESS_LOG=1)."""
+
+    async def dispatch(self, request: Request, call_next) -> Response:  # type: ignore[no-untyped-def]
+        if _env_truthy("RUNSTREAM_DISABLE_ACCESS_LOG"):
+            return await call_next(request)
+        t0 = time.perf_counter()
+        response = await call_next(request)
+        ms = (time.perf_counter() - t0) * 1000
+        client = request.client.host if request.client else "-"
+        logger.info(
+            "%s %s %s %d %.2fms",
+            client,
+            request.method,
+            request.url.path,
+            response.status_code,
+            ms,
+        )
+        return response
+
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    """
+    Sliding window per client IP when RUNSTREAM_ENABLE_RATE_LIMIT=1.
+    RUNSTREAM_RATE_LIMIT_RPM (default 120). /health is exempt.
+    """
+
+    def _rpm(self) -> int:
+        try:
+            return max(1, int(os.getenv("RUNSTREAM_RATE_LIMIT_RPM", "120")))
+        except ValueError:
+            return 120
+
+    async def dispatch(self, request: Request, call_next) -> Response:  # type: ignore[no-untyped-def]
+        if not _env_truthy("RUNSTREAM_ENABLE_RATE_LIMIT"):
+            return await call_next(request)
+        if request.url.path in ("/health", "/docs", "/openapi.json", "/redoc"):
+            return await call_next(request)
+
+        now = time.monotonic()
+        window = 60.0
+        rpm = self._rpm()
+        ip = request.client.host if request.client else "unknown"
+
+        bucket = _rate_limit_hits[ip]
+        bucket[:] = [t for t in bucket if now - t < window]
+        if len(bucket) >= rpm:
+            return JSONResponse(
+                status_code=429,
+                content={"detail": "rate limit exceeded", "retry_after_sec": 60},
+            )
+        bucket.append(now)
+
+        return await call_next(request)

src/runstream/parquet_export.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from .store import connect
+
+
+def export_runs_parquet(db_path: Path, out_path: Path) -> int:
+    """
+    Dump the `runs` table to Parquet (raw SQLite column layout: JSON as strings).
+    Requires optional dependency: pip install 'runstream[parquet]'.
+    """
+    try:
+        import pyarrow as pa
+        import pyarrow.parquet as pq
+    except ImportError as e:  # pragma: no cover - exercised when pyarrow missing
+        raise SystemExit(
+            "Parquet export needs pyarrow. Install: pip install 'runstream[parquet]'"
+        ) from e
+
+    conn = connect(db_path)
+    try:
+        cur = conn.execute("SELECT * FROM runs ORDER BY started_at DESC")
+        col_names = [c[0] for c in cur.description]
+        rows = cur.fetchall()
+        if not rows:
+            table = pa.table(
+                {c: pa.array([], type=pa.string()) for c in col_names}
+            )
+        else:
+            data = {col: [r[col] for r in rows] for col in col_names}
+            table = pa.table(data)
+        out_path = out_path.resolve()
+        out_path.parent.mkdir(parents=True, exist_ok=True)
+        pq.write_table(table, out_path)
+        return len(rows)
+    finally:
+        conn.close()

tests/conftest.py

@@ -0,0 +1,10 @@
+import pytest
+
+from runstream.http_middleware import clear_rate_limit_state_for_tests
+
+
+@pytest.fixture(autouse=True)
+def _clear_rate_limit_between_tests() -> None:
+    clear_rate_limit_state_for_tests()
+    yield
+    clear_rate_limit_state_for_tests()

tests/test_export_parquet.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+
+import pyarrow.parquet as pq
+
+from runstream.ingest import ingest_path
+from runstream.parquet_export import export_runs_parquet
+
+
+def test_export_parquet_roundtrip(tmp_path: Path) -> None:
+    db = tmp_path / "t.db"
+    ingest_path(Path("fixtures/example_run"), db)
+    out = tmp_path / "runs.parquet"
+    n = export_runs_parquet(db, out)
+    assert n >= 1
+    assert out.is_file()
+    table = pq.read_table(out)
+    assert "run_id" in table.column_names
+    assert table.num_rows == n
+
+
+def test_export_parquet_empty_db(tmp_path: Path) -> None:
+    db = tmp_path / "empty.db"
+    db.touch()
+    from runstream.store import connect
+
+    connect(db).close()
+    out = tmp_path / "empty.parquet"
+    n = export_runs_parquet(db, out)
+    assert n == 0
+    t = pq.read_table(out)
+    assert t.num_rows == 0