🚧 WIP

.gitignore

@@ -14,3 +14,6 @@ dist/
 # Local runtime artifacts (copy config.example.yaml -> config.yaml)
 config.yaml
 logs/
+
+# Generated by scripts/build_exe.ps1 at build time
+scripts/.exe_version_info.txt

.vscode/launch.json

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

README.md

@@ -33,8 +33,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 +51,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 +87,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
@@ -119,6 +140,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,6 +212,47 @@ 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

config.example.yaml

@@ -31,6 +31,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:

src/opcda_diagnozer/config.py

@@ -66,6 +66,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"
@@ -96,10 +103,11 @@ class Config:
     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", "latency", "quality", "stale", "browse", "report"}
 _TAG_KEYS = {"name", "expect_change"}
 
 
@@ -207,6 +215,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"):
@@ -246,6 +256,7 @@ def load_config(path: str | Path) -> Config:
         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:

src/opcda_diagnozer/main.py

@@ -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)")
@@ -211,61 +267,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 +368,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 +553,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} "

src/opcda_diagnozer/opc/automation.py

@@ -17,23 +17,31 @@ 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"}
@@ -98,9 +106,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 +129,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 +224,92 @@ 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=str(as_com_error(f"cannot add tag {tag!r}", exc)),
+                            )
+                        )
+                for tag, item in live_items:
+                    try:
+                        item.Read(int(source))
+                        raw_ts = item.TimeStamp
+                        # pywintypes.datetime is a datetime subclass; normalize
+                        # to UTC for the event model (SPECIFICATION §6).
+                        device_time = (
+                            raw_ts.astimezone(timezone.utc) if raw_ts is not None else None
+                        )
+                        results.append(
+                            ReadResult(
+                                tag=tag,
+                                value=item.Value,
+                                quality_raw=int(item.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=str(as_com_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()

src/opcda_diagnozer/opc/base.py

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

src/opcda_diagnozer/opc/browse.py

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

src/opcda_diagnozer/opc/fake.py

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

tests/test_browse.py

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

tests/test_build_exe.py

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

tests/test_config.py

@@ -18,7 +18,7 @@ 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

tests/test_read_cli.py

@@ -0,0 +1,210 @@
+"""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
+
+
+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()