using Invercargill.DataStructures; using Inversion; namespace ClankerSaddle.Services { /** * Process-wide saddle poll bookkeeping: the block answers and human * comments already handed to a saddle (each delivered exactly once; * a restart re-delivers and the daemon deduplicates), plus the * plan-10 ask/wake liveness bookkeeping: the outcome of a saddle's * last `POST /api/saddle/next-task` ask (a saddle counts as SLEEPING * when its last ask returned 204) and the pending wake each sleeping * saddle was sent (the lost-wake guard's timer), and the phase-2 * push-liveness clock: the last chunks/results arrival per session * (the staleness probe's timer — one exceptional probe per silent * episode, reset by the next push). * * The plan-13 entry-ingestion dedup deliberately does NOT live * here: it is daemon-global and persistent (the * `saddle_entry_dedup` table), so it survives restarts and session * churn — see SaddleService.store_pushed_entries. * * A singleton by necessity: {@link SaddleService} is request-scoped, * so this object is where the cross-request cursors live. Ids are * stringified: the Invercargill collections cannot key on value types. */ public class SaddlePollState : Object { private Dictionary probe_state = new Dictionary(); private Dictionary probe_uptime = new Dictionary(); private Dictionary probe_activity = new Dictionary(); private Dictionary probe_contact = new Dictionary(); private HashSet delivered_answers = new HashSet(); private HashSet delivered_comments = new HashSet(); private HashSet resumed_notified = new HashSet(); private Dictionary ask_outcome = new Dictionary(); private HashSet ask_seen = new HashSet(); private Dictionary wake_at_ms = new Dictionary(); private Dictionary wake_task = new Dictionary(); private Dictionary push_at_ms = new Dictionary(); private HashSet push_seen = new HashSet(); private HashSet stale_probed = new HashSet(); // Plan 15 C4: the daemon-reported per-repo push flags of the // goal branches, as the LAST /status probe saw them — live // probe state only, NEVER a stored snapshot on the goal row. private Dictionary probe_goal_branches = new Dictionary(); private Mutex state_lock = Mutex(); /** * Records a daemon /status observation for the saddles-page * health panel: reported state, uptime, last-activity epoch and * the web-side contact time (monotonic ms). */ public void record_probe(int64 saddle_id, string state, int64 uptime_seconds, int64 last_activity_epoch) { state_lock.lock(); string key = saddle_id.to_string(); probe_state.set(key, state); probe_uptime.set(key, uptime_seconds.to_string()); probe_activity.set(key, last_activity_epoch.to_string()); probe_contact.set(key, (get_monotonic_time() / 1000).to_string()); state_lock.unlock(); } /** * Records the daemon-reported goal-branch push flags (plan 15 * C4): raw JSON as the probed daemon sent it — the goal page * and the completion-accept gate read it back live. A missing * member clears the record (nothing to render or enforce). */ public void record_goal_branches(int64 saddle_id, string raw_json) { state_lock.lock(); probe_goal_branches.set(saddle_id.to_string(), raw_json); state_lock.unlock(); } /** The last probed goal-branch flags ("" when never reported). */ public string goal_branches_for(int64 saddle_id) { state_lock.lock(); string stored = ""; bool found = probe_goal_branches.try_get(saddle_id.to_string(), out stored); state_lock.unlock(); return found ? stored : ""; } /** The last probe snapshot for `saddle_id` ("" when never probed). */ public void health_for(int64 saddle_id, out string state, out int64 uptime, out int64 activity, out int64 contact_ms) { state_lock.lock(); string key = saddle_id.to_string(); string stored = ""; state = probe_state.try_get(key, out stored) ? stored : ""; uptime = int64.parse(probe_uptime.get_or_default(key) ?? "0"); activity = int64.parse(probe_activity.get_or_default(key) ?? "0"); contact_ms = int64.parse(probe_contact.get_or_default(key) ?? "0"); state_lock.unlock(); } /** * Marks `session_id`'s daemon-side resume as notified; returns * false when the progress entry already went out. */ public bool mark_resumed(int64 session_id) { state_lock.lock(); bool already = resumed_notified.has(session_id.to_string()); if (!already) { resumed_notified.add(session_id.to_string()); } state_lock.unlock(); return !already; } /** * Marks `block_id`'s answer as delivered; returns false when it * had already been handed over. */ public bool mark_delivered(int64 block_id) { state_lock.lock(); bool already = delivered_answers.has(block_id.to_string()); if (!already) { delivered_answers.add(block_id.to_string()); } state_lock.unlock(); return !already; } /** True when `block_id`'s answer was already handed over. */ public bool is_delivered(int64 block_id) { state_lock.lock(); bool already = delivered_answers.has(block_id.to_string()); state_lock.unlock(); return already; } /** * Marks `comment_id` as delivered to its saddle; returns false * when it had already been handed over. */ public bool mark_comment_delivered(int64 comment_id) { state_lock.lock(); bool already = delivered_comments.has(comment_id.to_string()); if (!already) { delivered_comments.add(comment_id.to_string()); } state_lock.unlock(); return !already; } /** True when `comment_id` was already handed over. */ public bool is_comment_delivered(int64 comment_id) { state_lock.lock(); bool already = delivered_comments.has(comment_id.to_string()); state_lock.unlock(); return already; } /** * Records an ask arriving from `saddle_id` (plan 10): the * pending wake is answered — the lost-wake guard must not fire * for it — and the saddle is awake until the ask resolves. The * first ask also marks the daemon ask-capable (post-flip): its * parked-task resumes flow through wake + next-task from then * on, never through /start pushes. */ public void mark_ask(int64 saddle_id) { state_lock.lock(); string key = saddle_id.to_string(); ask_outcome.set(key, "awake"); ask_seen.add(key); wake_at_ms.remove(key); wake_task.remove(key); state_lock.unlock(); } /** * Records the outcome of `saddle_id`'s ask: `204` puts it to * sleep (a later wake targets it), anything else leaves it * awake. */ public void mark_ask_result(int64 saddle_id, bool assigned) { state_lock.lock(); ask_outcome.set(saddle_id.to_string(), assigned ? "assigned" : "204"); state_lock.unlock(); } /** * True when `saddle_id`'s last ask returned 204 — the sleeping * state a wake targets. */ public bool is_sleeping(int64 saddle_id) { state_lock.lock(); string stored = ""; bool sleeping = ask_outcome.try_get(saddle_id.to_string(), out stored) && stored == "204"; state_lock.unlock(); return sleeping; } /** * True once `saddle_id` has ever asked (post-flip daemon): it * asks whenever idle-awake, so its work — fresh or resumed — * always flows through the allocation. */ public bool has_ever_asked(int64 saddle_id) { state_lock.lock(); bool seen = ask_seen.has(saddle_id.to_string()); state_lock.unlock(); return seen; } /** Forgets the ask state (the saddle went offline / was removed). */ public void clear_ask(int64 saddle_id) { state_lock.lock(); string key = saddle_id.to_string(); ask_outcome.remove(key); wake_at_ms.remove(key); wake_task.remove(key); state_lock.unlock(); } /** * Records a wake sent to `saddle_id` for `task_id` — starts the * lost-wake guard's timer. */ public void note_wake(int64 saddle_id, int64 task_id) { state_lock.lock(); string key = saddle_id.to_string(); wake_at_ms.set(key, (get_monotonic_time() / 1000).to_string()); wake_task.set(key, task_id.to_string()); state_lock.unlock(); } /** * True when `saddle_id` has a pending wake at least * `guard_ms` old — an ask should have arrived by now; the * guard declares the saddle offline and frees the work for * the next capable one. */ public bool wake_overdue(int64 saddle_id, uint guard_ms) { state_lock.lock(); string stored = ""; bool overdue = wake_at_ms.try_get(saddle_id.to_string(), out stored) && (get_monotonic_time() / 1000 - int64.parse(stored)) >= guard_ms; state_lock.unlock(); return overdue; } /** * True when `saddle_id` carries a pending wake younger than * `guard_ms` — the wake pass skips it instead of re-waking on * every tick while the guard window runs. */ public bool wake_pending(int64 saddle_id, uint guard_ms) { state_lock.lock(); string stored = ""; bool pending = wake_at_ms.try_get(saddle_id.to_string(), out stored) && (get_monotonic_time() / 1000 - int64.parse(stored)) < guard_ms; state_lock.unlock(); return pending; } /** The task id `saddle_id` was last woken for (0 when none). */ public int64 wake_task_of(int64 saddle_id) { state_lock.lock(); string stored = ""; bool found = wake_task.try_get(saddle_id.to_string(), out stored); state_lock.unlock(); return found ? int64.parse(stored) : 0; } /** * Records a chunks/results push arriving for `session_id` * (plan 10 phase 2): the running daemon is alive, and any * running staleness episode ends — a later silent episode * earns its own exceptional probe. */ public void note_push(int64 session_id) { state_lock.lock(); string key = session_id.to_string(); push_at_ms.set(key, (get_monotonic_time() / 1000).to_string()); push_seen.add(key); stale_probed.remove(key); state_lock.unlock(); } /** * Atomically decides the one exceptional staleness probe for * `session_id`: true exactly when pushes have been seen, the * silence has reached `stale_ms` and no probe fired for this * episode yet (the flag flips here, so a slow probe is never * double-fired by the next tick). A session with no recorded * push (a web restart wiped the clock) is seeded now and * starts its window instead of firing immediately — a fresh * baseline must not masquerade as 90 s of silence. */ public bool take_stale_probe(int64 session_id, uint stale_ms) { state_lock.lock(); string key = session_id.to_string(); int64 now_ms = get_monotonic_time() / 1000; if (!push_seen.has(key)) { push_seen.add(key); push_at_ms.set(key, now_ms.to_string()); state_lock.unlock(); return false; } if (stale_probed.has(key)) { state_lock.unlock(); return false; } string stored = ""; bool known = push_at_ms.try_get(key, out stored); bool due = known && (now_ms - int64.parse(stored)) >= stale_ms; if (due) { stale_probed.add(key); } state_lock.unlock(); return due; } } }