From 963f5277762e632f3c3ee5123799bef6e6ce290f Mon Sep 17 00:00:00 2001
From: ascottDI <ascott@homer.aquaq.co.uk>
Date: Thu, 13 Aug 2026 09:44:23 +0100
Subject: [PATCH 1/7] Initial implimentation

---
 di/tickerplant/VERSION       |   1 +
 di/tickerplant/init.q        |  11 ++
 di/tickerplant/tickerplant.q | 200 +++++++++++++++++++++++++++++++++++
 3 files changed, 212 insertions(+)
 create mode 100644 di/tickerplant/VERSION
 create mode 100644 di/tickerplant/init.q
 create mode 100644 di/tickerplant/tickerplant.q

diff --git a/di/tickerplant/VERSION b/di/tickerplant/VERSION
new file mode 100644
index 00000000..6e8bf73a
--- /dev/null
+++ b/di/tickerplant/VERSION
@@ -0,0 +1 @@
+0.1.0
diff --git a/di/tickerplant/init.q b/di/tickerplant/init.q
new file mode 100644
index 00000000..4a796242
--- /dev/null
+++ b/di/tickerplant/init.q
@@ -0,0 +1,11 @@
+/ di.tickerplant - tick-capture: log and publish incoming updates, roll at end of day
+/ hard dependencies - imported here as module-local handles (before the impl loads), used by tickerplant.q
+pubsub:use`di.pubsub
+eodtime:use`di.eodtime
+tplog:use`di.tplog
+\l ::tickerplant.q
+/ module version, read from the VERSION file (one plain-text file to bump per release). read
+/ module-relative at load (`:::` resolves to di/tickerplant) and BEFORE export, since export:([...])
+/ evaluates each name; version stays in the export so di.depcheck reads it from the export dict
+version:trim first read0`:::VERSION
+export:([init;upd;subscribe;endofday;getcounts;gettables;getapimeta;version])
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
new file mode 100644
index 00000000..2aa3de0f
--- /dev/null
+++ b/di/tickerplant/tickerplant.q
@@ -0,0 +1,200 @@
+/ core tick-capture for the modular torq world: receive updates from feeds, write them to a
+/ tickerplant log for recovery, and publish them to subscribers, rolling the log at end of day.
+/ orchestrates three hard deps - di.pubsub (subscribe/publish), di.eodtime (roll timing), and
+/ di.tplog (log check/repair on recovery) - with an injected log and timer. the pubsub / eodtime /
+/ tplog module handles are imported in init.q (module-local) and used here.
+/ .
+/ NB the tickerplant's data tables live at ROOT, not in .z.m. a tickerplant process owns its tables
+/ (feeds insert into them, subscribers replay them), and di.pubsub reads them by name via `value`,
+/ so they cannot be module-local. this is the one deliberate root-state exception for this process
+/ module; all other mutable state is module-local in .z.m.
+/ .
+/ NB subscriber-disconnect cleanup is handled by di.pubsub's own .z.pc (it self-assigns it). this
+/ module therefore takes no di.handlers dependency. FLAG: di.pubsub should migrate to di.handlers so
+/ .z.pc is not assigned outside the central registry - out of scope here, tracked separately.
+
+/ ============================================================
+/ constants (load-time)
+/ ============================================================
+
+/ default batch publish interval (matches TorQ's 1s system timer); di.timer periods are whole seconds
+defaultbatchperiod:0D00:00:01;
+
+/ optional config keys forwarded verbatim to di.eodtime.init
+eodtimekeys:`rolltimezone`datatimezone`rolltimeoffset;
+
+/ ============================================================
+/ internal helpers
+/ ============================================================
+
+raiseerror:{[ctx;msg]
+  / internal - log an error under ctx then signal it, so failures are observable as well as thrown
+  .z.m.logerr[ctx;msg];
+  '"di.tickerplant: ",string[ctx],": ",msg;
+  };
+
+stamp:{[x]
+  / prepend a data-timestamp column unless the update already carries one (first value is a timestamp)
+  if[-12h=type first first x;:x];
+  a:.z.p+eodtime.getdailyadj[];
+  :$[0>type first x;a,x;(enlist(count first x)#a),x];
+  };
+
+openlog:{[date]
+  / open the tp log for `date`, checking/repairing a pre-existing log via di.tplog; 0i if logging off
+  if[0=count .z.m.logdir;:0i];
+  l:hsym `$ .z.m.logdir,"/",.z.m.logname,string date;
+  if[count key l;l:tplog.check[l;0]];
+  if[not count key l;l set ()];
+  .z.m.logfile:l;
+  h:hopen l;
+  .z.m.loginfo[`openlog;"logging to ",string l];
+  :h;
+  };
+
+publishbuffer:{[]
+  / batch mode: publish each root table's buffered rows to subscribers then clear them; i catches j
+  pubsub.pubclear[.z.m.tabs];
+  .z.m.i:.z.m.j;
+  };
+
+publishrows:{[t;x]
+  / zero-latency mode: publish one stamped update immediately, as a table keyed by t's columns
+  f:cols t;
+  pubsub.publish[t;$[0>type first x;enlist f!x;flip f!x]];
+  };
+
+writelog:{[t;x]
+  / append the update to the tp log (if enabled) and bump the total message count
+  if[.z.m.logfile>0i;.z.m.logfile enlist (`upd;t;x);.z.m.j+:1];
+  };
+
+rollcheck:{[now]
+  / trigger end-of-day if we have passed the next scheduled roll timestamp
+  if[eodtime.getnextroll[]<now;endofday[]];
+  };
+
+tick:{[]
+  / timer job: in batch mode flush the buffer to subscribers, then check for an end-of-day roll
+  if[.z.m.batch;publishbuffer[]];
+  rollcheck .z.p;
+  };
+
+rolllog:{[]
+  / close the current tp log and open a fresh one for the new date; reset log message counts
+  if[.z.m.logfile>0i;hclose .z.m.logfile];
+  .z.m.i:.z.m.j:0;
+  .z.m.logfile:openlog .z.m.d;
+  };
+
+/ ============================================================
+/ public api
+/ ============================================================
+
+init:{[deps]
+  / wire the injected log + timer, initialise the dep modules, materialise the tables at root, open
+  / the tp log and schedule the batch/roll timer job.
+  / deps: a dict with `log (required), `timer (required), `schemas (required, tablename!schema) and
+  /   optional `batch (1b), `batchperiod (timespan), `logdir (string, "" disables logging),
+  /   `logname (string), `subtables (symbol list), plus di.eodtime keys (rolltimezone/datatimezone/
+  /   rolltimeoffset) forwarded verbatim.
+  / example: tp.init[`log`timer`schemas!(logdep;timerdep;`trade`quote!(tradeschema;quoteschema))]
+  if[99h<>type deps;
+    '"di.tickerplant: deps must be a dict with `log, `timer and `schemas keys"];
+  if[not `log in key deps;
+    '"di.tickerplant: log dependency is required; pass `info`warn`error functions keyed on `log"];
+  if[99h<>type deps`log;
+    '"di.tickerplant: log value must be a dict; pass `info`warn`error functions"];
+  if[not all `info`warn`error in key deps`log;
+    '"di.tickerplant: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)];
+  if[not `timer in key deps;
+    '"di.tickerplant: timer dependency is required; pass di.timer's exports keyed on `timer"];
+  if[99h<>type deps`timer;
+    '"di.tickerplant: timer value must be a dict exposing addjob"];
+  if[not `addjob in key deps`timer;
+    '"di.tickerplant: timer dict must expose addjob"];
+  if[not `custom in key deps[`timer]`addjob;
+    '"di.tickerplant: timer addjob must expose the custom variant"];
+  if[not `schemas in key deps;
+    '"di.tickerplant: schemas is required; pass a tablename!schema dict keyed on `schemas"];
+  if[99h<>type deps`schemas;
+    '"di.tickerplant: schemas must be a dict of tablename!schema"];
+  .z.m.loginfo:(deps`log)`info;
+  .z.m.logwarn:(deps`log)`warn;
+  .z.m.logerr:(deps`log)`error;
+  .z.m.timer:deps`timer;
+  / optional config with defaults
+  .z.m.batch:$[`batch in key deps;deps`batch;1b];
+  .z.m.batchperiod:$[`batchperiod in key deps;deps`batchperiod;defaultbatchperiod];
+  .z.m.logdir:$[`logdir in key deps;deps`logdir;""];
+  .z.m.logname:$[`logname in key deps;deps`logname;"tp"];
+  / materialise the schemas as root tables (see the header note on root state), applying `g# to sym
+  .z.m.schemas:deps`schemas;
+  .z.m.tabs:key deps`schemas;
+  {[nm;s] (nm) set $[`sym in cols s;@[s;`sym;`g#];s]}'[.z.m.tabs;value deps`schemas];
+  / initialise the dependency modules: eodtime with the injected log + tz passthrough, pubsub over
+  / the subscribable tables
+  eodtime.init[(enlist[`log]!enlist deps`log),(key[deps] inter eodtimekeys)#deps];
+  pubsub.setsubtables[$[`subtables in key deps;deps`subtables;.z.m.tabs]];
+  pubsub.init[];
+  / date, counts, and the tp log for today
+  .z.m.d:eodtime.getd[];
+  .z.m.i:.z.m.j:0;
+  .z.m.logfile:openlog .z.m.d;
+  / schedule the timer job that flushes the buffer (batch) and checks the roll; mode 1 = fixed period.
+  / guarded so a re-init does not re-add (di.timer.addjob throws on a duplicate id); tick reads
+  / .z.m.batch live, so the one job serves both modes across re-inits
+  if[not `scheduled in key .z.m;
+    .z.m.timer[`addjob][`custom][`tickerplant;tick;();`int$.z.m.batchperiod%0D00:00:01;1h;()!()];
+    .z.m.scheduled:1b];
+  .z.m.loginfo[`init;"di.tickerplant initialised (",$[.z.m.batch;"batch";"zero-latency"]," mode)"];
+  };
+
+upd:{[t;x]
+  / feed entry point: stamp the update, then buffer+log (batch) or publish+log (zero-latency).
+  / t is the table name, x the column data. wired to root `upd` by di.torq so feeds can call it.
+  if[not -11h=type t;raiseerror[`upd;"table must be a symbol"]];
+  if[not t in .z.m.tabs;raiseerror[`upd;"unknown table ",string t]];
+  rollcheck .z.p;
+  x:stamp x;
+  if[.z.m.batch;t insert x;writelog[t;x]];
+  if[not .z.m.batch;publishrows[t;x];writelog[t;x]];
+  };
+
+subscribe:{[tabs;filters]
+  / register a subscriber (delegates to di.pubsub); called by downstream processes over IPC
+  :pubsub.subscribe[tabs;filters];
+  };
+
+endofday:{[]
+  / flush any buffer, notify subscribers, roll the tp log, advance eodtime state and reset counts
+  if[.z.m.batch;publishbuffer[]];
+  pubsub.callendofday[.z.m.d];
+  .z.m.d+:1;
+  rolllog[];
+  eodtime.setnextroll eodtime.getroll[.z.p];
+  eodtime.setdailyadj eodtime.getdailyadjustment[];
+  .z.m.loginfo[`endofday;"rolled to ",string .z.m.d];
+  };
+
+getcounts:{[]
+  / current message counts and trading date - i (in the log), j (log plus buffered), d (date)
+  :`i`j`d!(.z.m.i;.z.m.j;.z.m.d);
+  };
+
+gettables:{[]
+  / the tables this tickerplant captures
+  :.z.m.tabs;
+  };
+
+getapimeta:{[]
+  / this module's api metadata, one row per CALLABLE api function (NOT init/getapimeta/version - those
+  / are plumbing/metadata di.torq handles by convention), for di.torq to collect and register with
+  / di.api. names are bare; di.torq applies process-wide qualification.
+  :flip `name`public`descrip`params`return!flip(
+    (`upd;       1b; "feed entry point - stamp, log and publish (or buffer) an update";  "[symbol: table; list: data]";      "null");
+    (`subscribe; 1b; "register a subscriber for tables/syms (delegates to di.pubsub)";   "[symbol|list: tables; filters]";   "subscription result");
+    (`endofday;  1b; "flush, notify subscribers, roll the log and advance eod state";    "[]";                               "null");
+    (`getcounts; 1b; "current log/buffer message counts and trading date";              "[]";                               "dict: `i`j`d");
+    (`gettables; 1b; "the tables this tickerplant captures";                            "[]";                               "symbol list: table names"));
+  };

From 5339964d097c488a738750e0b0bc24c89d2016fe Mon Sep 17 00:00:00 2001
From: ascottDI <ascott@homer.aquaq.co.uk>
Date: Tue, 18 Aug 2026 10:23:53 +0100
Subject: [PATCH 2/7] connecting tplog and tickerplant

---
 di/tickerplant/deps.q        |  4 ++++
 di/tickerplant/tickerplant.q | 24 +++++++++++-------------
 2 files changed, 15 insertions(+), 13 deletions(-)
 create mode 100644 di/tickerplant/deps.q

diff --git a/di/tickerplant/deps.q b/di/tickerplant/deps.q
new file mode 100644
index 00000000..736b8306
--- /dev/null
+++ b/di/tickerplant/deps.q
@@ -0,0 +1,4 @@
+/ hard module dependencies and their minimum versions, validated by di.depcheck.
+/ these are the modules di.tickerplant `use`s in init.q. the injected log and timer are not declared
+/ here - di.depcheck validates them through its core-contract check.
+deps:`di.pubsub`di.eodtime`di.tplog!("0.1.0";"0.1.0";"0.1.0");
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
index 2aa3de0f..28be2683 100644
--- a/di/tickerplant/tickerplant.q
+++ b/di/tickerplant/tickerplant.q
@@ -44,7 +44,7 @@ openlog:{[date]
   / open the tp log for `date`, checking/repairing a pre-existing log via di.tplog; 0i if logging off
   if[0=count .z.m.logdir;:0i];
   l:hsym `$ .z.m.logdir,"/",.z.m.logname,string date;
-  if[count key l;l:tplog.check[l;0]];
+  if[count key l;l:tplog.check l];
   if[not count key l;l set ()];
   .z.m.logfile:l;
   h:hopen l;
@@ -98,7 +98,6 @@ init:{[deps]
   /   optional `batch (1b), `batchperiod (timespan), `logdir (string, "" disables logging),
   /   `logname (string), `subtables (symbol list), plus di.eodtime keys (rolltimezone/datatimezone/
   /   rolltimeoffset) forwarded verbatim.
-  / example: tp.init[`log`timer`schemas!(logdep;timerdep;`trade`quote!(tradeschema;quoteschema))]
   if[99h<>type deps;
     '"di.tickerplant: deps must be a dict with `log, `timer and `schemas keys"];
   if[not `log in key deps;
@@ -131,10 +130,11 @@ init:{[deps]
   / materialise the schemas as root tables (see the header note on root state), applying `g# to sym
   .z.m.schemas:deps`schemas;
   .z.m.tabs:key deps`schemas;
-  {[nm;s] (nm) set $[`sym in cols s;@[s;`sym;`g#];s]}'[.z.m.tabs;value deps`schemas];
-  / initialise the dependency modules: eodtime with the injected log + tz passthrough, pubsub over
-  / the subscribable tables
+  {[nm;s] nm set $[`sym in cols s;@[s;`sym;`g#];s]}'[.z.m.tabs;value deps`schemas];
+  / initialise the dependency modules: eodtime (log + tz passthrough), tplog (log), pubsub (over the
+  / subscribable tables). tplog now takes an injected log and must be init'd before check is called.
   eodtime.init[(enlist[`log]!enlist deps`log),(key[deps] inter eodtimekeys)#deps];
+  tplog.init[enlist[`log]!enlist deps`log];
   pubsub.setsubtables[$[`subtables in key deps;deps`subtables;.z.m.tabs]];
   pubsub.init[];
   / date, counts, and the tp log for today
@@ -188,13 +188,11 @@ gettables:{[]
   };
 
 getapimeta:{[]
-  / this module's api metadata, one row per CALLABLE api function (NOT init/getapimeta/version - those
-  / are plumbing/metadata di.torq handles by convention), for di.torq to collect and register with
-  / di.api. names are bare; di.torq applies process-wide qualification.
+  / callable api for di.torq to register with di.api (init/getapimeta/version are plumbing, omitted)
   :flip `name`public`descrip`params`return!flip(
-    (`upd;       1b; "feed entry point - stamp, log and publish (or buffer) an update";  "[symbol: table; list: data]";      "null");
-    (`subscribe; 1b; "register a subscriber for tables/syms (delegates to di.pubsub)";   "[symbol|list: tables; filters]";   "subscription result");
-    (`endofday;  1b; "flush, notify subscribers, roll the log and advance eod state";    "[]";                               "null");
-    (`getcounts; 1b; "current log/buffer message counts and trading date";              "[]";                               "dict: `i`j`d");
-    (`gettables; 1b; "the tables this tickerplant captures";                            "[]";                               "symbol list: table names"));
+    (`upd;1b;"feed entry point - stamp, log and publish (or buffer) an update";"[symbol table; list data]";"null");
+    (`subscribe;1b;"register a subscriber for tables/syms (delegates to di.pubsub)";"[symbol|list tables; filters]";"subscription result");
+    (`endofday;1b;"flush, notify subscribers, roll the log and advance eod state";"[]";"null");
+    (`getcounts;1b;"current log/buffer message counts and trading date";"[]";"dict `i`j`d");
+    (`gettables;1b;"the tables this tickerplant captures";"[]";"symbol list of table names"));
   };

From a5ac34d442909d2ccfa2a7e708a3bbfe312b38f8 Mon Sep 17 00:00:00 2001
From: ascottDI <ascott@homer.aquaq.co.uk>
Date: Tue, 18 Aug 2026 11:48:52 +0100
Subject: [PATCH 3/7] adding dependent tests that will only pass when both
 TPlog and tickerplant are merged to the same branch

---
 di/tickerplant/test.csv       |  31 ++++++++++
 di/tickerplant/test.q         | 105 ++++++++++++++++++++++++++++++++++
 di/tickerplant/tickerplant.md |  96 +++++++++++++++++++++++++++++++
 3 files changed, 232 insertions(+)
 create mode 100644 di/tickerplant/test.csv
 create mode 100644 di/tickerplant/test.q
 create mode 100644 di/tickerplant/tickerplant.md

diff --git a/di/tickerplant/test.csv b/di/tickerplant/test.csv
new file mode 100644
index 00000000..cc3e8f1b
--- /dev/null
+++ b/di/tickerplant/test.csv
@@ -0,0 +1,31 @@
+action,ms,bytes,lang,code,repeat,minver,comment
+before,0,0,q,tp:use`di.tickerplant,1,,load the module under test
+before,0,0,q,timer:use`di.timer,1,,real di.timer (addjob works without timer.init - no live .z.ts)
+before,0,0,q,ntp:use`di.tplog,1,,di.tplog, to replay a tickerplant-written log
+before,0,0,q,os:use`di.os,1,,os module for portable path resolution
+before,0,0,q,"system ""l "",os.abspath[""di/tickerplant/test.q""]",1,,load fixture helpers
+before,0,0,q,setupfixture[],1,,create the temp fixture root
+comment,,,,,,,module metadata - version and getapimeta
+true,0,0,q,10h=type tp`version,1,1,version is a string
+true,0,0,q,0<count tp`version,1,1,version is non-empty
+true,0,0,q,`name`public`descrip`params`return~cols tp[`getapimeta][],1,1,getapimeta rows carry the registry columns
+true,0,0,q,(asc (key tp) except `init`getapimeta`version)~asc exec name from tp[`getapimeta][],1,1,getapimeta documents exactly the callable exports
+true,0,0,q,not any `init`getapimeta`version in exec name from tp[`getapimeta][],1,1,plumbing and version are not registered in the api
+comment,,,,,,,init dependency validation
+fail,0,0,q,tp.init[(::)],1,1,init rejects a non-dict deps
+fail,0,0,q,tp.init depsnolog[],1,1,init rejects deps without a log key
+fail,0,0,q,tp.init depsonlylog[],1,1,init rejects deps without a timer
+fail,0,0,q,tp.init depsbadtimer[],1,1,init rejects a timer that does not expose addjob
+fail,0,0,q,tp.init depsnoschemas[],1,1,init rejects deps without schemas
+comment,,,,,,,init / lifecycle
+true,0,0,q,testinit[],1,1,init materialises root tables (g# sym) and schedules the timer job
+true,0,0,q,testupdbatch[],1,1,batch upd stamps, buffers and logs
+true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering
+true,0,0,q,testendofday[],1,1,endofday flushes, rolls the log and resets counts
+comment,,,,,,,di.tplog integration
+true,0,0,q,testlogroundtrip[],1,1,a tickerplant-written log replays through di.tplog
+true,0,0,q,testcheckrepaironroll[],1,1,rolling into a corrupt log repairs it via di.tplog.check
+comment,,,,,,,upd input validation
+fail,0,0,q,tp[`upd][1;()],1,1,upd rejects a non-symbol table
+fail,0,0,q,tp[`upd][`nope;()],1,1,upd rejects an unknown table
+after,0,0,q,teardownfixture[],1,,remove the temp fixture root
diff --git a/di/tickerplant/test.q b/di/tickerplant/test.q
new file mode 100644
index 00000000..cd618d6e
--- /dev/null
+++ b/di/tickerplant/test.q
@@ -0,0 +1,105 @@
+/ fixture helpers for di.tickerplant's k4unit tests.
+/ tickerplant orchestrates the real di.pubsub / di.eodtime / di.tplog / di.timer modules with an
+/ injected log, so these tests wire the real modules (tplog resolved via QPATH) and drive a full
+/ init -> upd -> roll cycle. the data tables live at ROOT (the tickerplant owns them and di.pubsub
+/ reads them by name), and replay runs the ROOT upd, so a schema + recorder upd live here at root.
+/ tp (di.tickerplant), timer (di.timer) and ntp (di.tplog) are bound by test.csv's before rows.
+
+base:"/tmp/di_tickerplant_k4unit";
+trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`long$());
+upd:{[t;x] t insert x};
+
+/ capturing logger shared by tickerplant and its dep modules, so any module's output is assertable
+logcap:([]lvl:`symbol$();ctx:`symbol$();msg:());
+caplog:{`info`warn`error!(
+  {[c;m]`logcap insert(`info;c;m);};
+  {[c;m]`logcap insert(`warn;c;m);};
+  {[c;m]`logcap insert(`error;c;m);})};
+
+freshdir:{[sub] dd:base,"/",sub; system"rm -rf ",dd; system"mkdir -p ",dd; dd};
+
+/ deps for init - real timer, capturing log, one trade table, logging into dd, batch flag as given
+mkdeps:{[dd;batch] `log`timer`schemas`logdir`logname`batch!(caplog[];timer;enlist[`trade]!enlist trade;dd;"tp";batch)};
+
+/ fresh init into a clean per-test dir; clears the root table and the log capture first
+freshinit:{[sub;batch]
+  dd:freshdir sub;
+  `trade set 0#trade;
+  `logcap set 0#logcap;
+  tp.init mkdeps[dd;batch];
+  dd};
+
+/ a feed update: sym, price, size - no time, so the tickerplant stamps it
+row:{[s] (s;1.0;100)};
+
+setupfixture:{system"rm -rf ",base; system"mkdir -p ",base;};
+teardownfixture:{system"rm -rf ",base;};
+
+/ deps builders for the init validation fail rows
+depsnolog:{(enlist`x)!enlist 1};
+depsonlylog:{enlist[`log]!enlist caplog[]};
+depsnoschemas:{`log`timer!(caplog[];timer)};
+depsbadtimer:{`log`timer`schemas!(caplog[];(enlist`x)!enlist 1;enlist[`trade]!enlist trade)};
+
+/ =============================================================================
+/ tests (each returns 1b on success)
+/ =============================================================================
+
+/ init materialises the tables at root (g# on sym), schedules the timer job, zeroes the counts
+testinit:{[]
+  freshinit["init";1b];
+  c:tp[`getcounts][];
+  (`g=attr exec sym from trade) and (enlist[`trade]~tp[`gettables][]) and (0=c`i) and (0=c`j)
+    and (-14h=type c`d) and `tickerplant in exec id from timer.getalljobs[]};
+
+/ batch mode: upd stamps, buffers into the root table, and logs (bumping j; i unchanged)
+testupdbatch:{[]
+  freshinit["updb";1b];
+  tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
+  c:tp[`getcounts][];
+  (2=count trade) and (`AAPL`MSFT~exec sym from trade) and (not any null exec time from trade)
+    and (2=c`j) and 0=c`i};
+
+/ zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs
+testzerolatency:{[]
+  freshinit["zl";0b];
+  tp[`upd][`trade;row`AAPL];
+  c:tp[`getcounts][];
+  (0=count trade) and 1=c`j};
+
+/ the tp log tickerplant writes replays through di.tplog - the two modules agree on the log format
+testlogroundtrip:{[]
+  dd:freshinit["rt";1b];
+  tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
+  lf:hsym`$dd,"/tp",string first tp[`getcounts][]`d;
+  `trade set 0#trade; `rcv set 0;
+  `upd set {[t;x] `rcv set rcv+1; t insert x;};
+  n:ntp[`replay] lf;
+  `upd set {[t;x] t insert x;};
+  (2=n) and (2=rcv) and 2=count trade};
+
+/ endofday flushes the buffer, rolls to the next day's log, and resets the counts
+testendofday:{[]
+  dd:freshinit["eod";1b];
+  tp[`upd][`trade;row`AAPL];
+  oldd:first tp[`getcounts][]`d;
+  tp[`endofday][];
+  c:tp[`getcounts][];
+  ((oldd+1)=c`d) and (0=c`i) and (0=c`j) and (0=count trade)
+    and not ()~key hsym`$dd,"/tp",string oldd+1};
+
+/ rolling into a pre-existing CORRUPT log makes openlog repair it via di.tplog.check
+testcheckrepaironroll:{[]
+  dd:freshinit["rep";1b];
+  oldd:first tp[`getcounts][]`d;
+  nl:hsym`$dd,"/tp",string oldd+1;
+  h:hopen nl;
+  h enlist (`upd;`trade;(enlist 2026.08.13D10:00;enlist`AAPL;enlist 1.0;enlist 100));
+  h enlist (`upd;`trade;(enlist 2026.08.13D10:01;enlist`IBM;enlist 2.0;enlist 200));
+  hclose h;
+  nl set (-8)_read1 nl;
+  `logcap set 0#logcap;
+  tp[`endofday][];
+  (not ()~key hsym`$dd,"/tp",(string oldd+1),".good") and `warn in exec lvl from logcap where ctx=`check};
diff --git a/di/tickerplant/tickerplant.md b/di/tickerplant/tickerplant.md
new file mode 100644
index 00000000..e3353c8c
--- /dev/null
+++ b/di/tickerplant/tickerplant.md
@@ -0,0 +1,96 @@
+# di.tickerplant
+
+Core tick-capture for the modular TorQ world: receive updates from feeds, stamp them, write them to a
+tickerplant log for recovery, and publish them to subscribers, rolling the log at end of day. It is
+the modular replacement for TorQ's `code/processes/tickerplant.q`.
+
+It orchestrates three hard dependencies — `di.pubsub` (subscribe/publish), `di.eodtime` (roll timing)
+and `di.tplog` (log check/repair) — with an injected logger and timer.
+
+## Import and init
+
+```q
+tp:use`di.tickerplant
+
+trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`long$())
+tp.init[`log`timer`schemas!(logdep;timerdep;enlist[`trade]!enlist trade)]
+```
+
+`init` takes a single deps dict:
+
+| Key | Required | Description |
+|---|---|---|
+| `log` | yes | `` `info`warn`error `` dict of `{[ctx;msg]}` functions |
+| `timer` | yes | `di.timer`'s exports (must expose `addjob`) |
+| `schemas` | yes | `tablename!schema` dict; the tables to capture |
+| `batch` | no | `1b` (default) buffers and publishes on a timer; `0b` publishes each update immediately |
+| `batchperiod` | no | batch publish interval (timespan, whole seconds; default `0D00:00:01`) |
+| `logdir` | no | directory for the tp log; `""` (default) disables logging |
+| `logname` | no | log filename prefix (default `"tp"`; file is `<logdir>/<logname><date>`) |
+| `subtables` | no | tables offered for subscription (default: all captured tables) |
+| `rolltimezone` / `datatimezone` / `rolltimeoffset` | no | forwarded to `di.eodtime` |
+
+`init` initialises the dependency modules (`di.eodtime`, `di.tplog`, `di.pubsub`), materialises the
+schemas as root tables (applying `` `g# `` to any `sym` column), opens today's log, and schedules a
+single timer job that flushes the buffer (batch mode) and checks for the end-of-day roll. It is
+idempotent — a re-init does not re-add the timer job.
+
+## Root tables and the upd contract
+
+The captured tables live at **root**, not in `.z.m`: a tickerplant owns its tables, feeds insert into
+them, and `di.pubsub` reads them by name, so they cannot be module-local. This is the one deliberate
+root-state exception; all other mutable state is module-local. `di.torq` wires the process's root
+`upd` to `tickerplant.upd` so feeds can publish to it.
+
+## Exported functions
+
+| Function | Signature | Description |
+|---|---|---|
+| `upd` | `[table;data]` | Feed entry point: stamp the update, then buffer+log (batch) or publish+log (zero-latency). |
+| `subscribe` | `[tables;filters]` | Register a subscriber (delegates to `di.pubsub`); called by downstream processes over IPC. |
+| `endofday` | `[]` | Flush the buffer, notify subscribers, roll the tp log, and advance the end-of-day state. |
+| `getcounts` | `[]` | `` `i`j`d `` — messages published (`i`), messages logged (`j`), and the trading date. |
+| `gettables` | `[]` | The tables this tickerplant captures. |
+
+`getapimeta[]` and `version` are also exported, as metadata for `di.torq` / `di.depcheck`.
+
+## Dependencies
+
+Hard (imported via `use` in `init.q`, declared in `deps.q`): `di.pubsub`, `di.eodtime`, `di.tplog`.
+Injected via `init`: `log` and `timer` (both required; validated by `di.depcheck`'s contract check,
+not declared in `deps.q`).
+
+`di.tplog` is used only for check/repair on recovery — when `openlog` finds a pre-existing log it runs
+it through `tplog.check`, repairing a corrupt one. The tickerplant appends to and rolls the log itself
+(opening for append, not replaying), since `di.tplog`'s `open`/`roll` replay through `upd`, which a
+tickerplant must not do to its own log.
+
+## Design notes
+
+- **Batch vs zero-latency.** In batch mode `upd` inserts into the root table and the timer job
+  publishes the accumulated rows every `batchperiod`, then clears them. In zero-latency mode `upd`
+  publishes each update immediately and does not buffer. Both modes log every message.
+- **End of day.** The roll fires when the current time passes `di.eodtime`'s next roll timestamp,
+  checked on every `upd` and on every timer tick. `endofday` flushes, notifies subscribers, rolls the
+  log to the next day, and refreshes the roll time and data-timestamp offset from `di.eodtime`.
+- **No `di.handlers` dependency.** Subscriber-disconnect cleanup is handled by `di.pubsub`'s own
+  `.z.pc`. (`di.pubsub` should migrate to `di.handlers` so `.z.*` is not assigned outside the central
+  registry — tracked separately, out of scope here.)
+
+## Testing
+
+`test.csv` / `test.q` (k4unit) run against the **real** `di.pubsub`, `di.eodtime`, `di.tplog` and
+`di.timer` — no dependencies are mocked (the timer is used without `init`, so no live `.z.ts`, and its
+job is exercised through `endofday`). A capturing logger is shared across the modules so their output
+is assertable.
+
+Coverage: the metadata/version contract; strict `init` dependency validation (a `fail` row per guard);
+init materialising the root tables and scheduling the timer job; batch and zero-latency `upd`;
+`endofday` flushing and rolling; `upd` input validation; and the two `di.tplog` integration points — a
+tickerplant-written log replaying through `di.tplog`, and rolling into a corrupt log repairing it via
+`tplog.check`.
+
+```q
+k4unit:use`di.k4unit
+k4unit.moduletest`di.tickerplant
+```

From a07ac5ccc7a62a0245178bc17b86a39c764e8a60 Mon Sep 17 00:00:00 2001
From: ascottDI <ascott@homer.aquaq.co.uk>
Date: Tue, 18 Aug 2026 12:12:56 +0100
Subject: [PATCH 4/7] updates following automated reviewer comments

---
 di/tickerplant/test.csv      |  2 ++
 di/tickerplant/test.q        | 18 ++++++++++++++++++
 di/tickerplant/tickerplant.q |  5 ++++-
 3 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/di/tickerplant/test.csv b/di/tickerplant/test.csv
index cc3e8f1b..86ef50f6 100644
--- a/di/tickerplant/test.csv
+++ b/di/tickerplant/test.csv
@@ -20,8 +20,10 @@ fail,0,0,q,tp.init depsnoschemas[],1,1,init rejects deps without schemas
 comment,,,,,,,init / lifecycle
 true,0,0,q,testinit[],1,1,init materialises root tables (g# sym) and schedules the timer job
 true,0,0,q,testupdbatch[],1,1,batch upd stamps, buffers and logs
+true,0,0,q,testemptyupd[],1,1,an empty update is a no-op (no throw)
 true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering
 true,0,0,q,testendofday[],1,1,endofday flushes, rolls the log and resets counts
+true,0,0,q,testreinitnoleak[],1,1,re-init closes the old log handle (no fd leak)
 comment,,,,,,,di.tplog integration
 true,0,0,q,testlogroundtrip[],1,1,a tickerplant-written log replays through di.tplog
 true,0,0,q,testcheckrepaironroll[],1,1,rolling into a corrupt log repairs it via di.tplog.check
diff --git a/di/tickerplant/test.q b/di/tickerplant/test.q
index cd618d6e..7dd87fd0 100644
--- a/di/tickerplant/test.q
+++ b/di/tickerplant/test.q
@@ -61,6 +61,13 @@ testupdbatch:{[]
   (2=count trade) and (`AAPL`MSFT~exec sym from trade) and (not any null exec time from trade)
     and (2=c`j) and 0=c`i};
 
+/ an empty update is a no-op: no throw, nothing buffered, nothing logged
+testemptyupd:{[]
+  freshinit["empty";1b];
+  tp[`upd][`trade;()];
+  c:tp[`getcounts][];
+  (0=count trade) and 0=c`j};
+
 / zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs
 testzerolatency:{[]
   freshinit["zl";0b];
@@ -68,6 +75,17 @@ testzerolatency:{[]
   c:tp[`getcounts][];
   (0=count trade) and 1=c`j};
 
+/ re-init is safe: it closes the previous log handle instead of leaking the descriptor, and logging
+/ keeps working. fd count (linux /proc, as the suite is already unix-coupled) must not grow.
+testreinitnoleak:{[]
+  fddir:"/proc/",(string .z.i),"/fd";
+  freshinit["reinit";1b];
+  b:"J"$first system"ls ",fddir," | wc -l";
+  freshinit["reinit";1b]; freshinit["reinit";1b]; freshinit["reinit";1b];
+  a:"J"$first system"ls ",fddir," | wc -l";
+  tp[`upd][`trade;row`AAPL];
+  (a=b) and 1=tp[`getcounts][]`j};
+
 / the tp log tickerplant writes replays through di.tplog - the two modules agree on the log format
 testlogroundtrip:{[]
   dd:freshinit["rt";1b];
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
index 28be2683..f976f197 100644
--- a/di/tickerplant/tickerplant.q
+++ b/di/tickerplant/tickerplant.q
@@ -137,9 +137,11 @@ init:{[deps]
   tplog.init[enlist[`log]!enlist deps`log];
   pubsub.setsubtables[$[`subtables in key deps;deps`subtables;.z.m.tabs]];
   pubsub.init[];
-  / date, counts, and the tp log for today
+  / date, counts, and the tp log for today. close a handle held from a previous init before
+  / reopening, so a re-init does not leak the old file descriptor (rolllog closes on its own path)
   .z.m.d:eodtime.getd[];
   .z.m.i:.z.m.j:0;
+  if[`logfile in key .z.m;if[.z.m.logfile>0i;hclose .z.m.logfile]];
   .z.m.logfile:openlog .z.m.d;
   / schedule the timer job that flushes the buffer (batch) and checks the roll; mode 1 = fixed period.
   / guarded so a re-init does not re-add (di.timer.addjob throws on a duplicate id); tick reads
@@ -155,6 +157,7 @@ upd:{[t;x]
   / t is the table name, x the column data. wired to root `upd` by di.torq so feeds can call it.
   if[not -11h=type t;raiseerror[`upd;"table must be a symbol"]];
   if[not t in .z.m.tabs;raiseerror[`upd;"unknown table ",string t]];
+  if[not count x;:()];
   rollcheck .z.p;
   x:stamp x;
   if[.z.m.batch;t insert x;writelog[t;x]];

From 87dbd12f8f3a428b18e5a10888458146d1cc471d Mon Sep 17 00:00:00 2001
From: ascottDI <ascott@homer.aquaq.co.uk>
Date: Tue, 18 Aug 2026 14:46:26 +0100
Subject: [PATCH 5/7] changes following auto reviewer comments

---
 di/tickerplant/test.csv      |  2 ++
 di/tickerplant/test.q        | 12 ++++++++++--
 di/tickerplant/tickerplant.q |  3 ++-
 3 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/di/tickerplant/test.csv b/di/tickerplant/test.csv
index 86ef50f6..c580f70a 100644
--- a/di/tickerplant/test.csv
+++ b/di/tickerplant/test.csv
@@ -2,6 +2,7 @@ action,ms,bytes,lang,code,repeat,minver,comment
 before,0,0,q,tp:use`di.tickerplant,1,,load the module under test
 before,0,0,q,timer:use`di.timer,1,,real di.timer (addjob works without timer.init - no live .z.ts)
 before,0,0,q,ntp:use`di.tplog,1,,di.tplog, to replay a tickerplant-written log
+before,0,0,q,eod:use`di.eodtime,1,,di.eodtime, to force a roll due in a test
 before,0,0,q,os:use`di.os,1,,os module for portable path resolution
 before,0,0,q,"system ""l "",os.abspath[""di/tickerplant/test.q""]",1,,load fixture helpers
 before,0,0,q,setupfixture[],1,,create the temp fixture root
@@ -21,6 +22,7 @@ comment,,,,,,,init / lifecycle
 true,0,0,q,testinit[],1,1,init materialises root tables (g# sym) and schedules the timer job
 true,0,0,q,testupdbatch[],1,1,batch upd stamps, buffers and logs
 true,0,0,q,testemptyupd[],1,1,an empty update is a no-op (no throw)
+true,0,0,q,testemptyupdtriggersroll[],1,1,an empty update still triggers an overdue roll
 true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering
 true,0,0,q,testendofday[],1,1,endofday flushes, rolls the log and resets counts
 true,0,0,q,testreinitnoleak[],1,1,re-init closes the old log handle (no fd leak)
diff --git a/di/tickerplant/test.q b/di/tickerplant/test.q
index 7dd87fd0..d44501e0 100644
--- a/di/tickerplant/test.q
+++ b/di/tickerplant/test.q
@@ -68,6 +68,14 @@ testemptyupd:{[]
   c:tp[`getcounts][];
   (0=count trade) and 0=c`j};
 
+/ an empty update still triggers an overdue roll (the roll check runs before the empty-data skip)
+testemptyupdtriggersroll:{[]
+  freshinit["emptyroll";1b];
+  oldd:first tp[`getcounts][]`d;
+  eod.setnextroll .z.p-0D01:00:00;
+  tp[`upd][`trade;()];
+  (oldd+1)=first tp[`getcounts][]`d};
+
 / zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs
 testzerolatency:{[]
   freshinit["zl";0b];
@@ -80,8 +88,8 @@ testzerolatency:{[]
 testreinitnoleak:{[]
   fddir:"/proc/",(string .z.i),"/fd";
   freshinit["reinit";1b];
-  b:"J"$first system"ls ",fddir," | wc -l";
-  freshinit["reinit";1b]; freshinit["reinit";1b]; freshinit["reinit";1b];
+  b:"J"$first system"ls ",fddir," | wc -l";   / baseline AFTER the first init - one log handle is open
+  freshinit["reinit";1b]; freshinit["reinit";1b]; freshinit["reinit";1b];   / re-init must not add fds
   a:"J"$first system"ls ",fddir," | wc -l";
   tp[`upd][`trade;row`AAPL];
   (a=b) and 1=tp[`getcounts][]`j};
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
index f976f197..27f28b7e 100644
--- a/di/tickerplant/tickerplant.q
+++ b/di/tickerplant/tickerplant.q
@@ -157,8 +157,9 @@ upd:{[t;x]
   / t is the table name, x the column data. wired to root `upd` by di.torq so feeds can call it.
   if[not -11h=type t;raiseerror[`upd;"table must be a symbol"]];
   if[not t in .z.m.tabs;raiseerror[`upd;"unknown table ",string t]];
-  if[not count x;:()];
   rollcheck .z.p;
+  / empty update carries no data - nothing to stamp/log, but the roll check above still runs
+  if[not count x;:()];
   x:stamp x;
   if[.z.m.batch;t insert x;writelog[t;x]];
   if[not .z.m.batch;publishrows[t;x];writelog[t;x]];

From 44ec616d58fae72996ddbeeb8254a089b0fd48ee Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Wed, 19 Aug 2026 16:29:02 +0100
Subject: [PATCH 6/7] di.tickerplant: speak the subdetails protocol, fix log
 path, counters and re-init

Publish subdetails and tablelist at root so di.subscriptions can attach.
A standard TorQ tickerplant never had these - only chainedtp.q and
segmentedtickerplant.q define subdetails - so this is a deliberate
capability addition rather than restored parity, recorded in the
design-divergences section of tickerplant.md.

- keep the tp log path in .z.m.logpath; .z.m.logfile stays the handle,
  so the path is no longer overwritten and can be reported to subscribers
- advance .z.m.i in zero-latency mode, as chainedtp.q's tickpub does;
  without it every subscriber was told to replay nothing
- report the PUBLISHED watermark i in logfilelist, not the logged total
  j: in batch mode a row logged but not yet flushed is still buffered and
  goes out at the next tick, so j would replay it AND deliver it again.
  TorQ sends .u.i for the same reason (chainedtp.q) and kdb+tick's r.q
  replays with .u`i
- seed runtime state only on a fresh init, so a re-init cannot rewind the
  trading date, zero the counts a subscriber replays against, or reopen
  the log; table materialisation follows the same rule, since re-running
  the schema over a live tickerplant discarded buffered rows
- track per-table published rowcounts for the protocol's rowcounts field
- fail loud on a missing, unreadable or empty VERSION
- guard the cold-path exports with requireinit; upd stays unguarded as
  the per-message hot path, and a test pins which side each is on

Adds test_integration.csv: a real di.subscriptions driven against a real
di.tickerplant over IPC, covering both batch and zero-latency modes plus
the VERSION guards. It needs di.tplog with init (feature-tplog) and
di.subscriptions on the same branch, as a5ac34d notes for the existing
dependent tests.
---
 di/tickerplant/init.q               |  12 +-
 di/tickerplant/test.csv             |  24 ++-
 di/tickerplant/test.q               | 167 +++++++++++++++--
 di/tickerplant/test_integration.csv | 137 ++++++++++++++
 di/tickerplant/tickerplant.md       | 194 ++++++++++++++++++--
 di/tickerplant/tickerplant.q        | 268 ++++++++++++++++++++++++----
 6 files changed, 728 insertions(+), 74 deletions(-)
 create mode 100644 di/tickerplant/test_integration.csv

diff --git a/di/tickerplant/init.q b/di/tickerplant/init.q
index 4a796242..b373f068 100644
--- a/di/tickerplant/init.q
+++ b/di/tickerplant/init.q
@@ -6,6 +6,12 @@ tplog:use`di.tplog
 \l ::tickerplant.q
 / module version, read from the VERSION file (one plain-text file to bump per release). read
 / module-relative at load (`:::` resolves to di/tickerplant) and BEFORE export, since export:([...])
-/ evaluates each name; version stays in the export so di.depcheck reads it from the export dict
-version:trim first read0`:::VERSION
-export:([init;upd;subscribe;endofday;getcounts;gettables;getapimeta;version])
+/ evaluates each name; version stays in the export so di.depcheck reads it from the export dict.
+/ trim, and fail LOUD on a missing/unreadable/empty VERSION, rather than a bare `first read0`:
+/ a raw OS error names no module, and an empty or whitespace-padded value is worse than an error -
+/ di.depcheck compares versions as STRINGS, so padding silently breaks the comparison and an empty
+/ value reads as "exports no version", failing every dependent module's check for a reason that
+/ points nowhere near the real cause. same shape as di.rdb, di.dbwrite and di.eodtime
+version:@[{trim first read0 x};`:::VERSION;{'"di.tickerplant: VERSION file missing or unreadable"}];
+if[0=count version;'"di.tickerplant: VERSION file is empty"];
+export:([init;teardown;upd;subscribe;subdetails;tablelist;endofday;getcounts;gettables;getapimeta;version])
diff --git a/di/tickerplant/test.csv b/di/tickerplant/test.csv
index c580f70a..bdd67a9e 100644
--- a/di/tickerplant/test.csv
+++ b/di/tickerplant/test.csv
@@ -1,8 +1,9 @@
 action,ms,bytes,lang,code,repeat,minver,comment
 before,0,0,q,tp:use`di.tickerplant,1,,load the module under test
 before,0,0,q,timer:use`di.timer,1,,real di.timer (addjob works without timer.init - no live .z.ts)
-before,0,0,q,ntp:use`di.tplog,1,,di.tplog, to replay a tickerplant-written log
-before,0,0,q,eod:use`di.eodtime,1,,di.eodtime, to force a roll due in a test
+before,0,0,q,ntp:use`di.tplog,1,,"di.tplog, to replay a tickerplant-written log"
+before,0,0,q,eod:use`di.eodtime,1,,"di.eodtime, to force a roll due in a test"
+before,0,0,q,ps:use`di.pubsub,1,,"di.pubsub, to release the self-subscription a subdetails test leaves on handle 0"
 before,0,0,q,os:use`di.os,1,,os module for portable path resolution
 before,0,0,q,"system ""l "",os.abspath[""di/tickerplant/test.q""]",1,,load fixture helpers
 before,0,0,q,setupfixture[],1,,create the temp fixture root
@@ -17,18 +18,29 @@ fail,0,0,q,tp.init[(::)],1,1,init rejects a non-dict deps
 fail,0,0,q,tp.init depsnolog[],1,1,init rejects deps without a log key
 fail,0,0,q,tp.init depsonlylog[],1,1,init rejects deps without a timer
 fail,0,0,q,tp.init depsbadtimer[],1,1,init rejects a timer that does not expose addjob
+fail,0,0,q,tp.init depsnodeletejobs[],1,1,init rejects a timer without deletejobs - teardown needs it
 fail,0,0,q,tp.init depsnoschemas[],1,1,init rejects deps without schemas
+true,0,0,q,"0<count (@[{tp.init depsnodeletejobs[];""""};::;{[e] e}]) ss ""must expose deletejobs""",1,1,the deletejobs guard fails for its OWN reason - a fail row alone proves only that something threw
 comment,,,,,,,init / lifecycle
 true,0,0,q,testinit[],1,1,init materialises root tables (g# sym) and schedules the timer job
-true,0,0,q,testupdbatch[],1,1,batch upd stamps, buffers and logs
+true,0,0,q,testupdbatch[],1,1,"batch upd stamps, buffers and logs; i lags j until the flush"
 true,0,0,q,testemptyupd[],1,1,an empty update is a no-op (no throw)
 true,0,0,q,testemptyupdtriggersroll[],1,1,an empty update still triggers an overdue roll
-true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering
-true,0,0,q,testendofday[],1,1,endofday flushes, rolls the log and resets counts
-true,0,0,q,testreinitnoleak[],1,1,re-init closes the old log handle (no fd leak)
+true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering and advances i with j
+true,0,0,q,testendofday[],1,1,"endofday flushes, rolls the log and resets counts"
+true,0,0,q,testreinitpreservesstate[],1,1,a re-init preserves the counts / buffer / log handle and leaks no fd
 comment,,,,,,,di.tplog integration
 true,0,0,q,testlogroundtrip[],1,1,a tickerplant-written log replays through di.tplog
 true,0,0,q,testcheckrepaironroll[],1,1,rolling into a corrupt log repairs it via di.tplog.check
+comment,,,,,,,the subdetails / tablelist protocol di.subscriptions speaks
+true,0,0,q,testsubdetails[],1,1,subdetails returns all four required keys in the shapes di.subscriptions validates
+true,0,0,q,testsubdetailscountispublished[],1,1,the logfilelist message count is the published watermark i - not the logged total j
+true,0,0,q,testsubdetailsnolog[],1,1,logfilelist is empty when logging is disabled
+true,0,0,q,testsubdetailsunknown[],1,1,subdetails names the unpublished table it was asked for and logs the error
+true,0,0,q,testtablelist[],1,1,"tablelist accepts the one argument di.subscriptions sends, (`tablelist;`)"
+true,0,0,q,testrootandteardown[],1,1,the protocol is published at root and teardown gives both names back
+true,0,0,q,testreinitafterteardown[],1,1,a re-init after teardown re-publishes the protocol and re-schedules the job
+true,0,0,q,testrequireinit[],1,1,every callable export except upd names the init guard; upd is the documented hot-path exception
 comment,,,,,,,upd input validation
 fail,0,0,q,tp[`upd][1;()],1,1,upd rejects a non-symbol table
 fail,0,0,q,tp[`upd][`nope;()],1,1,upd rejects an unknown table
diff --git a/di/tickerplant/test.q b/di/tickerplant/test.q
index d44501e0..7611166c 100644
--- a/di/tickerplant/test.q
+++ b/di/tickerplant/test.q
@@ -3,7 +3,8 @@
 / injected log, so these tests wire the real modules (tplog resolved via QPATH) and drive a full
 / init -> upd -> roll cycle. the data tables live at ROOT (the tickerplant owns them and di.pubsub
 / reads them by name), and replay runs the ROOT upd, so a schema + recorder upd live here at root.
-/ tp (di.tickerplant), timer (di.timer) and ntp (di.tplog) are bound by test.csv's before rows.
+/ tp (di.tickerplant), timer (di.timer), ntp (di.tplog), eod (di.eodtime) and ps (di.pubsub) are
+/ bound by test.csv's before rows.
 
 base:"/tmp/di_tickerplant_k4unit";
 trade:([]time:`timestamp$();sym:`symbol$();price:`float$();size:`long$());
@@ -21,45 +22,71 @@ freshdir:{[sub] dd:base,"/",sub; system"rm -rf ",dd; system"mkdir -p ",dd; dd};
 / deps for init - real timer, capturing log, one trade table, logging into dd, batch flag as given
 mkdeps:{[dd;batch] `log`timer`schemas`logdir`logname`batch!(caplog[];timer;enlist[`trade]!enlist trade;dd;"tp";batch)};
 
+/ force the module back to a pre-init state, so the NEXT init is a fresh one.
+/ init deliberately seeds runtime state (date, counts, open log) only on the first call - see
+/ tickerplant.md - so a suite that wants a genuinely clean tickerplant per test has to clear the name
+/ initialised[] probes. done here rather than by exporting a reset verb: a state-wiping function that
+/ exists only for tests does not belong in the public api. also drops the self-subscription a
+/ subdetails test leaves behind on handle 0, so a later flush does not publish into it
+resetmodule:{[]
+  if[`schemas in key `.m.di.0tickerplant;
+    tp[`teardown][];
+    @[{if[.m.di.0tickerplant.logfile>0i;hclose .m.di.0tickerplant.logfile]};::;{[e] :(::)}];
+    ![`.m.di.0tickerplant;();0b;`schemas`scheduled]];
+  @[ps`closesub;0i;{[e] :(::)}];
+  };
+
 / fresh init into a clean per-test dir; clears the root table and the log capture first
 freshinit:{[sub;batch]
+  resetmodule[];
   dd:freshdir sub;
   `trade set 0#trade;
   `logcap set 0#logcap;
   tp.init mkdeps[dd;batch];
   dd};
 
+/ drive one batch flush without the roll check tick[] also does - the timer is never started here
+flushbuffer:{[] .m.di.0tickerplant.publishbuffer[]};
+
 / a feed update: sym, price, size - no time, so the tickerplant stamps it
 row:{[s] (s;1.0;100)};
 
 setupfixture:{system"rm -rf ",base; system"mkdir -p ",base;};
-teardownfixture:{system"rm -rf ",base;};
+teardownfixture:{resetmodule[]; system"rm -rf ",base;};
 
 / deps builders for the init validation fail rows
 depsnolog:{(enlist`x)!enlist 1};
 depsonlylog:{enlist[`log]!enlist caplog[]};
 depsnoschemas:{`log`timer!(caplog[];timer)};
 depsbadtimer:{`log`timer`schemas!(caplog[];(enlist`x)!enlist 1;enlist[`trade]!enlist trade)};
+depsnodeletejobs:{`log`timer`schemas!(caplog[];`deletejobs _ timer;enlist[`trade]!enlist trade)};
 
 / =============================================================================
 / tests (each returns 1b on success)
 / =============================================================================
 
-/ init materialises the tables at root (g# on sym), schedules the timer job, zeroes the counts
+/ init materialises the tables at root (g# on sym), publishes the subscription protocol at root,
+/ schedules the timer job and zeroes the counts
 testinit:{[]
   freshinit["init";1b];
   c:tp[`getcounts][];
   (`g=attr exec sym from trade) and (enlist[`trade]~tp[`gettables][]) and (0=c`i) and (0=c`j)
-    and (-14h=type c`d) and `tickerplant in exec id from timer.getalljobs[]};
+    and (-14h=type c`d) and (`tickerplant in exec id from timer.getalljobs[])
+    and all `subdetails`tablelist in key `.};
 
-/ batch mode: upd stamps, buffers into the root table, and logs (bumping j; i unchanged)
+/ batch mode: upd stamps, buffers into the root table and logs (bumping j). i is the PUBLISHED
+/ watermark and must NOT move while the rows are still buffered - it catches up at the flush
 testupdbatch:{[]
   freshinit["updb";1b];
   tp[`upd][`trade;row`AAPL];
   tp[`upd][`trade;row`MSFT];
   c:tp[`getcounts][];
-  (2=count trade) and (`AAPL`MSFT~exec sym from trade) and (not any null exec time from trade)
-    and (2=c`j) and 0=c`i};
+  n:count trade;
+  s:exec sym from trade;
+  stamped:not any null exec time from trade;
+  flushbuffer[];
+  a:tp[`getcounts][];
+  (2=n) and (`AAPL`MSFT~s) and stamped and (2=c`j) and (0=c`i) and (2=a`i) and 2=a`j};
 
 / an empty update is a no-op: no throw, nothing buffered, nothing logged
 testemptyupd:{[]
@@ -76,23 +103,33 @@ testemptyupdtriggersroll:{[]
   tp[`upd][`trade;()];
   (oldd+1)=first tp[`getcounts][]`d};
 
-/ zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs
+/ zero-latency mode: upd publishes immediately, does NOT buffer into the root table, still logs - and
+/ i tracks j, because every logged message has already gone out. without that a zero-latency
+/ tickerplant tells every subscriber to replay nothing
 testzerolatency:{[]
   freshinit["zl";0b];
   tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
   c:tp[`getcounts][];
-  (0=count trade) and 1=c`j};
+  (0=count trade) and (2=c`j) and 2=c`i};
 
-/ re-init is safe: it closes the previous log handle instead of leaking the descriptor, and logging
-/ keeps working. fd count (linux /proc, as the suite is already unix-coupled) must not grow.
-testreinitnoleak:{[]
+/ a re-init refreshes config but must NOT rewind runtime state: the date, the message counts a
+/ subscriber replays against, the buffered rows, or the open log handle. it must not leak a
+/ descriptor either, which it cannot now that it no longer reopens the log
+testreinitpreservesstate:{[]
   fddir:"/proc/",(string .z.i),"/fd";
-  freshinit["reinit";1b];
-  b:"J"$first system"ls ",fddir," | wc -l";   / baseline AFTER the first init - one log handle is open
-  freshinit["reinit";1b]; freshinit["reinit";1b]; freshinit["reinit";1b];   / re-init must not add fds
-  a:"J"$first system"ls ",fddir," | wc -l";
+  dd:freshinit["reinit";1b];
   tp[`upd][`trade;row`AAPL];
-  (a=b) and 1=tp[`getcounts][]`j};
+  tp[`upd][`trade;row`MSFT];
+  c0:tp[`getcounts][];
+  lp0:.m.di.0tickerplant.logpath;
+  b:"J"$first system"ls ",fddir," | wc -l";
+  tp.init mkdeps[dd;1b]; tp.init mkdeps[dd;1b]; tp.init mkdeps[dd;1b];
+  a:"J"$first system"ls ",fddir," | wc -l";
+  c1:tp[`getcounts][];
+  tp[`upd][`trade;row`IBM];
+  (c0~c1) and (a=b) and (lp0~.m.di.0tickerplant.logpath) and (3=count trade)
+    and 3=tp[`getcounts][]`j};
 
 / the tp log tickerplant writes replays through di.tplog - the two modules agree on the log format
 testlogroundtrip:{[]
@@ -129,3 +166,99 @@ testcheckrepaironroll:{[]
   `logcap set 0#logcap;
   tp[`endofday][];
   (not ()~key hsym`$dd,"/tp",(string oldd+1),".good") and `warn in exec lvl from logcap where ctx=`check};
+
+/ =============================================================================
+/ the subscription protocol di.subscriptions speaks
+/ =============================================================================
+
+/ subdetails carries all four required keys, in the shapes di.subscriptions' guards check:
+/ schemalist as (tablename;schema) pairs, logfilelist as (integer count;symbol file) pairs,
+/ rowcounts as a dict keyed by table, date as a date
+testsubdetails:{[]
+  freshinit["subd";1b];
+  tp[`upd][`trade;row`AAPL];
+  flushbuffer[];
+  d:tp[`subdetails][`;`];
+  sl:d`schemalist;
+  lfl:d`logfilelist;
+  (all `schemalist`logfilelist`rowcounts`date in key d)
+    and (1=count sl) and (`trade~first first sl) and (.Q.qt last first sl)
+    and (1=count lfl) and ((type first first lfl) in -7 -6h) and (-11h=type last first lfl)
+    and (99h=type d`rowcounts) and (1=(d`rowcounts)`trade) and (-14h=type d`date)};
+
+/ the message count is the PUBLISHED watermark i, not the logged total j: rows logged but not yet
+/ flushed are still in the buffer and go out to a NEW subscriber at the next flush, so reporting j
+/ would have it replay them from the log AND receive them again
+testsubdetailscountispublished:{[]
+  freshinit["subdi";1b];
+  tp[`upd][`trade;row`AAPL];
+  tp[`upd][`trade;row`MSFT];
+  flushbuffer[];
+  tp[`upd][`trade;row`IBM];
+  c:tp[`getcounts][];
+  n:first first tp[`subdetails][`;`]`logfilelist;
+  (2=c`i) and (3=c`j) and 2=n};
+
+/ with logging disabled there is no log to replay, so logfilelist is EMPTY rather than naming a file
+testsubdetailsnolog:{[]
+  resetmodule[];
+  `trade set 0#trade;
+  `logcap set 0#logcap;
+  tp.init[`log`timer`schemas`logdir!(caplog[];timer;enlist[`trade]!enlist trade;"")];
+  d:tp[`subdetails][`;`];
+  (()~d`logfilelist) and 0=tp[`getcounts][]`j};
+
+/ subdetails for a table this tickerplant does not publish fails with a message that NAMES the table
+/ and logs it. a bare `fail` row would pass on any throw at all, including an unrelated one
+testsubdetailsunknown:{[]
+  freshinit["subdunk";1b];
+  e:string @[{tp[`subdetails][x;`];`NOTHROW};`nope;{[e] `$e}];
+  (0<count e ss "no requested table is published") and (0<count e ss "nope")
+    and `error in exec lvl from logcap where ctx=`subdetails};
+
+/ tablelist must accept the ONE argument di.subscriptions sends - (`tablelist;`). a niladic form
+/ throws 'rank, which di.subscriptions catches and silently downgrades to asking for `
+testtablelist:{[]
+  freshinit["tl";1b];
+  (enlist[`trade]~tp[`tablelist][`]) and enlist[`trade]~tp[`tablelist][`trade]};
+
+/ both protocol names live at ROOT, where the default .z.pg reaches them for an IPC caller, and are
+/ callable in the message shape an IPC caller sends. teardown gives them back and drops the timer job
+testrootandteardown:{[]
+  freshinit["root";1b];
+  installed:all `subdetails`tablelist in key `.;
+  callable:enlist[`trade]~value (`tablelist;`);
+  tp[`teardown][];
+  installed and callable and (not any `subdetails`tablelist in key `.)
+    and not `tickerplant in exec id from timer.getalljobs[]};
+
+/ a re-init after teardown re-publishes the protocol and re-schedules the job - the flag is what
+/ makes that work, since a re-init is not fresh and di.timer throws on a duplicate job id
+testreinitafterteardown:{[]
+  dd:freshinit["rat";1b];
+  tp[`teardown][];
+  tp.init mkdeps[dd;1b];
+  (all `subdetails`tablelist in key `.) and `tickerplant in exec id from timer.getalljobs[]};
+
+/ every callable export except upd reports the module's own "init must be called" message rather than
+/ a bare unresolved-name error. upd is the documented exception - it is the per-message hot path and
+/ carries no guard, so this pins WHICH functions are guarded rather than asserting they all are.
+/ NB the calls are wrapped in NILADIC lambdas: tp[`subscribe][`;`] written bare would be evaluated
+/ while the list is built, throwing there instead of inside the protected apply
+testrequireinit:{[]
+  resetmodule[];
+  / return the ERROR STRING, or "NOTHROW" - never `string` of the result. a successful call can
+  / return a symbol vector or (::), and stringifying either yields a LIST of strings, which then
+  / throws 'type inside ss rather than failing the assertion (measured)
+  msg:{[f] :@[{[g] g[]; "NOTHROW"};f;{[e] e}]};
+  calls:({[] tp[`teardown][]};{[] tp[`subscribe][`;`]};{[] tp[`subdetails][`;`]};
+    {[] tp[`tablelist][`]};{[] tp[`endofday][]};{[] tp[`getcounts][]};{[] tp[`gettables][]});
+  saidinit:all {[m] 0<count m ss "init must be called before any other function"} each msg each calls;
+  / upd is deliberately unguarded, so it does NOT report the init guard. NB resetmodule clears only
+  / the names initialised[] probes, so .z.m.tabs survives and upd may simply return rather than
+  / throw - either outcome is correct here, and only the absence of the guard message is asserted
+  unguarded:0=count (msg {[] tp[`upd][`trade;()]}) ss "init must be called";
+  / leave the module INITIALISED again: the upd validation rows that follow are `fail` rows, and an
+  / uninitialised module makes them throw on an unset .z.m.logerr - passing for the wrong reason
+  freshinit["reqinit";1b];
+  saidinit and unguarded};
diff --git a/di/tickerplant/test_integration.csv b/di/tickerplant/test_integration.csv
new file mode 100644
index 00000000..5eb06e7e
--- /dev/null
+++ b/di/tickerplant/test_integration.csv
@@ -0,0 +1,137 @@
+action,ms,bytes,lang,code,repeat,minver,comment
+comment,,,,,,,integration suite - proves di.tickerplant speaks the subdetails/tablelist protocol di.subscriptions
+comment,,,,,,,"requires, by driving a REAL di.subscriptions against a REAL di.tickerplant over genuine IPC."
+comment,,,,,,,"the tickerplant is the spawned PEER and di.subscriptions runs here, not the other way round: a q"
+comment,,,,,,,process blocked in a sync call does not accept a new inbound connection (measured - the subscriber
+comment,,,,,,,"dialling back into a busy parent times out), so the side that DIALS has to be the side that drives."
+comment,,,,,,,that is also how di/rdb and di/subscriptions arrange their own suites.
+comment,,,,,,,"kept separate because moduletest only ever loads test.csv, so leaving these rows there would make"
+comment,,,,,,,every unit run spawn processes and bind ports.
+comment,,,,,,,RUN IT IN A FRESH q SESSION: after moduletest the unit rows are still loaded and would re-run
+comment,,,,,,,"against dirty module state, reporting phantom failures."
+comment,,,,,,,needs a q binary reachable via QHOME - each peer is launched with $QHOME/bin/q
+comment,,,,,,,"fixtures - the subscriber side, which is the code di.torq will eventually contain"
+before,0,0,q,sub:use`di.subscriptions,1,1,the consumer whose protocol this suite exists to satisfy
+before,0,0,q,hnd:use`di.handlers,1,1,di.subscriptions needs a handlers dep for its .z.pc observer
+before,0,0,q,caprows:([]lvl:`symbol$();ctx:`symbol$();msg:()),1,1,capturing log sink
+before,0,0,q,caplog:`info`warn`error!({[c;m] `caprows insert (`info;c;m)};{[c;m] `caprows insert (`warn;c;m)};{[c;m] `caprows insert (`error;c;m)}),1,1,the binary {[c;m]} log dep contract
+before,0,0,q,hnd.init[enlist[`log]!enlist caplog],1,1,wire di.handlers
+before,0,0,q,sub.init[`log`handlers!(caplog;`register`remove!(hnd`register;hnd`remove))],1,1,"wire di.subscriptions - one dict, the call shape di.torq uses"
+before,0,0,q,"upd:{[t;x] c:cols value t; @[`.;t;,;$[98h=type x;x;99h=type x;flip x;0>type first x;flip c!enlist each x;flip c!x]]}",1,1,"a root upd, which the -11! replay drives. di.subscriptions refuses to replay without one. it normalises the same payload shapes di.subscriptions' own payloadtable does - a table, a column dict, a list of atoms, or a list of column vectors"
+before,0,0,q,"tmpdir:{[nm] b:getenv`TMPDIR; if[0=count b;b:""/tmp""]; :b,""/"",nm,string .z.i}",1,1,honours TMPDIR and appends the pid so two runs cannot collide
+before,0,0,q,"BASE:tmpdir""ditickerplantintegration""",1,1,fixture: BASE
+before,0,0,q,"system""rm -rf "",BASE",1,1,clean slate
+before,0,0,q,"system""mkdir -p "",BASE",1,1,create the working directory
+before,0,0,q,"@[system;""pkill -f '"",BASE,""' 2>/dev/null || true"";{[e] :(::)}]",1,1,kill any peer left behind by an aborted earlier run
+comment,,,,,,,NO PORT NUMBER APPEARS IN THIS SUITE. each peer is started with -p 0W so the OS assigns a free
+comment,,,,,,,port and the peer writes it to a file we read back. scanning from a hardcoded base is not portable
+comment,,,,,,,and still races - a port held by a blocked non-q process reads as free
+before,0,0,q,"readport:{[f;timeoutms] deadline:.z.p+`timespan$1000000*timeoutms; p:0N; while[(.z.p<deadline) and null p; if[count key hsym`$f;p:""J""$first read0 hsym`$f]; if[null p;system""sleep 0.05""]]; :p}",1,1,wait for a peer to report the port the OS gave it
+comment,,,,,,,"the tickerplant peer source, one complete line per row - a split literal would become several expressions"
+before,0,0,q,PBODY:(),1,1,the peer body
+before,0,0,q,"PBODY,:enlist ""tp:use`di.tickerplant;""",1,1,"the module under test - the REAL di.tickerplant, not a hand-rolled subdetails stand-in"
+before,0,0,q,"PBODY,:enlist ""timer:use`di.timer;""",1,1,"the real timer - addjob works without timer.init, so no live .z.ts fires here"
+before,0,0,q,"PBODY,:enlist ""caprows:([]lvl:`symbol$();ctx:`symbol$();msg:());""",1,1,"capturing log sink, so the tickerplant's own errors can be read back"
+before,0,0,q,"PBODY,:enlist ""caplog:`info`warn`error!({[c;m] `caprows insert (`info;c;m)};{[c;m] `caprows insert (`warn;c;m)};{[c;m] `caprows insert (`error;c;m)});""",1,1,the binary {[c;m]} log dep contract
+before,0,0,q,"PBODY,:enlist ""trade:([]time:`timestamp$();sym:`g#`symbol$();price:`float$();size:`long$());""",1,1,"the captured table - a g# attribute, so the whole schema handoff is exercised end to end"
+before,0,0,q,"PBODY,:enlist ""tp.init[`log`timer`schemas`logdir`logname`batch!(caplog;timer;enlist[`trade]!enlist trade;LOGDIR;\""tp\"";BATCH)];""",1,1,"stand the tickerplant up - init publishes subdetails and tablelist at ROOT, where our .z.pg finds them"
+before,0,0,q,"PBODY,:enlist ""priorpc:@[value;`.z.pc;{[e] (::)}];""",1,1,"capture di.pubsub's own .z.pc, which it self-assigns at load - clobbering it would break subscriber cleanup"
+before,0,0,q,"PBODY,:enlist "".z.pc:{[w] if[not (::)~priorpc;priorpc w]; if[w>0;exit 0]};""",1,1,"self-exit backstop, CHAINED onto di.pubsub's handler. GUARDED on w>0: a child started with </dev/null hits stdin EOF as soon as the script ends and that fires .z.pc[0], so an unguarded backstop kills the peer before the parent can connect (measured)"
+before,0,0,q,"PBODY,:enlist ""mkrow:{[i] (enlist `$\""S\"",string i;enlist 1.0*i;enlist `long$10*i)};""",1,1,"a feed update: sym, price, size as one-element COLUMN VECTORS - the shape a real feed sends, and what the log replay on the far side can normalise. no time column, so the tickerplant stamps it"
+before,0,0,q,"PBODY,:enlist ""feed:{[ns] {[i] tp[`upd][`trade;mkrow i]} each ns; :count ns};""",1,1,drive the feed entry point
+before,0,0,q,"PBODY,:enlist ""flushtp:{[] .m.di.0tickerplant.publishbuffer[]; :`flushed};""",1,1,force one batch flush - the timer is never started in this peer
+before,0,0,q,"PBODY,:enlist ""counts:{[] :tp[`getcounts][]};""",1,1,read the published (i) and logged (j) counts back
+before,0,0,q,"PBODY,:enlist ""errs:{[] :exec ctx from caprows where lvl=`error};""",1,1,read back anything the tickerplant logged at error
+before,0,0,q,"peerlines:{[d;tok;batch] :((""(hsym`$\"""",d,""/port\"") 0: enlist string system\""p\"";"");(""TOKEN:`"",(string tok),"";"");(""LOGDIR:\"""",d,""/tplogs\"";"");(""system\""mkdir -p \"",LOGDIR;"");(""BATCH:"",$[batch;""1b"";""0b""],"";"")),PBODY}",1,1,"prepend the port report, an identity token, the log directory and the batch mode, then the body"
+before,0,0,q,"spawnpeer:{[nm;tok;batch] d:BASE,""/"",nm; system""rm -rf "",d; system""mkdir -p "",d; (hsym`$d,""/tp.q"") 0: peerlines[d;tok;batch]; system (getenv[`QHOME]),""/bin/q "",d,""/tp.q -p 0W -q </dev/null >/dev/null 2>&1 &""; p:readport[d,""/port"";8000]; if[null p;'""integration: peer "",nm,"" never reported a port""]; h:hopen (`$"":localhost:"",string p;5000); if[not tok~@[h;""TOKEN"";{[e] :`}];hclose h;'""integration: the peer that answered is not this run's""]; :h}",1,1,spawn one tickerplant peer and return a handle to it. NB the script path comes BEFORE the flags - q silently ignores a script that follows them
+before,0,0,q,"settle:{[h] system""sleep 0.4""; h""1+1""; system""sleep 0.2""; h""1+1"";}",1,1,let an async publish arrive and our own inbound queue drain
+comment,,,,,,,SCENARIO 1 - batch mode (the default). we subscribe with rows already published AND rows still
+comment,,,,,,,buffered on the tickerplant. that is the case which separates the two candidate message counts:
+comment,,,,,,,"subdetails must report the PUBLISHED watermark i (4), so we replay 4 and then receive the 2"
+comment,,,,,,,buffered rows at the next flush. reporting the logged total j would replay 6 and deliver 2 again
+before,0,0,q,"H1:spawnpeer[""tp1"";`tp1tok;1b]",1,1,"spawn a batch-mode tickerplant. this ONE handle carries the control calls, the subscription and the live feed - exactly as a real subscriber's does"
+before,0,0,q,"PID1:H1"".z.i""",1,1,"its pid, for the leak check"
+before,0,0,q,H1 (`feed;til 4),1,1,"four updates: logged, and buffered in the peer's root table"
+before,0,0,q,"PRE:H1""counts[]""",1,1,capture: counts before anything is published
+before,0,0,q,"H1""flushtp[]""",1,1,publish the buffer to nobody - this is what moves i up to j
+before,0,0,q,"MID:H1""counts[]""",1,1,capture: counts after the flush
+before,0,0,q,H1 (`feed;4 5),1,1,"two more updates: logged, still buffered, NOT yet published"
+before,0,0,q,"ATSUB:H1""counts[]""",1,1,capture: the counts a subscriber arriving now is answered from
+before,0,0,q,TPD:ATSUB`d,1,1,capture: the trading date the tickerplant reports
+before,0,0,q,R1:sub.subscribe[H1;`;`;1b;1b],1,1,"THE CALL UNDER TEST - ` for all tables, so tablelist is exercised too; define the schemas at root and replay the log"
+before,0,0,q,REPLAYED:count trade,1,1,"capture: rows recovered from the tp log, before any live delivery"
+before,0,0,q,ATT1:exec first a from meta trade where c=`sym,1,1,capture: the attribute the tickerplant's schema carried across
+before,0,0,q,SUBSCRIBED1:sub.subscribed[],1,1,capture: di.subscriptions reports a live subscription
+before,0,0,q,"H1""flushtp[]""",1,1,"publish the two buffered rows - we are a registered subscriber now, so they come to us"
+before,0,0,q,settle[H1],1,1,let them arrive
+before,0,0,q,FINAL:count trade,1,1,capture: total rows here
+before,0,0,q,SYMS:exec sym from trade,1,1,"capture: the syms, in order - proves no gap and no duplicate"
+before,0,0,q,"PERR1:H1""errs[]""",1,1,capture: anything the tickerplant logged at error
+before,0,0,q,"TABS1:H1""tablelist[`]""",1,1,"capture: the tablelist reply, in the arity di.subscriptions actually sends"
+comment,,,,,,,"SCENARIO 2 - zero-latency mode. every logged message has already gone out, so i must equal j and"
+comment,,,,,,,the subscriber must replay all three. with i stuck at its seeded 0 this replayed nothing
+before,0,0,q,sub.unsubscribe[H1],1,1,release the first subscription so the duplicate guard does not reject the second
+before,0,0,q,@[hclose;H1;{[e] :(::)}],1,1,close the peer - the backstop makes it exit
+before,0,0,q,![`.;();0b;enlist`trade],1,1,"drop the table, so scenario 2 starts from nothing and its row counts stand alone"
+before,0,0,q,"system""sleep 0.4""",1,1,let the peer exit
+before,0,0,q,"H2:spawnpeer[""tp2"";`tp2tok;0b]",1,1,spawn a ZERO-LATENCY tickerplant
+before,0,0,q,"PID2:H2"".z.i""",1,1,"its pid, for the leak check"
+before,0,0,q,H2 (`feed;til 3),1,1,"three updates: published immediately to nobody, and logged"
+before,0,0,q,"PRE2:H2""counts[]""",1,1,capture: i must already equal j - nothing is ever buffered in this mode
+before,0,0,q,"BUF2:H2""count trade""",1,1,capture: zero-latency does not buffer into the peer's root table
+before,0,0,q,R2:sub.subscribe[H2;`trade;`;1b;1b],1,1,"subscribe against the zero-latency tickerplant, naming the table explicitly this time"
+before,0,0,q,REPLAYED2:count trade,1,1,capture: rows recovered from the log
+before,0,0,q,H2 (`feed;3 4),1,1,"two more updates, published live this time"
+before,0,0,q,settle[H2],1,1,let them arrive
+before,0,0,q,FINAL2:count trade,1,1,capture: total rows here
+before,0,0,q,SYMS2:exec sym from trade,1,1,"capture: the syms, in order"
+before,0,0,q,"PERR2:H2""errs[]""",1,1,capture: anything the tickerplant logged at error
+comment,,,,,,,teardown - close the peer and prove neither leaked
+before,0,0,q,@[sub.unsubscribe;H2;{[e] :(::)}],1,1,release the second subscription
+before,0,0,q,@[hclose;H2;{[e] :(::)}],1,1,close the peer - the backstop makes it exit
+before,0,0,q,"system""sleep 0.5""",1,1,let it exit
+before,0,0,q,"gone:{[pid] :0=count @[{system"""" sv (""ps -p "";string x;"" -o pid="")};pid;{[e] :()}]}",1,1,"ps throws 'os when the pid is absent, so the probe has to be wrapped"
+before,0,0,q,GONE1:gone PID1,1,1,capture: no leaked first peer
+before,0,0,q,GONE2:gone PID2,1,1,capture: no leaked second peer
+before,0,0,q,SERR:exec ctx from caprows where lvl=`error,1,1,capture: anything di.subscriptions logged at error on this side
+before,0,0,q,"system""rm -rf "",BASE",1,1,remove the working directory
+comment,,,,,,,scenario 1 - batch mode
+true,0,0,q,(0=PRE`i) and 4=PRE`j,1,1,"i does not move while the rows are only buffered - four logged, none published"
+true,0,0,q,4=MID`i,1,1,the flush brings i up to j
+true,0,0,q,(4=ATSUB`i) and 6=ATSUB`j,1,1,at subscribe time two more messages are logged but not yet published
+true,0,0,q,4=REPLAYED,1,1,"THE TEST - di.subscriptions replayed exactly the PUBLISHED watermark, not the logged total"
+true,0,0,q,6=FINAL,1,1,and then received the two buffered rows once each: no gap and no duplicate. reporting j would give 8
+true,0,0,q,"SYMS~`$""S"",/:string til 6",1,1,"every row arrived exactly once, in order - a row count alone could hide an offsetting error"
+true,0,0,q,`g=ATT1,1,1,the g# attribute reached this side through schemalist
+true,0,0,q,(enlist`trade)~R1`subtables,1,1,di.subscriptions resolved the ` request through tablelist and subscribed to trade
+true,0,0,q,(enlist`trade)~TABS1,1,1,tablelist answers the (`tablelist;`) call di.subscriptions sends - a niladic form would throw 'rank and be silently downgraded
+true,0,0,q,TPD~R1`tplogdate,1,1,tplogdate matches the date the tickerplant reported and named its log after
+true,0,0,q,(99h=type R1`rowcounts) and 4=(R1`rowcounts)`trade,1,1,"rowcounts reports rows PUBLISHED for the table, so it agrees with the message count in the same reply"
+true,0,0,q,SUBSCRIBED1,1,1,di.subscriptions holds a live subscription on the handle
+true,0,0,q,0=count PERR1,1,1,the tickerplant logged nothing at error during the whole exchange
+comment,,,,,,,scenario 2 - zero-latency mode
+true,0,0,q,(3=PRE2`i) and 3=PRE2`j,1,1,i tracks j when every message is published as it is logged
+true,0,0,q,0=BUF2,1,1,zero-latency does not buffer into the tickerplant's root table
+true,0,0,q,3=REPLAYED2,1,1,all three pre-subscription messages were replayed - with i stuck at 0 this would replay none
+true,0,0,q,5=FINAL2,1,1,"the two live updates arrived on top, once each"
+true,0,0,q,"SYMS2~`$""S"",/:string til 5",1,1,"every row arrived exactly once, in order"
+true,0,0,q,(enlist`trade)~R2`subtables,1,1,the explicitly-named table was subscribed
+true,0,0,q,0=count PERR2,1,1,the tickerplant logged nothing at error
+comment,,,,,,,teardown
+true,0,0,q,GONE1 and GONE2,1,1,neither tickerplant peer leaked
+true,0,0,q,0=count SERR,1,1,di.subscriptions logged nothing at error during the entire run
+comment,,,,,,,"VERSION guards. these are LOAD-time, so they can only be observed in a CHILD process - the module"
+comment,,,,,,,is already loaded here and cannot be reloaded with a different VERSION file. a temp QPATH tree
+comment,,,,,,,symlinks every other module and copies di/tickerplant so only its VERSION is altered
+before,0,0,q,"VBASE:BASE,""/vtree""",1,1,root for the VERSION-guard trees
+before,0,0,q,"mkvtree:{[nm;mut] d:VBASE,""/"",nm; realdi:""/"" sv -1_ ""/"" vs .Q.m.mp`di.tickerplant; system""rm -rf "",d; system""mkdir -p "",d,""/di""; system""ln -s "",realdi,""/* "",d,""/di/ 2>/dev/null""; system""rm -f "",d,""/di/tickerplant""; system""cp -r "",realdi,""/tickerplant "",d,""/di/tickerplant""; mut d; :d}",1,1,build a temp module tree whose di/tickerplant VERSION can be mutated. links EVERY module and overwrites only tickerplant - hand-listing deps would rot as the graph changes
+before,0,0,q,"loadin:{[d] f:d,""/probe.q""; (hsym`$f) 0: (""r:@[{use`di.tickerplant; :`OK};::;{[e] `$\""THREW: \"",e}];"";""if[r~`OK; tpx:use`di.tickerplant; r:`$\""OK [\"",tpx.version,\""]\""];"";""(hsym`$\"""",d,""/out\"") 0: enlist string r;"";""exit 0;""); system ""QPATH="",d,"" "",(getenv[`QHOME]),""/bin/q "",f,"" -q </dev/null >/dev/null 2>&1""; :$[count key hsym`$d,""/out"";first read0 hsym`$d,""/out"";""NO OUTPUT""]}",1,1,load di.tickerplant in a child q against that tree. QPATH MUST be set for the child - inheriting ours makes it load the REAL module and every case passes vacuously
+before,0,0,q,"VGOOD:loadin mkvtree[""good"";{[d] }]",1,1,control: an untouched VERSION must load cleanly
+before,0,0,q,"VMISSING:loadin mkvtree[""missing"";{[d] system""rm -f "",d,""/di/tickerplant/VERSION""}]",1,1,a missing VERSION must fail loudly and name the module
+before,0,0,q,"VEMPTY:loadin mkvtree[""empty"";{[d] (hsym`$d,""/di/tickerplant/VERSION"") 0: enlist """"}]",1,1,an empty VERSION must fail loudly and name the module
+before,0,0,q,"VPAD:loadin mkvtree[""pad"";{[d] (hsym`$d,""/di/tickerplant/VERSION"") 0: enlist ""0.1.0  ""}]",1,1,"a padded VERSION must load and be TRIMMED. written from q, not shell printf - the quoting does not survive system[] and silently produces an empty file, which tests the wrong guard"
+before,0,0,q,"system""rm -rf "",VBASE",1,1,remove the VERSION-guard trees
+true,0,0,q,"""OK [0.1.0]""~VGOOD",1,1,"control - a clean VERSION loads and reports the exact version. BRACKETED because writing a bare string through 0:/read0 strips trailing spaces, which would destroy the padding the last assert looks for"
+true,0,0,q,"0<count VMISSING ss ""di.tickerplant: VERSION file missing or unreadable""",1,1,a missing VERSION names di.tickerplant rather than surfacing a raw OS error that points nowhere
+true,0,0,q,"0<count VEMPTY ss ""di.tickerplant: VERSION file is empty""",1,1,"an empty VERSION is rejected, not accepted as a version that reads to di.depcheck as 'exports no version'"
+true,0,0,q,"""OK [0.1.0]""~VPAD",1,1,"a padded VERSION is TRIMMED to the bare semver. asserts the VALUE, not just that it loaded - the naive read loads happily and hands di.depcheck a padded string di.depcheck compares as a string"
diff --git a/di/tickerplant/tickerplant.md b/di/tickerplant/tickerplant.md
index e3353c8c..84a8dd35 100644
--- a/di/tickerplant/tickerplant.md
+++ b/di/tickerplant/tickerplant.md
@@ -21,7 +21,7 @@ tp.init[`log`timer`schemas!(logdep;timerdep;enlist[`trade]!enlist trade)]
 | Key | Required | Description |
 |---|---|---|
 | `log` | yes | `` `info`warn`error `` dict of `{[ctx;msg]}` functions |
-| `timer` | yes | `di.timer`'s exports (must expose `addjob`) |
+| `timer` | yes | `di.timer`'s exports; must expose `addjob` (with the `custom` variant) and `deletejobs` |
 | `schemas` | yes | `tablename!schema` dict; the tables to capture |
 | `batch` | no | `1b` (default) buffers and publishes on a timer; `0b` publishes each update immediately |
 | `batchperiod` | no | batch publish interval (timespan, whole seconds; default `0D00:00:01`) |
@@ -31,24 +31,101 @@ tp.init[`log`timer`schemas!(logdep;timerdep;enlist[`trade]!enlist trade)]
 | `rolltimezone` / `datatimezone` / `rolltimeoffset` | no | forwarded to `di.eodtime` |
 
 `init` initialises the dependency modules (`di.eodtime`, `di.tplog`, `di.pubsub`), materialises the
-schemas as root tables (applying `` `g# `` to any `sym` column), opens today's log, and schedules a
-single timer job that flushes the buffer (batch mode) and checks for the end-of-day roll. It is
-idempotent — a re-init does not re-add the timer job.
+schemas as root tables (applying `` `g# `` to any `sym` column), opens today's log, **publishes
+`subdetails` and `tablelist` at root**, and schedules a single timer job that flushes the buffer
+(batch mode) and checks for the end-of-day roll.
 
-## Root tables and the upd contract
+`deletejobs` is required even though only `teardown` calls it: a dict-valued dependency returns a
+null-shaped value for an absent key rather than erroring, so a missing `deletejobs` would fail
+silently inside `teardown` rather than at wiring time.
 
-The captured tables live at **root**, not in `.z.m`: a tickerplant owns its tables, feeds insert into
-them, and `di.pubsub` reads them by name, so they cannot be module-local. This is the one deliberate
-root-state exception; all other mutable state is module-local. `di.torq` wires the process's root
-`upd` to `tickerplant.upd` so feeds can publish to it.
+## Root state: the tables and the subscription protocol
+
+Two things this module owns live at **root**, not in `.z.m`:
+
+- **The captured tables.** A tickerplant owns its tables, feeds insert into them, and `di.pubsub`
+  reads them by name, so they cannot be module-local.
+- **`subdetails` and `tablelist`.** A subscriber reaches these over IPC through the default
+  `.z.pg`/`.z.ps`, which resolve names at root. A bare assignment in module code lands in the
+  module's private namespace and would never be found, so `init` publishes them explicitly with
+  ``@[`.;nm;:;f]`` and `teardown` gives them back — the same install/uninstall pair `di.rdb` uses.
+  Installing over a name that holds something neither this module nor the caller installed logs a
+  warning first; `teardown` removes a name only while it still holds this module's function.
+
+All other mutable state is module-local. The process's root `upd` is **not** published here —
+`di.torq` wires the feed entry point, which may legitimately be a caller-supplied wrapper around
+this module's `upd`.
+
+## The subdetails / tablelist contract
+
+This is the seam every downstream process depends on. `di.subscriptions` drives both.
+
+### `tablelist[x]` → symbol list
+
+The tables offered for subscription (`di.pubsub`'s subscribable list). `di.subscriptions` resolves a
+`` ` `` (all tables) request by sending `` (`tablelist;`) `` — so this is **unary**, and the argument
+is accepted and ignored. TorQ's own `` tablelist:{.stpps.t} `` is unary for the same reason. A
+niladic `{[] …}` would throw `'rank`, which `di.subscriptions` catches and silently downgrades to
+asking for `` ` `` — which a segmented tickerplant cannot answer.
+
+### `subdetails[tabs;syms]` → dict
+
+Subscribes the calling handle **and** returns everything a subscriber needs to recover. Asking for
+the schemas *is* the subscription: `di.pubsub` registers the handle for live delivery as a side
+effect, and there is no unsubscribe verb to undo it.
+
+| Key | Shape | Meaning |
+|---|---|---|
+| `schemalist` | list of `(tablename;schema)` pairs | the table name and its empty schema, attributes included |
+| `logfilelist` | list of `(messagecount;logfile)` pairs | at most one entry — this tickerplant writes a single log. Empty when logging is disabled |
+| `rowcounts` | dict keyed by table name | rows **published** for each subscribed table so far today |
+| `date` | date | the tickerplant's current trading date |
+
+Behaviour on a partial or empty match, driven by `di.pubsub.subscribe`'s three reply shapes:
+
+- every requested table exists → the dict above;
+- **some** exist → the dict for those, plus a `warn` naming the ones that do not. It must still
+  succeed: those tables really were subscribed, and signalling would report failure to a caller that
+  is now registered;
+- **none** exist → signals. `di.pubsub` registered nothing in that case, so failing leaves no half
+  subscription behind, and `di.subscriptions` reports the message verbatim.
+
+### `logfilelist`'s message count is `i`, the published watermark — not `j`
+
+This is the one part of the contract that is easy to get backwards, so the reasoning is recorded
+here.
+
+`getcounts` reports two counters: `i`, messages **published** to subscribers, and `j`, messages
+**written to the log**. In zero-latency mode they move together. In batch mode `j` runs ahead: a row
+is logged by `upd` but stays in the buffer until the next timer tick.
+
+`subdetails` reports **`i`**. In batch mode a row that is logged but not yet flushed is still in the
+buffer, and `di.pubsub.pubclear` publishes the *whole table* to every registered handle — including
+one that registered after that row was buffered. So a subscriber told to replay `j` messages would
+replay those rows from the log **and** receive them again at the next tick:
+
+| t | event | i | j | buffered | subscriber told `i=0` | subscriber told `j=1` |
+|---|---|---|---|---|---|---|
+| 0.1 | `upd` A | 0 | 1 | A | — | — |
+| 0.2 | subscriber calls `subdetails` | 0 | 1 | A | replays nothing | replays A |
+| 1.0 | tick publishes the buffer | 1 | 1 | — | receives A | receives A **again** |
+
+TorQ reports the same field for the same reason — `chainedtp.q`'s `.ctp.sub` puts `.u.i` in the
+reply, and kdb+tick's `r.q` replays with `` .u`i ``.
+
+`rowcounts` is the per-table counterpart (TorQ's `.u.icounts`) and is bumped on the same events, so
+the two numbers in one reply always agree about what has been published.
 
 ## Exported functions
 
 | Function | Signature | Description |
 |---|---|---|
-| `upd` | `[table;data]` | Feed entry point: stamp the update, then buffer+log (batch) or publish+log (zero-latency). |
-| `subscribe` | `[tables;filters]` | Register a subscriber (delegates to `di.pubsub`); called by downstream processes over IPC. |
+| `upd` | `[table;data]` | Feed entry point: stamp the update, then buffer+log (batch) or log+publish (zero-latency). |
+| `subscribe` | `[tables;filters]` | The kdb+tick `.u.sub` entry point (delegates to `di.pubsub`), for a caller that wants the raw `(tables;schemas)` reply. |
+| `subdetails` | `[tables;syms]` | Subscribe and return the schemas, log details and counts `di.subscriptions` needs. Published at root. |
+| `tablelist` | `[ignored]` | The tables offered for subscription. Published at root; unary. |
 | `endofday` | `[]` | Flush the buffer, notify subscribers, roll the tp log, and advance the end-of-day state. |
+| `teardown` | `[]` | Remove the root protocol and the timer job `init` installed. |
 | `getcounts` | `[]` | `` `i`j`d `` — messages published (`i`), messages logged (`j`), and the trading date. |
 | `gettables` | `[]` | The tables this tickerplant captures. |
 
@@ -69,14 +146,68 @@ tickerplant must not do to its own log.
 
 - **Batch vs zero-latency.** In batch mode `upd` inserts into the root table and the timer job
   publishes the accumulated rows every `batchperiod`, then clears them. In zero-latency mode `upd`
-  publishes each update immediately and does not buffer. Both modes log every message.
+  logs and then publishes each update immediately and does not buffer. Both modes log every message.
+  The log-then-publish order matches TorQ's `chainedtp.q` and means a message that fails to publish
+  is still recoverable from the log.
+- **The log handle and the log path are separate names.** `.z.m.logfile` holds the handle (`writelog`
+  tests it with `>0i`); `.z.m.logpath` holds the path, which `logfilelist` reports. `openlog` sets
+  the path and returns the handle for the caller to store. They were one name, and the path never
+  survived — every caller overwrote it with the handle.
 - **End of day.** The roll fires when the current time passes `di.eodtime`'s next roll timestamp,
   checked on every `upd` and on every timer tick. `endofday` flushes, notifies subscribers, rolls the
-  log to the next day, and refreshes the roll time and data-timestamp offset from `di.eodtime`.
+  log to the next day, resets the message and row counts, and refreshes the roll time and
+  data-timestamp offset from `di.eodtime`.
+- **A re-init preserves runtime state.** Dependencies and config are refreshed on every `init`, but
+  the trading date, the message and row counts, and the open log are seeded only on the first. A
+  re-init — `di.torq` re-applying config, a config reload, a second wiring — must not rewind the date,
+  zero the counts a subscriber replays against, or reopen (and so leak) the log already being written
+  to. Table materialisation follows the same rule: a fresh `init` defines every captured table from
+  its schema, attributes included, while a re-init defines only names not already at root, because
+  re-running `nm set schema` over a live tickerplant discards every buffered row that has been logged
+  but not yet published — leaving the counts describing data that had just been thrown away. Same
+  precedent as `di.rdb` and `di.subscriptions`, which seed their runtime state only when fresh.
+  **Consequence:** a `logdir` or `logname` change on a re-init takes effect at the next roll, not
+  immediately.
+- **`init` must be called first, and every callable export says so — except `upd`.** `teardown`,
+  `subscribe`, `subdetails`, `tablelist`, `endofday`, `getcounts` and `gettables` each guard with
+  `requireinit` and report `di.tickerplant: <fn>: init must be called before any other function`.
+  `upd` deliberately does not: it runs once per feed message, and the `initialised[]` probe is a
+  protected apply costing ~0.7µs a call (measured) — a permanent per-message tax to catch a wiring
+  mistake that can only happen at startup and surfaces immediately when it does. Called before `init`
+  it throws a bare `'.m.di.0tickerplant.tabs` instead. `di.rdb`'s `updfn` is unguarded for the same
+  reason. A test pins which functions are on each side of that line, so the split cannot drift.
+- **The `VERSION` read fails loud.** `init.q` signals `di.tickerplant: VERSION file missing or
+  unreadable` / `... is empty` rather than doing a bare `first read0`. `di.depcheck` compares versions
+  as **strings**, so a whitespace-padded value silently breaks every dependent's check and an empty
+  one reads to it as "exports no version" — both failing far from the real cause. The value is
+  trimmed. Same shape as `di.rdb`, `di.dbwrite` and `di.eodtime`.
+- **`teardown` is idempotent and narrow.** It withdraws the process-global bindings — the root
+  protocol and the timer job — and deliberately leaves module state and the captured tables intact,
+  so a shutdown path can still inspect or save what is buffered. A re-init after a teardown
+  re-publishes the protocol and re-schedules the job.
 - **No `di.handlers` dependency.** Subscriber-disconnect cleanup is handled by `di.pubsub`'s own
   `.z.pc`. (`di.pubsub` should migrate to `di.handlers` so `.z.*` is not assigned outside the central
   registry — tracked separately, out of scope here.)
 
+## Design divergences from TorQ
+
+- **The `subdetails` protocol is a deliberate capability addition, not restored parity.** TorQ's
+  standard `code/processes/tickerplant.q` defines **no** `subdetails` — a grep over the TorQ tree
+  finds it only in `code/processes/chainedtp.q` and `code/processes/segmentedtickerplant.q`, which
+  `di.subscriptions`' own source comment also attributes it to. A standard TorQ tickerplant never
+  spoke this protocol.
+
+  It is implemented here anyway because there is no `di.chainedtp` or `di.segmentedtp` for `di.rdb`
+  to subscribe to instead, so without it no modular subscriber can attach to a modular tickerplant at
+  all. Both existing integration harnesses (`di/rdb/test_integration.csv`,
+  `di/subscriptions/test_integration.csv`) already hand-roll exactly this adapter in their spawned
+  peer, and both label it *"the subdetails adapter a modular tickerplant would own"*. This module now
+  owns it. Nobody should read its presence as evidence that a standard TorQ tickerplant had one.
+- **One log, not one per table.** `logfilelist` is a list because the protocol also serves a
+  segmented tickerplant, which writes one log per table. This module writes a single log, so it
+  reports at most one entry.
+- **Removed entirely:** all `.finspace.*` / `.aws.*` code. FinSpace is end-of-life.
+
 ## Testing
 
 `test.csv` / `test.q` (k4unit) run against the **real** `di.pubsub`, `di.eodtime`, `di.tplog` and
@@ -84,13 +215,40 @@ tickerplant must not do to its own log.
 job is exercised through `endofday`). A capturing logger is shared across the modules so their output
 is assertable.
 
-Coverage: the metadata/version contract; strict `init` dependency validation (a `fail` row per guard);
-init materialising the root tables and scheduling the timer job; batch and zero-latency `upd`;
-`endofday` flushing and rolling; `upd` input validation; and the two `di.tplog` integration points — a
-tickerplant-written log replaying through `di.tplog`, and rolling into a corrupt log repairing it via
-`tplog.check`.
+Coverage: the metadata/version contract; strict `init` dependency validation (a `fail` row per guard,
+plus a message assertion so a guard cannot pass on an unrelated throw); init materialising the root
+tables and scheduling the timer job; batch and zero-latency `upd`, including that `i` lags `j` while
+rows are buffered and tracks it when they are not; a re-init preserving the counts, the buffer and the
+log handle; `endofday` flushing and rolling; the `subdetails`/`tablelist` shapes, the published-not-
+logged message count, an empty `logfilelist` when logging is off, and the error naming an unpublished
+table; root publication and `teardown`; `upd` input validation; and the two `di.tplog` integration
+points.
 
 ```q
 k4unit:use`di.k4unit
 k4unit.moduletest`di.tickerplant
 ```
+
+It also carries the **`VERSION` guard** checks. Those fire at *load* time, so they can only be
+observed in a child process: the suite builds a temp `QPATH` tree that symlinks every other module
+and copies `di/tickerplant`, mutates only its `VERSION`, and loads the module in a child q — covering
+an untouched file, a missing one, an empty one, and a whitespace-padded one.
+
+`test_integration.csv` drives a **real `di.subscriptions`** against a **real `di.tickerplant`** over
+genuine IPC — the only test that actually proves the protocol gap is closed. The tickerplant is the
+spawned peer and `di.subscriptions` runs in the test process, not the other way round: a q process
+blocked in a sync call does not accept a new inbound connection, so the side that *dials* has to be
+the side that drives. Two scenarios, on OS-assigned ports with no port number anywhere in the file:
+
+- **batch mode** — subscribe with four rows already published and two still buffered, then flush.
+  The subscriber must replay 4 and end with 6 rows in order. Reporting `j` gives 8.
+- **zero-latency mode** — subscribe with three rows published-and-logged, then two more live. The
+  subscriber must replay 3 and end with 5. With `i` not advancing it replays 0.
+
+`moduletest` only ever loads `test.csv`, so load and run this suite directly, in a fresh session:
+
+```q
+k4unit:use`di.k4unit
+.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.tickerplant;`test_integration.csv]
+.m.di.0k4unit.KUrt[]
+```
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
index 27f28b7e..e9fc09e7 100644
--- a/di/tickerplant/tickerplant.q
+++ b/di/tickerplant/tickerplant.q
@@ -9,6 +9,10 @@
 / so they cannot be module-local. this is the one deliberate root-state exception for this process
 / module; all other mutable state is module-local in .z.m.
 / .
+/ NB init also publishes subdetails and tablelist at ROOT - see installroot. they are the subscription
+/ protocol di.subscriptions speaks, and an IPC caller reaches them through the default .z.pg/.z.ps,
+/ so a module-local binding would be invisible. teardown gives them back.
+/ .
 / NB subscriber-disconnect cleanup is handled by di.pubsub's own .z.pc (it self-assigns it). this
 / module therefore takes no di.handlers dependency. FLAG: di.pubsub should migrate to di.handlers so
 / .z.pc is not assigned outside the central registry - out of scope here, tracked separately.
@@ -23,10 +27,28 @@ defaultbatchperiod:0D00:00:01;
 / optional config keys forwarded verbatim to di.eodtime.init
 eodtimekeys:`rolltimezone`datatimezone`rolltimeoffset;
 
+/ the root names init publishes and teardown gives back - the subscription protocol di.subscriptions
+/ calls over IPC. see installroot
+rootnames:`subdetails`tablelist;
+
 / ============================================================
 / internal helpers
 / ============================================================
 
+initialised:{[]
+  / has init run? a direct (module-rewritten) reference detects prior setup without touching root.
+  / schemas has no module-level default - its only value comes from init, so this probe cannot be
+  / fooled by a load-time constant of the same name
+  :@[{.z.m.schemas;1b};::;{[e] :0b}];
+  };
+
+requireinit:{[ctx]
+  / every exported function except init depends on init having wired the logger. there is no default
+  / logger, so without this an early call dies with a bare 'type instead of a usable message
+  if[not initialised[];
+    '"di.tickerplant: ",string[ctx],": init must be called before any other function"];
+  };
+
 raiseerror:{[ctx;msg]
   / internal - log an error under ctx then signal it, so failures are observable as well as thrown
   .z.m.logerr[ctx;msg];
@@ -41,27 +63,40 @@ stamp:{[x]
   };
 
 openlog:{[date]
-  / open the tp log for `date`, checking/repairing a pre-existing log via di.tplog; 0i if logging off
-  if[0=count .z.m.logdir;:0i];
+  / open the tp log for `date`, checking/repairing a pre-existing log via di.tplog; 0i if logging off.
+  / records the PATH in .z.m.logpath and RETURNS the handle, which the caller stores in .z.m.logfile.
+  / the two are deliberately separate names: writelog tests the handle with >0i, and subdetails has to
+  / report the path to subscribers, so one name cannot carry both - it used to try, and the path
+  / assignment was dead because every caller overwrote it with the handle
+  if[0=count .z.m.logdir;.z.m.logpath:`;:0i];
   l:hsym `$ .z.m.logdir,"/",.z.m.logname,string date;
   if[count key l;l:tplog.check l];
   if[not count key l;l set ()];
-  .z.m.logfile:l;
+  .z.m.logpath:l;
   h:hopen l;
   .z.m.loginfo[`openlog;"logging to ",string l];
   :h;
   };
 
 publishbuffer:{[]
-  / batch mode: publish each root table's buffered rows to subscribers then clear them; i catches j
+  / batch mode: publish each root table's buffered rows to subscribers then clear them; i catches j.
+  / the row counts are taken BEFORE pubclear, which publishes and then empties each table.
+  / `. is the explicit ROOT namespace - the captured tables live there, not in .z.m
+  .z.m.rowcounts:.z.m.rowcounts+.z.m.tabs!count each `. .z.m.tabs;
   pubsub.pubclear[.z.m.tabs];
   .z.m.i:.z.m.j;
   };
 
 publishrows:{[t;x]
-  / zero-latency mode: publish one stamped update immediately, as a table keyed by t's columns
+  / zero-latency mode: publish one stamped update immediately, as a table keyed by t's columns, then
+  / mark it published. i:j is what makes i the PUBLISHED-message watermark in this mode - without it
+  / i never leaves its seeded 0 and every subscriber is told to replay nothing. TorQ does exactly this
+  / in chainedtp.q's tickpub (:96-99), and upd calls writelog FIRST so j is already bumped here
   f:cols t;
-  pubsub.publish[t;$[0>type first x;enlist f!x;flip f!x]];
+  d:$[0>type first x;enlist f!x;flip f!x];
+  pubsub.publish[t;d];
+  .z.m.rowcounts:@[.z.m.rowcounts;t;+;count d];
+  .z.m.i:.z.m.j;
   };
 
 writelog:{[t;x]
@@ -69,6 +104,21 @@ writelog:{[t;x]
   if[.z.m.logfile>0i;.z.m.logfile enlist (`upd;t;x);.z.m.j+:1];
   };
 
+logfilelist:{[]
+  / internal - the (messagecount;logfile) pairs subdetails reports. a LIST because the protocol also
+  / serves a segmented tickerplant, which writes one log per table; this one writes a single log, so
+  / there is at most one entry. EMPTY when logging is disabled - di.subscriptions accepts that
+  / (subscriptions.q:307) and replays nothing.
+  / the count is .z.m.i, the PUBLISHED watermark, NOT .z.m.j. in batch mode a row that has been logged
+  / but not yet flushed is still sitting in the buffer, and pubclear publishes the WHOLE table to every
+  / registered handle - including one that registered after the row was buffered. so reporting j would
+  / have the subscriber replay those rows from the log AND receive them again at the next tick.
+  / TorQ sends the same field for the same reason: chainedtp.q:169 puts .u.i in the reply, and
+  / kdb+tick's r.q replays with .u`i
+  if[null .z.m.logpath;:()];
+  :enlist (.z.m.i;.z.m.logpath);
+  };
+
 rollcheck:{[now]
   / trigger end-of-day if we have passed the next scheduled roll timestamp
   if[eodtime.getnextroll[]<now;endofday[]];
@@ -81,23 +131,82 @@ tick:{[]
   };
 
 rolllog:{[]
-  / close the current tp log and open a fresh one for the new date; reset log message counts
+  / close the current tp log and open a fresh one for the new date; reset message and row counts, as
+  / TorQ's refreshtp does (chainedtp.q:136). openlog updates .z.m.logpath alongside the handle
   if[.z.m.logfile>0i;hclose .z.m.logfile];
   .z.m.i:.z.m.j:0;
+  .z.m.rowcounts:.z.m.tabs!(count .z.m.tabs)#0;
   .z.m.logfile:openlog .z.m.d;
   };
 
+createtables:{[schemas;fresh]
+  / internal - materialise the captured tables at ROOT, applying `g# to any sym column.
+  / a FRESH init owns the tables outright and defines every one of them from its schema, attributes
+  / included. a RE-INIT defines only names not already at root - a schema key added since the first
+  / call - because re-running `nm set schema` over a live tickerplant would discard every buffered row
+  / that had been logged but not yet published, leaving the message counts describing data that had
+  / just been thrown away
+  nms:$[fresh;key schemas;(key schemas) where not (key schemas) in tables[`.]];
+  if[0=count nms;:()];
+  {[nm;s] nm set $[`sym in cols s;@[s;`sym;`g#];s]}'[nms;schemas nms];
+  };
+
+publishroot:{[nm;f]
+  / internal - publish ONE root entry point, warning first when the name already holds something that
+  / is neither the function about to be installed nor the one this module installed last time.
+  / uninstallroot deliberately refuses to delete a binding that is not ours; installing over one
+  / silently is that same asymmetry in reverse, and it is how a co-hosted process loses its own
+  / bindings without a word. same shape as di.rdb's publishroot
+  if[nm in key `.;
+    cur:`. nm;
+    if[not any (f;.z.m.rootinstalled nm)~\:cur;
+      .z.m.logwarn[`installroot;"root ",(string nm)," was already bound to something di.tickerplant ",
+        "did not install - replacing it. teardown will not give the previous binding back"]]];
+  @[`.;nm;:;f];
+  };
+
+installroot:{[]
+  / publish the subscription protocol at ROOT, where an IPC caller reaches it through the default
+  / .z.pg/.z.ps. a bare module-level assignment lands in this module's private namespace and would
+  / never be found:
+  /   subdetails - di.subscriptions sends (`subdetails;tabs;syms) and requires the reply keys
+  /   tablelist  - di.subscriptions sends (`tablelist;`) to resolve a ` (all tables) request
+  / NB the root `upd` is deliberately NOT published here - di.torq wires the process's feed entry
+  / point, which may legitimately be a caller-supplied wrapper around this module's upd
+  fs:(subdetails;tablelist);
+  publishroot'[rootnames;fs];
+  / record what was published, so the NEXT install can tell "someone else took this name" apart from
+  / a legitimate re-init. written as ONE dict rather than amended per name
+  .z.m.rootinstalled:rootnames!fs;
+  };
+
+dropifours:{[nm;f]
+  / internal - delete a root name only if it still holds the function we installed there
+  if[not nm in key `.;:()];
+  if[not f~`. nm;:()];
+  ![`.;();0b;enlist nm];
+  };
+
+uninstallroot:{[]
+  / internal - give back exactly what installroot published. only the names still bound to THIS
+  / module's functions are removed: a later module that has taken over one of these root names owns it
+  / now, and silently deleting its binding would be a worse outcome than leaving ours behind
+  dropifours'[rootnames;(subdetails;tablelist)];
+  };
+
 / ============================================================
 / public api
 / ============================================================
 
 init:{[deps]
   / wire the injected log + timer, initialise the dep modules, materialise the tables at root, open
-  / the tp log and schedule the batch/roll timer job.
+  / the tp log, publish the subscription protocol at root and schedule the batch/roll timer job.
   / deps: a dict with `log (required), `timer (required), `schemas (required, tablename!schema) and
   /   optional `batch (1b), `batchperiod (timespan), `logdir (string, "" disables logging),
   /   `logname (string), `subtables (symbol list), plus di.eodtime keys (rolltimezone/datatimezone/
   /   rolltimeoffset) forwarded verbatim.
+  / dependencies and config are re-applied on EVERY init; RUNTIME state - the date, the message and
+  / row counts, and the open log - is seeded only on the FIRST. see the fresh block below
   if[99h<>type deps;
     '"di.tickerplant: deps must be a dict with `log, `timer and `schemas keys"];
   if[not `log in key deps;
@@ -114,47 +223,89 @@ init:{[deps]
     '"di.tickerplant: timer dict must expose addjob"];
   if[not `custom in key deps[`timer]`addjob;
     '"di.tickerplant: timer addjob must expose the custom variant"];
+  / deletejobs is required because teardown deletes the job init schedules. a dict-valued dep returns
+  / a null-shaped value for an absent key rather than erroring, so a missing deletejobs would fail
+  / silently inside teardown's protected apply - same reasoning as di.rdb's timer validation
+  if[not `deletejobs in key deps`timer;
+    '"di.tickerplant: timer dict must expose deletejobs - teardown needs it"];
+  if[not (type deps[`timer]`deletejobs) within 100 112h;
+    '"di.tickerplant: timer deletejobs must be a function [ids]"];
   if[not `schemas in key deps;
     '"di.tickerplant: schemas is required; pass a tablename!schema dict keyed on `schemas"];
   if[99h<>type deps`schemas;
     '"di.tickerplant: schemas must be a dict of tablename!schema"];
+  / is this the FIRST init in this process? read it BEFORE any write, because initialised[] probes
+  / schemas and this must reflect the state on entry
+  fresh:not initialised[];
   .z.m.loginfo:(deps`log)`info;
   .z.m.logwarn:(deps`log)`warn;
   .z.m.logerr:(deps`log)`error;
   .z.m.timer:deps`timer;
-  / optional config with defaults
+  / optional config with defaults - one explicit write per key, so a reader can see at a glance which
+  / keys reach module state
   .z.m.batch:$[`batch in key deps;deps`batch;1b];
   .z.m.batchperiod:$[`batchperiod in key deps;deps`batchperiod;defaultbatchperiod];
   .z.m.logdir:$[`logdir in key deps;deps`logdir;""];
   .z.m.logname:$[`logname in key deps;deps`logname;"tp"];
-  / materialise the schemas as root tables (see the header note on root state), applying `g# to sym
   .z.m.schemas:deps`schemas;
   .z.m.tabs:key deps`schemas;
-  {[nm;s] nm set $[`sym in cols s;@[s;`sym;`g#];s]}'[.z.m.tabs;value deps`schemas];
+  / materialise the schemas as root tables (see the header note on root state), applying `g# to sym
+  createtables[deps`schemas;fresh];
   / initialise the dependency modules: eodtime (log + tz passthrough), tplog (log), pubsub (over the
-  / subscribable tables). tplog now takes an injected log and must be init'd before check is called.
+  / subscribable tables). tplog now takes an injected log and must be init'd before check is called
   eodtime.init[(enlist[`log]!enlist deps`log),(key[deps] inter eodtimekeys)#deps];
   tplog.init[enlist[`log]!enlist deps`log];
   pubsub.setsubtables[$[`subtables in key deps;deps`subtables;.z.m.tabs]];
   pubsub.init[];
-  / date, counts, and the tp log for today. close a handle held from a previous init before
-  / reopening, so a re-init does not leak the old file descriptor (rolllog closes on its own path)
-  .z.m.d:eodtime.getd[];
-  .z.m.i:.z.m.j:0;
-  if[`logfile in key .z.m;if[.z.m.logfile>0i;hclose .z.m.logfile]];
-  .z.m.logfile:openlog .z.m.d;
+  / RUNTIME state is seeded only on a FRESH init. a re-init - di.torq re-applying config, a config
+  / reload, a second wiring - must not rewind a live tickerplant's date, zero the message counts a
+  / subscriber replays against, or reopen (and so leak) the log it is already writing to. di.rdb and
+  / di.subscriptions set the same precedent; the dependency and config writes above are refreshed
+  / unconditionally. consequence, documented in tickerplant.md: a logdir/logname change applies at the
+  / next roll rather than immediately
+  if[fresh;
+    .z.m.d:eodtime.getd[];
+    .z.m.i:.z.m.j:0;
+    .z.m.rowcounts:.z.m.tabs!(count .z.m.tabs)#0;
+    / what installroot last published at root, keyed by name. seeded here rather than at module load
+    / because publishroot reads it on the FIRST install, before installroot has written it
+    .z.m.rootinstalled:(`$())!();
+    .z.m.scheduled:0b;
+    .z.m.logfile:openlog .z.m.d];
+  installroot[];
   / schedule the timer job that flushes the buffer (batch) and checks the roll; mode 1 = fixed period.
-  / guarded so a re-init does not re-add (di.timer.addjob throws on a duplicate id); tick reads
-  / .z.m.batch live, so the one job serves both modes across re-inits
-  if[not `scheduled in key .z.m;
+  / guarded on the flag rather than on `fresh` so a teardown-then-init pair re-schedules it - di.timer
+  / throws on a duplicate id, and teardown has deleted it. tick reads .z.m.batch live, so the one job
+  / serves both modes across re-inits
+  if[not .z.m.scheduled;
     .z.m.timer[`addjob][`custom][`tickerplant;tick;();`int$.z.m.batchperiod%0D00:00:01;1h;()!()];
     .z.m.scheduled:1b];
   .z.m.loginfo[`init;"di.tickerplant initialised (",$[.z.m.batch;"batch";"zero-latency"]," mode)"];
   };
 
+teardown:{[]
+  / release everything init installed process-wide: the root subscription protocol and the timer job.
+  / paired with init's side effects, the way di.rdb's teardown is paired with its root entry points.
+  / module state and the CAPTURED TABLES are deliberately left intact - a shutdown path may still need
+  / to inspect or save what is buffered; only the process-global bindings are withdrawn
+  requireinit[`teardown];
+  uninstallroot[];
+  / deleting a job that was never scheduled is a no-op in di.timer (a delete-where over the jobs
+  / table), so the id does not have to exist. protected only because a timer whose deletejobs throws
+  / must not take down a shutdown path - init has already established that it is a function
+  @[.z.m.timer[`deletejobs];enlist`tickerplant;{[e] :(::)}];
+  .z.m.scheduled:0b;
+  .z.m.loginfo[`teardown;"di.tickerplant root entry points and timer job removed"];
+  };
+
 upd:{[t;x]
-  / feed entry point: stamp the update, then buffer+log (batch) or publish+log (zero-latency).
+  / feed entry point: stamp the update, then buffer+log (batch) or log+publish (zero-latency).
   / t is the table name, x the column data. wired to root `upd` by di.torq so feeds can call it.
+  / NB the ONLY callable export without a requireinit guard, deliberately: this runs once per feed
+  / message, and initialised[] is a protected apply costing ~0.7us a call (measured) - a permanent
+  / per-message tax to catch a wiring mistake that can only happen at startup and shows up instantly
+  / when it does. calling it before init throws a bare '.m.di.0tickerplant.tabs instead of this
+  / module's own message; that is the accepted trade. di.rdb's updfn is unguarded for the same reason
   if[not -11h=type t;raiseerror[`upd;"table must be a symbol"]];
   if[not t in .z.m.tabs;raiseerror[`upd;"unknown table ",string t]];
   rollcheck .z.p;
@@ -162,16 +313,57 @@ upd:{[t;x]
   if[not count x;:()];
   x:stamp x;
   if[.z.m.batch;t insert x;writelog[t;x]];
-  if[not .z.m.batch;publishrows[t;x];writelog[t;x]];
+  / log BEFORE publishing, so publishrows can copy the bumped j into i, and so a message that fails to
+  / publish is at least recoverable from the log. TorQ orders it the same way - chainedtp.q:159-161
+  if[not .z.m.batch;writelog[t;x];publishrows[t;x]];
   };
 
 subscribe:{[tabs;filters]
-  / register a subscriber (delegates to di.pubsub); called by downstream processes over IPC
+  / register a subscriber (delegates to di.pubsub); the kdb+tick .u.sub entry point, for a caller that
+  / wants the raw (tables;schemas) reply rather than the subdetails protocol
+  requireinit[`subscribe];
   :pubsub.subscribe[tabs;filters];
   };
 
+subdetails:{[tabs;instruments]
+  / TorQ's subdetails protocol, published at ROOT by init. di.subscriptions sends
+  / (`subdetails;tabs;syms) and requires `schemalist`logfilelist`rowcounts`date
+  / (subscriptions.q:20,266,272); asking for the schemas IS the subscription, so this registers the
+  / calling handle for live delivery as a side effect.
+  / di.pubsub.subscribe returns THREE shapes (pubsub.q:126-137): (tables;schemas) when every requested
+  / table exists, (errmsg;(tables;schemas)) when only some do, and a bare errmsg SYMBOL when none do.
+  / flip turns the pair into the (tablename;schema) rows schemalist wants
+  requireinit[`subdetails];
+  r:pubsub.subscribe[tabs;instruments];
+  / nothing matched: di.pubsub registered NOTHING (pubsub.q:28,44), so failing here leaves no half
+  / subscription behind. raiseerror rather than passing the symbol back - di.subscriptions wraps this
+  / call in a protected apply and reports the message verbatim, whereas a bare symbol reaches it as
+  / the far less useful "subdetails must return a dictionary, got type -11h"
+  if[-11h=type r;raiseerror[`subdetails;"no requested table is published: ",string r]];
+  / a PARTIAL match still succeeds: the tables that did match are already registered, so signalling
+  / would tell the caller it failed while leaving it subscribed. say so at warn instead
+  partial:-11h=type first r;
+  if[partial;.z.m.logwarn[`subdetails;string first r]];
+  pairs:flip $[partial;last r;r];
+  nms:pairs[;0];
+  :`schemalist`logfilelist`rowcounts`date!
+    (pairs;logfilelist[];nms!0^.z.m.rowcounts nms;.z.m.d);
+  };
+
+tablelist:{[x]
+  / the tables offered for subscription, so di.subscriptions can resolve a ` (all tables) request
+  / (subscriptions.q:31,365). published at ROOT by init.
+  / UNARY on purpose - di.subscriptions sends (`tablelist;`), so a niladic {[] ...} would throw 'rank,
+  / which it CATCHES and downgrades to asking for ` (subscriptions.q:366-369): a silent degradation
+  / that a segmented tickerplant cannot answer. TorQ's own tablelist:{.stpps.t} is unary for the same
+  / reason. the argument is accepted and ignored, exactly as legacy does
+  requireinit[`tablelist];
+  :pubsub.getsubtables[];
+  };
+
 endofday:{[]
   / flush any buffer, notify subscribers, roll the tp log, advance eodtime state and reset counts
+  requireinit[`endofday];
   if[.z.m.batch;publishbuffer[]];
   pubsub.callendofday[.z.m.d];
   .z.m.d+:1;
@@ -182,21 +374,37 @@ endofday:{[]
   };
 
 getcounts:{[]
-  / current message counts and trading date - i (in the log), j (log plus buffered), d (date)
+  / message counts and trading date - i (published to subscribers), j (written to the log), d (date).
+  / i is the watermark subdetails reports for replay; in batch mode it lags j by whatever is still
+  / buffered, and in zero-latency mode the two move together
+  requireinit[`getcounts];
   :`i`j`d!(.z.m.i;.z.m.j;.z.m.d);
   };
 
 gettables:{[]
   / the tables this tickerplant captures
+  requireinit[`gettables];
   :.z.m.tabs;
   };
 
 getapimeta:{[]
   / callable api for di.torq to register with di.api (init/getapimeta/version are plumbing, omitted)
   :flip `name`public`descrip`params`return!flip(
-    (`upd;1b;"feed entry point - stamp, log and publish (or buffer) an update";"[symbol table; list data]";"null");
-    (`subscribe;1b;"register a subscriber for tables/syms (delegates to di.pubsub)";"[symbol|list tables; filters]";"subscription result");
-    (`endofday;1b;"flush, notify subscribers, roll the log and advance eod state";"[]";"null");
-    (`getcounts;1b;"current log/buffer message counts and trading date";"[]";"dict `i`j`d");
-    (`gettables;1b;"the tables this tickerplant captures";"[]";"symbol list of table names"));
+    (`upd;        1b; "feed entry point - stamp, log and publish (or buffer) an update";
+       "[symbol: table; list: column data]";                    "null");
+    (`subscribe;  1b; "register a subscriber for tables/syms - the kdb+tick .u.sub entry point";
+       "[symbol(list): tables (` for all); filters]";           "list: (tables;schemas), or an error symbol");
+    (`subdetails; 1b; "subscribe and return the schemas, log details and counts di.subscriptions needs";
+       "[symbol(list): tables (` for all); symbol(list)|dict: syms (` for all)]";
+       "dict: schemalist, logfilelist, rowcounts and date");
+    (`tablelist;  1b; "the tables offered for subscription, so a ` (all tables) request can be resolved";
+       "[ignored - unary because di.subscriptions sends (`tablelist;`)]"; "symbol list: table names");
+    (`endofday;   1b; "flush, notify subscribers, roll the log and advance eod state";
+       "[]";                                                    "null");
+    (`teardown;   1b; "remove the root subscription protocol and the timer job installed by init";
+       "[]";                                                    "null");
+    (`getcounts;  1b; "published (i) and logged (j) message counts, and the trading date";
+       "[]";                                                    "dict: i, j and d");
+    (`gettables;  1b; "the tables this tickerplant captures";
+       "[]";                                                    "symbol list: table names"));
   };

From 5a1e32cb9f3c70409451b076126fb38427ffb1f2 Mon Sep 17 00:00:00 2001
From: alowrydi <aaron.lowry@dataintellect.com>
Date: Mon, 24 Aug 2026 10:40:30 +0100
Subject: [PATCH 7/7] Forward an optional handlers dep through to
 di.pubsub.init

di.pubsub's init now accepts an optional handlers key that registers
its .z.pc hook with di.handlers' own registry. This module forwards
its own optional handlers dep straight through, unvalidated, the same
way log is forwarded to di.eodtime/di.tplog. Omitting the key keeps
existing behaviour unchanged.
---
 di/tickerplant/test.csv       |  2 ++
 di/tickerplant/test.q         | 12 ++++++++++++
 di/tickerplant/tickerplant.md | 15 +++++++++++----
 di/tickerplant/tickerplant.q  | 19 +++++++++++++------
 4 files changed, 38 insertions(+), 10 deletions(-)

diff --git a/di/tickerplant/test.csv b/di/tickerplant/test.csv
index bdd67a9e..c6cdd8a9 100644
--- a/di/tickerplant/test.csv
+++ b/di/tickerplant/test.csv
@@ -4,6 +4,7 @@ before,0,0,q,timer:use`di.timer,1,,real di.timer (addjob works without timer.ini
 before,0,0,q,ntp:use`di.tplog,1,,"di.tplog, to replay a tickerplant-written log"
 before,0,0,q,eod:use`di.eodtime,1,,"di.eodtime, to force a roll due in a test"
 before,0,0,q,ps:use`di.pubsub,1,,"di.pubsub, to release the self-subscription a subdetails test leaves on handle 0"
+before,0,0,q,handlers:use`di.handlers,1,,"di.handlers, to verify the OPTIONAL handlers dep init forwards to di.pubsub is actually wired through"
 before,0,0,q,os:use`di.os,1,,os module for portable path resolution
 before,0,0,q,"system ""l "",os.abspath[""di/tickerplant/test.q""]",1,,load fixture helpers
 before,0,0,q,setupfixture[],1,,create the temp fixture root
@@ -29,6 +30,7 @@ true,0,0,q,testemptyupdtriggersroll[],1,1,an empty update still triggers an over
 true,0,0,q,testzerolatency[],1,1,zero-latency upd publishes without buffering and advances i with j
 true,0,0,q,testendofday[],1,1,"endofday flushes, rolls the log and resets counts"
 true,0,0,q,testreinitpreservesstate[],1,1,a re-init preserves the counts / buffer / log handle and leaks no fd
+true,0,0,q,testhandlersforwarded[],1,1,init's optional handlers dep is forwarded to di.pubsub so its .z.pc hook is visible in di.handlers' own registry
 comment,,,,,,,di.tplog integration
 true,0,0,q,testlogroundtrip[],1,1,a tickerplant-written log replays through di.tplog
 true,0,0,q,testcheckrepaironroll[],1,1,rolling into a corrupt log repairs it via di.tplog.check
diff --git a/di/tickerplant/test.q b/di/tickerplant/test.q
index 7611166c..2ce689db 100644
--- a/di/tickerplant/test.q
+++ b/di/tickerplant/test.q
@@ -76,6 +76,18 @@ testinit:{[]
 
 / batch mode: upd stamps, buffers into the root table and logs (bumping j). i is the PUBLISHED
 / watermark and must NOT move while the rows are still buffered - it catches up at the flush
+/ init's OPTIONAL `handlers dep is forwarded verbatim to di.pubsub.init - see the design note in
+/ tickerplant.md. a fresh init WITH a handlers dep must make di.pubsub's own .z.pc hook visible in
+/ di.handlers' own registry (di.handlers.list[`.z.pc] naming `pubsub), which omitting the key (every
+/ other test here) never triggers
+testhandlersforwarded:{[]
+  resetmodule[];
+  dd:freshdir"handlers";
+  `trade set 0#trade;
+  handlers.init[enlist[`log]!enlist caplog[]];
+  tp.init mkdeps[dd;1b],enlist[`handlers]!enlist handlers;
+  0<count select from handlers.list[`.z.pc] where name=`pubsub};
+
 testupdbatch:{[]
   freshinit["updb";1b];
   tp[`upd][`trade;row`AAPL];
diff --git a/di/tickerplant/tickerplant.md b/di/tickerplant/tickerplant.md
index 84a8dd35..4b966608 100644
--- a/di/tickerplant/tickerplant.md
+++ b/di/tickerplant/tickerplant.md
@@ -135,7 +135,8 @@ the two numbers in one reply always agree about what has been published.
 
 Hard (imported via `use` in `init.q`, declared in `deps.q`): `di.pubsub`, `di.eodtime`, `di.tplog`.
 Injected via `init`: `log` and `timer` (both required; validated by `di.depcheck`'s contract check,
-not declared in `deps.q`).
+not declared in `deps.q`), plus an **optional** `handlers` (`di.handlers`' register dict) forwarded
+verbatim to `di.pubsub.init` — see the design note below.
 
 `di.tplog` is used only for check/repair on recovery — when `openlog` finds a pre-existing log it runs
 it through `tplog.check`, repairing a corrupt one. The tickerplant appends to and rolls the log itself
@@ -185,9 +186,15 @@ tickerplant must not do to its own log.
   protocol and the timer job — and deliberately leaves module state and the captured tables intact,
   so a shutdown path can still inspect or save what is buffered. A re-init after a teardown
   re-publishes the protocol and re-schedules the job.
-- **No `di.handlers` dependency.** Subscriber-disconnect cleanup is handled by `di.pubsub`'s own
-  `.z.pc`. (`di.pubsub` should migrate to `di.handlers` so `.z.*` is not assigned outside the central
-  registry — tracked separately, out of scope here.)
+- **`handlers` is optional and only ever forwarded, never consumed directly.** Subscriber-disconnect
+  cleanup is handled by `di.pubsub`'s own `.z.pc` (it self-assigns the event, chaining onto whatever
+  already owned it — `di.tickerplant` itself still takes no direct `di.handlers` dependency and
+  registers nothing of its own). `di.pubsub.init` now accepts an optional `handlers` key (`di.handlers`'
+  register dict) that additionally registers that hook with the central registry, so
+  `di.handlers.list[`.z.pc]` names `pubsub` explicitly instead of only chaining through it invisibly.
+  This module forwards its own optional `handlers` dep straight through, unvalidated — `di.pubsub`
+  validates its own shape — the same way `log` is forwarded to `di.eodtime`/`di.tplog` above. Omitting
+  `handlers` is fully supported: `di.pubsub.init[(::)]` behaves exactly as it always has.
 
 ## Design divergences from TorQ
 
diff --git a/di/tickerplant/tickerplant.q b/di/tickerplant/tickerplant.q
index e9fc09e7..93cb8495 100644
--- a/di/tickerplant/tickerplant.q
+++ b/di/tickerplant/tickerplant.q
@@ -13,9 +13,13 @@
 / protocol di.subscriptions speaks, and an IPC caller reaches them through the default .z.pg/.z.ps,
 / so a module-local binding would be invisible. teardown gives them back.
 / .
-/ NB subscriber-disconnect cleanup is handled by di.pubsub's own .z.pc (it self-assigns it). this
-/ module therefore takes no di.handlers dependency. FLAG: di.pubsub should migrate to di.handlers so
-/ .z.pc is not assigned outside the central registry - out of scope here, tracked separately.
+/ NB subscriber-disconnect cleanup is handled by di.pubsub's own .z.pc (it self-assigns it, chaining
+/ onto whatever already owned the event). di.pubsub's init now ALSO takes an optional `handlers key
+/ (di.handlers' register dict) to register that hook with the central registry so it is visible via
+/ di.handlers.list[`.z.pc], not just chained through it - this module forwards its own OPTIONAL
+/ `handlers dep straight through, unvalidated (di.pubsub validates its own shape), the same way it
+/ forwards `log to di.eodtime/di.tplog below. omitting `handlers here is still fully supported -
+/ di.pubsub.init[(::)] behaves exactly as before
 
 / ============================================================
 / constants (load-time)
@@ -203,8 +207,9 @@ init:{[deps]
   / the tp log, publish the subscription protocol at root and schedule the batch/roll timer job.
   / deps: a dict with `log (required), `timer (required), `schemas (required, tablename!schema) and
   /   optional `batch (1b), `batchperiod (timespan), `logdir (string, "" disables logging),
-  /   `logname (string), `subtables (symbol list), plus di.eodtime keys (rolltimezone/datatimezone/
-  /   rolltimeoffset) forwarded verbatim.
+  /   `logname (string), `subtables (symbol list), `handlers (di.handlers' register dict, forwarded
+  /   verbatim to di.pubsub.init - see the header note), plus di.eodtime keys (rolltimezone/
+  /   datatimezone/rolltimeoffset) forwarded verbatim.
   / dependencies and config are re-applied on EVERY init; RUNTIME state - the date, the message and
   / row counts, and the open log - is seeded only on the FIRST. see the fresh block below
   if[99h<>type deps;
@@ -256,7 +261,9 @@ init:{[deps]
   eodtime.init[(enlist[`log]!enlist deps`log),(key[deps] inter eodtimekeys)#deps];
   tplog.init[enlist[`log]!enlist deps`log];
   pubsub.setsubtables[$[`subtables in key deps;deps`subtables;.z.m.tabs]];
-  pubsub.init[];
+  / `handlers is OPTIONAL and forwarded as-is - di.pubsub.init validates its own shape if given, and
+  / treats the generic-null argument below exactly like the old niladic pubsub.init[] call
+  pubsub.init[$[`handlers in key deps;enlist[`handlers]!enlist deps`handlers;(::)]];
   / RUNTIME state is seeded only on a FRESH init. a re-init - di.torq re-applying config, a config
   / reload, a second wiring - must not rewind a live tickerplant's date, zero the message counts a
   / subscriber replays against, or reopen (and so leak) the log it is already writing to. di.rdb and