🚧 WIP

2026-10-08 06:30:54 +02:00 · 2026-10-07 17:57:30 +02:00 · 2026-10-07 17:13:51 +02:00
19 changed files with 2214 additions and 489 deletions
@@ -13,4 +13,7 @@ dist/

 # Local runtime artifacts (copy config.example.yaml -> config.yaml)
 config.yaml
-logs/
+logs/
+
+# Generated by scripts/build_exe.ps1 at build time
+scripts/.exe_version_info.txt
@@ -25,6 +25,21 @@
                "PYTHONUNBUFFERED": "1"
            }
        },
+        {
+            // One-shot read of every configured tag — the pre-deployment check.
+            "name": "opcda-diagnozer: probe",
+            "type": "debugpy",
+            "request": "launch",
+            "python": "${workspaceFolder}/.venv/Scripts/python.exe",
+            "module": "opcda_diagnozer",
+            "args": ["probe", "--config", "config.example.yaml"],
+            "cwd": "${workspaceFolder}",
+            "console": "integratedTerminal",
+            "justMyCode": false,
+            "env": {
+                "PYTHONUNBUFFERED": "1"
+            }
+        },
        {
            // Basic use: validate the example config without connecting.
            "name": "opcda-diagnozer: check-config",
@@ -90,14 +105,14 @@
            }
        },
        {
-            // List the devices (address-space branches) of the configured
-            // server — one level, with per-device tag counts.
-            "name": "opcda-diagnozer: list-devices",
+            // Full structure tree of the configured server: every branch
+            // AND every tag in place, whole subtree.
+            "name": "opcda-diagnozer: list-items (whole tree)",
            "type": "debugpy",
            "request": "launch",
            "python": "${workspaceFolder}/.venv/Scripts/python.exe",
            "module": "opcda_diagnozer",
-            "args": ["list-devices", "--config", "config.example.yaml"],
+            "args": ["list-items", "all", "--depth", "0", "--config", "config.example.yaml"],
            "cwd": "${workspaceFolder}",
            "console": "integratedTerminal",
            "justMyCode": false,
@@ -106,20 +121,42 @@
            }
        },
        {
-            // List the tags of one device (whole subtree) as JSON — the
-            // item_id values are what go into config `tags:`. Adjust the
-            // device path to the server your config points at.
-            "name": "opcda-diagnozer: list-tags (recursive, JSON)",
+            // Flat ItemID list of one subtree as JSON — the item_id values
+            // are what go into config `tags:`. Adjust the branch path to the
+            // server your config points at.
+            "name": "opcda-diagnozer: list-items tags (JSON)",
            "type": "debugpy",
            "request": "launch",
            "python": "${workspaceFolder}/.venv/Scripts/python.exe",
            "module": "opcda_diagnozer",
            "args": [
-                "list-tags",
-                "--config", "config.example.yaml",
-                "Simulation Items.Bucket Brigade",
-                "--recursive",
-                "--json"
+                "list-items",
+                "tags",
+                "--branch", "Simulation Items.Bucket Brigade",
+                "--json",
+                "--config", "config.example.yaml"
+            ],
+            "cwd": "${workspaceFolder}",
+            "console": "integratedTerminal",
+            "justMyCode": false,
+            "env": {
+                "PYTHONUNBUFFERED": "1"
+            }
+        },
+        {
+            // One-shot read of tags with a --server override — no config
+            // file needed. Adjust the tags (and optionally the ProgID) to
+            // whichever server is registered on your machine.
+            "name": "opcda-diagnozer: read (--server override)",
+            "type": "debugpy",
+            "request": "launch",
+            "python": "${workspaceFolder}/.venv/Scripts/python.exe",
+            "module": "opcda_diagnozer",
+            "args": [
+                "read",
+                "--server", "Matrikon.OPC.Simulation.1",
+                "Bucket Brigade.Int1",
+                "Bucket Brigade.Real8"
            ],
            "cwd": "${workspaceFolder}",
            "console": "integratedTerminal",
@@ -0,0 +1,90 @@
+# Changelog
+
+All notable changes to the OPC DA Diagnozer are documented here. The format
+follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versions
+bump per [SemVer](https://semver.org/). Dates are YYYY-MM-DD.
+
+## [Unreleased]
+
+### Added — server discovery & browsing
+- `list-servers` verb: enumerates OPC DA servers locally or on a DCOM host via
+  the Automation Wrapper (`GetOPCServers`), with a no-COM registry-scan fallback
+  (`--source registry|automation|auto`), JSON output, and DA-version metadata.
+- Browsing of the connected server's address space, unified in the
+  `list-items [all|branches|tags]` verb: `all` prints the full structure tree
+  (branches with `/`, tags with `[ItemID: ...]`), `branches` the structure-only
+  flat list, `tags` the flat ItemID list for config `tags:`. `--branch`
+  (dotted path or repeated segments), `--depth N` (1 default, 0 = whole
+  subtree), `--json`. Replaces the earlier split `list-devices`/`list-tags`.
+- `browse.max_depth` config key (default 32) as the address-space walk guard,
+  validated by `check-config`; cycle-safe tree walk in `opc/browse.py`.
+- `SpaceNode` tree data model (branch/leaf) replacing the earlier
+  device-centric `DeviceInfo`.
+
+### Added — reading & CLI
+- `read` verb: one-shot read of explicit tags (or the config's tags) with a
+  table or `--json`, `--source device|cache`, and a non-zero exit when any tag
+  doesn't read GOOD — scriptable as a health check.
+- `probe` verb: connect once and read every configured tag once.
+- `--server` / `--host` overrides on every connecting verb; with `--server` no
+  config file is needed at all.
+- Frozen-exe default config path: `config.yaml` next to the exe (a Windows
+  service starts with `CWD = System32`), not the working directory.
+
+### Added — monitoring design (D16) & config
+- `devices:` config section with `DeviceEntry(name, tag, ok_values)`: D16
+  device-link indicators. The normal (connected) reading is configurable per
+  indicator (`ok_values`, default `[true]`) — inverted fault flags
+  `[false, 0]` and string states `["RUN"]` are first-class. Strict validation:
+  duplicate indicators and indicator/tag-list overlap are rejected;
+  `check-config` prints the indicators.
+- Detection rules D16 (`DEVICE_LINK_DOWN`/`DEVICE_LINK_UP`) and the plain-text
+  anomaly log reporter (`report/textlog.py` task) specified in the docs
+  (SPECIFICATION §6.1, §9.1; IMPLEMENTATION_PLAN Phases 1–2) — implementation
+  pending, see TODO.md.
+- Clear-text status rule (spec §7): every reporter writes `NAME (0xNN)` —
+  e.g. `BAD / communication failure (0x18)`.
+
+### Added — packaging
+- `scripts/build_exe.ps1`: PyInstaller build of the 32-bit console exe with
+  hidden imports for the deferred COM modules (incl. `win32timezone` — see
+  Fixed), embedded version metadata, a PE-bitness check, and smoke tests
+  including a **frozen-read canary** against a live server.
+- `--version` and this CHANGELOG.md / TODO.md.
+
+### Fixed
+- COM lifecycle: all COM references (browsers, groups, items) are released
+  **inside** the apartment before `CoUninitialize`; retained exception
+  tracebacks are cleared in `as_com_error` — both caused
+  "Win32 exception occurred releasing IUnknown" noise and hard crashes at exit.
+- Frozen exe: every tag read failed with
+  `SystemError: ... InvokeTypes ... exception set` because pywin32 lazily
+  imports `win32timezone` during VARIANT-DATE conversion and PyInstaller never
+  bundles it — now a hidden import, with the build canary guarding it.
+- Reads: per-tag failures now carry the underlying exception's own type and
+  text (`_per_tag_error`) instead of a bare "failed"; a degenerate device
+  timestamp (never-updated alias, DST math) can no longer fail a read — the
+  timestamp is metadata and is dropped to `None`.
+- Matrikon browse quirk: servers may return an empty ItemID for special leaves
+  (alias roots); the adapter falls back to the bare browse name.
+- HRESULT labels: `OPC_E_UNKNOWNITEMID` (0xC0040007) and
+  `OPC_E_INVALIDITEMID` (0xC0040008) were swapped; now corrected everywhere.
+
+### Verified
+- All CLI verbs live-tested against Matrikon OPC Simulation Server
+  (`Matrikon.OPC.Simulation.1`): discovery, browse (all kinds/depths), reads
+  (incl. alias tags after server-side alias repair), config-driven
+  `max_depth`, and the frozen exe (onedir + onefile) including the read canary.
+- Test suite: 108 tests, all passing (pure logic + fake-client driven, runs on
+  any OS; COM paths are live-smoke-tested on dev boxes only).
+
+## [0.1.0] - 2026-10-07
+
+### Added
+- Initial project skeleton: typed YAML config with strict actionable
+  validation (`check-config`), OPC DA quality decode tables (spec §7), event
+  model skeletons (`DiagEvent`, D1–D15 `EventKind`s), client/protocol
+  abstractions, and the fake in-process client powering cross-OS tests.
+- Phase 0 decisions recorded: 32-bit interpreter mandatory (Automation Wrapper
+  is a 32-bit in-proc DLL), ruamel.yaml over PyYAML (no 32-bit cp314 wheels),
+  COM imports import-guarded for cross-OS test runs.
@@ -10,9 +10,12 @@ disconnects, hung RPC calls, tags marked BAD/UNCERTAIN, stale timestamps, frozen
 and flapping quality — as structured events (JSONL logs, Windows Event Log, optional
 HTTP webhook).

-> **Status:** planning. Read the
-> [specification](docs/SPECIFICATION.md) and the
-> [implementation plan](docs/IMPLEMENTATION_PLAN.md) first.
+> **Status:** in active development — discovery, browsing (`list-items`), one-shot
+> reads (`read`/`probe`) and the exe build are **live and field-verified** against a
+> real server; the continuous monitoring loop, reporters and Windows service host are
+> next. See the [TODO](TODO.md) for current status and the
+> [implementation plan](docs/IMPLEMENTATION_PLAN.md) for the phases; read the
+> [specification](docs/SPECIFICATION.md) for detection rules and behavior.

 ## Why

@@ -33,8 +36,8 @@ See the full interpretation matrix in the [specification](docs/SPECIFICATION.md)

 - Windows service (auto-start, SCM crash recovery) with a foreground `run` mode for testing
 - Local or remote (DCOM) connections to any OPC DA server
- Server browsing: `list-servers` (discovery), `list-devices` and `list-tags`
-  (address-space walk) — live now
+- Server browsing: `list-servers` (discovery) and `list-items`
+  (address-space browse: full tree / branches / tags) — live now
 - Latency monitoring: spikes, streaks, and p95 trends against configurable thresholds
 - Hang watchdog: flags RPC calls that never return — the silent killer
 - Quality monitoring: BAD/UNCERTAIN transitions with decoded subcodes, persistent-bad
@@ -51,16 +54,21 @@ opcda-diagnozer list-servers
 opcda-diagnozer list-servers --host CONTROL-HOST01        # remote, via DCOM
 opcda-diagnozer list-servers --source registry --json     # no-COM local scan

-# list the devices (address-space branches) of the configured server
-opcda-diagnozer list-devices
-opcda-diagnozer list-devices --depth 2                    # include sub-devices
-opcda-diagnozer list-devices --json                       # script-friendly output
+# browse the server's item space (list-items all|branches|tags)
+#   all      = the full structure tree: every branch AND every tag in place
+#   branches = the structure nodes (channels/devices/folders — all "branches" to OPC DA)
+#   tags     = the flat ItemID list that goes into config `tags:`
+opcda-diagnozer list-items                       # top level of the tree
+opcda-diagnozer list-items --depth 0              # the whole tree (browse.max_depth guards)
+opcda-diagnozer list-items branches --depth 2     # structure, two levels
+opcda-diagnozer list-items tags --branch "Simulation Items"   # one subtree, flat ItemIDs
+opcda-diagnozer list-items --json                 # script-friendly output

-# list the tags of one device (the item_id values go into config `tags:`)
-opcda-diagnozer list-tags "Simulation Items.Bucket Brigade"
-opcda-diagnozer list-tags "Simulation Items" "Bucket Brigade"   # equivalent
-opcda-diagnozer list-tags --recursive "Simulation Items"  # whole subtree
-opcda-diagnozer list-tags                                 # every tag on the server
+# read tags right now (validation tool) — no config needed with --server
+opcda-diagnozer read --server Matrikon.OPC.Simulation.1 "Bucket Brigade.Int1"
+opcda-diagnozer read --json "Bucket Brigade.Int1" "Bucket Brigade.Real8"
+opcda-diagnozer read                                     # the config's tags
+opcda-diagnozer read --source cache "Bucket Brigade.Int1" # server cache, no device trip

 # validate config
 opcda-diagnozer check-config
@@ -82,13 +90,29 @@ hosts or when the wrapper DLL is not registered. `auto` tries the first, falls b
 the second. Note the bitness rule still applies: a 32-bit process enumerates the
 32-bit registry view — exactly the servers this tool can load.

-`list-devices` / `list-tags` browse the **connected server's** address space through
-the Automation Wrapper's browser (`CreateBrowser`): devices are *branches*, tags are
-*leaves*. They work locally and over DCOM (same Launch+Activation requirements). A
-device path is either one dotted string (`"Simulation Items.Bucket Brigade"`) or
-several quoted branch names — use the latter when a branch name itself contains a
-dot. On flat-namespace servers (`organization: flat`) every tag hangs directly off
-the root; `list-tags` without a device lists them all.
+`list-items` browses the **connected server's** address space through the
+Automation Wrapper's browser (`CreateBrowser`). OPC DA defines exactly two
+element kinds — **branches** (the structure: what SCADA vendors call channels,
+devices, groups or folders; the protocol cannot tell them apart) and **leaves**
+(the tags). The `all` view renders the full tree — including leaves that hang
+directly at the server root and belong to no branch (server-internal items
+like `@ClientCount`). It works locally and over DCOM (same Launch+Activation
+requirements). A `--branch` path is either one dotted string
+(`"Simulation Items.Bucket Brigade"`) or repeated flag segments — use the
+latter when a branch name itself contains a dot. `--depth 1` (default) lists
+one level; `--depth 0` walks the whole subtree, guarded by `browse.max_depth`
+from the config (default 32) against server-side cycles.
+
+Every connecting verb (`read`, `probe`, `list-items`) accepts
+`--server PROGID` and `--host HOST` to **override the config's `opc` section** —
+point a one-shot check at any server without touching the deployed config.yaml.
+With `--server` given, no config file is needed at all; without it, the config must
+exist (and `probe`/`read` read the config's tag list unless tags are passed).
+
+`read` prints one row per tag with the decoded quality and the device timestamp;
+a non-GOOD or failing tag makes the exit code non-zero, so it doubles as a
+scriptable health check. `--source cache` reads the server cache instead of
+forcing a device round-trip.
 ### Configuration example (`config.example.yaml`)

 ```yaml
@@ -105,6 +129,15 @@ tags:
  - { name: "Channel1.Device1.Counter", expect_change: true }
  - { name: "Channel1.Device1.Level" }

+# D16 device-link indicators: tags saying whether the OPC server can reach
+# the device. DOWN (DEVICE_LINK_DOWN, CRITICAL) when the tag reads non-GOOD,
+# errors, or its GOOD value is NOT in ok_values. ok_values defaults to [true]
+# (boolean connected flags); inverted fault flags -> [false, 0]; string
+# states -> ["RUN"]. Do NOT also list these under tags:.
+devices:
+  - { name: "Conveyor 3", tag: "Channel1.Device3.Connected" }
+  - { name: "Pump A", tag: "Channel1.PumpA.Fault", ok_values: [false, 0] }
+
 latency:
  warn_ms: 1000
  critical_ms: 5000
@@ -119,6 +152,9 @@ stale:
  enabled: true
  max_age_s: 120

+browse:
+  max_depth: 32                         # address-space walk guard (list-items --depth 0)
+
 report:
  heartbeat_s: 300
  jsonl:  { dir: "C:/ProgramData/OPCDA-Diagnozer/logs", max_mb: 50, backups: 10 }
@@ -188,10 +224,53 @@ Notes:
  (`py install 3.14 -a x86`) or the installer from python.org ("Download Windows
  installer (32-bit)").

+## Building the Windows executable
+
+`scripts/build_exe.ps1` packages the CLI as a standalone **32-bit** console exe
+with [PyInstaller](https://pyinstaller.org) — no Python installation needed on
+the target host:
+
+```powershell
+& ".\.venv\Scripts\python.exe" -m pip install pyinstaller   # once
+.\scripts\build_exe.ps1                                     # dist\opcda-diagnozer\
+.\scripts\build_exe.ps1 -Onefile                            # single exe alternative
+.\scripts\build_exe.ps1 -Clean                              # fresh rebuild
+```
+
+The script refuses to run from a 64-bit interpreter (the exe must load the
+32-bit OPC DA wrapper DLLs, SPECIFICATION §11), re-verifies the produced PE
+really is 32-bit, embeds the project version in the exe properties, declares
+the hidden imports the code loads lazily (pywin32 COM, browse modules, the
+Phase 3 service host), and smoke-tests the artifact (`--version`,
+`check-config`, `list-servers --source registry`) after building. It also
+installs PyInstaller into the venv automatically if it is missing.
+
+If your policy blocks the script, run it with
+`powershell -NoProfile -ExecutionPolicy Bypass -File scripts\build_exe.ps1`.
+
+Deployment notes:
+- The **default config path** of a frozen exe is `config.yaml` **next to the
+  exe** — not the working directory. A Windows service starts with
+  `CWD = C:\Windows\System32`, so the relative default would silently point
+  there; the exe-dir default keeps the deployment layout simple
+  (`C:\Program Files\OPCDA-Diagnozer\{opcda-diagnozer.exe, config.yaml}`).
+- The exe is a console application that hosts every CLI verb, including the
+  future `run` / `probe` / service verbs. The Phase 3 Windows service
+  (`install` verb) will point `sc.exe` at this same exe with the service-host
+  entry point.
+- The Automation Wrapper DLL (`OPCAutomation.dll`) is **not** bundled — it is
+  a machine-installed COM component (registered in the registry), exactly like
+  the OPC DA servers the tool connects to. Only Python code and its
+  dependencies are frozen.
+- `onedir` is the default (faster startup, AV-friendlier); `--onefile` trades
+  startup time for a single file. `--clean` removes `build/` + `dist/` first.
+
 ## Documentation

 - [Specification](docs/SPECIFICATION.md) — detection rules, event model, operational behavior, acceptance criteria
 - [Implementation plan](docs/IMPLEMENTATION_PLAN.md) — phases, milestones, testing, risks
+- [TODO](TODO.md) — implementation status: what's done, what's next
+- [Changelog](CHANGELOG.md) — notable changes per release

 ## License

@@ -0,0 +1,98 @@
+# TODO — implementation status
+
+Companion to `docs/IMPLEMENTATION_PLAN.md` (the authoritative plan) and
+`CHANGELOG.md` (what shipped). One row per plan task; updated as work lands.
+Legend: **[done]** complete & verified · **[partial]** some of it landed ·
+**[todo]** not started · **[skip]** dropped by decision.
+
+## Phase 0 — Feasibility & environment **[done]**
+
+- [done] Wrapper backend decision: OPC DA Automation Wrapper (`OPC.Automation`)
+  → 32-bit interpreter mandatory (spec §11).
+- [done] Test server: Matrikon OPC Simulation Server, local ProgID
+  `Matrikon.OPC.Simulation.1`; baseline read RTT + wrapper error shapes
+  recorded (repo memory).
+- [done] Dependencies chosen for a 32-bit, no-compiler install:
+  ruamel.yaml (PyYAML has no cp314 win32 wheels), pywin32.
+
+## Phase 1 — Core engine
+
+- [done] 1. `config.py` — typed YAML config + strict validation; **D16
+  `devices:` indicators with configurable `ok_values`**; `check-config`
+  prints them.
+- [done] 2. `quality.py` — decode tables (spec §7).
+  - [todo] `status_text(raw) -> "NAME (0xNN)"` clear-text helper (spec §7
+    closing rule; used by every reporter).
+- [done] 3. `opc/base.py` — client/browser protocols, `ReadResult`,
+  `OpcError` with decoded HRESULT.
+- [done] 4. `opc/automation.py` — connect / browse / read (per-item
+  `Item.Read`; `SyncRead` is unmarshallable from pywin32 — verified), per-tag
+  errors, safe COM teardown.
+  - [skip] Single persistent group + static items (poller will create its own
+    per client; the throwaway-group read pattern is verified).
+- [todo] 5. `monitor/state.py` — connection state machine, exponential
+  backoff (1 s → 60 s cap), reconnect accounting (only the enum exists).
+- [todo] 6. `monitor/detectors.py` — D1–D15 pure detectors + rolling window
+  (only the `PollOutcome` dataclass exists).
+  - [todo] **D16 device-link detector** (`ok_values` semantics, edge-level,
+    `DEVICE_LINK_DOWN`/`DEVICE_LINK_UP`).
+- [todo] 7. `monitor/poller.py` — the COM thread poll loop + hang watchdog
+  (abandon & rebuild past `hard_hang_s`).
+  - [todo] Startup tag validation: probe every configured tag (incl. device
+    indicators) before entering the loop; fail fast on invalid ItemIDs.
+  - [open] Poll source default (CACHE vs DEVICE) — needs a spec decision:
+    Matrikon shows BAD qualities on some tags only via CACHE.
+- [todo] 8. `run` verb — foreground monitoring loop, event pipeline to
+  reporters, heartbeat (D15).
+- [done] 9. `opc/fake.py` — scriptable fake client (reads already scriptable
+  via `read_results`; hang/error sequences land with the poller).
+
+## Phase 2 — Reporters
+
+- [todo] 1. `report/jsonl.py` — rotating JSONL (size + backups), config-digest
+  header, atomic-ish append.
+- [todo] 2. `report/textlog.py` — **plain-text anomaly log** (spec §9.1):
+  `anomalies-YYYYMMDD.log`, one line per detected anomaly, clear-text status
+  `NAME (0xNN)`, recoveries logged so it reads as an incident narrative.
+- [todo] 3. `report/eventlog.py` — Windows Event Log (WARN+), source
+  registration via `install`.
+- [todo] 4. `report/http.py` — optional webhook, bounded queue, never blocks
+  polling.
+- [todo] 5. Heartbeat assembly (D15) + reporter fan-out dispatcher.
+
+## Phase 3 — Windows service
+
+- [todo] 1. `service.py` — `ServiceFramework` host; clean stop
+  (`STOP_PENDING`, join-with-timeout, abandon hung COM thread).
+- [todo] 2. CLI verbs `install`/`uninstall`/`start`/`stop` (sc.exe semantics).
+- [todo] 3. SCM recovery (`sc failure ... restart`), auto-start, service
+  account docs.
+- [todo] 4. `%ProgramData%\OPCDA-Diagnozer\` config/log dir conventions
+  (frozen exe already defaults there).
+- [open] Frozen-exe SCM hosting: console exe must expose `ServiceMain` — one
+  careful build pass (pywin32 `ServiceControlDispatcher`).
+
+## Phase 4 — Integration & chaos testing
+
+- [todo] Scenario matrix vs the sim server: server kill / service stop /
+  firewall silent drop / latency injection / device comm failure / quality
+  oscillation / stale+frozen / mass bad / config errors / **D16: flip a device
+  indicator (e.g. a `P500x.Connected` alias)**.
+- [todo] 24 h soak (AC10): no leaks, reconnect accounting correct, heartbeats
+  every 5 min, ≤ ~50 MB RAM, ~0 % CPU idle.
+
+## Phase 5 — Packaging & docs
+
+- [done] PyInstaller exe build (`scripts/build_exe.ps1`, onedir + onefile,
+  read canary) — live-verified.
+- [done] README (install, config, DCOM checklist, HRESULT table).
+- [todo] Ops runbook: reading events, kind meanings (spec §12 matrix),
+  threshold tuning, evidence locations.
+- [todo] 1.0.0 tag once AC1–AC10 pass.
+
+## Housekeeping
+
+- [todo] Wire `status_text` into `read`/`probe` output once it exists
+  (currently shows `raw_hex` + decoded name separately).
+- [todo] Record the CACHE-vs-DEVICE poll-source decision in the spec once
+  made; add fake-client scripted reads for both sources to the poller tests.
@@ -16,6 +16,19 @@ tags:
  - { name: "Channel1.Device1.Counter", expect_change: true }   # D13 frozen-value detection applies
  - { name: "Channel1.Device1.Level" }

+# D16 device-link indicators: tags saying whether the OPC server can reach the
+# device. DOWN (= DEVICE_LINK_DOWN, CRITICAL, clear-text status in the log) when
+# the tag reads non-GOOD, errors, or its GOOD value is NOT in ok_values.
+# ok_values defaults to [true] (boolean connected-flags); set it per system
+# convention: inverted fault flags -> [false, 0], string states -> ["RUN"].
+# Compared by string equality on the rendered value ("True"/"1"/"RUN").
+# Do NOT also list these under tags: - they are polled automatically.
+devices:
+  - { name: "Conveyor 3", tag: "Channel1.Device3.Connected" }              # ok: true/1
+  - { name: "Conveyor 4", tag: "Channel1.Device4.Connected", ok_values: [true, 1] }
+  - { name: "Pump A", tag: "Channel1.PumpA.Fault", ok_values: [false, 0] } # inverted: 0 = ok
+  - { name: "Boiler 2", tag: "Channel1.Boiler2.State", ok_values: ["RUN"] }
+
 latency:
  warn_ms: 1000                         # D4 single-spike threshold
  critical_ms: 5000                     # D5 sustained-slow severity
@@ -31,6 +44,9 @@ stale:
  enabled: true
  max_age_s: 120                        # D12 device timestamp older than this while GOOD

+browse:
+  max_depth: 32                         # address-space walk guard (list-items --depth 0)
+
 report:
  heartbeat_s: 300                      # D15 periodic stats snapshot
  jsonl:
@@ -27,7 +27,7 @@ opcda-diagnozer/
 ├── src/opcda_diagnozer/
 │   ├── __init__.py
 │   ├── __main__.py            # python -m opcda_diagnozer
-│   ├── main.py                # CLI: run / install / uninstall / check-config / probe / list-devices / list-tags
+│   ├── main.py                # CLI: run / install / uninstall / check-config / probe / read / list-items
 │   ├── config.py              # pydantic-free typed config (dataclasses) + YAML load + validation
 │   ├── events.py              # DiagEvent, Severity, EventKind
 │   ├── quality.py             # OPC DA quality decode tables
@@ -44,6 +44,7 @@ opcda-diagnozer/
 │   ├── report/
 │   │   ├── base.py            # Reporter protocol
 │   │   ├── jsonl.py           # rotating JSONL file
+│   │   ├── textlog.py         # plain-text anomaly log (human-readable, §9.1)
 │   │   ├── eventlog.py        # Windows Event Log (WARN+)
 │   │   └── http.py            # optional webhook
 │   └── service.py             # pywin32 ServiceFramework wrapper
@@ -82,7 +83,13 @@ Tasks:

 1. `config.py` — YAML config (see `config.example.yaml` sketch in README), dataclass
   model, strict validation, `check-config` messages that explain exactly what is wrong.
-2. `quality.py` — decode tables (spec §7), `decode_quality(raw: int) -> QualityInfo`.
+   Includes the **device-link indicators** (spec §6.1 / D16): `devices:` — each entry a
+   human-readable device name plus its connection-indicator tag ItemID. Indicators are
+   added to the polled tag set automatically; `check-config` warns when an indicator tag
+   is also listed in `tags:` (it should not be) and fails on an empty device name.
+2. `quality.py` — decode tables (spec §7), `decode_quality(raw: int) -> QualityInfo`,
+   plus the **clear-text status helper**: `status_text(raw) -> "NAME (0xNN)"` used by
+   every reporter (spec §7 closing rule — e.g. `BAD / communication failure (0x18)`).
 3. `opc/base.py` — client abstraction:
   ```python
   class OpcClient(Protocol):
@@ -99,6 +106,14 @@ Tasks:
 6. `monitor/detectors.py` — D1–D15 as small pure functions/classes over
   `PollOutcome` (list[ReadResult] + rtt_ms + error), using an injected clock for
   testability. Rolling window (deque, size `window_polls`) for p95/streak logic.
+   **D16 device-link detector** (spec §6.1): per configured indicator — quality
+   not GOOD, a read error, or a GOOD read whose value is **not in the
+   configurable `ok_values` list** (default `[true]`; string-equality compare
+   of the rendered value, so inverted fault flags `[false, 0]` and string states
+   `["RUN"]` both work) ⇒ `DEVICE_LINK_DOWN` (CRITICAL) carrying the device name,
+   indicator tag, clear-text status and value; recovery (`DEVICE_LINK_UP`) when
+   it reads GOOD with an ok value again. Edge-level per device, so flapping
+   links don't spam (one event per transition).
 7. `monitor/poller.py` — the single COM thread; watchdog logic for hang detection
   (spec §10.4): a second thread timestamps poll starts/ends; on `hard_hang_s` the COM
   thread is abandoned and rebuilt.
@@ -107,17 +122,26 @@ Tasks:
   hangs, quality flips) → powers the entire unit-test suite without COM/hardware.

 **Exit:** `opcda-diagnozer run` (foreground console) produces correct events against the
-fake client; unit tests for all detectors.
+fake client; unit tests for all detectors — including D16 edge semantics (false /
+non-GOOD / read-error / misconfigured value / recovery) against scripted indicator reads.

 ## 5. Phase 2 — Reporters (≈1 day)

 1. `report/jsonl.py` — rotation (size + backups), atomic-ish append, header line with
   config digest for provenance.
-2. `report/eventlog.py` — pywin32 `win32evtlog` usage; map severities; source
+2. `report/textlog.py` — **plain-text anomaly log** (spec §9.1): `anomalies-YYYYMMDD.log`
+   in the same directory as the JSONL files, one human-readable line per detected
+   anomaly (`timestamp  SEVERITY  KIND  message`) with clear-text status `NAME (0xNN)`.
+   Anomalies only — heartbeats/CONNECTED are filtered out (a small `is_anomaly(event)`
+   predicate on `EventKind`); recoveries (D2, `DEVICE_LINK_UP`, quality back to GOOD)
+   **are** logged, so the file reads as an incident narrative. Daily + size rotation
+   with `backups` kept, same scheme as the JSONL reporter. Every line corresponds to
+   a JSONL record (text = convenience view, JSONL = source of truth).
+3. `report/eventlog.py` — pywin32 `win32evtlog` usage; map severities; source
   registration via `install` step.
-3. `report/http.py` — optional webhook with timeout + simple retry/drop on failure
+4. `report/http.py` — optional webhook with timeout + simple retry/drop on failure
   (never blocks the poll loop; queue with bound).
-4. Heartbeat assembly (D15): uptime, reconnects, RTT percentiles, tags-bad counts.
+5. Heartbeat assembly (D15): uptime, reconnects, RTT percentiles, tags-bad counts.

 **Exit:** events visible in file + Event Log; heartbeat cadence verified.

@@ -150,6 +174,7 @@ Scenario matrix against a simulation server (and, when available, the real serve
 | Quality oscillation | toggling sim items | D11 flapping |
 | Stale/frozen tag | simulation server static item marked `expect_change` | D12 / D13 |
 | Mass bad quality | all sim items to bad | D14 |
+| **Device link down** | indicator tag flipped to `false` / set BAD (alias a bool sim item per device, e.g. `P5001.Connected` on Matrikon) | D16 `DEVICE_LINK_DOWN`, recovery line when restored |
 | Config errors | bad ProgID, unknown tag | fast fail + actionable message; `OPC_E_INVALIDITEMID` surfaced at startup |
 | Soak | 24 h healthy run | no leaks, heartbeats every 5 min, zero false positives |

@@ -99,9 +99,44 @@ diagnosis.
 | D13 | Frozen value | `expect_change: true` tag with unchanged value+timestamp for `max_age_s` | WARN |
 | D14 | Mass quality degradation | ≥ 50% of monitored tags BAD simultaneously while connection OK | CRITICAL (`CONTROL_LINK_DEGRADED` — points at OPC-server↔device link, not DCOM) |
 | D15 | Heartbeat / summary | Periodic stats snapshot every `heartbeat_s` (300) | INFO |
+| D16 | Device link down | A configured device-indicator tag reads `false`, or any quality other than GOOD, or a read error | CRITICAL per device (`DEVICE_LINK_DOWN`); D1/D2 do **not** imply device-link state |

 All thresholds are configurable; severities are INFO / WARN / CRITICAL.

+### 6.1 Device connection-indicator tags (D16)
+
+The config names one or more tags as **device-link indicators**: values the
+control system maintains that say whether the OPC server can reach the device.
+The **normal (connected) reading is configurable per indicator** — conventions
+differ per control system, so it is never hardcoded:
+
+```yaml
+devices:
+  - { name: "Conveyor 3", tag: "...Connected" }              # any truthy: true/1
+  - { name: "Pump A",     tag: "...PumpAFault", ok_values: [false, 0] }   # inverted fault flag
+  - { name: "Boiler 2",   tag: "...State",   ok_values: ["RUN"] }     # enum/string state
+```
+
+Default `ok_values: [true]` (boolean connected-flags are the common case).
+Value comparison is by **string equality** on the read value rendered as text
+(`"True"`, `"1"`, `"RUN"`), so type quirks across systems cannot cause false
+positives; quality must also be GOOD. Semantics (all DOWN cases are logged as
+`DEVICE_LINK_DOWN`):
+
+| Indicator reading | Interpretation |
+|-------------------|------------------|
+| GOOD + value ∈ `ok_values` | Device connected — normal |
+| GOOD + value ∉ `ok_values` | **Device link down / fault** — CRITICAL (works for inverted fault-flags: a fault `1` is not in `[false, 0]`) |
+| Any quality other than GOOD (BAD/UNCERTAIN) | **Something is wrong with the device link** — CRITICAL |
+| Read error (tag unreadable) | **Something is wrong with the device link** — CRITICAL |
+
+Device-link state is per indicator tag: the config maps each indicator to a
+human-readable device name for the log lines (see `config.example.yaml`).
+Recovery (`DEVICE_LINK_UP`) is logged when the same indicator reads GOOD with
+an ok value again. A value that is neither in `ok_values` nor a recognized
+fault shape still logs DOWN — the normal state is explicit, and everything
+else is "not normal" by definition.
+
 ## 7. OPC DA quality codes (reference)

 OPC DA quality is a 16-bit field. Bits 7–6 of the low byte give the class:
@@ -126,6 +161,13 @@ decodes and names:
 Remaining UNCERTAIN subcodes (DA 2.05a) are decoded by the same table in `quality.py`.
 Every quality event carries the raw hex code **and** the decoded name.

+**Clear-text status rule:** wherever a status or quality is reported — JSONL
+fields, the text log, Event Log, console — it is written in clear text with
+the raw code in parentheses: `BAD / communication failure (0x18)`,
+`GOOD (0xC0)`. Nobody should have to decode hex codes by hand, but the raw
+code stays for unambiguous correlation with vendor documentation and the
+HRESULT table.
+
 ## 8. Event model

 Every event is a structured record:
@@ -163,12 +205,36 @@ alongside and clearly labeled.
 | Reporter | Purpose | Notes |
 |----------|---------|-------|
 | JSONL rotating file | Primary evidence trail | One JSON event per line; rotation by size; default dir `%ProgramData%\OPCDA-Diagnozer\logs` |
+| **Plain-text anomaly log** | Human-readable evidence | One line per **detected anomaly** (not heartbeats): timestamp, kind, tag/device, message with clear-text status `NAME (0xNN)`; same rotation scheme as JSONL. See §9.1 for the line format |
 | Windows Event Log | Ops visibility | WARN+ only; source `OPCDA-Diagnozer` |
 | Console | Foreground/debug runs | Human-readable |
 | HTTP webhook (optional) | Alert integration | POST one JSON event per request, optional bearer token |

 Reporters are pluggable (`Reporter.emit(event)`); a heartbeat event every 5 minutes
 proves liveness and enables baseline graphs even when everything is healthy.
+The JSONL file stays the machine-readable source of truth; the text log is a
+human-first convenience view of the same events — every text line corresponds
+to a JSONL record.
+
+### 9.1 Plain-text anomaly log format
+
+File `anomalies-YYYYMMDD.log` (daily rotation, size-capped) in the same log
+directory. One line per detected anomaly — anomalies only; heartbeats and
+reconnect-OK events do not appear here. Clear text first, codes in parentheses:
+
+```
+2026-10-07 14:23:01.123 UTC  WARN      QUALITY_BAD       tag 'Channel1.Device1.Counter' changed GOOD (0xC0) -> BAD / communication failure (0x18)
+2026-10-07 14:24:31.500 UTC  CRITICAL  DEVICE_LINK_DOWN  device 'Conveyor 3': link indicator 'Channel1.Device3.Connected' read false (GOOD)
+2026-10-07 14:26:02.780 UTC  CRITICAL  DEVICE_LINK_DOWN  device 'Conveyor 3': link indicator 'Channel1.Device3.Connected' quality BAD / not connected (0x08)
+2026-10-07 14:27:10.001 UTC  CRITICAL  POLL_HANG        read call hung > 5000 ms; COM thread abandoned and rebuilt
+2026-10-07 14:30:00.000 UTC  CRITICAL  CONTROL_LINK_DEGRADED  12 of 14 monitored tags BAD while connection OK
+```
+
+Fields (pipe-free, fixed order): `timestamp  SEVERITY  KIND  message`.
+Timestamps are the local receive time in UTC (§8). The message carries the
+clear-text status with the code between parentheses per §7. A matching
+recovery (link back up, quality back to GOOD) is also logged as an anomaly
+line, so the log reads as a continuous incident narrative.

 ## 10. Operational behavior

@@ -0,0 +1,229 @@
+# Build the Windows executable (dist\opcda-diagnozer.exe) with PyInstaller.
+#
+# Usage (from the repo root):
+#   powershell -NoProfile -ExecutionPolicy Bypass -File scripts\build_exe.ps1
+#   .\scripts\build_exe.ps1            # onedir (default): dist\opcda-diagnozer\
+#   .\scripts\build_exe.ps1 -Onefile   # single exe:       dist\opcda-diagnozer.exe
+#   .\scripts\build_exe.ps1 -Clean     # remove build\ + dist\ first
+#
+# The exe is a console application hosting every CLI verb (check-config,
+# list-servers, list-devices, list-tags, the future run/probe/service verbs).
+# It is not itself a service binary: the Phase 3 service host will register
+# this same exe with the SCM via the 'install' verb (IMPLEMENTATION_PLAN 6).
+#
+# Hard rules enforced here:
+#   1. 32-bit interpreter (SPECIFICATION 11): the OPC DA Automation Wrapper
+#      (OPCAutomation.dll) is a 32-bit in-proc COM DLL - a 64-bit exe cannot
+#      load it and every COM call fails at runtime. The produced PE is
+#      re-verified to be 32-bit (IMAGE_FILE_MACHINE_I386) after the build.
+#   2. PyInstaller must see the modules the code imports lazily inside
+#      functions (COM, browse, service host) - declared via -HiddenImport.
+[CmdletBinding()]
+param(
+    [switch]$Onefile,
+    [switch]$Clean
+)
+
+$ErrorActionPreference = 'Stop'
+
+$Repo        = Split-Path -Parent $PSScriptRoot
+$VenvPython  = Join-Path $Repo '.venv\Scripts\python.exe'
+$Entry       = Join-Path $Repo 'src\opcda_diagnozer\__main__.py'
+$Dist        = Join-Path $Repo 'dist'
+$Build       = Join-Path $Repo 'build'
+$Name        = 'opcda-diagnozer'
+$VersionFile = Join-Path $PSScriptRoot '.exe_version_info.txt'
+
+# Modules imported lazily inside functions (import guards, IMPLEMENTATION_PLAN 9);
+# PyInstaller's static analysis misses those, so they are declared explicitly.
+$HiddenImports = @(
+    # pywin32 / COM - deferred into methods in opc/automation.py, opc/discovery.py
+    'pythoncom',
+    'pywintypes',
+    'win32com.client',
+    # pywintypes.datetime(VARIANT DATE -> datetime conversion) lazily imports
+    # win32timezone at CONVERSION time - missing from the bundle, every read of
+    # item.TimeStamp raised SystemError('InvokeTypes ... exception set') frozen
+    # (observed 2026-10-07: all tags failed in the exe while the venv worked).
+    'win32timezone',
+    # opc backend modules imported lazily in main.py CLI verbs
+    'opcda_diagnozer.opc.automation',
+    'opcda_diagnozer.opc.browse',
+    # future service host (Phase 3) - keep it in the bundle from day one
+    'opcda_diagnozer.service'
+)
+
+function Fail {
+    param([string]$Message)
+    Write-Host 'build_exe: FAILED' -ForegroundColor Red
+    Write-Host "  $Message" -ForegroundColor Red
+    exit 1
+}
+
+function Invoke-Captured {
+    # Run a native command and capture stdout+stderr with its exit code.
+    # $ErrorActionPreference must be relaxed around 2>&1 on PS 5.1: with 'Stop',
+    # redirecting native stderr raises a NativeCommandError and kills the script.
+    param([string]$FilePath, [string[]]$Arguments)
+    $prev = $ErrorActionPreference
+    $ErrorActionPreference = 'Continue'
+    try {
+        $lines = & $FilePath @Arguments 2>&1
+        $text = ($lines | ForEach-Object { "$_" }) -join [Environment]::NewLine
+        [pscustomobject]@{ ExitCode = $LASTEXITCODE; Output = $text }
+    }
+    finally {
+        $ErrorActionPreference = $prev
+    }
+}
+
+# --- environment checks ------------------------------------------------------
+
+if (-not (Test-Path $VenvPython)) {
+    Fail ("the 32-bit venv was not found: $VenvPython - create it first " +
+          "(README 'Development setup'): " +
+          '& "C:\...\Python314-32\python.exe" -m venv .venv; ' +
+          '& .\.venv\Scripts\python.exe -m pip install -e ".[dev]"')
+}
+
+$bitness = (Invoke-Captured -FilePath $VenvPython -Arguments @('-c', "import struct; print(struct.calcsize('P') * 8)")).Output.Trim()
+if ($bitness -ne '32') {
+    Fail ("the venv interpreter is ${bitness}-bit, but the OPC DA wrappers are " +
+          '32-bit in-proc DLLs (SPECIFICATION 11): a 64-bit exe cannot load ' +
+          "them. Rebuild the venv with the 32-bit Python (README 'Development setup').")
+}
+Write-Host 'build_exe: OK - 32-bit interpreter (SPECIFICATION 11 bitness rule)'
+
+$pyi = Invoke-Captured -FilePath $VenvPython -Arguments @('-m', 'PyInstaller', '--version')
+if ($pyi.ExitCode -ne 0) {
+    Write-Host 'build_exe: pyinstaller missing - installing into the venv ...'
+    & $VenvPython -m pip install pyinstaller
+    if ($LASTEXITCODE -ne 0) { Fail 'pip install pyinstaller failed' }
+}
+
+if ($Clean) {
+    foreach ($folder in @($Build, $Dist)) {
+        if (Test-Path $folder) { Remove-Item -Recurse -Force $folder }
+    }
+    Write-Host 'build_exe: cleaned build/ and dist/'
+}
+
+# --- version metadata (exe properties -> Details tab) -------------------------
+
+$initText = Get-Content (Join-Path $Repo 'src\opcda_diagnozer\__init__.py') -Raw
+if ($initText -notmatch '__version__\s*=\s*"([^"]+)"') {
+    Fail 'could not read __version__ from src\opcda_diagnozer\__init__.py'
+}
+$Version = $Matches[1]
+$parts = $Version.Split('.')
+$major = $parts[0]
+$minor = $parts[1]
+
+# Keyword names must match PyInstaller's FixedFileInfo.__init__ exactly:
+# filevers, prodvers, mask, flags, OS, fileType, subtype, date.
+# [IO.File]::WriteAllText writes UTF-8 WITHOUT a BOM - Set-Content -Encoding
+# UTF8 on PS 5.1 prepends one, and the BOM breaks PyInstaller's eval().
+$VersionInfo = @"
+# UTF-8
+VSVersionInfo(
+  ffi=FixedFileInfo(
+    filevers=($major, $minor, 0, 0),
+    prodvers=($major, $minor, 0, 0),
+    mask=0x3F,
+    flags=0x0,
+    OS=0x40004,
+    fileType=0x1,
+    subtype=0x0,
+    date=(0, 0)
+  ),
+  kids=[
+    StringFileInfo(
+      [
+        StringTable(
+          '040904B0',
+          [
+            StringStruct('CompanyName', 'Vivab'),
+            StringStruct('FileDescription', 'OPC DA Diagnozer - read-only OPC DA reference diagnostic client'),
+            StringStruct('FileVersion', '$Version.0'),
+            StringStruct('InternalName', 'opcda-diagnozer'),
+            StringStruct('OriginalFilename', 'opcda-diagnozer.exe'),
+            StringStruct('ProductName', 'OPC DA Diagnozer'),
+            StringStruct('ProductVersion', '$Version')
+          ]
+        )
+      ]
+    ),
+    VarFileInfo([VarStruct('Translation', [1033, 1200])])
+  ]
+)
+"@
+[IO.File]::WriteAllText($VersionFile, $VersionInfo)
+
+# --- the build itself ---------------------------------------------------------
+
+$pyArgs = @(
+    '-m', 'PyInstaller',
+    '--noconfirm', '--clean',
+    '--name', $Name,
+    '--distpath', $Dist,
+    '--workpath', $Build,
+    '--specpath', $Build,
+    '--console',
+    '--version-file', $VersionFile
+)
+foreach ($hidden in $HiddenImports) { $pyArgs += @('--hidden-import', $hidden) }
+$pyArgs += $(if ($Onefile) { '--onefile' } else { '--onedir' })
+$pyArgs += $Entry
+
+Write-Host 'build_exe: running PyInstaller...'
+& $VenvPython @pyArgs
+if ($LASTEXITCODE -ne 0) { Fail "PyInstaller exited $LASTEXITCODE" }
+
+$Exe = if ($Onefile) { Join-Path $Dist "$Name.exe" } else { Join-Path $Dist "$Name\$Name.exe" }
+if (-not (Test-Path $Exe)) { Fail "expected exe not found: $Exe" }
+Write-Host "build_exe: OK - $Exe"
+
+# --- final bitness guard: the produced PE must be 32-bit (SPECIFICATION 11) --
+
+$bytes = [System.IO.File]::ReadAllBytes($Exe)
+$peOffset = [System.BitConverter]::ToInt32($bytes, 0x3C)
+$machine = [System.BitConverter]::ToUInt16($bytes, $peOffset + 4)
+if ($machine -ne 0x14C) {
+    Fail ("the built exe is not 32-bit (PE machine 0x{0:X4}) - build from the 32-bit venv" -f $machine)
+}
+Write-Host 'build_exe: OK - exe is 32-bit (PE machine 0x014C, IMAGE_FILE_MACHINE_I386)'
+
+# --- smoke test ---------------------------------------------------------------
+
+Write-Host "build_exe: smoke-testing $Name.exe ..."
+$ExampleConfig = Join-Path $Repo 'config.example.yaml'
+$checks = @(
+    @{ Label = '--version';               Arguments = @('--version') },
+    @{ Label = 'check-config';            Arguments = @('check-config', '--config', $ExampleConfig) },
+    @{ Label = 'list-servers (registry)'; Arguments = @('list-servers', '--source', 'registry') }
+)
+foreach ($check in $checks) {
+    $result = Invoke-Captured -FilePath $Exe -Arguments $check.Arguments
+    $first = ($result.Output -split '\r?\n' | Where-Object { $_.Trim() } | Select-Object -First 1)
+    Write-Host ("  {0} -> exit {1}: {2}" -f $check.Label, $result.ExitCode, $first)
+    if ($result.ExitCode -ne 0) { Fail ("smoke test failed: " + $check.Label) }
+}
+
+# Frozen-read canary: the COM timestamp conversion (item.TimeStamp) lazily
+# imports win32timezone - if it is missing from the bundle, EVERY read fails
+# with SystemError('InvokeTypes ... exception set'). The registry/discovery
+# smoke tests cannot catch this; only a real read does. Skipped gracefully
+# when no COM server is reachable on this build host.
+$canary = Invoke-Captured -FilePath $Exe -Arguments @(
+    'read', '--server', 'Matrikon.OPC.Simulation.1', 'Bucket Brigade.Int1'
+)
+$canaryFirst = ($canary.Output -split '\r?\n' | Where-Object { $_.Trim() } | Select-Object -First 2) -join ' | '
+if ($canary.ExitCode -eq 0) {
+    Write-Host "  read canary (Matrikon sim) -> exit 0: $canaryFirst"
+} elseif ($canary.Output -match 'SystemError.*InvokeTypes') {
+    Fail ("frozen read canary failed with the missing-win32timezone signature: $canaryFirst")
+} else {
+    Write-Host "  read canary skipped (no Matrikon sim reachable): exit $($canary.ExitCode)"
+}
+Write-Host 'build_exe: smoke test OK'
+exit 0
@@ -38,6 +38,23 @@ class TagEntry:
    expect_change: bool = False


+@dataclass(frozen=True)
+class DeviceEntry:
+    """D16 device-link indicator: a tag saying if the device is reachable.
+
+    DOWN (``DEVICE_LINK_DOWN``, CRITICAL) when ``tag`` reads non-GOOD, errors,
+    or its GOOD value is **not** in ``ok_values`` — the normal (connected)
+    reading, configurable per indicator because conventions differ per
+    control system: connected-flags ``[true]`` (default), inverted fault-flags
+    ``[false, 0]``, string states ``["RUN"]``. Compared by string equality on
+    the rendered value (SPECIFICATION §6.1).
+    """
+
+    name: str
+    tag: str
+    ok_values: tuple[str, ...] = ("true",)
+
+
@dataclass(frozen=True)
 class PollConfig:
    interval_s: float = 1.0
@@ -66,6 +83,13 @@ class StaleConfig:
    max_age_s: float = 120.0  # D12 / D13


+@dataclass(frozen=True)
+class BrowseConfig:
+    # Address-space walk guard: deeper SCADA trees are fine, but a walk this
+    # deep is almost certainly a server-side cycle (list-items --depth 0).
+    max_depth: int = 32
+
+
@dataclass(frozen=True)
 class JsonlConfig:
    dir: str = "C:/ProgramData/OPCDA-Diagnozer/logs"
@@ -93,14 +117,49 @@ class Config:
    opc: OpcConfig
    poll: PollConfig = field(default_factory=PollConfig)
    tags: tuple[TagEntry, ...] = ()
+    devices: tuple[DeviceEntry, ...] = ()
    latency: LatencyConfig = field(default_factory=LatencyConfig)
    quality: QualityConfig = field(default_factory=QualityConfig)
    stale: StaleConfig = field(default_factory=StaleConfig)
+    browse: BrowseConfig = field(default_factory=BrowseConfig)
    report: ReportConfig = field(default_factory=ReportConfig)


-_ALLOWED_TOP_LEVEL = {"opc", "poll", "tags", "latency", "quality", "stale", "report"}
+_ALLOWED_TOP_LEVEL = {"opc", "poll", "tags", "devices", "latency", "quality", "stale", "browse", "report"}
 _TAG_KEYS = {"name", "expect_change"}
+_DEVICE_KEYS = {"name", "tag", "ok_values"}
+
+
+def _normalize_ok_values(raw: Any, where: str) -> tuple[str, ...]:
+    """ok_values as a tuple of rendered strings ("true", "1", "RUN").
+
+    The D16 comparison is string equality on the value rendered as text, so
+    booleans/ints/strings from YAML all work uniformly: `[true]` -> `("true",)`,
+    `[false, 0]` -> `("false", "0")`.
+    """
+    if raw is None:
+        return ("true",)
+    if isinstance(raw, (list, tuple)):
+        if not raw:
+            raise ConfigError(f"{where}: ok_values must not be empty - set the normal (connected) value(s)")
+        values = []
+        for item in raw:
+            if isinstance(item, bool):
+                values.append("true" if item else "false")
+            elif isinstance(item, (int, float, str)):
+                values.append(str(item).strip())
+            else:
+                raise ConfigError(
+                    f"{where}: ok_values entries must be scalars (bool/int/str), got {type(item).__name__}"
+                )
+        cleaned = tuple(v for v in values if v)
+        if not cleaned:
+            raise ConfigError(f"{where}: ok_values must not be empty - set the normal (connected) value(s)")
+        return cleaned
+    # a single scalar is shorthand for a one-element list
+    if isinstance(raw, (bool, int, float, str)):
+        return _normalize_ok_values([raw], where)
+    raise ConfigError(f"{where}: ok_values must be a list of scalars, got {type(raw).__name__}")


 def _build_section(cls: type[Any], data: Any, where: str) -> Any:
@@ -161,6 +220,37 @@ def _build_tags(data: Any) -> tuple[TagEntry, ...]:
    return tuple(entries)


+def _build_devices(data: Any) -> tuple[DeviceEntry, ...]:
+    """Parse the devices: section (D16 link indicators) with actionable errors."""
+    if data is None:
+        return ()
+    if not isinstance(data, list):
+        raise ConfigError(f"devices: must be a list, got {type(data).__name__}")
+    entries: list[DeviceEntry] = []
+    for i, item in enumerate(data):
+        if not isinstance(item, dict):
+            raise ConfigError(
+                f"devices[{i}]: must be a mapping with 'name' and 'tag'"
+            )
+        unknown = sorted(set(item) - _DEVICE_KEYS)
+        if unknown:
+            raise ConfigError(
+                f"devices[{i}]: unknown key(s) {unknown}; accepted keys: {sorted(_DEVICE_KEYS)}"
+            )
+        name = item.get("name")
+        tag = item.get("tag")
+        if not isinstance(name, str) or not name.strip():
+            raise ConfigError(f"devices[{i}]: 'name' is required and must be a non-empty string")
+        if not isinstance(tag, str) or not tag.strip():
+            raise ConfigError(
+                f"devices[{i}]: 'tag' is required and must be the connection-indicator "
+                "tag ItemID (see list-items)"
+            )
+        ok_values = _normalize_ok_values(item.get("ok_values"), f"devices[{i}]")
+        entries.append(DeviceEntry(name=name, tag=tag, ok_values=ok_values))
+    return tuple(entries)
+
+
 def _validate(config: Config) -> None:
    """Cross-field validation; each message names the offending key."""
    if not config.opc.server or not config.opc.server.strip():
@@ -178,6 +268,24 @@ def _validate(config: Config) -> None:
    duplicates = sorted({n for n in names if names.count(n) > 1})
    if duplicates:
        raise ConfigError(f"tags: duplicate tag name(s) {duplicates}")
+    device_names = [d.name for d in config.devices]
+    duplicate_devices = sorted({n for n in device_names if device_names.count(n) > 1})
+    if duplicate_devices:
+        raise ConfigError(f"devices: duplicate device name(s) {duplicate_devices}")
+    indicator_tags = [d.tag for d in config.devices]
+    duplicate_indicators = sorted({t for t in indicator_tags if indicator_tags.count(t) > 1})
+    if duplicate_indicators:
+        raise ConfigError(
+            f"devices: duplicate indicator tag(s) {duplicate_indicators} - one device per indicator"
+        )
+    # An indicator doubles as a monitored tag; listing it in tags: too would double-poll
+    # it and double-report its quality transitions (D8 AND D16 fire).
+    overlap = sorted(set(indicator_tags) & set(names))
+    if overlap:
+        raise ConfigError(
+            f"tags: {overlap} also appear as device indicators - remove them from "
+            "tags: (devices: entries are polled automatically)"
+        )
    if config.poll.interval_s <= 0:
        raise ConfigError(f"poll.interval_s: must be > 0, got {config.poll.interval_s}")
    if config.poll.timeout_ms <= 0:
@@ -207,6 +315,8 @@ def _validate(config: Config) -> None:
        raise ConfigError(f"quality.flapping_window_s: must be > 0, got {config.quality.flapping_window_s}")
    if config.stale.enabled and config.stale.max_age_s <= 0:
        raise ConfigError(f"stale.max_age_s: must be > 0, got {config.stale.max_age_s}")
+    if config.browse.max_depth < 1:
+        raise ConfigError(f"browse.max_depth: must be >= 1, got {config.browse.max_depth}")
    if config.report.heartbeat_s <= 0:
        raise ConfigError(f"report.heartbeat_s: must be > 0, got {config.report.heartbeat_s}")
    if config.report.eventlog.min_severity not in ("INFO", "WARN", "CRITICAL"):
@@ -243,9 +353,11 @@ def load_config(path: str | Path) -> Config:
        opc=_build_section(OpcConfig, data.get("opc"), "opc"),
        poll=_build_section(PollConfig, data.get("poll"), "poll"),
        tags=_build_tags(data.get("tags")),
+        devices=_build_devices(data.get("devices")),
        latency=_build_section(LatencyConfig, data.get("latency"), "latency"),
        quality=_build_section(QualityConfig, data.get("quality"), "quality"),
        stale=_build_section(StaleConfig, data.get("stale"), "stale"),
+        browse=_build_section(BrowseConfig, data.get("browse"), "browse"),
        report=_build_report(data.get("report")),
    )
    try:
@@ -2,9 +2,9 @@

 Verbs: ``run``, ``probe``, ``check-config``, ``list-items`` plus the service
 verbs ``install`` / ``uninstall`` / ``start`` / ``stop``. Currently functional:
-``check-config``, ``list-servers`` (OPC DA server discovery) and the browse
-verbs ``list-devices`` / ``list-tags`` (devices = branches, tags = leaves of
-the connected server's address space); the remaining verbs land with Phases 1–3.
+``check-config``, ``list-servers`` (discovery), ``list-items`` (address-space
+browsing: ``all`` / ``branches`` / ``tags``) and ``read`` / ``probe`` (one-shot
+tag reads); the remaining verbs land with Phases 1–3.
 """

 from __future__ import annotations
@@ -12,6 +12,7 @@ from __future__ import annotations
 import argparse
 import sys
 from collections.abc import Sequence
+from pathlib import Path

 from opcda_diagnozer import __version__
 from opcda_diagnozer.config import ConfigError, load_config
@@ -20,10 +21,21 @@ from opcda_diagnozer.opc.discovery import ServerInfo, list_servers

 DEFAULT_CONFIG = "config.yaml"

+
+def _default_config_path() -> str:
+    """Where the config lives by default: CWD in source runs, exe dir when frozen.
+
+    A frozen exe must not rely on the CWD: a Windows service starts with
+    ``CWD = C:\\Windows\\System32``, so the relative name would silently resolve
+    into System32. Frozen runs therefore default to ``config.yaml`` **next to
+    the exe** — the deployment layout the Phase 3 service will use (SPEC §6.4).
+    """
+    if getattr(sys, "frozen", False):
+        return str(Path(sys.executable).resolve().parent / DEFAULT_CONFIG)
+    return DEFAULT_CONFIG
+
 _NOT_IMPLEMENTED: dict[str, str] = {
    "run": "Phase 1 (core engine)",
-    "probe": "Phase 1 (core engine)",
-    "list-items": "superseded by list-devices / list-tags",
    "install": "Phase 3 (Windows service)",
    "uninstall": "Phase 3 (Windows service)",
    "start": "Phase 3 (Windows service)",
@@ -34,11 +46,30 @@ _NOT_IMPLEMENTED: dict[str, str] = {
 def _add_config_option(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        "--config", "-c",
-        default=DEFAULT_CONFIG,
+        default=_default_config_path(),
        help="path to the YAML config (default: %(default)s)",
    )


+def _add_server_overrides(parser: argparse.ArgumentParser) -> None:
+    """--server/--host override the config's opc section (SPECIFICATION §6).
+
+    They make one-shot CLI use possible without a config file — and let an
+    ops engineer point a validation verb at a different server without
+    touching the deployed config.yaml.
+    """
+    parser.add_argument(
+        "--server",
+        metavar="PROGID",
+        help="OPC DA server ProgID, overriding the config (e.g. Matrikon.OPC.Simulation.1)",
+    )
+    parser.add_argument(
+        "--host",
+        metavar="HOST",
+        help="remote DCOM host, overriding the config (omit for a local server)",
+    )
+
+
 def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="opcda-diagnozer",
@@ -53,15 +84,48 @@ def _build_parser() -> argparse.ArgumentParser:
    p = sub.add_parser("run", help="run in the foreground (console)")
    _add_config_option(p)

-    p = sub.add_parser("probe", help="connect once and print one read of every configured tag")
+    p = sub.add_parser(
+        "read",
+        help="connect once and read the given tags (a quick validation tool)",
+    )
    _add_config_option(p)
+    _add_server_overrides(p)
+    p.add_argument(
+        "tags",
+        nargs="*",
+        metavar="TAG",
+        help="tag ItemIDs to read (e.g. 'Bucket Brigade.Int1'); omitted -> the "
+        "config's tags",
+    )
+    p.add_argument(
+        "--source",
+        choices=("device", "cache"),
+        default="device",
+        help="read from the device (round-trip to the PLC) or the server cache "
+        "(default: %(default)s)",
+    )
+    p.add_argument(
+        "--json",
+        action="store_true",
+        help="emit one JSON object per tag (for scripts) instead of a table",
+    )
+
+    p = sub.add_parser(
+        "probe",
+        help="connect once and print one read of every configured tag",
+    )
+    _add_config_option(p)
+    _add_server_overrides(p)
+    p.add_argument(
+        "--source",
+        choices=("device", "cache"),
+        default="device",
+        help="read from the device or the server cache (default: %(default)s)",
+    )

    p = sub.add_parser("check-config", help="validate the configuration and exit")
    _add_config_option(p)

-    p = sub.add_parser("list-items", help="browse the server item space (validation tool)")
-    _add_config_option(p)
-
    p = sub.add_parser(
        "list-servers",
        help="enumerate the OPC DA servers available locally or on a DCOM host",
@@ -86,50 +150,42 @@ def _build_parser() -> argparse.ArgumentParser:
    )

    p = sub.add_parser(
-        "list-devices",
-        help="list the devices (address-space branches) exposed by the configured server",
+        "list-items",
+        help="browse the server's item space: the full structure tree, or a "
+        "flat view of its branches / tags",
    )
    _add_config_option(p)
+    _add_server_overrides(p)
+    p.add_argument(
+        "kind",
+        nargs="?",
+        choices=("all", "branches", "tags"),
+        default="all",
+        help="all = the full structure tree (branches AND tags in place); "
+        "branches = the structure nodes only (SCADA channels/devices/folders "
+        "are all branches to OPC DA); tags = the flat ItemID list that goes "
+        "into config `tags:` (default: %(default)s)",
+    )
+    p.add_argument(
+        "--branch",
+        action="append",
+        metavar="PATH",
+        help="limit the listing to this branch subtree: one dotted string "
+        "('--branch \"Simulation Items.Bucket Brigade\"') or repeat the flag "
+        "for explicit segment names containing dots (default: the server root)",
+    )
    p.add_argument(
        "--depth",
        type=int,
        default=1,
        metavar="N",
-        help="how many branch levels to descend (default: %(default)s)",
-    )
-    p.add_argument(
-        "--no-counts",
-        action="store_true",
-        help="skip fetching per-device tag counts (faster on huge servers)",
+        help="how many branch levels to descend; 0 = the whole subtree "
+        "(default: %(default)s)",
    )
    p.add_argument(
        "--json",
        action="store_true",
-        help="emit one JSON object per device (for scripts) instead of a table",
-    )
-
-    p = sub.add_parser(
-        "list-tags",
-        help="list the tags (leaves) of one device, or every tag on the server",
-    )
-    _add_config_option(p)
-    p.add_argument(
-        "device",
-        nargs="*",
-        metavar="BRANCH",
-        help="device path: one or more branch names — either a single dotted string "
-        "in quotes ('\"Simulation Items.Bucket Brigade\"') or separate quoted "
-        "segments ('\"Simulation Items\" \"Bucket Brigade\"'). Default: the server root",
-    )
-    p.add_argument(
-        "--recursive", "-r",
-        action="store_true",
-        help="walk the whole subtree, not just the device's direct tags",
-    )
-    p.add_argument(
-        "--json",
-        action="store_true",
-        help="emit one JSON object per tag (for scripts) instead of a table",
+        help="emit JSON objects (for scripts) instead of a tree/list",
    )

    p = sub.add_parser("install", help="install as a Windows service (requires admin)")
@@ -161,6 +217,10 @@ def _cmd_check_config(config_path: str) -> int:
    print("check-config: OK")
    print(f"  server: {config.opc.server!r}{host}")
    print(f"  tags:   {len(config.tags)} ({expect_change} with expect_change)")
+    if config.devices:
+        print(f"  devices: {len(config.devices)} link indicator(s):")
+        for device in config.devices:
+            print(f"    {device.name} <- {device.tag}")
    print(f"  poll:   every {config.poll.interval_s} s, hang flagged after {config.poll.timeout_ms} ms")
    return 0

@@ -211,61 +271,100 @@ def _cmd_list_servers(host: str | None, source: str, as_json: bool) -> int:
    return 0


-def _connect_from_config(config_path: str):
-    """Load the config and connect to its configured server (Automation Wrapper).
+def _connect_from_config(config_path: str, server: str | None = None, host: str | None = None):
+    """Load the config (when needed) and connect; --server/--host override opc.
+
+    Overrides make one-shot CLI use possible without a config file: when
+    ``--server`` is given, a missing config file is not an error and the
+    remaining settings fall back to their defaults. Without ``--server`` the
+    config must exist (its ``opc`` section decides where to connect).

    Returns the connected client; raises :class:`ConfigError` / :class:`OpcError`
    with actionable messages for the caller to print.
    """
    from opcda_diagnozer.opc.automation import AutomationOpcClient

-    config = load_config(config_path)
+    connect_timeout_s = 10.0  # OpcConfig default
+    config_exists = Path(config_path).is_file()
+    if config_exists:
+        config = load_config(config_path)
+        server = server or config.opc.server
+        if host is None:
+            host = config.opc.host
+        connect_timeout_s = config.opc.connect_timeout_s
+    elif server is None:
+        raise ConfigError(
+            f"config file not found: {config_path} - pass --server (and optionally "
+            "--host) to connect without a config, or create the config first"
+        )
    client = AutomationOpcClient(
-        server=config.opc.server,
-        host=config.opc.host,
-        connect_timeout_s=config.opc.connect_timeout_s,
+        server=server,
+        host=host,
+        connect_timeout_s=connect_timeout_s,
    )
    client.connect()
    return client


-def _print_devices(space, as_json: bool) -> None:
-    """Human table / JSON-lines output for the list-devices verb."""
-    if as_json:
-        import json
-
-        for device in space.devices:
-            payload = {
-                "name": device.name,
-                "path": list(device.path),
-                "qualified_name": device.qualified_name,
-                "sub_devices": list(device.sub_devices),
-                "tags": list(device.tag_names),
-            }
-            print(json.dumps(payload, ensure_ascii=False))
-        return
-    if not space.devices:
-        print(f"no devices (branches) found — organization: {space.organization}")
-        return
-    print(f"address space organization: {space.organization}")
-    print("devices:")
-    width = max(len(d.qualified_name) for d in space.devices)
-    for device in space.devices:
-        notes = [f"{len(device.tag_names)} tags"]
-        if device.sub_devices:
-            notes.append(f"{len(device.sub_devices)} sub-devices")
-        print(f"  {device.qualified_name:<{width}}  ({', '.join(notes)})")
+def _print_item_tree(nodes, indent: int = 0) -> None:
+    """Indented structure view: branches with a trailing /, tags with their ItemID."""
+    for node in nodes:
+        prefix = "  " * indent
+        if node.kind == "branch":
+            print(f"{prefix}{node.name}/")
+            _print_item_tree(node.children, indent + 1)
+        else:
+            print(f"{prefix}{node.name}  [ItemID: {node.item_id}]")


-def _cmd_list_devices(config_path: str, depth: int, include_counts: bool, as_json: bool) -> int:
-    from opcda_diagnozer.opc.browse import list_devices
+def _node_payload(node) -> dict:
+    """JSON payload for one SpaceNode (branches carry their children)."""
+    return {
+        "kind": node.kind,
+        "name": node.name,
+        "path": list(node.path),
+        "item_id": node.item_id,
+        "children": [_node_payload(c) for c in node.children],
+    }

+
+def _cmd_list_items(
+    config_path: str, server: str | None, host: str | None,
+    kind: str, branch: Sequence[str] | None, depth: int, as_json: bool,
+) -> int:
+    from opcda_diagnozer.opc.browse import (
+        DEFAULT_MAX_DEPTH, flatten_branches, flatten_leaves,
+        split_branch_path, walk_tree,
+    )
+
+    import json
+
+    branch_list = list(branch) if branch else None
+    # One --branch occurrence is a *dotted* branch path ("Simulation Items.Bucket
+    # Brigade"); several occurrences are explicit segment names (those may
+    # contain dots). Same convention the fake and browse helpers share.
+    branch_arg = (
+        branch_list[0]
+        if branch_list and len(branch_list) == 1
+        else branch_list
+    )
+    # --depth 0 means the whole subtree; the hard guard is browse.max_depth
+    # from the config (default 32) — a missing config is fine when --server
+    # is given, so fall back to the default guard.
+    max_depth = DEFAULT_MAX_DEPTH
+    if Path(config_path).is_file():
+        try:
+            max_depth = load_config(config_path).browse.max_depth
+        except ConfigError:
+            pass  # an invalid config surfaces in _connect_from_config below
    try:
-        client = _connect_from_config(config_path)
+        client = _connect_from_config(config_path, server, host)
        browser = None
        try:
            browser = client.create_browser()
-            space = list_devices(browser, depth=depth, include_tag_counts=include_counts)
+            nodes = walk_tree(
+                browser, branch_arg, depth=depth, max_depth=max_depth
+            )
        finally:
            # Release the browser's COM reference *before* close() tears down the
            # apartment — releasing an RCW after CoUninitialize crashes the process.
@@ -273,86 +372,184 @@ def _cmd_list_devices(config_path: str, depth: int, include_counts: bool, as_jso
                browser.close()
            client.close()
    except ConfigError as exc:
-        print(f"list-devices: FAILED\n  {exc}", file=sys.stderr)
+        print(f"list-items: FAILED\n  {exc}", file=sys.stderr)
        return 2
    except (OpcError, ValueError) as exc:
-        print(f"list-devices: FAILED\n  {exc}", file=sys.stderr)
+        print(f"list-items: FAILED\n  {exc}", file=sys.stderr)
        return 2
-    _print_devices(space, as_json)
-    if not as_json and not space.devices:
-        hint = (
-            "hint: the server address space is flat — every tag hangs directly off "
-            "the root; use: opcda-diagnozer list-tags"
-            if space.organization == "flat"
-            else "hint: no branches at the root — the server may expose only tags; "
-            "try: opcda-diagnozer list-tags"
+
+    path = split_branch_path(branch_arg)
+    where = ".".join(path) or "(server root)"
+
+    if kind == "all":
+        if not nodes:
+            print(f"nothing defined under {where}")
+            return 0
+        if as_json:
+            for node in nodes:
+                print(json.dumps(_node_payload(node), ensure_ascii=False))
+        else:
+            scope = "whole subtree" if depth == 0 else f"depth {depth}"
+            print(f"items under {where} ({scope}):")
+            _print_item_tree(nodes)
+        return 0
+
+    if kind == "tags":
+        tags = flatten_leaves(nodes)
+        if as_json:
+            for tag in tags:
+                print(json.dumps(
+                    {"name": tag.name, "item_id": tag.item_id, "path": list(tag.path)},
+                    ensure_ascii=False,
+                ))
+            return 0
+        if not tags:
+            print(
+                f"no tags found under {where} — increase --depth (0 = whole subtree) "
+                "or pick another --branch"
+            )
+            return 0
+        print(f"tags under {where}: {len(tags)}")
+        width = max(len(t.item_id) for t in tags)
+        for tag in tags:
+            print(f"  {tag.item_id:<{width}}  (leaf: {tag.name})")
+        return 0
+
+    # kind == "branches"
+    branches = flatten_branches(nodes)
+    if as_json:
+        for node in branches:
+            print(json.dumps({
+                "name": node.name,
+                "path": list(node.path),
+                "tags": sum(1 for c in node.children if c.kind == "leaf"),
+                "sub_branches": sum(1 for c in node.children if c.kind == "branch"),
+            }, ensure_ascii=False))
+        return 0
+    if not branches:
+        print(
+            f"no branches found under {where} — the space may be flat; "
+            "try kind 'all' or 'tags'"
        )
-        print(hint, file=sys.stderr)
+        return 0
+    print(f"branches under {where}: {len(branches)}")
+    width = max(len(".".join(b.path)) for b in branches)
+    for node in branches:
+        dotted = ".".join(node.path)
+        tags = sum(1 for c in node.children if c.kind == "leaf")
+        subs = sum(1 for c in node.children if c.kind == "branch")
+        notes = [f"{tags} tags"]
+        if subs:
+            notes.append(f"{subs} sub-branches")
+        print(f"  {dotted:<{width}}  ({', '.join(notes)})")
    return 0


-def _print_tags(tags, device_path: tuple[str, ...], recursive: bool, as_json: bool) -> None:
-    """Human table / JSON-lines output for the list-tags verb."""
-    where = ".".join(device_path) or "(server root)"
-    scope = " (recursive)" if recursive else ""
+def _print_read_results(results, as_json: bool) -> int:
+    """Human table / JSON-lines output for the read/probe verbs."""
+    from opcda_diagnozer.quality import decode_quality
+
+    failures = 0
    if as_json:
        import json

-        for tag in tags:
+        for result in results:
+            quality = decode_quality(result.quality_raw)
            payload = {
-                "name": tag.name,
-                "item_id": tag.item_id,
-                "path": list(tag.path),
+                "tag": result.tag,
+                "value": result.value,
+                "quality_raw": result.quality_raw,
+                "quality": quality.name,
+                "quality_class": quality.quality_class.value,
+                "device_time_utc": (
+                    result.device_time.isoformat() if result.device_time else None
+                ),
+                "error": result.error,
            }
-            print(json.dumps(payload, ensure_ascii=False))
-        return
-    if not tags:
-        print(f"no tags found under {where}{scope}")
-        return
-    print(f"tags under {where}{scope}: {len(tags)}")
-    width = max(len(t.item_id) for t in tags)
-    for tag in tags:
-        print(f"  {tag.item_id:<{width}}  (leaf: {tag.name})")
+            if result.error or quality.quality_class.value != "GOOD":
+                failures += 1
+            print(json.dumps(payload, ensure_ascii=False, default=str))
+        return failures
+    if not results:
+        print("no tags to read")
+        return 0
+    width = max(len(r.tag) for r in results)
+    print(f"{'tag':<{width}}  value                          quality  device time (UTC)")
+    for result in results:
+        quality = decode_quality(result.quality_raw)
+        if result.error:
+            failures += 1
+            print(f"  {result.tag:<{width}}  ERROR: {result.error}")
+            continue
+        marker = "" if quality.quality_class.value == "GOOD" else "  <-- " + quality.name
+        if quality.quality_class.value != "GOOD":
+            failures += 1
+        time_str = (
+            result.device_time.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+            if result.device_time
+            else "-"
+        )
+        print(
+            f"  {result.tag:<{width}}  {str(result.value):<30} {quality.raw_hex}   {time_str}{marker}"
+        )
+    return failures


-def _cmd_list_tags(
+def _cmd_read(
    config_path: str,
-    device: Sequence[str],
-    recursive: bool,
+    server: str | None,
+    host: str | None,
+    tags: Sequence[str],
+    source: str,
    as_json: bool,
 ) -> int:
-    from opcda_diagnozer.opc.browse import list_tags, split_device_path
+    """One-shot read of explicit tags (or the config's tags) from the server."""
+    from opcda_diagnozer.opc.base import OPCDataSource

-    # A single CLI argument is a *dotted* branch path; several arguments are
-    # explicit branch names (names may contain dots then).
-    device_arg = device[0] if len(device) == 1 else list(device)
+    requested = list(tags)
    try:
-        client = _connect_from_config(config_path)
-        browser = None
-        try:
-            browser = client.create_browser()
-            tags = list_tags(browser, device=device_arg or None, recursive=recursive)
-        finally:
-            if browser is not None:
-                browser.close()
-            client.close()
+        # connect first: a missing config with no --server must fail here with
+        # the actionable hint (before the tags load below can complain).
+        client = _connect_from_config(config_path, server, host)
+        if not requested:
+            # no tags given -> use the config's tags (config is mandatory then)
+            config = load_config(config_path)
+            requested = [t.name for t in config.tags]
    except ConfigError as exc:
-        print(f"list-tags: FAILED\n  {exc}", file=sys.stderr)
+        print(f"read: FAILED\n  {exc}", file=sys.stderr)
        return 2
-    except (OpcError, ValueError) as exc:
-        print(f"list-tags: FAILED\n  {exc}", file=sys.stderr)
+    except OpcError as exc:
+        print(f"read: FAILED\n  {exc}", file=sys.stderr)
        return 2
-    device_path = split_device_path(device_arg or None)
-    _print_tags(tags, device_path, recursive, as_json)
-    if not as_json and not tags and not recursive:
+    data_source = OPCDataSource.DEVICE if source == "device" else OPCDataSource.CACHE
+    try:
+        results = client.read(requested, source=data_source)
+    except OpcError as exc:
+        print(f"read: FAILED\n  {exc}", file=sys.stderr)
+        client.close()
+        return 2
+    finally:
+        client.close()
+    failures = _print_read_results(results, as_json)
+    if failures:
        print(
-            "hint: the branch exists but exposes no direct leaves — try --recursive "
-            "to include the whole subtree",
+            f"read: {failures} of {len(results)} tag(s) did not read GOOD",
            file=sys.stderr,
        )
+        return 1
    return 0


+def _cmd_probe(
+    config_path: str,
+    server: str | None,
+    host: str | None,
+    source: str,
+) -> int:
+    """Connect once, read every configured tag once, print — the pre-deploy check."""
+    return _cmd_read(config_path, server, host, (), source, as_json=False)
+
+
 def main(argv: Sequence[str] | None = None) -> int:
    """CLI entry point (console script + ``python -m opcda_diagnozer``)."""
    args = _build_parser().parse_args(argv)
@@ -360,10 +557,16 @@ def main(argv: Sequence[str] | None = None) -> int:
        return _cmd_check_config(args.config)
    if args.verb == "list-servers":
        return _cmd_list_servers(args.host, args.source, args.json)
-    if args.verb == "list-devices":
-        return _cmd_list_devices(args.config, args.depth, not args.no_counts, args.json)
-    if args.verb == "list-tags":
-        return _cmd_list_tags(args.config, args.device, args.recursive, args.json)
+    if args.verb == "list-items":
+        return _cmd_list_items(
+            args.config, args.server, args.host, args.kind, args.branch, args.depth, args.json
+        )
+    if args.verb == "read":
+        return _cmd_read(
+            args.config, args.server, args.host, args.tags, args.source, args.json
+        )
+    if args.verb == "probe":
+        return _cmd_probe(args.config, args.server, args.host, args.source)
    phase = _NOT_IMPLEMENTED[args.verb]
    print(
        f"{args.verb}: not implemented yet — planned for {phase} "
@@ -17,28 +17,50 @@ Matrikon.OPC.Simulation.1): ``Connect(progId)`` / ``Connect(progId, node)``,
 ``CurrentPosition``; ``Organization`` 1 = hierarchical, 2 = flat. Bad branch
 names raise ``E_INVALIDARG``; ``MoveTo`` requires a *sequence* (a bare string
 is a pywin32 SAFEARRAY ``TypeError``).
+
+**Reads** (verified live): the wrapper's ``SyncRead`` cannot be marshalled by
+pywin32 (SAFEARRAY ByRef out-params fail in every invocation shape), but
+``OPCItem.Read(Source)`` + the ``Value`` / ``Quality`` / ``TimeStamp``
+properties work — per-item reads with only scalar out-params. Data-source
+constants come from the wrapper typelib: ``OPCCache = 1``, ``OPCDevice = 2``.
+``TimeStamp`` is a tz-aware ``pywintypes.datetime`` (converted with
+``astimezone(timezone.utc)`` for the UTC event model, SPECIFICATION §6).
 """

 from __future__ import annotations

 from collections.abc import Sequence
+from datetime import datetime, timezone
 from typing import Any

 from opcda_diagnozer.opc.base import (
    Browser,
+    OPCDataSource,
    OpcError,
    ReadResult,
    TagInfo,
    as_com_error,
 )

-_NOT_IMPL = "Phase 1: Automation Wrapper adapter (docs/IMPLEMENTATION_PLAN.md §4.4)"
-
 #: OPC_BROWSE organization constants (Automation Wrapper): 1 = hierarchical,
 #: 2 = flat (tags addressed by bare ItemID, no branches to walk).
 _ORGANIZATION_LABELS = {1: "hierarchical", 2: "flat"}


+def _per_tag_error(what: str, exc: BaseException) -> str:
+    """Render a per-tag failure without swallowing the underlying cause.
+
+    COM errors (``com_error`` with an ``hresult``) keep their decoded HRESULT
+    suffix via :func:`as_com_error`. **Plain Python exceptions** (e.g. a
+    ``ValueError`` from timezone math on a degenerate device timestamp) must
+    show their own type and text — early builds reduced them to a bare
+    "read of tag 'X' failed", which made failures undiagnosable.
+    """
+    if getattr(exc, "hresult", None) is None:
+        return f"{what}; {type(exc).__name__}: {exc}"
+    return str(as_com_error(what, exc))
+
+
 class _AutomationBrowser:
    """Thin, read-only :class:`Browser` over one wrapper browser object.

@@ -49,7 +71,10 @@ class _AutomationBrowser:
    """

    def __init__(self, raw_browser: Any) -> None:
-        self._browser = raw_browser
+        # Explicit ``Any`` declaration: close() assigns None to drop the COM
+        # reference, and the inferred ``Any | None`` would flag every access
+        # below (the dispatch object is dynamic — always accessed raw).
+        self._browser: Any = raw_browser

    @property
    def organization(self) -> str:
@@ -98,9 +123,10 @@ class _AutomationBrowser:
                if not name:
                    continue
                item_id = str(self._browser.GetItemID(name)).strip()
-                if not item_id:
-                    # Some servers return an empty ItemID for filtered leaves;
-                    # fall back to the browse name so config tags stay usable.
+                if not item_id or item_id == f".{name}":
+                    # Some servers return an empty ItemID for filtered/special
+                    # leaves (observed: Matrikon's alias-root items); fall back
+                    # to the browse name so config tags stay usable.
                    item_id = name
                leaves.append(TagInfo(name=name, item_id=item_id, path=tuple(path)))
            return leaves
@@ -120,9 +146,10 @@ class _AutomationBrowser:
 class AutomationOpcClient:
    """``win32com``-based :class:`OpcClient`.

-    ``connect`` / ``create_browser`` / ``close`` are live (they power the
-    ``list-devices`` / ``list-tags`` validation verbs); ``read`` lands with the
-    Phase 1 core engine (IMPLEMENTATION_PLAN §4.4).
+    ``connect`` / ``create_browser`` / ``read`` / ``close`` are live: they
+    power the ``list-devices`` / ``list-tags`` validation verbs and the
+    one-shot ``read`` / ``probe`` CLI verbs. The Phase 1 poller builds on this
+    same adapter (IMPLEMENTATION_PLAN §4.4).
    """

    def __init__(self, server: str, host: str | None = None, connect_timeout_s: float = 10.0) -> None:
@@ -214,9 +241,104 @@ class AutomationOpcClient:
            self._com_initialized = False
            pythoncom.CoUninitialize()

-    def read(self, tags: Sequence[str]) -> list[ReadResult]:
-        self._require_pywin32()
-        raise NotImplementedError(_NOT_IMPL)
+    def read(
+        self, tags: Sequence[str], source: OPCDataSource = OPCDataSource.DEVICE
+    ) -> list[ReadResult]:
+        """Read the given tags from the connected server.
+
+        Uses one throwaway group with one item per tag and per-item
+        ``Item.Read(Source)`` (see the module docstring: the wrapper's
+        ``SyncRead`` cannot be marshalled by pywin32). A tag the server does
+        not know yields a :class:`ReadResult` with ``error`` set and a
+        ``BAD_CONFIG_ERROR``-style quality of 0 — never an exception, matching
+        the per-tag error contract of :class:`ReadResult` (IMPLEMENTATION_PLAN
+        §4.3). The group (and its items) is removed afterwards so repeated
+        calls never leak groups on the server.
+        """
+        if self._opc is None:
+            raise OpcError("AutomationOpcClient: read() before connect()")
+        results: list[ReadResult] = []
+        group = None
+        items_collection = None
+        live_items: list[tuple[str, Any]] = []  # (tag, item) in request order
+        try:
+            try:
+                group = self._opc.OPCGroups.Add("DiagnozerRead")
+                items_collection = group.OPCItems
+                errors: list[str | None] = [None] * len(tags)
+                for index, tag in enumerate(tags):
+                    try:
+                        item = items_collection.AddItem(tag, index + 1)
+                        live_items.append((tag, item))
+                    except Exception as exc:  # noqa: BLE001 — per-tag error
+                        results.append(
+                            ReadResult(
+                                tag=tag,
+                                value=None,
+                                quality_raw=0,
+                                device_time=None,
+                                error=_per_tag_error(f"cannot add tag {tag!r}", exc),
+                            )
+                        )
+                for tag, item in live_items:
+                    try:
+                        item.Read(int(source))
+                        # Primary data first: value + quality are the read's
+                        # purpose. The device timestamp is METADATA and is
+                        # fetched last, fully guarded — a broken timestamp
+                        # (missing win32timezone in a frozen exe, zero FILETIME
+                        # 1601 on a never-updated alias, DST math on a
+                        # historical local-time wrapper) must never fail a read
+                        # whose value/quality are good.
+                        value = item.Value
+                        quality = int(item.Quality)
+                        device_time = None
+                        try:
+                            raw_ts = item.TimeStamp
+                            # pywintypes.datetime is a datetime subclass; normalize
+                            # to UTC for the event model (SPECIFICATION §6).
+                            if raw_ts is not None:
+                                device_time = raw_ts.astimezone(timezone.utc)
+                        except Exception:  # noqa: BLE001 — timestamp is metadata
+                            device_time = None
+                        results.append(
+                            ReadResult(
+                                tag=tag,
+                                value=value,
+                                quality_raw=quality,
+                                device_time=device_time,
+                            )
+                        )
+                    except Exception as exc:  # noqa: BLE001 — per-tag error
+                        results.append(
+                            ReadResult(
+                                tag=tag,
+                                value=None,
+                                quality_raw=0,
+                                device_time=None,
+                                error=_per_tag_error(f"read of tag {tag!r} failed", exc),
+                            )
+                        )
+            except Exception as exc:  # noqa: BLE001 — group-level failure
+                raise as_com_error(
+                    f"could not read from server {self._server!r} "
+                    "(group/level failure)",
+                    exc,
+                ) from None
+        finally:
+            # Release every COM reference INSIDE the apartment, in reverse
+            # creation order, before close()/CoUninitialize can see them.
+            live_items.clear()
+            items_collection = None
+            if group is not None:
+                try:
+                    group.Parent.OPCGroups.RemoveAll()
+                except Exception:  # noqa: BLE001 — best-effort cleanup
+                    pass
+            group = None
+        # preserve request order regardless of which tags failed to add
+        by_tag = {r.tag: r for r in results}
+        return [by_tag[tag] for tag in tags]

    def close(self) -> None:
        self._release_and_uninit()
@@ -5,9 +5,12 @@ are normalized to :class:`OpcError` carrying the raw HRESULT and a decoded name.
 There is deliberately **no write path** — the tool is read-only by construction
 (SPECIFICATION §10.5).

-Browsing (devices = branches, tags = leaves) is modelled by :class:`Browser`;
-the ``list-devices`` / ``list-tags`` CLI verbs walk the address space through
-it without any recursion into the monitored poll path.
+Browsing is modelled by :class:`Browser`. The OPC DA browse world defines
+exactly two element kinds: **branches** (what SCADA vendors call channels,
+devices, groups or folders — indistinguishable at the protocol level) and
+**leaves** (the tags). :class:`SpaceNode` is the tree node the ``list-items``
+CLI verb renders; leaves may hang directly at the server root, belonging to
+no branch at all (server-internal items like ``@ClientCount``).
 """

 from __future__ import annotations
@@ -15,7 +18,8 @@ from __future__ import annotations
 from collections.abc import Sequence
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Protocol, runtime_checkable
+from enum import IntEnum
+from typing import Any, Literal, Protocol, runtime_checkable


@dataclass(frozen=True)
@@ -29,6 +33,19 @@ class ReadResult:
    error: str | None = None


+class OPCDataSource(IntEnum):
+    """Automation Wrapper data-source constants (OPCDataSource enum).
+
+    ``CACHE`` reads the server's last-known value (fast, may be stale),
+    ``DEVICE`` forces a round-trip to the underlying device/controller — the
+    one the diagnozer cares about (SPECIFICATION §10: the measured path is
+    server → device, not server → cache).
+    """
+
+    CACHE = 1
+    DEVICE = 2
+
+
 # Common HRESULTs the tool decodes and reports (README "DCOM troubleshooting").
 _HRESULT_NAMES: dict[int, str] = {
    0x800706BA: "RPC_S_SERVER_UNAVAILABLE — server process down, firewall, or DCOM activation refused",
@@ -38,7 +55,8 @@ _HRESULT_NAMES: dict[int, str] = {
    0x80040154: "REGDB_E_CLASSNOTREG — wrong ProgID, or wrapper/server registered under a different bitness",
    0x80020009: "DISP_E_EXCEPTION — the server rejected the call (e.g. no such branch/tag)",
    0x80070057: "E_INVALIDARG — invalid argument (e.g. wrong branch or item name)",
-    0xC0040007: "OPC_E_INVALIDITEMID — tag name wrong in config (surfaced by probe)",
+    0xC0040007: "OPC_E_UNKNOWNITEMID — the server does not know this tag",
+    0xC0040008: "OPC_E_INVALIDITEMID — tag name wrong in config (surfaced by probe)",
 }


@@ -90,25 +108,25 @@ ACCESS_WRITE = 0x02


@dataclass(frozen=True)
-class DeviceInfo:
-    """One browsable device: a branch in the server's hierarchical address space.
+class SpaceNode:
+    """One node of the browsed address space: a branch or a leaf (tag).

-    The Automation Wrapper exposes no per-branch metadata beyond the name, so a
-    "device" here is simply the branch path plus the counts the CLI prints.
+    OPC DA's browse model has exactly these two element kinds — there is no
+    separate "device" or "folder" type: SCADA channels, devices, groups and
+    folders are all plain branches, indistinguishable at the protocol level.
+    A branch may hold leaves, sub-branches, or both; leaves may also hang
+    directly at the root, belonging to no branch at all.
    """

+    kind: Literal["branch", "leaf"]
    name: str
-    #: Branch path from the server root, e.g. ``("Simulation Items", "Bucket Brigade")``.
+    #: For a branch: its own path from the root, e.g. ``("Simulation Items",)``.
+    #: For a leaf: the branch it hangs under — ``()`` means directly at the root.
    path: tuple[str, ...]
-    #: Sub-branches directly under this device (usually channels/groups).
-    sub_devices: tuple[str, ...] = ()
-    #: Leaf (tag) names directly under this device.
-    tag_names: tuple[str, ...] = ()
-
-    @property
-    def qualified_name(self) -> str:
-        """Human-friendly dotted rendering: ``Simulation Items.Bucket Brigade``."""
-        return ".".join(self.path) if self.path else self.name
+    #: Fully qualified ItemID — leaves only; what goes into config ``tags:``.
+    item_id: str | None = None
+    #: Child nodes — branches only; empty when the depth limit stopped here.
+    children: tuple["SpaceNode", ...] = ()


@dataclass(frozen=True)
@@ -1,51 +1,45 @@
-"""Browse the connected server's address space: devices (branches) and tags (leaves).
+"""Browse the connected server's address space (IMPLEMENTATION_PLAN §6.2).

-OPC DA has no formal "device" concept — servers publish a *hierarchical
-address space* whose top-level branches almost always represent devices (or
-device families), with tags as the leaves. This module maps that tree onto the
-two questions ops asks (SPECIFICATION §8, validation tooling):
+OPC DA's browse model defines exactly two element kinds:

- *Which devices does this server expose?* → :func:`list_devices`
- *Which tags does device X expose?* → :func:`list_tags`
+- **branches** — the structure nodes. SCADA vendors map channels, devices,
+  groups and folders onto branches; the protocol cannot tell them apart, so
+  this tool reports them uniformly as branches.
+- **leaves** — the tags (items), each with a fully qualified ItemID that goes
+  straight into config ``tags:``.

-The Automation Wrapper browser (``opc.CreateBrowser()``) is the only portable
-traversal API — it works over DCOM for remote servers too, with the same
-Launch+Activation requirements as ``list-servers``. Traversal is one level at
-a time; the ``max_depth`` guard keeps a pathological (cyclic/AB-tree) server
-from stalling the CLI.
+Leaves may also hang directly at the server root, belonging to no branch —
+server-internal items (``@ClientCount`` on the Matrikon sim) look "wrong" in
+a device-oriented view but are ordinary root-level leaves.

-Flat-namespace servers report ``organization == "flat"``; there every leaf is
-addressed by its bare ItemID, the root is treated as one implicit device.
+The ``list-items`` CLI verb renders whatever the caller asks for
+(``all`` / ``branches`` / ``tags``) through :func:`walk_tree`, bounded by
+``--depth`` (1 = one level, :data:`WHOLE_SUBTREE` = everything) and by the
+configured ``browse.max_depth`` (default 32) against server-side cycles.
 """

 from __future__ import annotations

 from collections.abc import Sequence
-from dataclasses import dataclass

-from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo
+from opcda_diagnozer.opc.base import Browser, OpcError, SpaceNode, TagInfo

-#: Walk guard: deep SCADA hierarchies exist (Channel → Device → Tag), but OPC
-#: trees deeper than this are almost certainly a server-side loop.
-MAX_DEPTH = 32
+#: Default walk guard: deep SCADA hierarchies exist (Channel → Device → Tag),
+#: but OPC trees deeper than this are almost certainly a server-side loop.
+#: Overridable via the ``browse.max_depth`` config key.
+DEFAULT_MAX_DEPTH = 32
+
+#: Depth value meaning "no limit" (the whole subtree, max_depth-guarded).
+WHOLE_SUBTREE = 0


-@dataclass(frozen=True)
-class Space:
-    """Result of a device-space walk: one entry per discovered branch node."""
-
-    organization: str
-    root: DeviceInfo
-    devices: tuple[DeviceInfo, ...]
-
-
-def split_device_path(path: Sequence[str] | str | None) -> tuple[str, ...]:
-    """Normalize a user-supplied device path into branch segments.
+def split_branch_path(path: Sequence[str] | str | None) -> tuple[str, ...]:
+    """Normalize a user-supplied branch path into segments.

    A **single** string is treated as dotted: ``"Simulation Items.Bucket Brigade"``
    → ``("Simulation Items", "Bucket Brigade")``. **Multiple** segments are
    taken as explicit branch names — no dot splitting, so names that contain
-    a dot stay intact (they just have to be passed as separate arguments).
+    a dot stay intact (pass them as separate ``--branch`` arguments).
    Empty segments and whitespace are dropped.
    """
    if path is None:
@@ -61,134 +55,96 @@ def split_device_path(path: Sequence[str] | str | None) -> tuple[str, ...]:
    return tuple(parts)


-def list_devices(
-    browser: Browser,
-    *,
-    depth: int = 1,
-    include_tag_counts: bool = True,
-) -> Space:
-    """Enumerate devices (branches) starting at the server root.
-
-    ``depth=1`` lists top-level branches (the typical "device list"); greater
-    depths reveal sub-devices (channels/groups). Each :class:`DeviceInfo`
-    carries the direct sub-branches and — when ``include_tag_counts`` is on —
-    the leaf names under it, so the CLI can print ``Device (3 tags, 2 sub)``.
-
-    Raises :class:`OpcError` when the server rejects the walk (COM failure or
-    unsupported browse interface). The walk is bounded by ``depth`` (≤
-    :data:`MAX_DEPTH`), so a pathological server can never stall the CLI.
-    """
-    if depth < 1:
-        raise ValueError("depth must be >= 1")
-    if depth > MAX_DEPTH:
-        raise ValueError(f"depth must be <= {MAX_DEPTH}")
-    organization = browser.organization
-    root_branches = browser.list_branches(())
-    root_leaves = browser.list_leaves(()) if include_tag_counts else []
-    root = DeviceInfo(
-        name="(root)",
-        path=(),
-        sub_devices=tuple(root_branches),
-        tag_names=tuple(t.name for t in root_leaves),
-    )
-    devices: list[DeviceInfo] = []
-    _walk_level(browser, (), root_branches, depth, devices, include_tag_counts)
-    return Space(organization=organization, root=root, devices=tuple(devices))
-
-
-def _walk_level(
-    browser: Browser,
-    path: tuple[str, ...],
-    branches: Sequence[str],
-    depth: int,
-    devices: list[DeviceInfo],
-    include_tag_counts: bool,
-) -> None:
-    if depth <= 0:
-        return
-    for branch in branches:
-        sub_path = path + (branch,)
-        sub_branches = tuple(browser.list_branches(sub_path))
-        tag_names = (
-            tuple(t.name for t in browser.list_leaves(sub_path))
-            if include_tag_counts
-            else ()
-        )
-        devices.append(
-            DeviceInfo(
-                name=branch,
-                path=sub_path,
-                sub_devices=sub_branches,
-                tag_names=tag_names,
-            )
-        )
-        _walk_level(browser, sub_path, sub_branches, depth - 1, devices, include_tag_counts)
-
-
-def list_tags(
-    browser: Browser,
-    device: str | Sequence[str] | None = None,
-    *,
-    recursive: bool = False,
-) -> list[TagInfo]:
-    """List the tags of one device (branch), or of the whole server.
-
-    ``device`` accepts either a dotted string (``"Simulation Items.Bucket
-    Brigade"``) or a pre-split sequence of branch names. ``None`` means the
-    server root; with ``recursive=True`` every leaf under it is returned
-    (bounded by :data:`MAX_DEPTH`). The returned :class:`TagInfo.item_id`
-    values are exactly what goes into the config ``tags:`` list. An unknown
-    device path raises :class:`OpcError` pointing at ``list-devices``.
-    """
-    path = _require_device(browser, device, what="list-tags")
-    leaves = browser.list_leaves(path)
-    tags: list[TagInfo] = list(leaves)
-    if not recursive:
-        return sorted(tags, key=lambda t: t.item_id.lower())
-    sub_branches = browser.list_branches(path)
-    _collect_recursive(browser, path, sub_branches, tags)
-    return sorted(tags, key=lambda t: t.item_id.lower())
-
-
-def _collect_recursive(
-    browser: Browser,
-    path: tuple[str, ...],
-    branches: Sequence[str],
-    tags: list[TagInfo],
-) -> None:
-    for branch in branches:
-        sub_path = path + (branch,)
-        if len(sub_path) > MAX_DEPTH:
-            raise OpcError(
-                f"address space deeper than {MAX_DEPTH} levels at "
-                f"{'.'.join(sub_path)} — refusing to walk further (possible "
-                "server-side cycle)"
-            )
-        tags.extend(browser.list_leaves(sub_path))
-        _collect_recursive(browser, sub_path, browser.list_branches(sub_path), tags)
-
-
-def _require_device(
-    browser: Browser,
-    device: str | Sequence[str] | None,
-    *,
-    what: str,
+def require_branch(
+    browser: Browser, branch: Sequence[str] | str | None, *, what: str
 ) -> tuple[str, ...]:
-    """Resolve and validate a user-supplied device path (exists, is a branch)."""
-    path = split_device_path(device)
+    """Resolve and validate a user-supplied branch path (must exist)."""
+    path = split_branch_path(branch)
    if not path:
        return ()
    siblings = browser.list_branches(path[:-1])
    if path[-1] not in siblings:
        raise OpcError(
-            f"{what}: device {path[-1]!r} not found under "
-            f"{'.'.join(path[:-1]) or '(root)'} — run list-devices to see the "
-            "available branches"
+            f"{what}: branch {path[-1]!r} not found under "
+            f"{'.'.join(path[:-1]) or '(root)'} — list the levels above it "
+            "with list-items to see the available branch names"
        )
    return path


-def tags_for_device(browser: Browser, device: str | Sequence[str]) -> list[TagInfo]:
-    """List tags of a **known** device path — validation helper (probe)."""
-    path = _require_device(browser, device, what="list-tags")
-    return sorted(browser.list_leaves(path), key=lambda t: t.item_id.lower())
+def walk_tree(
+    browser: Browser,
+    branch: Sequence[str] | str | None = None,
+    *,
+    depth: int = 1,
+    max_depth: int = DEFAULT_MAX_DEPTH,
+) -> list[SpaceNode]:
+    """Return the nodes the server defines under ``branch``, keeping structure.
+
+    ``depth`` is the number of branch levels to descend: 1 (the default)
+    returns the direct leaves and branches under ``branch`` (branches come
+    back with empty ``children``); :data:`WHOLE_SUBTREE` walks the entire
+    subtree. ``max_depth`` is the hard guard against server-side cycles —
+    it defaults to :data:`DEFAULT_MAX_DEPTH` and comes from the
+    ``browse.max_depth`` config key.
+    """
+    if depth < 0:
+        raise ValueError(f"depth must be >= 0 (0 = whole subtree), got {depth}")
+    if max_depth < 1:
+        raise ValueError(f"max_depth must be >= 1, got {max_depth}")
+    if depth > max_depth:
+        raise ValueError(f"depth must be <= max_depth ({max_depth}), got {depth}")
+    root = require_branch(browser, branch, what="list-items")
+    return _walk_level(browser, root, depth, max_depth)
+
+
+def _walk_level(
+    browser: Browser, path: tuple[str, ...], depth: int, max_depth: int
+) -> list[SpaceNode]:
+    nodes: list[SpaceNode] = [
+        SpaceNode(kind="leaf", name=leaf.name, path=path, item_id=leaf.item_id)
+        for leaf in browser.list_leaves(path)
+    ]
+    for branch in browser.list_branches(path):
+        sub = path + (branch,)
+        if len(sub) > max_depth:
+            raise OpcError(
+                f"address space deeper than {max_depth} levels at {'.'.join(sub)} "
+                "— refusing to walk further (possible server-side cycle)"
+            )
+        if depth == 1:
+            children: tuple[SpaceNode, ...] = ()
+        else:
+            next_depth = WHOLE_SUBTREE if depth == WHOLE_SUBTREE else depth - 1
+            children = tuple(_walk_level(browser, sub, next_depth, max_depth))
+        nodes.append(SpaceNode(kind="branch", name=branch, path=sub, children=children))
+    return nodes
+
+
+def flatten_leaves(nodes: Sequence[SpaceNode]) -> list[TagInfo]:
+    """Every leaf in the walked tree as a flat, case-insensitively sorted list."""
+    tags: list[TagInfo] = []
+    _collect_leaves(nodes, tags)
+    return sorted(tags, key=lambda t: t.item_id.lower())
+
+
+def _collect_leaves(nodes: Sequence[SpaceNode], tags: list[TagInfo]) -> None:
+    for node in nodes:
+        if node.kind == "leaf" and node.item_id is not None:
+            tags.append(TagInfo(name=node.name, item_id=node.item_id, path=node.path))
+        else:
+            _collect_leaves(node.children, tags)
+
+
+def flatten_branches(nodes: Sequence[SpaceNode]) -> list[SpaceNode]:
+    """Every branch in the walked tree, depth-first, as a flat list."""
+    branches: list[SpaceNode] = []
+    _collect_branches(nodes, branches)
+    return branches
+
+
+def _collect_branches(nodes: Sequence[SpaceNode], out: list[SpaceNode]) -> None:
+    for node in nodes:
+        if node.kind == "branch":
+            out.append(node)
+            _collect_branches(node.children, out)
@@ -15,7 +15,7 @@ from __future__ import annotations
 from collections.abc import Sequence
 from datetime import datetime, timezone

-from opcda_diagnozer.opc.base import Browser, OpcError, ReadResult, TagInfo
+from opcda_diagnozer.opc.base import Browser, OPCDataSource, OpcError, ReadResult, TagInfo

 #: The fake address space: branch path -> (sub-branches, leaf names).
 #: Mirrors the shape of a real SCADA server (Channel → Device → Tags).
@@ -99,15 +99,19 @@ class FakeOpcClient:
    """In-process :class:`OpcClient`; every read returns GOOD-quality zeros.

    ``space`` is injectable so tests can exercise flat namespaces, empty
-    servers, and deep trees; the default mirrors a Matrikon-like hierarchy.
+    servers, and deep trees; ``read_results`` scripts the next read() result
+    (Phase 1 scenario engine seed, IMPLEMENTATION_PLAN §4.9).
    """

    def __init__(
        self,
        space: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] | None = None,
+        read_results: list[ReadResult] | None = None,
    ) -> None:
        self.connected = False
        self._space = FAKE_SPACE if space is None else space
+        self._scripted_reads = read_results
+        self.reads: list[tuple[tuple[str, ...], OPCDataSource]] = []  # call log

    def connect(self) -> None:
        self.connected = True
@@ -117,9 +121,27 @@ class FakeOpcClient:
            raise OpcError("FakeOpcClient: create_browser() before connect()")
        return _FakeBrowser(self._space)

-    def read(self, tags: Sequence[str]) -> list[ReadResult]:
+    def read(
+        self, tags: Sequence[str], source: OPCDataSource = OPCDataSource.DEVICE
+    ) -> list[ReadResult]:
        if not self.connected:
            raise OpcError("FakeOpcClient: read() before connect()")
+        self.reads.append((tuple(tags), source))
+        if self._scripted_reads is not None:
+            scripted = {r.tag: r for r in self._scripted_reads}
+            return [
+                scripted.get(
+                    tag,
+                    ReadResult(
+                        tag=tag,
+                        value=None,
+                        quality_raw=0,
+                        device_time=None,
+                        error=f"no scripted result for {tag!r}",
+                    ),
+                )
+                for tag in tags
+            ]
        now = datetime.now(timezone.utc)
        return [
            ReadResult(tag=tag, value=0, quality_raw=0xC0, device_time=now)
@@ -1,24 +1,28 @@
-"""Browse tests (opc/browse.py + fake browser + CLI verbs).
+"""Browse tests (opc/browse.py walk_tree + the list-items CLI verb).

 Pure tests run on any OS against the fake client's address space; the COM
-path is Windows-only and covered by the live smoke tests (IMPLEMENTATION_PLAN
+path is Windows-only and covered by the live smoke test (IMPLEMENTATION_PLAN
 §9: COM adapters are import-guarded).
 """

+import json
 import sys
+from pathlib import Path

 import pytest

 from opcda_diagnozer.main import main as cli_main
-from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo
+from opcda_diagnozer.opc.base import Browser, OpcError, SpaceNode
 from opcda_diagnozer.opc.browse import (
-    MAX_DEPTH,
-    list_devices,
-    list_tags,
-    split_device_path,
-    tags_for_device,
+    DEFAULT_MAX_DEPTH,
+    WHOLE_SUBTREE,
+    flatten_branches,
+    flatten_leaves,
+    require_branch,
+    split_branch_path,
+    walk_tree,
 )
-from opcda_diagnozer.opc.fake import FAKE_SPACE, FakeOpcClient
+from opcda_diagnozer.opc.fake import FakeOpcClient


@pytest.fixture
@@ -29,7 +33,7 @@ def browser() -> Browser:


 # --------------------------------------------------------------------------- #
-# split_device_path
+# split_branch_path
 # --------------------------------------------------------------------------- #


@@ -50,57 +54,99 @@ def browser() -> Browser:
        (("Bucket. Inc", "Device 1"), ("Bucket. Inc", "Device 1")),
    ],
 )
-def test_split_device_path(raw, expected: tuple[str, ...]) -> None:
-    assert split_device_path(raw) == expected
+def test_split_branch_path(raw, expected: tuple[str, ...]) -> None:
+    assert split_branch_path(raw) == expected


 # --------------------------------------------------------------------------- #
-# list_devices
+# walk_tree
 # --------------------------------------------------------------------------- #


-def test_list_devices_top_level(browser: Browser) -> None:
-    space = list_devices(browser)
-    assert space.organization == "hierarchical"
-    assert space.root.sub_devices == ("Simulation Items", "Configured Aliases")
-    names = [d.name for d in space.devices]
-    assert names == ["Simulation Items", "Configured Aliases"]
-    assert all(d.path == (d.name,) for d in space.devices)
+def test_walk_tree_depth_one(browser: Browser) -> None:
+    """Default depth: direct leaves + branches, branches have no children."""
+    nodes = walk_tree(browser)
+    assert [(n.kind, n.name) for n in nodes] == [
+        ("leaf", "#MonitorACLFile"),
+        ("leaf", "@ClientCount"),
+        ("leaf", "@Clients"),
+        ("branch", "Simulation Items"),
+        ("branch", "Configured Aliases"),
+    ]
+    # depth 1: the walk stops at the branches themselves
+    assert all(n.children == () for n in nodes if n.kind == "branch")


-def test_list_devices_depth_two_includes_sub_devices(browser: Browser) -> None:
-    space = list_devices(browser, depth=2)
-    paths = [d.path for d in space.devices]
-    assert ("Simulation Items", "Bucket Brigade") in paths
-    assert ("Simulation Items", "Random") in paths
-    # depth 2: the root's direct children plus their direct children
-    assert len(paths) == 2 + 8  # 2 top-level + 8 under Simulation Items
+def test_walk_tree_depth_one_includes_root_leaves(browser: Browser) -> None:
+    """Root-level leaves (belonging to no branch) are ordinary tree items."""
+    nodes = walk_tree(browser)
+    root_leaves = [n for n in nodes if n.kind == "leaf"]
+    assert {n.name for n in root_leaves} == {"@Clients", "@ClientCount", "#MonitorACLFile"}
+    assert all(n.path == () for n in root_leaves)  # they hang directly at the root


-def test_list_devices_reports_tag_counts(browser: Browser) -> None:
-    space = list_devices(browser)
-    by_name = {d.name: d for d in space.devices}
-    assert by_name["Simulation Items"].tag_names == ()  # no direct leaves
-    assert by_name["Configured Aliases"].tag_names == ("Alias1", "Alias2")
-    assert by_name["Simulation Items"].sub_devices[:4] == (
+def test_walk_tree_depth_two(browser: Browser) -> None:
+    nodes = walk_tree(browser, depth=2)
+    sim = next(n for n in nodes if n.name == "Simulation Items")
+    child_names = [c.name for c in sim.children]
+    assert child_names == [
        "Bucket Brigade", "Random", "Read Error", "Saw-toothed Waves",
-    )
+        "Square Waves", "Triangle Waves", "Write Error", "Write Only",
+    ]
+    # depth 2 stops inside those sub-branches too
+    bucket = next(c for c in sim.children if c.name == "Bucket Brigade")
+    assert bucket.kind == "branch" and bucket.children == ()
+    assert len(flatten_leaves(nodes)) == 5  # root leaves only; nothing deeper read


-def test_list_devices_no_counts_skips_leaves(browser: Browser) -> None:
-    space = list_devices(browser, include_tag_counts=False)
-    assert space.root.tag_names == ()
-    assert all(d.tag_names == () for d in space.devices)
+def test_walk_tree_whole_subtree(browser: Browser) -> None:
+    """depth=0: everything defined in the tree, in structure."""
+    nodes = walk_tree(browser, depth=WHOLE_SUBTREE)
+    tags = flatten_leaves(nodes)
+    item_ids = {t.item_id for t in tags}
+    assert "Bucket Brigade.Int1" in item_ids          # device tags
+    assert "Configured Aliases.Alias1" in item_ids     # alias tags
+    assert "@ClientCount" in item_ids                  # root-level, no branch
+    # the structure is preserved: Bucket Brigade is nested under Simulation Items
+    sim = next(n for n in nodes if n.name == "Simulation Items")
+    bucket = next(c for c in sim.children if c.name == "Bucket Brigade")
+    assert bucket.path == ("Simulation Items", "Bucket Brigade")
+    assert any(c.name == "Int1" and c.item_id == "Bucket Brigade.Int1" for c in bucket.children)


-@pytest.mark.parametrize("depth", [0, -1, MAX_DEPTH + 1])
-def test_list_devices_rejects_bad_depth(browser: Browser, depth: int) -> None:
-    with pytest.raises(ValueError):
-        list_devices(browser, depth=depth)
+def test_walk_tree_from_subtree(browser: Browser) -> None:
+    """--branch limits the walk to that subtree."""
+    nodes = walk_tree(browser, branch="Simulation Items.Bucket Brigade", depth=WHOLE_SUBTREE)
+    # only the subtree: no root leaves, no other branches
+    kinds = [(n.kind, n.name) for n in nodes]
+    assert all(n.kind == "leaf" for n in nodes)
+    assert {n.name for n in nodes} == {
+        "Int1", "Int2", "Int4", "Real4", "Real8", "String", "Boolean",
+    }


-def test_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None:
-    """A pathological self-nesting branch must not hang the recursive walk."""
+def test_walk_tree_missing_branch_is_actionable(browser: Browser) -> None:
+    with pytest.raises(OpcError, match="list-items"):
+        walk_tree(browser, branch="Simulation Items.No Such Device")
+    with pytest.raises(OpcError, match="not found under"):
+        walk_tree(browser, branch="Nope")
+
+
+@pytest.mark.parametrize("depth", [-1, -5])
+def test_walk_tree_rejects_negative_depth(browser: Browser, depth: int) -> None:
+    with pytest.raises(ValueError, match=">= 0"):
+        walk_tree(browser, depth=depth)
+
+
+def test_walk_tree_rejects_bad_max_depth(browser: Browser) -> None:
+    with pytest.raises(ValueError, match="max_depth"):
+        walk_tree(browser, max_depth=0)
+    with pytest.raises(ValueError, match="max_depth"):
+        walk_tree(browser, depth=33, max_depth=32)
+
+
+def test_walk_tree_configurable_max_depth_guards_cycles() -> None:
+    """A pathological self-nesting branch must hit the max_depth guard, not hang."""

    class _LoopBrowser:
        @property
@@ -119,53 +165,38 @@ def test_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None:
        def close(self) -> None:
            return None

-    with pytest.raises(OpcError, match="deeper than"):
-        list_tags(_LoopBrowser(), recursive=True)  # type: ignore[arg-type]
+    with pytest.raises(OpcError, match="deeper than 2"):
+        walk_tree(_LoopBrowser(), depth=WHOLE_SUBTREE, max_depth=2)  # type: ignore[arg-type]
+
+
+def test_walk_tree_default_max_depth_constant() -> None:
+    """The default guard matches the documented config default (32)."""
+    assert DEFAULT_MAX_DEPTH == 32


 # --------------------------------------------------------------------------- #
-# list_tags
+# flatten helpers
 # --------------------------------------------------------------------------- #


-def test_list_tags_root_direct(browser: Browser) -> None:
-    tags = list_tags(browser)
-    assert [t.name for t in tags] == ["#MonitorACLFile", "@ClientCount", "@Clients"]
-    assert all(t.item_id == t.name for t in tags)  # root leaves: bare ItemIDs
-    assert all(t.path == () for t in tags)
+def test_flatten_leaves_sorted_case_insensitively(browser: Browser) -> None:
+    tags = flatten_leaves(walk_tree(browser, depth=WHOLE_SUBTREE))
+    ids = [t.item_id for t in tags]
+    assert ids == sorted(ids, key=str.lower)


-def test_list_tags_of_device_dotted_string(browser: Browser) -> None:
-    tags = list_tags(browser, device="Simulation Items.Bucket Brigade")
-    item_ids = [t.item_id for t in tags]
-    assert "Bucket Brigade.Int1" in item_ids
-    assert "Bucket Brigade.Real8" in item_ids
-    assert len(tags) == 7
+def test_flatten_branches_depth_first(browser: Browser) -> None:
+    branches = flatten_branches(walk_tree(browser, depth=WHOLE_SUBTREE))
+    paths = [".".join(b.path) for b in branches]
+    # depth-first: parents before their children
+    assert paths.index("Simulation Items") < paths.index("Simulation Items.Bucket Brigade")
+    assert paths.index("Simulation Items") < paths.index("Configured Aliases")


-def test_list_tags_of_device_segments(browser: Browser) -> None:
-    tags = list_tags(browser, device=["Simulation Items", "Random"])
-    assert {t.name for t in tags} == {"Int1", "Int4", "Money", "Time"}
-    assert all(t.path == ("Simulation Items", "Random") for t in tags)
-
-
-def test_list_tags_recursive(browser: Browser) -> None:
-    tags = list_tags(browser, device="Simulation Items", recursive=True)
-    item_ids = {t.item_id for t in tags}
-    # direct children of "Simulation Items" have no leaves; recursion gathers all
-    assert "Bucket Brigade.Int1" in item_ids
-    assert "Random.Money" in item_ids
-    assert "Read Error.Int4" in item_ids
-    assert "Saw-toothed Waves.Real8" in item_ids
-    assert all(t.path for t in tags)
-
-
-
-def test_tags_are_sorted_case_insensitively(browser: Browser) -> None:
-    tags = list_tags(browser, device="Simulation Items.Bucket Brigade")
-    assert [t.item_id for t in tags] == sorted(
-        (t.item_id for t in tags), key=lambda s: s.lower()
-    )
+def test_spacenode_is_frozen() -> None:
+    node = SpaceNode(kind="leaf", name="X", path=(), item_id="X")
+    with pytest.raises(AttributeError):
+        node.name = "Y"  # type: ignore[misc]


 # --------------------------------------------------------------------------- #
@@ -174,7 +205,6 @@ def test_tags_are_sorted_case_insensitively(browser: Browser) -> None:


 def test_fake_browser_matches_protocol(browser: Browser) -> None:
-    # runtime_checkable protocol: structural check
    assert isinstance(browser, Browser)


@@ -199,17 +229,24 @@ def test_fake_flat_space_reports_flat_organization() -> None:
    client.connect()
    browser = client.create_browser()
    assert browser.organization == "flat"
-    space = list_devices(browser)
-    assert space.devices == ()
-    assert space.root.tag_names == ("Tag1", "Tag2")
+    nodes = walk_tree(browser)
+    assert all(n.kind == "leaf" for n in nodes)
+    assert {n.name for n in nodes} == {"Tag1", "Tag2"}
+
+
+def test_require_branch_root_ok_and_missing(browser: Browser) -> None:
+    assert require_branch(browser, None, what="list-items") == ()
+    assert require_branch(browser, "Simulation Items", what="list-items") == ("Simulation Items",)
+    with pytest.raises(OpcError, match="list-items"):
+        require_branch(browser, "Missing", what="list-items")


 # --------------------------------------------------------------------------- #
-# CLI verbs (against the fake backend through a config file)
+# CLI verb (against the fake backend through a config file)
 # --------------------------------------------------------------------------- #


-def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int:
+def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, args: list[str]) -> int:
    """Run the CLI with the fake client injected in place of the COM adapter."""
    import opcda_diagnozer.main as main_mod
    from opcda_diagnozer.opc.fake import FakeOpcClient
@@ -221,7 +258,7 @@ def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int:
    )
    client = FakeOpcClient()

-    def _fake_connect(config_path: str):
+    def _fake_connect(config_path: str, server=None, host=None):
        client.connect()
        return client

@@ -229,69 +266,117 @@ def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int:
    return cli_main(args)


-def test_cli_list_devices_table(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None:
-    code = _run_cli(monkeypatch, tmp_path, ["list-devices", "-c", str(tmp_path / "config.yaml")])
-    out = capsys.readouterr().out
-    assert code == 0
-    assert "Simulation Items" in out
-    assert "Configured Aliases" in out
-    assert "2 tags" in out
-
-
-def test_cli_list_devices_json(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None:
+def test_cli_list_items_all_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
    code = _run_cli(monkeypatch, tmp_path, [
-        "list-devices", "-c", str(tmp_path / "config.yaml"), "--json",
+        "list-items", "all", "--depth", "0", "-c", str(tmp_path / "config.yaml"),
    ])
    out = capsys.readouterr().out
    assert code == 0
-    import json
-
-    lines = [json.loads(line) for line in out.splitlines() if line.strip()]
-    assert {d["name"] for d in lines} == {"Simulation Items", "Configured Aliases"}
-    assert lines[0]["qualified_name"] == "Simulation Items"  # walk order = root order
+    # tree structure: root leaves visible, branches with trailing /
+    assert "@ClientCount  [ItemID: @ClientCount]" in out
+    assert "Simulation Items/" in out
+    assert "  Bucket Brigade/" in out  # nested one level
+    assert "    Int1  [ItemID: Bucket Brigade.Int1]" in out  # nested two levels


-def test_cli_list_tags_dotted(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None:
+def test_cli_list_items_default_is_all_depth1(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    code = _run_cli(monkeypatch, tmp_path, ["list-items", "-c", str(tmp_path / "config.yaml")])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Simulation Items/" in out
+    assert "Bucket Brigade" not in out  # depth 1 stops at the top branches
+
+
+def test_cli_list_items_branches(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
    code = _run_cli(monkeypatch, tmp_path, [
-        "list-tags", "-c", str(tmp_path / "config.yaml"),
-        "Simulation Items.Bucket Brigade",
+        "list-items", "branches", "--depth", "0", "-c", str(tmp_path / "config.yaml"),
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Simulation Items" in out
+    assert "Simulation Items.Bucket Brigade" in out
+    assert "8 sub-branches" in out
+    assert "[ItemID" not in out  # no tags in the branches view
+
+
+def test_cli_list_items_tags_flat(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-items", "tags", "--depth", "0", "-c", str(tmp_path / "config.yaml"),
    ])
    out = capsys.readouterr().out
    assert code == 0
    assert "Bucket Brigade.Int1" in out
-    assert "Bucket Brigade.Real8" in out
+    assert "Configured Aliases.Alias1" in out
+    assert "@ClientCount" in out  # root-level tags are tags too


-def test_cli_list_tags_segments_and_recursive(
-    monkeypatch: pytest.MonkeyPatch, tmp_path, capsys
+def test_cli_list_items_branch_subtree(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
 ) -> None:
    code = _run_cli(monkeypatch, tmp_path, [
-        "list-tags", "-c", str(tmp_path / "config.yaml"),
-        "Simulation Items", "--recursive", "--json",
+        "list-items", "tags", "--branch", "Simulation Items.Bucket Brigade",
+        "-c", str(tmp_path / "config.yaml"),
    ])
    out = capsys.readouterr().out
    assert code == 0
-    import json
-
-    item_ids = {json.loads(line)["item_id"] for line in out.splitlines() if line.strip()}
-    assert "Bucket Brigade.Int1" in item_ids
-    assert "Random.Money" in item_ids
+    assert "Bucket Brigade.Int1" in out
+    assert "Random.Money" not in out  # only the selected subtree
+    assert "Configured Aliases" not in out


-def test_cli_list_tags_missing_device_fails_cleanly(
-    monkeypatch: pytest.MonkeyPatch, tmp_path, capsys
+def test_cli_list_items_branch_repeated_segments(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    """Repeating --branch gives explicit segments (names may contain dots)."""
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-items", "tags", "--branch", "Simulation Items", "--branch", "Bucket Brigade",
+        "-c", str(tmp_path / "config.yaml"),
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Bucket Brigade.Int1" in out
+
+
+def test_cli_list_items_json_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code = _run_cli(monkeypatch, tmp_path, [
+        "list-items", "--json", "--depth", "0", "-c", str(tmp_path / "config.yaml"),
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    lines = [json.loads(line) for line in out.splitlines() if line.strip()]
+    assert lines[0]["kind"] == "leaf"
+    sim = next(l for l in lines if l["name"] == "Simulation Items")
+    assert sim["kind"] == "branch"
+    assert any(c["name"] == "Bucket Brigade" for c in sim["children"])
+
+
+def test_cli_list_items_missing_branch_fails_cleanly(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
 ) -> None:
    code = _run_cli(monkeypatch, tmp_path, [
-        "list-tags", "-c", str(tmp_path / "config.yaml"), "No.Such.Device",
+        "list-items", "--branch", "No.Such.Branch", "-c", str(tmp_path / "config.yaml"),
    ])
    captured = capsys.readouterr()
    assert code == 2
    assert "FAILED" in captured.err


+def test_cli_old_verbs_are_gone(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    """list-devices / list-tags were replaced by list-items."""
+    for gone in ("list-devices", "list-tags"):
+        with pytest.raises(SystemExit) as exc_info:
+            _run_cli(monkeypatch, tmp_path, [gone, "-c", str(tmp_path / "config.yaml")])
+        assert exc_info.value.code == 2  # argparse: unknown verb
+        assert "invalid choice" in capsys.readouterr().err
+
+
@pytest.mark.skipif(sys.platform != "win32", reason="COM browse needs Windows + pywin32")
 def test_live_automation_browse_smoke() -> None:
-    """Live smoke test — requires the Matrikon simulation server on this dev box."""
+    """Live smoke test — requires a registered Automation Wrapper (dev box)."""
    pytest.importorskip("pythoncom")
    from opcda_diagnozer.opc.automation import AutomationOpcClient

@@ -299,11 +384,13 @@ def test_live_automation_browse_smoke() -> None:
    try:
        client.connect()
        browser = client.create_browser()
-        devices = list_devices(browser)
-        assert devices.organization == "hierarchical"
-        assert any(d.name == "Simulation Items" for d in devices.devices)
-        tags = list_tags(browser, device="Simulation Items.Bucket Brigade")
+        nodes = walk_tree(browser, depth=WHOLE_SUBTREE)
+        tags = flatten_leaves(nodes)
        assert any(t.item_id == "Bucket Brigade.Int1" for t in tags)
+        assert any(t.name == "@ClientCount" for t in tags)  # root-level leaf
+        sim = next(n for n in nodes if n.name == "Simulation Items")
+        assert sim.kind == "branch"
+        assert any(c.name == "Bucket Brigade" for c in sim.children)
    except OpcError as exc:
        pytest.skip(f"simulation server not usable on this machine: {exc}")
    finally:
@@ -0,0 +1,76 @@
+"""Tests for the frozen-exe build script + frozen-aware config default.
+
+The PyInstaller run itself is Windows-only and slow; these tests cover the
+pure logic the build script and the CLI rely on. The build script is
+PowerShell (``scripts/build_exe.ps1``); the tests assert on its content and
+on the CLI behaviors the produced exe depends on.
+"""
+
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+from opcda_diagnozer.main import DEFAULT_CONFIG, _default_config_path
+
+BUILD_SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "build_exe.ps1"
+
+
+def test_default_config_path_in_source_mode() -> None:
+    """Not frozen: the default stays the plain relative name (CWD lookup)."""
+    with mock.patch.object(sys, "frozen", False, create=True):
+        assert _default_config_path() == DEFAULT_CONFIG
+
+
+def test_default_config_path_when_frozen_resolves_next_to_exe(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Frozen: config.yaml lives next to the exe — never the service CWD (System32)."""
+    exe_dir = tmp_path / "Program Files" / "OPCDA-Diagnozer"
+    exe_dir.mkdir(parents=True)
+    exe = exe_dir / "opcda-diagnozer.exe"
+    exe.touch()
+    monkeypatch.setattr(sys, "frozen", True, raising=False)
+    monkeypatch.setattr(sys, "executable", str(exe))
+    assert _default_config_path() == str(exe_dir / DEFAULT_CONFIG)
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="the exe is Windows-only")
+class TestBuildScript:
+    """Static assertions on scripts/build_exe.ps1 (no PyInstaller invocation)."""
+
+    def test_hidden_imports_cover_lazy_modules(self) -> None:
+        """Every lazy import in the package must be declared to PyInstaller."""
+        content = BUILD_SCRIPT.read_text(encoding="utf-8")
+        # the deferred COM imports (opc/automation.py, opc/discovery.py)
+        assert "'pythoncom'" in content
+        assert "'win32com.client'" in content
+        assert "'opcda_diagnozer.opc.automation'" in content
+        # the lazily-imported browse module (main.py verbs)
+        assert "'opcda_diagnozer.opc.browse'" in content
+        # the Phase 3 service host stays in the bundle from day one
+        assert "'opcda_diagnozer.service'" in content
+
+    def test_version_info_template_uses_pyi_keywords(self) -> None:
+        """The embedded version-info must use PyInstaller's exact keyword names.
+
+        FixedFileInfo.__init__ takes filevers/prodvers/mask/flags/OS/fileType/
+        subtype/date — anything else (fileVers, fileDate, subType) raises a
+        TypeError inside PyInstaller's eval at build time.
+        """
+        content = BUILD_SCRIPT.read_text(encoding="utf-8")
+        assert "filevers=(" in content
+        assert "fileType=0x1" in content
+        assert "subtype=0x0" in content
+        assert "date=(0, 0)" in content
+        # the rejected spellings must NOT appear
+        assert "fileVers=" not in content
+        assert "fileDate=" not in content
+        assert "subType=" not in content
+
+    def test_bitness_guards_present(self) -> None:
+        """The script must refuse a 64-bit interpreter and verify the PE."""
+        content = BUILD_SCRIPT.read_text(encoding="utf-8")
+        assert "struct.calcsize('P') * 8" in content  # interpreter bitness check
+        assert "0x14C" in content  # PE machine = IMAGE_FILE_MACHINE_I386 check
@@ -18,11 +18,18 @@ def _write(tmp_path: Path, text: str) -> Path:
 def test_example_config_loads() -> None:
    config = load_config(EXAMPLE)
    assert config.opc.server == "Matrikon.OPC.Simulation.1"
-    assert config.opc.host == "CONTROL-HOST01"
+    assert config.opc.host is None  # host commented out -> local server
    assert len(config.tags) == 2
    assert config.tags[0].name == "Channel1.Device1.Counter"
    assert config.tags[0].expect_change is True
    assert config.tags[1].expect_change is False
+    # D16 device-link indicators (spec 6.1); normal reading configurable per device
+    assert [d.name for d in config.devices] == ["Conveyor 3", "Conveyor 4", "Pump A", "Boiler 2"]
+    assert config.devices[0].tag == "Channel1.Device3.Connected"
+    assert config.devices[0].ok_values == ("true",)  # default: boolean connected-flag
+    assert config.devices[1].ok_values == ("true", "1")
+    assert config.devices[2].ok_values == ("false", "0")  # inverted fault flag: 0 = ok
+    assert config.devices[3].ok_values == ("RUN",)  # string state
    assert config.poll.interval_s == 1.0
    assert config.poll.timeout_ms == 5000
    assert config.latency.warn_ms == 1000
@@ -65,6 +72,95 @@ def test_duplicate_tags_rejected(tmp_path: Path) -> None:
        load_config(p)


+def test_devices_missing_keys_rejected(tmp_path: Path) -> None:
+    p = _write(tmp_path, "opc:\n  server: X\ntags:\n  - name: t\ndevices:\n  - name: Device\n")
+    with pytest.raises(ConfigError, match="'tag' is required"):
+        load_config(p)
+
+
+def test_devices_blank_name_rejected(tmp_path: Path) -> None:
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\ndevices:\n  - { name: '', tag: Ind.Tag }\n",
+    )
+    with pytest.raises(ConfigError, match="non-empty"):
+        load_config(p)
+
+
+def test_devices_duplicate_indicator_rejected(tmp_path: Path) -> None:
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\n"
+        "devices:\n  - { name: A, tag: Ind.Tag }\n  - { name: B, tag: Ind.Tag }\n",
+    )
+    with pytest.raises(ConfigError, match="one device per indicator"):
+        load_config(p)
+
+
+def test_devices_indicator_in_tags_rejected(tmp_path: Path) -> None:
+    """An indicator is polled automatically; listing it in tags: double-reports."""
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\n"
+        "tags:\n  - name: Ind.Tag\ndevices:\n  - { name: A, tag: Ind.Tag }\n",
+    )
+    with pytest.raises(ConfigError, match="remove them from"):
+        load_config(p)
+
+
+def test_devices_empty_is_allowed(tmp_path: Path) -> None:
+    """devices: is optional (no link monitoring) and may be an empty list."""
+    p = _write(tmp_path, "opc:\n  server: X\ntags:\n  - name: t\ndevices: []\n")
+    assert load_config(p).devices == ()
+
+
+def test_devices_ok_values_default_is_true(tmp_path: Path) -> None:
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\ndevices:\n  - { name: A, tag: I.Tag }\n",
+    )
+    device = load_config(p).devices[0]
+    assert device.ok_values == ("true",)
+
+
+def test_devices_ok_values_normalized(tmp_path: Path) -> None:
+    """Booleans/ints/strings all render to comparison strings; scalar = shorthand."""
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\n"
+        "devices:\n"
+        "  - { name: A, tag: I.One, ok_values: [true, 1] }\n"
+        "  - { name: B, tag: I.Two, ok_values: [false, 0] }\n"
+        "  - { name: C, tag: I.Three, ok_values: [\"RUN\"] }\n"
+        "  - { name: D, tag: I.Four, ok_values: 1 }\n",
+    )
+    devices = {d.name: d for d in load_config(p).devices}
+    assert devices["A"].ok_values == ("true", "1")
+    assert devices["B"].ok_values == ("false", "0")
+    assert devices["C"].ok_values == ("RUN",)
+    assert devices["D"].ok_values == ("1",)  # single scalar shorthand
+
+
+def test_devices_ok_values_empty_rejected(tmp_path: Path) -> None:
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\n"
+        "devices:\n  - { name: A, tag: I.Tag, ok_values: [] }\n",
+    )
+    with pytest.raises(ConfigError, match="must not be empty"):
+        load_config(p)
+
+
+def test_devices_ok_values_non_scalar_rejected(tmp_path: Path) -> None:
+    p = _write(
+        tmp_path,
+        "opc:\n  server: X\ntags:\n  - name: t\n"
+        "devices:\n  - { name: A, tag: I.Tag, ok_values: [[1]] }\n",
+    )
+    with pytest.raises(ConfigError, match="scalars"):
+        load_config(p)
+
+
 def test_latency_threshold_order_rejected(tmp_path: Path) -> None:
    p = _write(
        tmp_path,
@@ -0,0 +1,390 @@
+"""Tests for the read/probe CLI verbs and the --server/--host overrides.
+
+All fake-client driven — no COM needed (IMPLEMENTATION_PLAN §9). The read path
+itself (OPCItem.Read) is verified live on a dev box with the Matrikon
+simulation server.
+"""
+
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+
+from opcda_diagnozer.main import main as cli_main
+from opcda_diagnozer.opc.base import OPCDataSource, OpcError, ReadResult
+from opcda_diagnozer.opc.fake import FakeOpcClient
+
+CONFIG = "opc:\n  server: Fake.Server.1\ntags:\n  - name: Bucket Brigade.Int1\n  - name: Bucket Brigade.Real8\n"
+
+
+def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, args: list[str], client: FakeOpcClient | None = None) -> tuple[int, FakeOpcClient]:
+    """Run the CLI with the fake client injected in place of the COM adapter."""
+    import opcda_diagnozer.main as main_mod
+
+    config = tmp_path / "config.yaml"
+    config.write_text(CONFIG, encoding="utf-8")
+    fake = client if client is not None else FakeOpcClient()
+
+    def _fake_connect(config_path: str, server=None, host=None):
+        assert isinstance(server, str | None)
+        fake.connect()
+        return fake
+
+    monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
+    return cli_main(args), fake
+
+
+def test_read_explicit_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code, fake = _run_cli(monkeypatch, tmp_path, [
+        "read", "-c", str(tmp_path / "config.yaml"), "Bucket Brigade.Int1", "Bucket Brigade.Real8",
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Bucket Brigade.Int1" in out
+    assert "0xC0" in out  # GOOD quality shown decoded
+    # request order preserved, source = DEVICE by default
+    assert fake.reads == [
+        (("Bucket Brigade.Int1", "Bucket Brigade.Real8"), OPCDataSource.DEVICE),
+    ]
+
+
+def test_read_from_config_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code, fake = _run_cli(monkeypatch, tmp_path, [
+        "read", "-c", str(tmp_path / "config.yaml"),
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert "Bucket Brigade.Int1" in out and "Bucket Brigade.Real8" in out
+    assert fake.reads[0][0] == ("Bucket Brigade.Int1", "Bucket Brigade.Real8")
+
+
+def test_read_cache_source_flag(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    _, fake = _run_cli(monkeypatch, tmp_path, [
+        "read", "-c", str(tmp_path / "config.yaml"), "--source", "cache", "Bucket Brigade.Int1",
+    ])
+    assert fake.reads[0][1] == OPCDataSource.CACHE
+
+
+def test_read_json_output(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code, _ = _run_cli(monkeypatch, tmp_path, [
+        "read", "-c", str(tmp_path / "config.yaml"), "--json", "Bucket Brigade.Int1",
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    payload = json.loads(out.strip().splitlines()[0])
+    assert payload["tag"] == "Bucket Brigade.Int1"
+    assert payload["quality"] == "GOOD"
+    assert payload["quality_class"] == "GOOD"
+    assert payload["device_time_utc"]
+
+
+def test_read_bad_quality_tag_fails_with_nonzero_exit(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    bad = ReadResult(
+        tag="Read Error.Int1",
+        value=None,
+        quality_raw=0x1C,  # BAD_OUT_OF_SERVICE (0x18 would be COMM_FAILURE)
+        device_time=datetime(2026, 10, 7, 9, 0, 0, tzinfo=timezone.utc),
+    )
+    code, _ = _run_cli(
+        monkeypatch, tmp_path,
+        ["read", "-c", str(tmp_path / "config.yaml"), "Read Error.Int1"],
+        client=FakeOpcClient(read_results=[bad]),
+    )
+    captured = capsys.readouterr()
+    assert code == 1  # non-GOOD quality -> exit 1
+    assert "BAD_OUT_OF_SERVICE" in captured.out
+    assert "did not read GOOD" in captured.err
+
+
+def test_read_error_tag_reports_error_not_crash(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    missing = ReadResult(
+        tag="No.Such.Tag",
+        value=None,
+        quality_raw=0,
+        device_time=None,
+        error="cannot add tag 'No.Such.Tag' (HRESULT 0xC0040008: ...)",
+    )
+    code, _ = _run_cli(
+        monkeypatch, tmp_path,
+        ["read", "-c", str(tmp_path / "config.yaml"), "No.Such.Tag"],
+        client=FakeOpcClient(read_results=[missing]),
+    )
+    captured = capsys.readouterr()
+    assert code == 1
+    assert "ERROR" in captured.out
+    assert "0xC0040008" in captured.out
+
+
+# --------------------------------------------------------------------------- #
+# per-tag error rendering (_per_tag_error) — no COM needed
+# --------------------------------------------------------------------------- #
+
+
+def _fake_com_error(hresult: int) -> Exception:
+    """A stand-in with the pywin32 com_error shape for the hresult check."""
+    err = Exception("server rejected the call")
+    err.hresult = hresult  # type: ignore[attr-defined]
+    return err
+
+
+def test_per_tag_error_keeps_hresult_for_com_errors() -> None:
+    from opcda_diagnozer.opc.automation import _per_tag_error
+
+    rendered = _per_tag_error("cannot add tag 'X'", _fake_com_error(0x80020009))
+    assert "cannot add tag 'X'" in rendered
+    assert "HRESULT 0x80020009" in rendered  # decoded COM suffix preserved
+
+
+def test_per_tag_error_names_plain_exceptions() -> None:
+    """A plain ValueError must show its type+text, not a bare 'failed'."""
+    from opcda_diagnozer.opc.automation import _per_tag_error
+
+    rendered = _per_tag_error("read of tag 'X' failed", ValueError("tz math broke"))
+    assert "read of tag 'X' failed" in rendered
+    assert "ValueError" in rendered
+    assert "tz math broke" in rendered
+    assert "HRESULT" not in rendered  # never fake a COM suffix for non-COM errors
+
+
+def test_timestamp_failure_never_fails_a_read() -> None:
+    """A degenerate device timestamp is metadata: value/quality stay readable.
+
+    Reproduces the alias-tag failure shape end-to-end through the adapter's
+    read() with a stubbed COM item: a TimeStamp whose astimezone() explodes
+    (zero FILETIME 1601 / DST math on Matrikon aliases) must yield a GOOD
+    result with device_time=None — never a per-tag error.
+    """
+    from datetime import timezone as _tz
+
+    from opcda_diagnozer.opc import automation
+    from opcda_diagnozer.opc.base import OPCDataSource
+
+    class _PoisonedItem:
+        """A COM item stand-in: reads succeed, TimeStamp cannot be normalized."""
+
+        ItemID = "P5002.Connected"
+
+        def __init__(self) -> None:
+            self.read_calls: list[int] = []
+
+        def Read(self, source: int) -> None:
+            self.read_calls.append(source)
+
+        @property
+        def Value(self):
+            return 1
+
+        @property
+        def Quality(self):
+            return 0xC0
+
+        @property
+        def TimeStamp(self):
+            class _Bad:
+                def astimezone(self, tz):
+                    raise ValueError("astimezone() cannot be applied to a naive datetime")
+
+            return _Bad()
+
+    class _Items:
+        def AddItem(self, item_id: str, handle: int):
+            assert item_id == "P5002.Connected"
+            return _PoisonedItem()
+
+    class _Group:
+        OPCItems = _Items()
+
+        class _Parent:
+            OPCGroups = None  # set below
+
+        Parent = _Parent()
+
+    class _Groups:
+        def Add(self, name: str):
+            return _Group()
+
+    class _Opc:  # minimal stand-in for the wrapper's dispatch object
+        OPCGroups = _Groups()
+
+    class _ClientStub(automation.AutomationOpcClient):
+        def __init__(self) -> None:
+            super().__init__(server="stub")
+            self._opc = _Opc()  # read() requires a connected-looking client
+
+        def _release_and_uninit(self) -> None:  # no COM to release on the stub
+            self._opc = None
+
+    client = _ClientStub()
+    results = client.read(["P5002.Connected"], source=OPCDataSource.DEVICE)
+    assert len(results) == 1
+    result = results[0]
+    assert result.error is None  # the poison never surfaced
+    assert result.value == 1
+    assert result.quality_raw == 0xC0
+    assert result.device_time is None  # metadata dropped, not the read
+
+
+def test_timestamp_property_explosion_never_fails_a_read() -> None:
+    """The frozen-exe failure shape: the TimeStamp PROPERTY itself explodes.
+
+    Missing win32timezone made pywin32 raise SystemError('InvokeTypes ...
+    exception set') when *accessing* item.TimeStamp (not in astimezone) — the
+    guard must therefore wrap the property fetch too, and the read must still
+    return the value/quality.
+    """
+    from opcda_diagnozer.opc import automation
+    from opcda_diagnozer.opc.base import OPCDataSource
+
+    class _ExplodingTsItem:
+        ItemID = "Bucket Brigade.Int1"
+
+        def Read(self, source: int) -> None:
+            return None
+
+        @property
+        def Value(self):
+            return 42
+
+        @property
+        def Quality(self):
+            return 0xC0
+
+        @property
+        def TimeStamp(self):
+            # the exact frozen-exe signature: a non-COM SystemError raised by
+            # pywin32's conversion inside the property access
+            raise SystemError(
+                "<built-in method InvokeTypes of PyIDispatch object> "
+                "returned a result with an exception set"
+            )
+
+    class _Items:
+        def AddItem(self, item_id: str, handle: int):
+            return _ExplodingTsItem()
+
+    class _Group:
+        OPCItems = _Items()
+
+        class _Parent:
+            OPCGroups = None
+
+        Parent = _Parent()
+
+    class _Groups:
+        def Add(self, name: str):
+            return _Group()
+
+    class _Opc:
+        OPCGroups = _Groups()
+
+    class _ClientStub(automation.AutomationOpcClient):
+        def __init__(self) -> None:
+            super().__init__(server="stub")
+            self._opc = _Opc()
+
+        def _release_and_uninit(self) -> None:
+            self._opc = None
+
+    client = _ClientStub()
+    results = client.read(["Bucket Brigade.Int1"], source=OPCDataSource.DEVICE)
+    assert len(results) == 1
+    result = results[0]
+    assert result.error is None  # the timestamp explosion never surfaced
+    assert result.value == 42
+    assert result.quality_raw == 0xC0
+    assert result.device_time is None
+
+
+def test_probe_reads_config_tags(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys) -> None:
+    code, fake = _run_cli(monkeypatch, tmp_path, [
+        "probe", "-c", str(tmp_path / "config.yaml"),
+    ])
+    out = capsys.readouterr().out
+    assert code == 0
+    assert fake.reads[0][0] == ("Bucket Brigade.Int1", "Bucket Brigade.Real8")
+    assert "Bucket Brigade.Int1" in out
+
+
+def test_probe_missing_config_is_actionable(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    # no monkeypatching: the real _connect_from_config must produce the hint
+    code = cli_main(["probe", "-c", str(tmp_path / "no-such-config.yaml")])
+    captured = capsys.readouterr()
+    assert code == 2
+    assert "FAILED" in captured.err
+    assert "--server" in captured.err  # the hint points at the override
+
+
+def test_server_override_flows_to_client(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    """--server/--host must reach the client even when a config exists."""
+    import opcda_diagnozer.main as main_mod
+
+    config = tmp_path / "config.yaml"
+    config.write_text(CONFIG, encoding="utf-8")
+    seen: dict[str, object] = {}
+
+    def _fake_connect(config_path: str, server=None, host=None):
+        seen["server"], seen["host"] = server, host
+        client = FakeOpcClient()
+        client.connect()
+        return client
+
+    monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect)
+    code = cli_main([
+        "read", "-c", str(config),
+        "--server", "Other.Server.1", "--host", "OTHER-HOST",
+        "Some.Tag",
+    ])
+    assert code == 0
+    assert seen == {"server": "Other.Server.1", "host": "OTHER-HOST"}
+
+
+def test_read_without_config_and_without_server_is_actionable(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys
+) -> None:
+    # no monkeypatching: the real connect path must fail with the hint
+    code = cli_main([
+        "read", "-c", str(tmp_path / "no-such.yaml"), "Some.Tag",
+    ])
+    captured = capsys.readouterr()
+    assert code == 2
+    assert "--server" in captured.err
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="live read needs Windows + pywin32")
+def test_live_automation_read_smoke() -> None:
+    """Live smoke test — requires the Matrikon simulation server on this dev box."""
+    pytest.importorskip("pythoncom")
+    from opcda_diagnozer.opc.automation import AutomationOpcClient
+
+    client = AutomationOpcClient(server="Matrikon.OPC.Simulation.1")
+    try:
+        client.connect()
+        results = client.read(
+            ["Bucket Brigade.Int1", "Read Error.Int1", "No.Such.Tag"],
+            source=OPCDataSource.DEVICE,
+        )
+        by_tag = {r.tag: r for r in results}
+        good = by_tag["Bucket Brigade.Int1"]
+        assert good.quality_raw == 0xC0
+        assert good.device_time is not None
+        # unknown tag -> per-tag error, never an exception
+        assert by_tag["No.Such.Tag"].error is not None
+        # "Read Error.Int1" from DEVICE returns the error code as the value
+        # with GOOD quality (verified live 2026-10-07); the BAD quality only
+        # appears on the CACHE path — the diagnozer's quality monitoring
+        # therefore reads CACHE for that tag in the poll loop (Phase 1).
+        assert by_tag["Read Error.Int1"].value is not None
+    except OpcError as exc:
+        pytest.skip(f"simulation server not usable on this machine: {exc}")
+    finally:
+        client.close()
Author	SHA1	Message	Date
Jonas Bengtsson	a1af8d68c9	🚧 WIP	2026-10-08 06:30:54 +02:00
Jonas Bengtsson	76b1cc5ccf	🚧 WIP	2026-10-07 17:57:30 +02:00
Jonas Bengtsson	2aa040e757	🚧 WIP	2026-10-07 17:13:51 +02:00