Reproduction steps:
- Use a contract with a window and a seen param:
inc = {"cursor_field":"id","seen_window_seconds":3600,
"since_param":"since","seen_param":"exclude_ids"}
- Write a watermark record with one in-window and one expired cursor
(now = 2026-08-14T16:00:00Z, window = 1h):
rec = {"watermark":"2026-08-14T09:00:00Z",
"seen":[["row_recent","2026-08-14T15:50:00Z"],
["row_expired","2026-08-14T10:00:00Z"]]}
- Scheduled path injection: watermark_store.active_seen(rec, inc, now)
- Manual path injection: place rec on disk under a schedule id and call
incremental_runtime.read_only_since(ws, meta, inc); take the returned set.
Expected:
Both paths exclude the same cursors — they read the same state under the same
contract, and a cursor past its window should be absent from both.
Actual:
Scheduled active_seen(...) -> ['row_recent'] (row_expired droppable)
Manual read_only_since(...) -> ['row_expired','row_recent'] (window ignored)
v0.99 made the scheduled seen_param window-filter via active_seen(), but the
manual branch of incremental_runtime.prepare() still injects sorted(seen)
straight from read_only_since(), which unions cursors across schedules and
applies no window filter. A row whose seen window has elapsed is re-deliverable
on a scheduled run but permanently suppressed on a manual one — and clicking Run
by hand is the action taken to re-deliver it.
Suggested fix:
Filter the manual path with the same helper. read_only_since() already reads
each record, so apply the window per record before the union:
for f in glob.glob(os.path.join(d, "*.json")):
rec = json.load(open(f)) or {}
if rec.get("watermark"):
marks.append(rec["watermark"])
seen |= WM.active_seen(rec, inc, now) # was: seen.add(str(_e[0]) ...)
Thread now into read_only_since() (or let active_seen default to it); the
manual injection then stays sorted(seen). Keeps the intended cross-schedule
union while making both paths agree on expiry.
Station version (railcall version): station-v0.99
Module slug + version: N/A (platform — workbench/primitives/incremental_runtime.py)