DEV Community

Riley Zhang
Riley Zhang

Posted on

Weekend Build Log: A Human Gate Between the Model and Your Database

Your side project watches a support inbox. One Friday night the model decides a paying customer should be downgraded. The write lands. Nobody approved it, and the logs do not say which prompt caused it.

That is the failure mode this build targets. Not model quality. Not latency. Unbounded write access from a component that cannot explain itself.

Below is the smallest version that works: a pending-actions table, a proposal worker, an approval CLI, and an idempotent applier. You can finish it in a weekend. The code is a runnable artifact, but it was not executed against a live endpoint while writing this post. Treat the test plan as your first job, not mine.

1. The scope cut comes first

You will be tempted to build a dashboard, roles, and Slack approvals. Do not. Every one of those is a second weekend, and none of them stop a bad write.

The gate needs three properties only. Proposals are stored, not executed. A human flips the status. The applier refuses to run twice.

Here is the cut I would make, written as a table you can argue with:

Option Effort Stops a bad write Skip it when
Human gate + audit row 1 weekend Yes, for allowlisted actions Nothing — start here
Dry-run diff in the CLI +2 hours Makes review fast Payloads are not human-readable
Full dashboard +2 weekends Nothing extra You have more than 3 reviewers
Auto-approve below a threshold +1 weekend Weakens the whole point You have real eval coverage

Pick the first two rows. Everything else is a later problem.

2. Step one: a table that remembers intent

SQLite is enough for a side project. The key design choice is the primary key: it is a hash of the action, not a random UUID.

That hash is what makes replay harmless.

CREATE TABLE IF NOT EXISTS proposals (
  id           TEXT PRIMARY KEY,          -- sha256(source_id, action_type, payload)
  source_id    TEXT NOT NULL,             -- the ticket, email, or job this came from
  action_type  TEXT NOT NULL,             -- must be on the allowlist
  payload      TEXT NOT NULL,             -- canonical JSON
  status       TEXT NOT NULL DEFAULT 'pending',
  created_at   TEXT NOT NULL,
  decided_by   TEXT,
  decided_at   TEXT
);

CREATE INDEX IF NOT EXISTS idx_pending
  ON proposals(status, created_at);
Enter fullscreen mode Exit fullscreen mode

Status values are fixed: pending, approved, rejected, applied, failed. Add a sixth only if you can name the state transition it represents.

3. Step two: the proposal worker never writes to your app tables

This worker reads a source record, asks the model for one action, validates it, and inserts a row. It has no other power. Enforce that with a database user that cannot touch the app tables.

Use an allowlist, not a blocklist. A model that invents delete_account fails validation instead of succeeding quietly.

import hashlib, json, os, sqlite3
from datetime import datetime, timezone

ALLOWED = {"set_plan_tier", "add_tag"}

DB = os.environ.get("APP_DB", "app.db")


def canonical(payload: dict) -> str:
    return json.dumps(payload, sort_keys=True, separators=(",", ":"))


def proposal_id(source_id: str, action_type: str, payload_json: str) -> str:
    raw = f"{source_id}|{action_type}|{payload_json}".encode()
    return hashlib.sha256(raw).hexdigest()[:32]


def record(source_id: str, action_type: str, payload: dict) -> str | None:
    if action_type not in ALLOWED:
        raise ValueError(f"blocked action_type: {action_type}")
    payload_json = canonical(payload)
    pid = proposal_id(source_id, action_type, payload_json)
    now = datetime.now(timezone.utc).isoformat()
    with sqlite3.connect(DB) as con:
        cur = con.execute(
            "INSERT OR IGNORE INTO proposals "
            "(id, source_id, action_type, payload, created_at) "
            "VALUES (?, ?, ?, ?, ?)",
            (pid, source_id, action_type, payload_json, now),
        )
        return pid if cur.rowcount == 1 else None
Enter fullscreen mode Exit fullscreen mode

INSERT OR IGNORE plus the hash means a retried job is a no-op. That single line saves you from duplicate approvals later.

4. Step three: parse strictly, fail loudly

The model returns JSON. Sometimes it returns JSON with a friendly sentence attached. Reject it and log the raw text; do not regex-repair it.

def parse_action(raw: str) -> tuple[str, dict]:
    doc = json.loads(raw)                      # raises, by design
    action_type = doc["action_type"]
    payload = doc["payload"]
    if not isinstance(payload, dict):
        raise ValueError("payload must be an object")
    if action_type == "set_plan_tier":
        if payload.get("tier") not in {"free", "pro", "team"}:
            raise ValueError("unknown tier")
        if not isinstance(payload.get("account_id"), str):
            raise ValueError("account_id must be a string")
    return action_type, payload
Enter fullscreen mode Exit fullscreen mode

One repair attempt is reasonable. Two is a sign your prompt is wrong. Zero is better than three.

5. Step four: the approval CLI is the product

Reviewers should not open a database browser. Give them four commands and no more.

python gate.py list --limit 20
python gate.py show 4f2a9c1e
python gate.py approve 4f2a9c1e --as riley
python gate.py reject 4f2a9c1e --as riley --note "wrong account"
Enter fullscreen mode Exit fullscreen mode

show prints the source record next to the proposed payload. A reviewer who cannot see the input cannot judge the output. Keep both in the same screen.

The approve path must be a conditional update, not a read-then-write.

def approve(pid: str, who: str) -> bool:
    now = datetime.now(timezone.utc).isoformat()
    with sqlite3.connect(DB) as con:
        cur = con.execute(
            "UPDATE proposals SET status='approved', decided_by=?, decided_at=? "
            "WHERE id=? AND status='pending'",
            (who, now, pid),
        )
        return cur.rowcount == 1
Enter fullscreen mode Exit fullscreen mode

Two reviewers clicking at once: one wins, one gets False. That is the correct outcome.

6. Step five: the applier is boring on purpose

Write the applier as a separate process. It claims approved rows, applies them in a transaction, and marks them applied. If the app fails, it marks them failed and leaves the payload intact for inspection.

BEGIN IMMEDIATE;
UPDATE proposals
   SET status = 'applied'
 WHERE id = :pid AND status = 'approved';
-- if changes() = 0, roll back: someone else got here first
COMMIT;
Enter fullscreen mode Exit fullscreen mode

Apply and mark inside the same transaction. An external API call in the middle breaks that guarantee, so put the external call after the commit and record its result separately.

7. Where the free tier actually fits

The proposal worker is a background job with bursty load. It idles most of the day and processes a few dozen records at night. That shape is a good fit for a free tier, because a fixed allowance is easy to reason about.

For this build I would point the worker at MonkeyCode's free model access and run the scheduler on its free server option. The outreach brief I received quotes a 10M-token allowance and a free server tier; check the current terms yourself before you depend on either. Point three environment variables at it and the rest of the code does not change:

MODEL_BASE_URL=...   # from the provider's docs
MODEL_API_KEY=...    # keep it out of git
MODEL_NAME=...       # whatever the free tier currently serves
Enter fullscreen mode Exit fullscreen mode

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

One practical rule: log tokens per proposal, not per day. The daily number hides the single record that cost 40x the median. You want that outlier visible the same night.

8. The test plan you run before trusting it

Five cases. Each is a few lines of pytest and none need a network call.

  1. Duplicate insert. Call record() twice with the same input. Assert one row and one None.
  2. Blocked action. Pass delete_account. Assert ValueError and zero rows.
  3. Malformed JSON. Feed Sure! Here you go: {...}. Assert json.JSONDecodeError and no insert.
  4. Double approve. Approve the same id twice. Assert True then False.
  5. Rejected is terminal. Try to apply a rejected row. Assert zero status change.

If case five fails, your conditional update is wrong. Fix it before you deploy.

9. What this build does not do

The gate does not make the model correct. It only makes wrong output cheap to catch. A reviewer who rubber-stamps twenty rows at midnight gives you no protection at all.

It also does not rate-limit the source. If your inbox produces 5,000 records a day, a human queue is the wrong architecture and you need sampling plus eval coverage instead. And it does not handle regulated data on its own; the payload column stores plain text, so redact before insert if that matters to you.

Skip this pattern if you need sub-second writes, or if no human can plausibly review the volume. It is for side projects where a wrong write costs more than a delayed one.

10. Ship it small

The whole thing is two tables, four CLI commands, and about 150 lines of Python. That is the point. A gate you finish beats a platform you abandon in week three.

Start with one allowlisted action. Add the second only after the first has survived a week of real review. If you want a free place to run the worker while you decide, the free tier mentioned above is one reasonable option, and a cron job on any small box is another.

Top comments (0)