diff --git a/influxdata/library/plugin_library.json b/influxdata/library/plugin_library.json
index bb8986b..a076235 100644
--- a/influxdata/library/plugin_library.json
+++ b/influxdata/library/plugin_library.json
@@ -358,8 +358,8 @@
                 "author": "InfluxData",
                 "docs_file_link": "https://github.com/influxdata/influxdb3_plugins/blob/main/influxdata/valuecounter/README.md",
                 "required_plugins": [],
-                "required_libraries": [],
-                "last_update": "2026-06-10",
+                "required_libraries": ["influxdata-plugin-utils>=0.3.0"],
+                "last_update": "2026-08-03",
                 "trigger_types_supported": ["scheduler", "data_writes"]
             },
             {
diff --git a/influxdata/valuecounter/README.md b/influxdata/valuecounter/README.md
index bdfbbfc..ae36508 100644
--- a/influxdata/valuecounter/README.md
+++ b/influxdata/valuecounter/README.md
@@ -25,11 +25,12 @@ This plugin includes a JSON metadata schema in its docstring that defines suppor
 
 ### Data write trigger parameters
 
-| Parameter        | Type    | Default          | Description                                                                                              |
-|------------------|---------|------------------|----------------------------------------------------------------------------------------------------------|
-| `period_seconds` | integer | `60`             | Emission period in seconds. Cache TTL is set to `2 * period_seconds`. Accepts `period=60s` form via TOML |
-| `output_suffix`  | string  | `_valuecounts`   | Suffix appended to the source measurement name for rollup output. Must be non-empty                      |
-| `dest_database`  | string  | trigger's own DB | Optional database to write rollups to via `write_sync_to_db`                                             |
+| Parameter        | Type    | Default          | Description                                                                                                                      |
+|------------------|---------|------------------|----------------------------------------------------------------------------------------------------------------------------------|
+| `period_seconds` | integer | `60`             | Emission period in seconds, at least `1`. Cache TTL is set to `2 * period_seconds`                                               |
+| `period`         | string  | `60s`            | Emission period as a duration, at least `1s`. Units: `s`, `min`, `h`, `d`, `w`. Overridden by `period_seconds` when both are set |
+| `output_suffix`  | string  | `_valuecounts`   | Suffix appended to the source measurement name for rollup output. Must be non-empty                                              |
+| `dest_database`  | string  | trigger's own DB | Optional database to write rollups to via `write_sync_to_db`                                                                     |
 
 ### Scheduled trigger parameters
 
@@ -46,9 +47,9 @@ Mode B is drift-based: the trigger's `every:<duration>` spec is the only cadence
 |--------------------|--------|---------|----------------------------------------------------------------------------------|
 | `config_file_path` | string | none    | TOML config file path relative to `PLUGIN_DIR` (required for TOML configuration) |
 
-*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3.
+*To use a TOML configuration file, set the `PLUGIN_DIR` environment variable and specify the `config_file_path` in the trigger arguments.* This is in addition to the `--plugin-dir` flag when starting InfluxDB 3. Relative paths are resolved against `PLUGIN_DIR`, then `INFLUXDB3_PLUGIN_DIR`, then the parent of `VIRTUAL_ENV`.
 
-If `config_file_path` is set, no other inline arguments may be set on the trigger. The TOML accepts the same keys as inline arguments; Mode A may use `period = "60s"` in place of `period_seconds = 60`.
+If `config_file_path` is set, no other inline arguments may be set on the trigger. The TOML accepts the same keys as inline arguments.
 
 #### Example TOML configuration
 
@@ -65,7 +66,7 @@ The rollup measurement name is always `<source_table><output_suffix>`. The defau
 ## Software Requirements
 
 - **InfluxDB 3 Core/Enterprise**: with the Processing Engine enabled
-- **Python packages**: none (standard library only)
+- **Python packages**: `influxdata-plugin-utils>=0.3.0`
 
 ## Installation steps
 
@@ -79,7 +80,11 @@ The rollup measurement name is always `<source_table><output_suffix>`. The defau
      --plugin-dir ~/.plugins
    ```
 
-2. No additional Python packages required for this plugin.
+2. Install required Python packages:
+
+   ```bash
+   influxdb3 install package "influxdata-plugin-utils>=0.3.0"
+   ```
 
 ## Trigger setup
 
@@ -207,8 +212,8 @@ The rollup measurement `payment_attempts_valuecounts` is written to the `analyti
 
 - `valuecounter.py`: The main plugin code containing `process_writes` (Mode A) and `process_scheduled_call` (Mode B)
 - `valuecounter_config.toml`: Example TOML configuration with both Mode A and Mode B shapes
-- `test_valuecounter.py`: Pytest suite (82 tests, runs without a live InfluxDB 3 server)
-- `requirements.txt`: Runtime dependencies (empty for this plugin)
+- `test_valuecounter.py`: Pytest suite (79 tests, runs without a live InfluxDB 3 server)
+- `requirements.txt`: Runtime dependencies (`influxdata-plugin-utils>=0.3.0`)
 - `requirements-dev.txt`: Development dependencies (`pytest`)
 
 ### Logging
@@ -251,7 +256,7 @@ Handles Mode B scheduled invocations. On the first fire, only the cadence anchor
 
 #### Issue: Source table tag list changed and rollups are missing tags
 
-**Solution**: The plugin caches tag-column names from `information_schema.columns` for one hour per source table. Schema changes are picked up within an hour. To force a refresh, delete the cache key `vc:tags:<table>`.
+**Solution**: The plugin caches tag-column names from `information_schema.columns` for one hour per source table. Schema changes are picked up within an hour. To force a refresh, delete the cache key `shared:tags:<table>`.
 
 #### Issue: Both modes installed against the same source table
 
diff --git a/influxdata/valuecounter/manifest.toml b/influxdata/valuecounter/manifest.toml
index b2b5e25..d4ee07f 100644
--- a/influxdata/valuecounter/manifest.toml
+++ b/influxdata/valuecounter/manifest.toml
@@ -2,7 +2,7 @@ manifest_schema_version = "1.2"
 
 [plugin]
 name = "valuecounter"
-version = "0.2.0"
+version = "0.3.0"
 description = "Counts unique field values from data-write or scheduled inputs and writes rollup measurements with per-value counts."
 triggers = ["process_writes", "process_scheduled_call"]
 homepage = "https://www.influxdata.com/"
@@ -16,4 +16,4 @@ exclude = [
 
 [dependencies]
 database_version = ">=3.8.2"
-python = []
+python = ["influxdata-plugin-utils>=0.3.0"]
diff --git a/influxdata/valuecounter/requirements-dev.txt b/influxdata/valuecounter/requirements-dev.txt
index e079f8a..2a50709 100644
--- a/influxdata/valuecounter/requirements-dev.txt
+++ b/influxdata/valuecounter/requirements-dev.txt
@@ -1 +1,2 @@
 pytest
+influxdata-plugin-utils>=0.3.0
diff --git a/influxdata/valuecounter/requirements.txt b/influxdata/valuecounter/requirements.txt
index e69de29..46a2877 100644
--- a/influxdata/valuecounter/requirements.txt
+++ b/influxdata/valuecounter/requirements.txt
@@ -0,0 +1 @@
+influxdata-plugin-utils>=0.3.0
diff --git a/influxdata/valuecounter/test_valuecounter.py b/influxdata/valuecounter/test_valuecounter.py
index 7354d4d..5b50851 100644
--- a/influxdata/valuecounter/test_valuecounter.py
+++ b/influxdata/valuecounter/test_valuecounter.py
@@ -6,21 +6,26 @@
 from collections import OrderedDict
 from typing import Optional
 
+
 class InfluxDBError(Exception):
     pass
 
+
 class InvalidMeasurementError(InfluxDBError):
     pass
 
+
 class InvalidKeyError(InfluxDBError):
     pass
 
+
 class InvalidLineError(InfluxDBError):
     pass
 
+
 class LineBuilder:
     def __init__(self, measurement: str):
-        if ' ' in measurement:
+        if " " in measurement:
             raise InvalidMeasurementError("Measurement name cannot contain spaces")
         self.measurement = measurement
         self.tags: OrderedDict[str, str] = OrderedDict()
@@ -31,32 +36,42 @@ def _validate_key(self, key: str, key_type: str) -> None:
         """Validate that a key does not contain spaces, commas, or equals signs."""
         if not key:
             raise InvalidKeyError(f"{key_type} key cannot be empty")
-        if ' ' in key:
+        if " " in key:
             raise InvalidKeyError(f"{key_type} key '{key}' cannot contain spaces")
-        if ',' in key:
+        if "," in key:
             raise InvalidKeyError(f"{key_type} key '{key}' cannot contain commas")
-        if '=' in key:
+        if "=" in key:
             raise InvalidKeyError(f"{key_type} key '{key}' cannot contain equals signs")
 
     def _escape_measurement(self, value: str) -> str:
         """Escape characters in measurement names according to line protocol."""
-        return value.replace(',', '\\,').replace(' ', '\\ ')
+        return value.replace(",", "\\,").replace(" ", "\\ ")
 
     def _escape_tag_value(self, value: str) -> str:
         """Escape characters in tag values according to line protocol."""
-        return value.replace('\\', '\\\\').replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ')
+        return (
+            value.replace("\\", "\\\\")
+            .replace(",", "\\,")
+            .replace("=", "\\=")
+            .replace(" ", "\\ ")
+        )
 
     def _escape_field_key(self, value: str) -> str:
         """Escape characters in field keys according to line protocol."""
-        return value.replace('\\', '\\\\').replace(',', '\\,').replace('=', '\\=').replace(' ', '\\ ')
+        return (
+            value.replace("\\", "\\\\")
+            .replace(",", "\\,")
+            .replace("=", "\\=")
+            .replace(" ", "\\ ")
+        )
 
-    def tag(self, key: str, value: str) -> 'LineBuilder':
+    def tag(self, key: str, value: str) -> "LineBuilder":
         """Add a tag to the line protocol."""
         self._validate_key(key, "tag")
         self.tags[key] = str(value)
         return self
 
-    def uint64_field(self, key: str, value: int) -> 'LineBuilder':
+    def uint64_field(self, key: str, value: int) -> "LineBuilder":
         """Add an unsigned integer field to the line protocol."""
         self._validate_key(key, "field")
         if value < 0:
@@ -64,34 +79,34 @@ def uint64_field(self, key: str, value: int) -> 'LineBuilder':
         self.fields[key] = f"{value}u"
         return self
 
-    def int64_field(self, key: str, value: int) -> 'LineBuilder':
+    def int64_field(self, key: str, value: int) -> "LineBuilder":
         """Add an integer field to the line protocol."""
         self._validate_key(key, "field")
         self.fields[key] = f"{value}i"
         return self
 
-    def float64_field(self, key: str, value: float) -> 'LineBuilder':
+    def float64_field(self, key: str, value: float) -> "LineBuilder":
         """Add a float field to the line protocol."""
         self._validate_key(key, "field")
         # Check if value has no decimal component
         self.fields[key] = f"{int(value)}.0" if value % 1 == 0 else str(value)
         return self
 
-    def string_field(self, key: str, value: str) -> 'LineBuilder':
+    def string_field(self, key: str, value: str) -> "LineBuilder":
         """Add a string field to the line protocol."""
         self._validate_key(key, "field")
         # Escape quotes and backslashes in string values
-        escaped_value = value.replace('\\', '\\\\').replace('"', '\\"')
+        escaped_value = value.replace("\\", "\\\\").replace('"', '\\"')
         self.fields[key] = f'"{escaped_value}"'
         return self
 
-    def bool_field(self, key: str, value: bool) -> 'LineBuilder':
+    def bool_field(self, key: str, value: bool) -> "LineBuilder":
         """Add a boolean field to the line protocol."""
         self._validate_key(key, "field")
-        self.fields[key] = 't' if value else 'f'
+        self.fields[key] = "t" if value else "f"
         return self
 
-    def time_ns(self, timestamp_ns: int) -> 'LineBuilder':
+    def time_ns(self, timestamp_ns: int) -> "LineBuilder":
         """Set the timestamp in nanoseconds."""
         self._timestamp_ns = timestamp_ns
         return self
@@ -103,7 +118,7 @@ def build(self) -> str:
 
         # Add tags if present
         if self.tags:
-            tags_str = ','.join(
+            tags_str = ",".join(
                 f"{key}={self._escape_tag_value(value)}"
                 for key, value in self.tags.items()
             )
@@ -113,7 +128,7 @@ def build(self) -> str:
         if not self.fields:
             raise InvalidLineError(f"At least one field is required: {line}")
 
-        fields_str = ','.join(
+        fields_str = ",".join(
             f"{self._escape_field_key(key)}={value}"
             for key, value in self.fields.items()
         )
@@ -130,11 +145,14 @@ class FakeCache:
     def __init__(self):
         self._d = {}
         self._ttls = {}
+
     def get(self, key, default=None, use_global=None):
         return self._d.get(key, default)
+
     def put(self, key, value, ttl=None, use_global=None):
         self._d[key] = value
         self._ttls[key] = ttl
+
     def delete(self, key, use_global=None):
         return self._d.pop(key, None) is not None
 
@@ -148,17 +166,26 @@ def __init__(self):
         self.fail_write = False
         self.query_responses = {}
         self.queries_run = []
-    def info(self, m):  self.logs.append(("info", m))
-    def warn(self, m):  self.logs.append(("warn", m))
-    def error(self, m): self.logs.append(("error", m))
+
+    def info(self, m):
+        self.logs.append(("info", m))
+
+    def warn(self, m):
+        self.logs.append(("warn", m))
+
+    def error(self, m):
+        self.logs.append(("error", m))
+
     def write_sync(self, line, no_sync=False):
         if self.fail_write:
             raise RuntimeError("simulated write failure")
         self.writes.append(line.build())
+
     def write_sync_to_db(self, db_name, line, no_sync=False):
         if self.fail_write:
             raise RuntimeError("simulated write failure")
         self.cross_db_writes.append((db_name, line.build()))
+
     def query(self, sql, params=None):
         self.queries_run.append((sql, params))
         for substr, rows in self.query_responses.items():
@@ -179,23 +206,30 @@ def test_vendored_line_builder_basic():
 def test_stringify_int():
     assert _stringify_value(200) == "200"
 
+
 def test_stringify_float():
     assert _stringify_value(1.5) == "1.5"
 
+
 def test_stringify_str():
     assert _stringify_value("OK") == "OK"
 
+
 def test_stringify_bool_true():
-    assert _stringify_value(True) == "true"   # telegraf-style lowercase
+    assert _stringify_value(True) == "true"  # telegraf-style lowercase
+
 
 def test_stringify_bool_false():
     assert _stringify_value(False) == "false"
 
+
 def test_stringify_none():
-    assert _stringify_value(None) is None     # signals "skip"
+    assert _stringify_value(None) is None  # signals "skip"
+
 
 def test_stringify_unknown_falls_back_to_repr():
     class Weird: ...
+
     out = _stringify_value(Weird())
     assert isinstance(out, str)
     assert "Weird" in out
@@ -207,14 +241,20 @@ class Weird: ...
 def test_sanitize_no_change_for_plain():
     assert _sanitize_field_name("status_200") == "status_200"
 
+
 def test_sanitize_spaces():
     assert _sanitize_field_name("not found") == "not_found"
 
+
 def test_sanitize_hyphen():
     assert _sanitize_field_name("User-Agent") == "User_Agent"
 
+
 def test_sanitize_punctuation():
-    assert _sanitize_field_name("500 Internal Server Error") == "500_Internal_Server_Error"
+    assert (
+        _sanitize_field_name("500 Internal Server Error") == "500_Internal_Server_Error"
+    )
+
 
 def test_sanitize_unicode_replaced():
     assert _sanitize_field_name("café") == "caf_"
@@ -227,211 +267,245 @@ def test_validate_accepts_plain():
     _validate_identifier("http_requests", "table")
     _validate_identifier("status", "field")
 
+
 def test_validate_accepts_leading_underscore():
     _validate_identifier("_internal", "table")
 
+
 def test_validate_rejects_leading_digit():
     with pytest.raises(ValueError):
         _validate_identifier("1status", "field")
 
+
 def test_validate_rejects_space():
     with pytest.raises(ValueError):
         _validate_identifier("status code", "field")
 
+
 def test_validate_rejects_dash():
     with pytest.raises(ValueError):
         _validate_identifier("user-agent", "field")
 
+
 def test_validate_rejects_semicolon():
     with pytest.raises(ValueError):
         _validate_identifier("status;DROP TABLE users", "table")
 
+
 def test_validate_rejects_quote():
     with pytest.raises(ValueError):
         _validate_identifier('status"', "field")
 
+
 def test_validate_rejects_too_long():
     with pytest.raises(ValueError):
         _validate_identifier("a" * 129, "table")
 
+
 def test_validate_message_includes_what_and_value():
     with pytest.raises(ValueError, match="invalid table:.*'bad name'"):
         _validate_identifier("bad name", "table")
 
 
-from valuecounter import _parse_duration
+from valuecounter import _parse_period_seconds
 
 
 def test_parse_seconds():
-    assert _parse_duration("60s") == 60
+    assert _parse_period_seconds("60s") == 60
+
+
+def test_parse_legacy_bare_minutes():
+    assert _parse_period_seconds("5m") == 300
+
 
 def test_parse_minutes():
-    assert _parse_duration("5m") == 300
+    assert _parse_period_seconds("5min") == 300
+
 
 def test_parse_hours():
-    assert _parse_duration("1h") == 3600
+    assert _parse_period_seconds("1h") == 3600
+
 
 def test_parse_days():
-    assert _parse_duration("1d") == 86400
+    assert _parse_period_seconds("1d") == 86400
+
 
 def test_parse_with_whitespace():
-    assert _parse_duration("  30 s  ") == 30
+    assert _parse_period_seconds("  30 s  ") == 30
+
 
 def test_parse_rejects_missing_unit():
-    with pytest.raises(ValueError, match="invalid duration"):
-        _parse_duration("60")
+    with pytest.raises(ValueError, match="Invalid duration"):
+        _parse_period_seconds("60")
+
+
+def test_parse_rejects_sub_second():
+    with pytest.raises(ValueError, match="at least 1 second"):
+        _parse_period_seconds("60ms")
 
-def test_parse_rejects_unknown_unit():
-    with pytest.raises(ValueError, match="invalid duration"):
-        _parse_duration("60ms")
 
 def test_parse_rejects_empty():
     with pytest.raises(ValueError):
-        _parse_duration("")
+        _parse_period_seconds("")
+
+
+from valuecounter import _resolve_config
 
 
-from valuecounter import Config, _parse_inline_args
+def _write_tmp_toml(tmp_path, content):
+    path = tmp_path / "vc.toml"
+    path.write_text(content)
+    return path
 
 
-def test_inline_args_mode_a_minimal():
-    cfg = _parse_inline_args({"fields": "status method"}, mode="wal")
+def test_resolve_inline_mode_a_minimal():
+    cfg = _resolve_config({"fields": "status method"}, mode="wal")
     assert cfg.fields == ["status", "method"]
     assert cfg.output_suffix == "_valuecounts"
     assert cfg.dest_database == ""
     assert cfg.period_seconds == 60
     assert cfg.table == ""
 
-def test_inline_args_mode_b_minimal():
-    cfg = _parse_inline_args({"table": "http_requests", "fields": "status"}, mode="scheduled")
+
+def test_resolve_inline_mode_b_minimal():
+    cfg = _resolve_config(
+        {"table": "http_requests", "fields": "status"}, mode="scheduled"
+    )
     assert cfg.table == "http_requests"
     assert cfg.fields == ["status"]
 
-def test_inline_args_mode_a_full():
-    cfg = _parse_inline_args(
-        {"fields": "status method", "output_suffix": "_vc",
-         "period_seconds": "30", "dest_database": "rollups"},
+
+def test_resolve_inline_mode_a_full():
+    cfg = _resolve_config(
+        {
+            "fields": "status method",
+            "output_suffix": "_vc",
+            "period_seconds": "30",
+            "dest_database": "rollups",
+        },
         mode="wal",
     )
     assert cfg.period_seconds == 30
     assert cfg.output_suffix == "_vc"
     assert cfg.dest_database == "rollups"
 
-def test_inline_args_rejects_unknown_key():
+
+def test_resolve_period_seconds_overrides_period():
+    cfg = _resolve_config(
+        {"fields": "a", "period": "5m", "period_seconds": "30"}, mode="wal"
+    )
+    assert cfg.period_seconds == 30
+
+
+def test_resolve_rejects_unknown_arg():
     with pytest.raises(ValueError, match="unknown arg"):
-        _parse_inline_args({"fields": "a", "bogus": "x"}, mode="wal")
+        _resolve_config({"fields": "a", "bogus": "x"}, mode="wal")
 
-def test_inline_args_mode_a_rejects_table():
-    with pytest.raises(ValueError, match="table"):
-        _parse_inline_args({"fields": "a", "table": "http_requests"}, mode="wal")
 
-def test_inline_args_mode_b_rejects_period_seconds():
-    with pytest.raises(ValueError, match="period"):
-        _parse_inline_args(
+def test_resolve_mode_a_rejects_table():
+    with pytest.raises(ValueError, match="trigger-spec.*inline args"):
+        _resolve_config({"fields": "a", "table": "http_requests"}, mode="wal")
+
+
+def test_resolve_mode_b_rejects_period_seconds():
+    with pytest.raises(ValueError, match="drift-based"):
+        _resolve_config(
             {"table": "t", "fields": "a", "period_seconds": "60"}, mode="scheduled"
         )
 
-def test_inline_args_mode_b_rejects_period():
-    with pytest.raises(ValueError, match="period"):
-        _parse_inline_args(
+
+def test_resolve_mode_b_rejects_period():
+    with pytest.raises(ValueError, match="drift-based"):
+        _resolve_config(
             {"table": "t", "fields": "a", "period": "60s"}, mode="scheduled"
         )
 
 
-import os
-import tempfile
-from valuecounter import _load_toml_config
-
-
-def _write_tmp_toml(content):
-    f = tempfile.NamedTemporaryFile(suffix=".toml", delete=False, mode="w")
-    f.write(content)
-    f.close()
-    return f.name
+def test_resolve_rejects_unknown_mode():
+    with pytest.raises(ValueError, match="unknown mode"):
+        _resolve_config({"fields": "a"}, mode="bogus")
 
 
-def test_toml_mode_a_minimal():
-    p = _write_tmp_toml('fields = ["status", "method"]\nperiod = "30s"\n')
-    cfg = _load_toml_config(p, mode="wal")
+def test_resolve_toml_mode_a(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
+    _write_tmp_toml(tmp_path, 'fields = ["status", "method"]\nperiod = "30s"\n')
+    cfg = _resolve_config({"config_file_path": "vc.toml"}, mode="wal")
     assert cfg.fields == ["status", "method"]
     assert cfg.period_seconds == 30
-    os.unlink(p)
 
 
-def test_toml_mode_b_minimal():
-    p = _write_tmp_toml('table = "http_requests"\nfields = ["status"]\n')
-    cfg = _load_toml_config(p, mode="scheduled")
+def test_resolve_toml_mode_b(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
+    _write_tmp_toml(tmp_path, 'table = "http_requests"\nfields = ["status"]\n')
+    cfg = _resolve_config({"config_file_path": "vc.toml"}, mode="scheduled")
     assert cfg.table == "http_requests"
     assert cfg.fields == ["status"]
-    os.unlink(p)
 
 
-def test_toml_rejects_unknown_key():
-    p = _write_tmp_toml('fields = ["a"]\nbogus = 1\n')
-    with pytest.raises(ValueError, match="unknown"):
-        _load_toml_config(p, mode="wal")
-    os.unlink(p)
+def test_resolve_toml_rejects_unknown_key(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
+    _write_tmp_toml(tmp_path, 'fields = ["a"]\nbogus = 1\n')
+    with pytest.raises(ValueError, match="unknown TOML key"):
+        _resolve_config({"config_file_path": "vc.toml"}, mode="wal")
 
 
-def test_toml_parse_error():
-    p = _write_tmp_toml('this is not valid toml = =')
+def test_resolve_toml_parse_error(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
+    _write_tmp_toml(tmp_path, "this is not valid toml = =")
     with pytest.raises(ValueError):
-        _load_toml_config(p, mode="wal")
-    os.unlink(p)
+        _resolve_config({"config_file_path": "vc.toml"}, mode="wal")
 
 
-def test_toml_missing_file():
+def test_resolve_toml_missing_file(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
     with pytest.raises(FileNotFoundError):
-        _load_toml_config("/tmp/definitely_not_here.toml", mode="wal")
+        _resolve_config({"config_file_path": "definitely_not_here.toml"}, mode="wal")
 
 
-from valuecounter import _resolve_config
+def test_resolve_toml_path_needs_plugin_dir(monkeypatch):
+    for name in ("PLUGIN_DIR", "INFLUXDB3_PLUGIN_DIR", "VIRTUAL_ENV"):
+        monkeypatch.delenv(name, raising=False)
+    with pytest.raises(ValueError, match="Cannot resolve plugin directory"):
+        _resolve_config({"config_file_path": "vc.toml"}, mode="wal")
 
 
-def test_resolve_inline_only():
-    cfg = _resolve_config({"fields": "status"}, plugin_dir="/tmp", mode="wal")
-    assert cfg.fields == ["status"]
-
 def test_resolve_neither_fields_nor_toml_raises():
     with pytest.raises(ValueError, match="fields"):
-        _resolve_config({}, plugin_dir="/tmp", mode="wal")
+        _resolve_config({}, mode="wal")
+
 
 def test_resolve_empty_fields_raises():
     with pytest.raises(ValueError, match="empty"):
-        _resolve_config({"fields": ""}, plugin_dir="/tmp", mode="wal")
+        _resolve_config({"fields": ""}, mode="wal")
+
 
 def test_resolve_mode_b_without_table_raises():
     with pytest.raises(ValueError, match="table"):
-        _resolve_config({"fields": "status"}, plugin_dir="/tmp", mode="scheduled")
+        _resolve_config({"fields": "status"}, mode="scheduled")
+
 
-def test_resolve_both_inline_and_toml_raises():
-    p = _write_tmp_toml('fields = ["status"]\n')
+def test_resolve_both_inline_and_toml_raises(monkeypatch, tmp_path):
+    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
+    _write_tmp_toml(tmp_path, 'fields = ["status"]\n')
     with pytest.raises(ValueError, match="either.*both"):
-        _resolve_config(
-            {"config_file_path": os.path.basename(p), "fields": "status"},
-            plugin_dir=os.path.dirname(p),
-            mode="wal",
-        )
-    os.unlink(p)
+        _resolve_config({"config_file_path": "vc.toml", "fields": "status"}, mode="wal")
+
 
 def test_resolve_validates_identifier_table():
     with pytest.raises(ValueError, match="invalid table"):
         _resolve_config(
-            {"table": "http requests", "fields": "status"},
-            plugin_dir="/tmp", mode="scheduled",
+            {"table": "http requests", "fields": "status"}, mode="scheduled"
         )
 
+
 def test_resolve_validates_identifier_field():
     with pytest.raises(ValueError, match="invalid field"):
-        _resolve_config(
-            {"fields": "status method-name"}, plugin_dir="/tmp", mode="wal",
-        )
+        _resolve_config({"fields": "status method-name"}, mode="wal")
 
 
 def test_resolve_rejects_empty_output_suffix():
     with pytest.raises(ValueError, match="output_suffix"):
-        _resolve_config(
-            {"fields": "status", "output_suffix": ""}, plugin_dir="/tmp", mode="wal",
-        )
+        _resolve_config({"fields": "status", "output_suffix": ""}, mode="wal")
 
 
 from valuecounter import _series_key
@@ -442,11 +516,13 @@ def test_series_key_stable_across_tag_order():
     k2 = _series_key("http_requests", {"endpoint": "/api", "host": "web-1"})
     assert k1 == k2
 
+
 def test_series_key_distinct_for_distinct_tags():
     k1 = _series_key("http_requests", {"host": "web-1"})
     k2 = _series_key("http_requests", {"host": "web-2"})
     assert k1 != k2
 
+
 def test_series_key_includes_table():
     k1 = _series_key("table_a", {"host": "x"})
     k2 = _series_key("table_b", {"host": "x"})
@@ -454,6 +530,7 @@ def test_series_key_includes_table():
     assert "table_a" in k1
     assert "table_b" in k2
 
+
 def test_series_key_empty_tags():
     k = _series_key("http_requests", {})
     assert k.startswith("http_requests:")
@@ -467,31 +544,40 @@ def test_extract_tags_subset():
     out = _extract_tags(row, ["host", "endpoint"])
     assert out == {"host": "web-1", "endpoint": "/api"}
 
+
 def test_extract_tags_missing_tag_returns_partial():
     row = {"host": "web-1", "status": 200}
     out = _extract_tags(row, ["host", "endpoint"])
     assert out == {"host": "web-1"}
 
+
 def test_extract_tags_empty_list():
     row = {"status": 200}
     assert _extract_tags(row, []) == {}
 
+
 def test_extract_tags_null_tag_returns_none():
     row = {"host": "web-1", "endpoint": None}
     assert _extract_tags(row, ["host", "endpoint"]) is None
 
 
-from valuecounter import _build_line
+import valuecounter
+from valuecounter import _build_rollup_line
+
+
+@pytest.fixture(autouse=True)
+def _inject_line_builder(monkeypatch):
+    """The engine injects LineBuilder as a global; tests use the vendored copy."""
+    monkeypatch.setattr(valuecounter, "LineBuilder", LineBuilder, raising=False)
 
 
 def test_build_line_basic():
-    lb = _build_line(
+    lb = _build_rollup_line(
         "http_requests",
         {"host": "web-1"},
         {"status_200": 3, "status_500": 1},
         "_valuecounts",
         1731000060000000000,
-        LineBuilder,
     )
     out = lb.build()
     assert out.startswith("http_requests_valuecounts,host=web-1 ")
@@ -499,34 +585,15 @@ def test_build_line_basic():
     assert "status_500=1i" in out
     assert out.endswith(" 1731000060000000000")
 
+
 def test_build_line_tagless():
-    lb = _build_line("t", {}, {"a_1": 2}, "_vc", 1, LineBuilder)
+    lb = _build_rollup_line("t", {}, {"a_1": 2}, "_vc", 1)
     assert lb.build() == "t_vc a_1=2i 1"
 
-def test_build_line_empty_counts_returns_none():
-    assert _build_line("t", {}, {}, "_vc", 1, LineBuilder) is None
-
 
-from valuecounter import _BatchLines
-
-
-def test_batchlines_joins():
-    lb1 = LineBuilder("m1"); lb1.int64_field("x", 1).time_ns(1)
-    lb2 = LineBuilder("m2"); lb2.int64_field("y", 2).time_ns(2)
-    b = _BatchLines([lb1, lb2])
-    assert b.build() == "m1 x=1i 1\nm2 y=2i 2"
-
-def test_batchlines_empty_raises():
-    b = _BatchLines([])
-    with pytest.raises(ValueError, match="no lines"):
-        b.build()
-
-def test_batchlines_caches_build():
-    lb = LineBuilder("m"); lb.int64_field("x", 1).time_ns(1)
-    b = _BatchLines([lb])
-    out1 = b.build()
-    out2 = b.build()
-    assert out1 == out2
+def test_build_line_empty_counts_raises():
+    with pytest.raises(ValueError, match="no fields"):
+        _build_rollup_line("t", {}, {}, "_vc", 1)
 
 
 from valuecounter import _build_scheduled_query
@@ -542,6 +609,7 @@ def test_build_query_with_tags():
     assert "GROUP BY" in sql
     assert "COUNT(*)" in sql.upper() or "count(*)" in sql
 
+
 def test_build_query_tagless():
     sql = _build_scheduled_query("t", [], "field_a")
     # No tag projection prefix or GROUP BY tag prefix; only the watched field
@@ -553,37 +621,6 @@ def test_build_query_tagless():
     assert "SELECT ," not in sql.replace(" ", "")[:8]
 
 
-from valuecounter import _get_tag_names
-
-
-def test_get_tag_names_cache_miss_runs_query_and_caches():
-    fake = FakeInfluxdb3Local()
-    fake.query_responses["information_schema.columns"] = [
-        {"column_name": "host"},
-        {"column_name": "endpoint"},
-    ]
-    out = _get_tag_names(fake, "http_requests")
-    assert out == ["host", "endpoint"]
-    assert len(fake.queries_run) == 1
-    # Cached for 1h
-    assert fake.cache._d.get("vc:tags:http_requests") == ["host", "endpoint"]
-    assert fake.cache._ttls.get("vc:tags:http_requests") == 3600
-
-def test_get_tag_names_cache_hit_skips_query():
-    fake = FakeInfluxdb3Local()
-    fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600)
-    out = _get_tag_names(fake, "http_requests")
-    assert out == ["host"]
-    assert fake.queries_run == []
-
-def test_get_tag_names_empty_result_caches_empty_list():
-    fake = FakeInfluxdb3Local()
-    fake.query_responses["information_schema.columns"] = []
-    out = _get_tag_names(fake, "tagless")
-    assert out == []
-    assert fake.cache._d.get("vc:tags:tagless") == []
-
-
 from valuecounter import _query_field_distribution
 
 
@@ -593,8 +630,12 @@ def test_query_field_distribution_calls_query_with_string_params():
         {"host": "a", "status": "200", "cnt": 3},
     ]
     rows = _query_field_distribution(
-        fake, "http_requests", ["host"], "status",
-        start_ns=1_000_000_000, end_ns=2_000_000_000,
+        fake,
+        "http_requests",
+        ["host"],
+        "status",
+        start_ns=1_000_000_000,
+        end_ns=2_000_000_000,
     )
     assert rows == [{"host": "a", "status": "200", "cnt": 3}]
     # Verify params dict was decimal strings
@@ -613,32 +654,34 @@ def _dt_ns(ns):
     return datetime.fromtimestamp(ns / 1_000_000_000, tz=timezone.utc)
 
 
-def test_scheduled_first_fire_sets_anchor_no_emit(monkeypatch, tmp_path):
+def test_scheduled_first_fire_sets_anchor_no_emit():
     fake = FakeInfluxdb3Local()
-    # Make process_scheduled_call see LineBuilder as a module-level injected global
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     call_time = _dt_ns(1_000_000_000_000_000_000)
     process_scheduled_call(
-        fake, call_time, args={"table": "http_requests", "fields": "status"},
+        fake,
+        call_time,
+        args={"table": "http_requests", "fields": "status"},
     )
 
     # Anchor written, no rollup emitted
-    assert fake.cache._d.get("vc:scheduled:last_call_ns:http_requests") == 1_000_000_000_000_000_000
+    assert (
+        fake.cache._d.get("vc:scheduled:last_call_ns:http_requests")
+        == 1_000_000_000_000_000_000
+    )
     assert fake.writes == []
     # Info log emitted
     assert any("first fire" in m for _, m in fake.logs)
 
 
-def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path):
+def test_scheduled_second_fire_emits_and_advances_anchor():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     # Pre-seed anchor and tag-name cache
-    fake.cache.put("vc:scheduled:last_call_ns:http_requests", 1_000_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600)
+    fake.cache.put(
+        "vc:scheduled:last_call_ns:http_requests", 1_000_000_000_000_000_000, ttl=None
+    )
+    fake.cache.put("shared:tags:http_requests", ["host"], ttl=3600)
     fake.query_responses['"status"'] = [
         {"host": "a", "status": "200", "cnt": 3},
         {"host": "a", "status": "500", "cnt": 1},
@@ -646,7 +689,9 @@ def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path):
 
     call_time = _dt_ns(2_000_000_000_000_000_000)
     process_scheduled_call(
-        fake, call_time, args={"table": "http_requests", "fields": "status"},
+        fake,
+        call_time,
+        args={"table": "http_requests", "fields": "status"},
     )
 
     assert len(fake.writes) == 1
@@ -654,21 +699,23 @@ def test_scheduled_second_fire_emits_and_advances_anchor(monkeypatch, tmp_path):
     assert "http_requests_valuecounts,host=a status_200=3i,status_500=1i" in out
     assert out.endswith(" 2000000000000000000")
     # Anchor advanced
-    assert fake.cache._d["vc:scheduled:last_call_ns:http_requests"] == 2_000_000_000_000_000_000
+    assert (
+        fake.cache._d["vc:scheduled:last_call_ns:http_requests"]
+        == 2_000_000_000_000_000_000
+    )
 
 
-def test_scheduled_write_failure_does_not_advance_anchor(monkeypatch, tmp_path):
+def test_scheduled_write_failure_does_not_advance_anchor():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
     fake.query_responses['"f"'] = [{"host": "a", "f": "x", "cnt": 1}]
     fake.fail_write = True
 
     process_scheduled_call(
-        fake, _dt_ns(2_000_000_000_000_000_000),
+        fake,
+        _dt_ns(2_000_000_000_000_000_000),
         args={"table": "t", "fields": "f"},
     )
 
@@ -676,17 +723,16 @@ def test_scheduled_write_failure_does_not_advance_anchor(monkeypatch, tmp_path):
     assert any("write failed" in m for _, m in fake.logs)
 
 
-def test_scheduled_empty_window_advances_anchor(monkeypatch, tmp_path):
+def test_scheduled_empty_window_advances_anchor():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
     # query_responses empty → no rows in window
 
     process_scheduled_call(
-        fake, _dt_ns(2_000_000_000_000_000_000),
+        fake,
+        _dt_ns(2_000_000_000_000_000_000),
         args={"table": "t", "fields": "f"},
     )
 
@@ -695,17 +741,16 @@ def test_scheduled_empty_window_advances_anchor(monkeypatch, tmp_path):
     assert any("no rows" in m for _, m in fake.logs)
 
 
-def test_scheduled_cross_db_write(monkeypatch, tmp_path):
+def test_scheduled_cross_db_write():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:t", [], ttl=3600)  # tagless
+    fake.cache.put("shared:tags:t", [], ttl=3600)  # tagless
     fake.query_responses['"f"'] = [{"f": "x", "cnt": 5}]
 
     process_scheduled_call(
-        fake, _dt_ns(2_000_000_000_000_000_000),
+        fake,
+        _dt_ns(2_000_000_000_000_000_000),
         args={"table": "t", "fields": "f", "dest_database": "rollups"},
     )
 
@@ -716,20 +761,19 @@ def test_scheduled_cross_db_write(monkeypatch, tmp_path):
     assert lp.startswith("t_valuecounts f_x=5i")
 
 
-def test_scheduled_sanitization_collision_sums_and_warns(monkeypatch, tmp_path):
+def test_scheduled_sanitization_collision_sums_and_warns():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     fake.cache.put("vc:scheduled:last_call_ns:t", 1_000_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
     fake.query_responses['"f"'] = [
-        {"host": "a", "f": "not found", "cnt": 2},   # sanitized: f_not_found
-        {"host": "a", "f": "not_found", "cnt": 3},   # already plain: f_not_found
+        {"host": "a", "f": "not found", "cnt": 2},  # sanitized: f_not_found
+        {"host": "a", "f": "not_found", "cnt": 3},  # already plain: f_not_found
     ]
 
     process_scheduled_call(
-        fake, _dt_ns(2_000_000_000_000_000_000),
+        fake,
+        _dt_ns(2_000_000_000_000_000_000),
         args={"table": "t", "fields": "f"},
     )
 
@@ -738,17 +782,16 @@ def test_scheduled_sanitization_collision_sums_and_warns(monkeypatch, tmp_path):
     assert any("collision" in m for lvl, m in fake.logs if lvl == "warn")
 
 
-def test_scheduled_call_time_ns_preserves_microseconds(monkeypatch, tmp_path):
+def test_scheduled_call_time_ns_preserves_microseconds():
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
 
     fake.cache.put("vc:scheduled:last_call_ns:t", 1_700_000_000_000_000_000, ttl=None)
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
     fake.query_responses['"f"'] = [{"host": "a", "f": "x", "cnt": 1}]
 
     # datetime with microseconds = 123456 -> expected ns suffix carries it exactly
     from datetime import datetime, timezone
+
     call_time = datetime(2026, 1, 1, 0, 0, 0, 123456, tzinfo=timezone.utc)
     expected_ns = int(call_time.timestamp()) * 1_000_000_000 + 123456 * 1000
 
@@ -758,30 +801,31 @@ def test_scheduled_call_time_ns_preserves_microseconds(monkeypatch, tmp_path):
     assert fake.writes[0].endswith(f" {expected_ns}")
 
 
-import time
 from valuecounter import process_writes
 
 
-def test_wal_first_batch_accumulates_no_emit(monkeypatch, tmp_path):
+def test_wal_first_batch_accumulates_no_emit(monkeypatch):
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
-    fake.cache.put("vc:tags:http_requests", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:http_requests", ["host"], ttl=3600)
 
     # Freeze time so period gate cannot fire
     fixed_now = 1_000_000_000_000_000_000
     monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now)
 
-    table_batches = [{
-        "table_name": "http_requests",
-        "rows": [
-            {"host": "web-1", "status": 200, "time": fixed_now - 1000},
-            {"host": "web-1", "status": 500, "time": fixed_now - 500},
-            {"host": "web-1", "status": 200, "time": fixed_now - 100},
-        ],
-    }]
+    table_batches = [
+        {
+            "table_name": "http_requests",
+            "rows": [
+                {"host": "web-1", "status": 200, "time": fixed_now - 1000},
+                {"host": "web-1", "status": 500, "time": fixed_now - 500},
+                {"host": "web-1", "status": 200, "time": fixed_now - 100},
+            ],
+        }
+    ]
 
-    process_writes(fake, table_batches, args={"fields": "status", "period_seconds": "60"})
+    process_writes(
+        fake, table_batches, args={"fields": "status", "period_seconds": "60"}
+    )
 
     # No emit yet — last_emit_ns == now_ns means period not elapsed
     assert fake.writes == []
@@ -795,34 +839,41 @@ def test_wal_first_batch_accumulates_no_emit(monkeypatch, tmp_path):
     assert series_entry["last_emit_ns"] == fixed_now
 
 
-def test_wal_period_elapsed_emits_and_subtracts(monkeypatch, tmp_path):
+def test_wal_period_elapsed_emits_and_subtracts(monkeypatch):
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
 
     # Pre-seed series state with last_emit_ns far in the past
     sh = _series_key("t", {"host": "a"})
-    fake.cache.put(f"vc:wal:{sh}", {
-        "table": "t", "tags": {"host": "a"},
-        "counts": {"status_200": 5},
-        "last_emit_ns": 0,
-    }, ttl=120)
+    fake.cache.put(
+        f"vc:wal:{sh}",
+        {
+            "table": "t",
+            "tags": {"host": "a"},
+            "counts": {"status_200": 5},
+            "last_emit_ns": 0,
+        },
+        ttl=120,
+    )
     fake.cache.put("vc:wal:_index:t", [sh], ttl=120)
 
     fixed_now = 1_000_000_000_000_000_000
     monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now)
 
     # New batch adds two more 200s (so during snapshot counts=5, after add counts=7)
-    table_batches = [{
-        "table_name": "t",
-        "rows": [
-            {"host": "a", "status": 200, "time": fixed_now - 100},
-            {"host": "a", "status": 200, "time": fixed_now - 50},
-        ],
-    }]
+    table_batches = [
+        {
+            "table_name": "t",
+            "rows": [
+                {"host": "a", "status": 200, "time": fixed_now - 100},
+                {"host": "a", "status": 200, "time": fixed_now - 50},
+            ],
+        }
+    ]
 
-    process_writes(fake, table_batches, args={"fields": "status", "period_seconds": "60"})
+    process_writes(
+        fake, table_batches, args={"fields": "status", "period_seconds": "60"}
+    )
 
     # One emit with counts=7 (5 pre + 2 new)
     assert len(fake.writes) == 1
@@ -833,26 +884,32 @@ def test_wal_period_elapsed_emits_and_subtracts(monkeypatch, tmp_path):
     assert new_state["last_emit_ns"] == fixed_now
 
 
-def test_wal_write_failure_retains_counts(monkeypatch, tmp_path):
+def test_wal_write_failure_retains_counts(monkeypatch):
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
 
     sh = _series_key("t", {"host": "a"})
-    fake.cache.put(f"vc:wal:{sh}", {
-        "table": "t", "tags": {"host": "a"},
-        "counts": {"status_200": 3},
-        "last_emit_ns": 0,
-    }, ttl=120)
+    fake.cache.put(
+        f"vc:wal:{sh}",
+        {
+            "table": "t",
+            "tags": {"host": "a"},
+            "counts": {"status_200": 3},
+            "last_emit_ns": 0,
+        },
+        ttl=120,
+    )
     fake.cache.put("vc:wal:_index:t", [sh], ttl=120)
 
     fixed_now = 1_000_000_000_000_000_000
     monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now)
     fake.fail_write = True
 
-    process_writes(fake, [{"table_name": "t", "rows": []}],
-                   args={"fields": "status", "period_seconds": "60"})
+    process_writes(
+        fake,
+        [{"table_name": "t", "rows": []}],
+        args={"fields": "status", "period_seconds": "60"},
+    )
 
     # Counts retained for retry; last_emit_ns NOT advanced
     state = fake.cache._d[f"vc:wal:{sh}"]
@@ -861,51 +918,63 @@ def test_wal_write_failure_retains_counts(monkeypatch, tmp_path):
     assert any("emit failed" in m for _, m in fake.logs)
 
 
-def test_wal_index_pruning_removes_expired_hashes(monkeypatch, tmp_path):
+def test_wal_index_pruning_removes_expired_hashes(monkeypatch):
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
 
     sh_live = _series_key("t", {"host": "a"})
     sh_dead = _series_key("t", {"host": "b"})
-    fake.cache.put(f"vc:wal:{sh_live}", {
-        "table": "t", "tags": {"host": "a"},
-        "counts": {}, "last_emit_ns": 0,
-    }, ttl=120)
+    fake.cache.put(
+        f"vc:wal:{sh_live}",
+        {
+            "table": "t",
+            "tags": {"host": "a"},
+            "counts": {},
+            "last_emit_ns": 0,
+        },
+        ttl=120,
+    )
     # sh_dead deliberately NOT in the cache (simulating TTL expiry)
     fake.cache.put("vc:wal:_index:t", [sh_live, sh_dead], ttl=120)
 
     fixed_now = 1_000_000_000_000_000_000
     monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now)
 
-    process_writes(fake, [{"table_name": "t", "rows": []}],
-                   args={"fields": "status", "period_seconds": "60"})
+    process_writes(
+        fake,
+        [{"table_name": "t", "rows": []}],
+        args={"fields": "status", "period_seconds": "60"},
+    )
 
     pruned = fake.cache._d["vc:wal:_index:t"]
     assert pruned == [sh_live]
 
 
-def test_wal_cross_db_write(monkeypatch, tmp_path):
+def test_wal_cross_db_write(monkeypatch):
     fake = FakeInfluxdb3Local()
-    monkeypatch.setattr("valuecounter.LineBuilder", LineBuilder, raising=False)
-    monkeypatch.setenv("PLUGIN_DIR", str(tmp_path))
-    fake.cache.put("vc:tags:t", ["host"], ttl=3600)
+    fake.cache.put("shared:tags:t", ["host"], ttl=3600)
 
     sh = _series_key("t", {"host": "a"})
-    fake.cache.put(f"vc:wal:{sh}", {
-        "table": "t", "tags": {"host": "a"},
-        "counts": {"status_200": 1},
-        "last_emit_ns": 0,
-    }, ttl=120)
+    fake.cache.put(
+        f"vc:wal:{sh}",
+        {
+            "table": "t",
+            "tags": {"host": "a"},
+            "counts": {"status_200": 1},
+            "last_emit_ns": 0,
+        },
+        ttl=120,
+    )
     fake.cache.put("vc:wal:_index:t", [sh], ttl=120)
 
     fixed_now = 1_000_000_000_000_000_000
     monkeypatch.setattr("valuecounter.time.time_ns", lambda: fixed_now)
 
-    process_writes(fake, [{"table_name": "t", "rows": []}],
-                   args={"fields": "status", "period_seconds": "60",
-                         "dest_database": "rollups"})
+    process_writes(
+        fake,
+        [{"table_name": "t", "rows": []}],
+        args={"fields": "status", "period_seconds": "60", "dest_database": "rollups"},
+    )
 
     assert fake.writes == []
     assert len(fake.cross_db_writes) == 1
diff --git a/influxdata/valuecounter/valuecounter.py b/influxdata/valuecounter/valuecounter.py
index 4827c9f..f035f6f 100644
--- a/influxdata/valuecounter/valuecounter.py
+++ b/influxdata/valuecounter/valuecounter.py
@@ -17,7 +17,13 @@
         {
             "name": "period_seconds",
             "example": "60",
-            "description": "Emission period in seconds. Cache TTL = 2x this value. Defaults to 60.",
+            "description": "Emission period in seconds, at least 1. Cache TTL = 2x this value. Defaults to 60.",
+            "required": false
+        },
+        {
+            "name": "period",
+            "example": "5min",
+            "description": "Emission period as a duration (e.g., '30s', '5min', '1h'), at least 1s. Units: 's', 'min', 'h', 'd', 'w'. Overridden by period_seconds when both are set.",
             "required": false
         },
         {
@@ -69,13 +75,19 @@
 """
 
 import hashlib
-import os
 import re
 import time
-import tomllib
 import uuid
 from dataclasses import dataclass
-from pathlib import Path
+
+from influxdata_plugin_utils.config import Validator, load_plugin_config
+from influxdata_plugin_utils.introspection import get_tag_names
+from influxdata_plugin_utils.parsing import (
+    parse_delimited_list,
+    parse_int,
+    parse_timedelta,
+)
+from influxdata_plugin_utils.write import build_line, write_data
 
 # At server runtime LineBuilder is injected as a builtin. In test environments
 # pytest patches this module-level name to a vendored copy. The reference in
@@ -110,15 +122,16 @@ def _validate_identifier(name, what):
         raise ValueError(f"invalid {what}: {name!r}")
 
 
-_DURATION_RE = re.compile(r"^\s*(\d+)\s*([smhd])\s*$")
-_DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
+_BARE_MINUTES_RE = re.compile(r"^\s*(\d+)\s*m\s*$")
 
 
-def _parse_duration(s):
-    m = _DURATION_RE.match(s)
-    if not m:
-        raise ValueError(f"invalid duration: {s!r} (expected e.g. '60s', '5m', '1h', '1d')")
-    return int(m.group(1)) * _DURATION_UNITS[m.group(2)]
+def _parse_period_seconds(raw):
+    match = _BARE_MINUTES_RE.match(str(raw))
+    text = f"{match.group(1)}min" if match else str(raw)
+    seconds = int(parse_timedelta(text).total_seconds())
+    if seconds < 1:
+        raise ValueError(f"invalid period: {raw!r} (must be at least 1 second)")
+    return seconds
 
 
 @dataclass
@@ -130,126 +143,82 @@ class Config:
     table: str = ""
 
 
-_MODE_A_ALLOWED = {"fields", "output_suffix", "period_seconds", "period", "dest_database", "config_file_path"}
-_MODE_B_ALLOWED = {"table", "fields", "output_suffix", "dest_database", "config_file_path"}
+_MODE_ALLOWED = {
+    "wal": {"fields", "output_suffix", "period", "period_seconds", "dest_database"},
+    "scheduled": {"table", "fields", "output_suffix", "dest_database"},
+}
 
 
-def _parse_inline_args(args, mode):
-    if mode == "wal":
-        allowed = _MODE_A_ALLOWED
-    elif mode == "scheduled":
-        allowed = _MODE_B_ALLOWED
-    else:
-        raise ValueError(f"unknown mode: {mode!r}")
+def _parse_fields(raw):
+    """Field names split on any whitespace; TOML may deliver a list already."""
+    if isinstance(raw, (list, tuple)):
+        return parse_delimited_list(raw)
+    return str(raw).split()
 
-    for key in args:
-        if key not in allowed:
-            # special-case better error messages for mode-incompatible knobs
-            if key == "table" and mode == "wal":
-                raise ValueError(
-                    "vc-wal: 'table' is determined by the trigger-spec, not inline args"
-                )
-            if key in ("period_seconds", "period") and mode == "scheduled":
-                raise ValueError(
-                    "vc-scheduled: 'period_seconds'/'period' is not used; Mode B is drift-based"
-                )
-            raise ValueError(f"unknown arg: {key!r}")
-
-    cfg = Config(fields=[])
-    if "fields" in args:
-        cfg.fields = args["fields"].split()
-    if "output_suffix" in args:
-        cfg.output_suffix = args["output_suffix"]
-    if "dest_database" in args:
-        cfg.dest_database = args["dest_database"]
-    if "table" in args:
-        cfg.table = args["table"]
-    if "period" in args:
-        cfg.period_seconds = _parse_duration(args["period"])
-    if "period_seconds" in args:
-        cfg.period_seconds = int(args["period_seconds"])
-    return cfg
 
+_COMMON_VALIDATORS = [
+    Validator("fields", default="", cast=_parse_fields),
+    Validator("output_suffix", default="_valuecounts", cast=str),
+    Validator("dest_database", default="", cast=str),
+]
 
-def _load_toml_config(path, mode):
-    if mode == "wal":
-        allowed = _MODE_A_ALLOWED - {"config_file_path"}
-    elif mode == "scheduled":
-        allowed = _MODE_B_ALLOWED - {"config_file_path"}
-    else:
-        raise ValueError(f"unknown mode: {mode!r}")
+# Only mode-valid keys get a validator, so registered defaults never trip the
+# unknown-key check below.
+_MODE_VALIDATORS = {
+    "wal": _COMMON_VALIDATORS,
+    "scheduled": _COMMON_VALIDATORS + [Validator("table", default="", cast=str)],
+}
 
-    try:
-        with open(path, "rb") as f:
-            data = tomllib.load(f)
-    except FileNotFoundError:
-        raise
-    except tomllib.TOMLDecodeError as e:
-        raise ValueError(f"TOML parse error in {path}: {e}")
-
-    for key in data:
-        if key not in allowed:
-            if key == "table" and mode == "wal":
-                raise ValueError(
-                    "vc-wal: 'table' is determined by the trigger-spec, not the TOML"
-                )
-            if key in ("period_seconds", "period") and mode == "scheduled":
-                raise ValueError(
-                    "vc-scheduled: 'period'/'period_seconds' is not used; Mode B is drift-based"
-                )
-            raise ValueError(f"unknown TOML key: {key!r}")
-
-    cfg = Config(fields=[])
-    if "fields" in data:
-        f = data["fields"]
-        if isinstance(f, str):
-            f = f.split()
-        cfg.fields = list(f)
-    if "output_suffix" in data:
-        cfg.output_suffix = data["output_suffix"]
-    if "dest_database" in data:
-        cfg.dest_database = data["dest_database"]
-    if "table" in data:
-        cfg.table = data["table"]
-    if "period" in data:
-        cfg.period_seconds = _parse_duration(data["period"])
-    if "period_seconds" in data:
-        cfg.period_seconds = int(data["period_seconds"])
-    return cfg
+
+def _reject_unknown_keys(loaded, mode, from_toml):
+    key_label = "TOML key" if from_toml else "arg"
+    source_label = "the TOML" if from_toml else "inline args"
+    allowed = _MODE_ALLOWED[mode]
+
+    for key in (name.lower() for name in loaded.as_dict()):
+        if key in allowed:
+            continue
+        # special-case better error messages for mode-incompatible knobs
+        if key == "table" and mode == "wal":
+            raise ValueError(
+                f"vc-wal: 'table' is determined by the trigger-spec, not {source_label}"
+            )
+        if key in ("period", "period_seconds") and mode == "scheduled":
+            raise ValueError(
+                "vc-scheduled: 'period'/'period_seconds' is not used; Mode B is drift-based"
+            )
+        raise ValueError(f"unknown {key_label}: {key!r}")
 
 
-def _resolve_config(args, plugin_dir, mode):
-    args = dict(args)  # defensive copy
-    cfp = args.pop("config_file_path", None)
+def _resolve_config(args, mode):
+    if mode not in _MODE_ALLOWED:
+        raise ValueError(f"unknown mode: {mode!r}")
 
-    if cfp is not None and args:
+    args = dict(args or {})  # defensive copy
+    config_file_path = args.get("config_file_path")
+    if config_file_path is not None and len(args) > 1:
         raise ValueError("set either config_file_path or inline args, not both")
 
-    if cfp is not None:
-        if plugin_dir is not None:
-            path = os.path.join(plugin_dir, cfp)
-        else:
-            # Fallbacks for servers where the operator has not exported PLUGIN_DIR:
-            #  - INFLUXDB3_PLUGIN_DIR: set when the server is configured via env var
-            #  - VIRTUAL_ENV: exported by the processing engine; default venv is <plugin-dir>/.venv
-            candidates = []
-            if influxdb3_plugin_dir := os.environ.get("INFLUXDB3_PLUGIN_DIR"):
-                candidates.append(influxdb3_plugin_dir)
-            if virtual_env := os.environ.get("VIRTUAL_ENV"):
-                candidates.append(str(Path(virtual_env).parent))
-
-            path = None
-            for base in candidates:
-                candidate = os.path.join(base, cfp)
-                if os.path.exists(candidate):
-                    path = candidate
-                    break
-            if path is None:
-                # Original default: resolve against the current directory.
-                path = os.path.join(".", cfp)
-        cfg = _load_toml_config(path, mode=mode)
-    else:
-        cfg = _parse_inline_args(args, mode=mode)
+    loaded = load_plugin_config(
+        args,
+        validators=_MODE_VALIDATORS[mode],
+        source="toml" if config_file_path else "args",
+    )
+    _reject_unknown_keys(loaded, mode, from_toml=bool(config_file_path))
+
+    cfg = Config(
+        fields=[str(f) for f in loaded.fields],
+        output_suffix=loaded.output_suffix,
+        dest_database=loaded.dest_database,
+        table=str(loaded.get("table") or ""),
+    )
+
+    if mode == "wal":
+        # 'period_seconds' wins when both spellings are present
+        if (period := loaded.get("period")) is not None:
+            cfg.period_seconds = _parse_period_seconds(period)
+        if (period_seconds := loaded.get("period_seconds")) is not None:
+            cfg.period_seconds = parse_int(period_seconds, minimum=1)
 
     if not cfg.fields:
         raise ValueError("config error: 'fields' is empty or missing")
@@ -264,7 +233,9 @@ def _resolve_config(args, plugin_dir, mode):
         _validate_identifier(f, "field")
 
     if cfg.output_suffix == "":
-        raise ValueError("config error: 'output_suffix' cannot be empty (would risk feedback loop in Mode A and ambiguity in Mode B)")
+        raise ValueError(
+            "config error: 'output_suffix' cannot be empty (would risk feedback loop in Mode A and ambiguity in Mode B)"
+        )
 
     return cfg
 
@@ -287,35 +258,14 @@ def _extract_tags(row, tag_names):
     return out
 
 
-def _build_line(table, tags, counts, output_suffix, ts_ns, line_builder_cls):
-    if not counts:
-        return None
-    lb = line_builder_cls(f"{table}{output_suffix}")
-    for k, v in tags.items():
-        lb.tag(k, str(v))
-    for k, v in counts.items():
-        lb.int64_field(k, int(v))
-    lb.time_ns(ts_ns)
-    return lb
-
-
-def _get_tag_names(influxdb3_local, table_name):
-    key = f"vc:tags:{table_name}"
-    cached = influxdb3_local.cache.get(key)
-    if cached is not None:
-        return cached
-    res = influxdb3_local.query(
-        """
-        SELECT column_name
-        FROM information_schema.columns
-        WHERE table_name = $tbl
-          AND data_type = 'Dictionary(Int32, Utf8)'
-        """,
-        {"tbl": table_name},
+def _build_rollup_line(table, tags, counts, output_suffix, ts_ns):
+    return build_line(
+        LineBuilder,
+        f"{table}{output_suffix}",
+        tags=tags,
+        fields=counts,
+        time_ns=ts_ns,
     )
-    tag_names = [r["column_name"] for r in res]
-    influxdb3_local.cache.put(key, tag_names, ttl=3600)
-    return tag_names
 
 
 def _build_scheduled_query(table, tag_names, field_name):
@@ -339,33 +289,23 @@ def _build_scheduled_query(table, tag_names, field_name):
     )
 
 
-def _query_field_distribution(influxdb3_local, table, tag_names, field_name, start_ns, end_ns):
+def _query_field_distribution(
+    influxdb3_local, table, tag_names, field_name, start_ns, end_ns
+):
     sql = _build_scheduled_query(table, tag_names, field_name)
     params = {"start_ns": str(start_ns), "end_ns": str(end_ns)}
     return influxdb3_local.query(sql, params)
 
 
-class _BatchLines:
-    def __init__(self, line_builders):
-        self._line_builders = list(line_builders)
-        self._built = None
-
-    def build(self):
-        if self._built is None:
-            lines = [b.build() for b in self._line_builders]
-            if not lines:
-                raise ValueError("batch_write received no lines to build")
-            self._built = "\n".join(lines)
-        return self._built
-
-
 def process_scheduled_call(influxdb3_local, call_time, args=None):
     task_id = uuid.uuid4().hex[:8]
 
     # call_time arrives as a PyDateTime per system_py.rs:847,867
-    call_time_ns = int(call_time.timestamp()) * 1_000_000_000 + call_time.microsecond * 1000
+    call_time_ns = (
+        int(call_time.timestamp()) * 1_000_000_000 + call_time.microsecond * 1000
+    )
 
-    cfg = _resolve_config(args or {}, os.environ.get("PLUGIN_DIR"), mode="scheduled")
+    cfg = _resolve_config(args, mode="scheduled")
 
     anchor_key = f"vc:scheduled:last_call_ns:{cfg.table}"
     last_call_ns = influxdb3_local.cache.get(anchor_key)
@@ -377,7 +317,7 @@ def process_scheduled_call(influxdb3_local, call_time, args=None):
         )
         return
 
-    tag_names = _get_tag_names(influxdb3_local, cfg.table)
+    tag_names = get_tag_names(influxdb3_local, cfg.table)
     for t in tag_names:
         _validate_identifier(t, "tag column")
 
@@ -390,8 +330,12 @@ def process_scheduled_call(influxdb3_local, call_time, args=None):
     for field_name in cfg.fields:
         try:
             rows = _query_field_distribution(
-                influxdb3_local, cfg.table, tag_names, field_name,
-                window_start_ns, window_end_ns,
+                influxdb3_local,
+                cfg.table,
+                tag_names,
+                field_name,
+                window_start_ns,
+                window_end_ns,
             )
         except Exception as e:
             influxdb3_local.error(
@@ -431,21 +375,20 @@ def process_scheduled_call(influxdb3_local, call_time, args=None):
         return
 
     builders = [
-        _build_line(ss["table"], ss["tags"], ss["counts"], cfg.output_suffix,
-                    call_time_ns, LineBuilder)
+        _build_rollup_line(
+            ss["table"], ss["tags"], ss["counts"], cfg.output_suffix, call_time_ns
+        )
         for ss in series.values()
     ]
-    builders = [b for b in builders if b is not None]
-    if not builders:
-        influxdb3_local.cache.put(anchor_key, call_time_ns, ttl=None)
-        return
 
     try:
-        batch = _BatchLines(builders)
-        if cfg.dest_database:
-            influxdb3_local.write_sync_to_db(cfg.dest_database, batch, no_sync=True)
-        else:
-            influxdb3_local.write_sync(batch, no_sync=True)
+        write_data(
+            influxdb3_local,
+            builders,
+            retries=0,
+            no_sync=True,
+            database=cfg.dest_database or None,
+        )
     except Exception as e:
         influxdb3_local.error(f"[{task_id}] vc-scheduled: write failed: {e}")
         return  # anchor unchanged → next fire's window covers two periods
@@ -457,7 +400,7 @@ def process_writes(influxdb3_local, table_batches, args=None):
     task_id = uuid.uuid4().hex[:8]
     now_ns = time.time_ns()
 
-    cfg = _resolve_config(args or {}, os.environ.get("PLUGIN_DIR"), mode="wal")
+    cfg = _resolve_config(args, mode="wal")
 
     period_ns = cfg.period_seconds * 1_000_000_000
     ttl = 2 * cfg.period_seconds
@@ -470,7 +413,7 @@ def process_writes(influxdb3_local, table_batches, args=None):
         by_table.setdefault(name, []).extend(batch["rows"])
 
     for table_name, rows in by_table.items():
-        tag_names = _get_tag_names(influxdb3_local, table_name)
+        tag_names = get_tag_names(influxdb3_local, table_name)
         for t in tag_names:
             _validate_identifier(t, "tag column")
 
@@ -527,7 +470,9 @@ def process_writes(influxdb3_local, table_batches, args=None):
                 continue
             if now_ns - state["last_emit_ns"] < period_ns:
                 continue
-            to_emit.append((sh, state["table"], dict(state["tags"]), dict(state["counts"])))
+            to_emit.append(
+                (sh, state["table"], dict(state["tags"]), dict(state["counts"]))
+            )
 
         # Prune index — only live hashes survive
         if live != active_hashes:
@@ -540,19 +485,18 @@ def process_writes(influxdb3_local, table_batches, args=None):
             continue
 
         builders = [
-            _build_line(t_, tags, counts, cfg.output_suffix, now_ns, LineBuilder)
+            _build_rollup_line(t_, tags, counts, cfg.output_suffix, now_ns)
             for _sh, t_, tags, counts in to_emit
         ]
-        builders = [b for b in builders if b is not None]
-        if not builders:
-            continue
 
         try:
-            batch = _BatchLines(builders)
-            if cfg.dest_database:
-                influxdb3_local.write_sync_to_db(cfg.dest_database, batch, no_sync=True)
-            else:
-                influxdb3_local.write_sync(batch, no_sync=True)
+            write_data(
+                influxdb3_local,
+                builders,
+                retries=0,
+                no_sync=True,
+                database=cfg.dest_database or None,
+            )
         except Exception as e:
             influxdb3_local.error(
                 f"[{task_id}] vc-wal: emit failed, counts retained for retry: {e}"
diff --git a/influxdata/valuecounter/valuecounter_config.toml b/influxdata/valuecounter/valuecounter_config.toml
index 80ad4a6..917ed75 100644
--- a/influxdata/valuecounter/valuecounter_config.toml
+++ b/influxdata/valuecounter/valuecounter_config.toml
@@ -7,7 +7,7 @@
 # Mode A — WAL trigger (process_writes)
 # ------------------------------------------------------------
 # 'table' is determined by the WAL trigger spec (`table:<name>`); do NOT set it here.
-# period = "60s"
+# period = "60s"   # or period_seconds = 60; units: s, min, h, d, w
 # fields = ["status", "method"]
 # output_suffix = "_valuecounts"
 # dest_database = "rollups"