engine: Varpulis at 4e3dbb5, windows close on event time (varpulis #284, #285) (#3)

A time window in a detect unit closed only when a later event reached that
same window: a count per address waited for the same address's next
failure, so a brute force that ended in a successful logon was never
raised in a running unit. The engine now closes a window when the event
time of the types feeding it passes its end, on any event of those types,
upstream first. A sequence step that names a derived stream now reads
what that stream outputs (over an aggregate it matched the raw events).

e2e/detect gains D6: three failures from one address, a success from it,
then a logon from another address after the window. On the previous
runtime nothing is ever raised; on this one the next logon raises the
brute force with its count. D4 now lists three units.


Claude-Session: https://claude.ai/code/session_01N3K1TGnWTvwYzt9rKuJXES

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: cpoder

CHANGELOG.md

@@ -7,6 +7,22 @@ is `0`, minor versions may carry breaking changes — they are called out here.
 
 ## [Unreleased]
 
+### Fixed — detect units, from the Varpulis engine (varpulis #284, #285)
+- **A time window closes when event time passes its end**, on the next event
+  of the types feeding it, not only when a later event reaches that same
+  window. A window over a filtered stream used to wait for the next event
+  passing the filter, a partitioned one for the next event of the same
+  partition: a brute force counted per address that ended in a successful
+  logon never raised its count in a running unit. A window fed by several
+  types waits for the slowest of them; a type that goes quiet holds back only
+  the windows it feeds (with `.watermark()` it used to hold back every
+  window). Windows close upstream first, so a count's alert reaches the
+  windows and correlations above it in time.
+- **A sequence step that names a derived stream reads what that stream
+  outputs**: over an aggregate it matched the raw events instead of the
+  results, a second `.where()` of the stream was lost, and a first step's own
+  filter replaced the stream's instead of adding to it.
+
 ## [0.3.1] — 2026-09-23
 
 ### Fixed — detect units, from the Varpulis engine (varpulis #277, #279, #281)

core/Cargo.lock

@@ -22,7 +22,7 @@ webpki-roots = "0.26"
 # The Varpulis CEP engine, embedded as a library (ADR-0031): compile a VPL
 # program, feed it events, publish its emits. No async runtime in its tree —
 # its own CI (scripts/check-engine-deps.py) fails if one ever appears.
-varpulis-engine = { git = "https://github.com/varpulis/varpulis", rev = "9e92c0bfaa748217edbb571b9fb4992d731ea0bf" }
+varpulis-engine = { git = "https://github.com/varpulis/varpulis", rev = "4e3dbb5eca2c985bfc5b4f568d1118a96123ef45" }
 
 [[bin]]
 name = "vejas-runtime"

core/Cargo.toml

@@ -50,6 +50,14 @@ machine it connected to: one alert, on `vx.alerts.lateral`.
   `.within()` and window is judged against it, never against the clock, so
   a replay of the same events gives the same alerts. A payload without any
   of them is stamped on arrival.
+- **A window closes when the event time of the types feeding it passes its
+  end**: on any event of those types, not only one of the window's own. A
+  brute force counted per address on Security events is raised by the next
+  Security event of any kind, even when the attacker got in and stopped. A
+  window fed by several types waits for the slowest; one that no event
+  follows stays open, so a source that can go quiet should send a heartbeat
+  of its own type. A stream whose events arrive late declares
+  `.watermark(out_of_order: 30s)` to hold its windows that much longer.
 - **Types.** A string `event_type` in the payload names the event type;
   without it, the type is the one the `.from()` binding declares for that
   subject. The engine's own decoder does this, the same one every Varpulis

docs/book/src/concepts/detects.md

@@ -15,6 +15,9 @@
 #                     come back from the unit's snapshot (round A), and what was
 #                     acked after the last snapshot is replayed from the bus
 #                     (round B) — resume by sequence, ADR-0031 points 4-5
+#   D6 count close    a count per address closes on the next event of its source
+#                     past the window, from any address: a brute force whose
+#                     attacker got in and stopped is still raised
 #   D3 vpl-check      the engine's verdict on the CLI: ok / refused, with exit codes
 #   D4 topology       /topology lists the units under "detects", lang vpl, running
 #
@@ -91,6 +94,30 @@ stream LateralMovement = SmbConnect as smb
     .to(Bus, topic: "vxt.lateral")
 VPL
 
+cat > "$ROOT/detects/bruteforce.vpl" <<'VPL'
+connector Bus = nats (
+    url: "ignored-by-vejas: the bus is NATS_URL"
+)
+
+event Auth:
+    ip: str
+    status: str
+
+stream Logons = Auth
+    .from(Bus, topic: "vxt.auth")
+
+stream Failed = Auth
+    .where(status == "failure")
+
+stream Brute = Failed
+    .partition_by(ip)
+    .window(5m)
+    .aggregate(ip: last(ip), n: count())
+    .where(n >= 3)
+    .emit(rule: "brute_force", ip: ip, n: n)
+    .to(Bus, topic: "vxt.brute")
+VPL
+
 start_nats() { "$NATSD" -js -sd "$WORK/js" -a 127.0.0.1 -p "$NATS_P" > /dev/null 2>&1 & NATS_PID=$!; sleep 0.5; }
 start_rt() {
   VEJAS_ROOT="$ROOT" NATS_URL="$URL" VEJAS_STREAM=TTEST VEJAS_SUBJECT_ROOT=vxt \
@@ -130,14 +157,14 @@ echo "── D4 topology"
 deadline=$((SECONDS+10))
 while [ "$SECONDS" -lt "$deadline" ]; do
   topo=$(curl -s "http://127.0.0.1:$HTTP_P/topology")
-  [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 2 ] && break
+  [ "$(printf '%s' "$topo" | grep -o '"status":"running"' | wc -l)" -ge 3 ] && break
   sleep 0.2
 done
-python3 - "$topo" <<'PY' && ok "two detects listed, lang vpl, running" || bad "topology: $topo"
+python3 - "$topo" <<'PY' && ok "three detects listed, lang vpl, running" || bad "topology: $topo"
 import json, sys
 t=json.loads(sys.argv[1]); d=t.get("detects", [])
 names=sorted(x["name"] for x in d)
-assert names==["detect:lateral","detect:threshold"], names
+assert names==["detect:bruteforce","detect:lateral","detect:threshold"], names
 assert all(x["lang"]=="vpl" for x in d), d
 assert all(x["status"]=="running" for x in d), [x["status"] for x in d]
 PY
@@ -198,6 +225,24 @@ lateral_round backlog 1 12
 [ "$got" -eq "$M" ] && ok "the same $M, and no more, when the second step arrives as one backlog" || { bad "backlog: expected $M lateral alerts, got $got"; diag_lateral; }
 kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.lateral" 2>/dev/null
 
+echo "── D6 a count closes on its source's event time (the attacker stops)"
+( timeout 30 "$NATS" -s "$URL" sub vxt.brute --raw > "$WORK/brute.txt" 2>/dev/null ) & SUB_PID=$!
+sleep 0.5
+auth() { "$NATS" -s "$URL" pub vxt.auth "{\"event_type\":\"Auth\",\"@timestamp\":\"2026-09-21T18:$1Z\",\"ip\":\"$2\",\"status\":\"$3\"}" > /dev/null 2>&1; }
+for t in 00:00 00:10 00:20; do auth "$t" 10.0.0.66 failure; done
+auth 00:30 10.0.0.66 success   # the attacker got in, and stops
+sleep 1.5
+[ "$(grep -c '^{' "$WORK/brute.txt")" -eq 0 ] && ok "nothing while the window runs" || bad "an alert before the window ended: $(cat "$WORK/brute.txt")"
+auth 06:00 10.0.0.7 success    # any later logon, from anyone
+deadline=$((SECONDS+10)); while [ "$SECONDS" -lt "$deadline" ]; do grep -q '10.0.0.66' "$WORK/brute.txt" && break; sleep 0.2; done
+sleep 1
+python3 - "$WORK/brute.txt" <<'PY' && ok "the next logon of anyone raises the brute force, with its count" || bad "brute force: $(cat "$WORK/brute.txt")"
+import json, sys
+alerts=[json.loads(l) for l in open(sys.argv[1]) if l.startswith("{")]
+assert len(alerts)==1 and alerts[0]["ip"]=="10.0.0.66" and alerts[0]["n"]==3, alerts
+PY
+kill "$SUB_PID" 2>/dev/null; pkill -f "[s]ub vxt.brute" 2>/dev/null
+
 echo "── D5 a stateful sequence survives kill -9 (snapshot, resume by sequence)"
 ( timeout 60 "$NATS" -s "$URL" sub vxt.lateral --raw > "$WORK/lateral5.txt" 2>/dev/null ) & SUB_PID=$!
 sleep 0.5