clanker-saddle/src/Services/SubtaskEventQueue.vala

using Invercargill;
using Invercargill.DataStructures;
using Inversion;
using ClankerSaddle.Data;

namespace ClankerSaddle.Services {

    /** One subtask status event awaiting delivery to the parent's daemon. */
    public class SubtaskStatusEvent : Object {
        public int64 parent_task_id { get; set; }
        public int64 subtask_id { get; set; }
        public string title { get; set; default = ""; }
        public string status { get; set; default = ""; }
        public string summary { get; set; default = ""; }
    }

    /**
     * Process-wide queue of subtask status events for daemons hosting
     * the parent's live or parked session (wire contract:
     * `POST /task-update {kind: "subtask-status"}`). A singleton by
     * necessity —
     * {@link TaskService} records events from any request or poll while
     * {@link SaddleService} drains them on the dispatcher tick, mirroring
     * the pending-comment bookkeeping in {@link SaddlePollState}.
     *
     * Coalescing: only the LATEST event per subtask stays queued — a
     * burst of transitions collapses to what the parent's daemon needs
     * (the current state), keeping the queue bounded by subtask count.
     * Ids are stringified: the Invercargill collections cannot key on
     * value types.
     */
    public class SubtaskEventQueue : Object {

        private Vector<SubtaskStatusEvent> pending = new Vector<SubtaskStatusEvent>();
        private HashSet<string> queued_ids = new HashSet<string>();
        private Mutex queue_lock = Mutex();

        /**
         * Enqueues `event`, replacing any earlier event for the same
         * subtask (the replacement moves to the back — newest last).
         */
        public void enqueue(SubtaskStatusEvent event) {
            queue_lock.lock();
            string key = event.subtask_id.to_string();
            if (queued_ids.has(key)) {
                pending.remove_first_where(e => e.subtask_id == event.subtask_id);
            } else {
                queued_ids.add(key);
            }
            pending.add(event);
            queue_lock.unlock();
        }

        /** A copy of the pending events, oldest first. */
        public ImmutableLot<SubtaskStatusEvent> snapshot() {
            queue_lock.lock();
            var copy = new Vector<SubtaskStatusEvent>();
            foreach (var event in pending) {
                copy.add(event);
            }
            queue_lock.unlock();
            return copy.to_immutable_buffer();
        }

        /** Removes the (delivered or dropped) event for `subtask_id`. */
        public void discard(int64 subtask_id) {
            queue_lock.lock();
            pending.remove_first_where(e => e.subtask_id == subtask_id);
            queued_ids.remove(subtask_id.to_string());
            queue_lock.unlock();
        }

        /** The number of distinct subtasks awaiting delivery. */
        public uint length {
            get {
                queue_lock.lock();
                uint count = pending.length;
                queue_lock.unlock();
                return count;
            }
        }

        /**
         * Records a dispatched subtask (the allocation claim writes the
         * assigned transition directly, bypassing
         * {@link TaskService.set_status} — this is its hook). The wire
         * value is `working`.
         */
        public void record_dispatch(TaskEntity subtask) {
            if (subtask.parent_task_id <= 0) {
                return;
            }
            enqueue(new SubtaskStatusEvent() {
                parent_task_id = subtask.parent_task_id,
                subtask_id = subtask.id,
                title = subtask.title,
                status = "working",
                summary = subtask.summary ?? ""
            });
        }
    }
}