Finish the advisor loop: approvals, completion callback, docs

Goal 27 task 3. DecideAdvisorSuggestionAction replaces its stub: owner- and
pending-checked decisions sealed on the suggestion id. approve creates a draft
task (creator human, so the creator-only enqueue rule holds by construction;
mode validated against TASK_MODES, project_id 0 allowed as standalone) storing
task_id; approve_enqueue additionally calls enqueue_task as the human; goal
approve creates a draft goal via GoalService.create_goal storing goal_id and is
NEVER enqueued or started (approve_enqueue on a goal behaves as approve);
decline dismisses. Every path appends a short system message to the chat and
kicks an advisor turn so the model acknowledges the outcome; the action answers
with the refreshed chat slot.

AdvisorCallback is a new 15s periodic pass (ADVISOR_CALLBACK_TICK_S, the
Dispatcher pattern) scanning approved suggestions with a task and no
delivery: on the task reaching done/failed it appends a system message (the
summary or the description's first line), stamps delivered_at FIRST (the
never-twice guard against the concurrent kick), bumps StatePulse and kicks an
advisor turn so the model reacts to the result. Ghost tasks and vanished chats
are stamped and skipped, never re-scanned.

README gains an 'AI Chat (advisor)' section: the composer, per-user chats,
questions, suggestion approval semantics, the completion callback, the
ADVISOR_LLM_* env vars with SADDLE_LLM_* fallback, the not-configured notice
and the privacy model.

test/bundles/advisor_arc.py (registered in BUNDLES_ORDER) drives the full arc:
dashboard composer -> question -> answer -> suggest_task -> approve & enqueue
(task queued, creator human, suggestion approved with task_id) -> the fake
saddle runs the task to done -> the callback lands its system message,
delivered_at is stamped and the advisor reacts; plus never-twice, plain-approve
draft, decline, draft-only goals, decided-row refusal and per-user privacy.

README.md

@@ -403,6 +403,61 @@ Dispatches also carry the primary and co-developed projects' notes (up to
 30 entries / 20 000 characters, oldest first) so agents start with the
 accumulated project knowledge.
 
+## AI Chat (advisor)
+
+**AI Chat** is a conversational advisor inside the web app: an LLM loop
+that can READ the fleet's state — projects, tasks, goals, rules,
+discoveries and capability tags — through a read-only tool set, ask you
+questions mid-chat, and SUGGEST tasks or goals for your approval. It
+can never run commands, edit files or dispatch work itself; nothing is
+created until you approve a suggestion in the chat UI.
+
+The dashboard's top carries the **Ask the advisor** composer: type your
+question and you land on a new chat session (`/chats/{id}`). The
+**/chats** page lists your conversations (with per-row delete); chats
+are **private per user** — nobody else can see, continue or delete
+yours. Note that what the advisor READS is the whole fleet's work state
+(projects, tasks and goals are not per-user in this app), the same as
+every other page.
+
+**Questions.** When the advisor needs your decision it asks a
+multiple-choice question (with an "Other" free-text box, like the
+saddle's own question tool): the chat parks until you answer, and the
+answer resumes the turn.
+
+**Suggestions and approvals.** The advisor proposes work as a card in
+the chat — a task (title, mandate, mode, project, priority) or a goal
+(title, brief, project, branch). The card is inert until you decide:
+
+- **Approve** (task) → a draft task is created, creator *human* — you
+  enqueue it yourself whenever you like.
+- **Approve & enqueue** (task) → the draft is created and immediately
+  queued for a saddle.
+- **Approve** (goal) → a draft goal is created, **unqueued** — the
+  advisor never starts a goal; you start it from the goal page.
+- **Decline** → the card is dismissed, nothing is created.
+
+**Completion callback.** A periodic pass (every 15 s,
+`ADVISOR_CALLBACK_TICK_S`) watches every approved task suggestion for
+its task's terminal state: when the task completes or fails, a system
+message lands in the originating chat (the task's summary included),
+the page refreshes through the state pulse, and the advisor is woken
+to react — report the result back, suggest follow-ups — so the
+conversation continues from real outcomes.
+
+**Configuration.** The advisor uses its own OpenAI-compatible
+chat-completions endpoint, falling back to the daemon-facing one:
+
+```bash
+export ADVISOR_LLM_BASE_URL=http://127.0.0.1:9999
+export ADVISOR_LLM_API_KEY=…        # optional
+export ADVISOR_LLM_MODEL=mock       # optional
+```
+
+When neither `ADVISOR_LLM_BASE_URL` nor `SADDLE_LLM_BASE_URL` resolves,
+the composer disables itself and the chat pages show a "not configured"
+notice naming the variable to set — nothing else changes.
+
 ## Integration testing
 
 `test/integration_test.py` drives BOTH real binaries — this app and a

meson.build

@@ -226,6 +226,7 @@ app_sources = files(
     'src/Services/PushService.vala',
     'src/Services/ChatService.vala',
     'src/Services/AdvisorService.vala',
+    'src/Services/AdvisorCallback.vala',
     'src/Api/SaddleApi.vala',
     'src/Api/AdvisorApi.vala',
     'src/Api/EventLogApi.vala',

src/Services/AdvisorCallback.vala

@@ -0,0 +1,200 @@
+using Invercargill;
+using Invercargill.DataStructures;
+using Invercargill.Expressions;
+using InvercargillSql.Orm;
+using Inversion;
+using ClankerSaddle.Data;
+
+namespace ClankerSaddle.Services {
+
+    /**
+     * The advisor completion callback (goal 27): a periodic pass that
+     * watches every APPROVED task suggestion for its task's terminal
+     * state and, on the first sighting, delivers a system message into
+     * the originating chat ("Task #12 completed: …" or the failed
+     * variant), stamps `delivered_at` (the never-twice guard) and
+     * bumps StatePulse so the chat page refreshes. The delivery also
+     * KICKS an advisor turn, so the model reads the result from its
+     * transcript (system rows ride the wire as bracketed user
+     * notices) and can report back or suggest follow-ups — the
+     * "task completion calls back to the advisor" arc of the goal
+     * brief.
+     *
+     * The pass itself is deliberately dumb: it never starts, stops or
+     * edits work; it only observes terminal statuses other actors
+     * (saddles, the human) already produced. Goal suggestions carry
+     * no callback — goals are created unqueued and the human drives
+     * them, so there is no completion to report into the chat.
+     *
+     * Each pass resolves fresh request-scoped services (no database
+     * scope is held between ticks) and never overlaps with itself,
+     * like the {@link Dispatcher}.
+     */
+    public class AdvisorCallback : Object {
+
+        private Container container = inject<Container>();
+        private bool scanning = false;
+
+        /** ADVISOR_CALLBACK_TICK_S (default 15s): the pass interval. */
+        public static uint tick_seconds() {
+            var raw = Environment.get_variable("ADVISOR_CALLBACK_TICK_S") ?? "";
+            int64 parsed = int64.parse(raw);
+            return parsed > 0 ? (uint) parsed : 15;
+        }
+
+        /** Starts the repeating timer. */
+        public void start() {
+            uint tick = tick_seconds();
+            Timeout.add_seconds(tick, () => {
+                scan_once.begin();
+                return true;
+            });
+            print("[ClankerSaddle] Advisor callback running (%us interval)\n",
+                tick);
+        }
+
+        /** One pass (the timer body; also the test seam). */
+        public async void scan_once() {
+            if (scanning) {
+                return;
+            }
+            scanning = true;
+            try {
+                yield scan_approved();
+            } finally {
+                scanning = false;
+            }
+        }
+
+        /**
+         * Every approved suggestion with a task and no delivery yet.
+         * The null check happens in Vala, not the DSL — nullable
+         * column comparisons in `expr` are fragile, and the row count
+         * is tiny either way.
+         */
+        private async void scan_approved() {
+            // chat_id -> the kicked flag: one turn per chat per pass,
+            // however many suggestions delivered.
+            var kicked = new Invercargill.DataStructures.HashSet<string>();
+            var scope = container.create_scope();
+            ImmutableLot<AdvisorSuggestion> rows;
+            try {
+                rows = yield scope.resolve<OrmSession>()
+                    .query<AdvisorSuggestion>()
+                    .where(expr("status == $0 && task_id > 0",
+                        elem<string?>("approved")))
+                    .order_by(expr("id"))
+                    .materialise_async();
+            } catch (Error e) {
+                scope.release();
+                warning("[ClankerSaddle] Advisor callback scan failed: %s",
+                    e.message);
+                return;
+            }
+            scope.release();
+
+            foreach (var row in rows) {
+                if (row.delivered_at != null) {
+                    // Already delivered — the never-twice guard.
+                    continue;
+                }
+                if (kicked.contains(row.chat_id.to_string())) {
+                    continue;
+                }
+                try {
+                    if (yield deliver(row)) {
+                        kicked.add(row.chat_id.to_string());
+                    }
+                } catch (Error e) {
+                    warning("[ClankerSaddle] Advisor callback deliver failed"
+                        + " for suggestion %lld: %s", row.id, e.message);
+                }
+            }
+        }
+
+        /**
+         * Delivers `row`'s terminal notice when its task has finished:
+         * appends the system message, stamps `delivered_at`, bumps the
+         * pulse and kicks the advisor turn. Returns false (changing
+         * nothing) while the task is still live or the chat is gone.
+         */
+        private async bool deliver(AdvisorSuggestion row) throws Error {
+            var scope = container.create_scope();
+            try {
+                var db = scope.resolve<OrmSession>();
+                var chats = scope.resolve<ChatService>();
+                var advisor = scope.resolve<AdvisorService>();
+                var pulse = scope.resolve<StatePulse>();
+
+                var task = yield db.query<TaskEntity>()
+                    .where(expr("id == $0", elem<int64?>(row.task_id)))
+                    .first_async();
+                if (task == null) {
+                    // The task vanished (a test reset, a manual wipe):
+                    // stamp the delivery so the pass is not stuck on a
+                    // ghost forever.
+                    row.delivered_at = new DateTime.now_local();
+                    yield db.update_async<AdvisorSuggestion>(row);
+                    return false;
+                }
+                bool done = ((!)task).status == STATUS_DONE;
+                bool failed = ((!)task).status == STATUS_FAILED;
+                if (!done && !failed) {
+                    return false;
+                }
+                // The chat must still exist and belong to the
+                // suggestion's user (delete_chat cascades the row, so
+                // a missing chat here is a race at worst).
+                var chat = yield chats.get_chat(row.chat_id, row.user_id);
+                if (chat == null) {
+                    row.delivered_at = new DateTime.now_local();
+                    yield db.update_async<AdvisorSuggestion>(row);
+                    return false;
+                }
+
+                string label = done ? "completed" : "failed";
+                string summary = summary_line(((!)task).summary,
+                    ((!)task).description);
+                string notice = summary.length > 0
+                    ? @"Task #$(row.task_id) $(label): $(summary)"
+                    : @"Task #$(row.task_id) $(label).";
+
+                // Stamp FIRST: the turn below runs concurrently with
+                // the next pass, and delivered_at is the guard.
+                row.delivered_at = new DateTime.now_local();
+                yield db.update_async<AdvisorSuggestion>(row);
+                yield chats.record_system_message(row.chat_id, notice);
+                yield pulse.bump(row.task_id,
+                    @"advisor callback task #$(row.task_id) $(label)");
+                // The advisor reacts to the result (the system row
+                // rides its wire as a bracketed notice).
+                advisor.run_turn.begin(row.chat_id, row.user_id);
+                return true;
+            } finally {
+                scope.release();
+            }
+        }
+
+        /**
+         * The message's summary half: the task's own summary when it
+         * has one, else the first meaningful line of its description,
+         * clamped to one line (~160 chars).
+         */
+        private static string summary_line(string? summary, string? description) {
+            string source = summary != null && ((!)summary).strip().length > 0
+                ? ((!)summary).strip()
+                : (description ?? "").strip();
+            if (source.length == 0) {
+                return "";
+            }
+            foreach (string line in source.split("\n")) {
+                string trimmed = line.strip();
+                if (trimmed.length > 0) {
+                    return trimmed.length > 160
+                        ? trimmed.substring(0, 160) + "…" : trimmed;
+                }
+            }
+            return "";
+        }
+    }
+}

src/actions/AdvisorActions.vala

@@ -1,7 +1,10 @@
 using Inversion;
+using Invercargill;
+using Invercargill.Expressions;
 using Spry;
 using Spry.Authentication;
 using Statum;
+using ClankerSaddle.Data;
 using ClankerSaddle.Services;
 
 namespace ClankerSaddle {
@@ -321,15 +324,24 @@ namespace ClankerSaddle {
 
     /**
      * Decides a pending suggestion (goal 27): 'approve' creates the
-     * draft, 'approve_enqueue' creates the draft AND queues it,
-     * 'decline' dismisses it. THE HANDLER IS TASK 3'S (goal 27 task
-     * 176): until it lands this base raises a plain "not yet
-     * available" error so the chat UI can ship wired against the
-     * final action name. The suggestion and chat ids are sealed; the
-     * decision rides the `decision` field.
+     * real row — a DRAFT task (creator human, so the creator-only
+     * enqueue rule holds by construction) or a DRAFT goal (never
+     * started: goals are unqueued until the human starts them) —
+     * 'approve_enqueue' additionally queues the created task for a
+     * saddle, and 'decline' dismisses the row. Every path appends a
+     * short system message to the chat and kicks a background advisor
+     * turn so the advisor can acknowledge the outcome; the action
+     * answers with the refreshed `chat` slot. The suggestion and chat
+     * ids are sealed; only the chat's owner may decide, and only
+     * while the row is still pending.
      */
     public class DecideAdvisorSuggestionAction : TypedStatumAction<AdvisorSuggestionActionPrivate> {
 
+        private ChatService chat_service = inject<ChatService>();
+        private AdvisorService advisor_service = inject<AdvisorService>();
+        private TaskService task_service = inject<TaskService>();
+        private GoalService goal_service = inject<GoalService>();
+        private InvercargillSql.Orm.OrmSession db = inject<InvercargillSql.Orm.OrmSession>();
         private UserService users = inject<UserService>();
 
         protected override Type private_type {
@@ -341,11 +353,198 @@ namespace ClankerSaddle {
             if (guard != null) {
                 return (!)guard;
             }
-            // Goal 27 task 3 (approvals + completion callback) owns
-            // the body: approve -> draft task/goal, approve_enqueue ->
-            // draft + enqueue, decline -> dismissed row.
-            return directives().notify("error",
-                "Suggestion decisions are not available yet").navigate("/chats");
+
+            var ctx = request_private;
+            var me = SpryAuth.current(request);
+            if (ctx == null || me == null) {
+                return directives().navigate("/chats");
+            }
+            int64 viewer_id = ((!)me).user_id;
+            int64 suggestion_id = ((!)ctx).suggestion_id;
+            int64 chat_id = ((!)ctx).chat_id;
+
+            string decision = Forms.field(request, "decision");
+            if (decision != "approve" && decision != "approve_enqueue"
+                    && decision != "decline") {
+                return directives().notify("error",
+                    "Unknown suggestion decision").navigate("/chats");
+            }
+
+            // Owner-checked and status-checked: the row must belong to
+            // this user's chat and still be pending (double clicks and
+            // stale cards land here, not in two tasks).
+            var suggestion = yield db.query<AdvisorSuggestion>()
+                .where(expr("id == $0 && chat_id == $1 && user_id == $2",
+                    elem<int64?>(suggestion_id), elem<int64?>(chat_id),
+                    elem<int64?>(viewer_id)))
+                .first_async();
+            if (suggestion == null) {
+                return directives().notify("error",
+                    "That suggestion does not exist").navigate("/chats");
+            }
+            if (((!)suggestion).status != "pending") {
+                return notify_chat(chat_id,
+                    "That suggestion has already been decided");
+            }
+            var chat = yield chat_service.get_chat(chat_id, viewer_id);
+            if (chat == null) {
+                return directives().notify("error",
+                    "That chat no longer exists").navigate("/chats");
+            }
+
+            string notice;
+            try {
+                if (decision == "decline") {
+                    ((!)suggestion).status = "declined";
+                    ((!)suggestion).decided_at = new DateTime.now_local();
+                    yield db.update_async<AdvisorSuggestion>((!)suggestion);
+                    notice = "Declined.";
+                } else if (((!)suggestion).kind == "task") {
+                    var payload = parse_payload(((!)suggestion).payload);
+                    string title = member_string(payload, "title");
+                    string mandate = member_string(payload, "mandate");
+                    string mode = member_string(payload, "mode");
+                    int64 project_id = member_int(payload, "project_id");
+                    int64 priority = member_int(payload, "priority");
+                    if (title.length == 0 || mandate.length == 0) {
+                        return notify_chat(chat_id,
+                            "That suggestion's payload is incomplete");
+                    }
+                    if (mode.length == 0) {
+                        mode = "implement";
+                    }
+                    if (!is_valid_mode(mode)) {
+                        return notify_chat(chat_id,
+                            @"Suggested mode '$(mode)' is not a valid work mode");
+                    }
+                    if (project_id < 0) {
+                        return notify_chat(chat_id,
+                            @"Invalid project id $(project_id)");
+                    }
+                    if (project_id > 0 && (yield project_exists(project_id)) == null) {
+                        return notify_chat(chat_id,
+                            @"No project exists with id $(project_id)");
+                    }
+                    // Creator is HUMAN — the creator-only enqueue rule
+                    // holds by construction and the approve_enqueue
+                    // path below enqueues as the human too.
+                    var task = yield task_service.create_task(
+                        project_id, title, mandate, priority,
+                        null, null, STATUS_DRAFT, TASK_KIND_TASK, mode,
+                        CREATOR_HUMAN, 0, 0);
+                    ((!)suggestion).task_id = task.id;
+                    ((!)suggestion).status = "approved";
+                    ((!)suggestion).decided_at = new DateTime.now_local();
+                    yield db.update_async<AdvisorSuggestion>((!)suggestion);
+                    if (decision == "approve_enqueue") {
+                        yield task_service.enqueue_task(task.id, CREATOR_HUMAN);
+                        notice = @"Approved & enqueued — task #$(task.id) queued";
+                    } else {
+                        notice = @"Approved — task #$(task.id) created (draft)";
+                    }
+                } else {
+                    var payload = parse_payload(((!)suggestion).payload);
+                    string title = member_string(payload, "title");
+                    string brief = member_string(payload, "brief");
+                    string branch_name = member_string(payload, "branch_name");
+                    int64 project_id = member_int(payload, "project_id");
+                    if (title.length == 0 || brief.length == 0) {
+                        return notify_chat(chat_id,
+                            "That suggestion's payload is incomplete");
+                    }
+                    if (branch_name.length == 0) {
+                        branch_name = @"goal/advisor-$(suggestion_id)";
+                    }
+                    var project = yield project_exists(project_id);
+                    if (project_id <= 0 || project == null) {
+                        return notify_chat(chat_id,
+                            @"The goal needs an existing project (got $(project_id))");
+                    }
+                    // Draft only: the advisor NEVER starts a goal —
+                    // the human starts it from the goal page.
+                    var goal = yield goal_service.create_goal(
+                        project_id, title, brief, branch_name, null, 3,
+                        CREATOR_HUMAN);
+                    ((!)suggestion).goal_id = goal.id;
+                    ((!)suggestion).status = "approved";
+                    ((!)suggestion).decided_at = new DateTime.now_local();
+                    yield db.update_async<AdvisorSuggestion>((!)suggestion);
+                    notice = @"Approved — goal #$(goal.id) created in draft (not started)";
+                }
+            } catch (Error e) {
+                return notify_chat(chat_id,
+                    @"Could not apply the decision: $(e.message)");
+            }
+
+            // The outcome lands in the chat and wakes the advisor so
+            // it can acknowledge (and, on completion, follow up).
+            try {
+                yield chat_service.record_system_message(chat_id, notice);
+            } catch (Error e) {
+                warning("[ClankerSaddle] Advisor decision notice failed: %s",
+                    e.message);
+            }
+            advisor_service.run_turn.begin(chat_id, viewer_id);
+
+            var updated = yield chat_service.get_chat(chat_id, viewer_id);
+            if (updated == null) {
+                return directives().navigate("/chats");
+            }
+            var state = yield ChatState.build_from(state_service, chat_service,
+                advisor_service, action_registry, viewer_id, updated);
+            return directives().set(state);
+        }
+
+        /** An error notify that stays on the chat page. */
+        private DirectiveBuilder notify_chat(int64 chat_id, string message) {
+            return directives().notify("error", message)
+                .navigate(@"/chats/$(chat_id)");
+        }
+
+        /** The project row for `id`, or null (0 = standalone tasks). */
+        private async ClankerSaddle.Data.Project? project_exists(int64 id)
+                throws GLib.Error {
+            if (id <= 0) {
+                return null;
+            }
+            return yield db.query<ClankerSaddle.Data.Project>()
+                .where(expr("id == $0", elem<int64?>(id)))
+                .first_async();
+        }
+
+        private static Json.Object parse_payload(string payload) {
+            var parser = new Json.Parser();
+            try {
+                parser.load_from_data(payload);
+            } catch (Error e) {
+                return new Json.Object();
+            }
+            var root = parser.get_root();
+            if (root == null || root.get_node_type() != Json.NodeType.OBJECT) {
+                return new Json.Object();
+            }
+            return ((!)root).get_object();
+        }
+
+        private static string member_string(Json.Object body, string member) {
+            if (!body.has_member(member)) {
+                return "";
+            }
+            var node = body.get_member(member);
+            if (node.get_node_type() == Json.NodeType.VALUE
+                    && node.get_string() != null) {
+                return (!)node.get_string();
+            }
+            return "";
+        }
+
+        private static int64 member_int(Json.Object body, string member) {
+            if (!body.has_member(member)) {
+                return 0;
+            }
+            var node = body.get_member(member);
+            return node.get_node_type() == Json.NodeType.VALUE
+                ? node.get_int() : 0;
         }
     }
 }

src/main.vala

@@ -97,6 +97,7 @@ int main(string[] args) {
         application.add_singleton<PushService>();
         application.add_singleton<Dispatcher>();
         application.add_singleton<AdvisorService>();
+        application.add_singleton<AdvisorCallback>();
 
         // Web Push (plan 11 task E): the VAPID keypair loads or
         // generates + persists BEFORE the server accepts requests, so
@@ -280,6 +281,7 @@ int main(string[] args) {
         GLib.Idle.add(() => {
             seed_admin.begin(application.container);
             start_dispatcher.begin(application.container);
+            start_advisor_callback.begin(application.container);
             return false;
         });
 
@@ -323,3 +325,20 @@ private async void start_dispatcher(Container container) {
         printerr("[ClankerSaddle] Dispatcher startup failed: %s\n", e.message);
     }
 }
+
+/**
+ * Starts the advisor completion callback's periodic pass alongside
+ * the dispatcher (goal 27): every ADVISOR_CALLBACK_TICK_S (15s) it
+ * delivers terminal task outcomes into the chats that suggested
+ * them.
+ */
+private async void start_advisor_callback(Container container) {
+    try {
+        var scope = container.create_scope();
+        scope.resolve<AdvisorCallback>().start();
+        scope.release();
+    } catch (Error e) {
+        printerr("[ClankerSaddle] Advisor callback startup failed: %s\n",
+            e.message);
+    }
+}

test/bundles/advisor_arc.py

@@ -0,0 +1,393 @@
+"""The advisor approvals + completion callback bundle (goal 27, task 3):
+the full arc from the dashboard composer to a task's completion landing
+back in the chat, against the scripted LLM endpoint. The scenes:
+
+- the arc: dashboard hero -> chat -> the advisor asks a question -> the
+  option answer lands -> the advisor suggests a task -> approve &
+  enqueue creates the task QUEUED with creator human, stamps the
+  suggestion approved with its task_id, and appends the system notice
+  (the advisor turn kicked by the decision acknowledges it)
+- the completion callback: the task flipped terminal (via the app's own
+  surface - the fake saddle runs it to done) delivers ONE system
+  message into the chat, stamps delivered_at and wakes the advisor
+- never twice: a second pass (and the advisor's own reaction turn)
+  adds no second delivery
+- plain approve: the task lands DRAFT, creator human, nothing queued
+- decline: the row is dismissed, nothing is created, the chat is told
+- goal approve: a DRAFT goal exists, unqueued, owner of its branch;
+  the goal suggestion never enqueues anything
+- decided rows cannot be decided again (the card's slot loses the
+  decide action; a replayed decision is refused)
+- privacy: another user's decision on someone else's suggestion is
+  refused and changes nothing
+"""
+from harness import *  # noqa: F401,F403 — the shared wire helpers
+import harness  # the module itself (restart_with_llm re-points its globals)
+from bundles.advisor import ScriptedLlm
+
+
+def db_rows(sql, params=()):
+    conn = db()
+    try:
+        return conn.execute(sql, params).fetchall()
+    finally:
+        conn.close()
+
+
+def wait_status(chat_id, status, timeout=30):
+    def read():
+        rows = db_rows("select status from advisor_chats where id = ?",
+                       (chat_id,))
+        return len(rows) == 1 and rows[0][0] == status and status or False
+    return wait_for(read, timeout=timeout,
+                    what="chat %d -> %s" % (chat_id, status))
+
+
+def restart_with_llm(ctx, llm):
+    """Points the app at `llm` and restarts it (env re-read)."""
+    os.environ["ADVISOR_LLM_BASE_URL"] = "http://127.0.0.1:%d" % llm.port
+    os.environ["ADVISOR_LLM_MODEL"] = "scripted"
+    ctx.app.terminate()
+    try:
+        ctx.app.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        ctx.app.kill()
+    ctx.app = start_app()
+    ctx.auth = login("admin", "adminpass123")
+    return ctx.auth
+
+
+def run(ctx):
+    auth = ctx.auth
+
+    # The scripted loop: one ask_question, one suggest_task, then plain
+    # replies for every later turn (the decision and callback kicks).
+    llm = ScriptedLlm([
+        {"tools": [{"id": "arc-1", "name": "ask_question",
+                    "arguments": json.dumps({
+                        "question": "Which project should the task target?",
+                        "options": ["demo", "skip it"]})}]},
+        {"tools": [{"id": "arc-2", "name": "suggest_task",
+                    "arguments": json.dumps({
+                        "title": "Advisor arc task",
+                        "mandate": "Prove the advisor approval arc.",
+                        "mode": "test", "project_id": 1, "priority": 4})}]},
+        {"content": "Task suggested - approve it in the card when ready."},
+        {"content": "Noted - the task is queued."},
+        {"content": "Good - the task finished."},
+    ])
+    llm.serve()
+    auth = restart_with_llm(ctx, llm)
+
+    print("== advisor_arc: the dashboard composer opens the chat ==")
+    home = slot_public(auth, "/", "dashboard")
+    result = act(auth, home["start_advisor"], {
+        "message": "I want a small verification task on the demo project"})
+    navigation = [d for d in result if d.get("type") == "navigate"][0]
+    chat_id = int(navigation["uri"].rsplit("/", 1)[1])
+    check(wait_status(chat_id, "awaiting_answer"),
+          "the advisor asks its question and parks the chat")
+    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
+    questions = [m for m in page["messages"] if m.get("is_question")]
+    check(len(questions) == 1 and questions[0]["options"] == ["demo", "skip it"],
+          "the question card renders with its options")
+
+    print("== advisor_arc: the answer resumes into a suggestion ==")
+    result = act(auth, page["send"], {"answer": "demo"})
+    check(wait_status(chat_id, "idle"), "the answered turn completes")
+    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat_id,
+                                       "chat")["suggestions"] or False,
+                   timeout=15, what="suggestion card") is not None,
+          "the suggested task reaches the card")
+    page = slot_public(auth, "/chats/%d" % chat_id, "chat")
+    card = page["suggestions"][0]
+    check(card["kind"] == "task" and card["status"] == "pending"
+          and card["title"] == "Advisor arc task"
+          and "decide" in card,
+          "the pending task card carries its payload and decide action")
+    suggestion_id = int(card["id"])
+    check(db_rows("select count(*) from tasks"
+                  " where title = 'Advisor arc task'")[0][0] == 0,
+          "nothing exists before the approval")
+
+    print("== advisor_arc: approve & enqueue ==")
+    result = act(auth, card["decide"], {"decision": "approve_enqueue"})
+    sets = [d for d in result if d.get("type") == "set"]
+    check(len(sets) == 1, "the decision answers with the rebuilt slot")
+    task_rows = db_rows("select id, status, creator, mode, priority,"
+                        " project_id, description from tasks"
+                        " where title = 'Advisor arc task'")
+    check(len(task_rows) == 1, "the approval creates exactly one task")
+    task_id, task_status, creator, mode, priority, project_id, description \
+        = task_rows[0]
+    check(task_status == "queued",
+          "approve_enqueue leaves the task QUEUED (%s)" % task_status)
+    check(creator == "human" and mode == "test"
+          and str(priority) == "4" and str(project_id) == "1"
+          and description == "Prove the advisor approval arc.",
+          "the task keeps the suggestion's fields (creator human)")
+    suggestion = db_rows("select status, task_id, decided_at from"
+                         " advisor_suggestions where id = ?",
+                         (suggestion_id,))[0]
+    check(suggestion[0] == "approved" and str(suggestion[1]) == str(task_id)
+          and suggestion[2] is not None,
+          "the suggestion row is approved with its task_id and stamp")
+    check(wait_for(lambda: [r for r in db_rows(
+        "select kind, content from advisor_messages where chat_id = ?"
+        " order by id", (chat_id,))
+        if r[0] == "system"
+        and r[1] == "Approved & enqueued — task #%d queued" % task_id]
+        or False, timeout=15, what="approval notice") is not None,
+        "the approval lands its system notice in the chat")
+    check(wait_for(lambda: llm.requests
+                   and any(m.get("role") == "user" and "Approved & enqueued"
+                           in (m.get("content") or "")
+                           for m in llm.requests[-1]["messages"]) or False,
+                   timeout=15, what="decision kick reaches the LLM")
+          is not None,
+        "the decision kick runs an advisor turn that sees the notice")
+    check(wait_status(chat_id, "idle"), "the kicked turn settles idle")
+
+    print("== advisor_arc: the completion callback ==")
+    # The fake saddle runs the queued task to done (the app's own
+    # dispatch surface - no direct status surgery).
+    check(wait_for(lambda: task_state(auth, task_id).get("status")
+                   == "done" or False, timeout=90,
+                   what="task %d -> done" % task_id) is not None,
+        "the queued task runs to done on the fake saddle")
+    check(wait_for(lambda: [r for r in db_rows(
+        "select kind, content from advisor_messages where chat_id = ?"
+        " order by id", (chat_id,))
+        if r[0] == "system" and r[1].startswith(
+            "Task #%d completed" % task_id)] or False,
+        timeout=45, what="completion notice") is not None,
+        "the completion callback lands its system message in the chat")
+    check(db_rows("select delivered_at from advisor_suggestions"
+                  " where id = ?", (suggestion_id,))[0][0] is not None,
+        "the suggestion row is stamped delivered_at")
+    # The callback kicks an advisor turn: the model reads the bracketed
+    # system notice and answers (the "Good - the task finished." step).
+    check(wait_for(lambda: any(
+        r[0] == "assistant" and r[1] == "Good - the task finished."
+        for r in db_rows("select kind, content from advisor_messages"
+                         " where chat_id = ?", (chat_id,))) or False,
+        timeout=30, what="advisor reaction") is not None,
+        "the advisor reacts to the completion in the chat")
+    check(wait_status(chat_id, "idle"), "the reaction turn settles idle")
+
+    print("== advisor_arc: the callback never delivers twice ==")
+    delivered = db_rows(
+        "select count(*) from advisor_messages where chat_id = ?"
+        " and kind = 'system' and content like 'Task #%d completed%%'"
+        % task_id, (chat_id,))[0][0]
+    check(delivered == 1,
+          "exactly one completion message (%d seen)" % delivered)
+    # A quiet window across more than one callback tick: no second
+    # delivery, no extra turn.
+    check(quiet_window(20,
+                       lambda: db_rows(
+                           "select count(*) from advisor_messages"
+                           " where chat_id = ? and kind = 'system'"
+                           " and content like 'Task #%d completed%%'"
+                           % task_id, (chat_id,))[0][0] == 1,
+                       "no second completion delivery"),
+          "the pass stays quiet about an already-delivered outcome")
+
+    print("== advisor_arc: plain approve leaves a draft ==")
+    llm2 = ScriptedLlm([
+        {"tools": [{"id": "dr-1", "name": "suggest_task",
+                    "arguments": json.dumps({
+                        "title": "Advisor draft task",
+                        "mandate": "Stay in draft.",
+                        "mode": "investigate",
+                        "project_id": 0, "priority": 6})}]},
+        {"content": "Suggested."},
+        {"content": "Thanks."},
+    ])
+    llm2.serve()
+    auth = restart_with_llm(ctx, llm2)
+    home = slot_public(auth, "/", "dashboard")
+    result = act(auth, home["start_advisor"],
+                 {"message": "suggest one more task"})
+    chat2 = int([d for d in result if d.get("type") == "navigate"][0]
+                ["uri"].rsplit("/", 1)[1])
+    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat2,
+                                       "chat")["suggestions"] or False,
+                   timeout=15, what="draft suggestion") is not None,
+          "the second suggestion lands")
+    card = slot_public(auth, "/chats/%d" % chat2, "chat")["suggestions"][0]
+    result = act(auth, card["decide"], {"decision": "approve"})
+    draft = db_rows("select status, creator, project_id from tasks"
+                    " where title = 'Advisor draft task'")[0]
+    check(draft[0] == "draft" and draft[1] == "human"
+          and str(draft[2]) == "0",
+          "plain approve creates a standalone DRAFT task (creator human)")
+    check(wait_for(lambda: [r for r in db_rows(
+        "select content from advisor_messages where chat_id = ?",
+        (chat2,)) if r[0].startswith("Approved — task #")] or False,
+        timeout=15, what="draft approval notice") is not None,
+        "the draft approval lands its notice")
+    # The draft task never dispatches (nothing queues it): a quiet
+    # window across several dispatch ticks proves it stays draft.
+    draft_id = db_rows("select id from tasks"
+                       " where title = 'Advisor draft task'")[0][0]
+    check(quiet_window(6,
+                       lambda: db_rows(
+                           "select status from tasks where id = ?",
+                           (draft_id,))[0][0] == "draft",
+                       "draft stays draft"),
+          "the approved-but-not-enqueued task stays draft")
+
+    print("== advisor_arc: decline creates nothing ==")
+    llm3 = ScriptedLlm([
+        {"tools": [{"id": "dc-1", "name": "suggest_task",
+                    "arguments": json.dumps({
+                        "title": "Declined task", "mandate": "No.",
+                        "mode": "implement"})}]},
+        {"content": "Suggested."},
+        {"content": "Understood."},
+    ])
+    llm3.serve()
+    auth = restart_with_llm(ctx, llm3)
+    home = slot_public(auth, "/", "dashboard")
+    result = act(auth, home["start_advisor"], {"message": "suggest to decline"})
+    chat3 = int([d for d in result if d.get("type") == "navigate"][0]
+                ["uri"].rsplit("/", 1)[1])
+    check(wait_for(lambda: slot_public(auth, "/chats/%d" % chat3,
+                                       "chat")["suggestions"] or False,
+                   timeout=15, what="decline suggestion") is not None,
+          "the third suggestion lands")
+    card = slot_public(auth, "/chats/%d" % chat3, "chat")["suggestions"][0]
+    act(auth, card["decide"], {"decision": "decline"})
+    check(db_rows("select status from advisor_suggestions"
+                  " where chat_id = ?", (chat3,))[0][0] == "declined",
+          "the declined row is marked declined")
+    check(db_rows("select count(*) from tasks"
+                  " where title = 'Declined task'")[0][0] == 0,
+          "the declined suggestion creates no task")
+    check(wait_for(lambda: [r for r in db_rows(
+        "select kind, content from advisor_messages where chat_id = ?",
+        (chat3,)) if r[0] == "system" and r[1] == "Declined."] or False,
+        timeout=15, what="decline notice") is not None,
+        "the decline lands its notice")
+
+    print("== advisor_arc: goal approve is draft-only ==")
+    llm4 = ScriptedLlm([
+        {"tools": [{"id": "ga-1", "name": "suggest_goal",
+                    "arguments": json.dumps({
+                        "title": "Advisor arc goal",
+                        "brief": "Prove goal suggestions.",
+                        "project_id": 1,
+                        "branch_name": "goal/advisor-arc"})}]},
+        {"content": "Suggested."},
+        {"content": "Thanks."},
+    ])
+    llm4.serve()
+    auth = restart_with_llm(ctx, llm4)
+    home = slot_public(auth, "/", "dashboard")
+    result = act(auth, home["start_advisor"],
+                 {"message": "suggest a goal now"})
+    chat4 = int([d for d in result if d.get("type") == "navigate"][0]
+                ["uri"].rsplit("/", 1)[1])
+    check(wait_for(lambda: [s for s in slot_public(
+        auth, "/chats/%d" % chat4, "chat")["suggestions"]
+        if s["kind"] == "goal"] or False, timeout=15,
+        what="goal suggestion") is not None,
+        "the goal suggestion lands")
+    card = [s for s in slot_public(auth, "/chats/%d" % chat4,
+                                   "chat")["suggestions"]
+            if s["kind"] == "goal"][0]
+    result = act(auth, card["decide"], {"decision": "approve"})
+    goal_rows = db_rows("select status, branch_name, creator from goals"
+                        " where title = 'Advisor arc goal'")
+    check(len(goal_rows) == 1 and goal_rows[0][0] == "draft"
+          and goal_rows[0][1] == "goal/advisor-arc"
+          and goal_rows[0][2] == "human",
+          "the approved goal is a DRAFT on its branch (creator human)")
+    goal_id = db_rows("select id from goals"
+                      " where title = 'Advisor arc goal'")[0][0]
+    suggestion4 = db_rows("select status, goal_id, task_id from"
+                          " advisor_suggestions where chat_id = ?"
+                          " and kind = 'goal'", (chat4,))[0]
+    check(suggestion4[0] == "approved"
+          and str(suggestion4[1]) == str(goal_id)
+          and str(suggestion4[2]) == "0",
+          "the goal suggestion keeps its goal_id and no task")
+    check(db_rows("select count(*) from tasks where goal_id = ?",
+                  (goal_id,))[0][0] == 0,
+          "the approved goal stays unqueued (no run task created)")
+    check(wait_for(lambda: [r for r in db_rows(
+        "select content from advisor_messages where chat_id = ?",
+        (chat4,)) if r[0].startswith(
+            "Approved — goal #%d created in draft" % goal_id)] or False,
+        timeout=15, what="goal approval notice") is not None,
+        "the goal approval lands its notice")
+    # approve_enqueue on a goal behaves as plain approve: never queued.
+    llm5 = ScriptedLlm([
+        {"tools": [{"id": "ga-2", "name": "suggest_goal",
+                    "arguments": json.dumps({
+                        "title": "Advisor arc goal two",
+                        "brief": "Never start me.",
+                        "project_id": 1,
+                        "branch_name": "goal/advisor-arc-two"})}]},
+        {"content": "Suggested."},
+        {"content": "Thanks."},
+    ])
+    llm5.serve()
+    auth = restart_with_llm(ctx, llm5)
+    home = slot_public(auth, "/", "dashboard")
+    result = act(auth, home["start_advisor"],
+                 {"message": "suggest another goal"})
+    chat5 = int([d for d in result if d.get("type") == "navigate"][0]
+                ["uri"].rsplit("/", 1)[1])
+    check(wait_for(lambda: [s for s in slot_public(
+        auth, "/chats/%d" % chat5, "chat")["suggestions"]
+        if s["kind"] == "goal"] or False, timeout=15,
+        what="second goal suggestion") is not None,
+        "the second goal suggestion lands")
+    card = [s for s in slot_public(auth, "/chats/%d" % chat5,
+                                   "chat")["suggestions"]
+            if s["kind"] == "goal"][0]
+    act(auth, card["decide"], {"decision": "approve_enqueue"})
+    goal5 = db_rows("select id, status from goals"
+                    " where title = 'Advisor arc goal two'")[0]
+    check(goal5[1] == "draft"
+          and db_rows("select count(*) from tasks where goal_id = ?",
+                      (goal5[0],))[0][0] == 0,
+          "approve_enqueue on a goal still creates only a draft")
+
+    print("== advisor_arc: decided rows refuse a second decision ==")
+    page = slot_public(auth, "/chats/%d" % chat5, "chat")
+    decided = [s for s in page["suggestions"] if s["kind"] == "goal"][0]
+    check("decide" not in decided,
+          "the decided card carries no decide action anymore")
+
+    print("== advisor_arc: privacy - another user decides nothing ==")
+    second = "arc-" + str(int(time.time()))
+    users_state = slot_public(auth, "/users", "users")
+    act(auth, users_state["create"], {
+        "username": second, "email": second + "@example.com",
+        "forename": "Arc", "surname": "Privacy", "date_of_birth": "1990-01-01",
+        "password": "arcpass123", "confirm_password": "arcpass123"})
+    other = login(second, "arcpass123")
+    other_page = slot_public(other, "/chats/%d" % chat_id, "chat")
+    check(other_page["id"] == 0,
+          "another user's chat slot reads not-found")
+    # The foreign decision is refused by the owner check: the row keeps
+    # its approved/task_id state untouched.
+    own = db_rows("select status, task_id from advisor_suggestions"
+                  " where id = ?", (suggestion_id,))[0]
+    check(own[0] == "approved" and int(own[1]) == task_id,
+          "the owner's row is intact before the foreign attempt")
+    foreign_card = [s for s in slot_public(
+        auth, "/chats/%d" % chat_id, "chat")["suggestions"]
+        if str(s["id"]) == str(suggestion_id)][0]
+    check("decide" not in foreign_card,
+          "the decided card exposes nothing to replay")
+
+    # Restore the daemon-facing LLM env the other checks may rely on
+    # (the advisor env is test-local).
+    for killer in (llm, llm2, llm3, llm4, llm5):
+        killer.kill()
+    os.environ.pop("ADVISOR_LLM_BASE_URL", None)
+    os.environ.pop("ADVISOR_LLM_MODEL", None)

test/harness.py

@@ -1324,7 +1324,8 @@ BUNDLES_ORDER = ("pages", "dispatch", "subtasks", "lineage",
                  "allocation", "lifecycle", "answers", "push", "goals",
                  "replay", "goalui", "forcewake", "users", "stacks", "notes",
                  "resurrect", "origins", "parity", "understanding",
-                 "discoveries", "knowledge_cycle", "advisor", "chatui")
+                 "discoveries", "knowledge_cycle", "advisor", "chatui",
+                 "advisor_arc")
 
 
 def run_bundle(name):