clanker-saddle/src/Services/AllocationService.vala

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

namespace ClankerSaddle.Services {

    /**
     * One offerable task for the wake computation: a fresh queued task
     * (dependencies met, {@link affine_saddle_id} 0) or a parked task
     * ready to resume — blocked/review-requested with its newest block
     * answered, or suspended with a due goal event behind it
     * parked — pinned to the saddle holding its session record
     * (plan 16 D4: parked rounds close 'parked'; the row still names
     * the workspace's saddle).
     */
    public class OfferableTask : Object {
        public TaskEntity task { get; set; }
        /** The saddle whose parked workspace holds the task (0 = fresh). */
        public int64 affine_saddle_id { get; set; }
        /** True when resuming a parked workspace (vs. a fresh dispatch). */
        public bool parked { get; set; }
        /** True when the parked task is a suspended parent's resume. */
        public bool suspended_resume { get; set; }
    }

    /**
     * A claimed allocation: the transactional half of
     * `POST /api/saddle/next-task` succeeded — the task is `assigned`
     * to the asking saddle and its open session row IS the busy record
     * (plan 16 item 23: no saddle-row write remains). The assignment
     * payload is assembled from it afterwards.
     */
    public class AllocationClaim : Object {
        public int64 task_id { get; set; }
        public int64 session_id { get; set; }
        public bool parked { get; set; }
        public bool suspended_resume { get; set; }
        /** The task had never started before this claim (its first dispatch). */
        public bool first_start { get; set; }
    }

    /**
     * The ask/allocation half of the plan-10 work-acquisition flip:
     * daemons ask (`POST /api/saddle/next-task`), the allocation picks
     * the best offerable task for the asking saddle and claims it in a
     * single write transaction, and the dispatcher's wake computation
     * uses the same offerable set to decide which sleeping daemons to
     * wake.
     *
     * Two filters, in priority order:
     *
     * 1. **Workspace affinity** — a task parked (blocked, awaiting
     *    review or suspended) on this saddle is offered ONLY to this
     *    saddle: its workspace lives here, and capabilities do not
     *    apply (the parked session pins the pairing, exactly like the
     *    pre-flip resume paths).
     * 2. **Capability match** — fresh queued tasks (dependencies met)
     *    go to the asking saddle only when its capability set is a
     *    superset of the task's required tags.
     *
     * The claim itself runs inside one `BEGIN IMMEDIATE` transaction
     * with no awaits, so concurrent asks cannot interleave between the
     * decision and the writes: exactly one ask wins a task even when
     * several fire simultaneously. Every candidate is re-validated
     * inside the transaction (a racing ask may have claimed it after
     * the candidate list was built), and a lost race simply falls
     * through to the next candidate.
     */
    public class AllocationService : Object {

        private OrmSession db = inject<OrmSession>();
        private Connection connection = inject<Connection>();
        private TaskService tasks = inject<TaskService>();
        private CapabilityService capabilities = inject<CapabilityService>();
        private GoalService goals = inject<GoalService>();
        private RuleService rules = inject<RuleService>();
        private StatePulse pulse = inject<StatePulse>();
        private PushService push = inject<PushService>();
        private SaddlePollState poll_state = inject<SaddlePollState>();

        /**
         * Allocates the best offerable task to `saddle`, or null when
         * nothing is offerable (the daemon sleeps). The transactional
         * claim runs first; the assignment payload (answers, resume
         * notice, effective rules, project notes, git identity, llm
         * config, session id) is assembled from the claimed rows after
         * the commit.
         */
        public async Json.Object? allocate_next(SaddleService saddles, ProjectService projects,
                Saddle saddle) throws Error {
            if (LlmEndpoint.from_environment() == null) {
                // No LLM endpoint configured: assignment stays paused
                // (the dispatcher logs it once); the daemon sleeps.
                return null;
            }

            var candidates = yield candidates_for(saddle);
            AllocationClaim? claim = null;
            foreach (var candidate in candidates) {
                claim = claim_candidate(saddle, candidate);
                if (claim != null) {
                    break;
                }
            }
            if (claim == null) {
                return null;
            }
            return yield assemble_assignment(saddles, projects, saddle, (!)claim);
        }

        /**
         * Every offerable task (see {@link OfferableTask}) — the
         * dispatcher's wake computation walks this list: parked tasks
         * target their affine saddle first, fresh tasks any capable
         * sleeping one. A parked task without a session record has no
         * workspace to resume and is not offerable.
         */
        public async ImmutableLot<OfferableTask> offerable_tasks() throws Error {
            var rows = new Vector<OfferableTask>();
            foreach (var status in new string[] { STATUS_BLOCKED, STATUS_REVIEW_REQUESTED }) {
                foreach (var task in yield tasks.list_tasks(status, null)) {
                    if (yield tasks.newest_block_answered(task.id)) {
                        var parked = yield parked_offerable(task, false);
                        if (parked != null) {
                            rows.add((!)parked);
                        }
                    }
                }
            }
            // Plan 12: a parked goal agent resumes on its own
            // machinery — task-terminal events under its goal,
            // coalesced ~10 s ({@link GoalService.resume_due}); plain
            // suspended tasks (without one) stay parked until a human
            // or the goal acts.
            foreach (var run in yield tasks.list_tasks(STATUS_SUSPENDED, null)) {
                if (run.kind == TASK_KIND_GOAL_RUN && (yield goals.resume_due(run))) {
                    var parked = yield parked_offerable(run, true);
                    if (parked != null) {
                        rows.add((!)parked);
                    }
                }
            }
            foreach (var task in yield tasks.get_ready_tasks()) {
                rows.add(new OfferableTask() {
                    task = task,
                    affine_saddle_id = 0,
                    parked = false,
                    suspended_resume = false
                });
            }
            return rows.to_immutable_buffer();
        }

        /**
         * The offerable tasks `saddle` may be allocated: affinity
         * candidates first (parked on this saddle, priority order),
         * then fresh queued tasks whose required capabilities are
         * covered (ready order — priority, then creation).
         */
        private async ImmutableLot<OfferableTask> candidates_for(Saddle saddle) throws Error {
            var rows = new Vector<OfferableTask>();
            foreach (var offerable in yield offerable_tasks()) {
                if (offerable.parked) {
                    if (offerable.affine_saddle_id == saddle.id) {
                        rows.add(offerable);
                    }
                } else if (yield capabilities.saddle_covers_task(saddle.id,
                        offerable.task.id)) {
                    rows.add(offerable);
                }
            }
            return rows.to_immutable_buffer();
        }

        /**
         * A parked offerable bound to its session's saddle, or null when
         * the task has no session record — nothing to resume.
         */
        private async OfferableTask? parked_offerable(TaskEntity task, bool suspended_resume)
                 throws Error {
            var session = yield tasks.newest_session_for_task(task.id);
            if (session == null) {
                return null;
            }
            return new OfferableTask() {
                task = task,
                affine_saddle_id = ((!)session).saddle_id,
                parked = true,
                suspended_resume = suspended_resume
            };
        }

        /**
         * The single-transaction claim. `BEGIN IMMEDIATE` takes the
         * SQLite write lock up front; the section is fully synchronous
         * (no awaits), so no other ask, action or tick can interleave
         * between the re-validation reads and the claim writes —
         * concurrent asks race to exactly one winner. Returns null when
         * the candidate was claimed by a racing ask or its saddle
         * became unavailable (the caller falls through to the next
         * candidate).
         */
        private AllocationClaim? claim_candidate(Saddle saddle, OfferableTask candidate) throws Error {
            connection.execute("BEGIN IMMEDIATE");
            try {
                var task = db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(candidate.task.id)))
                    .first();
                if (task == null || !still_claimable((!)task, candidate)) {
                    connection.execute("ROLLBACK");
                    return null;
                }

                var row = db.query<Saddle>()
                    .where(expr("id == $0", elem<int64?>(saddle.id)))
                    .first();
                if (row == null || ((!)row).status == SADDLE_OFFLINE) {
                    // An offline row never wins a claim (the probe
                    // stamp outranks the candidate list); everything
                    // else about availability is DERIVED from the
                    // sessions (plan 16 item 23): the saddle is free
                    // iff it holds no running session — an open
                    // session on an assigned/working task. The
                    // transaction makes the check atomic against
                    // every other ask: two asks racing for one saddle
                    // resolve to exactly one winner, and the parked
                    // resume needs no special case anymore (its own
                    // session closed 'parked' at the park — D4).
                    connection.execute("ROLLBACK");
                    return null;
                }
                foreach (var open in db.query<Session>()
                        .where(expr("saddle_id == $0", elem<int64?>(saddle.id)))
                        .order_by(expr("id"))
                        .materialise()) {
                    if (open.ended_at != null) {
                        continue;
                    }
                    var holder = db.query<TaskEntity>()
                        .where(expr("id == $0", elem<int64?>(open.task_id)))
                        .first();
                    if (holder != null && (((!)holder).status == STATUS_ASSIGNED
                            || ((!)holder).status == STATUS_WORKING)) {
                        connection.execute("ROLLBACK");
                        return null;
                    }
                }

                int64 session_id = 0;
                if (candidate.parked) {
                    var session = db.query<Session>()
                        .where(expr("task_id == $0 && saddle_id == $1",
                            elem<int64?>(candidate.task.id), elem<int64?>(saddle.id)))
                        .order_by(expr("id"))
                        .materialise();
                    Session? newest = null;
                    foreach (var prior in session) {
                        newest = prior;
                    }
                    if (newest == null) {
                        // The parked workspace's session record is gone
                        // (a concurrent teardown); nothing to resume
                        // onto.
                        connection.execute("ROLLBACK");
                        return null;
                    }
                    if (((!)newest).ended_at == null) {
                        // Plan 16 D4: a straggler open row (a raise
                        // racing this claim) closes 'parked' like
                        // every park — the record never dangles.
                        ((!)newest).ended_at = new DateTime.now_local();
                        ((!)newest).outcome = OUTCOME_PARKED;
                        db.update<Session>((!)newest);
                    }
                    // Plan 11 + D4: the resuming assignment carries a
                    // FRESH session id — the parked row closed with
                    // outcome 'parked' at its park (the park record)
                    // and the daemon keys every telemetry push of the
                    // resumed round on the new row.
                    var resumed = new Session() {
                        task_id = candidate.task.id,
                        saddle_id = saddle.id,
                        started_at = new DateTime.now_local()
                    };
                    db.insert<Session>(resumed);
                    session_id = resumed.id;
                } else {
                    var session = new Session() {
                        task_id = candidate.task.id,
                        saddle_id = saddle.id,
                        started_at = new DateTime.now_local()
                    };
                    db.insert<Session>(session);
                    session_id = session.id;
                }

                bool first_start = ((!)task).started_at == null;
                ((!)task).status = STATUS_ASSIGNED;
                ((!)task).started_at = new DateTime.now_local();
                db.update<TaskEntity>((!)task);

                // No saddle-row write remains (plan 16 item 23): the
                // open session this claim just created IS the saddle's
                // busy record — every reader derives it.

                connection.execute("COMMIT");
                return new AllocationClaim() {
                    task_id = candidate.task.id,
                    session_id = session_id,
                    parked = candidate.parked,
                    suspended_resume = candidate.suspended_resume,
                    first_start = first_start
                };
            } catch (Error e) {
                try {
                    connection.execute("ROLLBACK");
                } catch (Error rollback_error) {
                    warning("[ClankerSaddle] Allocation rollback failed: %s",
                        rollback_error.message);
                }
                throw e;
            }
        }

        /**
         * The in-transaction re-validation: the task must still be in
         * its candidate state (a racing ask that claimed it flipped the
         * status), and a fresh candidate's dependencies must still be
         * met.
         */
        private bool still_claimable(TaskEntity task, OfferableTask candidate) throws Error {
            if (candidate.parked) {
                return task.status == STATUS_BLOCKED
                    || task.status == STATUS_REVIEW_REQUESTED
                    || (candidate.suspended_resume && task.status == STATUS_SUSPENDED);
            }
            if (task.status != STATUS_QUEUED) {
                return false;
            }
            if (task.enqueue_after_task_id > 0) {
                // A queued dependent whose predecessor is not done yet
                // (a predecessor set after a manual enqueue) waits.
                var predecessor = db.query<TaskEntity>()
                    .where(expr("id == $0", elem<int64?>(task.enqueue_after_task_id)))
                    .first();
                if (predecessor == null || ((!)predecessor).status != STATUS_DONE) {
                    return false;
                }
            }
            return true;
        }

        /**
         * Assembles the claimed task's assignment payload (today's
         * /start shape): pending block answers ride the task object,
         * and every parked resume carries a `resume_notice` — a
         * goal agent's the coalesced batched report, an answered
         * block's the verdict text in the stable wording ({@link
         * TaskService.verdict_notice_for}) — plus the
         * `branch_advanced` verdict. The hand-over itself starts the
         * session's push-liveness clock (the daemon cannot push before
         * it owns the assignment), so the staleness probe's window
         * opens here.
         */
        private async Json.Object assemble_assignment(SaddleService saddles,
                ProjectService projects, Saddle saddle, AllocationClaim claim) throws Error {
            var detail = yield tasks.get_task(claim.task_id);
            if (detail == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 48,
                    @"Task $(claim.task_id) vanished after its allocation claim");
            }
            var task = ((!)detail).task;

            string? resume_notice = null;
            bool? branch_advanced = null;
            string? context_update = null;
            if (claim.parked) {
                branch_advanced = yield tasks.branch_advanced_since(task.id);
                // The instruction-context delta since the run's last
                // round (plan approval, rules, notes) rides the
                // resume as a turn — null when nothing changed.
                context_update = yield context_update_for(task, claim.session_id, projects);
                if (claim.suspended_resume) {
                    // Plan 12: the goal agent's batched report — every
                    // plan task's status plus the completion reports
                    // that landed while it was parked. Consumes the
                    // pending goal events (the coalescing watermark)
                    // and ends the parking window.
                    resume_notice = yield goals.take_resume_notice(task.goal_id);
                    yield tasks.clear_suspension(task.id);
                } else {
                    // Answering = allocating (plan 11): the verdict rides
                    // the resuming assignment's resume_notice, injected as
                    // the restored conversation's opening turn. Plan 15
                    // C1: the goal-run task's block IS the goal's block
                    // — its own answer is the whole verdict, no mirror
                    // sweep exists.
                    var newest = yield tasks.newest_block(task.id);
                    if (newest != null && ((!)newest).answer != null) {
                        resume_notice = TaskService.verdict_notice_for((!)newest);
                    }
                }
                // Plan 14 B4: persist what the resumed agent first
                // received on the session row — the terminal page's
                // resume-info panel reads it. Fresh sessions keep a
                // null notice.
                if (resume_notice != null && resume_notice.length > 0) {
                    var resumed_row = yield db.query<Session>()
                        .where(expr("id == $0", elem<int64?>(claim.session_id)))
                        .first_async();
                    if (resumed_row != null) {
                        ((!)resumed_row).resume_notice = resume_notice;
                        yield db.update_async<Session>((!)resumed_row);
                    }
                }
            }

            poll_state.note_push(claim.session_id);
            var payload = yield saddles.build_assignment_payload(saddle, task,
                claim.session_id, projects, resume_notice, branch_advanced,
                context_update);
            payload.set_int_member("session_id", claim.session_id);

            yield saddles.mark_task_comments_delivered(task.id);
            yield pulse.bump(task.id, "task dispatched");
            // Web Push (plan 11 task E): the FIRST dispatch of a task
            // notifies its opted-in users — fired from the same place
            // the dispatch pulse bumps (single funnel).
            if (claim.first_start) {
                push.notify_opted_in(task.id, PushService.lifecycle_notification(
                    task.title, task.id, "started", null));
            }
            print(@"[ClankerSaddle] Allocated task $(task.id) to saddle '$(saddle.name)' on ask (session $(claim.session_id))\n");
            return payload;
        }

        // ------------------------------------------------------------------
        // Resume-time context delta (`context_update`)
        // ------------------------------------------------------------------

        /** Clip lengths keeping the delta a SHORT markdown summary. */
        private const int CONTEXT_PLAN_CLIP = 4000;
        private const int CONTEXT_RULE_CLIP = 800;
        private const int CONTEXT_NOTE_CLIP = 800;

        /**
         * The instruction-context delta since the run's last round,
         * as the resuming assignment's `context_update` markdown — or
         * null when nothing changed. The watermark is the previous
         * session's last recorded activity (its newest entry's stamp,
         * else its start): anything the effective instruction set did
         * after that lands in the summary. Covered, simplest-correct:
         *
         * - the goal's plan approved since the run's last round (the
         *   approved plan rides along whole — it is the operative
         *   document the agent must now follow);
         * - rules in the effective set that were created or edited,
         *   or whose admitting attachment (to the task or an involved
         *   project) was created, after the watermark;
         * - project notes of the primary + co-developed projects
         *   created after the watermark.
         *
         * Detaching a rule or deleting a note is NOT detected (the
         * rows carry no such timestamps) — the daemon still receives
         * the current effective `rules`/`project_notes` arrays in the
         * same payload for anything that rebuilds them wholesale.
         */
        private async string? context_update_for(TaskEntity task, int64 session_id,
                ProjectService projects) throws Error {
            // The previous round: the newest session before this
            // claim's fresh one; its last activity is the watermark.
            DateTime? watermark = null;
            int64 previous_session = 0;
            foreach (var session in yield db.query<Session>()
                    .where(expr("task_id == $0", elem<int64?>(task.id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (session.id < session_id && session.started_at != null) {
                    previous_session = session.id;
                    watermark = session.started_at;
                }
            }
            if (previous_session == 0 || watermark == null) {
                return null;
            }
            foreach (var command in yield db.query<SessionCommand>()
                    .where(expr("session_id == $0", elem<int64?>(previous_session)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                DateTime? at = command.completed_at ?? command.started_at;
                if (at != null && ((!)at).compare((!)watermark) > 0) {
                    watermark = at;
                }
            }

            var report = new StringBuilder();

            // Plan approval (goal work only): an approved plan BLOCK
            // (plan 15 C3 — the answered run-task row IS the record)
            // after the watermark names the round that missed it.
            if (task.goal_id > 0) {
                bool approved = false;
                var approval_run = yield db.query<TaskEntity>()
                    .where(expr("goal_id == $0 && kind == $1",
                        elem<int64?>(task.goal_id), elem(TASK_KIND_GOAL_RUN)))
                    .first_async();
                if (approval_run != null) {
                    foreach (var block in yield tasks.list_blocks(
                            ((!)approval_run).id)) {
                        if (block.block_type == BLOCK_PLAN
                                && block.answer == NOTICE_PLAN_APPROVED
                                && block.answered_at != null
                                && ((!)block.answered_at).compare((!)watermark) > 0) {
                            approved = true;
                        }
                    }
                }
                if (approved) {
                    string plan = "";
                    foreach (var goal in yield db.query<Goal>()
                            .where(expr("id == $0", elem<int64?>(task.goal_id)))
                            .materialise_async()) {
                        plan = goal.plan ?? "";
                    }
                    if (plan.length > 0) {
                        report.append("\n## Plan approved\n\n");
                        report.append("The goal's plan was approved since your last round.");
                        report.append(" The approved plan now reads:\n\n");
                        report.append(clip_text(plan, CONTEXT_PLAN_CLIP));
                        report.append("\n");
                    }
                }
            }

            // Rules: anything effective whose row was created/edited,
            // or whose admitting attachment was created, after the
            // watermark.
            var involved = yield co_developed_project_ids(task, projects);
            var changed_rules = new Vector<Rule>();
            foreach (var rule in yield rules.effective_rules_for(task, projects,
                    task.goal_id)) {
                DateTime? rule_stamp = rule.updated_at ?? rule.created_at;
                if (rule_stamp != null && ((!)rule_stamp).compare((!)watermark) > 0) {
                    changed_rules.add(rule);
                    continue;
                }
                if (yield rule_attached_since(rule.id, task.id, involved, (!)watermark)) {
                    changed_rules.add(rule);
                }
            }
            if (changed_rules.length > 0) {
                report.append("\n## Rules changed\n\n");
                report.append("New or updated rules now apply to your work:\n\n");
                foreach (var rule in changed_rules) {
                    report.append_printf("- **%s**: %s\n", rule.name,
                        clip_text(rule.body, CONTEXT_RULE_CLIP));
                }
            }

            // Project notes of the primary + co-developed projects.
            var changed_notes = new Vector<ProjectNote>();
            var note_project = new Dictionary<string, string>();
            foreach (var project_id in involved) {
                var project = yield db.query<Project>()
                    .where(expr("id == $0", elem<int64?>(int64.parse(project_id))))
                    .first_async();
                if (project == null) {
                    continue;
                }
                note_project.set(project_id, ((!)project).name);
                foreach (var note in yield projects.list_notes(int64.parse(project_id))) {
                    if (note.created_at != null
                            && ((!)note.created_at).compare((!)watermark) > 0) {
                        changed_notes.add(note);
                    }
                }
            }
            if (changed_notes.length > 0) {
                report.append("\n## New project notes\n\n");
                foreach (var note in changed_notes) {
                    report.append_printf("- **%s**: %s\n",
                        note_project.get_or_default(note.project_id.to_string()) ?? "",
                        clip_text(note.body, CONTEXT_NOTE_CLIP));
                }
            }

            if (report.len == 0) {
                return null;
            }
            var summary = new StringBuilder();
            summary.append("Context changes since your last round (your instructions changed while you were away):\n");
            summary.append(report.str);
            return summary.str;
        }

        /**
         * The primary + co-developed project ids (as id strings — the
         * container-safe element type) of `task`'s effective context:
         * through the GOAL's links for goal work, the project's links
         * otherwise (the note/rule scopes of the assignment payload).
         */
        private async Vector<string> co_developed_project_ids(TaskEntity task,
                ProjectService projects) throws Error {
            var ids = new Vector<string>();
            ids.add(task.project_id.to_string());
            if (task.goal_id > 0) {
                foreach (var link in yield goals.list_links(task.goal_id)) {
                    if (link.relationship == LINK_CO_DEVELOPED) {
                        ids.add(link.project_id.to_string());
                    }
                }
            } else {
                foreach (var link in yield projects.list_links(task.project_id)) {
                    if (link.relationship == LINK_CO_DEVELOPED) {
                        ids.add(link.linked_project_id.to_string());
                    }
                }
            }
            return ids;
        }

        /**
         * True when one of the rule's admitting attachments — a
         * `task_rules` row for this task, or a `rule_projects` row
         * for an involved project — was created after `watermark`.
         */
        private async bool rule_attached_since(int64 rule_id, int64 task_id,
                Vector<string> involved_projects, DateTime watermark) throws Error {
            foreach (var link in yield db.query<TaskRule>()
                    .where(expr("rule_id == $0 && task_id == $1",
                        elem<int64?>(rule_id), elem<int64?>(task_id)))
                    .materialise_async()) {
                if (link.created_at != null
                        && ((!)link.created_at).compare(watermark) > 0) {
                    return true;
                }
            }
            foreach (var project_id in involved_projects) {
                foreach (var link in yield db.query<RuleProject>()
                        .where(expr("rule_id == $0 && project_id == $1",
                            elem<int64?>(rule_id), elem<int64?>(int64.parse(project_id))))
                        .materialise_async()) {
                    if (link.created_at != null
                            && ((!)link.created_at).compare(watermark) > 0) {
                        return true;
                    }
                }
            }
            return false;
        }

        /** A body clipped to `limit` chars (ellipsis when it cut). */
        private static string clip_text(string body, int limit) {
            return body.length > limit ? body.substring(0, limit) + "…" : body;
        }
    }
}