Online shop server: what is inside

HEART for software: a shop's web server. Three Playbooks the owner adopted fix known problems within daily limits; everything else is escalated, and nothing destructive is possible. HEART runs as a sidecar on the shop's own server, beside the web workers, reporting outbound only.

Specifications it follows

• Part 2: Autonomy Levels (A1, Minimum Tending)
• Part 2: Autonomy Levels (A4, Fixes known problems: Playbooks inside an envelope)
• Part 3: Information Model (messages, fault codes)
• Part 6: Human Control and Safety (the gate, proof of presence, staying out of the safety chain)
• API Standard (the Valve, Section 4; the gate, Section 6.3)
• HEART Hub profile (claim codes, reports, Flatline)

Customers  ──►  Shop  ◄──(3 capabilities only)──  Heart  ──►  local HEART API      (127.0.0.1)
 arrivals       workers, queue, cache,             watches,    └►  HEART Hub    (outbound only)
                database, disk, payment, shipping  Playbooks

Run it yourself, and register it on HEART Hub

1. Download README.md, onlineShopSimulator.py, test_e2e.py.
2. Run python3 onlineShopSimulator.py --scenario all. It prints a device ID and a claim code.
3. Sign in, open Register a device, enter them, and watch your own copy on My devices.

Now do it for your own device and monitor it here: https://heartcortex.ai/monitor. Start from a template, follow the HEART specifications, then register your device with its claim code.

Full instructions: README · case study: Online shop server.

The code: onlineShopSimulator.py

1407 lines of Python, standard library only. The live page above replays the same design in the site's own code; this file is the version you run.

#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Template: heart/embeddedstandard/online-shop-simulator/v1.2611.2
"""onlineShopSimulator.py: a simulated online shop server with a HEART inside, that fixes known problems by itself
and reports to HEART Monitor.

A DEMONSTRATION ONLY. It is not a shop, takes no real orders and holds no customer data. Its numbers are plausible, not
measured. See README.md beside this file, and the case study guides/Online-Shop-Server.md.

HEART for software rather than hardware. What it does, in one process and with only the Python standard library:

  * simulated CUSTOMERS: requests arrive on random timers, with a daily curve (quiet at night, busy in the
    afternoon) and now and then a flash sale that brings six times as many;
  * a simulated SHOP SERVER (the product): a web tier of 8 worker processes behind a queue, a cache, a database,
    a disk that fills with logs, and two outside services it needs (a payment provider and a shipping API). Workers
    hang, the cache leaks, logs pile up and outside services go down, each on its own random timer;
  * a HEART that WATCHES the shop and, for three known problems only, FIXES it. The owner adopted three Playbooks
    ("when this happens, do that") and put each fix on the allowlist for initiator rule, with a daily limit and a
    cooldown: restart a worker stuck for 30 s, clear the cache when memory is above 90 % and the cache is mostly
    junk, rotate and compress the logs when the disk is nearly full. Each run is an Action record. A fix that does
    not help twice in a row suspends its Playbook, which then escalates instead of looping. Everything else
    (payment provider down, database slow, a flash sale) is only noticed, explained and escalated to a person.
    Destructive things (delete orders, refund, deploy a new version, change the payment provider) are never
    possible through HEART: they are refused with 403 capability-not-allowed, even with the owner key.
  * LIFECYCLE (API Standard 13.2 rule 3): the owner can drain the shop (no new orders, the running ones finish) and
    restart it. A graceful restart drains first and fails, never forced, if an order cannot finish by timeoutSec.
    A force restart is destructive: it waits for a person at the server (`touch button.press`).

This is a demonstration of Autonomy Level A4 behaviour ("fixes known problems"). The Card claims only A1, the
level of the HEART as built: a real product claims A4 only at HEART Full, after it passed HEART-Rating.

Run it:

  python3 onlineShopSimulator.py                       # report to https://heartcortex.ai
  python3 onlineShopSimulator.py --scenario all        # also stage each fault, one after another
  python3 onlineShopSimulator.py --no-monitor          # local only: http://127.0.0.1:43278/heart/v1/pulse
  python3 onlineShopSimulator.py --monitor-url http://127.0.0.1:8787   # a local monitor site (heart-monitor/test/serve.mjs)

At first start it prints a device ID and a claim code. Register them at https://heartcortex.ai/register, signed in
with your email, then watch the server at https://heartcortex.ai/device/<ID>. The ID, claim code and key are kept
in online-shop-monitor.json (mode 600), so the same simulated server comes back after a restart.

Scenarios (--scenario, comma-separated): stuck, disk, crashloop, leak, payment, flash, slowdb, all. Each fault starts
at a set time, lasts a while, and is then fixed (by a Playbook, or "by the owner"), so you can watch alerts open,
be tended or escalated, and resolve.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. Traffic,
memory, disk and every other measurement stay local (internal); orders and revenue are private; customer data is
never part of HEART at all: the HEART has no way to read it.
"""
import argparse
import hashlib
import json
import math
import os
import random
import secrets
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SYSTEM_ID = "org.heartaispec.demo.online-shop"
NAME = "Online Shop Server Simulator (demo)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/online-shop-simulator/v1.2611.2"
PULSE_SEC = 30
PROMISED_LEVEL = 4                 # the level its control offers: max 4 is mode act (API Standard 13.4)
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
EXT_SIM = "org.heartaispec.demo.sim-sec"          # Action ext: simulated time of the Action, for the limits

# The simulated server.
WORKERS, WORKER_RPS = 8, 15                       # worker processes; requests per second each can handle
QUEUE_WAIT_MAX = 10                               # seconds a request may wait in the queue before it times out
CHECKOUT_SHARE, ORDER_SHARE = 0.04, 0.01          # requests that are checkouts; requests that end in an order

# The HEART's thresholds. CHANGE for a real shop: take them from its own service levels.
STUCK_SEC = 30                                    # a worker that has not answered for this long is stuck
MEMORY_HIGH, HIT_LOW = 90, 50                     # memory used %, cache hit rate %
DISK_LOW, DISK_FULL = 10, 2                       # disk free %
SLOW_MS, BACKLOG_HIGH = 1500, 500                 # p95 handling time; queued requests

SCENARIOS = ["stuck", "disk", "crashloop", "leak", "payment", "flash", "slowdb"]
# The capabilities that are never possible through HEART, for any caller. They are not even listed in the Card.
DESTRUCTIVE = {"deleteOrders": "deleting orders", "refundOrder": "refunds", "deployVersion": "deploying a new version",
               "changePaymentProvider": "changing the payment provider"}

# Lifecycle (API Standard 13.2 rule 3): drain, and restart. Times here are real seconds, as a caller asks for them.
STANDARD_OPS = ("restart", "shutdown", "sleep", "wake", "drain", "reload", "power-cycle", "safe-mode", "maintenance-on",
                "maintenance-off", "reset-config", "reset-network", "clear-data", "factory-reset", "decommission")
LIFECYCLE_OPS = ["drain", "restart"]
LIFECYCLE_FIELDS = {"operation", "mode", "target", "timeoutSec", "drainSec", "delaySec", "depth", "wakeAt",
                    "durationSec", "reason"}
WAKE_SOURCES = ["local"]          # a stopped shop is started again by its service manager or a person at the server
RESTART_SEC = 5                   # how long the simulated restart takes: the API does not answer meanwhile
                                  # (nextReachableAt adds 2 s: the main loop looks about once a second)
ORDER_SEC = (1, 4)                # how long a simulated checkout takes (the payment call)


def log(msg):
    print(time.strftime("%H:%M:%S ") + msg, file=sys.stderr, flush=True)


def utc(t=None):
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t))


# ------------------------------------------------------------------------------------------------ the simulation

class Timer:
    """A value that changes on its own random timer (in simulated seconds)."""

    def __init__(self, rng, lo, hi):
        self.rng, self.lo, self.hi, self.at = rng, lo, hi, 0.0

    def due(self, t):
        if t < self.at:
            return False
        self.at = t + self.rng.uniform(self.lo, self.hi)
        return True


class Episode:
    """Something that happens now and then, for a while: next start after a random gap, lasting a random time."""

    def __init__(self, rng, gap, length):
        self.rng, self.gap, self.length = rng, gap, length
        self.start = rng.uniform(*gap)
        self.end = 0.0

    def active(self, t):
        if t >= self.start and self.end <= self.start:
            self.end = self.start + self.rng.uniform(*self.length)
        if self.end > self.start and t >= self.end:
            self.start = t + self.rng.uniform(*self.gap)
        return self.start <= t < self.end


def staged(sc, name, t, start, length):
    return name in sc and start <= t < start + length


class Customers:
    """The simulated customers: only how many requests arrive. Who they are is not simulated, and never reaches HEART."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.base, self.arriving, self.flash = 30.0, 30.0, False
        self.t_base = Timer(rng, 5, 20)
        self.sale = Episode(rng, (8 * 3600, 36 * 3600), (600, 1800))     # a flash sale, now and then

    def step(self, t, wall):
        lt = time.localtime(wall)
        h = lt.tm_hour + lt.tm_min / 60
        curve = 0.2 + max(0.0, math.sin(math.pi * (h - 6) / 18)) if h >= 6 else 0.2   # quiet at night, peak mid-afternoon
        if self.t_base.due(t):
            self.base += (40 * curve - self.base) * 0.5 + self.rng.gauss(0, 2)
            self.base = max(2.0, self.base)
        self.flash = self.sale.active(t) or staged(self.sc, "flash", t, 880, 100)
        rate = max(self.base * 6, 220) if self.flash else self.base
        self.arriving = max(0.0, rate + self.rng.gauss(0, math.sqrt(rate)))


class Worker:
    def __init__(self):
        self.state, self.since, self.ok_since, self.restarts = "ok", 0.0, 0.0, []


class Shop:
    """The simulated server. The HEART reads its values, and may call only restart_worker, clear_cache and
    rotate_logs, through the capabilities below. There is nothing here that deletes, refunds or deploys."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc, self.t = rng, scenarios, 0.0
        self.workers = [Worker() for _ in range(WORKERS)]
        self.queue = self.rps = self.served = 0.0
        self.error_rate, self.p95, self.db_ms, self.db_base = 0.002, 100.0, 8.0, 8.0
        self.capacity = WORKERS * WORKER_RPS
        self.warm, self.junk = 1.0, 0.0             # cache: how warm it is (0 to 1), and leaked junk (memory %)
        self.memory = self.hit_rate = 0.0
        self.logs = 10.0                            # disk used by logs, %; the rest of the system uses 55 %
        self.deps = {"payment": True, "shipping": True}
        self.orders_today, self.revenue_today, self._orders, self.day = 0, 0.0, 0.0, None   # private
        self.t_db = Timer(rng, 10, 60)
        self.hang_ep = [Episode(rng, (2 * 3600, 12 * 3600), (1, 2)) for _ in range(WORKERS)]
        self.debug_ep = Episode(rng, (3 * 86400, 10 * 86400), (1800, 7200))   # someone left debug logging on
        self.pay_ep = Episode(rng, (12 * 3600, 48 * 3600), (120, 600))
        self.ship_ep = Episode(rng, (8 * 3600, 36 * 3600), (300, 1200))
        self.slow_ep = Episode(rng, (6 * 3600, 24 * 3600), (60, 300))
        self.done = set()
        self.taking_orders, self.in_flight, self.refused_orders = True, [], 0   # checkouts running: when each ends
        self.order_rng = random.Random(1)           # its own random source: the simulated life stays as it was
        self.cache()

    # -- what the HEART reads
    def healthy(self):
        return sum(w.state == "ok" for w in self.workers)

    def stuck_for(self, i):
        w = self.workers[i]
        return self.t - w.since if w.state == "stuck" else 0.0

    def longest_stuck(self):
        return max(range(WORKERS), key=self.stuck_for)

    @property
    def stuck_max(self):
        return self.stuck_for(self.longest_stuck())

    @property
    def disk_free(self):
        return max(0.0, 100 - 55 - self.logs)

    def cache(self):
        good = 20 * self.warm                       # memory % held by useful cache entries
        self.hit_rate = 93 * self.warm * good / (good + self.junk) if good + self.junk > 0 else 0.0
        self.memory = min(99.5, 40 + good + self.junk)

    # -- the three things the HEART may do (capabilities restartWorker, clearCache, rotateLogs)
    def restart_worker(self, i):
        w = self.workers[i]
        was = w.state
        w.state, w.ok_since = "ok", self.t
        w.restarts = [r for r in w.restarts if self.t - r < 600] + [self.t]
        return {"worker": i, "was": was, "now": "ok"}

    def clear_cache(self, percent):
        before = round(self.memory, 1)
        self.junk *= 1 - percent / 100
        self.warm *= 1 - percent / 100
        self.cache()
        return {"memoryBefore": before, "memoryAfter": round(self.memory, 1)}

    def rotate_logs(self, keep_days):
        before = round(self.disk_free, 1)
        self.logs = 1.0 + (self.logs - 1.0) * (0.15 + 0.005 * keep_days)    # compress old logs, keep keepDays
        return {"diskFreeBefore": before, "diskFreeAfter": round(self.disk_free, 1)}

    # -- what the lifecycle operations need: the checkouts running, a storefront that can stop taking orders
    def running_orders(self):
        now = time.time()
        self.in_flight = [f for f in self.in_flight if f > now]
        return len(self.in_flight)

    def checkout(self, seconds=None):
        """A customer pays. Refused while the shop drains or the payment provider is down. Returns why not, or None."""
        if not self.taking_orders:
            self.refused_orders += 1
            return ("The shop is about to restart and takes no new orders for a moment. Your basket is kept: "
                    "please try again in a minute.")
        if not self.deps["payment"]:
            return "Payments cannot be taken right now. Your basket is kept: please try again later."
        self.in_flight.append(time.time() + (seconds or self.order_rng.uniform(*ORDER_SEC)))
        self.orders_today += 1
        self.revenue_today += self.order_rng.uniform(15, 120)
        return None

    def reload(self, t):
        """After a restart: new worker processes, an empty queue and a cold cache (the leaked junk went with the old
        process). Logs, the disk and the day's totals are kept: they were on disk and in the database."""
        self.t = t
        self.workers = [Worker() for _ in range(WORKERS)]
        for w in self.workers:
            w.ok_since = t
        self.queue, self.warm, self.junk = 0.0, 0.1, 0.0
        self.in_flight, self.taking_orders = [], True
        self.cache()

    def hang(self, i, why):
        w = self.workers[i]
        if w.state == "ok":
            w.state, w.since = "stuck", self.t
            log(f"shop: worker {i} hangs ({why})")

    def step(self, t, dt, cust, wall):
        self.t, sc, rng = t, self.sc, self.rng
        # Staged faults, for the demo. Each starts at a set time and is then fixed, by a Playbook or "by the owner".
        if "stuck" in sc and t >= 60 and "stuck" not in self.done:
            self.done.add("stuck")
            self.hang(3, "staged: a request that never finishes")
        poison = staged(sc, "crashloop", t, 300, 300)
        w5 = self.workers[5]
        if poison and w5.state == "ok" and t - w5.ok_since >= 10:
            self.hang(5, "staged: a poison job it picks up again after every restart")
        if "crashloop" in sc and t >= 600 and "crashloop" not in self.done:
            self.done.add("crashloop")
            if w5.state == "stuck":
                w5.state, w5.ok_since = "ok", t
            log("shop: the owner removed the poison job and restarted worker 5 (not through HEART)")
        for i, w in enumerate(self.workers):
            if self.hang_ep[i].active(t):
                self.hang(i, "random")
        debug = self.debug_ep.active(t) or staged(sc, "disk", t, 150, 300)
        if staged(sc, "leak", t, 500, 450):
            self.junk += 0.25 * dt                  # a cache eviction bug: entries that are never used, never freed
        if "leak" in sc and t >= 950 and "leak" not in self.done:
            self.done.add("leak")
            self.junk = 0.0
            log("shop: the owner deployed a fix for the cache eviction bug (not through HEART)")
        self.deps["payment"] = not (self.pay_ep.active(t) or staged(sc, "payment", t, 700, 150))
        self.deps["shipping"] = not self.ship_ep.active(t)
        slow = self.slow_ep.active(t) or staged(sc, "slowdb", t, 1010, 100)

        # Cache and memory.
        self.warm = min(1.0, self.warm + (1 - self.warm) * dt / 40)
        self.cache()
        self.memory = min(99.5, max(0.0, self.memory + rng.gauss(0, 0.3)))
        # Database: slower on cache misses, much slower in a slow-database episode.
        if self.t_db.due(t):
            self.db_base = min(14.0, max(5.0, self.db_base + rng.gauss(0, 0.8)))
        self.db_ms = self.db_base * (1 + 2 * (1 - self.hit_rate / 100)) + (rng.uniform(800, 1000) if slow else 0)
        # The web tier: requests arrive, wait in the queue, and are handled by the workers that are not stuck.
        self.capacity = self.healthy() * WORKER_RPS / (1 + self.db_ms / 2000)
        incoming = cust.arriving * dt
        q = self.queue + incoming
        served = min(q, self.capacity * dt)
        self.queue = q - served
        dropped = max(0.0, self.queue - self.capacity * QUEUE_WAIT_MAX)          # waited too long: timed out
        self.queue -= dropped
        stuck = WORKERS - self.healthy()
        errors = (dropped + incoming * stuck / WORKERS * 0.3 + served * 0.002
                  + (served * CHECKOUT_SHARE if not self.deps["payment"] else 0))
        k = min(1.0, dt / 60)
        self.error_rate += (min(1.0, errors / incoming if incoming else 0) - self.error_rate) * k
        self.rps, self.served = cust.arriving, served / dt
        self.p95 = max(20.0, 60 + 4 * self.db_ms + rng.gauss(0, 8))            # handling time, without queue wait
        # Logs fill the disk; much faster with debug logging left on.
        self.logs += dt * (0.15 if debug else 0.0004 * cust.arriving / 40)
        # Orders and revenue: private, never read by HEART's rules, never sent anywhere.
        day = time.localtime(wall).tm_yday
        if day != self.day:
            self.day, self.orders_today, self.revenue_today = day, 0, 0.0
        if self.deps["payment"]:
            self._orders += served * ORDER_SHARE
            while self._orders >= 1:
                self._orders -= 1
                if not self.taking_orders:          # draining: the customer is asked to try again in a minute
                    self.refused_orders += 1
                    continue
                self.orders_today += 1
                self.revenue_today += rng.uniform(15, 120)
                self.in_flight.append(time.time() + self.order_rng.uniform(*ORDER_SEC))


# ------------------------------------------------------------------------------------------------ the HEART

RULES = [
    # code, severity, condition, summary, component
    ("system.main-function-lost", "critical", lambda s: s.healthy() == 0 or (s.rps > 1 and s.served < 0.1),
     "The shop is not serving any customer", None),
    ("software.crashed", "warning", lambda s: s.stuck_max >= STUCK_SEC,
     "A web worker has stopped answering", lambda s: f"worker-{s.longest_stuck()}"),
    ("software.crash-loop", "error",
     lambda s: any(w.state == "stuck" and len([r for r in w.restarts if s.t - r < 600]) >= 2 for w in s.workers),
     "A web worker hangs again after every restart", None),
    ("performance.memory-exhausted", "error", lambda s: s.memory >= MEMORY_HIGH, "Memory is nearly used up", None),
    ("storage.nearly-full", "warning", lambda s: DISK_FULL < s.disk_free <= DISK_LOW, "The disk is nearly full", None),
    ("storage.full", "critical", lambda s: s.disk_free <= DISK_FULL, "The disk is full: logs and uploads fail", None),
    ("connectivity.dependency-unreachable", "error", lambda s: not all(s.deps.values()),
     "An outside service the shop needs cannot be reached", None),
    ("performance.overloaded", "warning", lambda s: s.rps > s.capacity * 1.05,
     "More customers are arriving than the shop can serve", None),
    ("performance.backlog", "warning", lambda s: s.queue > BACKLOG_HIGH, "Requests are piling up in the queue", None),
    ("performance.slow", "warning", lambda s: s.p95 > SLOW_MS, "Pages take much longer than normal", None),
]

EXPLAIN = {
    "system.main-function-lost": lambda s: f"{s.healthy()} of {WORKERS} workers answer and {s.served:.0f} requests/s are served. Look now.",
    "software.crashed": lambda s: f"Worker {s.longest_stuck()} has not answered for {s.stuck_max:.0f} s. Its requests time out.",
    "software.crash-loop": lambda s: "Restarting does not help: it hangs again within seconds. Likely a bad job, request or release. A person must look.",
    "performance.memory-exhausted": lambda s: f"Memory {s.memory:.0f} % used, cache hit rate {s.hit_rate:.0f} %.",
    "storage.nearly-full": lambda s: f"{s.disk_free:.1f} % free (threshold {DISK_LOW} %). Logs take {s.logs:.0f} % of the disk.",
    "storage.full": lambda s: f"{s.disk_free:.1f} % free. New logs, sessions and uploads cannot be written.",
    "connectivity.dependency-unreachable": lambda s: " ".join(
        {"payment": "The payment provider is down: customers can browse but cannot pay.",
         "shipping": "The shipping API is down: orders are taken, labels wait."}[d] for d, up in s.deps.items() if not up),
    "performance.overloaded": lambda s: f"{s.rps:.0f} requests/s arriving, {s.capacity:.0f} can be served. A flash sale? Adding servers costs money: the owner decides.",
    "performance.backlog": lambda s: f"{s.queue:.0f} requests waiting. Those waiting over {QUEUE_WAIT_MAX} s time out.",
    "performance.slow": lambda s: f"p95 handling time {s.p95:.0f} ms (database {s.db_ms:.0f} ms per page). The database is slow: no Playbook covers it.",
}

# The capabilities the HEART offers: three harmless writes, each with its input range. Not physical (software only).
CAPS = {
    "restartWorker": {
        "title": "Restart a web worker", "description": "Restart one worker process. Requests in it are retried by the queue.",
        "inputSchema": {"type": "object", "required": ["worker"], "additionalProperties": False,
                        "properties": {"worker": {"type": "integer", "minimum": 0, "maximum": WORKERS - 1}}}},
    "clearCache": {
        "title": "Clear the cache", "description": "Drop this share of the cache. Pages are slower until it is warm again.",
        "inputSchema": {"type": "object", "required": ["percent"], "additionalProperties": False,
                        "properties": {"percent": {"type": "integer", "minimum": 1, "maximum": 100}}},
        "guards": {"interlocks": [{"signal": f"{SYSTEM_ID}.cache_hit_rate", "op": "<", "value": HIT_LOW}]}},
    "rotateLogs": {
        "title": "Rotate and compress the logs", "description": "Start new log files and compress the old ones, keeping keepDays days uncompressed.",
        "inputSchema": {"type": "object", "required": ["keepDays"], "additionalProperties": False,
                        "properties": {"keepDays": {"type": "integer", "minimum": 1, "maximum": 30}}}},
}

# The owner's allowlist (control policy): what may run without asking each time, for initiator rule only.
ALLOWLIST = [
    {"target": "restartWorker", "initiators": ["rule"],
     "envelope": {"arguments": {"worker": {"min": 0, "max": WORKERS - 1}}, "maxPerDay": 6, "cooldownSec": 60}},
    {"target": "clearCache", "initiators": ["rule"],
     "envelope": {"arguments": {"percent": {"min": 1, "max": 100}}, "maxPerDay": 2, "cooldownSec": 120}},
    {"target": "rotateLogs", "initiators": ["rule"],
     "envelope": {"arguments": {"keepDays": {"min": 3, "max": 30}}, "maxPerDay": 4, "cooldownSec": 300}},
]

# The owner's three Playbooks. Each runs when the HEART has raised its alert (after the 5-second hold).
PLAYBOOKS = [
    {"id": "restart-stuck-worker", "codes": ["software.crashed"], "capability": "restartWorker",
     "description": f"When a web worker has not answered for {STUCK_SEC} s, restart that worker.",
     "when": {"signal": f"{SYSTEM_ID}.worker_stuck_max", "op": ">=", "value": STUCK_SEC},
     "args": lambda s: {"worker": s.longest_stuck()},
     "recurred": lambda s, a: s.workers[a["worker"]].state == "stuck", "judgeSec": 120},
    {"id": "clear-bloated-cache", "codes": ["performance.memory-exhausted"], "capability": "clearCache",
     "description": f"When memory is above {MEMORY_HIGH} % and the cache hit rate is below {HIT_LOW} % (the clearCache interlock), clear the cache.",
     "when": {"signal": f"{SYSTEM_ID}.memory_used", "op": ">=", "value": MEMORY_HIGH},
     "args": lambda s: {"percent": 100},
     "recurred": lambda s, a: s.memory >= MEMORY_HIGH, "judgeSec": 60},
    {"id": "rotate-full-logs", "codes": ["storage.nearly-full", "storage.full"], "capability": "rotateLogs",
     "description": f"When the disk has {DISK_LOW} % or less free, rotate and compress the logs.",
     "when": {"signal": f"{SYSTEM_ID}.disk_free", "op": "<=", "value": DISK_LOW},
     "args": lambda s: {"keepDays": 7},
     "recurred": lambda s, a: s.disk_free <= DISK_LOW, "judgeSec": 60},
]
GIVE_UP_AFTER = 2          # a fix that did not help this many times in a row suspends its Playbook


class Heart:
    @property
    def current(self):
        """The current level: the lower of the ceiling and the promised level (API Standard 13.4)."""
        return min(self.max, PROMISED_LEVEL)

    @property
    def mode(self):
        """Never set: it follows from the current level."""
        return "observe" if self.current <= 2 else "suggest" if self.current == 3 else "act"

    """Notice, change status, raise, explain, escalate; and, for the three known problems, run the owner's Playbooks
    within the allowlist. No network or model is needed for any of it."""

    def __init__(self, shop):
        self.shop = shop
        self.lock = threading.RLock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key = secrets.token_urlsafe(18)
        self.t, self.max, self.stopped_at = 0.0, PROMISED_LEVEL, None   # max: the autonomy ceiling a person set
        self.actions, self.events, self.refused, self.idem = [], [], set(), {}
        self.prev = ["ok"] * WORKERS
        self.started_t, self.phase, self.phase_since = 0.0, "running", utc()   # lifecycle
        self.life, self.drained_until, self.down_until = None, None, 0.0     # the lifecycle Action under way
        now = utc()
        # In this demo the owner adopted the Playbooks at setup (a real product: PUT /heart/v1/tending/rules/{id},
        # by a person, with proof of presence; every Playbook starts proposed).
        self.rules = {p["id"]: {"status": "adopted", "enabled": True, "adoptedAt": now, "inARow": 0, "judging": None}
                      for p in PLAYBOOKS}
        self.event("system.started", "system", "info", "The shop server and its HEART started")

    # -- history (Rhythm): what happened, and every action with why, who and the result
    def event(self, name, category, severity, summary, component=None, attributes=None):
        e = {"sequence": len(self.events), "name": name, "category": category, "severity": severity, "summary": summary}
        if component:
            e["component"] = component
        if attributes:
            e["attributes"] = attributes
        self.events = (self.events + [e])[-500:]

    def tend(self, t):
        with self.lock:
            s = self.shop
            self.t = t
            self.sequence += 1
            for i, w in enumerate(s.workers):
                if w.state != self.prev[i]:
                    self.event("component.failed" if w.state == "stuck" else "component.recovered", "component",
                               "warning" if w.state == "stuck" else "info", f"Worker {i} is {w.state}", f"worker-{i}")
                    self.prev[i] = w.state
            self.judge(t)
            for code, severity, cond, summary, comp in RULES:
                hit = cond(s)
                # A condition must hold for 5 simulated seconds before it is raised: one odd reading is not a fault.
                prev = self.since.get(code)
                self.since[code] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[code] >= 5
                open_ = self.open_alert(code)
                if held and open_ is None:
                    a = {"id": str(uuid.uuid4()), "code": code, "severity": severity, "status": "open",
                         "detectedAt": utc(), "summary": summary, "explanation": EXPLAIN[code](s), "actions": []}
                    if comp:
                        a["component"] = comp(s)
                    self.alerts = (self.alerts + [a])[-100:]
                    pb = self.playbook_for(code)
                    if pb and self.rules[pb["id"]]["status"] == "adopted" and self.mode == "act":
                        a["status"] = "tending"
                        log(f"HEART tending  {code}: {summary}. {a['explanation']} Playbook {pb['id']} will act.")
                    else:
                        self.escalate(a, self.why_not(pb))
                elif not hit and open_ is not None:
                    open_.update(status="resolved", resolvedAt=utc())
                    log(f"HEART resolved   {code}: {open_['summary'].lower()}")
            self.playbooks(t)
            sev = {a["severity"] for a in self.alerts if a["status"] != "resolved"}
            self.status = "failing" if "critical" in sev else "degraded" if sev else "healthy"

    def open_alert(self, code):
        return next((a for a in self.alerts if a["code"] == code and a["status"] != "resolved"), None)

    def playbook_for(self, code):
        return next((p for p in PLAYBOOKS if code in p["codes"]), None)

    def why_not(self, pb):
        if pb is None:
            return "No Playbook covers this problem: a person must decide what to do"
        if self.mode != "act":
            return "Automatic action is stopped: a person must fix this"
        return f"Playbook {pb['id']} is suspended because its fix did not help: a person must look"

    def escalate(self, a, reason):
        """CHANGE for a product: the on-call route (pager, chat, email). Here: the console."""
        if a["status"] == "escalated":
            return
        a["status"] = "escalated"
        a["escalation"] = {"at": utc(), "reason": reason}
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']} Escalated: {reason}.")

    # -- the Tending engine: Playbooks, the gate, the allowlist and its limits
    def playbooks(self, t):
        for pb in PLAYBOOKS:
            alert = next((self.open_alert(c) for c in pb["codes"] if self.open_alert(c)), None)
            if alert is None:
                continue
            state, cap = self.rules[pb["id"]], pb["capability"]
            args = pb["args"](self.shop)
            if state["status"] != "adopted" or not state["enabled"]:
                self.refuse(pb, alert, args, "suspended", f"Playbook {pb['id']} is suspended: its fix did not help "
                            f"{GIVE_UP_AFTER} times in a row. Only a person can adopt it again")
                continue
            if self.mode != "act":
                self.refuse(pb, alert, args, "autonomy-stopped", "autonomy-stopped: automatic action was stopped by the owner")
                continue
            entry = next((e for e in ALLOWLIST if e["target"] == cap and "rule" in e["initiators"]), None)
            if entry is None:
                self.refuse(pb, alert, args, "allowlist", f"{cap} is not on the owner's allowlist for rule")
                continue
            env = entry["envelope"]
            box = env["arguments"]
            out = [k for k, v in args.items() if k not in box or not box[k]["min"] <= v <= box[k]["max"]]
            if out:
                self.refuse(pb, alert, args, "envelope", f"tending.envelope.exceeded: {', '.join(out)} outside the allowlist envelope")
                continue
            runs = [a for a in self.actions if a.get("capability") == cap and a["initiator"]["kind"] == "rule"
                    and a["status"] in ("succeeded", "failed")]
            if runs and t - runs[-1]["ext"][EXT_SIM] < env["cooldownSec"]:
                continue                                 # cooldown: the rule waits; nothing is refused yet
            if len([a for a in runs if t - a["ext"][EXT_SIM] < 86400]) >= env["maxPerDay"]:
                self.refuse(pb, alert, args, "maxPerDay", f"maxPerDay: {cap} already ran {env['maxPerDay']} times "
                            "automatically in the last 24 hours. The Playbook stops here")
                continue
            if state["judging"]:
                continue                                 # the last fix is still being judged
            a = self.run(cap, args, {"kind": "rule", "id": pb["id"]}, alert,
                         f"Playbook {pb['id']}: {pb['description']} Allowed by the owner's allowlist entry for rule.")
            if a["status"] == "succeeded":
                state["judging"] = (a, t + pb["judgeSec"])
                log(f"HEART fixed    {alert['code']}: Playbook {pb['id']} ran {cap}({json.dumps(a['arguments'])}): {json.dumps(a['result'])}")
            else:
                self.escalate(alert, f"Playbook {pb['id']} could not run: {a['error']}")

    def judge(self, t):
        """Did the last automatic fix help? Not if the problem came back within judgeSec."""
        for pb in PLAYBOOKS:
            st = self.rules[pb["id"]]
            if not st["judging"]:
                continue
            a, until = st["judging"]
            if pb["recurred"](self.shop, a["arguments"]):
                st["judging"], st["inARow"] = None, st["inARow"] + 1
                a["ext"]["org.heartaispec.demo.helped"] = False
                self.event("tending.rule.failed", "tending", "warning", f"Playbook {pb['id']}: the fix did not help "
                           f"({st['inARow']} in a row)", attributes={"action": a["id"]})
                log(f"HEART          Playbook {pb['id']}: the problem came back. The fix did not help ({st['inARow']} in a row).")
                if st["inARow"] >= GIVE_UP_AFTER:
                    st["status"] = "suspended"
                    self.event("tending.rule.suspended", "tending", "error", f"Playbook {pb['id']} suspended: its fix "
                               f"did not help {GIVE_UP_AFTER} times in a row")
                    log(f"HEART          Playbook {pb['id']} SUSPENDED: it escalates instead of looping. A person must look.")
            elif t >= until:
                st["judging"], st["inARow"] = None, 0
                a["ext"]["org.heartaispec.demo.helped"] = True

    def refuse(self, pb, alert, args, why, reason):
        """Record one denied Action per alert and reason, tell a person, and do nothing."""
        key = (alert["id"], why)
        if key in self.refused:
            return
        self.refused.add(key)
        a = self.record(pb["capability"], args, {"kind": "rule", "id": pb["id"]}, alert, "denied",
                        f"Playbook {pb['id']} wanted to run {pb['capability']}; refused.")
        a["error"] = reason
        self.event("tending.rule.refused", "tending", "warning", f"Playbook {pb['id']}: {reason}", attributes={"action": a["id"]})
        self.escalate(alert, f"Playbook {pb['id']} refused: {reason}")

    def record(self, cap, args, initiator, alert, status, explanation):
        now = utc()
        a = {"id": str(uuid.uuid4()), "capability": cap, "status": status, "risk": "write", "physical": False,
             "createdAt": now, "updatedAt": now, "initiator": initiator, "arguments": dict(args),
             "explanation": explanation, "ext": {EXT_SIM: round(self.t, 1)}}
        if alert:
            a["arrhythmiaId"] = alert["id"]
            alert["actions"].append(a["id"])
        self.actions = (self.actions + [a])[-500:]
        return a

    def run(self, cap, args, initiator, alert, explanation):
        """Run one capability after the gate: guards (interlocks) first, then the product's own function."""
        s = self.shop
        a = self.record(cap, args, initiator, alert, "running", explanation)
        for chk in CAPS[cap].get("guards", {}).get("interlocks", []):
            value = self.signal(chk["signal"])
            if not OPS[chk["op"]](value, chk["value"]):
                a.update(status="failed", updatedAt=utc(), error=f"guard-failed: {chk['signal'].rsplit('.', 1)[1]} is "
                         f"{value:.0f}, the interlock needs {chk['op']} {chk['value']}. The cache is still useful: "
                         "clearing it would only slow the shop down")
                self.event("tending.guard.refused", "tending", "warning", a["error"], attributes={"action": a["id"]})
                return a
        if cap == "restartWorker":
            a["result"] = s.restart_worker(args["worker"])
            self.event("component.restarted", "component", "info", f"Worker {args['worker']} restarted by "
                       f"{initiator['kind']} {initiator['id']}", f"worker-{args['worker']}", {"action": a["id"]})
        elif cap == "clearCache":
            a["result"] = s.clear_cache(args["percent"])
        elif cap == "rotateLogs":
            a["result"] = s.rotate_logs(args["keepDays"])
        a.update(status="succeeded", updatedAt=utc())
        if cap != "restartWorker":
            self.event(f"shop.{cap.lower()}.done", "domain", "info", f"{CAPS[cap]['title']} by {initiator['kind']} "
                       f"{initiator['id']}", attributes={"action": a["id"], "result": a["result"]})
        return a

    def signal(self, name):
        return {p[0]: p[1] for p in self.values()}[name.rsplit(".", 1)[1]]

    # -- messages
    def envelope(self, kind, data, privacy="public"):
        return {"specversion": "1.0", "id": str(uuid.uuid4()), "source": f"/heart/{SYSTEM_ID}/{INSTANCE}", "type": kind,
                "time": utc(), "datacontenttype": "application/json", "heartversion": "1.0", "heartprivacy": privacy,
                "data": data}

    def pulse(self, owner=False):
        with self.lock:
            s = self.shop
            d = {"system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "status": self.status,
                 "sequence": self.sequence, "intervalSec": PULSE_SEC}
            if owner:
                d["reconnaissance"] = {
                    "state": "draining" if not s.taking_orders else "overloaded" if s.rps > s.capacity else "serving",
                    "lastEvent": self.events[-1]["name"] if self.events else "none",
                    "vitals": {"uptimeSec": int(self.t - self.started_t), "errorRate": round(min(1.0, s.error_rate), 4),
                               "latencyMsP95": round(s.p95), "memoryUtilization": round(s.memory / 100, 3)}}
                d["tending"] = {"openIssues": sum(a["status"] != "resolved" for a in self.alerts),
                                "lastAction": self.actions[-1]["id"] if self.actions else None}
                d["autonomy"] = {"max": self.max, "current": self.current}
                if self.current < PROMISED_LEVEL:
                    d["autonomy"]["limitedBy"] = "stopped"
        return self.envelope("heart.pulse", d, "internal" if owner else "public")

    def public_alerts(self, n=16):
        with self.lock:
            keep = ("id", "code", "severity", "status", "detectedAt", "resolvedAt")
            return [self.envelope("heart.arrhythmia", {k: a[k] for k in keep if k in a}) for a in reversed(self.alerts[-n:])]

    def values(self):
        s = self.shop
        storefront = "closed" if s.healthy() == 0 else "degraded" if self.status != "healthy" else "open"
        return [  # name, value, unit, privacy
            ("storefront", storefront, None, "public"),
            ("requests_per_sec", round(s.rps, 1), "/s", "internal"),
            ("error_rate", round(100 * s.error_rate, 2), "%", "internal"),
            ("latency_p95", round(s.p95), "ms", "internal"),
            ("queue_backlog", round(s.queue), None, "internal"),
            ("workers_healthy", s.healthy(), None, "internal"),
            ("worker_stuck_max", round(s.stuck_max), "s", "internal"),
            ("cache_hit_rate", round(s.hit_rate, 1), "%", "internal"),
            ("memory_used", round(s.memory, 1), "%", "internal"),
            ("disk_free", round(s.disk_free, 1), "%", "internal"),
            ("database_latency", round(s.db_ms), "ms", "internal"),
            ("payment_provider", "up" if s.deps["payment"] else "down", None, "internal"),
            ("shipping_api", "up" if s.deps["shipping"] else "down", None, "internal"),
            ("orders_today", s.orders_today, None, "internal"),
            ("revenue_today_usd", round(s.revenue_today, 2), None, "private"),
        ]

    def properties(self, owner=False):
        out = []
        with self.lock:
            for name, value, unit, privacy in self.values():
                if privacy != "public" and not owner:
                    continue
                d = {"name": f"{SYSTEM_ID}.{name}", "privacy": privacy, "value": value, "writable": False, "time": utc()}
                if unit:
                    d["unit"] = unit
                out.append(self.envelope("heart.property", d, privacy))
        return out

    def card(self, owner=False, port=0):
        d = {"system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "level": "basic",
             "bindings": ["http-server"], "endpoints": {"api": f"http://127.0.0.1:{port}/heart/v1"},
             "pulse": {"intervalSec": PULSE_SEC}, "modules": ["properties", "control", "lifecycle"],
             "shares": {"pulse": "public", "arrhythmia": "public", "property": "private", "action": "internal",
                        "rule": "internal", "control": "internal", "rhythm": "internal"},
             "autonomy": {"level": 1}, "template": TEMPLATE}
        if owner:   # the capabilities are internal: only a caller with that clearance sees them
            d["capabilities"] = [dict(name=n, risk="write", privacy="internal", physical=False, reversible=True,
                                      idempotent=False, callers=["human", "rule"], kind="product",
                                      **{k: v for k, v in c.items()}) for n, c in CAPS.items()]
            d["lifecycle"] = {"operations": LIFECYCLE_OPS, "wakeSources": WAKE_SOURCES}
        return self.envelope("heart.card", d, "internal" if owner else "public")

    def rule_messages(self):
        with self.lock:
            out = []
            for pb in PLAYBOOKS:
                st = self.rules[pb["id"]]
                d = {"id": pb["id"], "enabled": st["enabled"], "origin": "owner", "description": pb["description"],
                     "when": pb["when"], "then": {"capability": pb["capability"]}, "status": st["status"],
                     "adoptedAt": st["adoptedAt"], "adoptedBy": "owner (at setup, simulated)", "adoptedPhysical": False}
                out.append(self.envelope("heart.rule", d, "private"))
            return out

    def control(self):
        d = {"max": self.max, "current": self.current, "mode": self.mode, "allowlist": ALLOWLIST, "approvalTimeoutSec": 86400}
        if self.stopped_at:
            d.update(stoppedAt=self.stopped_at, reason="Stop pressed by the owner")
        return self.envelope("heart.control", d, "internal")

    def stop(self):
        with self.lock:
            if self.max > 2:
                self.max, self.stopped_at = 2, utc()
                self.event("tending.stopped", "tending", "warning", "Stop pressed: all automatic action stopped")
                log("HEART          STOP pressed: Playbooks no longer act. Noticing and escalating go on.")
            return self.control()

    # -- Lifecycle (API Standard 13.2 rule 3): drain, and restart gracefully or by force. The product carries the
    #    operation out (rule 8): here the main loop closes the API, reloads the shop and opens the API again.
    def lifecycle_view(self, owner):
        with self.lock:
            d = {"phase": self.phase}
            if owner:
                d.update(allowed=[] if self.life or self.phase != "running" else LIFECYCLE_OPS, since=self.phase_since,
                         wakeSources=WAKE_SOURCES, ext={"org.heartaispec.demo.taking-orders": self.shop.taking_orders,
                                                        "org.heartaispec.demo.orders-running": self.shop.running_orders()})
            return self.envelope("heart.lifecycle", d, "private" if owner else "public")

    def lifecycle_post(self, op, body):
        """POST /heart/v1/lifecycle/{op} with the owner key (a human initiator). The path is already checked.
        Returns (status, body, content type[, headers])."""
        if set(body) - LIFECYCLE_FIELDS:
            return problem(400, "invalid-request", f"Unknown fields: {', '.join(sorted(set(body) - LIFECYCLE_FIELDS))}")
        if body.get("operation", op) != op:
            return problem(400, "invalid-request", "operation in the body must equal the one in the path")
        for k, top in (("timeoutSec", 86400), ("drainSec", 86400), ("delaySec", 2592000)):
            v = body.get(k, 1)
            if not isinstance(v, int) or isinstance(v, bool) or not (1 if k == "timeoutSec" else 0) <= v <= top:
                return problem(400, "invalid-request", f"{k} must be a whole number of seconds, "
                                                       f"{1 if k == 'timeoutSec' else 0} to {top}")
        if body.get("delaySec", 0) > 0:
            return problem(400, "invalid-request", "delaySec needs the schedules module, which this shop does not offer. "
                                                   "Send the request when it should happen")
        if "mode" in body and (op != "restart" or body["mode"] not in ("graceful", "force")):
            return problem(400, "invalid-request", "mode is for restart only: graceful (the default) or force")
        if body.get("target", "system") != "system" or ("target" in body and op != "restart"):
            return problem(400, "invalid-request", "This shop restarts as a whole: target system. To restart one "
                                                   "worker, use the capability restartWorker")
        extra = sorted({"depth", "wakeAt", "durationSec"} & set(body))
        if extra:
            return problem(400, "invalid-request", f"{', '.join(extra)}: only for sleep and maintenance-on, which this "
                                                   "shop does not offer")
        if not isinstance(body.get("reason", ""), str) or len(body.get("reason", "")) > 500:
            return problem(400, "invalid-request", "reason must be text, at most 500 characters")
        if self.life:
            return problem(409, "conflict", f"Another lifecycle operation is under way (Action {self.life['id']}, "
                                            f"{self.life['status']}). Wait for it, or cancel it first")
        force = body.get("mode") == "force"
        now, operation, aid = time.time(), f"POST /heart/v1/lifecycle/{op}", str(uuid.uuid4())
        limit = min(body.get("timeoutSec", 60), body.get("drainSec", 86400)) if op == "restart" else \
            min(body.get("drainSec", 60), body.get("timeoutSec", 86400))
        a = {"id": aid, "operation": operation, "status": "running", "risk": "destructive" if force else "write",
             "physical": False, "digest": hashlib.sha256(json.dumps(
                 {"system": SYSTEM_ID, "id": aid, "operation": operation, "arguments": body, "physical": False},
                 sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest(),
             "createdAt": utc(now), "updatedAt": utc(now), "initiator": {"kind": "human", "id": "owner"},
             "arguments": dict(body), "explanation": body.get("reason") or {
                 "drain": "Drain asked for by the owner: stop taking new orders and finish the ones running",
                 "restart": "Force restart asked for by the owner: now, cutting off the orders running" if force else
                 "Graceful restart asked for by the owner: finish the orders running, then restart"}[op],
             "cancellable": True, "ext": {EXT_SIM: round(self.t, 1)},
             "_op": op, "_force": force, "_limit": limit, "_until": now + limit, "_cancel": False}
        if op == "drain":
            a["_reopen"] = now + body.get("drainSec", 60)
        if force:   # destructive: held for a person at the server; nothing runs before that (API Standard 6.3)
            a.update(status="awaiting-approval", expiresAt=utc(now + 86400), nextReachableAt=None,
                     nextReachableNote="Not restarting yet: a person must approve it at the server first")
            a["_expires"] = now + 86400
            log(f"HEART          a FORCE RESTART waits for a person: press the button at the server (`touch {BUTTON}` "
                f"in its working directory) to approve Action {aid} (digest {a['digest'][:16]}...). Deny or cancel "
                "it through the API.")
        else:
            if op == "restart":
                a["nextReachableAt"] = utc(now + limit + RESTART_SEC + 2)
            self.start_drain(a)
        self.life = a
        self.actions = (self.actions + [a])[-500:]
        what = "Force restart" if force else op.capitalize()
        self.event(f"heart.lifecycle.{op}-asked", "system", "info", f"{what} asked for by the owner", attributes={"action": aid})
        return 202, self.envelope("heart.action", view(a), "private"), "application/cloudevents+json", \
            {"Location": f"/heart/v1/actions/{aid}"}

    def start_drain(self, a):
        s = self.shop
        s.taking_orders = False
        a["_n0"] = s.running_orders()
        log(f"HEART          {a['_op']}: no new orders; {a['_n0']} running, {a['_limit']} s to finish them")

    def life_end(self, a, status, **fields):
        """A lifecycle Action ends. The shop takes orders again unless a finished drain still holds it."""
        now = time.time()
        a.update(status=status, updatedAt=utc(now), cancellable=False, **fields)
        if a["_op"] == "restart" and status != "succeeded":
            a["nextReachableAt"] = utc(now)         # it never stopped answering
            a.pop("nextReachableNote", None)
        if self.life is a:
            self.life = None
        self.shop.taking_orders = not (self.drained_until and now < self.drained_until)
        self.event(f"heart.lifecycle.{a['_op']}-{status}", "system", "info" if status == "succeeded" else "warning",
                   f"{a['_op'].capitalize()} {status}" + (f": {a['error']}" if a.get("error") else ""),
                   attributes={"action": a["id"]})
        log(f"HEART          {'force ' if a['_force'] else ''}{a['_op']} {status}" + (f": {a['error']}" if a.get("error") else ""))

    def decide(self, aid, body):
        """POST /heart/v1/actions/{id}/decision with the owner key: approve, deny or cancel (API Standard 6.3)."""
        a = next((x for x in self.actions if x["id"] == aid), None)
        if a is None:
            return problem(404, "not-found", "No such Action")
        d = body.get("decision")
        if set(body) - {"decision", "note", "presence"} or d not in ("approve", "deny", "cancel") or \
                not isinstance(body.get("note", ""), str):
            return problem(400, "invalid-request", "Send {\"decision\": \"approve\", \"deny\" or \"cancel\"}, with an "
                                                   "optional note, and presence to approve")
        st, rec = a["status"], {"decision": d, "at": utc(), "by": "owner"}
        if body.get("note"):
            rec["note"] = body["note"]
        if d == "cancel":
            if st in ("awaiting-approval", "approved"):
                self.life_end(a, "cancelled", decision=rec)
            elif st == "running" and a.get("cancellable"):
                a.update(status="cancelling", updatedAt=utc(), decision=rec)
                a["_cancel"] = True                 # it stops at the next safe point: the next check, within 1 s
            elif st != "cancelling":
                return problem(409, "conflict", "The restart point is passed: it can no longer be stopped"
                               if st == "running" else f"This Action is {st}: there is nothing left to cancel")
            return 200, self.envelope("heart.action", view(a), "private")
        if st != "awaiting-approval":
            return problem(409, "conflict", f"This Action is {st}, not awaiting approval")
        if d == "deny":
            self.life_end(a, "denied", decision=rec)
            return 200, self.envelope("heart.action", view(a), "private")
        pr = body.get("presence")
        at_server = f"Press the button at the server: run `touch {BUTTON}` in its working directory. The press approves this Action"
        if pr is None:
            return problem(403, "presence-required", f"A force restart needs a person, not only a key. {at_server}")
        if not isinstance(pr, dict) or set(pr) != {"method", "digest", "proof"} or not isinstance(pr["proof"], str) or \
                pr["method"] not in ("webauthn", "ciba", "device-button", "remote-site"):
            return problem(400, "invalid-request", "presence is {\"method\", \"digest\", \"proof\"}")
        if pr["digest"] != a["digest"]:
            return problem(403, "presence-invalid", "This proof is for another digest: it does not approve this Action")
        if pr["method"] == "device-button":
            return problem(403, "presence-invalid", f"A button press is not accepted through the API (API Standard 6.3). {at_server}")
        return problem(403, "presence-invalid", f"This simulator cannot check {pr['method']} proofs: it has no passkeys, "
                                                f"CIBA or pinned monitor-site key. {at_server}")

    def lifecycle_tick(self):
        """Called by the main loop about once a second; each call is a safe point. Returns "restart" when the shop
        must restart now."""
        now = time.time()
        with self.lock:
            s, a = self.shop, self.life
            if self.drained_until and now >= self.drained_until and a is None:
                self.drained_until, s.taking_orders = None, True
                self.event("heart.lifecycle.drain-ended", "system", "info", "The drain ended: new orders are taken again")
                log("HEART          the drain ended: new orders are taken again")
            if a is None:
                return None
            if a["status"] == "awaiting-approval":
                if now >= a["_expires"]:
                    self.life_end(a, "expired")
                elif pressed():                     # a person at the server: the simulator's proof of presence
                    a.update(status="approved", updatedAt=utc(), nextReachableAt=utc(now + RESTART_SEC + 2),
                             decision={"decision": "approve", "at": utc(), "by": "a person at the server (local button)",
                                       "presenceMethod": "device-button"})
                    a.pop("nextReachableNote", None)
                    log("HEART          a person at the server approved the force restart")
                else:
                    return None
            if a["status"] == "approved":           # force: no drain, the orders running are cut off
                cut = s.running_orders()
                a.update(status="running", updatedAt=utc(), cancellable=False, progress=1.0,
                         _result={"ordersCutOff": cut, "ordersFinished": 0})
                self.phase, self.phase_since = "stopping", utc()
                return "restart"
            if a["status"] not in ("running", "cancelling") or not a["cancellable"]:
                return None
            left = s.running_orders()
            n0 = max(a["_n0"], left)
            a["progress"] = round((n0 - left) / n0, 2) if n0 else 1.0
            if a["_cancel"]:
                self.life_end(a, "cancelled", result={"ordersFinished": n0 - left, "ordersRunning": left})
            elif left == 0 and a["_op"] == "drain":
                self.drained_until = a["_reopen"]
                self.life_end(a, "succeeded", progress=1.0, result={
                    "ordersFinished": n0, "takesOrdersAgainAt": utc(a["_reopen"]),
                    "note": "Drained: no order is running. New orders are refused until takesOrdersAgainAt"})
            elif left == 0:                         # the restart point: from here it can no longer be cancelled
                a.update(updatedAt=utc(), cancellable=False, progress=1.0, nextReachableAt=utc(now + RESTART_SEC + 2),
                         _result={"ordersFinished": n0, "ordersCutOff": 0})
                self.phase, self.phase_since = "stopping", utc()
                return "restart"
            elif now >= a["_until"]:
                what = "drain did not finish" if a["_op"] == "drain" else "restart did not happen"
                self.life_end(a, "failed", result={"ordersFinished": n0 - left, "ordersRunning": left}, error=(
                    f"{left} order(s) were still running after {a['_limit']} s, so the {what}. "
                    + ("Nothing was stopped: they go on, and the shop takes new orders again."
                       if a["_op"] == "drain" else "It was not forced: the orders go on and the shop takes new orders "
                       "again. Ask again later, or ask for mode force, which a person must approve.")))
            return None

    def restarted(self, t, down_sec):
        """The shop is back after a restart: state reloaded, the HEART's records kept (as if read back from disk)."""
        with self.lock:
            self.shop.reload(t)
            self.t, self.started_t, self.phase, self.phase_since, self.drained_until = t, t, "running", utc(), None
            self.prev = ["ok"] * WORKERS
            self.event("system.started", "system", "info", "The shop server restarted: workers new, cache cold")
            a = self.life
            if a:
                self.life_end(a, "succeeded", result=dict(a.pop("_result"), downSec=round(down_sec, 1),
                                                          note="Restarted: running again, with its state reloaded"))


def view(a):
    """An Action as served: without the simulator's own working fields (those starting with _)."""
    return {k: v for k, v in a.items() if not k.startswith("_")}


def problem(status, code, detail):
    return status, {"type": f"https://heartcortex.ai/problems/{code}", "title": code.replace("-", " "), "status": status,
                    "detail": detail, "code": code}, "application/problem+json"


OPS = {"<": lambda a, b: a < b, "<=": lambda a, b: a <= b, ">": lambda a, b: a > b, ">=": lambda a, b: a >= b,
       "==": lambda a, b: a == b, "!=": lambda a, b: a != b}


# ------------------------------------------------------------------------------------------------ HEART Monitor

def identity(path, url, new=False):
    """Made once from a cryptographic random source, never from anything public."""
    if not new and os.path.exists(path):
        with open(path) as f:
            return json.load(f)
    ident = {"id": secrets.token_hex(8), "claimCode": "".join(secrets.choice(CROCKFORD) for _ in range(20)),
             "key": secrets.token_urlsafe(32)}
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(ident, f)
    return ident


def show_claim(ident, url):
    code = "-".join(ident["claimCode"][i:i + 5] for i in range(0, 20, 5))
    print(f"""
  HEART Monitor: register this simulated shop server
    1. Sign in at {url}/login with your email (a 6-digit code is sent to you).
    2. Open {url}/register?id={ident['id']}
    3. Enter the claim code  {code}
    4. Watch it at {url}/device/{ident['id']}  (private to you until you share it)
""", file=sys.stderr, flush=True)


def post(url, path, body, key=None):
    req = urllib.request.Request(url.rstrip("/") + path, data=json.dumps(body).encode(), method="POST",
                                 headers={"Content-Type": "application/json", "User-Agent": "heart-online-shop-simulator/1.0",
                                          **({"Authorization": f"Bearer {key}"} if key else {})})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except ValueError:
            return e.code, {}


def monitor(heart, url, path, stop):
    """Outbound only. Reports at most every 10 seconds, and at once when something changes."""
    host = url.split("://", 1)[-1].split("/")[0].split(":")[0]
    if not url.startswith("https://") and host not in ("127.0.0.1", "localhost"):
        log("HEART Monitor: refusing a site without https")
        return
    ident, phase, next_at, last, sent, backoff, claimed = identity(path, url), "hello", 0.0, 0.0, None, 0, None
    show_claim(ident, url)
    while not stop.is_set():
        if heart.down_until:                     # restarting: a real process would be silent too
            stop.wait(1)
            continue
        with heart.lock:
            seen = (heart.status, tuple((a["code"], a["status"]) for a in heart.alerts[-16:]))
        due = time.time() >= next_at or (phase == "report" and seen != sent and time.time() - last >= 10)
        if not due:
            stop.wait(1)
            continue
        try:
            if phase == "hello":
                status, reply = post(url, f"/v1/devices/{ident['id']}/hello", {
                    "claimHash": hashlib.sha256(f"heart-claim:{ident['claimCode']}".encode()).hexdigest(),
                    "keyHash": hashlib.sha256(ident["key"].encode()).hexdigest(), "template": TEMPLATE})
                if status in (200, 201):
                    phase, next_at = "report", 0
                    log(f"HEART Monitor: {url} knows this device")
                elif status == 409:
                    ident = identity(path, url, new=True)
                    show_claim(ident, url)
                else:
                    log(f"HEART Monitor: hello answered {status}; trying again in 10 minutes")
                    next_at = time.time() + 600
                continue
            last, sent = time.time(), seen
            status, reply = post(url, f"/v1/devices/{ident['id']}/report",
                                 {"pulse": heart.pulse(), "alerts": heart.public_alerts()}, ident["key"])
            backoff = 0
            if status in (200, 202) and reply.get("claimed") != claimed:
                claimed = reply.get("claimed")
                log("HEART Monitor: registered, reports are now stored" if claimed else
                    "HEART Monitor: not registered yet, nothing is stored (see the claim code above)")
            if status in (200, 202, 429):
                next_at = time.time() + int(reply.get("nextSec") or PULSE_SEC)
            elif status == 404:
                phase = "hello"
            elif status == 401:
                log("HEART Monitor: the site refused this device's key. Reporting has stopped.")
                return
            else:
                log(f"HEART Monitor: report answered {status} {reply.get('detail', '')}; not retrying it unchanged")
                next_at = time.time() + 600
        except (OSError, ValueError) as e:
            backoff = min(backoff * 2 or 30, 600)
            log(f"HEART Monitor: offline ({e.__class__.__name__}); trying again in {backoff} s. Noticing goes on.")
            next_at = time.time() + backoff


# ------------------------------------------------------------------------------------------------ the HEART API

BUTTON = "button.press"      # proof of presence in this simulator: `touch button.press` in the working directory


def pressed():
    """The local button: a file touched in the last 120 seconds, used once."""
    try:
        fresh = time.time() - os.path.getmtime(BUTTON) <= 120
        os.remove(BUTTON)
        return fresh
    except OSError:
        return False


def handler(heart):
    class H(BaseHTTPRequestHandler):
        def send(self, status, body, ctype="application/cloudevents+json", headers=None):
            raw = json.dumps(body, indent=1).encode()
            self.send_response(status)
            self.send_header("Content-Type", ctype)
            self.send_header("Content-Length", str(len(raw)))
            self.send_header("Heart-Version", "1.0")
            for k, v in (headers or {}).items():
                self.send_header(k, v)
            self.end_headers()
            self.wfile.write(raw)

        def problem(self, status, code, detail):
            self.send(status, {"type": f"https://heartcortex.ai/problems/{code}", "title": code.replace("-", " "),
                               "status": status, "detail": detail, "code": code}, "application/problem+json")

        def owner(self):
            return secrets.compare_digest(self.headers.get("Authorization", ""), f"Bearer {heart.owner_key}")

        def body(self):
            n = int(self.headers.get("Content-Length") or 0)
            try:
                return json.loads(self.rfile.read(n) or b"{}") if n else {}
            except ValueError:
                return None

        def do_GET(self):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            port = self.server.server_address[1]
            if path == "/.well-known/heart-card.json":
                return self.send(200, heart.card(False, port))
            if path == "/heart/v1/card":
                return self.send(200, heart.card(owner, port))
            if path == "/heart/v1/pulse":
                return self.send(200, heart.pulse(owner))
            if path == "/heart/v1/arrhythmias":
                if owner:
                    with heart.lock:
                        return self.send(200, {"items": [heart.envelope("heart.arrhythmia", dict(a), "internal")
                                                         for a in reversed(heart.alerts)]}, "application/json")
                return self.send(200, {"items": heart.public_alerts(100)}, "application/json")
            if path == "/heart/v1/properties":
                return self.send(200, {"items": heart.properties(owner)}, "application/json")
            if path == "/heart/v1/lifecycle":    # the phase is public; what is allowed, and the orders, need the key
                return self.send(200, heart.lifecycle_view(owner))
            private = {"/heart/v1/actions", "/heart/v1/tending/rules", "/heart/v1/tending/autonomy", "/heart/v1/rhythm",
                       "/heart/v1/capabilities", "/heart/v1/sales"}
            if (path in private or path.startswith("/heart/v1/actions/")) and not owner:
                return self.problem(403, "insufficient-clearance", "Actions, Playbooks, history and sales need the owner key")
            with heart.lock:
                if path == "/heart/v1/actions":
                    return self.send(200, {"items": [heart.envelope("heart.action", view(a), "private")
                                                     for a in reversed(heart.actions)]}, "application/json")
                if path.startswith("/heart/v1/actions/"):
                    a = next((a for a in heart.actions if a["id"] == path.rsplit("/", 1)[1]), None)
                    return self.send(200, heart.envelope("heart.action", view(a), "private")) if a else \
                        self.problem(404, "not-found", "No such Action")
                if path == "/heart/v1/tending/rules":
                    return self.send(200, {"items": heart.rule_messages()}, "application/json")
                if path == "/heart/v1/tending/autonomy":
                    return self.send(200, heart.control())
                if path == "/heart/v1/rhythm":
                    return self.send(200, {"items": [heart.envelope("heart.rhythm", dict(e), "private")
                                                     for e in reversed(heart.events)]}, "application/json")
                if path == "/heart/v1/capabilities":
                    return self.send(200, {"items": heart.card(True, port)["data"]["capabilities"]}, "application/json")
                if path == "/heart/v1/sales":     # the product's private view: totals only, never a customer
                    s = heart.shop
                    return self.send(200, {"ordersToday": s.orders_today, "revenueTodayUsd": round(s.revenue_today, 2),
                                           "note": "Totals only. Customer data is never part of HEART."}, "application/json")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties, /lifecycle, "
                                "/actions, /tending/rules, /tending/autonomy, /rhythm or /sales")

        def refuse(self, what="This"):
            self.problem(403, "capability-not-allowed",
                         f"{what} is never possible through HEART. HEART may only restart a worker, clear the cache "
                         "or rotate the logs. Orders, refunds, deploys and the payment provider are the owner's, "
                         "in the shop's own admin tools.")

        def idempotent(self, run):
            """Every non-read POST or PUT carries an Idempotency-Key; a repeat returns the first answer."""
            k = self.headers.get("Idempotency-Key", "")
            if not 8 <= len(k) <= 255:
                return self.problem(400, "invalid-request", "Send an Idempotency-Key header (8 to 255 characters)")
            body = self.body()
            if body is None:
                return self.problem(400, "invalid-request", "The body is not JSON")
            with heart.lock:
                if k in heart.idem:
                    same, answer = heart.idem[k]
                    if same != json.dumps(body, sort_keys=True):
                        return self.problem(422, "idempotency-conflict", "This key was used with another body")
                    return self.send(*answer)
                answer = run(body)
                if answer[0] < 400 or answer[0] == 409:
                    heart.idem[k] = (json.dumps(body, sort_keys=True), answer)
                return self.send(*answer)

        def do_POST(self):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            parts = path.split("/")
            if path.startswith("/heart/v1/capabilities/") and path.endswith("/invoke") and len(parts) == 6:
                name = parts[4]
                if name in DESTRUCTIVE:
                    return self.refuse(DESTRUCTIVE[name][0].upper() + DESTRUCTIVE[name][1:])
                if name not in CAPS:
                    return self.problem(404, "not-found", f"No capability {name}. Try restartWorker, clearCache or rotateLogs")
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Invoking needs the owner key")
                return self.idempotent(lambda b: self.invoke(name, b))
            if path == "/heart/v1/tending/stop":
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Stop needs the owner key")
                return self.send(200, heart.stop())
            if path.startswith("/heart/v1/lifecycle/") and len(parts) == 5:
                op = parts[4]
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Lifecycle operations need the owner key")
                if op not in STANDARD_OPS:
                    return self.problem(404, "not-found", f"No lifecycle operation {op}. This shop offers "
                                        f"{' and '.join(LIFECYCLE_OPS)}, and no vendor (x-) operations")
                if op not in LIFECYCLE_OPS:
                    return self.problem(501, "not-implemented", f"This shop does not offer {op}: only "
                                        f"{' and '.join(LIFECYCLE_OPS)} (card.lifecycle.operations)")
                return self.idempotent(lambda b: heart.lifecycle_post(op, b) if isinstance(b, dict) else
                                       problem(400, "invalid-request", "The body must be a JSON object"))
            if path.startswith("/heart/v1/actions/") and path.endswith("/decision") and len(parts) == 6:
                if not owner:
                    return self.problem(403, "insufficient-clearance", "Decisions need the owner key")
                return self.idempotent(lambda b: heart.decide(parts[4], b) if isinstance(b, dict) else
                                       problem(400, "invalid-request", "The body must be a JSON object"))
            if path == "/shop/checkout":
                return self.checkout()
            return self.refuse()

        def checkout(self):
            """The shop's own storefront, not HEART: a simulated customer pays. No customer data, only an optional
            "seconds", how long this checkout takes (a demo of an order that is slow to finish)."""
            b = self.body()
            sec = b.get("seconds") if isinstance(b, dict) else None
            if b is None or not isinstance(b, dict) or (sec is not None and (not isinstance(sec, int) or isinstance(sec, bool)
                                                                            or not 1 <= sec <= 3600)):
                return self.send(400, {"accepted": False, "message": "Send {} or {\"seconds\": 1 to 3600}"}, "application/json")
            with heart.lock:
                why = heart.shop.checkout(sec)
            if why:
                return self.send(503, {"accepted": False, "message": why}, "application/json", {"Retry-After": "60"})
            return self.send(201, {"accepted": True, "order": "demo-" + secrets.token_hex(4)}, "application/json")

        def invoke(self, name, body):
            """A person (the owner key) runs a capability by hand: human and not physical, so it runs at once."""
            args = body.get("arguments")
            props = CAPS[name]["inputSchema"]["properties"]
            if not isinstance(args, dict) or set(args) != set(props) or \
                    not all(isinstance(v, int) and not isinstance(v, bool) for v in args.values()):
                return 400, {"type": "https://heartcortex.ai/problems/invalid-request", "title": "invalid request",
                             "status": 400, "code": "invalid-request", "detail": f"arguments must be {list(props)}, whole numbers"}, \
                    "application/problem+json"
            bad = [k for k, v in args.items() if not props[k]["minimum"] <= v <= props[k]["maximum"]]
            if bad:
                return 422, {"type": "https://heartcortex.ai/problems/out-of-range", "title": "out of range",
                             "status": 422, "code": "out-of-range", "detail": f"{', '.join(bad)} out of range"}, \
                    "application/problem+json"
            a = heart.run(name, args, {"kind": "human", "id": "owner"}, None,
                          body.get("reason") or f"Run by hand by the owner: {CAPS[name]['title'].lower()}")
            if a["status"] == "failed":
                return 409, {"type": "https://heartcortex.ai/problems/guard-failed", "title": "guard failed",
                             "status": 409, "code": "guard-failed", "detail": a["error"], "action": a["id"]}, \
                    "application/problem+json"
            log(f"HEART          the owner ran {name}({json.dumps(args)}) by hand: {json.dumps(a['result'])}")
            return 200, heart.envelope("heart.action", dict(a), "private")

        def do_PUT(self):
            path = self.path.split("?")[0].rstrip("/")
            if path != "/heart/v1/tending/autonomy":
                return self.refuse()
            if not self.owner():
                return self.problem(403, "insufficient-clearance", "Changing autonomy needs the owner key")
            return self.idempotent(self.set_autonomy)

        def set_autonomy(self, body):
            m = body.get("max")
            if (not isinstance(m, int) or isinstance(m, bool) or not (0 <= m <= 2 or m == PROMISED_LEVEL)
                    or set(body) - {"max", "reason", "actionId", "presence"}):
                return 400, {"type": "https://heartcortex.ai/problems/invalid-request", "title": "invalid request",
                             "status": 400, "code": "invalid-request",
                             "detail": f"This simulator offers a max of 0 to 2 (Stop) or {PROMISED_LEVEL}"}, \
                    "application/problem+json"
            if m > heart.max:
                # Raising the ceiling is destructive (API Standard 13.4): a person must be present, not just a key.
                if not pressed():
                    return 403, {"type": "https://heartcortex.ai/problems/presence-required", "title": "presence required",
                                 "status": 403, "code": "presence-required",
                                 "detail": f"Turning automatic action back on needs a person at the server: run "
                                           f"`touch {BUTTON}` in its working directory, then send this again"}, \
                        "application/problem+json"
                heart.max, heart.stopped_at = m, None
                heart.event("tending.autonomy.changed", "tending", "info",
                            f"A person raised the autonomy ceiling to {m}: automatic action is back on (act)")
                log("HEART          a person turned automatic action back on: Playbooks act again.")
            else:
                if m <= 2:
                    heart.stop()
                heart.max = m
            return 200, heart.control()

        do_PATCH = do_DELETE = lambda self: self.refuse()

        def log_message(self, *a):
            pass
    return H


# ------------------------------------------------------------------------------------------------ main

INSTANCE = "sim"


class Api:
    """The local HEART API, which a restart closes and opens again on the same port."""

    def __init__(self, heart, port):
        self.heart = heart
        try:
            self.srv = ThreadingHTTPServer(("127.0.0.1", port), handler(heart))
        except OSError:
            self.srv = ThreadingHTTPServer(("127.0.0.1", 7685), handler(heart))   # the standard fallback port
        self.port = self.srv.server_address[1]
        self.start()

    def start(self):
        if self.srv is None:
            self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), handler(self.heart))
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()

    def stop(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.srv = None


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated online shop server with a HEART, reporting to HEART Monitor. Demo only.")
    ap.add_argument("--monitor-url", default=os.environ.get("HEART_MONITOR_URL", "https://heartcortex.ai"))
    ap.add_argument("--no-monitor", action="store_true", help="do not report to a monitor site")
    ap.add_argument("--state-file", default="online-shop-monitor.json", help="the device ID, claim code and key")
    ap.add_argument("--port", type=int, default=43278, help="local HEART API port (0: none)")
    ap.add_argument("--scenario", default="", help=",".join(SCENARIOS) + " or all")
    ap.add_argument("--speed", type=float, default=1.0, help="simulated seconds per real second (default 1)")
    ap.add_argument("--seed", type=int, default=None, help="random seed, to repeat a run")
    ap.add_argument("--duration", type=float, default=0, help="stop after this many real seconds (0: run until Ctrl-C)")
    args = ap.parse_args()
    scenarios = set(SCENARIOS) if args.scenario == "all" else {s for s in args.scenario.split(",") if s}
    unknown = scenarios - set(SCENARIOS)
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")

    rng = random.Random(args.seed)
    cust, shop = Customers(rng, scenarios), Shop(rng, scenarios)
    heart = Heart(shop)
    if not args.no_monitor and os.path.exists(args.state_file):
        with open(args.state_file) as f:
            INSTANCE = json.load(f).get("id", INSTANCE)
    elif not args.no_monitor:
        INSTANCE = identity(args.state_file, args.monitor_url)["id"]

    print(f"""
  {NAME}
  A DEMONSTRATION. Simulated customers and server; no real orders or customer data.
  HEART fixes three known problems by the owner's Playbooks (restart a stuck worker, clear a bloated cache,
  rotate full logs), within the allowlist's limits. Anything destructive is never possible through HEART.
  Scenarios: {', '.join(sorted(scenarios)) or 'none (random life only)'} · speed {args.speed}x
""", file=sys.stderr, flush=True)

    stop = threading.Event()
    api = Api(heart, args.port) if args.port else None
    if api:
        log(f"HEART API on http://127.0.0.1:{api.port}/heart/v1/pulse  (owner key for private views: {heart.owner_key})")
    if not args.no_monitor:
        threading.Thread(target=monitor, args=(heart, args.monitor_url, args.state_file, stop), daemon=True).start()

    t, wall0, last_print = 0.0, time.time(), 0.0
    try:
        while not stop.is_set():
            dt = args.speed
            t += dt
            if heart.down_until:                 # restarting: nothing is served, the API does not answer
                if time.time() >= heart.down_until:
                    heart.restarted(t, time.time() - down_at)
                    heart.down_until = 0.0
                    if api:
                        api.start()
                    log(f"shop: restarted after {time.time() - down_at:.0f} s. Workers new, cache cold; HEART API answers again")
                time.sleep(1)
                continue
            with heart.lock:
                cust.step(t, wall0 + t)
                shop.step(t, dt, cust, wall0 + t)
            heart.tend(t)
            if heart.lifecycle_tick() == "restart":
                down_at = time.time()
                log(f"shop: restarting now (about {RESTART_SEC} s): the HEART API stops answering")
                if api:
                    api.stop()
                heart.down_until = down_at + RESTART_SEC
                time.sleep(1)
                continue
            if time.time() - last_print >= 15:
                last_print = time.time()
                s = shop
                log(f"{s.rps:5.0f} req/s · errors {100 * s.error_rate:4.1f}% · p95 {s.p95:5.0f} ms · queue {s.queue:5.0f} · "
                    f"workers {s.healthy()}/{WORKERS} · cache {s.hit_rate:3.0f}% · memory {s.memory:3.0f}% · disk free "
                    f"{s.disk_free:4.1f}% · payment {'up' if s.deps['payment'] else 'DOWN'} · HEART {heart.status}")
            if args.duration and time.time() - wall0 >= args.duration:
                break
            time.sleep(1)
    except KeyboardInterrupt:
        pass
    stop.set()
    log("stopped. The monitor site will show a Flatline if this server stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo