Building an Autonomous AI Operator That Runs a Business While You Sleep

Most "AI automation" is a Zapier workflow with a ChatGPT step in the middle. It fires when triggered, does one thing, and stops. If the Shopify webhook breaks, nobody notices until a customer complains. If inventory changes overnight, the marketing content is stale by morning.

I wanted something different: an AI agent that operates like a junior employee with a checklist. Every hour, it wakes up, checks every system the business depends on, identifies what needs attention, takes action on what it can, and escalates what it cannot. No triggers, no webhooks, no human in the loop unless something is genuinely wrong.

This is how I built the autonomous operator running LuxuriousComputers.com — a system that has been health-checking infrastructure, analyzing inventory for marketing opportunities, generating social content, and posting to Facebook, all on a scheduled loop, for months.

The Architecture: Three Layers

The system has three layers, each building on the one below it:

1. Health Check — probes every endpoint the business depends on: storefront, product API, chat agent, search listings, SSL certs. Produces a structured brief with pass/fail status and response times.
2. Growth Analyzer — reads the product catalog and identifies marketing opportunities: items on deep discount, categories with thin inventory, products missing photos, listings with stale descriptions.
3. Action Pipeline — takes the growth analyzer's output and does something with it: queues social media posts, generates video content, pushes alerts to the owner's phone.

Each layer is a pure-stdlib Python script. No frameworks, no dependencies, no virtualenvs to break. They run on a Windows VM inside Proxmox, scheduled as Windows Task Scheduler jobs, and survive reboots because the VM is set to auto-start with the hypervisor.

Layer 1: The Health Check

The health check script (ceo_daily.py) runs every hour during business hours. It probes five endpoints and produces a structured JSON brief:

import json, urllib.request, time

STOREFRONT = "https://luxuriouscomputers.com/"
PRODUCTS   = "https://luxuriouscomputers.com/products.json?limit=250"
CHAT       = "https://lc-next.pages.dev/api/chat"
LISTINGS   = "https://lc-next.pages.dev/api/listings"

# Cloudflare returns 403 "error code 1010" to the default
# python-urllib User-Agent. A browser UA passes.
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
      "AppleWebKit/537.36 Chrome/126.0 Safari/537.36")

def _get(url, timeout=30, data=None, headers=None):
    """Return (status, body, elapsed_seconds).
    Status 0 = connection failure."""
    t0 = time.time()
    h = {"User-Agent": UA}
    if headers:
        h.update(headers)
    req = urllib.request.Request(url, data=data, headers=h)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read().decode("utf-8", "replace")
            return r.status, body, round(time.time() - t0, 1)
    except urllib.error.HTTPError as e:
        return e.code, "", round(time.time() - t0, 1)
    except Exception as e:
        return 0, str(e), round(time.time() - t0, 1)

Each probe is its own function that returns a dict with an ok boolean and diagnostic data:

def check_storefront():
    st, _, el = _get(STOREFRONT, timeout=25)
    return {"ok": st == 200, "status": st, "elapsed_s": el}

def check_products():
    st, body, el = _get(PRODUCTS, timeout=25)
    count = None
    if st == 200:
        try:
            count = len(json.loads(body).get("products", []))
        except Exception:
            count = -1
    return {"ok": st == 200 and (count or 0) > 0,
            "status": st, "count": count, "elapsed_s": el}

The chat endpoint is the trickiest. It is a streaming SSE endpoint with cold starts that can exceed 50 seconds. Naive timeout-based checking would false-alarm every time the Cloudflare Worker cold-starts. Instead, the probe reads the stream incrementally and passes the instant the first delta arrives:

def check_chat():
    """SSE endpoint — cold start can exceed 50s.
    Read incrementally, pass on first delta."""
    payload = json.dumps({
        "messages": [{"role": "user", "content": "hi"}]
    }).encode()
    hdrs = {"Content-Type": "application/json",
            "User-Agent": UA}
    for attempt in (1, 2):
        t0 = time.time()
        req = urllib.request.Request(
            CHAT, data=payload, headers=hdrs)
        try:
            with urllib.request.urlopen(req, timeout=70) as r:
                for _ in range(200):
                    chunk = r.read(256).decode("utf-8", "replace")
                    if not chunk:
                        break
                    if "delta" in chunk:
                        return {"ok": True,
                                "elapsed_s": round(time.time()-t0, 1),
                                "attempt": attempt}
        except Exception:
            pass
    return {"ok": False, "elapsed_s": 0, "attempt": 2}

Two attempts: the first pays the cold-start cost, the second runs warm. This eliminates the most common false alarm in health checking serverless AI endpoints.

Structured Logging and State File

Every run appends to logs/health.jsonl — one JSON line per check, timestamped, with every probe result. But the real power is the state file: a Markdown document (BUSINESS_STATE.md) that the script overwrites on every run. The top half is hand-authored policy — KPIs, standing decisions, editing rules. The bottom half (below a ## LIVE STATE marker) is auto-updated by the health check.

This design means a fresh AI instance — one that has never seen the business before — can read a single file and know: what the business is, what its rules are, what its current health is, and what happened on the last check. No conversation history needed.

def update_state_file(brief):
    """Overwrite LIVE STATE section, preserve policy above."""
    with open(STATE_FILE, "r") as f:
        full = f.read()
    marker = "## LIVE STATE"
    idx = full.find(marker)
    if idx < 0:
        return  # no marker — don't corrupt
    policy = full[:idx]
    live = f"""{marker}
_Auto-updated {brief['ts']} by ceo_daily.py_

### System Health
| Endpoint   | Status | Latency |
|------------|--------|---------|
| Storefront | {'UP' if brief['storefront']['ok'] else 'DOWN'} | {brief['storefront']['elapsed_s']}s |
| Products   | {'UP' if brief['products']['ok'] else 'DOWN'} ({brief['products']['count']} items) | {brief['products']['elapsed_s']}s |
| Chat Agent | {'UP' if brief['chat']['ok'] else 'DOWN'} | {brief['chat']['elapsed_s']}s |
"""
    with open(STATE_FILE, "w") as f:
        f.write(policy + live)

Push Notifications via ntfy

Every health check pushes a one-line summary to ntfy, a self-hosted push notification service. I run my own ntfy instance behind a Cloudflare tunnel, with a public ntfy.sh fallback:

def push_summary(brief, topic="lc-ceo-brandon"):
    """Push one-line summary to ntfy."""
    status = "ALL GREEN" if brief["all_ok"] else "ISSUES FOUND"
    body = f"CEO Check: {status} | " \
           f"Store={'UP' if brief['storefront']['ok'] else 'DOWN'} | " \
           f"Products={brief['products']['count']} | " \
           f"Chat={'UP' if brief['chat']['ok'] else 'DOWN'}"
    for base in ["https://ntfy.beamvideos.com",
                 "https://ntfy.sh"]:
        try:
            req = urllib.request.Request(
                f"{base}/{topic}",
                data=body.encode(),
                headers={
                    "Title": f"LC CEO: {status}",
                    "Priority": "4" if not brief["all_ok"] else "3",
                    "Tags": "warning" if not brief["all_ok"] else "white_check_mark"
                })
            urllib.request.urlopen(req, timeout=10)
            return True
        except Exception:
            continue
    return False

The self-hosted instance is the primary (faster, no rate limits). ntfy.sh is the fallback. If the storefront is down, the priority bumps to urgent and the phone buzzes immediately.

Layer 2: The Growth Analyzer

The growth analyzer (ceo_growth.py) runs after the health check passes. It reads the Shopify product catalog and classifies every item into opportunity tiers:

• P1 — Revenue at risk: items listed as in-stock but missing photos, products with broken compare-at prices that trigger Google Merchant Center disapprovals, items marked available but returning 404 on their product page.
• P2 — Quality and compliance: products with thin descriptions (under 50 words), missing alt text on images, categories with only one or two SKUs.
• P3 — Marketing opportunities: items on deep discount (15%+ off compare-at price), seasonal relevance, new arrivals that have not been promoted yet.

def analyze_catalog(products):
    """Classify every product into opportunity tiers."""
    opportunities = {"p1": [], "p2": [], "p3": []}

    for p in products:
        price = float(p.get("variants", [{}])[0]
                       .get("price", "0"))
        compare = float(p.get("variants", [{}])[0]
                         .get("compare_at_price") or "0")
        has_image = bool(p.get("images"))
        desc_len = len(p.get("body_html", "").split())
        available = p.get("status") == "active"

        # P1: revenue-at-risk
        if available and not has_image:
            opportunities["p1"].append({
                "product": p["title"],
                "handle": p["handle"],
                "issue": "NO_PHOTO",
                "impact": "high"
            })
        if compare > 0 and price >= compare:
            opportunities["p1"].append({
                "product": p["title"],
                "handle": p["handle"],
                "issue": "COMPARE_AT_INVALID",
                "impact": "high"  # GMC misrep risk
            })

        # P2: quality
        if desc_len < 50 and available:
            opportunities["p2"].append({
                "product": p["title"],
                "issue": "THIN_DESCRIPTION",
                "word_count": desc_len
            })

        # P3: marketing
        if compare > 0 and price < compare:
            discount_pct = round(
                (1 - price / compare) * 100, 1)
            if discount_pct >= 15:
                opportunities["p3"].append({
                    "product": p["title"],
                    "handle": p["handle"],
                    "price": price,
                    "compare_at": compare,
                    "discount_pct": discount_pct,
                    "has_image": has_image,
                    "url": f"https://luxuriouscomputers.com/"
                           f"products/{p['handle']}"
                })

The output is a ranked opportunity list. P3 items with images and deep discounts feed directly into the action pipeline.

The Promo Queue

The growth analyzer does not just report — it acts. The build_promo_queue() function filters P3 opportunities to items that are in-stock, have a product photo, and have a discount worth promoting. It writes a ready-to-schedule queue file:

def build_promo_queue(opportunities, output_path):
    """Write a promo queue for the Reels pipeline."""
    candidates = [
        opp for opp in opportunities["p3"]
        if opp["has_image"] and opp["discount_pct"] >= 15
    ]
    # Sort by discount depth — deepest deals first
    candidates.sort(key=lambda x: -x["discount_pct"])

    queue = []
    for c in candidates[:20]:  # cap at 20 pending reels
        queue.append({
            "product": c["product"],
            "handle": c["handle"],
            "url": c["url"],
            "price": c["price"],
            "compare_at": c["compare_at"],
            "save": round(c["compare_at"] - c["price"], 2),
            "pct": c["discount_pct"],
            "source_image": f"https://luxuriouscomputers.com/"
                            f"products/{c['handle']}",
            "status": "queued"
        })

    with open(output_path, "w") as f:
        json.dump(queue, f, indent=2)
    return len(queue)

The first time I ran this on the live catalog, it found 48 items at 15%+ discount, including an Apple TV at 60% off and a MacBook Pro M1 Pro at 56% off ($1,120 savings). These were products already on the storefront that nobody was actively promoting — found money.

Layer 3: The Reels Pipeline

The action pipeline turns the promo queue into Facebook Reels. This is the most complex layer because it involves video generation, Facebook's Graph API, and rate limiting.

Video Generation

Each Reel is a short promotional video: the product image with price overlays, animated text, and a call to action. The render pipeline runs on a local GPU (RTX 3060) using a video generation model. The rendered MP4s land in a renders/ directory, named by queue position.

The Post Scheduler

The scheduler (post_scheduler.py) is the bridge between rendered videos and Facebook. It does two things:

1. Publishes due reels — any reel that is rendered, due, and not yet posted gets published via the Graph API, capped at 2 per day to avoid spam flags.
2. Fills the scheduled queue — queues upcoming reels into Facebook's native scheduled publishing system, up to a ceiling of 56 pending, spread across the next 28 days.

SCHED_PER_RUN = 2   # reels to schedule per execution
QUEUE_CEILING = 56  # max pending in FB's queue
DAILY_POST_CAP = 2  # max published per calendar day

def main():
    token = load_page_token()
    posted_today = count_posts_today(token)
    budget = max(0, DAILY_POST_CAP - posted_today)

    # Phase 1: publish due reels (oldest first)
    due = get_due_rendered_unposted()
    for reel in due[:budget]:
        video_id = upload_and_publish(reel, token)
        if video_id:
            verify_published(video_id, token)
            mark_posted(reel, video_id)

    # Phase 2: fill scheduled queue
    pending = count_pending_scheduled(token)
    slots_available = QUEUE_CEILING - pending
    if slots_available > 0:
        unscheduled = get_rendered_unscheduled()
        for reel in unscheduled[:min(SCHED_PER_RUN,
                                      slots_available)]:
            schedule_future_slot(reel, token)

Rate Limiting and Spam Protection

Facebook's spam detection is aggressive. Error code 368 means "you are posting too fast" — and once triggered, continuing to post makes it worse. The scheduler implements a cooldown mechanism:

COOLDOWN_FILE = "logs/.fb_cooldown"
COOLDOWN_HOURS = 6

def check_cooldown():
    """Return True if we're still in cooldown."""
    if not os.path.exists(COOLDOWN_FILE):
        return False
    try:
        stamp = float(open(COOLDOWN_FILE).read().strip())
        if time.time() - stamp < COOLDOWN_HOURS * 3600:
            return True
        os.remove(COOLDOWN_FILE)
    except Exception:
        pass
    return False

def trigger_cooldown():
    """Set a cooldown after error 368."""
    with open(COOLDOWN_FILE, "w") as f:
        f.write(str(time.time()))

def upload_and_publish(reel, token):
    """Upload video and publish. Handle spam blocks."""
    # ... upload logic ...
    if response.get("error", {}).get("code") == 368:
        trigger_cooldown()
        return None
    # ...

When the 368 fires, the scheduler writes a timestamp file and skips all posting for 6 hours. The next scheduled run checks the file, sees the cooldown is active, and exits immediately. No retry loops, no exponential backoff — just a hard stop, because Facebook's spam detector penalizes rapid retries harder than it penalizes waiting.

Verification by Observation

One critical lesson: never trust the local log. After publishing a reel, the scheduler verifies it is actually live by querying the Graph API:

def verify_published(video_id, token):
    """Verify a reel is actually live on Facebook."""
    url = (f"https://graph.facebook.com/v20.0/"
           f"{video_id}?fields=status"
           f"&access_token={token}")
    st, body, _ = _get(url, timeout=15)
    if st == 200:
        data = json.loads(body)
        status = data.get("status", {})
        return status.get("video_status") == "ready" \
            and status.get("publishing_status") in \
                ("published", "scheduled")
    return False

This caught a real bug: videos that the local upload call reported as successful but that Facebook silently dropped because they exceeded the resolution or duration limits for Reels. Without verification, those would have been logged as "posted" and never noticed.

The Scheduling Infrastructure

The whole system runs on Windows Task Scheduler inside a Proxmox VM. This sounds low-tech, and it is — intentionally. The VM auto-starts with the hypervisor (onboot=1), so a power outage recovers automatically.

# Task configuration (via schtasks)
# CEO health check: hourly, 8am-10pm
schtasks /create /tn "LC-CEO-Hourly" \
  /tr "python C:\lc-ceo\ceo_daily.py" \
  /sc hourly /st 08:00 /et 22:00 \
  /ru SYSTEM /rl HIGHEST

# Reels scheduler: hourly, same window
schtasks /create /tn "LC-Reels-Hourly" \
  /tr "python C:\lc-reels\post_scheduler.py" \
  /sc hourly /st 08:00 /et 22:00 \
  /ru SYSTEM /rl HIGHEST

The SYSTEM Context Gotcha

Running as SYSTEM on Windows means ~ resolves to C:\WINDOWS\system32\config\systemprofile, not the user's home directory. Every file path that uses ~ — including the Facebook access token — needs to be duplicated to the SYSTEM profile, or the script silently fails with "token not found."

This bit me on day one. The health check ran perfectly interactively but produced empty results when Task Scheduler fired it. The fix was storing all config in absolute paths and copying credentials to both locations.

Surviving Memory Loss

The most interesting design constraint: the AI operator has no persistent memory. Each hourly run is a fresh instance. It does not know what it did last hour, what it decided yesterday, or what the business looked like a week ago.

This is a feature, not a bug. Stateless agents cannot accumulate drift. They cannot develop cargo-cult habits from a lucky success three weeks ago. Every decision is made from current data.

But you still need continuity. The solution is the decision journal — a Markdown file (logs/decisions.md) that the agent appends to at the end of every session:

### 2026-09-10 08:30 — HOURLY CHECK
- **Health:** ALL GREEN (storefront 200 in 1.2s,
  products 247 items, chat UP in 8.3s)
- **Growth:** 48 items ≥15% off, top=Apple TV 60%
- **Action:** queued 2 reels (reel_0042 MacBook Air,
  reel_0043 AirPods Pro)
- **Escalation:** none

### 2026-09-10 09:30 — HOURLY CHECK
- **Health:** ALL GREEN
- **Growth:** no change
- **Action:** 0 reels (2/day cap reached)
- **Escalation:** none

A fresh instance reads this file and knows exactly what happened — not from memory, but from the audit trail. It can see that the daily post cap was hit at 9:30, that no escalation was needed, and that the growth opportunities are unchanged.

The BUSINESS_STATE.md file serves the same purpose for strategic context: a hand-authored top half (business rules, KPIs, standing decisions that the agent must not re-litigate) and a machine-updated bottom half (current health, recent decisions). One file, complete situational awareness.

What Goes Wrong (And How the System Handles It)

The Chat Agent Dies

The most common failure is the AI chat agent going down. This has happened three times in production, each with a different root cause:

1. OpenRouter credits exhausted — the paid model returned 402, and the free fallback models had been retired by OpenRouter (404). The health check caught this within one hour and pushed an urgent notification.
2. Cold start timeout — a Cloudflare Worker cold start took 62 seconds, exceeding the previous 30-second timeout. The two-attempt probe pattern now handles this.
3. Reasoning model leak — a fallback model leaked chain-of-thought to customers (<think>Let me craft a response...</think>). The fix was a streaming sanitizer that strips reasoning tags in real-time, including tags split across SSE deltas.

In every case, the autonomous operator detected the issue before any customer reported it. The mean time to detection was under 60 minutes — the length of one scheduled cycle.

Facebook Token Expiration

Facebook page tokens expire. When they do, every post call fails silently (the Graph API returns a helpful 190 error, but only if you check). The scheduler now validates the token on startup and pushes an alert if it is expired or expiring within 24 hours.

KV Quota Exhaustion

Cloudflare's free-tier KV has write quotas that reset at midnight UTC. During high-traffic periods, the lead capture APIs were crashing with HTTP 1101 (unhandled exception) when KV writes threw on quota exhaustion — silently losing real leads. The fix: wrap every KV write in try/catch, gracefully degrade (still send notifications, just skip storage), and log the quota hit.

This same bug existed across five different APIs in the fleet. The autonomous operator's health check caught it on one site, and the fix was propagated to all five in the same session.

Cost Analysis

The autonomous operator runs on a Windows VM that is already paid for (home Proxmox server). Incremental costs:

• Compute: $0/month — VM uses 2 cores and 4GB RAM of existing server capacity
• ntfy notifications: $0/month — self-hosted on existing infrastructure
• Video generation: $0/month — renders on local RTX 3060 GPU
• Facebook API: $0/month — Graph API is free for page owners
• Cloudflare Workers/KV: $0/month — within free tier
• LLM API calls: Variable — the chat agent uses OpenRouter; the autonomous operator itself is pure Python with zero LLM calls

Total incremental cost for the autonomous operator: $0/month. The health check, growth analysis, promo queue, and scheduling pipeline are all deterministic Python — no AI inference required. AI is in the chat agent and video generation, not in the decision loop itself.

This is a deliberate design choice. The operator's decisions are simple enough to codify in rules: "if discount > 15% and has_image, queue for promotion." The AI lives at the edges (customer conversations, content generation) where rules cannot capture the complexity.

What I Would Do Differently

Move to Linux

Running on Windows Task Scheduler works but creates friction: the SYSTEM context gotcha, PowerShell encoding issues when driving the VM remotely, and the Windows-specific file path handling. A Linux cron job with systemd timers would be cleaner. The reason it is on Windows is path dependency — the VM was cloned from an existing Windows desktop setup, and it works, so migrating is low priority.

Add an LLM Decision Layer

The current system is rule-based: fixed thresholds for discount depth, hard caps on daily posts, deterministic priority tiers. An LLM layer could make smarter decisions — "this product just went viral on TikTok, promote it now even though the discount is only 12%" — but the reliability tradeoff is not worth it yet. Rules are debuggable. LLM decisions are not.

When I do add it, the architecture is ready: the growth analyzer produces structured data that an LLM could reason over, and the action pipeline accepts a queue file in a known format. The LLM would sit between analysis and action, replacing the threshold filter with contextual judgment.

Better Observability

JSONL logs are fine for debugging, but they are not queryable. A lightweight time-series database (InfluxDB or even SQLite with timestamps) would let me answer questions like "how many times did the chat agent go down this month?" or "what is the average discount depth of promoted items?" without grep.

The Pattern for Your Business

The specific implementation — Shopify products, Facebook Reels, Cloudflare Workers — is particular to this business. The pattern is universal:

1. Health check layer: probe every endpoint your business depends on. Output a structured brief. Push alerts on failure. This layer takes a day to build and saves you from finding out your site is down from a customer email.
2. Analysis layer: read your data and surface opportunities. You already have the data (inventory, analytics, CRM) — you just need a script that reads it on a schedule and tells you what is interesting.
3. Action layer: automate the obvious actions. Queue social posts. Send digest emails. Update pricing. Start small — one action per cycle — and expand as you build trust in the system.
4. State file: write a single document that a new operator (human or AI) can read and immediately understand the business. Policy at the top, live state at the bottom. This is the glue that makes stateless execution work.

The whole thing runs on pure Python, stdlib only, no dependencies. It deploys to any machine that has Python installed. It costs nothing to operate. And it catches problems faster than a human checking dashboards — because it checks every hour, every day, without getting bored or distracted.

That is the bar for autonomous AI operations: not replacing human judgment, but automating the checklist that humans are bad at following consistently. The AI is the discipline, not the intelligence.