clanker-saddle/src/Services/Dispatcher.vala

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

namespace ClankerSaddle.Services {

    /**
     * The task dispatcher: a five-second timer that pairs ready tasks with
     * idle saddles, resumes answered-but-paused tasks on the saddle that
     * holds their parked workspace, pushes human comments to busy daemons
     * and keeps busy saddles' output in sync.
     *
     * Each tick resolves fresh request-scoped services (so no database
     * scope is held between ticks) and never overlaps with itself: a slow
     * tick coalesces into the next one.
     */
    public class Dispatcher : Object {

        private Container container = inject<Container>();
        private bool ticking = false;
        private bool warned_no_llm = false;

        /**
         * Task ids that already logged a capability wait. The skip note
         * prints once per task (cleared on dispatch) instead of on every
         * five-second tick.
         */
        private Invercargill.DataStructures.HashSet<string> capability_wait_logged =
            new Invercargill.DataStructures.HashSet<string>();

        /**
         * Consecutive communication failures per saddle id. Transient
         * errors (a pooled connection dropped mid-reuse, a momentary
         * SQLite lock) must not tear down a running session, so a saddle
         * is only declared unreachable after three strikes in a row.
         */
        private Dictionary<string, string> strikes = new Dictionary<string, string>();

        /** Starts the repeating timer. */
        public void start() {
            Timeout.add_seconds(5, () => {
                run_tick.begin();
                return true;
            });
            print("[ClankerSaddle] Dispatcher running (5s interval)\n");
        }

        private async void run_tick() {
            if (ticking) {
                return;
            }
            ticking = true;
            try {
                yield tick();
            } catch (Error e) {
                warning("[ClankerSaddle] Dispatch tick failed: %s", e.message);
            }
            ticking = false;
        }

        private int strike_count(int64 saddle_id) {
            string stored = "";
            return strikes.try_get(saddle_id.to_string(), out stored) ? int.parse(stored) : 0;
        }

        private void record_strike(int64 saddle_id, int count) {
            strikes.set(saddle_id.to_string(), count.to_string());
        }

        /**
          * One dispatch cycle: for every ready task (priority order) grab an
          * idle saddle whose capability set is a superset of the task's
          * required capabilities (tag-based routing — a task with no
          * requirements matches any saddle; a task that cannot be matched
          * is skipped with a log note and lower-priority tasks get their
          * chance), then resume answered-but-paused tasks on the saddle
          * that holds their parked workspace, and poll busy saddles for
          * output and status changes. Assignment pauses while
          * {@link LlmEndpoint} is unconfigured (polling continues).
          */
        public async void tick() throws Error {
            var scope = container.create_scope();
            var orm = scope.resolve<OrmSession>();
            var tasks = scope.resolve<TaskService>();
            var saddles = scope.resolve<SaddleService>();
            var projects = scope.resolve<ProjectService>();
            var capabilities = scope.resolve<CapabilityService>();
            var subtask_events = scope.resolve<SubtaskEventQueue>();

            if (LlmEndpoint.from_environment() == null) {
                if (!warned_no_llm) {
                    warning("[ClankerSaddle] SADDLE_LLM_BASE_URL is not set; task assignment is paused");
                    warned_no_llm = true;
                }
            } else {
                warned_no_llm = false;
                // Stale-busy healing comes first so a freed saddle is
                // dispatchable within the same tick.
                yield saddles.heal_stale_busy_saddles(tasks);
                var ready = yield tasks.get_ready_tasks();
                foreach (var task in ready) {
                    var saddle = yield pick_idle_capable_saddle(saddles, capabilities, task);
                    if (saddle == null) {
                        bool any_idle = (yield saddles.list_idle_saddles()).length > 0;
                        if (!any_idle) {
                            break;
                        }
                        // A capable saddle may free up later; meanwhile
                        // lower-priority tasks still get their turn.
                        if (capability_wait_logged.add(task.id.to_string())) {
                            var required = new Invercargill.DataStructures.Vector<string>();
                            foreach (var capability in yield capabilities.list_for_task(task.id)) {
                                required.add(capability.name);
                            }
                            string required_text = string.joinv(", ", required.to_array());
                            print(@"[ClankerSaddle] Task $(task.id) waits for a capable saddle (requires: $(required_text))\n");
                        }
                        continue;
                    }
                    capability_wait_logged.remove(task.id.to_string());

                    var session = new Session() {
                        task_id = task.id,
                        saddle_id = ((!)saddle).id,
                        started_at = new DateTime.now_local()
                    };
                    yield orm.insert_async<Session>(session);

                    task.status = STATUS_ASSIGNED;
                    task.started_at = new DateTime.now_local();
                    yield orm.update_async<TaskEntity>(task);
                    yield saddles.update_saddle_status(((!)saddle).id, "busy", task.id);
                    subtask_events.record_dispatch(task);

                    try {
                        yield saddles.seed_session_cursor(((!)saddle).id, session.id);
                        yield saddles.push_task_to_saddle(((!)saddle).id, task, session.id, projects);
                        yield saddles.mark_task_comments_delivered(task.id);
                        record_strike(((!)saddle).id, 0);
                        print(@"[ClankerSaddle] Dispatched task $(task.id) to saddle '$(((!)saddle).name)' (session $(session.id))\n");
                    } catch (Error e) {
                        warning(@"[ClankerSaddle] Pushing task $(task.id) to saddle $(((!)saddle).id) failed: $(e.message)");
                        yield abandon_saddle(((!)saddle).id, saddles, tasks);
                    }
                }
            }

            yield resume_answered_tasks(tasks, saddles, projects);
            yield resume_suspended_parents(tasks, saddles, projects);
            yield saddles.deliver_pending_subtask_events(tasks);

            foreach (var saddle in yield saddles.list_busy_saddles()) {
                try {
                    yield saddles.poll_saddle_updates(saddle.id, tasks);
                    yield saddles.deliver_pending_comments(saddle.id, saddle.current_task_id);
                    record_strike(saddle.id, 0);
                } catch (Error e) {
                    int count = strike_count(saddle.id) + 1;
                    record_strike(saddle.id, count);
                    warning(@"[ClankerSaddle] Polling saddle $(saddle.id) failed (attempt $(count)/3): $(e.message)");
                    if (count >= 3) {
                        yield abandon_saddle(saddle.id, saddles, tasks);
                    }
                }
            }
        }

        /**
         * Resumes blocked/review-requested tasks whose newest block has
         * been answered, on the saddle holding their open (parked)
         * session — the workspace-pause counterpart of the dispatch loop.
         * Capabilities do not apply here: the parked workspace pins the
         * task to the saddle that holds it.
         */
        private async void resume_answered_tasks(TaskService tasks, SaddleService saddles,
                ProjectService projects) throws Error {
            var candidates = new Vector<TaskEntity>();
            foreach (var status in new string[] { STATUS_BLOCKED, STATUS_REVIEW_REQUESTED }) {
                foreach (var task in yield tasks.list_tasks(status, null)) {
                    candidates.add(task);
                }
            }

            foreach (var task in candidates) {
                TaskBlock? newest = null;
                foreach (var block in yield tasks.list_blocks(task.id)) {
                    newest = block;
                }
                if (newest == null || ((!)newest).answer == null) {
                    continue;
                }

                var session = yield tasks.open_session_for_task(task.id);
                if (session == null) {
                    continue;
                }
                try {
                    yield saddles.resume_answered_task(task, (!)session, tasks, projects);
                } catch (Error e) {
                    warning(@"[ClankerSaddle] Resuming task $(task.id) failed: $(e.message)");
                }
            }
        }

        /**
         * Resumes suspended parents whose subtasks produced events since
         * they parked (a subtask completed or failed): a fresh /start on
         * the saddle holding the parked workspace, carrying the coalesced
         * markdown status report of every subtask as the resume notice.
         * Resumes on EACH such event — not only when all subtasks finish.
         */
        private async void resume_suspended_parents(TaskService tasks, SaddleService saddles,
                ProjectService projects) throws Error {
            foreach (var parent in yield tasks.list_tasks(STATUS_SUSPENDED, null)) {
                if (!(yield tasks.has_terminal_subtask_since(parent))) {
                    continue;
                }
                var session = yield tasks.open_session_for_task(parent.id);
                if (session == null) {
                    continue;
                }
                try {
                    string notice = yield tasks.subtask_status_report(parent.id);
                    bool resumed = yield saddles.resume_answered_task(parent, (!)session,
                        tasks, projects, notice);
                    if (resumed) {
                        // The parking window is over: a later suspension
                        // starts a fresh event window.
                        yield tasks.clear_suspension(parent.id);
                        print(@"[ClankerSaddle] Resumed suspended task $(parent.id) with a subtask report\n");
                    }
                } catch (Error e) {
                    warning(@"[ClankerSaddle] Resuming suspended task $(parent.id) failed: $(e.message)");
                }
            }
        }

        /**
         * Declares a saddle unreachable once its failures accumulated:
         * the task it held returns to the queue (with a bot comment) and
         * its session closes.
         */
        private async void abandon_saddle(int64 saddle_id, SaddleService saddles,
                TaskService tasks) throws Error {
            record_strike(saddle_id, 0);
            yield saddles.mark_saddle_unreachable(saddle_id, tasks);
        }

        /**
         * The first idle saddle whose capability set covers `task`'s
         * required capabilities (registration order), or null when every
         * idle saddle is missing a tag — the tag-based routing half of
         * the pairing loop.
         */
        private async Saddle? pick_idle_capable_saddle(SaddleService saddles,
                CapabilityService capabilities, TaskEntity task) throws Error {
            foreach (var saddle in yield saddles.list_idle_saddles()) {
                if (yield capabilities.saddle_covers_task(saddle.id, task.id)) {
                    return saddle;
                }
            }
            return null;
        }
    }
}