DEV Community: Abhijat Chaturvedi

Fail Closed, Not Open: Designing an AI Gateway for Regulated Enterprises

Abhijat Chaturvedi — Sun, 26 Jul 2026 07:46:29 +0000

"Why did this request go to that model?"

If your company uses LLMs across more than one team, someone will eventually ask you this question. If you work in financial services, healthcare, or anywhere else with a regulator, someone will ask it under oath-adjacent conditions.

Most organizations can't answer it. Teams integrate directly with whatever provider they like, sensitive data handling varies by team, regional restrictions are applied inconsistently, and there's no record of why a model was selected — or, more importantly, why the alternatives weren't.

I recently built Aegis Gateway to explore what a proper answer looks like: a single front door for AI requests that decides which model may serve each request — and can prove why. This article is about the three design decisions that matter, and the one failure mode that I organized the entire architecture around preventing.

The failure mode that matters: silent governance downgrade

Here's the scenario that keeps compliance officers up at night.

Your gateway is well-behaved on the happy path: restricted client data goes to the on-prem model, everything's fine. Then one Tuesday the on-prem cluster goes down. Your fallback logic kicks in, does what fallback logic does — finds something that works — and quietly routes restricted data to an external provider. The request succeeds. Everyone's happy. Nobody notices.

That's a silent governance downgrade: a system that's compliant until it's inconvenient. And the insidious part is that it usually isn't a bug — it's a reasonable-looking design where availability and compliance compete in the same decision, and availability wins because that's what gets paged about.

My design goal was to make this failure mode structurally impossible — not discouraged by code review, not caught by an alert, but unreachable by construction.

Decision 1: Policy is a hard gate, ranking is soft optimization — and the order is the architecture

The routing core has three stages with a strict, non-negotiable order:

catalogue ──► Policy Engine ──► Ranking ──► Fallback Executor
              (hard gate)       (eligible    (eligible list
                                 models       only)
                                 only)

Stage 1 — Policy filtering. Every model in the catalogue is checked against governance rules: Does it have the capability for this task type? Is its provider approved for this business unit? Is it certified for this data classification? Does restricted data stay on private infrastructure? Is it served in this region? Every model that fails records the specific rule it failed.

def evaluate(self, model: ModelSpec, ctx: RequestContext) -> list[str]:
    """Return the list of violated rules (empty list = eligible)."""
    reasons = []
    if ctx.task_type not in model.capabilities:
        reasons.append(f"model does not support task '{ctx.task_type.value}'")
    if model.provider not in unit["approved_providers"]:
        reasons.append(f"provider '{model.provider}' is not approved "
                       f"for business unit '{ctx.business_unit}'")
    if req_level > CLASSIFICATION_LEVEL[model.max_classification]:
        reasons.append(f"model is certified only up to "
                       f"{model.max_classification.value} data")
    # ... region residency, restricted→private, availability
    return reasons

Stage 2 — Ranking. A weighted score over normalized cost, quality, and latency, with weights driven by the request's declared priority (Cost / Balanced / Quality). Plain arithmetic. The crucial property: the ranker's input is the output of the policy gate. It cannot see, and therefore cannot resurrect, a non-compliant model.

Stage 3 — Fallback. If the selected model fails at call time, walk down the ranked list. Which list? The eligible list. When it's exhausted:

if trace.status != "ROUTED":
    trace.rationale = (
        "No model satisfies current policy for this request context. "
        "Request refused rather than downgrading governance."
    )

NO_COMPLIANT_ROUTE. The gateway refuses — even if healthy external models are one function call away.

This is the whole trick, and it's almost embarrassingly simple: the compliance guarantee doesn't live in a rule that says "don't fall back to non-compliant models." It lives in the data flow. Fallback iterates a list that non-compliant models were never in. There is no code path where the trade-off can even be expressed.

An outage is recoverable. A compliance breach is not. The architecture should not be able to make that trade, and in this design, it can't.

Decision 2: Explainability is an output, not a log line

Most systems treat decision rationale as something you grep for after the incident. Aegis treats it as part of the API contract. Every response carries a decision trace:

{
  "request_id": "f188db34-…",
  "policy_version": "2026.07-r4",
  "status": "ROUTED",
  "selected_model": "llama-3-onprem-70b",
  "fallback_model": "llama-3-onprem-8b",
  "rejected_models": [
    { "model_id": "chatgpt",
      "reasons": ["Restricted data may only route to private providers"] },
    { "model_id": "gemini-nano-banana",
      "reasons": ["model does not support task 'text-generation'"] }
  ],
  "eligible_ranked": [
    { "model_id": "llama-3-onprem-70b", "score": 0.7,
      "breakdown": { "cost": 0.0, "quality": 0.7, "latency": 0.0 } }
  ]
}

Notice two things.

First, the trace distinguishes "not allowed" from "not needed." When a public, cost-priority request routes to a cheap model, the premium models appear in eligible_ranked — they were permitted, just out-scored. When a restricted request rejects an external model, it appears in rejected_models with the exact rule it broke. Those are different answers to "why not GPT?", and auditors care about the difference.

Second, the trace carries the policy version. Policies change; six months later, the audit question is "was this decision correct under the rules in force at the time?" Without versioning, that question is unanswerable.

The audit copy of the trace has one deliberate omission: the prompt. It stores a SHA-256 fingerprint and a character count — enough to prove which request a decision belongs to, without the audit log becoming a second, less-protected copy of your most sensitive data. Your audit trail should never be the leak.

Decision 3: Everything tenant-specific is data, not code

The killer question for any multi-tenant governance system: a second business unit wants different provider approvals, different regional restrictions, and different routing weights. How do you support that without business-unit-specific code paths?

If your answer involves if business_unit == "wealth-mgmt", you've already lost — every new tenant is a code change, a release, and a new place for governance bugs to hide.

In Aegis, a tenant is a policy entry:

business_units:
  retail-banking:
    approved_providers: [InHouseAI, OpenAI, Anthropic, Google]

  wealth-mgmt:
    approved_providers: [InHouseAI, Anthropic]
    weight_overrides:
      Balanced: {cost: 0.20, quality: 0.60, latency: 0.20}

Same for models — a new model is one YAML entry declaring its provider, capabilities, regions, certification ceiling, and cost/quality/latency profile. The routing core never mentions a tenant or a model by name. Onboarding either one is a config change with policy review, not a deployment.

(Confession: the first version of my demo UI hardcoded the business-unit dropdown. The routing core was clean, but the presentation layer had grown exactly the tenant-specific code path the architecture forbids. It's a good reminder that governance invariants erode from the edges — the UI, the batch job, the "temporary" script — not the core.)

The decision I'd defend hardest: no LLM in the routing loop

It's 2026, so the reflexive design is "use an LLM to pick the best model." I think that's exactly wrong for this layer, and it's worth being precise about why.

A governance gateway's routing decision must be:

Deterministic — same request, same config, same decision, every time;
Explainable rule-by-rule — "the model felt this route was compliant" is not something you can tell a regulator;
Fast and cheap — it's on the critical path of every request.

Rules and arithmetic give you all three. An LLM judge gives you none of them. In this architecture, LLMs are the workload, never the judge.

There's one legitimately hard sub-problem here: classifying the request itself (is this a text task or an image task? is this data really "Internal"?). My position: classification is an upstream contract, not a gateway guess. Applications declare task type and data classification when they integrate — it's validated, versioned, and auditable. Inferring it from prompt content would mean reading the prompt before deciding who's allowed to read the prompt, which is a privacy paradox you want no part of. If you must infer, do it inside the trust boundary, record the confidence in the trace, and never let the inference touch the governance rules — only the capability match. Bound the blast radius of a wrong guess to "suboptimal route," never "compliance breach."

What's simulated, and why that's a feature

Full disclosure: the provider calls in the repo are simulated — execute() returns canned text, and outages are injected via a test hook. That's deliberate. The simulation boundary is exactly one function; everything the project actually demonstrates — the hard gate, the ranking, the fail-closed fallback, the trace — is real, deterministic, and covered by tests, and the whole thing runs anywhere in seconds with zero API keys.

The production hardening path is the predictable one: YAML files become registry and policy services with approval workflows, the availability flag becomes health checks and circuit breakers feeding the eligibility gate, the JSONL audit file becomes an append-only store, and the business unit comes from the caller's authenticated identity instead of the request body. None of it changes the routing core — which is the point of getting the seams right.

The three sentences I'd want you to take away

Put policy before optimization, structurally — the ranker and the fallback should be physically unable to see non-compliant options, so the trade-off can't be expressed, let alone taken.
Fail closed — a refusal with reasons is a feature; a silent downgrade is an incident you'll find out about from a regulator.
Make explainability the contract — if "why this model?" isn't answerable from the response itself, with a policy version attached, you don't have governance; you have vibes.

The full implementation — routing core, Streamlit UI, FastAPI endpoint, CLI demo, and the test suite that pins the invariants — is at github.com/abhijatchaturvedi/aegis-gateway. The scenario I'd point you at first: mark every compliant model as down and watch it refuse to route while healthy external models sit right there. That refusal is the entire architecture in one response.

What's your organization's answer to "why did this request go to that model?" I'd genuinely like to hear how others are handling this — especially the task-classification contract, which I think is the least-settled part of this space.

The Two-Map Party Game Server: Building GameNight Without a Database

Abhijat Chaturvedi — Sat, 25 Jul 2026 15:21:13 +0000

Every party game app I'd used before building this one wanted an account, a lobby website, or a subscription. I wanted the opposite: plug a laptop into the TV, run one command, and have everyone's phone connected in under thirty seconds — no internet required once the LAN is up.

That constraint ends up dictating almost every architectural decision in GameNight: a Node/Express/Socket.io server that runs five real-time party games — a Mafia-style social deduction game I call Mongolpuri, UNO, a trivia quiz, Scribble, and Tic-Tac-Toe with tournament brackets — entirely from two in-memory Maps, no database, no auth, no build step on the frontend.

Decision 1: A room is a plain object, not a schema

const rooms = new Map();          // roomCode -> room
const playerRooms = new Map();    // socketId -> roomCode

const room = {
  code, gameType,
  host: socket.id,
  players: new Map([[socket.id, { id: socket.id, name, avatar }]]),
  status: 'lobby',
  gameState: null,
  timers: [],
  settings: defaultSettings(gameType),
  sessionStats: {},
};

Every game's state — the UNO deck, the Killer/Doctor night phase, the Scribble canvas buffer — lives in room.gameState, an untyped bag shaped differently per gameType. There's no ORM, no room class hierarchy, no GameEngine interface every game implements. Each game gets its own set of top-level functions (startKD, kdResolveNight, startUno, unoPlayCard, …) that read and mutate room.gameState directly, dispatched through one handleAction switch:

function handleAction(room, socket, data) {
  const gs = room.gameState; if (!gs) return;
  switch (room.gameType) {
    case 'tictactoe':    /* ... */ break;
    case 'killerdoctor': kdAction(room, socket, data); break;
    case 'scribble':     scribbleAction(room, socket, data); break;
    case 'uno':          unoAction(room, socket, data); break;
    case 'quiz':         quizAction(room, socket, data); break;
  }
}

For a five-game server built by one person, this is the right amount of abstraction: zero. A shared GameEngine interface across UNO's card-matching rules and Mongolpuri's night/day phase machine would have forced an artificial common shape onto two games that don't actually share behavior — and every future game (I'd like to add a sixth eventually) gets to invent its own state shape too, at the cost of one new switch case in four places.

Decision 2: Timers are room-scoped, not global

Every phase transition — night falling in Mongolpuri, a Scribble round ending, a quiz question closing — is a setTimeout. With five concurrent games potentially running across multiple rooms, a naive implementation leaks timers the moment a game ends early (all-but-one players disconnect, host force-restarts, tie-breaks resolve early).

function clearTimers(room) { (room.timers||[]).forEach(clearTimeout); room.timers = []; }
function addTimer(room, fn, ms) {
  if (!room.timers) room.timers = [];
  const t = setTimeout(fn, ms);
  room.timers.push(t);
  return t;
}

Every scheduled callback in the server goes through addTimer, and every phase transition that can be short-circuited (a game:restart, a host kicking a player, a win condition resolving early) calls clearTimers first. It's a two-function pattern, but it's applied with total consistency — there's no path in the codebase that calls raw setTimeout and skips the room's timer bookkeeping. That consistency is what makes it trustworthy; a timer helper only two of five games remember to use is worse than no helper at all.

Decision 3: Randomized resolution delay as a social-engineering tool

Mongolpuri (my take on Mafia/Werewolf) has a real UX problem shared by every social deduction game: if the night phase resolves the instant the last player acts, whoever acts last effectively announces "I was the one making the final decision" through pure timing. Over several rounds, alert players correlate resolution speed with who's still deciding — deanonymizing the Killer through nothing but network timing.

function checkNightDone(room) {
  const gs = room.gameState; if (gs?.phase !== 'night') return;
  if (!kdAlive(gs).every(p => p.hasActedNight)) return;
  clearTimers(room);
  const delay = 1000 + Math.random() * 4000;
  addTimer(room, () => kdResolveNight(room), delay);
}

A random 1–5 second pause between "everyone has acted" and "night resolves" breaks that correlation. It costs nothing in gameplay terms — nobody is waiting on a meaningful decision during that window — and it closes a side channel that would otherwise undermine the entire hidden-role mechanic. It's a small line, but it's the single most important line in the file for keeping the game actually fair.

Villagers also get a decoy: every living player, not just the Killer and Doctors, submits a night pick.

case 'night_kill':
  // Any living player may submit a night pick; the screen is identical for all.
  // The pick is stored per-player and only matters for the killer (kill) and
  // doctor (save) at resolution — a villager's pick is an ignored decoy.

Villagers' picks are never read at resolution time. The point isn't game logic — it's that every player's client looks and behaves identically during the night phase, so nobody can infer a role from which players even see a target-selection screen in the first place.

Decision 4: Fully declarative win detection

Doctor coverage scales with room size instead of being fixed:

const maxDoctors = Math.max(1, Math.floor(players.length / 5));
const numDoctors = Math.floor(Math.random() * maxDoctors) + 1;

Roughly one Doctor per five players, randomized within that ceiling each game, keeps the Killer-to-protection ratio sane whether four people or fifteen are playing — without a lookup table of "at N players, use M doctors" to maintain. And the whole win-condition check is two lines with no hidden state to keep in sync:

function kdCheckWin(gs) {
  const killer = kdRole(gs,'killer');
  if (!killer?.alive) return { winner: 'villagers', reason: 'The Killer has been eliminated!' };
  if (kdAlive(gs).length <= 2) return { winner: 'killer', reason: 'The Killer cannot be outvoted!' };
  return null;
}

It's called after every night resolution and every vote resolution — never cached, never tracked incrementally. Deriving it fresh from gs.playerData each time means there's no "did I forget to update the win-check after this state change" class of bug, because there's nothing to forget.

The honest gap: reconnect works by name, and one game exposes the seam

The README advertises "refresh the page and jump back in," and for four of the five games that's true because of how mid-game rejoin works:

if (room.status === 'playing') {
  const existing = [...room.players.values()].find(p => p.name === playerName?.trim());
  if (!existing) { socket.emit('room:error', { msg: 'Game in progress — cannot join now.' }); return; }
  room.players.delete(existing.id);
  if (room.host === existing.id) room.host = socket.id;
  existing.id = socket.id;
  room.players.set(socket.id, existing);
  ...
  sendReconnectState(room, socket);

There's no session token, no cookie, no persisted connection ID — Socket.io hands out a fresh socket.id on every reconnect, and GameNight re-identifies a returning player purely by matching their typed name against the room roster. room.players gets its key swapped from the old socket ID to the new one, and sendReconnectState replays whatever that game needs: the UNO hand, the Scribble canvas buffer, the quiz question in flight.

Here's the seam: for UNO, a disconnect during play evicts the player from playerOrder and deletes their hand outright —

gs.playerOrder.splice(idx, 1);
delete gs.hands[sid];

— so by the time they rejoin, there's nothing stale to reconcile; they simply re-enter as a fresh observer of unoPublic(gs). But Killer/Doctor doesn't evict on disconnect — it marks the player alive: false and leaves their gs.playerData[oldSocketId] entry in place, keyed to the old socket ID:

case 'killerdoctor':
  if (gs.playerData?.[sid]) gs.playerData[sid].alive = false;

room.players gets re-keyed to the new socket ID on rejoin, but gs.playerData never does — nothing in the file rewrites that dictionary's keys. So sendReconnectState's Killer/Doctor branch, which looks up gs.playerData?.[socket.id] using the new ID, finds nothing, and the if (pd) guard silently skips sending any reconnect payload at all. In practice this is close to moot — a disconnected player is already marked dead, so there's little "state" left worth restoring — but it's a real inconsistency between what four of the five games guarantee and what the fifth actually delivers, and it only shows up in the one game where identity (who's the Killer, who already died) matters most. I'd rather say that plainly than let the README's blanket "reconnect support" line imply more than the code backs up.

What I'd take away

Match your abstraction to your team size, not your ambition. Five games, one developer, zero shared GameEngine interface — and that's correct, not lazy, because the games don't actually share behavior worth abstracting.
A consistently-applied simple pattern beats an inconsistently-applied clever one. The addTimer/clearTimers pair isn't sophisticated; its value is that literally every timer in the file goes through it.
In games with hidden information, network timing is a side channel. A random resolution delay isn't cosmetic — it's closing a leak that would otherwise let attentive players deduce roles from server latency alone.

The full server — all five games, the tournament bracket logic, the Bonjour/mDNS LAN discovery, the settings validation — is at github.com/abhijatchaturvedi/gamenight. If you want to see the reconnect seam yourself: start a Killer/Doctor game, let the Killer's tab close mid-round, then reopen it with the same name and watch what does (and doesn't) come back.

No Backend, No Build Step: A Spaced-Repetition Chrome Extension That Runs on chrome.storage.sync Alone

Abhijat Chaturvedi — Sat, 25 Jul 2026 15:21:02 +0000

Most "save this for later" tools I've used eventually want a server: an account system, a database for your notes, a sync service with its own outage history. I wanted something narrower — capture text or a whole page while browsing, turn it into a spaced-repetition flashcard, and have it show up on my other machine — without running any infrastructure at all.

MindStack is a Manifest V3 Chrome extension that does exactly that: capture, spaced-repetition scheduling, a full dashboard, and cross-device sync, built entirely on chrome.storage.sync and chrome.identity. No backend, no bundler, no npm install before you can load it unpacked. Here's what that constraint forces you to get right.

Decision 1: The scheduler is SM-2-shaped, not SM-2

Spaced repetition apps usually reach for a full SuperMemo SM-2 implementation — ease factors computed from response quality on a 0–5 scale, per-review interval history. MindStack's actual scheduler is a compressed version that captures the two properties that matter for a lightweight capture tool and drops the rest:

const scoreReview = async (score) => {
  const memory = state.memories.find((item) => item.id === activeReviewId);
  const interval = {
    forgot: 1,
    hard: Math.max(1, Math.round((memory.reviewCount || 1) * 1.5)),
    good: Math.max(2, Math.round((memory.reviewCount || 1) * (memory.ease || 2.5))),
    easy: Math.max(4, Math.round((memory.reviewCount || 1) * ((memory.ease || 2.5) + 1)))
  }[score];

  const updated = {
    ...memory,
    reviewCount: (memory.reviewCount || 0) + 1,
    successCount: (memory.successCount || 0) + (score === "forgot" ? 0 : 1),
    ease: Math.min(3.4, Math.max(1.3,
      (memory.ease || 2.5) + ({ forgot: -0.35, hard: -0.12, good: 0.05, easy: 0.16 }[score])
    )),
    nextReviewAt: addDays(interval),
  };

Two properties, deliberately preserved from SM-2: intervals grow multiplicatively with review count (so a card you keep getting right gets reviewed exponentially less often, not linearly less often), and the ease factor is clamped to [1.3, 3.4] — the same floor SuperMemo uses, because below ~1.3 a card's interval stops growing at all and it becomes a permanent daily nuisance instead of graduating out of the queue.

What's missing on purpose: SM-2's interval formula is previous_interval × ease_factor, which requires storing the previous interval per card. MindStack derives the next interval from reviewCount instead — a proxy, not the real thing. For a browsing-habit tool where you're scheduling "did I actually retain this article" rather than cramming vocabulary for an exam, that approximation is close enough, and it means one fewer field to store, sync, and keep consistent across devices. Building the textbook version would have been the wrong amount of engineering for what this tool is actually for.

Decision 2: The account gate is a UI convention, not a backend contract

MindStack uses chrome.identity.getProfileUserInfo to read whichever Google account the browser is signed into — it never authenticates against MindStack's own servers, because there are none:

if (!isExtension() || !chrome.identity?.getProfileUserInfo) { /* preview-mode fallback */ }
chrome.identity.getProfileUserInfo({ accountStatus: "ANY" }, async (profile) => { ... });

Saving is disabled until that check reports a connected profile:

const accountConnected = (data) => Boolean(data?.account?.connected && data?.account?.email);
const connectRequiredMessage = "Connect your Google account in MindStack before saving.";

This is honest about what it actually verifies: which Chrome profile you're using, not an authenticated session with any MindStack backend, because chrome.storage.sync already keys data to that same Chrome profile automatically. The gate exists so a user never captures a dozen notes locally, only to discover afterward they were never going to sync anywhere. It's a UX guardrail against a foot-gun, not a security boundary — and I designed it as exactly that, rather than dressing it up as authentication it isn't.

Decision 3: One storage key, one deep-merge read

Every surface of the extension — the popup, the dashboard, the content-script resurfacing layer, the options page — reads and writes the exact same shape from the exact same key:

const STORE_KEY = "mindstack:data";

const readData = async () => {
  const stored = await storageArea().get(STORE_KEY);
  return {
    ...DEFAULT_DATA,
    ...(stored[STORE_KEY] || {}),
    settings: { ...DEFAULT_DATA.settings, ...((stored[STORE_KEY] || {}).settings || {}) },
    account: { ...DEFAULT_DATA.account, ...((stored[STORE_KEY] || {}).account || {}) },
  };
};

chrome.storage.sync has a 100KB total quota and an 8KB per-item limit, which rules out one storage key per memory card at any real scale of usage. A single JSON blob avoids that ceiling entirely and gives every surface (background service worker, dashboard, content script) the same read path with the same default-filling behavior — nobody has to remember to backfill settings.dailyTarget on a fresh install; readData() does it unconditionally, every time, by spreading DEFAULT_DATA underneath whatever's actually stored. New settings fields added in later versions appear for existing users automatically, without a migration step, because the defaults are re-applied on every read rather than written once at install time.

The trade-off is real too: every write round-trips the entire memory library through chrome.storage.sync, so at genuinely large card counts this stops being free. For a personal knowledge-capture tool measured in hundreds of cards, not tens of thousands, that trade-off is the right one — it buys atomic-enough writes and zero schema-migration code, which matters far more at this scale than storage efficiency does.

Decision 4: Resurfacing lives in the content script, gated by both settings and a delay

The most Chrome-extension-specific problem here: how do you remind someone of a due memory while they're browsing, without becoming the extension that nags on every page load and gets uninstalled in a week?

const maybeShowResurface = () => {
  if (!chrome.storage) return;
  area.get(key, (stored) => {
    const data = stored[key];
    if (!data?.settings?.resurfaceEnabled) return;

    const due = (data.memories || [])
      .filter((memory) => !memory.archived && new Date(memory.nextReviewAt) <= new Date())
      .sort((a, b) => new Date(a.nextReviewAt) - new Date(b.nextReviewAt));

    if (!due.length) return;
    // ...build and append the toast card
  });
};

setTimeout(maybeShowResurface, 1800);

Two small choices carry the whole feature: a per-user resurfaceEnabled setting checked before touching the DOM at all, and an 1800ms delay before the check even runs. The delay isn't decoration — it means the toast never competes with a page's own load-time layout shifts, and it means a quick tab-through (open, glance, close) never triggers it. The card shows the oldest due memory, not a random one, so the on-page nudge and the dashboard's review queue always agree on what's next — there's no separate "what to show in the toast" ranking to keep in sync with "what to show in the queue."

What I'd take away

Approximate the algorithm you actually need, not the one with the famous name. A compressed SM-2 that captures "harder answers shrink the interval, easier answers grow it multiplicatively" gets you 90% of spaced repetition's value with a fraction of the state to store and sync.
Don't let a UI convenience gate pretend to be a security boundary — but don't skip it either. chrome.identity.getProfileUserInfo proves nothing about a backend that doesn't exist; it exists purely to stop a user from capturing notes that were never going to sync.
One storage key with defaults re-applied on every read beats a schema you have to migrate. It costs some write efficiency at scale; for a personal tool, that's the correct trade.

The full extension — manifest, background service worker, content script, dashboard — is at github.com/abhijatchaturvedi/MindStack. Load it unpacked via chrome://extensions, capture a page with Alt+Shift+W, and watch the review queue pick it up on its own schedule.

Every EnvCastError Tells You How to Fix It: Designing Error Messages as a Feature

Abhijat Chaturvedi — Sat, 25 Jul 2026 15:20:47 +0000

int(os.environ.get("PORT", "8080")) fails constantly in ways that waste your time: ValueError: invalid literal for int() with base 10: 'abc'. No variable name. No hint about what a valid value looks like. You grep the codebase for PORT to even find where the read happened.

specenv is a zero-runtime-dependency Python library for typed environment variable loading — casting, validation, schema grouping, prefix namespacing. All of that is useful, but none of it is the actual design decision worth writing about. The decision that shaped everything else was: every error must name the variable and say how to fix it, unconditionally, with no opt-out.

Decision 1: The error message is generated at the failure site, not templated afterward

It would be easy to build one generic EnvCastError(var_name, raw_value, target_type) and format a message from those three fields in __str__. specenv doesn't do that — each cast failure builds its own message inline, at the point where the specific failure is known:

if cast_type is int:
    try:
        return int(raw)
    except ValueError:
        raise EnvCastError(
            f'Cannot cast {name}={raw!r} to int.\n'
            f'    → Set {name} to a valid integer (e.g. {name}=8080)'
        ) from None

if cast_type is bool:
    ...
    raise EnvCastError(
        f'Cannot cast {name}={raw!r} to bool.\n'
        f'    → Set {name} to one of: 1/0, true/false, yes/no, on/off'
    )

The generic version would produce "Cannot cast PORT='abc' to int" and stop there. The inline version gets to add (e.g. PORT=8080) for ints, 1/0, true/false, yes/no, on/off for bools, a namespaced hint for prefixed variables — because at the point of failure, you know exactly what a correct value looks like for that type, and a generic formatter three calls up the stack doesn't. The cost is a few lines of duplication across _caster.py's type branches. That's a fair trade for every single error message being genuinely actionable instead of generically accurate.

Decision 2: Missing-and-required collapses to the same error type as bad-cast

if raw is None:
    if var.required or var.default is _MISSING:
        errors.append(EnvCastError(
            f"Required variable {name} is not set.\n"
            f"    → Add {name} to your environment or .env file."
        ))

A separate EnvMissingError would be the conventional choice — missing and malformed are conceptually different failures. specenv treats them as the same EnvCastError deliberately: from the caller's perspective, both mean "I cannot produce a usable value for this variable," and code that wraps Schema.load() in a single try/except EnvCastError handles both without needing to know there are two exception types to catch. The library only introduces a second exception class, EnvValidationError, for a genuinely different failure mode — a value that cast successfully but failed a caller-supplied predicate. That's the right place to split: not "how did it fail" but "does the caller need to react differently."

Decision 3: Multi-field validation batches errors instead of failing on the first one

Schema.load() walks every declared Var field and, when several are broken at once, doesn't stop at the first failure:

errors: list[Exception] = []
for klass in cls.__mro__:
    for attr_name, val in vars(klass).items():
        ...
        try:
            value = cast_value(name, raw, var.cast_type)
        except EnvCastError as exc:
            errors.append(exc)
            continue
        ...

if errors:
    if len(errors) == 1:
        raise errors[0]
    combined = "\n".join(str(e) for e in errors)
    raise EnvCastError(f"Multiple configuration errors:\n{combined}")

Fail-on-first-error is the obvious implementation and the wrong one for this use case: environment misconfiguration on a fresh deploy is rarely a single missing variable. It's usually three or four, because someone copied a .env.example that's stale. Failing on the first one means the deploy loop is: fix PORT, redeploy, discover DB_URL is also missing, redeploy, discover TIMEOUT doesn't parse. Batching every failure into one report means one redeploy fixes everything the schema can see wrong. The single-error case still raises that error directly rather than wrapping it — no "Multiple configuration errors: " noise when there's genuinely only one thing wrong.

Decision 4: `Var` is a descriptor, and the schema instance is built with `new`, not `init`

class Var:
    def __set_name__(self, owner: type, name: str) -> None:
        self.name = name

    def __get__(self, obj: Any, objtype: type | None = None) -> Any:
        if obj is None:
            return self
        try:
            return obj.__dict__[self.name]
        except KeyError:
            if self.default is not _MISSING:
                return self.default
            raise AttributeError(self.name) from None

instance = cls.__new__(cls)
...
instance.__dict__[attr_name] = value

Using the descriptor protocol (__set_name__ / __get__ / __set__) rather than a metaclass or __init_subclass__ hook keeps Schema subclasses looking like plain dataclasses at the point of declaration — PORT = Var(int, default=8080) reads exactly like a type annotation with a default, no special base-class machinery visible in the subclass body. Building the populated instance via cls.__new__(cls) instead of calling __init__ sidesteps a chicken-and-egg problem: if Schema.__init__ tried to read the environment itself, subclasses could never add their own __init__ for other purposes, and testing with an injected env dict would mean either mutating os.environ around the constructor or threading the dict through a constructor argument that every subclass would need to forward. load() as a classmethod that builds the instance directly keeps the constructor free for subclasses to use however they want, and keeps the "where does this data come from" question answered in exactly one place.

What I'd take away

Generate the specific error at the point where you know the specific failure, not a generic one afterward. The few lines of duplication across cast branches buy every message a concrete, correct fix suggestion instead of a generic restatement of the problem.
Split exception types by how the caller needs to react, not by how the failure occurred. Missing and malformed both mean "no usable value" to a caller wrapping load() in one try/except — validation failure is genuinely different, so it gets its own type.
Batch validation errors when the realistic failure mode is "several things are wrong at once." Fail-fast is the right default almost everywhere except config loading at deploy time, where it just means more redeploy round-trips to discover the same list of problems one at a time.

The full library — casting, Schema/Var, namespacing, the test suite — is at github.com/abhijatchaturvedi/envspec, and on PyPI as pip install specenv. Try triggering three failures in the same schema at once — a missing required var, a bad int, and a validation failure — and you'll get one combined report instead of three separate deploy attempts.

Building a Timing Utility That Can't Corrupt Its Own Stats — Even When Your Code Throws

Abhijat Chaturvedi — Sat, 25 Jul 2026 15:20:15 +0000

Most ad-hoc timing code in Python looks like this:

start = time.perf_counter()
result = do_work()
elapsed = time.perf_counter() - start
stats[name].append(elapsed)

It works, until do_work() raises. Then the line that records the timing never runs, the exception propagates, and the one call that was probably slowest — the one that failed — is silently missing from your stats. If you're using timing data to find what's expensive, the failing case is exactly the one you can least afford to lose.

timerx is a small, dependency-free Python timing library — a decorator, a context manager, and named stopwatches, all backed by one stats store. The one rule that shapes the whole implementation: a timing gets recorded whether or not the timed code raised.

Decision 1: `finally`, everywhere, no exceptions to the rule

@functools.wraps(target)def wrapper(*args: Any, **kwargs: Any) -> Any:
    started = self._clock()
    try:
        return target(*args, **kwargs)
    finally:
        elapsed = self._clock() - started
        with self._lock:
            self._record(label, elapsed)
return wrapper

The async wrapper is the identical shape with await added. The context manager (_Lap) does the same thing structurally, just split across __enter__/__exit__ instead of try/finally:

def __exit__(self, *exc_info: object) -> bool:
    if self._started is None:
        raise RuntimeError("timerx lap exited before it was entered")
    elapsed = self._timer._clock() - self._started
    with self._timer._lock:
        self._timer._record(self._name, elapsed)
    return False

Note the return False — __exit__ deliberately never swallows the exception. It records the timing and lets the exception continue propagating unchanged, because a timing library has exactly one job here: observe, not intervene. A version that suppressed exceptions to "clean up" would be actively dangerous to drop into someone else's codebase.

Three entry points — decorator, context manager, stopwatch — and all three guarantee the same thing, because all three route through the same _record call inside a finally (or its structural equivalent). There's no code path in the library where a timing can be silently dropped because the timed code failed.

Decision 2: `start`/`stop` names are a stack, not a single slot

Named stopwatches need to answer a question decorators and context managers don't: what happens if you call start("x") twice before calling stop("x")? A single-slot implementation (self._running[name] = t) would let the second start silently overwrite the first — the first call's start time is gone, and its eventual stop measures the wrong duration.

def start(self, name: str) -> None:
    t = self._clock()
    with self._lock:
        self._running.setdefault(name, []).append(t)

def stop(self, name: str) -> float:
    t = self._clock()
    try:
        with self._lock:
            stack = self._running[name]
            started = stack.pop()
            if not stack:
                del self._running[name]
    except KeyError as exc:
        raise KeyError(f"timerx stopwatch {name!r} was not started") from exc
    elapsed = t - started
    with self._lock:
        self._record(name, elapsed)
    return elapsed

_running[name] is a list used as a stack: each start pushes, each stop pops the most recent start. This means recursive or re-entrant timing under the same name — nested retries, a recursive function you're profiling — behaves correctly without the caller having to invent unique names for each nesting level. It also means stop() on a name that was never started raises KeyError with a clear message rather than a TypeError from popping None, or worse, silently returning a garbage duration.

Decision 3: Global convenience functions and isolated instances share one implementation

class TimerX:
    def __init__(self, clock: Callable[[], float] | None = None) -> None:
        self._clock = clock or time.perf_counter
        self._records: dict[str, _Record] = {}
        self._running: dict[str, list[float]] = {}
        self._lock = threading.Lock()

The module-level timerx.track, timerx.lap, timerx.start, timerx.stop that most users reach for are just methods on one process-wide TimerX() instance created at import time. from timerx import TimerX; tx = TimerX() gives you a completely separate instance with its own _records, its own _running stack, its own lock.

This matters for exactly one reason: a library that imports timerx to profile its own internals shouldn't pollute the timing stats of the application that imports the same library. If timerx only exposed the global functions, every consumer would share one namespace, and a library's @timerx.track on some internal helper would show up in the application's timerx.summary() output next to the application's own timings, with no way to separate them. Building the instance-based API first and deriving the global convenience functions from a single default instance — rather than building the global API first and bolting on isolation later — is what makes that separation free instead of an afterward patch.

Decision 4: `_clock` is injectable, and that's what makes the test suite honest

def __init__(self, clock: Callable[[], float] | None = None) -> None:
    self._clock = clock or time.perf_counter

Every place that needs "now" calls self._clock(), never time.perf_counter() directly. That single seam is what lets the 33-test suite assert exact elapsed durations — tx = TimerX(clock=fake_clock) where fake_clock returns a scripted sequence of values — instead of asserting "elapsed is roughly 0.2 seconds, give or take scheduler jitter" and occasionally flaking in CI under load. A timing library whose own test suite has to tolerate timing noise to test its timing logic would be a bad sign; injecting the clock sidesteps that entirely.

What I'd take away

A timing library's one non-negotiable invariant is "always record, even on failure." Once you decide that, try/finally (or its __enter__/__exit__ equivalent) isn't an implementation detail — it's the whole point, and it should appear identically in every entry point, not just the common-case one.
Model "start twice, stop twice" as a stack from the start. Retrofitting stack semantics onto a single-slot dict later is a breaking API change; building it as a stack from the first line costs nothing and quietly supports recursion nobody explicitly designed for.
Make time itself an injectable dependency. It's a few characters at the constructor, and it's the difference between a test suite that asserts exact values and one that asserts approximate ranges and hopes the CI runner isn't busy.

The full library — decorator, context manager, stopwatches, formatted summaries, isolated instances — is at github.com/abhijatchaturvedi/timerx, and on PyPI as pip install timerx. Try decorating a function that raises on every third call and check timerx.get_stats() afterward — the failed calls are all still there.

DEV Community: Abhijat Chaturvedi

Fail Closed, Not Open: Designing an AI Gateway for Regulated Enterprises

The failure mode that matters: silent governance downgrade

Decision 1: Policy is a hard gate, ranking is soft optimization — and the order is the architecture

Decision 2: Explainability is an output, not a log line

Decision 3: Everything tenant-specific is data, not code

The decision I'd defend hardest: no LLM in the routing loop

What's simulated, and why that's a feature

The three sentences I'd want you to take away

The Two-Map Party Game Server: Building GameNight Without a Database

Decision 1: A room is a plain object, not a schema

Decision 2: Timers are room-scoped, not global

Decision 3: Randomized resolution delay as a social-engineering tool

Decision 4: Fully declarative win detection

The honest gap: reconnect works by name, and one game exposes the seam

What I'd take away

No Backend, No Build Step: A Spaced-Repetition Chrome Extension That Runs on chrome.storage.sync Alone

Decision 1: The scheduler is SM-2-shaped, not SM-2

Decision 2: The account gate is a UI convention, not a backend contract

Decision 3: One storage key, one deep-merge read

Decision 4: Resurfacing lives in the content script, gated by both settings and a delay

What I'd take away

Every EnvCastError Tells You How to Fix It: Designing Error Messages as a Feature

Decision 1: The error message is generated at the failure site, not templated afterward

Decision 2: Missing-and-required collapses to the same error type as bad-cast

Decision 3: Multi-field validation batches errors instead of failing on the first one

Decision 4: Var is a descriptor, and the schema instance is built with __new__, not __init__

What I'd take away

Building a Timing Utility That Can't Corrupt Its Own Stats — Even When Your Code Throws

Decision 1: finally, everywhere, no exceptions to the rule

Decision 2: start/stop names are a stack, not a single slot

Decision 3: Global convenience functions and isolated instances share one implementation

Decision 4: _clock is injectable, and that's what makes the test suite honest

What I'd take away

Decision 4: `Var` is a descriptor, and the schema instance is built with `new`, not `init`

Decision 1: `finally`, everywhere, no exceptions to the rule

Decision 2: `start`/`stop` names are a stack, not a single slot

Decision 4: `_clock` is injectable, and that's what makes the test suite honest