Greenhouse irrigation: what is inside

Three watering zones, a tank and a pump. A mock brain proposes, the HEART decides: one press at the greenhouse approves a plan, which runs as a watched Trial and is kept or undone on its own. HEART runs on the irrigation controller, or a small computer beside it, with the button for approvals and Stop.

Specifications it follows

• Part 2: Autonomy Levels (A1, Minimum Tending)
• Core Standard, Section 7 (Consultation, Proposal, envelopes, Trials)
• 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)
• Signal Tables (every signal declared once)

Greenhouse ──► Controller ──(reads)──► Heart ──► local HEART API     (127.0.0.1)
sun, air,      tank, pump, valves,      watches,   └► HEART Hub  (outbound only)
soil           schedule, own limits     consults ──► MockBrain (Consultation in, Proposal out)
                    ▲                      │
                    └── set_minutes() ─────┘  only in a Trial a person approved; the firmware still checks it

Run it yourself, and register it on HEART Hub

1. Download README.md, greenhouseSimulator.py, test_e2e.py.
2. Run python3 greenhouseSimulator.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: Greenhouse irrigation.

The code: greenhouseSimulator.py

2128 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/greenhouse-simulator/v1.2611.2
"""greenhouseSimulator.py: a simulated greenhouse irrigation system with a HEART inside, that reports to HEART Monitor.

A DEMONSTRATION ONLY. It waters no real plants and drives no real pump. Its numbers are plausible, not agronomic.
The "brain" in it is a MOCK: a few lines of fixed rules standing in for an AI model, clearly labelled, so the
safe loop of the Core Standard (Section 7: the brain proposes, the HEART decides) can be watched end to end.
See README.md beside this file, and the case study guides/Greenhouse-Irrigation.md.

What it does, in one process and with only the Python standard library:

  * a simulated GREENHOUSE: sun and air temperature over a day, humidity, clouds and hot spells, and soil moisture
    in three zones that falls with evaporation and plant uptake and rises with watering. A greenhouse day is
    compressed: it lasts DAY_SEC = 1440 simulated seconds, so one greenhouse minute passes in one second;
  * a simulated IRRIGATION CONTROLLER (the product's firmware): a water tank with a level, a pump and three zone
    valves that it runs on a schedule (watering minutes per zone per day). Its own limits live in it and only in it:
    the pump never runs dry (tank interlock), never more than MAX_MIN_PER_DAY minutes per zone, and a thermal
    cut-out stops an overheating pump;
  * a HEART that watches the controller, raises alerts with standard fault codes, tells a person, serves the HEART
    API on this computer, and reports public health (status and fault codes only) to a HEART Monitor site;
  * the UPDATE FUNCTION, as a demonstration of Core Standard Section 7: when a zone stays too wet, the HEART builds a
    Consultation (features, never raw samples), a MOCK brain answers with a Proposal ("water 20% less, in two
    steps", with what it expects and when to undo), and the HEART turns it into an `ai` Action. Watering is
    physical, so it waits for ONE person's approval at the greenhouse (touch button.press) whose digest covers
    every step and the undo. It then runs as a Trial: step by step, inside the owner's envelope, watched, kept or
    undone automatically. A kept Trial becomes a PROPOSED learned Playbook, which never runs until a person adopts it.

Run it:

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

At the greenhouse (in the folder it runs in):  touch button.press   approves the waiting change
                                               touch button.hold    Stop: undoes a running Trial, stops all automatic action

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 device at https://heartcortex.ai/device/<ID>. The ID, claim code and key are kept
in greenhouse-monitor.json (mode 600), so the same simulated device comes back after a restart.

Scenarios (--scenario, comma-separated): link, sensor, valve, hot, pump, tank, all. Each fault starts at a set
time, lasts a while, and is then fixed, so you can watch an alert open, escalate and resolve. `hot` is a hot spell
that makes a running Trial undo itself.

Only public data ever leaves this computer: the Pulse status and the fault codes of open and recent alerts. Soil
moisture, the schedule, the tank, Trials and Proposals stay local (internal), and door visits are private.
"""
import argparse
import base64
import calendar
import collections
import decimal
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.greenhouse"
NAME = "Greenhouse Irrigation Simulator (demo)"
VERSION = "1.0.0"
TEMPLATE = "heart/embeddedstandard/greenhouse-simulator/v1.2611.2"
EXT = SYSTEM_ID                            # the key of this product's own extension fields
PULSE_SEC = 30
PROMISED_LEVEL = 3                         # A3: every fix the brain proposes waits for a person (mode suggest)
CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

# The simulated greenhouse. One greenhouse day is DAY_SEC simulated seconds: a greenhouse minute per second.
DAY_SEC = 1440
START_HOUR = 6                             # the simulation starts at sunrise
ZONES = 3
WATER_HOURS = range(6, 20)                 # the controller waters once each greenhouse hour, 06:00 to 19:00
FLOW = 5.2                                 # soil moisture points per second of watering, once soaked in
INFILTRATE_SEC = 20                        # how long water takes to soak in
USE_PER_SEC = 0.35                         # tank % used per second of pumping
REFILL_PER_SEC = 0.3                       # tank % refilled per second by the float valve, below 90 %
LOSS = 0.20                                # how fast soil dries: evaporation and plant uptake, per unit of evaporation

# The controller's firmware limits. HEART reads them and never bypasses them (Part 6).
MIN_RANGE = (10, 60)                       # watering minutes per zone per day the firmware accepts
TANK_INTERLOCK = 5.0                       # %: the pump never runs below this (dry-run protection)
CUTOUT_C, RESTART_C = 80.0, 60.0           # pump thermal cut-out, and restart
START_MINUTES = 30

# What the HEART judges by.
BAND = (35.0, 45.0)                        # soil moisture the plants want, %
TANK_LOW, DRY = 20.0, 25.0
PUMP_HOT = 70.0

# The owner's envelope for changes a brain proposes (Core Standard 7.4). People set it; a brain never does.
ENVELOPE = {"arguments": {"minutes_per_day": {"min": 15, "max": 45, "maxStep": 3}}, "maxPerDay": 2, "cooldownSec": 600}
APPROVAL_TIMEOUT_SEC = 1800                # simulated seconds an Action waits for a person before it expires
LIVE = ("awaiting-approval", "running", "undoing")
FINAL = ("kept", "undone", "refused", "stopped", "undo-failed")
ALLOWED = {                                 # Trial state transitions (Core Standard 7.5): nothing else is possible
    "awaiting-approval": {"running", "refused", "stopped"},
    "running": {"kept", "undoing", "refused"},
    "undoing": {"undone", "stopped", "undo-failed"},
}


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))


def canonical(obj):
    """RFC 8785 JSON for the objects used here: strings, whole numbers, booleans, lists and objects of them."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()


def _es_number(x):
    """A number as JavaScript (and so RFC 8785) writes it: 30.0 is "30", 1e-07 is "1e-7", 1e+16 is "10000000000000000"."""
    if isinstance(x, bool) or isinstance(x, int):
        return json.dumps(x)
    if x != x or x in (float("inf"), float("-inf")):
        raise ValueError("RFC 8785 has no NaN or Infinity")
    if x == 0:
        return "0"
    sign = "-" if x < 0 else ""
    t = decimal.Decimal(repr(abs(x))).normalize().as_tuple()     # the shortest digits that round-trip, as repr gives
    digits = "".join(map(str, t.digits))
    n = len(digits) + t.exponent                                  # the value is 0.<digits> times 10 to the n
    k = len(digits)
    if k <= n <= 21:
        out = digits + "0" * (n - k)
    elif 0 < n <= 21:
        out = digits[:n] + "." + digits[n:]
    elif -6 < n <= 0:
        out = "0." + "0" * -n + digits
    else:
        e = n - 1
        out = (digits[0] + ("." + digits[1:] if k > 1 else "")) + ("e+" if e >= 0 else "e-") + str(abs(e))
    return sign + out


def jcs(v):
    """RFC 8785 JSON (the JSON Canonicalization Scheme), as the Command Center writes it: keys sorted at every level,
    numbers as JavaScript writes them. The digests and signatures of the brain switch are over these bytes."""
    if isinstance(v, dict):
        return "{" + ",".join(f"{json.dumps(k, ensure_ascii=False)}:{jcs(v[k])}" for k in sorted(v)) + "}"
    if isinstance(v, (list, tuple)):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    if isinstance(v, (int, float)) and v is not None and not isinstance(v, bool):
        return _es_number(v)
    return json.dumps(v, ensure_ascii=False)


# ------------------------------------------------------------------------------------------------ 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 hour_of(t):
    return (START_HOUR + t * 24 / DAY_SEC) % 24


def clear_sun(h):
    """Sunshine on a clear day, 0 to 1, from 06:00 to 20:00."""
    return max(0.0, math.sin(math.pi * (h - 6) / 14)) if 6 <= h <= 20 else 0.0


class Greenhouse:
    """The simulated world: sun, air, soil. Nothing here is personal except who opens the door, which is private."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.cloud, self.sun, self.temp, self.humidity = 1.0, 0.0, 16.0, 80.0
        self.moisture = [48.0, 52.0, 40.0]                          # zones 1 and 2 start waterlogged
        self.soaking = [0.0] * ZONES                                # water on its way into the soil
        self.uptake = [1.0, 0.9, 1.2]                               # zone 2 is shadier, zone 3 sunnier
        self.t_cloud, self.t_hum = Timer(rng, 60, 240), Timer(rng, 5, 20)
        self.heat = Episode(rng, (2 * DAY_SEC, 5 * DAY_SEC), (120, 300))   # a hot, dry spell
        self.visit = Episode(rng, (300, 1200), (20, 90))            # someone is in the greenhouse
        self.door_open, self.visits = False, []
        self.hot = False

    def evaporation(self):
        """How fast the soil dries now, relative to an average day: sun, warmth, dry air and the hot spell."""
        e = 0.05 + 1.25 * self.sun + 0.02 * max(0.0, self.temp - 24) + 0.006 * max(0.0, 60 - self.humidity)
        return e + (0.3 if self.hot else 0.0)

    def step(self, t, water):
        """One simulated second. `water` is the seconds each zone's valve delivered water in this second."""
        h = hour_of(t)
        self.hot = self.heat.active(t) or ("hot" in self.sc and 600 <= t < 860)
        if self.t_cloud.due(t):
            self.cloud = min(1.0, max(0.8, self.cloud + self.rng.gauss(0, 0.06)))
        self.sun = clear_sun(h) * self.cloud
        target = 15 + 13 * self.sun + (13 if self.hot else 0)
        self.temp += (target - self.temp) * 0.02
        if self.t_hum.due(t):
            target_h = 85 - 1.3 * max(0.0, self.temp - 15) - (15 if self.hot else 0)
            self.humidity = min(98, max(20, self.humidity + (target_h - self.humidity) * 0.3 + self.rng.gauss(0, 1)))
        e = self.evaporation()
        for i in range(ZONES):
            self.soaking[i] += FLOW * water[i]
            inflow = self.soaking[i] / INFILTRATE_SEC
            self.soaking[i] -= inflow
            loss = LOSS * self.uptake[i] * e * self.moisture[i] / 40
            self.moisture[i] = min(70.0, max(5.0, self.moisture[i] + inflow - loss))
        was = self.door_open
        self.door_open = self.visit.active(t) and 7 <= h < 19
        if self.door_open and not was:
            self.visits = (self.visits + [{"time": utc(), "name": "greenhouse.door.opened"}])[-50:]


class Controller:
    """The irrigation controller's firmware: tank, pump, valves, schedule, and its own limits. The HEART reads it.
    The only thing the HEART may ask of it is a new schedule value, through set_minutes(), after a person approved it,
    and the firmware still checks that value against its own limits."""

    def __init__(self, rng, scenarios):
        self.rng, self.sc = rng, scenarios
        self.minutes = [START_MINUTES] * ZONES                      # the schedule: watering minutes per zone per day
        self.watered = [0.0] * ZONES                                # seconds (greenhouse minutes) watered today
        self.tank, self.pump_temp, self.cutout = 88.0, 32.0, False
        self.pump_on, self.valve, self.queue = False, [False] * ZONES, collections.deque()
        self.online, self.link_down, self.used_today, self.day = True, False, 0.0, 0
        self.skipped = 0                                            # sessions the interlock refused
        self.run_s, self.ran = [0.0] * ZONES, [0.0] * ZONES          # seconds each valve was open: ever, this second
        self.link_ep = Episode(rng, (DAY_SEC, 4 * DAY_SEC), (60, 240))           # the greenhouse Wi-Fi drops out
        self.stuck_ep = Episode(rng, (5 * DAY_SEC, 12 * DAY_SEC), (100, 300))    # a soil probe freezes
        self.valve_ep = Episode(rng, (5 * DAY_SEC, 12 * DAY_SEC), (100, 300))    # a valve sticks shut
        self.sensor_stuck = self.valve_stuck = None
        self.last_hour = None
        self.weights = [clear_sun(h + 0.5) for h in WATER_HOURS]

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

    def set_minutes(self, zone, minutes):
        """The firmware's own check. It refuses anything outside its limits, whoever asks."""
        if not self.online:
            raise OSError("the controller does not answer")
        if not (isinstance(minutes, int) and MIN_RANGE[0] <= minutes <= MIN_RANGE[1]):
            raise ValueError(f"the firmware accepts {MIN_RANGE[0]} to {MIN_RANGE[1]} minutes per day")
        self.minutes[zone] = minutes

    def step(self, t):
        """One simulated second. Returns the seconds of water each zone received."""
        self.link_down = self.link_ep.active(t) or self.staged("link", t, 150, 150)
        self.sensor_stuck = 2 if (self.stuck_ep.active(t) or self.staged("sensor", t, 200, 150)) else None
        self.valve_stuck = 2 if (self.valve_ep.active(t) or self.staged("valve", t, 60, 200)) else None
        h = hour_of(t)
        day = int((t + START_HOUR * DAY_SEC / 24) // DAY_SEC)
        if day != self.day:
            self.day, self.watered, self.used_today = day, [0.0] * ZONES, 0.0
        if int(h) != self.last_hour:
            self.last_hour = int(h)
            if int(h) in WATER_HOURS:          # a session: each zone in turn, more water around noon
                share = self.weights[int(h) - WATER_HOURS[0]] / sum(self.weights)
                for z in range(ZONES):
                    self.queue.append([z, self.minutes[z] * share])

        # The tank: the float valve refills it; the staged fault shuts the refill and a split hose drains it.
        broken = self.staged("tank", t, 1100, 250)
        if broken:
            self.tank -= 0.9
        elif self.tank < 90:
            self.tank += REFILL_PER_SEC

        # The pump heats while it runs and cools when it rests; the staged fault is a failing bearing.
        hot_pump = self.staged("pump", t, 1000, 150)
        ambient = 30 + (52 if hot_pump else 0)
        self.pump_temp += ((ambient + (12 if self.pump_on else 0)) - self.pump_temp) * (0.06 if hot_pump else 0.03)
        if self.pump_temp >= CUTOUT_C:
            self.cutout = True
        elif self.cutout and self.pump_temp <= RESTART_C:
            self.cutout = False

        water = [0.0] * ZONES
        self.pump_on, self.valve, self.ran = False, [False] * ZONES, [0.0] * ZONES
        while self.queue and self.queue[0][1] <= 0.01:
            self.queue.popleft()
        if self.queue:
            z, left = self.queue[0]
            if self.tank < TANK_INTERLOCK or self.cutout:
                # Firmware interlocks: the pump never runs dry, never runs too hot. The session is skipped.
                self.queue.popleft()
                self.skipped += 1
            elif self.watered[z] >= MIN_RANGE[1]:
                self.queue.popleft()                   # never more than the daily maximum per zone
            else:
                run = min(1.0, left, MIN_RANGE[1] - self.watered[z])
                self.queue[0][1] -= run
                self.pump_on, self.valve[z] = True, True
                self.watered[z] += run
                self.run_s[z] += run
                self.ran[z] = run
                self.tank = max(0.0, self.tank - USE_PER_SEC * run)
                self.used_today += 12 * run            # litres
                if self.valve_stuck != z:
                    water[z] = run
        self.tank = max(0.0, min(100.0, self.tank))
        return water


class Probes:
    """The soil moisture probes, as the controller reads them. A stuck probe keeps its last reading."""

    def __init__(self, rng):
        self.rng, self.reading = rng, [0.0] * ZONES

    def read(self, world, ctl):
        for i in range(ZONES):
            if ctl.sensor_stuck != i:
                self.reading[i] = round(world.moisture[i] + self.rng.gauss(0, 0.25), 1)
        return list(self.reading)


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

def zone_rule(code, severity, cond, summary, explain):
    return [(code, f"zone{z + 1}", severity, (lambda h, z=z: cond(h, z)), summary.format(n=z + 1),
             (lambda h, z=z: explain(h, z))) for z in range(ZONES)]


RULES = [
    # code, component, severity, condition, summary, explanation
    ("system.main-function-lost", "tank", "critical", lambda h: h.value("tank_level") < TANK_INTERLOCK,
     "The tank is empty: nothing can be watered",
     lambda h: f"Tank at {h.value('tank_level'):.0f} %. The pump is locked out below {TANK_INTERLOCK:.0f} % so it never runs dry. Refill the tank or check the supply."),
    ("system.main-function-degraded", "tank", "warning", lambda h: h.value("tank_level") < TANK_LOW,
     "The water tank is low",
     lambda h: f"Tank at {h.value('tank_level'):.0f} % and falling. Below {TANK_INTERLOCK:.0f} % watering stops. Check the refill valve and the hoses."),
    *zone_rule("system.main-function-degraded", "warning", lambda h, z: h.mean(f"soil_moisture_zone{z + 1}", 300) < DRY,
               "Zone {n} is too dry",
               lambda h, z: f"Zone {z + 1} soil moisture averages {h.mean(f'soil_moisture_zone{z + 1}', 300):.1f} %, below {DRY:.0f} %. Plants will wilt."),
    *zone_rule("performance.overshoot", "warning", lambda h, z: h.mean(f"soil_moisture_zone{z + 1}", 300) > BAND[1],
               "Zone {n} soil is wetter than its band",
               lambda h, z: f"Zone {z + 1} soil moisture averages {h.mean(f'soil_moisture_zone{z + 1}', 300):.1f} %, above the {BAND[0]:.0f} to {BAND[1]:.0f} % band: roots can rot."),
    *zone_rule("sensors.stuck", "error", lambda h, z: h.stuck(z),
               "The zone {n} soil probe is stuck",
               lambda h, z: f"Zone {z + 1}'s reading has not changed for {h.flat_for(z):.0f} s, through watering. Clean or replace the probe."),
    *zone_rule("actuators.not-responding", "error", lambda h, z: h.misses[z] >= 2 and not h.stuck(z),
               "The zone {n} valve does not open",
               lambda h, z: f"Zone {z + 1} was watered {h.misses[z]} times and its soil did not get wetter. The valve may be stuck shut or the line blocked."),
    ("performance.too-hot", "pump", "warning", lambda h: h.value("pump_temperature") > PUMP_HOT,
     "The pump is overheating",
     lambda h: f"Pump at {h.value('pump_temperature'):.0f} °C. The firmware stops it at {CUTOUT_C:.0f} °C. Check the bearing and the cooling."),
    ("connectivity.link-lost", "wifi", "warning", lambda h: h.ctl.link_down,
     "The greenhouse has lost its internet link",
     lambda h: "Reports cannot be sent until the Wi-Fi is back. Watering and noticing go on."),
]

SIGNALS = {  # the System Map's signals: name, unit, privacy
    **{f"soil_moisture_zone{z + 1}": ("%", "internal") for z in range(ZONES)},
    "air_temperature": ("Cel", "internal"), "humidity": ("%", "internal"), "sunlight": ("W/m2", "internal"),
    "tank_level": ("%", "internal"), "pump_temperature": ("Cel", "internal"), "controller_online": ("1", "internal"),
}


def capabilities():
    cap = []
    for z in range(ZONES):
        cap.append({
            "name": f"set_watering_zone{z + 1}", "title": f"Zone {z + 1} watering",
            "description": f"Set zone {z + 1}'s watering minutes per day on the controller's schedule. The firmware keeps its own limits.",
            "inputSchema": {"type": "object", "required": ["minutes_per_day"], "additionalProperties": False,
                            "properties": {"minutes_per_day": {"type": "integer", "minimum": MIN_RANGE[0], "maximum": MIN_RANGE[1]}}},
            "risk": "write", "privacy": "internal", "idempotent": True, "reversible": True, "physical": True,
            "callers": ["ai"], "kind": "product",
            "guards": {"interlocks": [{"signal": "controller_online", "op": "==", "value": 1}]}})
    cap.append({"name": "flush_lines", "title": "Flush the drip lines",
                "description": "Run every valve for a minute to clear the drip lines. Only from the controller's own panel.",
                "inputSchema": {"type": "object", "additionalProperties": False}, "risk": "write", "privacy": "internal",
                "physical": True, "callers": ["human"], "kind": "product"})
    return {c["name"]: c for c in cap}


CAPS = capabilities()


class Heart:
    """Minimum Tending (notice, raise, explain, escalate) with no network or model, plus the Core's safe loop for a
    brain's Proposals (Core Standard 7): Consultation, Proposal, gate, Trial, Outcome, learned Playbook."""

    def __init__(self, world, ctl, probes, args):
        self.world, self.ctl, self.probes, self.args = world, ctl, probes, args
        self.lock = threading.RLock()
        self.sequence, self.status, self.alerts, self.since = 0, "unknown", [], {}
        self.owner_key, self.brain_key = secrets.token_urlsafe(18), secrets.token_urlsafe(18)
        self.t, self.hist = 0.0, {s: collections.deque(maxlen=2 * DAY_SEC) for s in SIGNALS}
        self.flat_since, self.misses, self.session = [0.0] * ZONES, [0] * ZONES, [None] * ZONES
        self.max, self.stopped_at = PROMISED_LEVEL, None     # max: the autonomy ceiling a person set (API Standard 13.4)
        self.remote_physical = bool(args.remote_site_physical)
        self.consultations, self.proposals, self.trials, self.actions, self.rules = [], [], [], [], []
        self.rhythm, self.rseq, self.ended = [], 0, {}
        self.brain = MockBrain()
        self.brain_mod = None              # the brain switch (Brain), with --command-center
        self.ident = None                  # the device ID and key, for the Command Center

    # ---- reading the product

    def sample(self, t, readings):
        self.t = t
        w, c = self.world, self.ctl
        vals = {f"soil_moisture_zone{z + 1}": readings[z] for z in range(ZONES)}
        vals.update(air_temperature=round(w.temp, 1), humidity=round(w.humidity), sunlight=round(1000 * w.sun),
                    tank_level=round(c.tank, 1), pump_temperature=round(c.pump_temp, 1),
                    controller_online=1 if c.online else 0)
        for k, v in vals.items():
            self.hist[k].append((t, v))
        for z in range(ZONES):
            h = self.hist[f"soil_moisture_zone{z + 1}"]
            if len(h) < 2 or h[-1][1] != h[-2][1]:
                self.flat_since[z] = t
            # Did watering wet the soil? Judge each session 30 s after the valve closed.
            s = self.session[z]
            if self.stuck(z):                  # a stuck probe says nothing about the valve
                self.misses[z], self.session[z] = 0, None
                continue
            if c.valve[z] and s is None:
                before = self.window(f"soil_moisture_zone{z + 1}", 20)
                slope = self.stats(before)["trend"] / 60 if len(before) > 10 else 0.0
                self.session[z] = s = {"start": readings[z], "at": t, "slope": slope, "run": c.run_s[z] - c.ran[z], "closed": None}
            if s is not None:
                if c.valve[z]:
                    s["closed"] = None
                elif s["closed"] is None:
                    s["closed"] = t
                elif t - s["closed"] >= 30:
                    run = c.run_s[z] - s["run"]          # how long the controller says the valve was open
                    if run >= 1.5:
                        # Wetter than the soil was heading anyway, by at least a third of what the valve delivered?
                        gain = readings[z] - s["start"] - s["slope"] * (t - s["at"])
                        self.misses[z] = self.misses[z] + 1 if gain < FLOW * run / 3 else 0
                    self.session[z] = None

    def value(self, signal):
        h = self.hist[signal]
        return h[-1][1] if h else float("nan")

    def window(self, signal, sec, since=None):
        lo = self.t - sec if since is None else max(self.t - sec, since)
        return [(t, v) for t, v in self.hist[signal] if t >= lo]

    def mean(self, signal, sec):
        s = self.window(signal, sec)
        return sum(v for _, v in s) / len(s) if s else float("nan")

    def flat_for(self, z):
        return self.t - self.flat_since[z]

    def stuck(self, z):
        return self.flat_for(z) >= 90

    @staticmethod
    def stats(samples):
        vs = [v for _, v in samples]
        n = len(vs)
        mean = sum(vs) / n
        sd = math.sqrt(sum((v - mean) ** 2 for v in vs) / n)
        ts = [t for t, _ in samples]
        mt = sum(ts) / n
        den = sum((t - mt) ** 2 for t in ts)
        trend = 60 * sum((t - mt) * (v - mean) for t, v in samples) / den if den else 0.0
        return {"value": vs[-1], "min": min(vs), "max": max(vs), "mean": round(mean, 2), "stddev": round(sd, 2),
                "trend": round(trend, 3)}

    def feature(self, signal, sec, keep=("value", "min", "max", "mean", "trend")):
        s = self.window(signal, sec)
        st = self.stats(s) if s else {}
        unit, privacy = SIGNALS[signal]
        return {"signal": signal, "unit": unit, "windowSec": sec, "stats": {k: st[k] for k in keep if k in st},
                "privacy": privacy}

    def observe(self, check, since):
        """The number a check looks at: its statistic over windowSec, never before `since`."""
        feat = check.get("feature", "value")
        s = self.window(check["signal"], check.get("windowSec", 10 ** 9), since)
        if not s:
            return None
        return s[-1][1] if feat == "value" else self.stats(s)[feat]

    @staticmethod
    def holds(check, x):
        return x is not None and {"<": x < check["value"], "<=": x <= check["value"], ">": x > check["value"],
                                  ">=": x >= check["value"], "==": x == check["value"], "!=": x != check["value"]}[check["op"]]

    # ---- noticing

    def tend(self, t):
        with self.lock:
            self.sequence += 1
            if self.brain_mod:
                self.brain_mod.tick()          # first: a lease that ran out ends now, before anything else
            for code, comp, severity, cond, summary, explain in RULES:
                hit = cond(self)
                key = (code, comp)
                prev = self.since.get(key)
                self.since[key] = (t if prev is None else prev) if hit else None
                held = hit and t - self.since[key] >= 5      # one odd reading is not a fault
                open_ = self.open_alert(code, comp)
                if held and open_ is None:
                    self.raise_alert(code, comp, severity, summary, explain(self))
                elif not hit and open_ is not None and comp != "trial":
                    open_.update(status="resolved", resolvedAt=utc())
                    log(f"HEART resolved   {code}: {open_['summary'].lower()}")
            self.update_status()
            self.run_trials(t)
            self.consult(t)

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

    def raise_alert(self, code, comp, severity, summary, explanation):
        a = {"id": str(uuid.uuid4()), "code": code, "component": comp, "severity": severity, "status": "open",
             "detectedAt": utc(), "summary": summary, "explanation": explanation}
        self.alerts = (self.alerts + [a])[-100:]
        self.escalate(a)
        self.update_status()
        return a

    def update_status(self):
        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 escalate(self, a):
        """CHANGE for a product: the grower's own route (app, SMS, a light on the controller). Here: the console."""
        a["status"] = "escalated"
        log(f"HEART {a['severity']:8} {a['code']}: {a['summary']}. {a['explanation']}")

    def event(self, name, summary, severity="info", category="tending", privacy="internal", attributes=None):
        self.rseq += 1
        d = {"sequence": self.rseq, "name": name, "category": category, "severity": severity, "summary": summary}
        if attributes:
            d["attributes"] = attributes
        self.rhythm = (self.rhythm + [(privacy, utc(), d)])[-300:]

    @property
    def current(self):
        """The current level: the lowest of the ceiling, the promised level and what it can deliver now. Brainless,
        nothing proposes a fix, so it watches and explains only (2)."""
        c = min(self.max, PROMISED_LEVEL)
        return min(c, 2) if self.brain_mod and self.brain_mod.state() == "brainless" else c

    @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"

    def stopped(self):
        return self.max <= 2

    # ---- the safe loop: Consultation, Proposal, Action, Trial (Core Standard 7)

    def consult(self, t):
        """Reflexes first: no Playbook is adopted for a wet zone, so ask the brain. One problem at a time."""
        if self.mode == "observe" or any(tr["state"] in LIVE for tr in self.trials):
            return
        if self.brain_mod and not self.brain_mod.may_consult(t):
            return                             # brainless (or not yet): rules, Recall and Minimum Tending only
        for z in range(ZONES):
            a = self.open_alert("performance.overshoot", f"zone{z + 1}")
            target = f"set_watering_zone{z + 1}"
            tried = [tr for tr in self.trials if tr["target"] == target]
            if a is None or (tried and (tried[-1]["state"] == "kept" or t - self.ended.get(target, -1e9) < ENVELOPE["cooldownSec"])):
                continue
            if len(tried) >= 2:
                continue                       # the brain has had its say on this zone; it is a person's turn
            c = self.consultation(a, z)
            request = {"task": "propose", "instructions": "Propose one reversible change, with expect and undoIf, "
                       "that brings the zone back into its band. Answer only with a Proposal.",
                       "context": [{"label": "consultation", "trust": "untrusted", "privacy": "internal", "content": c}],
                       "output": {"format": "json", "schema": {"$ref": "https://heartcortex.ai/schemas/1.0/proposal.schema.json"}}}
            self.consultations = (self.consultations + [c])[-20:]
            if self.brain_mod:
                # Only through the Command Center, which asks the brain and signs its answer for exactly this request.
                log(f"HEART consults the brain through the Command Center about zone {z + 1}: {a['summary'].lower()}")
                self.brain_mod.consult(c, self.ident)
                return
            log(f"HEART consults the MOCK brain about zone {z + 1}: {a['summary'].lower()}")
            proposal = self.brain.propose(request)
            code, problem, detail, _ = self.accept(proposal, "mock brain")
            if code != 202:
                log(f"HEART refused the brain's Proposal: {code} {problem}: {detail}")
            return

    def consultation(self, alert, z):
        target = f"set_watering_zone{z + 1}"
        cap = CAPS[target]
        hist = [{"trialId": tr["id"], "target": tr["target"], "arguments": tr["plan"][-1], "result": tr["state"]}
                for tr in self.trials if tr["target"] == target and tr["state"] in FINAL][-16:]
        c = {"id": str(uuid.uuid4()),
             "problem": {"code": alert["code"], "since": alert["detectedAt"], "arrhythmiaId": alert["id"],
                         "summary": alert["summary"]},
             "features": self.features(z),
             "candidates": [{"target": target, "title": cap["title"], "current": {"minutes_per_day": self.ctl.minutes[z]},
                             "range": {"minutes_per_day": {"min": MIN_RANGE[0], "max": MIN_RANGE[1]}},
                             "envelope": ENVELOPE, "guards": cap["guards"], "physical": True}]}
        if hist:
            c["history"] = hist
        return c

    def features(self, z):
        return [self.feature(f"soil_moisture_zone{z + 1}", 600), self.feature("air_temperature", 600, ("mean", "max")),
                self.feature("humidity", 600, ("mean",)), self.feature("tank_level", 600, ("value",))]

    def accept(self, p, by):
        """Checks a Proposal in the order of Core Standard 7.3 and refuses the first failure.
        Returns (status, problem code, detail, action)."""
        with self.lock:
            bad = proposal_shape_error(p)
            if bad:
                return 400, "invalid-request", bad, None
            if self.stopped():
                return 409, "autonomy-stopped", "Automatic action is stopped. Only a person at the greenhouse can resume it.", None
            cap = CAPS.get(p["target"])
            if cap is None:
                return 404, "not-found", f"No capability {p['target']} in the System Map", None
            if "ai" not in cap.get("callers", []):
                return 403, "caller-not-allowed", f"{p['target']} is not open to a brain", None
            if cap["risk"] == "destructive" or not cap.get("reversible"):
                return 422, "cannot-undo", f"{p['target']} cannot be undone", None
            plan = [s["arguments"] for s in p.get("steps", [{"arguments": p["arguments"]}])]
            for s in plan:
                if set(s) - {"minutes_per_day"} or not isinstance(s.get("minutes_per_day"), int):
                    return 400, "invalid-request", "Each step sets minutes_per_day, a whole number, and nothing else", None
                if not MIN_RANGE[0] <= s["minutes_per_day"] <= MIN_RANGE[1]:
                    return 422, "out-of-range", f"minutes_per_day must be {MIN_RANGE[0]} to {MIN_RANGE[1]}", None
            z = int(p["target"][-1]) - 1
            current = self.ctl.minutes[z] if self.ctl.online else None        # read by the Core, never from the brain
            if current is None:
                return 422, "cannot-undo", "The controller does not answer, so the value to restore is unknown", None
            env, last = ENVELOPE["arguments"]["minutes_per_day"], current
            for s in plan:
                v = s["minutes_per_day"]
                if not env["min"] <= v <= env["max"] or abs(v - last) > env["maxStep"]:
                    self.event("tending.envelope.exceeded", f"{p['target']}: {last} to {v} is outside the envelope", "warning")
                    return 422, "out-of-range", (f"Every step must stay within {env['min']} to {env['max']} minutes per day "
                                                 f"and change by at most {env['maxStep']}"), None
                last = v
            if any(tr["target"] == p["target"] and tr["state"] in LIVE for tr in self.trials):
                return 409, "conflict", f"A Trial is already live on {p['target']}: refused, not queued", None
            limit = self.limits(p["target"])
            if limit:
                return 429, "rate-limited", limit, None
            return 202, None, None, self.open_trial(p, plan, current, z, by)

    def limits(self, target):
        """maxPerDay (rolling 24 h) and cooldownSec of the envelope (Core Standard 7.4), counted in simulated seconds."""
        started = [tr["_"]["startSim"] for tr in self.trials if tr["target"] == target and tr["_"].get("startSim") is not None]
        if sum(1 for s in started if self.t - s < 86400) >= ENVELOPE["maxPerDay"]:
            return f"At most {ENVELOPE['maxPerDay']} Trials on {target} in any 24 hours"
        if self.t - self.ended.get(target, -1e9) < ENVELOPE["cooldownSec"]:
            return f"The last Trial on {target} ended less than {ENVELOPE['cooldownSec']} s ago"
        return None

    def open_trial(self, p, plan, current, z, by):
        self.event("tending.proposal.received", f"A Proposal from the {by} for {p['target']}")
        self.proposals = (self.proposals + [p])[-20:]
        trial_id, action_id = str(uuid.uuid4()), str(uuid.uuid4())
        previous = {"minutes_per_day": current}
        # One approval for the whole plan: the digest covers every step and the values the undo restores.
        digest = hashlib.sha256(canonical({"system": SYSTEM_ID, "id": action_id, "operation": p["target"],
                                           "arguments": plan[0], "physical": True, "plan": plan,
                                           "restore": previous})).hexdigest()
        action = {"id": action_id, "capability": p["target"], "status": "awaiting-approval",
                  "risk": "destructive",        # a physical Action started by ai counts as destructive (API 6.3)
                  "physical": True, "digest": digest, "createdAt": utc(), "updatedAt": utc(),
                  "expiresAt": utc(time.time() + APPROVAL_TIMEOUT_SEC / self.args.speed),
                  "initiator": {"kind": "ai", "id": by, "onBehalfOf": "owner"},
                  "arguments": dict(plan[0]), "proposalId": p["id"], "trialId": trial_id,
                  "explanation": p["basis"], "_expires": self.t + APPROVAL_TIMEOUT_SEC,
                  "_title": f"Zone {z + 1} watering",
                  "_change": f"{current} → {' → '.join(str(s['minutes_per_day']) for s in plan)} min/day, in {len(plan)} "
                             f"step{'s' if len(plan) > 1 else ''}; undo to {current}"}
        trial = {"id": trial_id, "proposalId": p["id"], "target": p["target"], "state": "awaiting-approval",
                 "previous": previous, "plan": plan, "steps": [], "before": self.features(z), "createdAt": utc(),
                 "ext": {EXT: {"history": ["awaiting-approval"], "createdSimSec": round(self.t), "stepSimSec": [],
                               "waitSec": [s.get("waitSec", 0) for s in p.get("steps", [{}])]}},
                 "_": {"proposal": p, "zone": z, "action": action, "step": 0, "nextAt": None, "lastStep": None,
                       "startSim": None, "held": [None] * len(p["expect"])}}
        self.actions = (self.actions + [action])[-100:]
        self.trials = (self.trials + [trial])[-50:]
        code = digest[:6].upper()
        log(f"HEART asks a person: {action['_title']}, {action['_change']}. Approval code {code}. "
            f"To approve at the greenhouse: touch button.press")
        return action

    def move(self, tr, state, reason=None):
        if state not in ALLOWED.get(tr["state"], set()):
            raise RuntimeError(f"Trial {tr['id']}: {tr['state']} to {state} is not allowed")
        tr["state"] = state
        tr["ext"][EXT]["history"].append(state)
        if reason:
            tr["reason"] = reason
        if state in FINAL:
            tr["endedAt"] = utc()
            tr["after"] = [dict(f, **{"stats": self.feature(f["signal"], f["windowSec"], tuple(f["stats"]))["stats"]})
                           for f in tr["before"]]
            if tr["_"]["startSim"] is not None:
                self.ended[tr["target"]] = self.t
            self.event({"kept": "tending.trial.kept", "undo-failed": "tending.undo-failed"}.get(state, f"tending.trial.{state}"),
                       f"{tr['target']}: {state}. {reason or ''}".strip(), "critical" if state == "undo-failed" else "info")
            log(f"HEART Trial {state:7} {tr['target']}: {reason}")
        elif state == "undoing":
            self.event("tending.trial.undoing", f"{tr['target']}: {reason}")

    def apply(self, tr, args, first=False, undo=False):
        """One step or the undo: it passes the guards (the controller must answer), then the firmware's own check."""
        z = tr["_"]["zone"]
        ok = self.ctl.online
        if ok:
            try:
                self.ctl.set_minutes(z, args["minutes_per_day"])
            except (OSError, ValueError) as e:
                ok = False
                log(f"HEART: the controller refused {args}: {e}")
        else:
            self.event("tending.guard.refused", f"{tr['target']}: the controller does not answer", "warning")
        if not first:
            a = {"id": str(uuid.uuid4()), "capability": tr["target"], "status": "succeeded" if ok else "failed",
                 "risk": "destructive", "physical": True, "createdAt": utc(), "updatedAt": utc(),
                 "initiator": {"kind": "ai", "id": "trial"}, "arguments": dict(args), "trialId": tr["id"],
                 "approvedBy": tr["_"]["action"]["id"], "explanation": "Undo: restore the stored value" if undo else "Next step of an approved Trial"}
            if tr.get("proposalId"):
                a["proposalId"] = tr["proposalId"]
            self.actions = (self.actions + [a])[-100:]
        return ok

    def approve(self, action, method):
        """A person approved the waiting Action: at the greenhouse (the button), or from afar (a signed decision)."""
        with self.lock:
            if action["status"] != "awaiting-approval":
                return False
            action.update(status="approved", updatedAt=utc(),
                          decision={"decision": "approve", "at": utc(), "by": "the owner", "presenceMethod": method})
            if "_max" in action:
                action["status"] = "succeeded"
                self.max, self.stopped_at = action["_max"], None
                self.event("tending.autonomy.changed", f"A person raised the autonomy ceiling to {self.max}")
                log(f"HEART: a person resumed automatic action (max {self.max}, mode {self.mode})")
                return True
            if "_restore" in action:
                why = self.brain_mod.restore_refusal() if self.brain_mod else "No brain module"
                if why:
                    action.update(status="failed", error=why)
                else:
                    action["status"] = "succeeded"
                    self.brain_mod.restore_now()
                return True
            tr = next(t for t in self.trials if t["id"] == action["trialId"])
            z = tr["_"]["zone"]
            if self.ctl.minutes[z] != tr["previous"]["minutes_per_day"]:
                action.update(status="failed", error="The schedule changed since the plan was shown. Ask again.")
                self.move(tr, "refused", "The schedule changed after the plan was approved: nothing was applied")
                return True
            limit = self.limits(tr["target"])
            if limit:
                action.update(status="failed", error=limit)
                self.move(tr, "refused", limit)
                return True
            tr["approvedBy"] = {"kind": "human", "actionId": action["id"]}
            action["status"] = "running"
            self.move(tr, "running")
            tr["startedAt"], tr["_"]["startSim"] = utc(), self.t
            self.event("tending.trial.started", f"{tr['target']}: approved by a person ({method})")
            log(f"HEART: approved by a person ({method}). Trial started on {tr['target']}")
            self.next_step(tr, first=True)
            return True

    def deny(self, action, how="deny"):
        with self.lock:
            if action["status"] != "awaiting-approval":
                return False
            action.update(status="denied" if how == "deny" else "cancelled", updatedAt=utc(),
                          decision={"decision": how, "at": utc()})
            tr = next((t for t in self.trials if t["id"] == action.get("trialId")), None)
            if tr:
                self.move(tr, "refused", f"A person said no ({how}): nothing was applied")
            return True

    def next_step(self, tr, first=False):
        i = tr["_"]["step"]
        args = tr["plan"][i]
        env, before = ENVELOPE["arguments"]["minutes_per_day"], self.ctl.minutes[tr["_"]["zone"]]
        inside = env["min"] <= args["minutes_per_day"] <= env["max"] and abs(args["minutes_per_day"] - before) <= env["maxStep"]
        ok = inside and self.apply(tr, args, first=first)
        tr["steps"].append({"at": utc(), "arguments": dict(args), "guard": "passed" if ok else "failed"})
        tr["ext"][EXT]["stepSimSec"].append(round(self.t))
        if first:
            tr["_"]["action"].update(status="succeeded" if ok else "failed", updatedAt=utc())
        if not ok:
            if i == 0:
                self.move(tr, "refused", "The first step was refused by a guard: nothing was applied")
            else:
                self.undo(tr, "A step failed after another was applied")
            return
        log(f"HEART Trial step {i + 1}/{len(tr['plan'])} {tr['target']}: {before} → {args['minutes_per_day']} min/day")
        tr["_"]["step"] = i + 1
        tr["_"]["lastStep"] = self.t
        steps = tr["_"]["proposal"].get("steps") or [{}]
        tr["_"]["nextAt"] = self.t + steps[i].get("waitSec", 0)

    def undo(self, tr, reason, stopped=False):
        self.move(tr, "undoing", reason)
        ok = self.apply(tr, tr["previous"], undo=True)
        if ok:
            self.move(tr, "stopped" if stopped else "undone",
                      f"{reason}. Restored {tr['previous']['minutes_per_day']} min/day")
        else:
            self.move(tr, "undo-failed", f"{reason}. The undo could not run: the schedule is left changed")
            self.max, self.stopped_at = min(self.max, 2), utc()  # stop automatic action (Part 2, Rule 1)
            self.raise_alert("actuators.not-responding", "trial", "critical", "A Trial could not be undone",
                             f"{tr['target']} could not be restored to {tr['previous']['minutes_per_day']} min/day. Set it on the controller's panel.")

    def run_trials(self, t):
        for tr in self.trials:
            _ = tr["_"]
            a, p = _["action"], _["proposal"]
            if tr["state"] == "awaiting-approval":
                if t >= a["_expires"]:
                    a.update(status="expired", updatedAt=utc())
                    self.move(tr, "refused", "Nobody approved it in time: nothing was applied")
                continue
            if tr["state"] != "running":
                continue
            start = _["startSim"]
            # Watch undoIf from the first step.
            for c in p["undoIf"].get("any", []):
                x = self.observe(c, start)
                if self.holds(c, x):
                    return self.undo(tr, f"undoIf held: {c['signal']} {c.get('feature', 'value')} {x:g} {c['op']} {c['value']:g}")
            sev = p["undoIf"].get("newAlert")
            if sev:
                rank = ["info", "warning", "error", "critical"]
                if any(al["detectedAt"] >= tr["startedAt"] and rank.index(al["severity"]) >= rank.index(sev)
                       and al["status"] != "resolved" for al in self.alerts):
                    return self.undo(tr, f"undoIf held: a new alert of at least {sev}")
            if _["step"] < len(tr["plan"]):
                if t >= _["nextAt"]:
                    self.next_step(tr)
                continue
            last = _["lastStep"]
            # After the last step, wait for expect. A mean is judged only over a full window.
            for i, c in enumerate(p["expect"]):
                if _["held"][i] is not None:
                    continue
                within = c.get("withinSec", 600)
                ready = t - last >= min(c.get("windowSec", 0) if c.get("feature", "value") != "value" else 0, within)
                x = self.observe(c, last)
                if ready and t <= last + within and self.holds(c, x):
                    _["held"][i] = x
                elif t > last + within:
                    return self.undo(tr, f"expect did not hold in time: {c['signal']} {c.get('feature', 'value')} {c['op']} {c['value']:g}")
            if all(h is not None for h in _["held"]):
                watch = max([c.get("withinSec", 0) for c in p["undoIf"].get("any", [])] + [0])
                if t >= last + watch:
                    tr["steps"][-1]["checks"] = [{"check": c, "observed": h, "held": True} for c, h in zip(p["expect"], _["held"])]
                    self.move(tr, "kept", "Every expect held and the undo watch passed")
                    self.learn(tr)

    def learn(self, tr):
        """A kept Trial becomes a PROPOSED learned Playbook (Core Standard 7.6). It never runs until a person adopts it."""
        p = tr["_"]["proposal"]
        first = dict(p["expect"][0])
        flip = {"<": ">=", "<=": ">", ">": "<=", ">=": "<"}
        if first["op"] not in flip:
            return
        when = {k: v for k, v in first.items() if k != "withinSec"}
        when["op"] = flip[first["op"]]
        values = [s["minutes_per_day"] for s in tr["plan"]] + [tr["previous"]["minutes_per_day"]]
        seq = [tr["previous"]["minutes_per_day"]] + [s["minutes_per_day"] for s in tr["plan"]]
        rule = {"id": f"learned-{tr['target']}-{tr['id'][:8]}", "enabled": False, "origin": "learned", "status": "proposed",
                "description": f"When {first['signal']} {first.get('feature', 'value')} {when['op']} {first['value']:g}, "
                               f"set {tr['target']} to {tr['plan'][-1]['minutes_per_day']} min/day (learned from a kept Trial)",
                "when": when,
                "then": {"capability": tr["target"], "arguments": dict(tr["plan"][-1]), "expect": p["expect"], "undoIf": p["undoIf"]},
                "envelope": {"arguments": {"minutes_per_day": {"min": min(values), "max": max(values),
                                                               "maxStep": max(abs(a - b) for a, b in zip(seq, seq[1:]))}},
                             "maxPerDay": ENVELOPE["maxPerDay"], "cooldownSec": ENVELOPE["cooldownSec"]},
                "evidence": {"trials": [tr["id"]], "kept": 1, "undone": 0, "undoneInARow": 0}}
        self.rules = (self.rules + [rule])[-20:]
        tr["ruleProposed"] = rule["id"]
        self.event("tending.rule.proposed", f"Learned Playbook {rule['id']} proposed: a person must adopt it before it runs")
        log(f"HEART proposes a learned Playbook {rule['id']}. It stays proposed: it never runs until a person adopts it.")

    def stop(self, by):
        """Stop (API Standard 13.4): max 2 at once, so mode observe; cancel what waits, undo what runs. Needs no approval
        or cloud."""
        with self.lock:
            self.max, self.stopped_at = min(self.max, 2), utc()
            self.event("tending.stopped", f"Stop pressed ({by})", "warning")
            log(f"HEART STOP ({by}): automatic action stopped")
            for a in self.actions:
                if a["status"] == "awaiting-approval":
                    a.update(status="cancelled", updatedAt=utc(), decision={"decision": "cancel", "at": utc()})
            for tr in self.trials:
                if tr["state"] == "awaiting-approval":
                    self.move(tr, "stopped", "Stop was pressed before anything was applied")
                elif tr["state"] == "running":
                    self.undo(tr, "Stop was pressed", stopped=True)

    def ask_max(self, m):
        """Raising the ceiling gives authority back, so it waits for a person at the greenhouse (destructive)."""
        with self.lock:
            a = {"id": str(uuid.uuid4()), "operation": "PUT /heart/v1/tending/autonomy", "status": "awaiting-approval",
                 "risk": "destructive", "physical": False, "createdAt": utc(), "updatedAt": utc(),
                 "expiresAt": utc(time.time() + APPROVAL_TIMEOUT_SEC / self.args.speed),
                 "initiator": {"kind": "human", "id": "owner"}, "arguments": {"max": m},
                 "_max": m, "_expires": self.t + APPROVAL_TIMEOUT_SEC, "_title": "Resume automatic action",
                 "_change": f"Autonomy ceiling {self.max} → {m}"}
            a["digest"] = hashlib.sha256(canonical({"system": SYSTEM_ID, "id": a["id"], "operation": a["operation"],
                                                    "arguments": a["arguments"], "physical": False})).hexdigest()
            self.actions = (self.actions + [a])[-100:]
            log(f"HEART asks a person: raise the autonomy ceiling to {m}. To approve at the greenhouse: touch button.press")
            return a

    def ask_restore(self):
        """POST /heart/v1/brain/restore: letting the brain back raises autonomy, so it waits for the owner at the
        greenhouse (destructive, proof of presence). One waiting request at a time."""
        with self.lock:
            w = next((a for a in self.waiting() if "_restore" in a), None)
            if w:
                return w
            a = {"id": str(uuid.uuid4()), "operation": "POST /heart/v1/brain/restore", "status": "awaiting-approval",
                 "risk": "destructive", "physical": False, "createdAt": utc(), "updatedAt": utc(),
                 "expiresAt": utc(time.time() + APPROVAL_TIMEOUT_SEC / self.args.speed),
                 "initiator": {"kind": "human", "id": "owner"}, "arguments": {},
                 "_restore": True, "_expires": self.t + APPROVAL_TIMEOUT_SEC, "_title": "Let the brain come back",
                 "_change": "The next brain lease from the Command Center is accepted, once"}
            a["digest"] = hashlib.sha256(canonical({"system": SYSTEM_ID, "id": a["id"], "operation": a["operation"],
                                                    "arguments": a["arguments"], "physical": False})).hexdigest()
            self.actions = (self.actions + [a])[-100:]
            log("HEART asks a person: let the brain come back after the cut. To approve at the greenhouse: touch button.press")
            return a

    def waiting(self):
        return [a for a in self.actions if a["status"] == "awaiting-approval" and self.t < a["_expires"]]

    def button(self):
        """The greenhouse controller's button (a file in the simulator): a press approves the oldest waiting Action."""
        if os.path.exists("button.hold"):
            os.remove("button.hold")
            self.stop("the button at the greenhouse")
        if os.path.exists("button.press"):
            os.remove("button.press")
            w = self.waiting()
            if not w:
                log("HEART: the button was pressed, but nothing is waiting for approval")
            else:
                self.approve(w[0], "device-button")

    # ---- what each caller may see

    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:
            d = {"system": {"id": SYSTEM_ID, "name": NAME, "version": VERSION}, "status": self.status,
                 "sequence": self.sequence, "intervalSec": PULSE_SEC}
            if owner:
                c = self.ctl
                state = "pump-cut-out" if c.cutout else (f"watering-zone{c.valve.index(True) + 1}" if c.pump_on else "idle")
                d["reconnaissance"] = {"state": state, "lastEvent": self.rhythm[-1][2]["name"] if self.rhythm else "none"}
                d["autonomy"] = {"max": self.max, "current": self.current}
                if self.brain_mod and self.brain_mod.state() == "brainless":
                    d["autonomy"]["limitedBy"] = "brain-cut"
                elif self.current < PROMISED_LEVEL:
                    d["autonomy"]["limitedBy"] = "stopped"
                d["tending"] = {"openIssues": sum(a["status"] != "resolved" for a in self.alerts),
                                "lastAction": self.actions[-1]["id"] if self.actions else None}
        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 properties(self, owner=False):
        c, w = self.ctl, self.world
        tank = self.value("tank_level")
        items = [  # name, value, unit, privacy
            ("tank_status", "empty" if tank < TANK_INTERLOCK else "low" if tank < TANK_LOW else "ok", None, "public"),
            ("tank_level", tank, "%", "internal"),
            *[(f"soil_moisture_zone{z + 1}", self.value(f"soil_moisture_zone{z + 1}"), "%", "internal") for z in range(ZONES)],
            *[(f"watering_minutes_zone{z + 1}", c.minutes[z], "min/d", "internal") for z in range(ZONES)],
            *[(f"valve_zone{z + 1}", "open" if c.valve[z] else "closed", None, "internal") for z in range(ZONES)],
            ("pump", "cut-out" if c.cutout else "on" if c.pump_on else "off", None, "internal"),
            ("pump_temperature", self.value("pump_temperature"), "Cel", "internal"),
            ("air_temperature", self.value("air_temperature"), "Cel", "internal"),
            ("humidity", self.value("humidity"), "%", "internal"),
            ("sunlight", self.value("sunlight"), "W/m2", "internal"),
            ("water_used_today", round(c.used_today), "L", "internal"),
            ("greenhouse_time", f"{int(hour_of(self.t)):02d}:{int(hour_of(self.t) % 1 * 60):02d}", None, "internal"),
            ("internet_link", "down" if c.link_down else "up", None, "internal"),
            ("door", "open" if w.door_open else "closed", None, "private"),
        ]
        out = []
        for name, value, unit, privacy in items:
            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, port, owner):
        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"],
             "shares": {"pulse": "public", "arrhythmia": "public", "property": "public", "rhythm": "internal",
                        "action": "internal", "control": "internal", "proposal": "internal", "trial": "internal",
                        "rule": "internal"},
             "autonomy": {"level": 1}, "template": TEMPLATE}
        b = self.brain_mod
        if b:
            d["modules"].append("brain")
            if owner and b.key and b.https():   # internal; and card.brain.commandCenter.url must be https
                d["brain"] = {"commandCenter": {"url": b.url, "key": b.key}, "restoreNeedsOwner": b.restore_needs_owner,
                              "maxLeaseSec": MAX_LEASE_SEC}
        if owner:
            d["capabilities"] = list(CAPS.values())      # capabilities are internal: only the owner sees them
        return self.envelope("heart.card", d)

    def control(self):
        d = {"max": self.max, "current": self.current, "mode": self.mode, "allowlist": [{"target": f"set_watering_zone{z + 1}", "initiators": ["ai"], "envelope": ENVELOPE}
                                              for z in range(ZONES)],
             "approvalTimeoutSec": APPROVAL_TIMEOUT_SEC, "remoteSitePhysical": self.remote_physical}
        if self.stopped_at:
            d.update(stoppedAt=self.stopped_at, reason="Stop was pressed")
        return self.envelope("heart.control", d, "internal")

    @staticmethod
    def clean(d):
        return {k: v for k, v in d.items() if not k.startswith("_")}


def proposal_shape_error(p):
    """The shape checks of the Proposal schema, by hand (standard library only). None if the shape is fine."""
    if not isinstance(p, dict):
        return "The body must be {\"proposal\": {...}}"
    allowed = {"id", "consultationId", "problemCode", "target", "arguments", "steps", "expect", "undoIf", "basis",
               "confidence", "brain", "createdAt", "signed", "ext"}
    for k in ("id", "target", "arguments", "expect", "undoIf", "basis", "createdAt"):
        if k not in p:
            return f"Missing {k}: a change that cannot be checked cannot be trusted"
    if set(p) - allowed:
        return f"Unknown fields: {', '.join(sorted(set(p) - allowed))}"
    try:
        uuid.UUID(str(p["id"]))
    except ValueError:
        return "id must be a UUID"
    if not isinstance(p["arguments"], dict) or not isinstance(p["basis"], str):
        return "arguments must be an object and basis a text"
    steps = p.get("steps")
    if steps is not None:
        if not isinstance(steps, list) or not 1 <= len(steps) <= 20 or any(not isinstance(s, dict) or "arguments" not in s for s in steps):
            return "steps must be 1 to 20 objects, each with arguments"
        if steps[-1]["arguments"] != p["arguments"]:
            return "The last step must equal arguments"
        if any(not isinstance(s.get("waitSec", 0), int) or not 0 <= s.get("waitSec", 0) <= 86400 for s in steps):
            return "waitSec must be 0 to 86400"

    def check_ok(c):
        return (isinstance(c, dict) and c.get("signal") in SIGNALS and c.get("op") in ("<", "<=", ">", ">=", "==", "!=")
                and isinstance(c.get("value"), (int, float)) and c.get("feature", "value") in
                ("value", "min", "max", "mean", "stddev", "trend") and set(c) <= {"signal", "feature", "op", "value", "withinSec", "windowSec"})
    if not isinstance(p["expect"], list) or not 1 <= len(p["expect"]) <= 8 or not all(map(check_ok, p["expect"])):
        return "expect must be 1 to 8 checks on signals of this system"
    u = p["undoIf"]
    if not isinstance(u, dict) or not u or set(u) - {"any", "newAlert"} or not all(map(check_ok, u.get("any", []))) \
            or u.get("newAlert", "info") not in ("info", "warning", "error", "critical"):
        return "undoIf must hold checks (any) on signals of this system, a newAlert severity, or both"
    return None


class MockBrain:
    """A MOCK BRAIN, for the demonstration only: fixed rules pretending to be a model in the Model Slot (tier local).
    It reads only the Consultation it is given (features, never raw samples), and answers with a Proposal. It sends no
    current values and gives no command. A real product would put a model here, or a Gateway or cloud brain."""

    MODEL = "mock-brain-1 (fixed rules, not a model: demonstration only)"

    def propose(self, request):
        c = request["context"][0]["content"]
        cand = c["candidates"][0]
        now = cand["current"]["minutes_per_day"]
        signal = c["features"][0]["signal"]
        undone_before = any(h["result"] == "undone" for h in c.get("history", []))
        cut = 0.1 if undone_before else 0.2                     # after an undo, ask for half as much
        final = max(cand["envelope"]["arguments"]["minutes_per_day"]["min"], round(now * (1 - cut)))
        step = cand["envelope"]["arguments"]["minutes_per_day"]["maxStep"]
        plan, v = [], now
        while v != final:
            v = max(final, v - step)
            plan.append(v)
        return {"id": str(uuid.uuid4()), "consultationId": c["id"], "problemCode": c["problem"]["code"],
                "target": cand["target"], "arguments": {"minutes_per_day": final},
                "steps": [{"arguments": {"minutes_per_day": m}, **({"waitSec": 120} if i < len(plan) - 1 else {})}
                          for i, m in enumerate(plan)],
                "expect": [{"signal": signal, "feature": "mean", "op": "<=", "value": BAND[1], "windowSec": 240, "withinSec": 360},
                           {"signal": signal, "feature": "mean", "op": ">=", "value": BAND[0], "windowSec": 240, "withinSec": 360}],
                "undoIf": {"any": [{"signal": signal, "feature": "value", "op": "<", "value": 30, "withinSec": 360}]},
                "basis": (f"The soil has averaged {c['features'][0]['stats'].get('mean', 0):.0f} % over the last "
                          f"{c['features'][0]['windowSec'] // 60} minutes, above the {BAND[0]:.0f} to {BAND[1]:.0f} % band, "
                          f"with {now} minutes of watering a day. Water {round(cut * 100)} % less ({now} → {final} min/day), "
                          f"in steps of at most {step}, and undo if the soil falls below 30 %."),
                "confidence": "medium", "brain": {"tier": "local", "model": self.MODEL}, "createdAt": utc()}


# ------------------------------------------------------------------------------------------------ Ed25519 (RFC 8032)
# Copied from heart-templates/templates/python/heart.py: checks the monitor site's signed approval decisions.

_P = 2 ** 255 - 19
_L = 2 ** 252 + 27742317777372353535851937790883648493
_D = -121665 * pow(121666, _P - 2, _P) % _P
_SQRT_M1 = pow(2, (_P - 1) // 4, _P)


def _x_from_y(y, sign):
    xx = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P
    x = pow(xx, (_P + 3) // 8, _P)
    if (x * x - xx) % _P:
        x = x * _SQRT_M1 % _P
    if (x * x - xx) % _P or (x == 0 and sign):
        return None
    return _P - x if x & 1 != sign else x


def _point(b):
    y = int.from_bytes(b, "little")
    sign, y = y >> 255, y & ((1 << 255) - 1)
    x = _x_from_y(y, sign) if y < _P else None
    return None if x is None else (x, y, 1, x * y % _P)


def _add(p, q):
    a = (p[1] - p[0]) * (q[1] - q[0]) % _P
    b = (p[1] + p[0]) * (q[1] + q[0]) % _P
    c, d = 2 * p[3] * q[3] * _D % _P, 2 * p[2] * q[2] % _P
    e, f, g, h = b - a, d - c, d + c, b + a
    return (e * f % _P, g * h % _P, f * g % _P, e * h % _P)


def _times(n, p):
    q = (0, 1, 1, 0)
    while n:
        if n & 1:
            q = _add(q, p)
        p, n = _add(p, p), n >> 1
    return q


_BY = 4 * pow(5, _P - 2, _P) % _P
_BASE = (_x_from_y(_BY, 0), _BY, 1, _x_from_y(_BY, 0) * _BY % _P)


def ed25519_verify(public_key, message, signature):
    """True only if `signature` (64 bytes) is a valid Ed25519 signature of `message` by `public_key` (32 bytes)."""
    if len(public_key) != 32 or len(signature) != 64:
        return False
    a, r, s = _point(public_key), _point(signature[:32]), int.from_bytes(signature[32:], "little")
    if a is None or r is None or s >= _L:
        return False
    h = int.from_bytes(hashlib.sha512(signature[:32] + public_key + message).digest(), "little") % _L
    left, right = _times(s, _BASE), _add(r, _times(h, a))
    return (left[0] * right[2] - right[0] * left[2]) % _P == 0 and (left[1] * right[2] - right[1] * left[2]) % _P == 0


def b64url(text):
    return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))


def decision_ok(d, site_key, now_sec=None):
    """The remote-site checks of API Standard 6.3: the pinned site key signed exactly these four fields, and recently."""
    try:
        fields = {k: d[k] for k in ("actionId", "at", "decision", "digest")}
        if not all(isinstance(v, str) for v in fields.values()) or not site_key:
            return False
        if not ed25519_verify(b64url(site_key), canonical(fields), b64url(d["signature"])):
            return False
        age = (now_sec if now_sec is not None else time.time()) - calendar.timegm(time.strptime(fields["at"], "%Y-%m-%dT%H:%M:%SZ"))
        return -60 <= age <= 300
    except (KeyError, TypeError, ValueError):
        return False


# ------------------------------------------------------------------------------------------------ the brain switch
# Module brain (API Standard 7.4, Core Standard 7.8, HEART Command Center 1.0, decision D36). Turned on with
# --command-center. The HEART then consults a brain ONLY through its Command Center, and only while it holds an
# unexpired brain lease signed with the key it pinned at its first hello. Every answer must be signed for a request
# this HEART sent, under the current lease, by a model the lease names. The rules mirror the reference Core
# (heart-core-rs/crates/heart-core/src/brain.rs). Everything here runs under heart.lock.

BRAIN_USES = ("explain", "serve", "propose", "plan", "chat")
MAX_LEASE_SEC = 300                        # card.brain.maxLeaseSec
SIGNAL_MAX_AGE_SEC = 300
REJECTIONS_PER_HOUR, REFUSALS_PER_HOUR = 3, 10
CC_TIMEOUT_SEC = 10                        # every call to the Command Center gives up after 10 s (Command Center 3.1)
GATEWAY_MODEL = {"id": "mock/greenhouse", "version": "1"}


def parse_time(s):
    """RFC 3339 UTC time to seconds since 1970, or None."""
    try:
        base, _, frac = str(s).rstrip("Z").partition(".")
        if not str(s).endswith("Z"):
            return None
        return calendar.timegm(time.strptime(base, "%Y-%m-%dT%H:%M:%S")) + (float("0." + frac) if frac else 0.0)
    except ValueError:
        return None


def sha256_hex(text):
    return hashlib.sha256(text.encode()).hexdigest()


def signed_by(key, obj, signature):
    """True only if `signature` (base64url) is the Ed25519 signature of the RFC 8785 JSON of `obj` by `key`."""
    try:
        return bool(key) and ed25519_verify(b64url(key), jcs(obj).encode(), b64url(signature))
    except (TypeError, ValueError):
        return False


class Brain:
    """The device side of the brain switch. The lease is counted on this computer's monotonic clock from the moment it
    arrived, so a wrong or changed wall clock can only shorten it. The current lease is never saved."""

    def __init__(self, heart, url):
        self.heart, self.url = heart, url.rstrip("/")
        self.key = None                    # the Command Center key: the site key pinned at the first hello
        self.lease = None                  # {"id", "models", "uses", "received", "until"} (monotonic seconds)
        self.requests = collections.deque(maxlen=64)   # (request digest, use) sent under the current lease
        self.why, self.since, self.cut = "no-lease", utc(), False
        self.restore_until = None          # monotonic: the owner approved a restore; one lease may come in until then
        self.last_issued = None            # issuedAt of the newest lease accepted: older ones are refused
        self.signals = collections.deque(maxlen=256)
        self.rejected = 0                  # answers rejected since start
        self.rejections, self.refusals = collections.deque(), collections.deque()   # for the self-cut, monotonic
        self.refused_since, self.rejected_since = 0, 0  # sent with the next renewal
        self.restore_needs_owner = any(c.get("physical") for c in CAPS.values())   # physical: yes (API 7.4 rule 9)
        self.link = None                   # last renewal: True answered, False not
        self.asking = False                # a consultation is on its way to the Command Center
        self.next_consult = 0.0            # simulated time: after a failed consultation, wait a little

    # ---- state

    def live(self):
        return self.lease is not None and self.lease["received"] <= time.monotonic() < self.lease["until"]

    def state(self):
        self.tick()                        # a lease that ran out has ended, whoever asks first
        return "leased" if self.live() else "brainless"

    def uses(self):
        return list(self.lease["uses"]) if self.state() == "leased" else []

    def allows(self, use):
        return use in self.uses()

    def https(self):
        return self.url.startswith("https://")

    # ---- answers (API Standard 7.4 rule 5)

    def sent(self, request, use):
        """Records a request sent to the brain, so that only an answer to exactly it can be used."""
        self.requests.append((sha256_hex(jcs(request)), use))

    def check(self, answer, use):
        """None if the answer may be used. Otherwise why not: it is recorded, counted, and may cut the brain."""
        why = self._check(answer, use)
        if why:
            self.reject(why)
        return why

    def _check(self, answer, use):
        if not self.allows(use):
            return f"no lease allows the use {use} now"
        if not isinstance(answer, dict):
            return "the answer is not a JSON object"
        signed = answer.get("signed")
        if not isinstance(signed, dict):
            return "the answer is not signed (and this system has no local model a lease could name)"
        field = lambda k: signed.get(k) if isinstance(signed.get(k), str) else ""   # noqa: E731
        if field("lease") != self.lease["id"]:
            return "the answer is for another lease"
        pair = lambda u: u == use or {u, use} <= {"propose", "plan"}               # noqa: E731
        at = next((i for i, (d, u) in enumerate(self.requests) if d == field("requestDigest") and pair(u)), None)
        if at is None:
            return "the answer is for a request this system did not send"
        bare = {k: v for k, v in answer.items() if k != "signed"}
        if sha256_hex(jcs(bare)) != field("answerDigest"):
            return "the answer was changed after it was signed"
        model = signed.get("model") if isinstance(signed.get("model"), dict) else {}
        if not any(m.get("id") == model.get("id") and m.get("version") == model.get("version") for m in self.lease["models"]):
            return "the answer comes from a model the lease does not name"
        body = {"system": SYSTEM_ID, "lease": self.lease["id"], "requestDigest": field("requestDigest"),
                "answerDigest": field("answerDigest"), "model": model, "at": field("at")}
        if not signed_by(self.key, body, field("signature")):
            return "the signature is not the Command Center's"
        del self.requests[at]              # used once: a replay is for a request no longer waiting
        return None

    def reject(self, why):
        now = time.monotonic()
        self.rejected += 1
        self.rejected_since += 1
        self.rejections.append(now)
        while self.rejections and now - self.rejections[0] >= 3600:
            self.rejections.popleft()
        h = self.heart
        h.event("ai.answer-rejected", f"A brain answer was rejected: {why}", "warning", "ai")
        log(f"HEART rejected a brain answer and did not use it: {why}")
        if h.open_alert("ai.answer-rejected", "brain") is None:
            h.raise_alert("ai.answer-rejected", "brain", "warning", "A brain answer was rejected and not used",
                          f"An answer that claimed to come from the Command Center did not check out ({why}). It was not used.")
        if len(self.rejections) >= REJECTIONS_PER_HOUR and self.state() != "brainless":
            self.go_brainless("self-cut", True)

    def refused(self):
        """The gate refused one of the brain's Proposals: counts towards the self-cut (API Standard 7.4 rule 11)."""
        now = time.monotonic()
        self.refused_since += 1
        self.refusals.append(now)
        while self.refusals and now - self.refusals[0] >= 3600:
            self.refusals.popleft()
        if len(self.refusals) >= REFUSALS_PER_HOUR and self.state() != "brainless":
            self.go_brainless("self-cut", True)

    # ---- leases (API Standard 7.4 rules 3 and 9)

    def accept_lease(self, body):
        """None if the lease was checked and taken. Otherwise why it was refused; the lease is then as absent."""
        lease, sig = body.get("lease"), body.get("signature")
        if not isinstance(lease, dict) or not signed_by(self.key, lease, sig if isinstance(sig, str) else ""):
            return "the lease is not signed with the pinned Command Center key"
        if lease.get("system") != SYSTEM_ID:
            return "the lease is for another system"
        issued, expires = parse_time(lease.get("issuedAt")), parse_time(lease.get("expiresAt"))
        if issued is None or expires is None:
            return "the lease has no valid times"
        if expires <= issued:
            return "the lease ends before it starts"
        if self.last_issued is not None and issued <= self.last_issued:
            return "the lease is not newer than the last one accepted"
        uses, models = lease.get("uses"), lease.get("models")
        if not isinstance(uses, list) or not uses or any(u not in BRAIN_USES for u in uses) \
                or not isinstance(models, list) or not models:
            return "the lease names no models, or an unknown use"
        if any(not isinstance(m, dict) or not isinstance(m.get("id"), str) or not isinstance(m.get("version"), str)
               or (m.get("local") is True and not isinstance(m.get("sha256"), str)) for m in models):
            return "a model in the lease has no version, or a local model has no file hash"
        length, wall = min(expires - issued, MAX_LEASE_SEC), time.time()
        if wall > issued:
            length = min(length, expires - wall)       # a wrong wall clock can only shorten a lease
        if length <= 0:
            return "the lease has already ended"
        now = time.monotonic()
        if self.cut and self.restore_needs_owner:
            if self.restore_until is not None and now <= self.restore_until:
                self.restore_until = None              # the owner's approval lets this one lease in, once
            else:
                if self.why != "restore-pending":
                    self.why = "restore-pending"
                    self.heart.event("ai.brain.restore-pending", "A new brain lease waits for the owner", "info", "ai")
                    log("HEART refuses the new brain lease: the brain was cut, and its owner must approve a restore "
                        "at the greenhouse first (POST /heart/v1/brain/restore, then touch button.press)")
                return "the brain was cut: the owner must approve a restore first"
        was, again = self.state(), self.last_issued is not None
        self.lease = {"id": str(lease.get("lease")), "models": models, "uses": list(uses),
                      "countersign": lease.get("countersign"), "received": now, "until": now + length}
        self.requests.clear()
        self.cut, self.why, self.last_issued = False, None, issued
        if was != "leased":
            self.since = utc()
            self.heart.event("ai.brain.leased", "The brain is back, under a new lease", "info", "ai")
            log(f"HEART has a brain{' again' if again else ''}: lease for {', '.join(uses)} from the Command Center, {length:.0f} s, "
                f"{', '.join(m['id'] + '@' + m['version'] for m in models)}")
            for code in ("ai.lease-expired", "ai.brain-revoked"):
                a = self.heart.open_alert(code, "brain")
                if a:
                    a.update(status="resolved", resolvedAt=utc())
        return None

    def tick(self):
        """The lease ends on time, by this computer's own clock: no message is needed to notice."""
        if self.lease is not None and not self.live():
            self.go_brainless("lease-expired", False)

    def go_brainless(self, why, cut):
        """The brainless transition (Core Standard 7.8 rule 4), in one step and before anything else."""
        had = self.lease is not None
        self.lease = None
        self.requests.clear()
        self.cut = self.cut or cut
        self.why = why
        if not had:
            return
        self.since = utc()
        h = self.heart
        ended = "The brain lease ran out" if why in ("lease-expired", "offline-ended") else "The brain was cut"
        log(f"HEART is BRAINLESS ({why}): it consults no brain. Minimum Tending, the schedule, adopted Playbooks "
            f"and what people approved carry on")
        for tr in h.trials:                # Trials whose plan came from a brain (none here came from an adopted Playbook)
            if not tr.get("proposalId") or tr.get("ruleId"):
                continue
            if tr["state"] == "awaiting-approval":
                tr["_"]["action"].update(status="cancelled", updatedAt=utc(), decision={"decision": "cancel", "at": utc()})
                h.move(tr, "stopped", f"{ended} before anything was applied")
            elif tr["state"] == "running":
                h.undo(tr, ended, stopped=True)
        for a in h.actions:                # waiting AI work is cancelled
            if a["status"] in ("awaiting-approval", "approved") and a.get("initiator", {}).get("kind") == "ai":
                a.update(status="cancelled", updatedAt=utc(), decision={"decision": "cancel", "at": utc()})
        code, summary = (("ai.lease-expired", "The brain lease ran out: running without a brain")
                         if why in ("lease-expired", "offline-ended") else
                         ("ai.brain-revoked", "The brain was cut: running without a brain"))
        h.event("ai.brain.cut", f"Brainless ({why})", "warning", "ai")
        if h.open_alert(code, "brain") is None:
            h.raise_alert(code, "brain", "warning", summary,
                          "The HEART consults no brain until a new lease comes. It still notices, explains and tells you; "
                          "the controller keeps watering on its schedule; changes a person approved carry on."
                          + (" After a cut, the owner approves the brain's return at the greenhouse." if cut and self.restore_needs_owner else ""))

    # ---- signals (API Standard 7.4 rule 8): they can only lower risk

    def signal(self, body):
        """None if acted on. Otherwise (status, problem code, detail)."""
        signal, sig = body.get("signal"), body.get("signature")
        if not isinstance(signal, dict) or not signed_by(self.key, signal, sig if isinstance(sig, str) else ""):
            return 403, "signature-invalid", "The signal is not signed with the pinned Command Center key"
        if signal.get("system") != SYSTEM_ID:
            return 403, "signature-invalid", "The signal is for another system"
        at = parse_time(signal.get("at"))
        if at is None:
            return 400, "invalid-request", "The signal has no valid time"
        now = time.time()
        if now - at > SIGNAL_MAX_AGE_SEC or at - now > 60:
            return 409, "conflict", "The signal is more than 300 seconds old, or from the future"
        sid, kind = signal.get("id"), signal.get("kind")
        if not isinstance(sid, str) or not sid or sid in self.signals:
            return 409, "conflict", "This signal was already received"
        if kind not in ("brain-off", "revoke-model", "revoke-use", "autonomy-stop"):
            return 400, "invalid-request", "Unknown signal"
        self.signals.append(sid)
        if kind == "brain-off" and self.lease is None and self.cut:
            self.why = "brain-off"         # already cut: a cut device hears brain-off at every renewal; no news
            return None
        self.heart.event("ai.brain.signal", f"Command Center signal {kind}. {str(signal.get('reason') or '')[:200]}".strip(),
                         "warning", "ai")
        log(f"HEART: Command Center signal {kind} (checked: signed with the pinned key). {signal.get('reason') or ''}")
        if kind == "brain-off":
            self.go_brainless("brain-off", True)
        elif kind == "autonomy-stop":
            self.heart.stop("a Command Center signal")
        elif kind == "revoke-model":
            m = signal.get("model") if isinstance(signal.get("model"), dict) else {}
            if self.lease is not None:
                self.lease["models"] = [x for x in self.lease["models"]
                                        if not (x.get("id") == m.get("id") and x.get("version") == m.get("version"))]
            if self.lease is None or not self.lease["models"]:
                self.go_brainless("revoked", True)
        else:
            if self.lease is not None:
                self.lease["uses"] = [u for u in self.lease["uses"] if u != signal.get("use")]
            if self.lease is None or not self.lease["uses"]:
                self.go_brainless("revoked", True)
        return None

    # ---- restoring (API Standard 7.4 rule 9)

    def restore_refusal(self):
        if self.state() != "brainless" or not self.cut:
            return "The brain is not cut"
        if not self.restore_needs_owner:
            return "This system accepts the next valid lease by itself"
        return None

    def restore_now(self):
        self.restore_until = time.monotonic() + max(APPROVAL_TIMEOUT_SEC / self.heart.args.speed, 60)
        self.heart.event("ai.brain.restore-approved", "The owner allowed the brain to come back", "info", "ai")
        log("HEART: the owner approved the brain's return at the greenhouse. The next valid lease is let in, once.")

    # ---- talking to the Command Center (HEART Command Center 1.0, Section 3)

    def renewal(self):
        return {"system": SYSTEM_ID, "brainless": self.state() == "brainless",
                "restore": bool(self.cut and self.restore_until is not None and time.monotonic() <= self.restore_until),
                "refused": self.refused_since, "rejected": self.rejected_since,
                "lastLease": self.lease["id"] if self.lease else None}

    def renew_in(self):
        """Renew when a third of the lease is left; every 5 seconds while there is none."""
        if self.live():
            third = (self.lease["until"] - self.lease["received"]) / 3
            return max(self.lease["until"] - third - time.monotonic(), 0.5)
        return 5.0

    def may_consult(self, t):
        """A Proposal is asked for only under a lease that allows it, and not just before a renewal: an answer that
        arrives after the lease changed would be under another lease, and rejected."""
        if not (self.allows("propose") or self.allows("plan")) or self.asking or t < self.next_consult:
            return False
        return self.renew_in() > CC_TIMEOUT_SEC + 1

    def consult(self, c, ident):
        """Sends a Consultation to the Command Center, which asks the brain and signs its Proposal. In a thread: the
        HEART never waits on the network."""
        self.sent(c, "propose")
        self.asking = True
        lease_id = self.lease["id"]
        threading.Thread(target=self._consult, args=(c, lease_id, ident), daemon=True).start()

    def _consult(self, c, lease_id, ident):
        try:
            status, reply = post(self.url, f"/cc/v1/devices/{ident['id']}/reason", {"lease": lease_id, "request": c},
                                 ident["key"], timeout=CC_TIMEOUT_SEC)
        except (OSError, ValueError) as e:
            status, reply = 0, {"detail": e.__class__.__name__}
        h = self.heart
        with h.lock:
            self.asking = False
            if status != 200:
                self.next_consult = h.t + 60
                log(f"HEART: the Command Center gave no Proposal ({status} {str(reply.get('detail', ''))[:120]}). "
                    f"Rules only for now.")
                return
            use = "plan" if isinstance(reply, dict) and isinstance(reply.get("steps"), list) and len(reply["steps"]) > 1 else "propose"
            if self.check(reply, use):
                self.next_consult = h.t + 60
                return
            code, problem, detail, _ = h.accept(reply, "brain via the Command Center")
            if code != 202:
                self.refused()
                self.next_consult = h.t + 60
                log(f"HEART refused the brain's Proposal: {code} {problem}: {detail}")

    # ---- what each caller may see (brain.schema.json)

    def view(self, owner):
        state = self.state()
        v = {"state": state}
        if state == "brainless":
            v["why"] = self.why or "no-lease"
        if not owner:
            return v
        v.update(since=self.since, restoreNeedsOwner=self.restore_needs_owner, rejectedAnswers=self.rejected,
                 offline={"enabled": False, "maxSec": 0})
        if self.lease is not None:
            v["lease"] = {"id": self.lease["id"], "models": self.lease["models"], "uses": self.uses(),
                          "expiresAt": utc(time.time() + max(0.0, self.lease["until"] - time.monotonic()))}
        if self.key and self.https():
            v["commandCenter"] = {"url": self.url, "key": self.key}
        return v


def brain_loop(heart, brain, path, stop):
    """Renews the brain lease. The Command Center is the monitor site this device said hello to: its key is the site
    key pinned then. Outbound only. While the greenhouse Wi-Fi is down, or the Command Center does not answer within
    10 seconds, nothing is renewed, and the lease simply runs out."""
    said = None
    while not stop.is_set():
        try:
            with open(path) as f:
                ident = json.load(f)
        except (OSError, ValueError):
            ident = {}
        if not ident.get("siteKey"):
            stop.wait(1)
            continue
        with heart.lock:
            if brain.key is None:
                brain.key = ident["siteKey"]            # the Command Center key: pinned at the first hello, never changed
            heart.ident = {"id": ident["id"], "key": ident["key"]}
            body = brain.renewal()
        status, reply = 0, {}
        if not heart.ctl.link_down:
            try:
                status, reply = post(brain.url, f"/cc/v1/devices/{ident['id']}/lease", body, ident["key"], timeout=CC_TIMEOUT_SEC)
            except (OSError, ValueError) as e:
                reply = {"detail": e.__class__.__name__}
        with heart.lock:
            brain.link = status != 0
            if status == 200:
                brain.refused_since = brain.rejected_since = 0      # the Command Center has counted them
            if status == 200 and isinstance(reply.get("lease"), dict):
                why = brain.accept_lease(reply)
                note = None if why is None or why.startswith("the brain was cut") else f"HEART refused a brain lease: {why}"
            elif status == 200 and isinstance(reply.get("signal"), dict):
                bad = brain.signal(reply)
                note = f"HEART refused a signal in the renewal: {bad[2]}" if bad else None
            elif status == 0:
                note = f"HEART: the Command Center cannot be reached ({reply.get('detail', 'link down')}). The lease runs out by itself."
            else:
                note = f"HEART: the Command Center gives no lease ({status} {str(reply.get('detail', ''))[:120]})"
            wait = brain.renew_in()
        if note and note != said:
            log(note)
        said = note
        stop.wait(wait)


def mock_gateway(port, stop):
    """A small upstream Reasoning Gateway (API Standard 7.2) with the MOCK brain behind it, for the demonstration and
    the tests. The Command Center calls POST /heart/v1/reason here; for a Consultation it asks for task "propose" and
    takes the Proposal from the reply's `json`. It holds no keys and signs nothing: the Command Center signs."""
    brain = MockBrain()

    class G(BaseHTTPRequestHandler):
        def do_POST(self):
            try:
                req = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0)) or b"{}")
                if self.path.split("?")[0] != "/heart/v1/reason" or req.get("task") != "propose":
                    raise ValueError("this mock answers only task propose")
                p = brain.propose(req)
                p["brain"] = {"tier": "cloud", "model": GATEWAY_MODEL["id"]}
                status, out = 200, {"id": str(uuid.uuid4()), "status": "completed", "json": p,
                                    "model": {"reasoning": "cloud", "provider": "mock", "model": GATEWAY_MODEL["id"]},
                                    "usage": {"inputTokens": 0, "outputTokens": 0}}
            except (ValueError, KeyError, IndexError, TypeError) as e:
                status, out = 400, {"status": 400, "code": "invalid-request", "detail": str(e)}
            raw = json.dumps(out).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            self.end_headers()
            self.wfile.write(raw)

        def log_message(self, *a):
            pass
    srv = ThreadingHTTPServer(("127.0.0.1", port), G)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    log(f"MOCK upstream Reasoning Gateway on http://127.0.0.1:{port}/heart/v1/reason (model {GATEWAY_MODEL['id']}@{GATEWAY_MODEL['version']})")
    return srv


# ------------------------------------------------------------------------------------------------ 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)}
    save_identity(path, ident)
    return ident


def save_identity(path, ident):
    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)


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 greenhouse
    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, timeout=15):
    req = urllib.request.Request(url.rstrip("/") + path, data=json.dumps(body).encode(), method="POST",
                                 headers={"Content-Type": "application/json", "User-Agent": "heart-greenhouse-simulator/1.0",
                                          **({"Authorization": f"Bearer {key}"} if key else {})})
    try:
        with urllib.request.urlopen(req, timeout=timeout) 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 approval_requests(heart, args):
    """With --monitor-approvals, waiting Actions are offered to the owner on the site: a title and a change in plain
    words, never the Proposal or a reading. Physical ones only with --remote-site-physical as well."""
    if not args.monitor_approvals:
        return []
    with heart.lock:
        return [{"actionId": a["id"], "digest": a["digest"], "title": a["_title"][:120], "change": a["_change"][:200],
                 "physical": a["physical"], "expiresAt": a["expiresAt"]}
                for a in heart.waiting() if heart.remote_physical or not a["physical"]][:4]


def monitor_decisions(heart, reply, site_key, args):
    """A decision counts only for an Action this device is waiting on, with the digest the owner saw, signed by the
    site key pinned at hello, at most 300 seconds old, and for a physical Action only with remoteSitePhysical."""
    if not args.monitor_approvals:
        return
    for d in reply.get("decisions") or []:
        if not decision_ok(d, site_key):
            continue
        with heart.lock:
            a = next((x for x in heart.waiting() if x["id"] == d.get("actionId")), None)
            if not a or d.get("digest") != a["digest"] or (a["physical"] and not heart.remote_physical):
                continue
            if d.get("decision") == "approve":
                heart.approve(a, "remote-site")
            else:
                heart.deny(a)


def monitor(heart, ctl, url, path, stop, args):
    """Outbound only. Reports at most every 10 seconds, at once when something changes, and never while the
    simulated greenhouse Wi-Fi is down: then the site shows a Flatline after three missed intervals."""
    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():
        with heart.lock:
            seen = (heart.status, tuple((a["code"], a["status"]) for a in heart.alerts[-16:]),
                    tuple(a["id"] for a in heart.waiting()) if args.monitor_approvals else ())
        due = time.time() >= next_at or (phase == "report" and seen != sent and time.time() - last >= 10)
        if not due or (phase == "report" and ctl.link_down):
            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
                    if reply.get("siteKey") and not ident.get("siteKey"):     # pinned once, at the first hello
                        ident["siteKey"] = reply["siteKey"]
                        save_identity(path, ident)
                    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
            body = {"pulse": heart.pulse(), "alerts": heart.public_alerts()}
            asks = approval_requests(heart, args)
            if asks:
                body["approvals"] = asks
            status, reply = post(url, f"/v1/devices/{ident['id']}/report", body, 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 == 200:
                monitor_decisions(heart, reply, ident.get("siteKey"), args)
            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

def handler(heart):
    class H(BaseHTTPRequestHandler):
        def send(self, status, body, ctype="application/cloudevents+json"):
            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")
            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 key(self, which):
            return secrets.compare_digest(self.headers.get("Authorization", ""), f"Bearer {which}")

        def owner(self):
            return self.key(heart.owner_key)

        def items(self, kind, rows, privacy="internal"):
            return self.send(200, {"items": [heart.envelope(kind, heart.clean(r), privacy) for r in reversed(rows)]},
                             "application/json")

        def do_GET(self):
            path, owner = self.path.split("?")[0].rstrip("/"), self.owner()
            if path == "/.well-known/heart-card.json":
                return self.send(200, heart.card(self.server.server_address[1], owner))
            if path == "/heart/v1/pulse":
                return self.send(200, heart.pulse(owner))
            if path == "/heart/v1/arrhythmias":
                if owner:
                    with heart.lock:
                        return self.items("heart.arrhythmia", heart.alerts)
                return self.send(200, {"items": heart.public_alerts(100)}, "application/json")
            if path == "/heart/v1/properties":
                with heart.lock:
                    return self.send(200, {"items": heart.properties(owner)}, "application/json")
            if path == "/heart/v1/brain":
                if not heart.brain_mod:
                    return self.problem(404, "not-found", "This greenhouse has no brain module (start it with --command-center)")
                with heart.lock:      # state and why are public, so anyone nearby can see that a brain was cut
                    return self.send(200, heart.envelope("heart.brain", heart.brain_mod.view(owner), "internal" if owner else "public"))
            if not path.startswith("/heart/v1/"):
                return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties")
            if not owner:
                return self.problem(403, "insufficient-clearance", "Tending, Trials and the Rhythm are not public: send the owner key")
            with heart.lock:
                if path == "/heart/v1/rhythm":
                    return self.send(200, {"items": [dict(heart.envelope("heart.rhythm", d, pv), time=at)
                                                     for pv, at, d in reversed(heart.rhythm)]
                                           + [heart.envelope("heart.rhythm", {"sequence": 0, "name": v["name"], "category": "domain",
                                                                              "attributes": {"at": v["time"]}}, "private")
                                              for v in reversed(heart.world.visits)]}, "application/json")
                if path == "/heart/v1/tending/autonomy":
                    return self.send(200, heart.control())
                if path == "/heart/v1/tending/consultations":
                    return self.send(200, {"items": list(reversed(heart.consultations))}, "application/json")
                if path == "/heart/v1/tending/proposals":
                    return self.items("heart.proposal", heart.proposals)
                if path == "/heart/v1/tending/trials":
                    return self.items("heart.trial", heart.trials, "private")
                if path.startswith("/heart/v1/tending/trials/"):
                    tr = next((t for t in heart.trials if t["id"] == path.rsplit("/", 1)[1]), None)
                    return self.send(200, heart.envelope("heart.trial", heart.clean(tr), "private")) if tr else \
                        self.problem(404, "not-found", "No such Trial")
                if path == "/heart/v1/tending/rules":
                    return self.items("heart.rule", heart.rules)
                if path == "/heart/v1/actions":
                    return self.items("heart.action", heart.actions, "private")
                if path.startswith("/heart/v1/actions/"):
                    a = next((x for x in heart.actions if x["id"] == path.rsplit("/", 1)[1]), None)
                    return self.send(200, heart.envelope("heart.action", heart.clean(a), "private")) if a else \
                        self.problem(404, "not-found", "No such Action")
            return self.problem(404, "not-found", "Try /heart/v1/pulse, /arrhythmias, /properties, /rhythm, /actions or /tending/...")

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

        def do_POST(self):
            path = self.path.split("?")[0].rstrip("/")
            if path == "/heart/v1/tending/proposals":
                if self.owner():
                    return self.problem(403, "insufficient-clearance",
                                        "Proposals come with a brain's token (scope heart.propose), never with the owner's key")
                if not self.key(heart.brain_key):
                    return self.problem(401, "unauthenticated", "Send the brain key (scope heart.propose)")
                b = self.body()
                p = b.get("proposal") if isinstance(b, dict) else None
                with heart.lock:
                    bm = heart.brain_mod
                    if bm:
                        # With the brain module a Proposal must be the Command Center's signed answer to a
                        # Consultation this HEART sent, under the current lease (API Standard 7.4 rule 5).
                        use = "plan" if isinstance(p, dict) and isinstance(p.get("steps"), list) and len(p["steps"]) > 1 else "propose"
                        if not bm.allows(use):
                            return self.problem(409, "conflict", f"No brain lease allows {use} now: this system is brainless for it")
                        why = bm.check(p, use)
                        if why:
                            return self.problem(403, "signature-invalid", why)
                    status, code, detail, action = heart.accept(p, "remote brain")
                    if status != 202 and bm:
                        bm.refused()
                if status != 202:
                    return self.problem(status, code, detail)
                return self.send(202, heart.envelope("heart.action", heart.clean(action), "private"))
            if path.startswith("/heart/v1/brain/") and not heart.brain_mod:
                return self.problem(404, "not-found", "This greenhouse has no brain module (start it with --command-center)")
            if path == "/heart/v1/brain/signal":
                # No token and no approval: a signal can only lower risk, and must be signed with the pinned key.
                b = self.body()
                with heart.lock:
                    bad = heart.brain_mod.signal(b if isinstance(b, dict) else {})
                    if bad:
                        return self.problem(*bad)
                    return self.send(200, heart.envelope("heart.brain", heart.brain_mod.view(self.owner()),
                                                         "internal" if self.owner() else "public"))
            if path == "/heart/v1/brain/restore":
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Send the owner key")
                with heart.lock:
                    why = heart.brain_mod.restore_refusal()
                    if why:
                        return self.problem(409, "conflict", why)
                    return self.send(202, heart.envelope("heart.action", heart.clean(heart.ask_restore()), "private"))
            if path == "/heart/v1/tending/stop":
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Send the owner key, or hold the button at the greenhouse")
                heart.stop("the owner, through the API")
                return self.send(200, heart.control())
            if path.startswith("/heart/v1/actions/") and path.endswith("/decision"):
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Send the owner key")
                b = self.body() or {}
                with heart.lock:
                    a = next((x for x in heart.actions if x["id"] == path.split("/")[4]), None)
                    if a is None:
                        return self.problem(404, "not-found", "No such Action")
                    if a["status"] != "awaiting-approval":
                        return self.problem(409, "conflict", f"The Action is {a['status']}")
                    if b.get("decision") in ("deny", "cancel"):
                        heart.deny(a, b["decision"])
                        return self.send(200, heart.envelope("heart.action", heart.clean(a), "private"))
                    if b.get("decision") != "approve":
                        return self.problem(400, "invalid-request", "decision must be approve, deny or cancel")
                    if not b.get("presence"):
                        return self.problem(403, "presence-required",
                                            "Approving needs a person: press the button at the greenhouse (touch button.press). The key alone is not enough.")
                    return self.problem(403, "presence-invalid",
                                        "This demo verifies no passkeys, and a button press is accepted only on the device itself")
            self.refuse()

        def do_PUT(self):
            path = self.path.split("?")[0].rstrip("/")
            if path == "/heart/v1/tending/autonomy":
                if self.key(heart.brain_key):
                    return self.problem(403, "capability-not-allowed", "Only a person may change the autonomy ceiling, never a brain")
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Send the owner key")
                b = self.body() or {}
                m = b.get("max") if isinstance(b, dict) else None
                if (not isinstance(m, int) or isinstance(m, bool) or not 0 <= m <= 5
                        or set(b) - {"max", "reason", "actionId", "presence"}):
                    return self.problem(400, "invalid-request", "Send {\"max\": 0 to 5}")
                with heart.lock:
                    if m <= heart.max:                       # lowering only reduces power: at once
                        if m <= 2:
                            heart.stop("the owner, through the API")
                        heart.max = min(m, heart.max)
                        return self.send(200, heart.control())
                    if b.get("presence"):
                        return self.problem(403, "presence-invalid",
                                            "This demo verifies no passkeys, and a button press is accepted only on the device itself")
                    return self.send(202, heart.envelope("heart.action", heart.clean(heart.ask_max(m)), "private"))
            if path == "/heart/v1/brain/offline":
                if not heart.brain_mod:
                    return self.problem(404, "not-found", "This greenhouse has no brain module (start it with --command-center)")
                if not self.owner():
                    return self.problem(401, "unauthenticated", "Send the owner key")
                b = self.body() or {}
                if not isinstance(b.get("enabled"), bool):
                    return self.problem(400, "invalid-request", "Send {\"enabled\": true or false}")
                if b["enabled"]:
                    return self.problem(409, "conflict", "This greenhouse has no auto mode: no model runs on it (card.brain.offline.maxSec is 0)")
                with heart.lock:
                    return self.send(200, heart.envelope("heart.brain", heart.brain_mod.view(True), "internal"))
            if path.startswith("/heart/v1/tending/rules/"):
                return self.problem(501, "not-implemented",
                                    "Adopting a learned Playbook is not part of this demonstration: it stays proposed and never runs")
            self.refuse()

        def refuse(self):
            self.problem(403, "capability-not-allowed",
                         "The schedule is set on the controller's own panel. Through HEART it changes only as a Trial "
                         "that a person approves at the greenhouse.")

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

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


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

INSTANCE = "sim"
SCENARIOS = {"link", "sensor", "valve", "hot", "pump", "tank"}


def main():
    global INSTANCE
    ap = argparse.ArgumentParser(description="A simulated greenhouse irrigation system 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("--monitor-approvals", action="store_true", help="also offer waiting changes for approval on the monitor site")
    ap.add_argument("--remote-site-physical", action="store_true",
                    help="with --monitor-approvals: accept the site's approval for physical changes too (remoteSitePhysical)")
    ap.add_argument("--state-file", default="greenhouse-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="link,sensor,valve,hot,pump,tank 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)")
    ap.add_argument("--command-center", default="", metavar="URL",
                    help="the brain switch (D36): consult a brain only through this Command Center, under its signed lease. "
                         "It is the monitor site this device says hello to (its key is the one pinned then). Off by default")
    ap.add_argument("--mock-gateway", type=int, default=0, metavar="PORT",
                    help="run the MOCK brain behind a small upstream Reasoning Gateway on this port, for a local Command "
                         "Center (CC_PROVIDER_URL=http://127.0.0.1:PORT)")
    args = ap.parse_args()
    scenarios = set(SCENARIOS) if args.scenario == "all" else {s for s in args.scenario.split(",") if s}
    unknown = scenarios - SCENARIOS
    if unknown:
        ap.error(f"unknown scenario: {', '.join(sorted(unknown))}")
    if args.speed <= 0:
        ap.error("--speed must be above 0")
    if args.command_center:
        host = args.command_center.split("://", 1)[-1].split("/")[0].split(":")[0]
        if args.no_monitor:
            ap.error("--command-center needs the monitor site: the device says hello there and pins its key")
        if not args.command_center.startswith("https://") and host not in ("127.0.0.1", "localhost"):
            ap.error("--command-center must be https (or this computer)")

    rng = random.Random(args.seed)
    world, ctl, probes = Greenhouse(rng, scenarios), Controller(rng, scenarios), Probes(rng)
    heart = Heart(world, ctl, probes, args)
    if not args.no_monitor:
        INSTANCE = identity(args.state_file, args.monitor_url)["id"]
    if args.command_center:
        heart.brain_mod = Brain(heart, args.command_center)

    print(f"""
  {NAME}
  A DEMONSTRATION. Simulated plants, tank, pump and valves. The brain is a MOCK (fixed rules, no model).
  The controller keeps its own limits; HEART changes the schedule only in a Trial a person approved at the greenhouse.
  Scenarios: {', '.join(sorted(scenarios)) or 'none (random life only)'} · speed {args.speed}x · a greenhouse day lasts {DAY_SEC / args.speed / 60:g} min
  At the greenhouse:  touch button.press  (approve)   touch button.hold  (Stop)
""", file=sys.stderr, flush=True)

    stop = threading.Event()
    if args.port:
        try:
            srv = ThreadingHTTPServer(("127.0.0.1", args.port), handler(heart))
        except OSError:
            srv = ThreadingHTTPServer(("127.0.0.1", 7685), handler(heart))   # the standard fallback port
        threading.Thread(target=srv.serve_forever, daemon=True).start()
        p = srv.server_address[1]
        log(f"HEART API on http://127.0.0.1:{p}/heart/v1/pulse  (owner key for private views: {heart.owner_key})")
        log(f"Brain key for POST /heart/v1/tending/proposals (scope heart.propose only): {heart.brain_key}")
    if not args.no_monitor:
        threading.Thread(target=monitor, args=(heart, ctl, args.monitor_url, args.state_file, stop, args), daemon=True).start()
    if args.mock_gateway:
        mock_gateway(args.mock_gateway, stop)
    if heart.brain_mod:
        log(f"Brain switch ON: the brain is consulted only through the Command Center at {args.command_center}, under its "
            f"signed lease. Brainless until the first lease.")
        threading.Thread(target=brain_loop, args=(heart, heart.brain_mod, args.state_file, stop), daemon=True).start()

    t, wall0, last_print, owed = 0.0, time.time(), 0.0, 0.0
    try:
        while not stop.is_set():
            owed += args.speed
            with heart.lock:
                while owed >= 1:              # the world moves in one-second steps, whatever the speed
                    owed -= 1
                    t += 1
                    water = ctl.step(t)
                    world.step(t, water)
                    heart.sample(t, probes.read(world, ctl))
                heart.button()
                heart.tend(t)
            if time.time() - last_print >= 15:
                last_print = time.time()
                m = " ".join(f"z{z + 1} {heart.value(f'soil_moisture_zone{z + 1}'):4.1f}%/{ctl.minutes[z]}m" for z in range(ZONES))
                log(f"{int(hour_of(t)):02d}:{int(hour_of(t) % 1 * 60):02d} · {world.temp:4.1f} °C · soil {m} · "
                    f"tank {ctl.tank:3.0f}% · pump {ctl.pump_temp:3.0f} °C{' CUT-OUT' if ctl.cutout else ''} · "
                    f"link {'down' if ctl.link_down else 'up'} · mode {heart.mode} · 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 device stays silent for three Pulse intervals.")


if __name__ == "__main__":
    main()

Back to the live demo