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

namespace ClankerSaddle.Services {

    /** Valid goal statuses (the `goals.status` values). */
    public const string GOAL_DRAFT = "draft";
    public const string GOAL_PLANNING = "planning";
    public const string GOAL_PLAN_REVIEW = "plan_review";
    public const string GOAL_ACTIVE = "active";
    public const string GOAL_PAUSED = "paused";
    public const string GOAL_COMPLETION_REVIEW = "completion_review";
    public const string GOAL_COMPLETE = "complete";
    public const string GOAL_FAILED = "failed";
    public const string GOAL_CANCELLED = "cancelled";

    /**
     * Valid goal block types (plan 15 C1: the values of the goal-run
     * task's `task_blocks.type` rows — question, rfc, plan and the
     * completion review).
     */
    public const string GOAL_BLOCK_QUESTION = "question";
    public const string GOAL_BLOCK_RFC = "rfc";
    public const string GOAL_BLOCK_PLAN = "plan";
    public const string GOAL_BLOCK_COMPLETION = "completion";

    /** Task kinds (the `tasks.kind` wire values). */
    public const string TASK_KIND_TASK = "task";
    public const string TASK_KIND_GOAL_RUN = "goal_run";

    /** Task creators: enqueueing is creator-only. */
    public const string CREATOR_HUMAN = "human";
    public const string CREATOR_GOAL = "goal";

    /**
     * The goal-run internal task's mode: the goal agent orchestrates
     * (prompt-level mode shaping is the daemon's job).
     */
    public const string MODE_ORCHESTRATE = "orchestrate";

    /** Valid work modes for human/goal-created tasks. */
    public const string[] TASK_MODES = { "implement", "investigate", "review", "test" };

    /** The priority every goal-run internal task dispatches at. */
    public const int64 GOAL_RUN_PRIORITY = 3;

    /**
     * Goal storage and lifecycle (plan 12; blocks consolidated plan
     * 15 C1): creation with branch-name uniqueness validation, the
     * draft → planning → plan review → active → completion review →
     * complete/failed/cancelled flow (with paused freezing
     * supervision in between), the goal agent's run task, plan/
     * completion review flows with revisions, goal comments, the
     * cancel cascade and the event-driven resume machinery (task-
     * terminal events coalesce ~10 s into one goal-agent resume
     * carrying the batched status report). The goal agent's BLOCKS
     * ride the run task's own `task_blocks` rows — one row per
     * question, answered in place through {@link answer_block}.
     */
    public class GoalService : Object {

        private OrmSession db = inject<OrmSession>();
        private StatePulse pulse = inject<StatePulse>();
        private PushService push = inject<PushService>();
        private TaskService tasks = inject<TaskService>();

        /**
         * SADDLE_GOAL_RESUME_S (default 10): the coalescing window —
         * a parked goal agent resumes once the OLDEST unconsumed task
         * event is at least this old, so bursts collapse into one
         * resume.
         */
        public static uint goal_resume_coalesce_seconds() {
            var raw = Environment.get_variable("SADDLE_GOAL_RESUME_S") ?? "";
            int64 parsed = int64.parse(raw);
            return parsed > 0 ? (uint) parsed : 10;
        }

        // ------------------------------------------------------------------
        // Creation and links
        // ------------------------------------------------------------------

        /**
         * Creates a goal in draft. The branch name is required and
         * uniqueness-validated: no other live (non-cancelled,
         * non-complete, non-failed) goal may use the same branch name
         * while sharing any involved project (primary or link) — one
         * branch of that name per involved repo. `base_branch` defaults
         * to the project's default branch.
         */
        public async Goal create_goal(int64 project_id, string title, string brief,
                string branch_name, string? base_branch, int64 priority,
                string creator = CREATOR_HUMAN) throws Error {
            var project = yield db.query<Project>()
                .where(expr("id == $0", elem<int64?>(project_id)))
                .first_async();
            if (project == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 60,
                    @"No project exists with id $(project_id)");
            }
            string trimmed_branch = branch_name.strip();
            if (title.strip().length == 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 61,
                    "A goal title is required");
            }
            if (brief.strip().length == 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 62,
                    "An initial brief is required");
            }
            if (trimmed_branch.length == 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 63,
                    "A goal branch name is required");
            }

            foreach (var other in yield db.query<Goal>().materialise_async()) {
                if (other.branch_name != trimmed_branch
                        || other.status == GOAL_CANCELLED
                        || other.status == GOAL_COMPLETE
                        || other.status == GOAL_FAILED) {
                    continue;
                }
                if ((yield involved_project_ids(other)).contains(project_id.to_string())) {
                    throw new Error(Quark.from_string("clanker-saddle"), 64,
                        @"Branch name '$(trimmed_branch)' is already used by a live goal of an involved project (#$(other.id) $(other.title)) — pick another name");
                }
            }

            var goal = new Goal() {
                project_id = project_id,
                title = title.strip(),
                brief = brief,
                branch_name = trimmed_branch,
                base_branch = base_branch != null && base_branch.strip().length > 0
                    ? base_branch.strip() : ((!)project).default_branch,
                priority = priority.clamp(1, 9),
                creator = creator,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<Goal>(goal);
            yield record_event(goal.id, "created", @"Goal created (branch $(trimmed_branch))");
            yield pulse.bump(0, "goal created");
            return goal;
        }

        /** Loads the goal identified by `id`, or null. */
        public async Goal? get_goal(int64 id) throws Error {
            return yield db.query<Goal>()
                .where(expr("id == $0", elem<int64?>(id)))
                .first_async();
        }

        /** Every goal, newest first. */
        public async ImmutableLot<Goal> list_goals() throws Error {
            return yield db.query<Goal>()
                .order_by_desc(expr("id"))
                .materialise_async();
        }

        /**
         * The primary + linked project ids of `goal` (stringified —
         * the Invercargill collections cannot hash value types).
         */
        public async Invercargill.DataStructures.HashSet<string> involved_project_ids(Goal goal) throws Error {
            var ids = new Invercargill.DataStructures.HashSet<string>();
            ids.add(goal.project_id.to_string());
            foreach (var link in yield list_links(goal.id)) {
                ids.add(link.project_id.to_string());
            }
            return ids;
        }

        /** Adds a project link to the goal; duplicates are refused. */
        public async void add_link(int64 goal_id, int64 project_id,
                string relationship) throws Error {
            if (relationship != LINK_CO_DEVELOPED && relationship != LINK_REFERENCE) {
                throw new Error(Quark.from_string("clanker-saddle"), 65,
                    @"Relationship must be '$(LINK_CO_DEVELOPED)' or '$(LINK_REFERENCE)'");
            }
            var goal = yield get_goal(goal_id);
            if (goal == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 66,
                    @"No goal exists with id $(goal_id)");
            }
            if (project_id == ((!)goal).project_id) {
                throw new Error(Quark.from_string("clanker-saddle"), 67,
                    "The primary project is implicitly linked");
            }
            var existing = yield db.query<GoalLink>()
                .where(expr("goal_id == $0 && project_id == $1",
                    elem<int64?>(goal_id), elem<int64?>(project_id)))
                .materialise_async();
            if (existing.length > 0) {
                throw new Error(Quark.from_string("clanker-saddle"), 68,
                    "That project is already linked to the goal");
            }
            yield db.insert_async<GoalLink>(new GoalLink() {
                goal_id = goal_id,
                project_id = project_id,
                relationship = relationship
            });
            yield pulse.bump(0, "goal link added");
        }

        /** Removes the goal link identified by `id`. */
        public async void remove_link(int64 link_id) throws Error {
            var link = yield db.query<GoalLink>()
                .where(expr("id == $0", elem<int64?>(link_id)))
                .first_async();
            if (link != null) {
                yield db.delete_async<GoalLink>((!)link);
                yield pulse.bump(0, "goal link removed");
            }
        }

        /** The goal's links, oldest first. */
        public async ImmutableLot<GoalLink> list_links(int64 goal_id) throws Error {
            return yield db.query<GoalLink>()
                .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        // ------------------------------------------------------------------
        // Lifecycle
        // ------------------------------------------------------------------

        /**
         * The goal's goal-run internal task (kind 'goal_run'), or null
         * before the goal starts.
         */
        public async TaskEntity? goal_run_task(int64 goal_id) throws Error {
            var rows = yield db.query<TaskEntity>()
                .where(expr("goal_id == $0 && kind == $1",
                    elem<int64?>(goal_id), elem(TASK_KIND_GOAL_RUN)))
                .order_by(expr("id"))
                .materialise_async();
            TaskEntity? first = null;
            foreach (var row in rows) {
                first = row;
                break;
            }
            return first;
        }

        /**
         * Starts a draft goal: the brief locks, the goal moves to
         * 'planning' and the goal agent's internal run task is created
         * (kind 'goal_run', mode 'orchestrate', priority 3, creator
         * 'goal') and queued — the dispatcher hands it to a saddle and
         * the agent begins investigating.
         */
        public async Goal start_goal(int64 goal_id) throws Error {
            var goal = yield require_goal(goal_id);
            if (((!)goal).status != GOAL_DRAFT) {
                throw new Error(Quark.from_string("clanker-saddle"), 69,
                    "Only draft goals can be started");
            }
            ((!)goal).status = GOAL_PLANNING;
            ((!)goal).started_at = new DateTime.now_local();
            yield db.update_async<Goal>((!)goal);

            var run = yield tasks.create_task(((!)goal).project_id,
                @"Goal agent — $(((!)goal).title)",
                ((!)goal).brief, GOAL_RUN_PRIORITY,
                null, ((!)goal).base_branch, STATUS_QUEUED,
                TASK_KIND_GOAL_RUN, MODE_ORCHESTRATE, CREATOR_GOAL,
                ((!)goal).id, 0);
            yield record_event(goal_id, "started",
                @"Goal started — the goal agent is planning (run task #$(run.id))", run.id);
            yield pulse.bump(run.id, "goal started");
            return goal;
        }

        /**
         * Pauses an active goal: supervision freezes — no goal-agent
         * resumes are scheduled while paused; task events keep
         * queueing and coalesce into the flush on resume.
         */
        public async Goal pause_goal(int64 goal_id) throws Error {
            var goal = yield require_goal(goal_id);
            if (((!)goal).status != GOAL_ACTIVE) {
                throw new Error(Quark.from_string("clanker-saddle"), 70,
                    "Only active goals can be paused");
            }
            ((!)goal).status = GOAL_PAUSED;
            ((!)goal).paused_at = new DateTime.now_local();
            yield db.update_async<Goal>((!)goal);
            yield record_event(goal_id, "paused", "Goal paused — supervision frozen (events queue)");
            yield pulse.bump(0, "goal paused");
            return goal;
        }

        /**
         * Resumes a paused goal: supervision restarts and any events
         * that queued while paused coalesce into the next resume (the
         * oldest pending event usually already satisfies the window).
         */
        public async Goal resume_goal(int64 goal_id) throws Error {
            var goal = yield require_goal(goal_id);
            if (((!)goal).status != GOAL_PAUSED) {
                throw new Error(Quark.from_string("clanker-saddle"), 71,
                    "Only paused goals can be resumed");
            }
            ((!)goal).status = GOAL_ACTIVE;
            ((!)goal).paused_at = null;
            yield db.update_async<Goal>((!)goal);
            yield record_event(goal_id, "resumed", "Goal resumed — queued events coalesce into the next agent wake");
            yield pulse.bump(0, "goal resumed");
            return goal;
        }

        /**
         * Cancels a goal: the cascade closes every one of its tasks —
         * non-terminal ones are cancelled (the running goal agent
         * included), draft ones deleted (they never ran). Terminal
         * goals refuse the action. Live daemons are NOT aborted here —
         * the cancel action (which can reach SaddleService without a
         * dependency cycle) aborts them first.
         */
        public async void cancel_goal(int64 goal_id) throws Error {
            var goal = yield require_goal(goal_id);
            string status = ((!)goal).status;
            if (status == GOAL_COMPLETE || status == GOAL_FAILED
                    || status == GOAL_CANCELLED) {
                throw new Error(Quark.from_string("clanker-saddle"), 72,
                    "The goal is already finished");
            }
            ((!)goal).status = GOAL_CANCELLED;
            ((!)goal).completed_at = new DateTime.now_local();
            yield db.update_async<Goal>((!)goal);

            foreach (var task in yield db.query<TaskEntity>()
                    .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (task.status == STATUS_DONE || task.status == STATUS_FAILED) {
                    continue;
                }
                if (task.status == STATUS_DRAFT) {
                    yield tasks.delete_task_quietly(task.id);
                } else {
                    yield tasks.cancel_task(task.id);
                }
            }
            yield record_event(goal_id, "cancelled", "Goal cancelled — its tasks were cancelled with it");
            yield pulse.bump(0, "goal cancelled");
        }

        // ------------------------------------------------------------------
        // The goal agent's pushes: plan, completion, failure
        // ------------------------------------------------------------------

        /**
         * The goal agent submitted (or revised) its plan: ONE
         * review-kind block of type 'plan' on the goal-run task
         * carries the markdown, the goal moves to 'plan_review' and
         * the run parks on it (the verdict resume rides the answered
         * block). The caller has already parked the task via the
         * results state. Plan 15 C1: the block lives on the run task
         * — there is no second row to keep in sync.
         */
        public async TaskBlock? submit_plan(int64 goal_id, string markdown) throws Error {
            var goal = yield require_goal(goal_id);
            // Idempotent on the pending submission: a redelivered push
            // (outbox retry, the entry + state pair of one `saddle
            // plan`) must not stack duplicate revisions — a NEW row
            // only appears once the previous one was answered
            // (rejected → the agent revises).
            TaskBlock? raised = yield raise_run_block(((!)goal), "Plan submitted for review",
                GOAL_BLOCK_PLAN, null, markdown, true);
            if (raised == null || ((!)raised).answer != null) {
                // No pending submission: a fresh revision supersedes
                // the answered history.
                raised = yield raise_run_block(((!)goal), "Plan submitted for review",
                    GOAL_BLOCK_PLAN, null, markdown, false);
            }
            if (raised == null) {
                return null;
            }
            ((!)goal).status = GOAL_PLAN_REVIEW;
            yield db.update_async<Goal>(((!)goal));
            yield record_event(goal_id, "plan-submitted",
                @"Plan revision #$(((!)raised).id) submitted for review");
            return raised;
        }

        /**
         * Records a goal-agent question/RFC as ONE block on the
         * goal-run task and parks it there. Idempotent on the pending
         * question: a redelivered blocked push (outbox retry, the
         * park-unwinding round-end push) must not stack duplicates.
         */
        public async TaskBlock? raise_agent_block(int64 goal_id, string question,
                string block_type, string[]? options = null, string? body = null) throws Error {
            var goal = yield require_goal(goal_id);
            TaskBlock? raised = yield raise_run_block(((!)goal), question,
                block_type.length > 0 ? block_type : GOAL_BLOCK_QUESTION,
                options, body, false);
            if (raised == null) {
                return null;
            }
            yield record_event(goal_id, "block-raised",
                @"$(((!)raised).block_type) raised: $(question)");
            return raised;
        }

        /**
         * The goal agent submitted its completion report (`saddle
         * done` in goal context): the goal moves to
         * 'completion_review', the report is stored (goal row + ONE
         * completion-kind block on the goal-run task) and the run
         * parks on it. The daemon enforces the branch-push rule
         * before letting `done` through. Idempotent on terminal goals
         * (plan 13 B2): a redelivered done push must not resurrect a
         * finished goal.
         */
        public async TaskBlock? submit_completion(int64 goal_id, string report) throws Error {
            var goal = yield require_goal(goal_id);
            if (((!)goal).status == GOAL_COMPLETE || ((!)goal).status == GOAL_FAILED
                    || ((!)goal).status == GOAL_CANCELLED) {
                // Terminal stands: the pending-report idempotency below
                // cannot apply (the verdict already landed).
                throw new Error(Quark.from_string("clanker-saddle"), 85,
                    @"The goal is already $(((!)goal).status) — its completion review is closed");
            }
            // Idempotent on the pending report (see submit_plan).
            TaskBlock? raised = yield raise_run_block(((!)goal), "Goal completion report",
                GOAL_BLOCK_COMPLETION, null, report, true);
            if (raised == null) {
                return null;
            }
            ((!)goal).status = GOAL_COMPLETION_REVIEW;
            ((!)goal).completion_report = report;
            yield db.update_async<Goal>(((!)goal));
            yield record_event(goal_id, "completion-submitted",
                "Completion report submitted for review");
            return raised;
        }

        /**
         * Fails the goal (`saddle failure` in goal context): status
         * 'failed', the report stored, the goal-run task failed (its
         * session closes and saddle frees), every non-terminal task
         * cascades exactly like a cancellation (plan 13 B3: drafts
         * are deleted — no orphan stays dispatchable — queued/working/
         * blocked tasks are cancelled) and the narrative records the
         * cascade. Every user is notified. Idempotent on terminal
         * goals (plan 13 B2): a replayed failure push — redelivered
         * OR a fresh entry id from the honest burst — re-runs no
         * bookkeeping, so exactly one goal-failed event, one cascade,
         * one notification, ever.
         */
        public async void fail_goal(int64 goal_id, string? report) throws Error {
            var goal = yield require_goal(goal_id);
            if (((!)goal).status == GOAL_COMPLETE || ((!)goal).status == GOAL_FAILED
                    || ((!)goal).status == GOAL_CANCELLED) {
                return;
            }
            ((!)goal).status = GOAL_FAILED;
            ((!)goal).completed_at = new DateTime.now_local();
            ((!)goal).completion_report = report ?? ((!)goal).completion_report;
            yield db.update_async<Goal>(((!)goal));

            var run = yield goal_run_task(goal_id);
            if (run != null && ((!)run).status != STATUS_DONE
                    && ((!)run).status != STATUS_FAILED) {
                yield tasks.fail_task(((!)run).id,
                    report != null && report.length > 0 ? report : "The goal agent failed the goal");
            }
            // Plan 13 B3: the failure cascade — the cancel-cascade
            // semantics (the run task just failed above, so the loop's
            // terminal skip covers it).
            int64 cascaded = 0;
            foreach (var task in yield db.query<TaskEntity>()
                    .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (task.status == STATUS_DONE || task.status == STATUS_FAILED) {
                    continue;
                }
                if (task.status == STATUS_DRAFT) {
                    yield tasks.delete_task_quietly(task.id);
                } else {
                    yield tasks.cancel_task(task.id);
                }
                cascaded++;
            }
            yield record_event(goal_id, "goal-failed",
                report != null && report.length > 0 ? report : "The goal agent failed the goal");
            if (cascaded > 0) {
                yield record_event(goal_id, "cascade",
                    @"Goal failure cancelled $(cascaded) outstanding task(s)");
            }
            push.notify_global(PushService.goal_terminal_notification(
                ((!)goal).title, goal_id, false));
            yield pulse.bump(0, "goal failed");
        }

        /**
         * Answers a goal's block (plan 15 C1: THE single answer path
         * every surface calls — the goal page's verdict forms, the
         * task-page deep link on a goal-run block, the dashboard's
         * link-through destination). The block is the goal-run task's
         * OWN row: answering it in place settles everything an
         * observer can read before the first pulse fires, so the
         * completion-review-never-leaves-the-dashboard bug class is
         * structurally impossible — no dual rows exist to fall out of
         * sync. The parked-resume machinery then wakes the agent with
         * the verdict as its resume_notice. Plan and completion
         * blocks apply their goal-state transitions:
         *
         * - plan + {@link NOTICE_PLAN_APPROVED} → the submission
         *   becomes goal.plan, status 'active';
         * - plan + feedback → status 'planning' (the agent revises);
         * - completion + {@link NOTICE_REVIEW_APPROVED} → the goal
         *   completes and the goal-run task closes done;
         * - completion + feedback → status 'active' (the agent keeps
         *   working).
         */
        public async void answer_block(int64 block_id, string answer,
                string? answered_by = null) throws Error {
            var block = yield db.query<TaskBlock>()
                .where(expr("id == $0", elem<int64?>(block_id)))
                .first_async();
            if (block == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 73,
                    @"No goal block exists with id $(block_id)");
            }
            if (((!)block).answer != null) {
                return;
            }
            var task = yield db.query<TaskEntity>()
                .where(expr("id == $0", elem<int64?>(((!)block).task_id)))
                .first_async();
            if (task == null || ((!)task).kind != TASK_KIND_GOAL_RUN
                    || ((!)task).goal_id <= 0) {
                // Not a goal-run block: the plain task answer path.
                yield tasks.answer_block(block_id, answer, answered_by);
                return;
            }
            var goal = yield get_goal(((!)task).goal_id);
            if (goal == null) {
                return;
            }

            // The verdict's events record FIRST, then the row settles
            // (below) before the first observer pulse fires — an
            // event-less resume is a lost context_update, and an
            // unanswered row past the pulse is exactly the stale-pile
            // shape plan 14 B3 defended against with mirror sweeps.
            // With one row there is nothing left to sweep.
            if (((!)block).block_type == GOAL_BLOCK_PLAN) {
                if (answer == NOTICE_PLAN_APPROVED) {
                    if (((!)block).body != null && ((!)block).body.length > 0) {
                        ((!)goal).plan = ((!)block).body;
                    }
                    ((!)goal).status = GOAL_ACTIVE;
                    yield record_event(((!)goal).id, "plan-approved",
                        @"Plan revision #$(((!)block).id) approved — the goal is active");
                } else {
                    ((!)goal).status = GOAL_PLANNING;
                    yield record_event(((!)goal).id, "plan-rejected",
                        @"Plan revision #$(((!)block).id) rejected with feedback — the agent revises");
                }
            } else if (((!)block).block_type == GOAL_BLOCK_COMPLETION) {
                if (answer == NOTICE_REVIEW_APPROVED) {
                    ((!)goal).status = GOAL_COMPLETE;
                    ((!)goal).completed_at = new DateTime.now_local();
                    yield record_event(((!)goal).id, "goal-approved",
                        "Completion report approved — the goal is complete");
                } else {
                    ((!)goal).status = GOAL_ACTIVE;
                    yield record_event(((!)goal).id, "completion-rejected",
                        "Completion report rejected with feedback — the agent continues");
                }
            }
            yield db.update_async<Goal>(((!)goal));

            yield tasks.answer_block(block_id, answer, answered_by);

            if (((!)block).block_type == GOAL_BLOCK_COMPLETION
                    && answer == NOTICE_REVIEW_APPROVED) {
                // Approving the completion closes the goal agent's
                // own task: the run is finished, its saddle frees.
                yield tasks.complete_task(((!)task).id,
                    ((!)goal).completion_report, null, "done");
                push.notify_global(PushService.goal_terminal_notification(
                    ((!)goal).title, ((!)goal).id, true));
            }
            yield pulse.bump(0, "goal block answered");
        }

        /**
         * Raises the goal's ONE block on the goal-run task (plan 15
         * C1): the parked-resume machinery (answered blocks ride the
         * resuming allocation) works untouched, and the goal-run task
         * parks as 'blocked' waiting for the human verdict. The raise
         * is QUIET on the task side — the goal-level notification
         * (goal title, /goals/<id> URL) fires here, once, and must
         * not double-notify with the run task's title.
         *
         * Deduplication (`dedupe_on_type`): a pending row of the same
         * question — or, for the review kinds whose question text is
         * fixed ("Plan submitted for review", "Goal completion
         * report"), of the same TYPE — means the push is a
         * redelivery and the existing row is returned unwritten. A
         * missing or terminal run task raises nothing (null).
         */
        private async TaskBlock? raise_run_block(Goal goal, string question,
                string block_type, string[]? options, string? body,
                bool dedupe_on_type) throws Error {
            var run = yield goal_run_task(goal.id);
            if (run == null || ((!)run).status == STATUS_DONE
                    || ((!)run).status == STATUS_FAILED) {
                warning("[ClankerSaddle] Goal %s block '%s' raised without a live run task",
                    goal.id.to_string(), question);
                return null;
            }
            foreach (var existing in yield tasks.list_blocks(((!)run).id)) {
                bool same = dedupe_on_type
                    ? existing.block_type == block_type : existing.question == question;
                if (same && existing.answer == null) {
                    return existing;
                }
            }
            var raised = yield tasks.add_block(((!)run).id, question, block_type,
                options, body, true);
            yield pulse.bump(0, "goal block raised");
            push.notify_global(PushService.goal_block_notification(
                goal.title, goal.id, block_type, question));
            return raised;
        }

        // ------------------------------------------------------------------
        // Comments and events
        // ------------------------------------------------------------------

        /** Appends a goal comment; `author` is 'agent' or 'user:<name>'. */
        public async GoalComment add_comment(int64 goal_id, string author, string body,
                string? quote = null) throws Error {
            var comment = new GoalComment() {
                goal_id = goal_id,
                author = author,
                body = body,
                quote = quote != null && quote.strip().length > 0 ? quote.strip() : null,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<GoalComment>(comment);
            yield pulse.bump(0, "goal comment");
            return comment;
        }

        /** A goal's comments, oldest first. */
        public async ImmutableLot<GoalComment> list_comments(int64 goal_id) throws Error {
            return yield db.query<GoalComment>()
                .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * A goal's blocks (plan 15 C1): the goal-run task's
         * `task_blocks` rows, oldest first — the run task is the
         * single home of the goal agent's questions, plans and
         * completion reports. Empty while the goal has no run task.
         */
        public async ImmutableLot<TaskBlock> list_blocks(int64 goal_id) throws Error {
            var run = yield goal_run_task(goal_id);
            if (run == null) {
                return new Vector<TaskBlock>().to_immutable_buffer();
            }
            return yield tasks.list_blocks(((!)run).id);
        }

        /** A goal's narrative events, oldest first. */
        public async ImmutableLot<GoalEvent> list_events(int64 goal_id) throws Error {
            return yield db.query<GoalEvent>()
                .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                .order_by(expr("id"))
                .materialise_async();
        }

        /**
         * Records one supervision-narrative event. Called by the
         * lifecycle flows, the goal agent's task operations and the
         * task-terminal hook — every goal_events row is also a
         * potential resume signal (see {@link resume_due}).
         */
        public async GoalEvent record_event(int64 goal_id, string kind, string text,
                int64 task_id = 0, int64 ref_task_id = 0) throws Error {
            var event = new GoalEvent() {
                goal_id = goal_id,
                kind = kind,
                text = text,
                task_id = task_id,
                created_at = new DateTime.now_local()
            };
            yield db.insert_async<GoalEvent>(event);
            return event;
        }

        // ------------------------------------------------------------------
        // Event-driven resume (the parked goal agent)
        // ------------------------------------------------------------------

        /**
         * True when the goal agent should be resumed: its run task is
         * parked (suspended), the goal is active (paused goals freeze
         * supervision; review states park on blocks instead), and at
         * least one task event has waited out the coalescing window
         * (the OLDEST unconsumed event is old enough — bursts collapse
         * into one resume).
         */
        public async bool resume_due(TaskEntity run_task) throws Error {
            if (run_task.kind != TASK_KIND_GOAL_RUN || run_task.status != STATUS_SUSPENDED) {
                return false;
            }
            var goal = yield get_goal(run_task.goal_id);
            if (goal == null || ((!)goal).status != GOAL_ACTIVE) {
                return false;
            }
            DateTime? oldest = null;
            // ONLY owned-task terminal events resume the parked agent —
            // the agent's own task-created/task-enqueued narrative (and
            // human comments) must never wake it; the first terminal
            // event ages the window and bursts coalesce behind it.
            foreach (var event in yield db.query<GoalEvent>()
                    .where(expr("goal_id == $0 && id > $1",
                        elem<int64?>(run_task.goal_id),
                        elem<int64?>(((!)goal).last_seen_event_id)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (event.kind != "task-done" && event.kind != "task-failed") {
                    continue;
                }
                if (event.created_at != null) {
                    oldest = ((!)event).created_at;
                    break;
                }
            }
            if (oldest == null) {
                return false;
            }
            double age = new DateTime.now_local().difference((!)oldest) / 1000000.0;
            return age >= goal_resume_coalesce_seconds();
        }

        /**
         * The batched markdown report riding a goal-agent resume (the
         * assignment's `resume_notice`): the current status of ALL of
         * the goal's tasks plus the full completion reports of the
         * tasks that finished since the agent parked. Consumes the
         * pending events (the watermark advances), so the window
         * starts fresh for the next park.
         */
        public async string take_resume_notice(int64 goal_id) throws Error {
            var goal = yield require_goal(goal_id);
            DateTime? since = null;
            var run = yield goal_run_task(goal_id);
            if (run != null) {
                since = ((!)run).suspended_at;
            }

            var report = new StringBuilder();
            report.append("Goal update — resuming your parked session:\n\n");
            report.append("## Task statuses\n");
            int64 newest_event = ((!)goal).last_seen_event_id;
            var finished = new Vector<TaskEntity>();
            foreach (var task in yield db.query<TaskEntity>()
                    .where(expr("goal_id == $0 && kind == $1",
                        elem<int64?>(goal_id), elem(TASK_KIND_TASK)))
                    .order_by(expr("id"))
                    .materialise_async()) {
                if (task.status == STATUS_DONE) {
                    report.append_printf("- #%s %s — done: %s\n", task.id.to_string(), task.title,
                        task.summary ?? "no summary reported");
                    if (since == null || task.completed_at == null
                            || ((!)task.completed_at).compare((!)since) > 0) {
                        finished.add(task);
                    }
                } else if (task.status == STATUS_FAILED) {
                    report.append_printf("- #%s %s — failed: %s\n", task.id.to_string(), task.title,
                        task.summary ?? "no summary reported");
                } else {
                    report.append_printf("- #%s %s — %s\n", task.id.to_string(), task.title, task.status);
                }
            }
            if (finished.length > 0) {
                report.append("\n## New completion reports\n");
                foreach (var task in finished) {
                    report.append_printf("\n### #%s %s\n%s\n", task.id.to_string(), task.title,
                        task.summary ?? "no summary reported");
                }
            }
            report.append("\nSupervise your tasks: merge completed work into the goal branch,")
                .append(" prune worktrees, create or enqueue follow-ups as needed, then")
                .append(" `saddle suspend` again when you are parked.\n");

            foreach (var event in yield db.query<GoalEvent>()
                    .where(expr("goal_id == $0", elem<int64?>(goal_id)))
                    .order_by_desc(expr("id"))
                    .limit(1)
                    .materialise_async()) {
                newest_event = event.id;
            }
            if (newest_event != ((!)goal).last_seen_event_id) {
                ((!)goal).last_seen_event_id = newest_event;
                yield db.update_async<Goal>((!)goal);
            }
            return report.str;
        }

        // ------------------------------------------------------------------
        // Branch push state (daemon /status exposure)
        // ------------------------------------------------------------------

        /**
         * Stores a daemon-reported goal-branch push state on the goal
         * (raw JSON as the daemon sent it — the web only renders it).
         * Called from the /status observation path when the probed
         * saddle is running a goal agent.
         */
        public async void store_branch_push_state(int64 goal_id, string state_json) throws Error {
            var goal = yield get_goal(goal_id);
            if (goal == null || ((!)goal).branch_push_state == state_json) {
                return;
            }
            ((!)goal).branch_push_state = state_json;
            yield db.update_async<Goal>((!)goal);
        }

        // ------------------------------------------------------------------
        // Helpers
        // ------------------------------------------------------------------

        private async Goal require_goal(int64 goal_id) throws Error {
            var goal = yield get_goal(goal_id);
            if (goal == null) {
                throw new Error(Quark.from_string("clanker-saddle"), 74,
                    @"No goal exists with id $(goal_id)");
            }
            return (!)goal;
        }
    }
}