🚧 Rewrite continous simulation.

.vscode/tasks.json

@@ -1476,6 +1476,233 @@
 			],
 			"isBackground": false,
 			"problemMatcher": []
+		},
+		{
+			"label": "delete old plantenv",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"import pathlib; p = pathlib.Path('src/wago_sim/scenario/plantenv.py'); p.unlink(); print('deleted:', not p.exists())"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: pytest env2",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests/test_continuous.py",
+				"-q",
+				"--tb=short"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "debug load_env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"from wago_sim.scenario.plantenv import load_env, PlantEnv; import envs.pump_station_env as m; print('class:', m.PumpStationEnv); print('is subclass:', issubclass(m.PumpStationEnv, PlantEnv)); print('module plantenv file:', __import__('wago_sim.scenario.plantenv', fromlist=['x']).__file__)"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "pytest load env detail",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests/test_continuous.py::TestPlantEnv::test_load_env_class_and_factory",
+				"-q",
+				"--tb=long",
+				"-s"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: pytest env3",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests/test_continuous.py",
+				"-q",
+				"--tb=short"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: pytest env4",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests/test_continuous.py",
+				"-q",
+				"--tb=short"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "debug missing env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"from wago_sim.scenario.plantenv import load_env; from wago_sim.modules import DI_8; from wago_sim.app import build_app; app = build_app(modules=[DI_8]);\ntry:\n    e = load_env('envs.missing_module', node=app.node)\n    print('NO RAISE; got:', type(e))\nexcept Exception as exc:\n    print('raised:', exc)"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "run debug env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"scripts\\debug_env.py"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "debug in pytest",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests/test_continuous.py::TestPlantEnv::test_load_env_class_and_factory",
+				"-q",
+				"--tb=long"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "rm debug script",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"import pathlib; pathlib.Path('scripts/debug_env.py').unlink(); print('ok')"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: pytest all env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests",
+				"-q",
+				"--tb=short"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: serve env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"wago_sim.cli",
+				"serve",
+				"--node",
+				"config/pump_node.yaml",
+				"--host",
+				"127.0.0.1",
+				"--port",
+				"5020",
+				"--web-port",
+				"8088",
+				"--env",
+				"envs.pump_station_env:PumpStationEnv"
+			],
+			"isBackground": true,
+			"problemMatcher": []
+		},
+		{
+			"label": "env closed-loop probe",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"scripts\\env_probe.py"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "headless sim probe",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"scripts\\sim_probe.py"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "headless sim probe2",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"scripts\\sim_probe.py"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "rm probes",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"import pathlib; [pathlib.Path(p).unlink() for p in ('scripts/env_probe.py', 'scripts/sim_probe.py')]; print('ok')"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "stop env serve",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-c",
+				"import subprocess; subprocess.run(['powershell','-NoProfile','-Command','Get-NetTCPConnection -LocalPort 5020,8088 -ErrorAction SilentlyContinue | Select-Object -ExpandProperty OwningProcess -Unique | ForEach-Object { Stop-Process -Id $_ -Force }'], capture_output=True); print('stopped')"
+			],
+			"isBackground": false,
+			"problemMatcher": []
+		},
+		{
+			"label": "wago-sim: pytest final env",
+			"type": "shell",
+			"command": ".venv\\Scripts\\python.exe",
+			"args": [
+				"-m",
+				"pytest",
+				"tests",
+				"-q",
+				"--tb=short"
+			],
+			"isBackground": false,
+			"problemMatcher": []
 		}
 	]
 }

README.md

@@ -185,6 +185,56 @@ Configuration options: `--no-web`, `--web-port <n>` or env
 benches (all mutating endpoints answer 403). Plain HTTP — keep the
 port inside the bench network.
 
+## Plant environments (continuous simulation)
+
+A **plant environment** simulates the physical world around the
+distributed I/O: per deterministic tick it reads what the PLC
+commanded (DO/AO), runs your plant physics, and writes what the field
+now shows (DI/AI) — the sensors the PLC reads back. Environments run
+**until manually stopped** (Ctrl+C / `docker stop`), optionally ending
+sooner via an `is_done()` criterion.
+
+Attach one to a live Modbus server (wall-clock-paced 10 ms ticks):
+
+```bash
+wago-sim serve --node config/pump_node.yaml --port 5020 \
+               --env envs.pump_station_env:PumpStationEnv
+```
+
+or run headless (unpaced, for plant-model development):
+
+```bash
+wago-sim sim --node config/pump_node.yaml \
+             --env envs.pump_station_env:PumpStationEnv
+```
+
+**Authoring one** is a small subclass — override `model()`:
+
+```python
+# myenvs/tank.py
+from wago_sim.scenario.plantenv import PlantEnv
+
+class TankEnv(PlantEnv):
+    def model(self, state, dt_s):
+        pump_on = bool(state.by_code("DO0.0"))   # PLC command
+        level   = float(state.by_code("AI0.0"))  # current field
+        return {"AI0.0": level + (-5.0 if pump_on else 1.0) * dt_s}
+
+Env = TankEnv        # `--env myenvs.tank` picks this up
+```
+
+API surface (spec FR-8.2.14–17): `observe()` returns every
+DI/DO/AI/AO channel (I/O codes, names, values) plus aliases and
+degradation parameters; `apply()` writes DI/AI (the field);
+`force()` writes DO/AO master-side (headless runs, fault setup);
+`is_done()` optional end; `reset()` deterministic replay. Reference
+forms: `pkg.mod:ClassName`, `pkg.mod:factory`, or `pkg.mod` with
+`Env`/`make_env`. The shipped example is
+`envs/pump_station_env.py` (tank + pump physics with a low-level
+switch). Scenario-continuous mode (§8 YAML: datasets, degradation,
+invariants) remains available for scripted soak runs on the same
+engine ticks.
+
 ## Documentation
 
 - `docs/spec/software-spec.md` — the software specification
@@ -404,17 +454,18 @@ reporting, snapshots, and the gym-style `PlantEnv`. 30 new tests
 Datasets are **optional**: signal logic runs in parallel with CSV
 streaming, or entirely without it.
 
-- [x] Plant environment `scenario/plantenv.py` (FR-8.2.14–16, requested
+- [x] Plant environment `scenario/plantenv.py` (FR-8.2.14–17, reworked
-      2026-10-04): gym-style `PlantEnv` — deterministic engine-owned
+      2026-10-04 to environment-first): standalone `PlantEnv` —
-      tick clock (never wall-clock); per tick:
+      `observe()` returns ALL modules (DI/DO/AI/AO channels, per-type
-      `observe()` returns the WAGO input image in engineering units
+      codes, names, parameters); `apply()` writes the FIELD side
-      (channel values, names, per-type I/O codes, ranges, sim-time);
+      (DI/AI) by default; `force()` writes DO/AO master-side (mirrors/
-      the decision source decides; `apply()` writes the decided
+      statistics/watchdog stay consistent); `model()`/`is_done()`
-      outputs through the datastore's internal paths (mirrors/
+      override hooks (subclass authoring, FR-8.2.17); `reset()`
-      statistics/watchdog stay consistent). Existing `plugins:`
+      deterministic. Runners: `serve --env ...` wall-clock-paced
-      become thin adapters; external (non-Python) decision sources
+      alongside Modbus, runs until stopped; `sim --env ...` headless
-      step the env over a socket; one decision source per output
+      unpaced until is_done()/Ctrl+C. Example: `envs/pump_station_env.py`.
-      alias per tick (ownership conflicts are errors).
+      Verified live: PLC coil write → env physics → level registers
+      drain/refill over Modbus.
 - [x] `scenario/datasets.py` — streaming CSV playback for designated
       inputs (OPTIONAL driver): header-driven time column,
       `interpolation: linear|hold`, `repeat` with seamless seams,

docs/spec/software-spec.md

@@ -580,44 +580,49 @@ unchanged whether or not CSV streaming is used.
   top-level `until:` bounding the run. All §8.3–8.4 machinery (degradation,
   invariants, rate limits, snapshots) applies identically, whether or not
   any CSV streaming is configured.
-- FR-8.2.14: **Plant environment abstraction** (requested 2026-10-04): the
+- FR-8.2.14: **Plant environment abstraction** (requested 2026-10-04,
-  continuous engine exposes a gymnasium-style **environment object** —
+  reworked same day to environment-first): `PlantEnv` is the **primary
-  `PlantEnv` — as the single programmatic interface to the simulated plant.
+  interface** for continuous plant simulation, standalone-first — it
-  Per deterministic tick (the engine's fixed tick clock, FR-4.3.9 — never
+  requires only a virtual node + datastore (no scenario), with the §8.3/8.4
-  wall-clock), the cycle is:
+  machinery (degradation, invariants) as optional attachments. Per
-  **`state = env.observe()`** → the caller (plant logic, an external
+  deterministic tick (the engine or the runner advances the tick clock,
-  agent, or the scenario) decides → **`env.apply(actions)`** writes the
+  never wall-clock arithmetic): **`state = env.observe()`** returns the
-  decided output values → the engine advances exactly one tick
+  state of ALL modules — DI/DO/AI/AO channels, engineering units, per-type
-  (edges, datasets, degradation, invariants — FR-8.2.12 order) → repeat.
+  I/O codes, configured names, plus active aliases and parameters — →
-  `observe()` returns the state of the **generic inputs** — the WAGO
+  `env.model(state, dt)` (the caller-supplied plant physics) → **`apply()`
-  input process image in engineering units (all DI/DO/AI/AO channel
+  writes the new field state**. The environment owns no policy; plant
-  values, per-type module/channel addressing, configurable channel
+  models are supplied by subclassing (FR-8.2.17).
-  names, plus the active aliases) — and `apply(actions)` maps decided
+- FR-8.2.15: **Action semantics**: `apply()` primarily writes the **input
-  values onto the **output process image** through the datastore's
+  side (DI/AI)** — the simulated field: what the sensors show, driven by
-  internal Modbus write paths, so mirrors, statistics, and watchdog
+  plant physics reacting to the PLC's output commands. Writing outputs
-  behavior stay consistent (FR-9.2.6 semantics). The environment owns
+  (DO/AO) is allowed as explicit **forcing** via a `force=True` keyword or
-  no policy: the plant model (physics, closed-loop control) is supplied
+  `env.force(...)` — for headless runs without a PLC and for bench
-  by the caller, exactly like `gym.Env` vs. its agents.
+  fault setup; forced outputs go through the datastore's internal write
-- FR-8.2.15: **Environment integration**: `PlantEnv` is the interface the
+  paths so mirrors/statistics/watchdog stay consistent (FR-9.2.6).
-  per-tick plugins (FR-4.3.8) are migrated onto: a plugin becomes
+  Exactly one decision source may own an alias per tick (ownership
-  `env.observe()` → decide → `env.apply()` per tick, and a scenario may
+  conflicts are validation errors, extending FR-8.2.2).
-  alternatively register an external `step(env)` agent — including a
-  non-Python one (e.g., a trained RL policy) communicating over a
-  socket/pipe — without touching the engine loop. The existing
-  `plugins:` mechanism keeps working unchanged (thin adapter); the
-  engine always owns the tick clock, and exactly one decision source
-  may drive an output alias per tick (ownership conflicts are validation
-  errors, extending FR-8.2.2).
 - FR-8.2.16: **Environment API surface** (concrete):
-  `env.observe() -> PlantState` (channel values, names, I/O codes,
+  `env.observe() -> PlantState` (all channels, aliases, parameters,
-  ranges, sim-time, wear/parameters); `env.apply(actions) -> None`
+  sim-time); `env.apply(values, force=False)` (keys: I/O code or alias
-  (actions keyed by I/O code or alias name; unknown keys are errors);
+  name; DI/AI default, DO/AO require force; unknown keys are errors);
-  `env.reset(config) -> PlantState` (fresh episode, deterministic
+  `env.model(state, dt_s)` — override point for plant physics, returning
-  seed); `env.tick_ms`, `env.sim_time_ms` (read-only clock); and the
+  new field values; `env.is_done(state) -> bool` — optional end
-  full §8.3/§8.4 machinery (parameters, degradation, invariants,
+  criterion, default False (runs until manually stopped);
-  snapshots) reachable as `env.parameters`, `env.invariants`. The same
+  `env.reset()` — deterministic fresh episode; `env.tick_ms` /
-  object is served over the scenario-run HTTP API in server mode (§9),
+  `env.sim_time_ms` (read-only clock); `env.parameters` /
-  so external agents can run against a live PLC-connected bench.
+  `env.invariants` attachments.
+- FR-8.2.17: **Environment authoring & runners**: environments are easy
+  to create and call — a subclass of `PlantEnv` overriding `model()`
+  (and optionally `is_done()`) in any importable module
+  (`--env myenvs.pump_station.PumpStationEnv`), or a module-level
+  factory function. Two runners: (a) `wago-sim serve --env ...` ticks
+  the environment on a **wall-clock-paced** periodic task (10 ms tick
+  alongside the Modbus server, real-time behavior for a live PLC) and
+  runs **until manually stopped** (SIGINT/SIGTERM — the existing
+  graceful-shutdown path); (b) `wago-sim sim --node ... --env ...`
+  runs headless unpaced (as fast as determinism allows) until
+  `is_done()` or SIGINT. The scenario-continuous mode (FR-8.2.1-13)
+  remains a scripted layer on top of the same engine ticks.
 
 ### 8.3 Requirements — degradation models (plant logic over time)
 

envs/__init__.py

@@ -0,0 +1,8 @@
+"""User-authored plant environments (spec FR-8.2.17).
+
+Each module here is loadable by reference:
+
+    wago-sim serve --node ... --env envs.pump_station_env:PumpStationEnv
+    wago-sim sim     --node ... --env envs.pump_station_env:make_env
+    wago-sim serve --node ... --env envs.pump_station_env   # picks Env
+"""

envs/pump_station_env.py

@@ -0,0 +1,74 @@
+"""Example pump-station plant environment (spec FR-8.2.17).
+
+Demonstrates environment authoring: subclass PlantEnv, override
+``model()`` — read what the PLC commanded (DO/AO), compute what the
+field now shows (DI/AI). Run it live against a Modbus master:
+
+    wago-sim serve --node config/pump_node.yaml --port 5020 \
+                   --env envs.pump_station_env:PumpStationEnv
+
+or headless for model development:
+
+    wago-sim sim --node config/pump_node.yaml \
+                 --env envs.pump_station_env:PumpStationEnv
+
+Physics (deliberately simple, for demonstration): one tank.
+- inflow is constant (or a small sinusoidal demand profile)
+- the pump drains at `flow_max * capacity * speed` while DO0.0 is ON
+- the level integrates the balance; AI0.0 reports it as 0.1 mm/count
+- a low-level switch DI0.1 opens when the tank drains below 500 mm
+- optional is_done: stop after a sim hour (default: run forever)
+"""
+
+from __future__ import annotations
+
+from wago_sim.scenario.plantenv import PlantEnv, PlantState
+
+TANK_AREA = 2.0          # m^2
+FLOW_MAX = 30.0          # m^3/h the pump moves at full speed
+INFLOW = 12.0            # m^3/h constant inflow
+LEVEL_MAX = 9000.0       # mm — tank brim
+SWITCH_AT = 500.0        # mm — low-level switch threshold
+
+
+class PumpStationEnv(PlantEnv):
+    """Closed-loop tank + pump physics over the pump_node layout."""
+
+    def __init__(self, *args, stop_after_ms: float | None = None, **kw):
+        super().__init__(*args, **kw)
+        self._level_mm = 3000.0          # initial tank level
+        self.stop_after_ms = stop_after_ms
+
+    def model(self, state: PlantState, dt_s: float) -> dict[str, object]:
+        dt_h = dt_s / 3600.0
+        # what the PLC currently commands
+        pump_on = bool(state.by_code("DO0.0"))
+        # optional speed setpoint from AO0.0 (counts -> 0..1 fraction)
+        speed_raw = float(state.by_code("AO0.0") or 0)
+        speed = min(1.0, max(0.0, speed_raw / 4095.0))
+        # physics: water balance with pump wear (parameters attachable)
+        capacity = state.parameters.get("pump_capacity", 1.0)
+        outflow = FLOW_MAX * capacity * speed if pump_on else 0.0
+        d_level_mm = (INFLOW - outflow) / TANK_AREA * 1000.0 * dt_h
+        self._level_mm = min(LEVEL_MAX, max(0.0, self._level_mm + d_level_mm))
+        # the field the PLC reads back
+        level_counts = int(self._level_mm / 0.1)      # AI0.0: 0.1 mm/count
+        low_switch = self._level_mm < SWITCH_AT       # DI0.1
+        return {
+            "AI0.0": level_counts,
+            "DI0.1": low_switch,
+        }
+
+    def is_done(self, state: PlantState) -> bool:
+        if self.stop_after_ms is None:
+            return False                 # default: run until stopped
+        return state.sim_time_ms >= self.stop_after_ms
+
+
+def make_env(node=None, engine=None) -> PumpStationEnv:
+    """Factory form (FR-8.2.17): envs.pump_station_env:make_env."""
+    return PumpStationEnv(node=node, engine=engine)
+
+
+# bare-module form: `--env envs.pump_station_env` picks this up
+Env = PumpStationEnv

src/wago_sim/cli.py

@@ -196,6 +196,12 @@ async def _serve(args: argparse.Namespace) -> int:
     app = build_app(modules=modules, host=args.host, port=args.port, web=web_cfg,
                     node_config_path=args.node, channel_names=channel_names,
                     channel_ranges=channel_ranges)
+    # optional plant environment (FR-8.2.17): ticked wall-clock-paced
+    # alongside the Modbus server, runs until manually stopped
+    env = None
+    if getattr(args, "env", None):
+        from .scenario.plantenv import load_env
+        env = load_env(args.env, node=app.node)
     await app.server.start()
     web_server = None
     if web_cfg is not None:
@@ -204,6 +210,27 @@ async def _serve(args: argparse.Namespace) -> int:
               f":{web_cfg.port} (user: {web_cfg.user})")
     print(f"Simulating WAGO 750-362 with {len(modules)} modules "
           f"on {args.host}:{args.port} (Ctrl+C to stop)")
+    env_task = None
+    if env is not None:
+        async def run_env():
+            """Paced ticker (FR-8.2.17): wall-clock-aligned ticks for a
+            live PLC; runs until cancelled by the shutdown path."""
+            import time as _time
+            period = env.TICK_MS / 1000.0
+            next_at = _time.monotonic()
+            while True:
+                env.tick()
+                if env.is_done(env.observe()):
+                    log.info("environment is_done() satisfied; env stopped")
+                    return
+                next_at += period
+                delay = next_at - _time.monotonic()
+                if delay > 0:
+                    await asyncio.sleep(delay)
+                else:
+                    next_at = _time.monotonic()   # catch up, no drift burst
+        env_task = asyncio.create_task(run_env())
+        print(f"Plant environment: {args.env} (10 ms paced tick, Ctrl+C to stop)")
 
     # Graceful shutdown on SIGINT/SIGTERM (docker stop, systemctl stop).
     # asyncio.run installs a SIGINT handler; SIGTERM arrives as a raw
@@ -220,6 +247,12 @@ async def _serve(args: argparse.Namespace) -> int:
     except asyncio.CancelledError:
         pass  # SIGINT via asyncio.run's handler
     finally:
+        if env_task is not None:
+            env_task.cancel()
+            try:
+                await env_task
+            except asyncio.CancelledError:
+                pass
         if web_server is not None:
             task = getattr(web_server, "_serve_task", None)
             web_server.should_exit = True
@@ -232,6 +265,34 @@ async def _serve(args: argparse.Namespace) -> int:
     return 0
 
 
+def _sim(args: argparse.Namespace) -> int:
+    """Headless environment run (FR-8.2.17): unpaced, until is_done()
+    or Ctrl+C — for plant-model development without a PLC."""
+    node_config = _load_yaml_file(args.node, "node config")
+    modules = load_node_modules(node_config)
+    channel_names = load_channel_names(node_config)
+    channel_ranges = load_channel_ranges(node_config)
+    app = build_app(modules=modules, channel_names=channel_names,
+                    channel_ranges=channel_ranges)
+    from .scenario.plantenv import load_env
+    env = load_env(args.env, node=app.node)
+    print(f"Running {args.env} headless (Ctrl+C to stop; is_done() to end)")
+    ticks = 0
+    try:
+        while True:
+            state = env.tick()
+            ticks += 1
+            if env.is_done(state):
+                print(f"is_done() at tick {ticks} (sim {state.sim_time_ms:.0f} ms)")
+                break
+            if ticks % 1000 == 0 and args.verbose:
+                print(f"tick {ticks}: {state.sim_time_ms:.0f} ms")
+    except KeyboardInterrupt:
+        print(f"\nstopped manually at tick {ticks} "
+              f"(sim {env.sim_time_ms:.0f} ms)")
+    return 0
+
+
 def _run_scenario(args: argparse.Namespace) -> int:
     node_config = _load_yaml_file(args.node, "node config")
     modules = load_node_modules(node_config)
@@ -332,6 +393,12 @@ def main(argv: list[str] | None = None) -> int:
         "--web-port", type=_port_type, default=_env_port("WAGO_SIM_WEB_PORT", 8080),
         help="web UI port (or env: WAGO_SIM_WEB_PORT; default 8080)",
     )
+    serve.add_argument(
+        "--env", default=None,
+        help=("plant environment module (FR-8.2.17): e.g. "
+              "envs.pump_station_env:PumpStationEnv — ticked wall-clock-"
+              "paced alongside Modbus; runs until stopped"),
+    )
     serve.set_defaults(func=_serve)
 
     runp = sub.add_parser("run", help="run a scenario headless (no server)")
@@ -345,6 +412,13 @@ def main(argv: list[str] | None = None) -> int:
     val.add_argument("--scenario", required=True)
     val.set_defaults(func=_validate)
 
+    sim = sub.add_parser(
+        "sim", help="run a plant environment headless (FR-8.2.17)")
+    sim.add_argument("--node", required=True, help="node config YAML")
+    sim.add_argument("--env", required=True,
+                     help="environment module, e.g. envs.pump_station_env:PumpStationEnv")
+    sim.set_defaults(func=_sim)
+
     args = parser.parse_args(argv)
     _build_logger(args.verbose)
     # `--node` may come from the environment when omitted (service mode);

src/wago_sim/scenario/engine.py

@@ -204,7 +204,7 @@ class ScenarioEngine:
         for model in self.degradation.models.values():
             self._claim_alias(model.observes, f"degradation:{model.name}")
         self.invariants = InvariantMonitor(self, self.scenario)
-        self.env = PlantEnv(self)
+        self.env = PlantEnv(engine=self)
 
         # FR-8.2.1: a continuous run needs at least one run-length driver
         if (not self.datasets and not self._scheduled
@@ -519,10 +519,10 @@ class ScenarioEngine:
                 return self._result
         # (2) edges
         self._detect_edges()
-        # (3) plant logic: plugins first, then queued env actions flush
+        # (3) plant logic: plugins first, then queued env decisions land
+        #     (the scenario layer drives env via observe/apply; queued
+        #     values are already applied by the decision source)
         self._run_tick_plugins()
-        if self.env is not None:
-            self.env.flush()
         self._advance_ramps(dt)
         # (4) degradation
         if self.degradation is not None:

src/wago_sim/scenario/plantenv.py

@@ -1,24 +1,32 @@
-"""PlantEnv: gymnasium-style environment over the simulated plant (spec §8, FR-8.2.14-16).
+"""PlantEnv: gymnasium-style plant environment (spec §8, FR-8.2.14-17).
 
-The environment is the decision-source interface to the distributed
+Standalone-first: requires only a virtual node + datastore — no
-I/O: per deterministic engine tick —
+scenario. The environment is the interface for continuous plant
+simulation; per deterministic tick (runner-owned clock, never
+wall-clock arithmetic):
 
-    state = env.observe()      # WAGO input image in engineering units
+    state = env.observe()      # ALL modules: DI/DO/AI/AO channels,
-    decision = policy(state)   # caller's plant model / agent (no policy here)
+                               #   engineering units, names, parameters
-    env.apply(actions)        # decided outputs -> output image via
+    values = env.model(state, dt)   # plant physics (subclass hook)
-                              #   the datastore's INTERNAL write paths
+    env.apply(values)          # new FIELD state -> DI/AI inputs;
-    engine.pump()             # engine advances exactly one tick
+                               #   DO/AO via env.force() (master-side)
 
-The engine always owns the tick clock (FR-4.3.9 deterministic sim
+`model()` is the override point for plant physics: read what the PLC
-time); the env never reads wall-clock. `apply()` writes through the
+commanded (DO/AO), compute what the field now shows (DI/AI), return
-datastore's Modbus-equivalent dispatch so mirrors, statistics, and
+it. `is_done(state)` gives an optional end criterion; the default is
-watchdog behavior stay consistent (FR-9.2.6 semantics).
+False — environments run until manually stopped (SIGINT/SIGTERM).
+
+Attachments: degradation (§8.3), invariants (§8.4) and aliases
+(scenario-continuous mode) hook onto the same tick when the env is
+constructed on an engine.
 """
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from dataclasses import dataclass, field
 
+from ..modules import VirtualNode
 from .engine import ScenarioError
 
 
@@ -31,43 +39,60 @@ class ChannelState:
     channel: int         # 0-based channel on the module
     io_code: str        # e.g. "DI0.0"
     name: str | None     # configured application name, if any
-    value: object        # bool for di/do, float for ai/ao
+    value: object        # bool for di/do, int/float for ai/ao
 
 
 @dataclass
 class PlantState:
-    """Full observation: every channel + aliases + parameters + clock."""
+    """Full observation: every channel of every module + attachments."""
 
     sim_time_ms: float
     channels: list[ChannelState] = field(default_factory=list)
     aliases: dict[str, object] = field(default_factory=dict)
     parameters: dict[str, float] = field(default_factory=dict)
 
+    def by_code(self, io_code: str) -> object:
+        """Value of one channel by I/O code, or KeyError."""
+        for ch in self.channels:
+            if ch.io_code == io_code:
+                return ch.value
+        raise KeyError(io_code)
+
+    def by_kind(self, kind: str) -> dict[str, object]:
+        """{io_code: value} for all channels of one kind."""
+        return {c.io_code: c.value for c in self.channels if c.kind == kind}
+
 
 class PlantEnv:
-    """gym-style observe/apply environment (FR-8.2.14-16).
+    """gym-style plant environment (FR-8.2.14-17).
 
-    Constructed by the continuous engine; the engine stays the clock
+    Constructed on a VirtualNode (standalone) or on an engine (the
-    owner — the env's `step()` convenience drives `engine.pump()` for
+    scenario-continuous layer). The runner owns the tick clock; `tick()`
-    exactly one tick (server mode may tick the engine externally and
+    advances sim-time deterministically by TICK_MS per call so both
-    only use observe/apply).
+    paced (serve) and unpaced (sim) runners behave identically.
     """
 
-    def __init__(self, engine) -> None:
+    TICK_MS = 10.0
+
+    def __init__(self, node: VirtualNode | None = None, *,
+                 engine=None) -> None:
         self.engine = engine
-        self._actions: dict[str, object] = {}
+        self.node: VirtualNode = node if node is not None else engine.node
+        self.datastore = self.node.datastore
+        self.sim_time_ms = 0.0
+        # attachments (optional; the scenario layer wires these)
+        self.degradation = getattr(engine, "degradation", None)
+        self.invariants = getattr(engine, "invariants", None)
+        self._aliases = getattr(engine, "_aliases", None) if engine else None
+        self._alias_value: Callable[[str], object] | None = (
+            engine.alias_value if engine is not None else None)
 
     # -- observation (FR-8.2.14) -----------------------------------------------
 
     def observe(self) -> PlantState:
-        """Snapshot of the generic inputs (WAGO input image + aliases).
+        """Snapshot ALL modules (DI/DO/AI/AO) in engineering units."""
-
+        node = self.node
-        DI/AI channels hold the simulated field state; DO/AO hold the
+        ds = self.datastore
-        PLC-written outputs as currently in the images — all values in
-        engineering units (alias scale/offset applied where defined).
-        """
-        node = self.engine.node
-        ds = self.engine.datastore
         channels: list[ChannelState] = []
         for kind, pool in (("di", node.di), ("do", node.do),
                            ("ai", node.ai), ("ao", node.ao)):
@@ -80,7 +105,8 @@ class PlantEnv:
                     value = ds.get_input_word(loc.word)
                 else:
                     value = ds.get_output_word(loc.word)
-                code = node.io_code_at_slot(kind, loc.module_index, loc.channel - 1)
+                code = node.io_code_at_slot(
+                    kind, loc.module_index, loc.channel - 1)
                 channels.append(ChannelState(
                     kind=kind,
                     module=node.kind_index(loc.module_index)[kind],
@@ -89,128 +115,226 @@ class PlantEnv:
                     name=node.channel_names.get(code),
                     value=value,
                 ))
-        aliases = {name: self.engine.alias_value(name)
+        aliases = ({name: self._alias_value(name)
-                   for name in self.engine._aliases}
+                    for name in (self._aliases or {})}
+                   if self._alias_value else {})
         parameters: dict[str, float] = {}
-        degradation = getattr(self.engine, "degradation", None)
+        if self.degradation is not None:
-        if degradation is not None:
+            parameters = {n: m.value
-            parameters = {n: m.value for n, m in degradation.models.items()}
+                          for n, m in self.degradation.models.items()}
         return PlantState(
-            sim_time_ms=self.engine.sim_time_ms,
+            sim_time_ms=self.sim_time_ms,
             channels=channels,
             aliases=aliases,
             parameters=parameters,
         )
 
-    # -- actions (FR-8.2.14) -----------------------------------------------------
+    # -- plant model hook (FR-8.2.16) -------------------------------------------
 
-    def apply(self, actions: dict[str, object]) -> None:
+    def model(self, state: PlantState, dt_s: float) -> dict[str, object]:
-        """Queue decided outputs; they land when the engine ticks.
+        """Plant physics: react to PLC commands, compute the new field.
 
-        Keys: I/O code ("DO0.0") or alias name ("pump_relay"). Writing
+        Override in a subclass (FR-8.2.17). Receives the full observed
-        a DI/AI through the env is an error — inputs are the field side
+        state and the tick delta in seconds; returns {I/O code or alias:
-        (datasets/scenario steps simulate them). Keys are validated
+        new value} for the channels it drives — DI/AI normally (the
-        EAGERLY so a bad decision surfaces immediately, not at flush.
+        simulated sensors); DO/AO entries are treated as forcing and go
+        through the master-side write paths. Returning {} (the default)
+        makes this a pure observation environment.
         """
-        if not isinstance(actions, dict):
+        return {}
-            raise ScenarioError("apply() actions must be a mapping")
-        validated: dict[str, object] = {}
-        for key, value in actions.items():
-            alias = self.engine._aliases.get(key) or self._alias_by_code(key)
-            if alias is None:
-                raise ScenarioError(f"apply(): unknown channel {key!r}")
-            if alias.kind not in ("do", "ao"):
-                raise ScenarioError(
-                    f"apply(): {key!r} is an input alias ({alias.kind}); "
-                    "only outputs (do/ao) are decided")
-            validated[str(key)] = value
-        self._actions.update(validated)
 
-    def flush(self) -> None:
+    def is_done(self, state: PlantState) -> bool:
-        """Write queued actions into the output image (engine-tick step).
+        """Optional end criterion; default runs until manually stopped."""
+        return False
 
-        Called by the engine after the observe/decide window and before
+    # -- actions (FR-8.2.15) ------------------------------------------------------
-        the pipeline advances — so the next observe() sees them
+
-        (FR-8.2.12 atomic-env-tick paragraph).
+    def apply(self, values: dict[str, object], *, force: bool = False) -> None:
+        """Write new channel values into the process images.
+
+        DI/AI keys (the field) write directly — this is the primary
+        direction. DO/AO keys (PLC outputs) require force=True and go
+        through the datastore's internal Modbus dispatch so mirrors,
+        statistics, and watchdog stay consistent (FR-9.2.6) — for
+        headless runs and fault setup only.
+        Keys: I/O code ("AI0.0") or alias name. Unknown keys raise
+        immediately (FR-8.2.16).
         """
-        for key, value in self._actions.items():
+        if not isinstance(values, dict):
-            self._write_one(key, value)
+            raise ScenarioError("apply() values must be a mapping")
-        self._actions = {}
+        for key, value in values.items():
+            self._write_one(key, value, force=force)
 
-    def _write_one(self, key: str, value: object) -> None:
+    def force(self, values: dict[str, object]) -> None:
-        # resolve by alias name first, then by I/O code
+        """Master-side forcing of DO/AO (FR-8.2.15 explicit form)."""
-        alias = self.engine._aliases.get(key)
+        self.apply(values, force=True)
-        if alias is None:
+
-            alias = self._alias_by_code(key)
+    def _resolve(self, key: str):
-        if alias is None:
+        """-> (kind, ChannelLocation, alias_or_None) for a code or alias."""
+        # alias name first (scenario layer)
+        if self._aliases and key in self._aliases and self.engine is not None:
+            alias = self._aliases[key]
+            return alias.kind, self.engine._alias_location(alias), alias
+        # any channel by I/O code (per-type codes, FR-4.3.11 v1.5)
+        code_up = str(key).strip().upper()
+        node = self.node
+        for kind, pool in (("di", node.di), ("do", node.do),
+                           ("ai", node.ai), ("ao", node.ao)):
+            for loc in pool:
+                if node.io_code_at_slot(
+                        kind, loc.module_index, loc.channel - 1) == code_up:
+                    return kind, loc, None
         raise ScenarioError(f"apply(): unknown channel {key!r}")
-        if alias.kind not in ("do", "ao"):
+
+    def _write_one(self, key: str, value: object, *, force: bool) -> None:
+        kind, loc, alias = self._resolve(key)
+        ds = self.datastore
+        if kind in ("do", "ao") and not force:
             raise ScenarioError(
-                f"apply(): {key!r} is an input alias ({alias.kind}); "
+                f"apply(): {key!r} is an output ({kind.upper()}); use "
-                "only outputs (do/ao) are decided")
+                "force=True / env.force() for master-side forcing")
-        loc = self.engine._alias_location(alias)
+        scale = getattr(alias, "scale", 1.0) or 1.0
-        ds = self.engine.datastore
+        offset = getattr(alias, "offset", 0.0) or 0.0
-        if alias.kind == "do":
+        if kind == "di":
+            ds.set_digital_input_channel(loc.bit + 1, bool(value))
+        elif kind == "ai":
+            raw = int(float(value) * scale + offset) & 0xFFFF
+            ds.set_input_word(loc.word, raw)
+        elif kind == "do":
             result = ds.write_bits(loc.bit, [bool(value)])
-        else:
+            if isinstance(result, Exception):
-            numeric = float(value)  # type: ignore[arg-type]
+                raise ScenarioError(f"force() write to {key!r}: {result}")
-            raw = int(numeric * alias.scale + alias.offset) & 0xFFFF
+        else:  # ao
+            raw = int(float(value) * scale + offset) & 0xFFFF
             result = ds.write_words(loc.word, [raw])
             if isinstance(result, Exception):
-            raise ScenarioError(f"apply() write to {key!r} rejected: {result}")
+                raise ScenarioError(f"force() write to {key!r}: {result}")
 
-    def _alias_by_code(self, code: str):
+    # -- one full tick (FR-8.2.14 cycle) ---------------------------------------------
-        """Find an alias whose I/O code matches (per-type codes)."""
+
-        code_up = str(code).strip().upper()
+    def tick(self, dt_ms: float | None = None) -> PlantState:
-        node = self.engine.node
+        """One environment tick: model() -> apply(field) -> advance.
-        for name, alias in self.engine._aliases.items():
+
-            if alias.kind not in ("do", "ao"):
+        Deterministic: sim-time advances by TICK_MS (or dt_ms) per call
-                continue
+        — never from wall-clock. Attachments advance with the tick:
-            try:
+        degradation (§8.3) then invariants (§8.4), FR-8.2.12 order.
-                loc = self.engine._alias_location(alias)
+        """
-            except Exception:
+        dt = dt_ms if dt_ms is not None else self.TICK_MS
-                continue
+        state = self.observe()
-            if node.io_code_at_slot(alias.kind, loc.module_index,
+        if self.is_done(state):
-                                    loc.channel - 1) == code_up:
+            return state
-                return alias
+        values = self.model(state, dt / 1000.0)
-        return None
+        if values:
+            self.apply(values)
+        self.sim_time_ms += dt
+        if self.degradation is not None:
+            self.degradation.tick(dt)
+        if self.invariants is not None:
+            self.invariants.tick()
+        return self.observe()
+
+    # -- episode control (FR-8.2.16) ------------------------------------------------
+
+    def reset(self) -> PlantState:
+        """Fresh episode: zero images, reset wear + parameters + clock
+        (deterministic — identical runs replay identically, NFR-8.7.3)."""
+        self.datastore.reset_images()
+        self.sim_time_ms = 0.0
+        if self.degradation is not None:
+            for model in self.degradation.models.values():
+                model.reset()
+                model.diagnosed = False
+            self.degradation._last_truth = {}
+        if self.invariants is not None:
+            self.invariants.episodes = []
+            for inv in self.invariants.invariants:
+                inv.active = None
+            for rl in self.invariants.rate_limits:
+                rl.samples = []
+                rl.breached = False
+                rl.peak_rate = 0.0
+            self.invariants.snapshots = []
+            self.invariants._last_snapshot_ms = 0.0
+        return self.observe()
 
     # -- clock (FR-8.2.16) ---------------------------------------------------------
 
     @property
     def tick_ms(self) -> float:
-        return float(self.engine.TICK_MS)
+        return self.TICK_MS
-
-    @property
-    def sim_time_ms(self) -> float:
-        return self.engine.sim_time_ms
-
-    # -- episode control (FR-8.2.16) ------------------------------------------------
-
-    def reset(self, config: dict | None = None) -> PlantState:
-        """Fresh episode: zero images, reset wear + parameters + clock
-        (deterministic — identical runs replay identically, NFR-8.7.3)."""
-        self.engine.reset_episode()
-        return self.observe()
-
-    def step(self, actions: dict[str, object] | None = None):
-        """One full env tick: apply -> engine pump -> observe (FR-8.2.14).
-
-        Convenience for headless agents; server mode ticks the engine
-        on its own scheduler and uses observe/apply separately.
-        """
-        if actions:
-            self.apply(actions)
-        self.engine.pump()
-        return self.observe()
-
-    # -- §8.3/§8.4 machinery (FR-8.2.16) ----------------------------------------------
 
     @property
     def parameters(self) -> dict[str, float]:
-        deg = getattr(self.engine, "degradation", None)
+        if self.degradation is None:
-        return {n: m.value for n, m in deg.models.items()} if deg else {}
+            return {}
+        return {n: m.value for n, m in self.degradation.models.items()}
 
-    @property
+
-    def invariants(self):
+def load_env(ref: str, node: VirtualNode | None = None, *,
-        return getattr(self.engine, "invariants", None)
+             engine=None) -> PlantEnv:
+    """Load an environment from a module reference (FR-8.2.17).
+
+    Accepts ``pkg.mod.ClassName``, ``pkg.mod.factory_fn`` (returns a
+    PlantEnv), or a bare ``pkg.mod`` whose module defines either a
+    class named ``Env`` or a function ``make_env(...)``.
+    """
+    import importlib
+
+    parts = ref.split(":")
+    attr_path = None
+    mod = None
+    leaf = None
+    if len(parts) == 2:
+        # explicit module:attr form — ref is the WHOLE module path
+        attr_path = parts[1] or None
+        mod_name = parts[0]
+    else:
+        # no colon: try the whole ref as a module first, then as
+        # pkg.mod containing an attr (e.g. ...:Class omitted)
+        try:
+            mod = importlib.import_module(ref)
+            mod_name = ref
+        except ImportError:
+            mod_name, _, leaf = ref.rpartition(".")
+    if not mod_name or ("." not in mod_name and attr_path is None
+                        and leaf is None):
+        raise ScenarioError(f"--env {ref!r}: reference must include a module path")
+
+    if mod is None:
+        try:
+            mod = importlib.import_module(mod_name)
+        except ImportError as exc:
+            raise ScenarioError(
+                f"--env {ref!r}: cannot import {mod_name!r}: {exc}")
+
+    obj = None
+    if attr_path is not None:
+        # explicit module:attr — must exist; never fall back silently
+        if not hasattr(mod, attr_path):
+            raise ScenarioError(
+                f"--env {ref!r}: no {attr_path!r} in {mod_name!r}")
+        obj = getattr(mod, attr_path)
+    elif leaf and hasattr(mod, leaf):
+        obj = getattr(mod, leaf)
+    elif hasattr(mod, "Env"):
+        obj = getattr(mod, "Env")
+    elif hasattr(mod, "make_env"):
+        obj = getattr(mod, "make_env")
+    else:
+        raise ScenarioError(
+            f"--env {ref!r}: no environment found in {mod_name!r} "
+            "(expected a class, Env, or make_env())")
+
+    env: PlantEnv
+    if isinstance(obj, type) and issubclass(obj, PlantEnv):
+        env = obj(node=node, engine=engine)
+    elif callable(obj):
+        made = obj(node=node, engine=engine) if node is not None \
+            else obj(engine=engine)
+        if not isinstance(made, PlantEnv):
+            raise ScenarioError(f"--env {ref!r}: did not return a PlantEnv")
+        env = made
+    else:
+        raise ScenarioError(
+            f"--env {ref!r}: not a PlantEnv subclass or factory")
+    return env

tests/test_continuous.py

@@ -9,6 +9,7 @@ import pytest
 
 from wago_sim.scenario.datasets import CsvDataset
 from wago_sim.scenario.engine import DataError, ScenarioEngine, ScenarioError, Verdict
+from wago_sim.scenario.plantenv import PlantEnv
 
 
 def _write_csv(path, text: str) -> str:
@@ -326,6 +327,8 @@ class TestInvariants:
 
 
 class TestPlantEnv:
+    """PlantEnv v2 (FR-8.2.14-17): standalone-first, apply -> DI/AI."""
+
     def _scenario(self):
         return {
             "mode": "continuous", "name": "x", "until": "500ms",
@@ -334,66 +337,130 @@ class TestPlantEnv:
         }
 
     def test_observe_returns_all_channels(self, node):
-        engine = _engine(node, self._scenario())
+        # standalone construction: no engine needed
-        state = engine.env.observe()
+        env = PlantEnv(node=node)
+        state = env.observe()
         kinds = {c.kind for c in state.channels}
         assert kinds == {"di", "do", "ai", "ao"}
         codes = [c.io_code for c in state.channels]
         assert "DI0.0" in codes and "DO0.0" in codes
-        assert state.aliases["start"] is False
         assert state.sim_time_ms == 0.0
 
-    def test_apply_writes_output_before_next_observe(self, node):
+    def test_apply_writes_input_side(self, node):
-        engine = _engine(node, self._scenario())
+        env = PlantEnv(node=node)
-        engine.env.apply({"pump": True})
+        env.apply({"DI0.0": True, "AI0.0": 1234})
-        state = engine.env.step()          # flush + pump + observe
+        state = env.observe()
-        pump_states = [c for c in state.channels if c.io_code == "DO0.0"]
+        assert state.by_code("DI0.0") is True
-        assert pump_states[0].value is True
+        assert state.by_code("AI0.0") == 1234
-        assert state.aliases["pump"] is True
 
-    def test_apply_by_io_code(self, node):
+    def test_apply_output_requires_force(self, node):
-        engine = _engine(node, self._scenario())
+        env = PlantEnv(node=node)
-        engine.env.apply({"DO0.0": True})
+        with pytest.raises(ScenarioError, match="force"):
-        state = engine.env.step()
+            env.apply({"DO0.0": True})
-        assert state.aliases["pump"] is True
+        # explicit forcing works and lands in the output image
-
+        env.force({"DO0.0": True, "AO0.0": 999})
-    def test_apply_input_rejected(self, node):
+        state = env.observe()
-        engine = _engine(node, self._scenario())
+        assert state.by_code("DO0.0") is True
-        with pytest.raises(ScenarioError, match="input alias"):
+        assert state.by_code("AO0.0") == 999
-            engine.env.apply({"start": True})
 
     def test_apply_unknown_rejected(self, node):
-        engine = _engine(node, self._scenario())
+        env = PlantEnv(node=node)
         with pytest.raises(ScenarioError, match="unknown channel"):
-            engine.env.apply({"nonexistent": True})
+            env.apply({"nonexistent": True})
+
+    def test_model_hook_drives_field(self, node):
+        """The intended loop: model() reads PLC outputs, returns field."""
+
+        class Tank(PlantEnv):
+            def model(self, state, dt_s):
+                pump_on = bool(state.by_code("DO0.0"))
+                level = float(state.by_code("AI0.0"))
+                return {"AI0.0": level + (5.0 if pump_on else 1.0)}
+
+        env = Tank(node=node)
+        env.force({"DO0.0": True})            # the "PLC" turns the pump on
+        state = env.tick()
+        assert state.by_code("AI0.0") == 5.0  # field advanced via model
+        state = env.tick()
+        assert state.by_code("AI0.0") == 10.0
+        assert state.sim_time_ms == 20.0
+
+    def test_is_done_stops_ticks(self, node):
+        class DoneAfterTwo(PlantEnv):
+            def __init__(self, *a, **kw):
+                super().__init__(*a, **kw)
+                self.n = 0
+
+            def is_done(self, state):
+                return self.n >= 2
+
+            def model(self, state, dt_s):
+                self.n += 1
+                return {}
+
+        env = DoneAfterTwo(node=node)
+        env.tick()
+        env.tick()
+        state = env.tick()                    # done: no further advance
+        assert env.n == 2
+        assert state.sim_time_ms == 20.0
 
     def test_reset_deterministic_replay(self, node):
-        engine = _engine(node, self._scenario())
+        env = PlantEnv(node=node)
-        engine.env.apply({"pump": True})
+        env.apply({"DI0.0": True, "AI0.0": 42})
-        for _ in range(10):
+        env.tick()
-            engine.pump()
+        assert env.sim_time_ms > 0
-        assert engine.sim_time_ms > 0
+        state = env.reset()
-        engine.env.reset()
+        assert env.sim_time_ms == 0.0
-        assert engine.sim_time_ms == 0.0
+        assert state.by_code("DI0.0") is False
-        state = engine.env.observe()
+        assert state.by_code("AI0.0") == 0
-        assert state.aliases["pump"] is False
 
-    def test_two_tick_cycle_deterministic(self, node):
+    def test_engine_attached_env_uses_aliases(self, node):
-        """Acceptance criterion 12: copy an input to an output; the
+        """Scenario layer: alias names + alias scale/offset still work."""
-        second observation contains it (FR-8.2.12 pipeline consistency)."""
         engine = _engine(node, self._scenario())
-        state1 = engine.env.observe()          # observe: DI0.0 still off
+        env = engine.env
-        di_then = next(c for c in state1.channels if c.io_code == "DI0.0")
+        state = env.observe()
-        assert di_then.value is False
+        assert state.aliases["start"] is False
-        # field-side change + decision AFTER observing, before the tick
+        # alias write to a do output requires force (PLC-side semantics)
-        engine.datastore.set_digital_input_channel(1, True)
+        with pytest.raises(ScenarioError, match="force"):
-        engine.env.apply({"pump": di_then.value})   # decide on observed state
+            env.apply({"pump": True})
-        state2 = engine.env.step()                  # flush + tick + observe
+        env.force({"pump": True})
-        di_now = next(c for c in state2.channels if c.io_code == "DI0.0")
+        assert env.observe().aliases["pump"] is True
-        assert di_now.value is True                  # visible next observe
+
-        # decided output applied from the OBSERVED state: pump still off
+    def test_pump_station_example_env(self, node):
-        do_now = next(c for c in state2.channels if c.io_code == "DO0.0")
+        """The shipped example env loads and runs closed-loop physics."""
-        assert do_now.value is False
+        from envs.pump_station_env import PumpStationEnv
+        env = PumpStationEnv(node=node)
+        # PLC turns the pump on at full speed
+        env.force({"DO0.0": True, "AO0.0": 4095})
+        env.tick()                                # physics applies the level
+        level0 = env.observe().by_code("AI0.0")
+        assert level0 > 0                         # tank starts at 3000 mm
+        # big dt (10 sim-minutes): the pump visibly drains the tank
+        env.tick(dt_ms=600_000)
+        level1 = env.observe().by_code("AI0.0")
+        assert level1 < level0                    # pump drains the tank
+        # pump OFF: inflow fills the tank instead
+        env.force({"DO0.0": False})
+        env.tick(dt_ms=600_000)
+        level2 = env.observe().by_code("AI0.0")
+        assert level2 > level1
+        # and it runs forever by default
+        assert env.is_done(env.observe()) is False
+
+    def test_load_env_class_and_factory(self, node):
+        from wago_sim.scenario.plantenv import load_env
+        env = load_env("envs.pump_station_env:PumpStationEnv", node=node)
+        assert isinstance(env, PlantEnv)
+        env = load_env("envs.pump_station_env:make_env", node=node)
+        assert isinstance(env, PlantEnv)
+        # bare module picks up Env
+        env = load_env("envs.pump_station_env", node=node)
+        assert isinstance(env, PlantEnv)
+        with pytest.raises(ScenarioError):
+            load_env("envs.pump_station_env:Nope", node=node)
+        with pytest.raises(ScenarioError):
+            load_env("envs.missing_module", node=node)
 
 
 class TestTickPipeline: