using Invercargill; using Invercargill.DataStructures; using Invercargill.Expressions; using InvercargillSql; using InvercargillSql.Orm; using Inversion; using Soup; using ClankerSaddle.Data; namespace ClankerSaddle.Services { /** Caps of the `project_notes` array in the /start payload. */ public const int64 PROJECT_NOTES_MAX_ENTRIES = 30; public const int64 PROJECT_NOTES_MAX_CHARS = 20000; /** A block answer awaiting collection by one saddle's next poll. */ public class PendingAnswer : Object { public int64 block_id { get; set; } public string answer { get; set; default = ""; } public bool branch_advanced { get; set; } } /** A human comment awaiting delivery to one saddle's conversation. */ public class PendingComment : Object { public int64 comment_id { get; set; } public string author { get; set; default = ""; } public string body { get; set; default = ""; } public string quote { get; set; default = ""; } } /** A saddle registration plus the API token generated for it. */ public class SaddleRegistration : Object { public Saddle saddle { get; set; } public string token { get; set; default = ""; } } /** * The OpenAI-compatible endpoint every saddle assignment is sent to, * read from the environment: `SADDLE_LLM_BASE_URL` (required — dispatch * pauses without it), `SADDLE_LLM_API_KEY` and `SADDLE_LLM_MODEL`. */ public class LlmEndpoint : Object { public string base_url { get; set; default = ""; } public string api_key { get; set; default = ""; } public string model { get; set; default = ""; } /** The configured endpoint, or null when the base URL is unset. */ public static LlmEndpoint? from_environment() { var base_url = (Environment.get_variable("SADDLE_LLM_BASE_URL") ?? "").strip(); if (base_url.length == 0) { return null; } return new LlmEndpoint() { base_url = base_url, api_key = (Environment.get_variable("SADDLE_LLM_API_KEY") ?? "").strip(), model = (Environment.get_variable("SADDLE_LLM_MODEL") ?? "").strip() }; } } /** * Saddle registry and control-plane client (plan 10 phase 2): registers * containers, finds idle capacity for the dispatcher's wake * computation, wakes sleeping daemons, and ingests the telemetry the * daemons PUSH — raw PTY chunks (`POST /api/saddle/chunks`) and round * results with state transitions (`POST /api/saddle/results`) — reusing * the storage logic the deleted `/output` + `/log` pollers ran. GET * /status against a daemon is exceptional only: the staleness probe * for a silent runner and the manual refresh button. Block answers are * handed out exactly once (delivered ids are remembered in memory; a * restart re-delivers and the daemon deduplicates). */ public class SaddleService : Object { private OrmSession db = inject(); private Connection connection = inject(); /** * Every daemon request shares this session and therefore its * timeouts: without them a single hung request (a daemon wedged * mid-response over the keep-alive pool) froze the dispatcher's * tick latch forever. `timeout` bounds each request * (SADDLE_HTTP_TIMEOUT_S, default 30s); `idle_timeout` reaps * pooled connections so a peer that vanished between requests is * not dialed again. */ private Soup.Session http = new Soup.Session() { timeout = http_timeout_seconds(), idle_timeout = 10 }; private static uint http_timeout_seconds() { var raw = Environment.get_variable("SADDLE_HTTP_TIMEOUT_S") ?? ""; int64 parsed = int64.parse(raw); return parsed > 0 ? (uint) parsed : 30; } private SaddlePollState poll_state = inject(); private StatePulse pulse = inject(); private PushService push = inject(); private RuleService rules = inject(); private CapabilityService capability_service = inject(); private TaskService task_service = inject(); private GoalService goal_service = inject(); private Astralis.WebConfig config = inject(); /** * Registers a saddle container: inserts the row with a freshly * minted per-saddle token (plan 10) and hands the credential to * the daemon via `POST /register {saddle_token, web_base_url}` * (best effort — an older daemon without the route stays * registered; the operator copies the token from the saddles * page instead). `host` is host:port of its control server; the * returned {@link SaddleRegistration} carries the token the * daemon must present as `Authorization: Bearer …` on * `/api/saddle/*` calls. */ public async SaddleRegistration register_saddle(string name, string host) throws Error { var saddle = new Saddle() { name = name, host = host, status = "idle", token = mint_saddle_token() }; yield db.insert_async(saddle); yield register_daemon(saddle); yield pulse.bump(0, "saddle registered"); return new SaddleRegistration() { saddle = saddle, token = saddle.token ?? "" }; } /** A fresh per-saddle bearer token (`saddle-`). */ private static string mint_saddle_token() { return @"saddle-$(Uuid.string_random().replace("-", ""))"; } /** * Delivers the per-saddle token to the daemon's * `POST /register` endpoint (wire contract): the daemon * persists it for its calls home and answers * `{ok, ssh_public_key, version}` — the key is stored on the * row the moment the daemon reports one. A daemon without the * route (HTTP 404/405) or unreachable at registration time is * tolerated: the row keeps the token, the operator can rety * via a refresh or copy the token manually. */ public async void register_daemon(Saddle saddle) throws Error { var payload = new Json.Builder(); payload.begin_object(); payload.set_member_name("saddle_token"); payload.add_string_value(saddle.token ?? ""); payload.set_member_name("web_base_url"); payload.add_string_value(callback_base_url()); payload.end_object(); try { var response = yield request_saddle_json(saddle.host, "POST", "/register", (owned) payload); string reported_key = string_member(response, "ssh_public_key"); if (reported_key.length > 0 && reported_key != (saddle.ssh_public_key ?? "")) { saddle.ssh_public_key = reported_key; yield db.update_async(saddle); } print(@"[ClankerSaddle] Registered with saddle '$(saddle.name)'s daemon (token delivered)\n"); } catch (Error e) { print(@"[ClankerSaddle] Saddle '$(saddle.name)'s daemon did not accept /register ($(e.message)); token stays on the row\n"); } } /** * Startup migration of the per-saddle token: every saddle row * gets its canonical token — the latest legacy `api_tokens` * value it already holds (daemons registered before plan 10 * keep their credential), or a freshly minted one. Existing * rows with a token are left alone. */ public async void ensure_saddle_tokens() throws Error { foreach (var saddle in yield list_saddles()) { if ((saddle.token ?? "").length > 0) { continue; } string? legacy = yield token_for_saddle(saddle.id); saddle.token = legacy != null && ((!)legacy).length > 0 ? (!)legacy : mint_saddle_token(); yield db.update_async(saddle); print(@"[ClankerSaddle] Saddle '$(saddle.name)' token ready ($(saddle.token.substring(0, 12))…)\n"); } } /** * Generates and stores a fresh API token for `saddle_id`, * replacing any earlier one on the wire (older rows remain valid * until removed; in practice each saddle holds one). */ public async string generate_token(int64 saddle_id) throws Error { var row = new ApiToken() { saddle_id = saddle_id, token = @"saddle-$(Uuid.string_random().replace("-", ""))", created_at = new DateTime.now_local() }; yield db.insert_async(row); return row.token; } /** * True when `token` matches the credential of `saddle_id` — its * per-saddle row token (plan 10), or any legacy `api_tokens` * row still held for it (tokens minted before the column * existed keep validating) — the credential check for the * `/api/saddle/*` endpoints. */ public async bool validate_token(int64 saddle_id, string token) throws Error { if (token.length == 0) { return false; } var saddle = yield get_saddle(saddle_id); if (saddle != null && ((!)saddle).token == token) { return true; } var rows = yield db.query() .where(expr("saddle_id == $0", elem(saddle_id))) .materialise_async(); foreach (var row in rows) { if (row.token == token) { return true; } } return false; } /** * The saddle a bearer token belongs to, or null — the * credential check for the id-less `/api/saddle/*` endpoints * (`next-task`, `note`, `task`): the row token * first, then any legacy `api_tokens` row (mapped to its * saddle). Removing the saddle revokes both. */ public async Saddle? saddle_for_token(string token) throws Error { if (token.length == 0) { return null; } var by_row = yield db.query() .where(expr("token == $0", elem(token))) .first_async(); if (by_row != null) { return by_row; } var legacy = yield db.query() .where(expr("token == $0", elem(token))) .first_async(); if (legacy == null) { return null; } return yield get_saddle(((!)legacy).saddle_id); } /** * The most recent API token issued for `saddle_id`, or null: the * row token first, then the latest legacy `api_tokens` row. */ public async string? token_for_saddle(int64 saddle_id) throws Error { var saddle = yield get_saddle(saddle_id); if (saddle != null && (((!)saddle).token ?? "").length > 0) { return ((!)saddle).token; } ApiToken? latest = null; var rows = yield db.query() .where(expr("saddle_id == $0", elem(saddle_id))) .order_by(expr("id")) .materialise_async(); foreach (var row in rows) { latest = row; } return latest != null ? ((!)latest).token : null; } /** * This app's externally reachable base URL — what daemons use to * POST notes back — read from the `saddle_callback_base_url` key * of web-config.json, falling back to the * `SADDLE_CALLBACK_BASE_URL` environment variable and then to * `http://localhost:8080`. The daemon typically runs in a * container or on another host, so its view of this app differs * from the browser's; set the key to whatever reaches this * process from there. */ public string callback_base_url() { string configured = config.get_string("saddle_callback_base_url", "").strip(); if (configured.length > 0) { return configured; } var from_env = (Environment.get_variable("SADDLE_CALLBACK_BASE_URL") ?? "").strip(); if (from_env.length > 0) { return from_env; } return "http://localhost:8080"; } /** * Removes a saddle registration; tasks it was running are * re-queued. Revocation: the row (and its per-saddle token) * dies here, and every legacy `api_tokens` row held for the * saddle is deleted — nothing authorises `/api/saddle/*` for * it afterwards. */ public async void unregister_saddle(int64 id) throws Error { var saddle = yield get_saddle(id); if (saddle == null) { return; } if (((!)saddle).current_task_id > 0) { var task = yield db.query() .where(expr("id == $0", elem(((!)saddle).current_task_id))) .first_async(); if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) { ((!)task).status = STATUS_QUEUED; yield db.update_async((!)task); } } foreach (var row in yield db.query() .where(expr("saddle_id == $0", elem(id))) .materialise_async()) { yield db.delete_async(row); } yield db.delete_async((!)saddle); yield pulse.bump(0, "saddle removed"); } /** Loads the saddle identified by `id`, or null. */ public async Saddle? get_saddle(int64 id) throws Error { return yield db.query() .where(expr("id == $0", elem(id))) .first_async(); } /** Lists every saddle ordered by name. */ public async ImmutableLot list_saddles() throws Error { return yield db.query() .order_by(expr("name")) .materialise_async(); } /** * Lists saddles with status 'busy' (the dispatcher's poll * candidates). */ public async ImmutableLot list_busy_saddles() throws Error { return yield db.query() .where(expr("status == $0", elem("busy"))) .materialise_async(); } /** * Records a /status observation into the health snapshot, * tolerating older daemons without the uptime/activity fields. */ /** * Persists the saddle's git identity — what its daemon commits * as (`git_identity` in every /start). Empty/null clears the * fields so the daemon's fallback applies. */ public async void update_git_identity(int64 id, string? git_name, string? git_email) throws Error { var saddle = yield get_saddle(id); if (saddle == null) { throw new Error(Quark.from_string("clanker-saddle"), 17, @"No saddle exists with id $(id)"); } ((!)saddle).git_name = git_name != null && git_name.length > 0 ? git_name : null; ((!)saddle).git_email = git_email != null && git_email.length > 0 ? git_email : null; yield db.update_async((!)saddle); yield pulse.bump(0, "saddle identity saved"); } /** * Persists the daemon's reported `ssh_public_key` onto the saddle * row (the plan-9 contract: the key travels on /status, empty * until the daemon generates one; the saddles page shows it for * adding as a deploy key). A single-column UPDATE — the * full-row write could stomp a concurrent ask claim. Only writes * when it changed. */ private async void store_reported_ssh_key(Saddle saddle, Json.Object status) throws Error { if (!status.has_member("ssh_public_key")) { return; } string reported = status.get_string_member_with_default("ssh_public_key", ""); string stored = saddle.ssh_public_key ?? ""; if (reported == stored) { return; } var command = connection.create_command( "UPDATE saddles SET ssh_public_key = :key WHERE id = :id"); command.with_parameter("key", elem(reported.length > 0 ? reported : null)); command.with_parameter("id", elem(saddle.id)); yield command.execute_non_query_async(); } /** * Records a /status observation: the health snapshot, the * daemon-reported ssh key, and — because the daemon surfaces * `resumed` on the FIRST status after a workspace restore, * whichever probe happens to receive it (the dispatcher's poll, * a heal pass, a manual refresh) — the once-per-session resume * progress entry. */ private async void record_status_probe(Saddle saddle, Json.Object status) throws Error { string state = status.has_member("state") ? status.get_string_member_with_default("state", "") : ""; int64 uptime = status.has_member("uptime_seconds") && status.get_member("uptime_seconds").get_value_type() == typeof(int64) ? status.get_int_member("uptime_seconds") : 0; int64 activity = status.has_member("last_activity_epoch") && status.get_member("last_activity_epoch").get_value_type() == typeof(int64) ? status.get_int_member("last_activity_epoch") : 0; poll_state.record_probe(saddle.id, state, uptime, activity); yield store_reported_ssh_key(saddle, status); // Plan 12: a daemon running (or parking — the workspace // stays while the saddle frees) a goal agent exposes the // push state of the goal branches on /status; cache it on // the goal row for the completion-review UI. if (status.has_member("goal_branches")) { var goal_branches = status.get_member("goal_branches"); if (goal_branches.get_node_type() == Json.NodeType.ARRAY || goal_branches.get_node_type() == Json.NodeType.OBJECT) { var generator = new Json.Generator(); generator.set_root(goal_branches); string raw = generator.to_data(null); int64 push_goal_id = 0; if (saddle.current_task_id > 0) { push_goal_id = yield goal_id_of_run(saddle.current_task_id); } else { // Parked: the open session still names the // goal agent's workspace. var open_session = yield open_session_on_saddle(saddle.id); if (open_session != null) { push_goal_id = yield goal_id_of_run(((!)open_session).task_id); } } if (push_goal_id > 0) { yield goal_service.store_branch_push_state(push_goal_id, raw); } } } // The "interrupted -> resumed" timeline arc. Absent on older // daemons; marked once per session in SaddlePollState. if (status.has_member("resumed") && status.get_member("resumed").get_value_type() == typeof(bool) && status.get_boolean_member("resumed") && saddle.current_task_id > 0) { var session = yield task_service.open_session_for_task(saddle.current_task_id); if (session != null && poll_state.mark_resumed(((!)session).id)) { yield task_service.add_progress(saddle.current_task_id, "Saddle restored the session context; task resumed"); } } } /** * Wakes a sleeping daemon (plan 10 wire contract): `POST /wake` * `{}` — a capable task is ready; sleeping daemons ask for work, * busy ones ignore it, idle-awake ones re-ask. Idempotent. The * caller records the pending wake for the lost-wake guard. */ public async void wake_saddle(Saddle saddle) throws Error { var payload = new Json.Builder(); payload.begin_object(); payload.end_object(); Json.Node? root = null; int status_code = yield request_saddle(saddle.host, "POST", "/wake", (owned) payload, out root); if (status_code >= 400) { throw new Error(Quark.from_string("clanker-saddle"), 49, @"wake returned HTTP $(status_code)"); } } /** * Cancelling / requeueing / stopping a task with a live session * reaches its daemon (wire contract): `POST /task-update * {kind: "cancel"}` — the authoritative stop through the daemon's * abort machinery, the workspace kept on disk per the park * contract. Success is narrated as "; daemon aborted"; a * transport failure marks the saddle offline (the daemon's next * ask proves it alive again and flips the row back to idle). The * daemon must never keep running a task the web has moved on * from as a zombie. */ public async void abort_live_session(TaskService tasks, int64 task_id, string verb) throws Error { var session = yield tasks.open_session_for_task(task_id); if (session == null) { return; } yield abort_session_on_its_saddle(tasks, (!)session, verb); } /** * SADDLE_STALE_S (default 90s): the silence window after which a * running daemon owes an exceptional GET /status probe. */ public static uint stale_after_seconds() { var raw = Environment.get_variable("SADDLE_STALE_S") ?? ""; int64 parsed = int64.parse(raw); return parsed > 0 ? (uint) parsed : 90; } /** * {@link abort_live_session} keyed by saddle instead of task — * the unregister path. The cancel is delivered to the saddle * whose own open session was found (NOT re-resolved through the * task's lowest-id open session, which a stale fixture row on * another saddle would redirect to the wrong daemon). */ public async void abort_saddle_session(TaskService tasks, int64 saddle_id, string verb) throws Error { var session = yield open_session_on_saddle(saddle_id); if (session == null) { return; } yield abort_session_on_its_saddle(tasks, (!)session, verb); } /** * The cancel delivery core shared by the task-keyed and * saddle-keyed abort paths: `POST /task-update {kind: cancel}` * to the daemon holding THIS session. */ private async void abort_session_on_its_saddle(TaskService tasks, ClankerSaddle.Data.Session session, string verb) throws Error { var saddle = yield get_saddle(session.saddle_id); if (saddle == null) { return; } var payload = new Json.Builder(); payload.begin_object(); payload.set_member_name("kind"); payload.add_string_value("cancel"); payload.end_object(); Json.Node? root = null; try { int status_code = yield request_saddle(((!)saddle).host, "POST", "/task-update", (owned) payload, out root); if (status_code >= 400) { throw new Error(Quark.from_string("clanker-saddle"), 45, @"task-update cancel returned HTTP $(status_code)"); } yield tasks.add_comment(session.task_id, "agent", @"$(verb); daemon aborted"); } catch (Error e) { warning(@"[ClankerSaddle] Abort for task $(session.task_id) failed: $(e.message); marking saddle '$(((!)saddle).name)' offline"); yield mark_saddle_unreachable(((!)saddle).id, tasks, @"abort delivery failed ($(e.message))"); yield tasks.add_comment(session.task_id, "agent", @"$(verb); daemon unreachable — saddle marked offline"); } } /** * Requeues tasks caught running on saddles that have since gone * offline (a manual status refresh, a restart) — without this * pass they would wait forever for a daemon that will not answer. * Only `assigned`/`working` tasks move: parked tasks * (blocked/review/suspended) keep their sessions, which the * resume machinery needs. Each move lands a agent comment so the * timeline shows why the task restarted. */ public async void requeue_tasks_on_offline_saddles(TaskService tasks) throws Error { foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) { foreach (var task in yield db.query() .where(expr("status == $0", elem(status))) .materialise_async()) { var session = yield tasks.open_session_for_task(task.id); if (session == null) { continue; } if (yield tasks.newest_block_answered(task.id)) { // Parked-with-answered: a verdict awaits the // resume machinery — never requeue it. continue; } var saddle = yield get_saddle(((!)session).saddle_id); if (saddle == null || ((!)saddle).status != "offline") { continue; } print(@"[ClankerSaddle] Task $(task.id) requeued: its saddle went offline\n"); yield tasks.requeue_task(task.id); yield tasks.add_comment(task.id, "agent", "saddle went offline; task requeued"); } } } /** * Parks a saddle (idle, no task) only when it still runs * `task_id` — a single conditional UPDATE, atomic against the * ask allocation claiming a NEW task for the saddle between the * poller's read and its write (the poller's "blocked"/"suspended" * parking races endpoint-side claims otherwise, leaving the * freshly-claimed task assigned to a busy@0 saddle nothing * polls). */ public async void park_if_current(int64 saddle_id, int64 task_id) throws Error { var command = connection.create_command( "UPDATE saddles SET status = 'idle', current_task_id = 0, allocated_at = NULL" + " WHERE id = :id AND current_task_id = :task"); command.with_parameter("id", elem(saddle_id)); command.with_parameter("task", elem(task_id)); yield command.execute_non_query_async(); yield pulse.bump(0, "saddle parked"); } /** * Every saddle with status 'idle', registration order — the * dispatcher's capability-matching candidate list. */ public async ImmutableLot list_idle_saddles() throws Error { return yield db.query() .where(expr("status == $0", elem("idle"))) .order_by(expr("id")) .materialise_async(); } /** * Marks a saddle offline when its control server cannot be * reached, releasing the task it was holding back to 'queued' * and closing that task's open session (outcome 'stopped'). The * requeue lands as a agent comment on the task, so the timeline * shows the interruption; combined with the daemon's * crash-resume, the next dispatch restores the workspace and * reports `resumed` on /status. */ public async void mark_saddle_unreachable(int64 id, TaskService tasks, string? reason = null) throws Error { var saddle = yield get_saddle(id); if (saddle == null) { return; } int64 held_task_at_entry = ((!)saddle).current_task_id; if (((!)saddle).current_task_id > 0) { int64 task_id = ((!)saddle).current_task_id; bool keep_parked = false; var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); if (task != null && (((!)task).status == STATUS_ASSIGNED || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED || ((!)task).status == STATUS_REVIEW_REQUESTED)) { keep_parked = yield tasks.newest_block_answered(task_id); if (keep_parked) { // Parked-with-answered: the verdict must survive // for the resume (answer delivery, or the // affinity allocation when the daemon asks // again) — the session stays open and the task // stays parked. print(@"[ClankerSaddle] Task $(task_id) keeps its parked session: its answered block awaits delivery\n"); } else { ((!)task).status = STATUS_QUEUED; yield db.update_async((!)task); string unreachable_text = reason != null && reason.length > 0 ? @"Saddle became unreachable ($(reason)); task requeued" : "Saddle became unreachable; task requeued"; yield tasks.add_comment(task_id, "agent", unreachable_text); } } if (!keep_parked) { var sessions = yield db.query() .where(expr("task_id == $0", elem(task_id))) .materialise_async(); foreach (var session in sessions) { if (session.ended_at != null) { continue; } session.ended_at = new DateTime.now_local(); session.outcome = "stopped"; yield db.update_async(session); } } } var offline_write = connection.create_command( "UPDATE saddles SET status = 'offline', current_task_id = 0, allocated_at = NULL" + " WHERE id = :id AND current_task_id = :seen"); offline_write.with_parameter("id", elem(id)); offline_write.with_parameter("seen", elem(held_task_at_entry)); yield offline_write.execute_non_query_async(); // An offline saddle is neither sleeping nor wake-awaiting; // a restarted daemon re-establishes both by asking. poll_state.clear_ask(id); yield pulse.bump(0, "saddle unreachable"); } /** * Assembles the assignment payload shared by the /start push and * the `POST /api/saddle/next-task` response — `{task, kind, * mode, creator, goal?, rules, project_notes, callback, * llm_config, session_id}` — with the project, links (the * goal's own links for goal work, else the project's), branch * fields and recent comments inside the task object, plus any * block answers not yet handed over (a resumed workspace picks * its pending answers up from the assignment itself), the * effective rule set ({@link RuleService.effective_rules}) as * the top-level `rules` array, the capped `project_notes` of * the primary and co-developed projects, the note-creation * `callback` (base URL + the saddle's own bearer token), the * saddle's `git_identity`, a `capabilities` context (plan 10 * phase 3: `saddle_tags` — the allocated saddle's own tags — * and `fleet_tags` — every tag name known across the fleet), * the plan-12 goal context (`goal` for goal work; the * worker's `ephemeral_branch`) and — for resumed workspaces — * the `resume_notice`/`branch_advanced` context. `session_id` * (the web-side run row the daemon reports against) rides * along so pushed telemetry can key on it. Project checkouts * are addressed by the `SADDLE_PROJECTS_ROOT` convention * (`root/task-/name`) and cloned from each project's * `git_url`; the LLM endpoint by the {@link LlmEndpoint} * environment. */ public async Json.Object build_assignment_payload(Saddle saddle, TaskEntity task, int64 session_id, ProjectService projects, string? resume_notice = null, bool? branch_advanced = null, string? context_update = null) throws Error { var endpoint = LlmEndpoint.from_environment(); if (endpoint == null) { throw new Error(Quark.from_string("clanker-saddle"), 24, "No LLM endpoint configured (set SADDLE_LLM_BASE_URL)"); } var project = yield db.query() .where(expr("id == $0", elem(task.project_id))) .first_async(); if (project == null) { throw new Error(Quark.from_string("clanker-saddle"), 12, @"Task $(task.id) has no project $(task.project_id)"); } // Plan 12: goal work (the agent's own run and its worker // tasks) scopes its co-developed/reference set through the // GOAL's links; standalone tasks through the project's. Goal? goal = null; if (task.goal_id > 0) { goal = yield db.query() .where(expr("id == $0", elem(task.goal_id))) .first_async(); } var co_developed = new Vector(); var references = new Vector(); if (goal != null) { foreach (var link in yield goal_service.list_links(task.goal_id)) { var linked = yield db.query() .where(expr("id == $0", elem(link.project_id))) .first_async(); if (linked == null) { continue; } if (link.relationship == LINK_CO_DEVELOPED) { co_developed.add((!)linked); } else if (link.relationship == LINK_REFERENCE) { references.add((!)linked); } } } else { foreach (var link in yield projects.list_links(task.project_id)) { var linked = yield db.query() .where(expr("id == $0", elem(link.linked_project_id))) .first_async(); if (linked == null) { continue; } if (link.relationship == LINK_CO_DEVELOPED) { co_developed.add((!)linked); } else if (link.relationship == LINK_REFERENCE) { references.add((!)linked); } } } var payload = new Json.Builder(); payload.begin_object(); payload.set_member_name("task"); payload.begin_object(); payload.set_member_name("id"); payload.add_int_value(task.id); payload.set_member_name("title"); payload.add_string_value(task.title); payload.set_member_name("description"); payload.add_string_value(task.description ?? ""); payload.set_member_name("project_name"); payload.add_string_value(((!)project).name); payload.set_member_name("project_branch"); payload.add_string_value(((!)project).default_branch); payload.set_member_name("task_branch"); payload.add_string_value(task.task_branch ?? ""); payload.set_member_name("branch_from"); payload.add_string_value(task.branch_from ?? ((!)project).default_branch); payload.set_member_name("co_developed"); payload.begin_array(); payload.add_value(project_node(((!)project), task.id)); foreach (var linked in co_developed) { payload.add_value(project_node(linked, task.id)); } payload.end_array(); payload.set_member_name("reference"); payload.begin_array(); foreach (var linked in references) { payload.add_value(project_node(linked, task.id, true)); } payload.end_array(); payload.set_member_name("comments"); payload.begin_array(); var comments = yield db.query() .where(expr("task_id == $0", elem(task.id))) .order_by(expr("id")) .materialise_async(); foreach (var comment in comments) { payload.begin_object(); payload.set_member_name("author"); payload.add_string_value(comment.author); payload.set_member_name("body"); payload.add_string_value(comment.body); if (comment.quote != null) { payload.set_member_name("quote"); payload.add_string_value(((!)comment.quote)); } payload.end_object(); } payload.end_array(); payload.set_member_name("answers"); payload.begin_array(); foreach (var pending in yield take_pending_answers_for_task(task.id)) { payload.begin_object(); payload.set_member_name("block_id"); payload.add_int_value(pending.block_id); payload.set_member_name("answer"); payload.add_string_value(raw_answer_for_saddle(pending.answer)); if (branch_advanced != null) { payload.set_member_name("branch_advanced"); payload.add_boolean_value((!)branch_advanced); } payload.end_object(); } payload.end_array(); payload.end_object(); // The plan-12 wire additions: kind ("task"|"goal_run"), // mode and creator ride the top level beside the task. payload.set_member_name("kind"); payload.add_string_value(task.kind); payload.set_member_name("mode"); payload.add_string_value(task.mode); payload.set_member_name("creator"); payload.add_string_value(task.creator); // The goal context (goal work only): the goal agent gets // the full brief + approved plan; a goal worker gets the // same object as read-only reference (the daemon adds the // stick-to-mandate shaping) plus its ephemeral branch. if (goal != null) { payload.set_member_name("goal"); payload.begin_object(); payload.set_member_name("id"); payload.add_int_value(((!)goal).id); payload.set_member_name("branch"); payload.add_string_value(((!)goal).branch_name); payload.set_member_name("base_branch"); payload.add_string_value(((!)goal).base_branch ?? ((!)project).default_branch); payload.set_member_name("brief"); payload.add_string_value(((!)goal).brief); payload.set_member_name("plan"); payload.add_string_value(((!)goal).plan ?? ""); payload.end_object(); if (task.kind == TASK_KIND_TASK) { payload.set_member_name("ephemeral_branch"); payload.add_string_value(@"task/$(task.goal_id)-$(task.id)"); } } // The wire contract's top-level `rules` addition: the // effective rule set (task ∪ primary ∪ co-developed, deduped, // stable order — see RuleService.effective_rules). Goal work // expands co-developed through the goal's links. payload.set_member_name("rules"); payload.begin_array(); foreach (var rule in yield rules.effective_rules_for(task, projects, goal != null ? task.goal_id : 0)) { payload.begin_object(); payload.set_member_name("name"); payload.add_string_value(rule.name); payload.set_member_name("body"); payload.add_string_value(rule.body); payload.end_object(); } payload.end_array(); // The wire contract's `git_identity` addition (plan 9): the // name/email the daemon commits as, from the saddle row. // Omitted entirely when unset so the daemon's fallback // identity applies. string identity_name = saddle.git_name ?? ""; string identity_email = saddle.git_email ?? ""; if (identity_name.length > 0 || identity_email.length > 0) { payload.set_member_name("git_identity"); payload.begin_object(); payload.set_member_name("name"); payload.add_string_value(identity_name); payload.set_member_name("email"); payload.add_string_value(identity_email); payload.end_object(); } // The wire contract's `capabilities` context (plan 10 // phase 3): the allocated saddle's own tags plus every tag // name known across the fleet (the operator-managed // catalogue) — the daemon renders both into its system // prompt and validates `task create --capability`. payload.set_member_name("capabilities"); payload.begin_object(); payload.set_member_name("saddle_tags"); payload.begin_array(); foreach (var carried in yield capability_service.list_for_saddle(saddle.id)) { payload.add_string_value(carried.name); } payload.end_array(); payload.set_member_name("fleet_tags"); payload.begin_array(); foreach (var known in yield capability_service.list_capabilities()) { payload.add_string_value(known.name); } payload.end_array(); payload.end_object(); // The wire contract's `resume_notice` addition: on a // workspace-adopting start (a suspended parent's resume), the // daemon injects it as the next user turn instead of the // opening prompt. if (resume_notice != null && resume_notice.length > 0) { payload.set_member_name("resume_notice"); payload.add_string_value(resume_notice); } // The wire's `branch_advanced` (resume pushes): whether the // shared branch moved while this task was parked — false or // absent means the daemon omits the pull instruction entirely. if (branch_advanced != null) { payload.set_member_name("branch_advanced"); payload.add_boolean_value((!)branch_advanced); } // The wire's `context_update` (resume pushes): a short // markdown delta of the instruction context that changed // while the workspace was parked — the goal's plan // approved since the run's last round, rules created, // edited or attached, or new project notes. The daemon // appends it as a TURN (prompt-cache friendly) instead of // refreshing the system prompt; absent when nothing // changed. if (context_update != null && context_update.length > 0) { payload.set_member_name("context_update"); payload.add_string_value(context_update); } // The wire contract's `project_notes` addition: notes of the // primary and co-developed projects, oldest first, capped at // 30 entries / 20 000 chars (head kept). var note_project_names = new Dictionary(); var project_notes_rows = new Vector(); note_project_names.set(((!)project).id.to_string(), ((!)project).name); foreach (var note in yield projects.list_notes(((!)project).id)) { project_notes_rows.add(note); } foreach (var linked in co_developed) { note_project_names.set(linked.id.to_string(), linked.name); foreach (var note in yield projects.list_notes(linked.id)) { project_notes_rows.add(note); } } var project_notes = project_notes_rows.sort((a, b) => (int)(a.id - b.id)); payload.set_member_name("project_notes"); payload.begin_array(); int64 note_entries = 0; int64 note_chars = 0; foreach (var note in project_notes) { if (note_entries >= PROJECT_NOTES_MAX_ENTRIES || note_chars >= PROJECT_NOTES_MAX_CHARS) { break; } string note_body = note.body; if (note_chars + note_body.length > PROJECT_NOTES_MAX_CHARS) { note_body = note_body.substring(0, (int)(PROJECT_NOTES_MAX_CHARS - note_chars)); } payload.begin_object(); payload.set_member_name("project"); payload.add_string_value(note_project_names.get_or_default(note.project_id.to_string()) ?? ""); payload.set_member_name("body"); payload.add_string_value(note_body); payload.end_object(); note_entries++; note_chars += note_body.length; } payload.end_array(); // The wire contract's `callback` addition: this app's // externally reachable base URL plus the bearer token the // daemon presents when posting back — its own per-saddle // credential (plan 10), which every `/api/saddle/*` call // carries. string callback_token = saddle.token ?? ""; if (callback_token.length == 0) { string? legacy = yield token_for_saddle(saddle.id); callback_token = legacy ?? ""; } payload.set_member_name("callback"); payload.begin_object(); payload.set_member_name("base_url"); payload.add_string_value(callback_base_url()); payload.set_member_name("token"); payload.add_string_value(callback_token); payload.end_object(); payload.set_member_name("llm_config"); payload.begin_object(); payload.set_member_name("base_url"); payload.add_string_value(((!)endpoint).base_url); payload.set_member_name("api_key"); payload.add_string_value(((!)endpoint).api_key); payload.set_member_name("model"); payload.add_string_value(((!)endpoint).model); payload.end_object(); // The web-side run row the daemon's pushed telemetry keys on // (plan-10 phase-1 wire addition: it rides every assignment // so /start consumers can ignore it harmlessly). payload.set_member_name("session_id"); payload.add_int_value(session_id); payload.end_object(); var root = payload.get_root().copy(); return root.get_object(); } /** * Probes a saddle's `GET /status` and refreshes its registered * status: 'idle' or 'busy' from the daemon state, 'offline' when * the daemon cannot be reached. The write is guarded on the * assignment seen at load — an ask allocation may claim the * saddle while the probe is in flight, and must not be stomped. */ public async void refresh_saddle_status(int64 id) throws Error { var saddle = yield get_saddle(id); if (saddle == null) { throw new Error(Quark.from_string("clanker-saddle"), 17, @"No saddle exists with id $(id)"); } int64 seen_task = ((!)saddle).current_task_id; string next_status = "idle"; int64 next_task = 0; DateTime? next_allocated = null; try { var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null); string state = status.has_member("state") ? status.get_string_member_with_default("state", "") : ""; yield record_status_probe(((!)saddle), status); if (state == "working") { // Reconcile with the web's own records: busy only when // a live session truly holds this saddle. var session = yield open_session_on_saddle(id); next_status = "busy"; next_task = session != null ? ((!)session).task_id : 0; next_allocated = session != null ? ((!)session).started_at : null; } // Idle AND the parked states (blocked/suspended, plus // terminal and unknown states): the workspace is // parked, so the saddle is logically free — marking it // busy here used to wedge assignment against a daemon // that would never poll. } catch (Error e) { next_status = "offline"; } var command = connection.create_command( "UPDATE saddles SET status = :status, current_task_id = :task," + " allocated_at = :allocated WHERE id = :id AND current_task_id = :seen"); command.with_parameter("status", elem(next_status)); command.with_parameter("task", elem(next_task)); command.with_parameter("allocated", elem(next_allocated)); command.with_parameter("id", elem(id)); command.with_parameter("seen", elem(seen_task)); yield command.execute_non_query_async(); yield pulse.bump(0, "saddle status refreshed"); } /** * The saddle's open (live or parked) session, or null — the * reconciliation source for whether a daemon is genuinely * running one of our tasks. */ private async ClankerSaddle.Data.Session? open_session_on_saddle(int64 saddle_id) throws Error { var sessions = yield db.query() .where(expr("saddle_id == $0", elem(saddle_id))) .order_by(expr("id")) .materialise_async(); ClankerSaddle.Data.Session? newest_open = null; foreach (var session in sessions) { if (session.ended_at == null) { newest_open = session; } } return newest_open; } /** * Answered blocks of the saddle's current task not yet handed over * the poll endpoint; marks them delivered. */ public async ImmutableLot take_pending_answers(int64 saddle_id) throws Error { var saddle = yield get_saddle(saddle_id); if (saddle == null || ((!)saddle).current_task_id == 0) { return new Vector().to_immutable_buffer(); } return yield take_pending_answers_for_task(((!)saddle).current_task_id); } /** * Answered blocks of `task_id` not yet handed over anywhere; * marks them delivered. Used both by the poll endpoint and by * {@link push_task_to_saddle} (a resumed assignment carries its * pending answers in the payload). */ public async ImmutableLot take_pending_answers_for_task(int64 task_id) throws Error { var blocks = yield db.query() .where(expr("task_id == $0", elem(task_id))) .materialise_async(); var rows = new Vector(); foreach (var block in blocks) { if (block.answer == null) { continue; } if (!poll_state.mark_delivered(block.id)) { continue; } rows.add(new PendingAnswer() { block_id = block.id, answer = raw_answer_for_saddle(((!)block).answer ?? "") }); } return rows.to_immutable_buffer(); } /** * Human comments of the saddle's current task not yet delivered to * the daemon; marks them delivered (the poll response hands them * over exactly once). */ public async ImmutableLot take_pending_comments(int64 saddle_id) throws Error { var saddle = yield get_saddle(saddle_id); if (saddle == null || ((!)saddle).current_task_id == 0) { return new Vector().to_immutable_buffer(); } var comments = yield db.query() .where(expr("task_id == $0", elem(((!)saddle).current_task_id))) .order_by(expr("id")) .materialise_async(); var rows = new Vector(); foreach (var comment in comments) { if (!comment.author.has_prefix("user:") || poll_state.is_comment_delivered(comment.id)) { continue; } poll_state.mark_comment_delivered(comment.id); rows.add(new PendingComment() { comment_id = comment.id, author = comment.author, body = comment.body, quote = comment.quote ?? "" }); } return rows.to_immutable_buffer(); } /** * Pushes the task's undelivered human comments to the RUNNING * daemon's `POST /task-update {kind: "comment"}` endpoint so * they land in the live conversation. Best effort: a daemon * without the route (HTTP 404/405) keeps the comments pending, * and other failures are logged and retried on the next * dispatcher tick. */ public async void deliver_pending_comments(int64 saddle_id, int64 task_id) throws Error { var saddle = yield get_saddle(saddle_id); if (saddle == null) { return; } var comments = yield db.query() .where(expr("task_id == $0", elem(task_id))) .order_by(expr("id")) .materialise_async(); foreach (var comment in comments) { if (!comment.author.has_prefix("user:") || poll_state.is_comment_delivered(comment.id)) { continue; } var payload = new Json.Builder(); payload.begin_object(); payload.set_member_name("kind"); payload.add_string_value("comment"); payload.set_member_name("text"); payload.add_string_value(comment.body); payload.set_member_name("quote"); payload.add_string_value(comment.quote ?? ""); payload.end_object(); Json.Node? root = null; int status_code = yield request_saddle(((!)saddle).host, "POST", "/task-update", (owned) payload, out root); if (status_code == 200 || status_code == 204) { poll_state.mark_comment_delivered(comment.id); } else if (status_code != 404 && status_code != 405) { warning(@"[ClankerSaddle] Comment task-update to saddle $(saddle_id) returned HTTP $(status_code)"); } } } /** * Marks the task's human comments as delivered — they were handed * over inside an assignment payload, so neither the poll endpoint * nor `/notify` should repeat them. */ public async void mark_task_comments_delivered(int64 task_id) throws Error { var comments = yield db.query() .where(expr("task_id == $0", elem(task_id))) .materialise_async(); foreach (var comment in comments) { if (comment.author.has_prefix("user:")) { poll_state.mark_comment_delivered(comment.id); } } } /** * Projects the agent's conversation-level events onto the task: * `saddle comment` lines become agent comments, `saddle plan` * lines update the plan, `saddle progress-update` lines append * progress, `saddle review` raises the review banner, the * `saddle done` entry completes the task with the summary the * agent gave, and the `saddle failure` entry fails it with the * explanation the agent piped in. `saddle yeehaw` entries are * handled explicitly but project nothing — the stored * `session_commands` row IS the celebration the card, badge * and timeline read back. Returns true once a done or failed * event finished the task. * * Plan 12: on a GOAL-RUN session the plan/done/failure verbs * are goal-context — a plan submission enters plan review, a * done submits the completion report (the goal-run task parks * instead of completing), a failure fails the goal — and * comments also land on the goal's thread. * * Plan 13 B2: the projections key on the DEDUPED entry stream * ({@link store_pushed_entries} claims every id once per * saddle, ever) and stop at terminal state — a task already * done/failed, a goal already finished, projects nothing * further. The live failure's 115-event/187-comment fan-out is * structurally impossible: replays never reach this function, * and the honest burst of DISTINCT failure ids dies at the * first one. */ public async bool apply_entry_events(int64 task_id, Vector entries, TaskService tasks) throws Error { var goal_id = yield goal_id_of_run(task_id); bool terminal = yield task_is_terminal(task_id); if (!terminal && goal_id > 0) { terminal = yield goal_is_terminal(goal_id); } if (terminal) { return false; } foreach (var entry in entries) { string command = string_member(entry, "command"); string output = string_member(entry, "output"); switch (string_member(entry, "type")) { case "comment": // Plan 13 B4: body-first — the entry's output IS // the comment body; the command echo (the // `saddle comment` invocation, heredoc shapes // included) must never become the projected // text. The parsed command stays as a legacy // fallback for pushes without output. string body = output; if (body.length == 0) { body = strip_quotes(strip_verb(command, "saddle comment")); } if (body.length > 0) { yield tasks.add_comment(task_id, "agent", body); if (goal_id > 0) { yield goal_service.add_comment(goal_id, "agent", body); yield goal_service.record_event(goal_id, "comment", snippet_text(body)); } } break; case "plan": // Plan 12: plans are goal-context only — a // goal-run's `saddle plan` enters plan review; // plain task runs carry no plan member. if (goal_id > 0) { string markdown = output.length > 0 ? output : strip_quotes(strip_verb(command, "saddle plan")); if (markdown.length > 0) { yield goal_service.submit_plan(goal_id, markdown); } } break; case "progress": string summary = strip_verb(command, "saddle progress-update"); if (summary.length == 0) { summary = output; } if (summary.length > 0) { yield tasks.add_progress(task_id, strip_quotes(summary)); } break; case "review": yield tasks.request_review(task_id, output.length > 0 ? output : strip_quotes(strip_verb(command, "saddle review"))); break; case "done": if (goal_id > 0) { string report = output.length > 0 ? output : strip_quotes(strip_verb(command, "saddle done")); yield goal_service.submit_completion(goal_id, report.length > 0 ? report : "The goal agent submitted its completion report"); return true; } { // the DONE entry's output IS the report // (the command is just "saddle done") — // falling back to it keeps the summary when // this entry path beats the transition push string report = output.length > 0 ? output : strip_quotes(strip_verb(command, "saddle done")); yield tasks.complete_task(task_id, report.length > 0 ? report : null, null, "done"); } return true; case "failed": string explanation = output.length > 0 ? output : strip_quotes(strip_verb(command, "saddle failure")); if (goal_id > 0) { yield goal_service.fail_goal(goal_id, explanation.length > 0 ? explanation : null); return true; } yield tasks.fail_task(task_id, explanation); return true; case "yeehaw": // The celebration needs no task-side projection: // store_pushed_entries already persisted the row // the yeehaw card, badge and timeline read from. break; default: break; } } return false; } /** * The goal id when `task_id` is a goal agent's run task (kind * 'goal_run'), else 0 — the discriminator for the goal-context * verb handling above. */ private async int64 goal_id_of_run(int64 task_id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); return task != null && ((!)task).kind == TASK_KIND_GOAL_RUN ? ((!)task).goal_id : 0; } /** A one-line snippet of a body (narrative event text). */ private static string snippet_text(string body) { string flat = body.replace("\n", " ").strip(); return flat.length > 160 ? flat.substring(0, 160) + "…" : flat; } /** * Raises the daemon's pushed blocking prompt as a `task_blocks` * row exactly once: a not-yet-answered row, or an answered row * whose answer has not been delivered to the daemon yet, means the * prompt is already represented. The results push's `block` * object carries the question plus the block type, options and * markdown body (the `/status` field names — `block_question` * &c — are accepted as a fallback). `saddle block rfc` carries * no question text, so rfc prompts surface under a stable * fallback title. */ private async void ensure_block(TaskService tasks, int64 task_id, Json.Object block) throws Error { string question = block_question_from_payload(block); if (question.length == 0) { return; } foreach (var existing in yield tasks.list_blocks(task_id)) { if (existing.question != question) { continue; } if (existing.answer == null || !poll_state.is_delivered(existing.id)) { return; } } yield tasks.add_block(task_id, question, block_string_or(block, "type", "block_type"), string_array_member(block, block.has_member("options") ? "options" : "block_options"), optional_string_member(block, block.has_member("body") ? "body" : "block_body")); } /** The first non-empty of two string members (empty when both are). */ private static string block_string_or(Json.Object obj, string primary, string fallback) { string value = string_member(obj, primary); return value.length > 0 ? value : string_member(obj, fallback); } /** * The question a daemon block is recognised by: `question` * (or the `/status`-style `block_question`) verbatim, or the * rfc fallback when the daemon raised an RFC (the * `saddle block rfc` shape carries its document as the body and * no question operand). Both the block-raising and answer- * delivery paths key blocks by this string. */ private static string block_question_from_payload(Json.Object block) { string question = block_string_or(block, "question", "block_question"); if (question.length > 0) { return question; } return block_string_or(block, "type", "block_type") == "rfc" ? "Request for comments" : ""; } /** * Stores a pushed batch of raw PTY chunks (`POST * /api/saddle/chunks`, plan 10 phase 2) as the open session's * `session_raw_log` rows — the same storage logic the deleted * `GET /log` poller ran, driven by the daemon's flusher instead * of the web's cursor. Idempotent on the unique * (session_id, seq) pair: a chunk whose row already exists (an * outbox redelivery after a lost response) is consumed, not an * error; anything else stops the pass so the daemon's retry * re-sends the tail. A truly empty text stores no row (it * renders as nothing); whitespace-only chunks DO store. Returns * the count of rows newly stored. */ public async int64 store_pushed_chunks(int64 session_id, Json.Array chunks) throws Error { int64 stored = 0; foreach (var element in chunks.get_elements()) { if (element.get_node_type() != Json.NodeType.OBJECT) { continue; } var chunk = element.get_object(); int64 seq = chunk.has_member("seq") && chunk.get_member("seq").get_value_type() == typeof(int64) ? chunk.get_int_member("seq") : 0; if (seq <= 0) { continue; } string text = string_member(chunk, "text"); if (text.length == 0) { continue; } try { yield db.insert_async(new SessionRawLog() { session_id = session_id, seq = seq, chunk = text }); stored++; } catch (Error insert_error) { // The unique (session_id, seq) pair is the idempotency // guarantee: a row already there means the push is a // redelivery — consume it. Any other failure is real. var existing = yield db.query() .where(expr("session_id == $0 && seq == $1", elem(session_id), elem(seq))) .first_async(); if (existing == null) { throw insert_error; } } } return stored; } /** * Stores a pushed batch of round entries (`POST * /api/saddle/results`, plan 10 phase 2) as `session_commands` * rows — the mapping the deleted `GET /output` poller ran. Plan * 13 B1: the dedup key is the DAEMON-GLOBAL (saddle, entry id) * pair ({@link SaddleEntryDedup}), persisted in the database so * it survives web restarts and ignores session churn entirely — * a redelivered entry is consumed silently under whatever * session id the payload carries, and a push may only CREATE * rows for entries whose ids have not already landed on this * saddle. Entries without an id always store. Returns the * entries stored, oldest first, for {@link apply_entry_events} * — the deduped entry stream every projection keys on (B2). */ public async Vector store_pushed_entries(Saddle saddle, int64 session_id, Json.Array entries) throws Error { var stored = new Vector(); foreach (var element in entries.get_elements()) { if (element.get_node_type() != Json.NodeType.OBJECT) { continue; } var entry = element.get_object(); int64 entry_id = entry.has_member("id") && entry.get_member("id").get_value_type() == typeof(int64) ? entry.get_int_member("id") : 0; if (entry_id > 0 && !(yield claim_entry_id(saddle.id, entry_id))) { // A redelivery under any session binding — consumed, // never duplicated, never projected. continue; } var command = new SessionCommand() { session_id = session_id, command_text = string_member(entry, "command"), output_text = entry.has_member("output") && entry.get_member("output").get_string() != null ? entry.get_string_member("output") : null, exit_code = entry.has_member("exit_code") && entry.get_member("exit_code").get_value_type() == typeof(int64) ? entry.get_int_member("exit_code") : 0, entry_type = string_member(entry, "type").length > 0 ? string_member(entry, "type") : "command", piped_input = optional_string_member(entry, "input"), started_at = parse_iso(entry, "started_at") ?? parse_iso(entry, "timestamp"), completed_at = parse_iso(entry, "completed_at"), duration_ms = entry.has_member("duration_ms") && entry.get_member("duration_ms").get_value_type() == typeof(int64) ? entry.get_int_member("duration_ms") : 0 }; yield db.insert_async(command); stored.add(entry); } return stored; } /** * Claims `(saddle_id, entry_id)` for ingestion: true when the * id is fresh on this saddle, false when it already landed (an * outbox redelivery — consumed silently regardless of the * payload's session binding). The dedup row is inserted BEFORE * the entry stores, so a mid-pass failure can never re-project * an entry the daemon retries (exactly-once projections beat * at-least-once storage). */ private async bool claim_entry_id(int64 saddle_id, int64 entry_id) throws Error { try { yield db.insert_async(new SaddleEntryDedup() { saddle_id = saddle_id, entry_id = entry_id }); return true; } catch (Error insert_error) { var existing = yield db.query() .where(expr("saddle_id == $0 && entry_id == $1", elem(saddle_id), elem(entry_id))) .first_async(); if (existing == null) { throw insert_error; } return false; } } /** * Applies a pushed state transition (`POST /api/saddle/results`, * plan 10 phase 2) — the same task/saddle bookkeeping the * deleted /status poll switch ran, driven by the daemon's push: * `done`/`failed` complete or fail the task with the pushed * summary (freeing the saddle and closing the session) — * except on a goal-run session, where they submit the * completion report or fail the goal (plan 12); `blocked` * raises the block payload as an answerable block on the * goal run's OWN task row (plan 15 C1: the goal agent's * blocks ride the run task — no mirrored second row exists) * and parks the saddle; `suspended` * parks the same way — for a goal agent the resume is driven * by the goal's task-terminal events; and `working` flips the * task to working (never downgrading a pending review). * * Plan 13 B2: a push against a task that is already terminal — * or a goal-context push against a finished goal — is an * outbox redelivery and is consumed silently, whatever state * it carries; terminal transitions and their narrative fire * once, ever. */ public async void apply_result_state(Saddle saddle, ClankerSaddle.Data.Session session, string state, Json.Object? block, string? summary, TaskService tasks) throws Error { int64 task_id = session.task_id; int64 goal_id = yield goal_id_of_run(task_id); bool terminal = yield task_is_terminal(task_id); if (!terminal && goal_id > 0) { terminal = yield goal_is_terminal(goal_id); } if (terminal) { return; } switch (state) { case "done": if (goal_id > 0) { // Goal context: `saddle done` submits the // completion report and enters completion // review — the run parks on the mirrored // block, it does not complete. yield goal_service.submit_completion(goal_id, summary != null && summary.length > 0 ? summary : "The goal agent submitted its completion report"); yield park_if_current(saddle.id, task_id); break; } yield tasks.complete_task(task_id, summary, null, "done"); break; case "failed": string explanation = summary ?? ""; if (goal_id > 0) { yield goal_service.fail_goal(goal_id, explanation.length > 0 ? explanation : null); break; } yield tasks.fail_task(task_id, explanation.length > 0 ? explanation : null); break; case "blocked": if (goal_id > 0) { // Goal context: the block lands on the run // task itself (plan 15 C1) — a plan payload // enters plan review, anything else raises a // goal question/RFC; the run parks on that // very row for the verdict resume. A BLOCKED // push WITHOUT a // block payload (the park-unwinding round-end // push, its trailing entries) raises nothing — // the payload-carrying transition already did, // and a synthesised fallback row would dupe it. if (block == null) { yield tasks.set_status(task_id, STATUS_BLOCKED); yield park_if_current(saddle.id, task_id); break; } string question = block_question_from_payload((!)block); if (question.length == 0) { question = "The goal agent raised a question"; } if (block_string_or((!)block, "type", "block_type") == BLOCK_PLAN) { yield goal_service.submit_plan(goal_id, optional_string_member((!)block, "body") ?? question); } else { yield goal_service.raise_agent_block(goal_id, question, block_string_or((!)block, "type", "block_type"), string_array_member((!)block, (!)block.has_member("options") ? "options" : "block_options"), optional_string_member((!)block, "body")); } yield tasks.set_status(task_id, STATUS_BLOCKED); yield park_if_current(saddle.id, task_id); break; } yield ensure_block(tasks, task_id, block != null ? (!)block : new Json.Object()); { // A review-kind block parks as review_requested // (the banner state); every other kind as // blocked. bool review = block != null && block_string_or((!)block, "type", "block_type") == BLOCK_REVIEW; yield tasks.set_status(task_id, review ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED); } // The daemon parks the task's workspace while blocked, // freeing the saddle for other work. The blocked task // plus its open session is the pause record the // affinity allocation resumes from once the block is // answered. yield park_if_current(saddle.id, task_id); break; case "suspended": // `saddle suspend`: the same parking as a block, but // the web drives resumption — a wake of the affine // saddle once a goal event lands, the resume // assignment carrying the coalesced report. yield tasks.suspend_task(task_id); yield park_if_current(saddle.id, task_id); break; case "working": // A pending review keeps its banner: started_at (and // the first-start push) only stamp for work that // never allocated. var current = yield tasks.set_status(task_id, STATUS_WORKING); if (current != null && ((!)current).status != STATUS_REVIEW_REQUESTED && ((!)current).started_at == null) { ((!)current).started_at = new DateTime.now_local(); yield db.update_async((!)current); // The first-start push lives at the allocation // (where started_at is stamped for dispatched // tasks); this path only covers legacy status // pushes for never-allocated work. push.notify_opted_in(task_id, PushService.lifecycle_notification( ((!)current).title, task_id, "started", null)); } break; default: break; } } /** True when the task's status is done or failed. */ private async bool task_is_terminal(int64 task_id) throws Error { var task = yield db.query() .where(expr("id == $0", elem(task_id))) .first_async(); return task != null && (((!)task).status == STATUS_DONE || ((!)task).status == STATUS_FAILED); } /** True when the goal's status is complete, failed or cancelled. */ private async bool goal_is_terminal(int64 goal_id) throws Error { var goal = yield db.query() .where(expr("id == $0", elem(goal_id))) .first_async(); return goal != null && (((!)goal).status == GOAL_COMPLETE || ((!)goal).status == GOAL_FAILED || ((!)goal).status == GOAL_CANCELLED); } /** * The running-liveness probe (plan 10 phase 2): a busy saddle's * session has seen neither chunks nor results for the staleness * window, so ONE exceptional `GET /status` decides its fate — * an answer means the daemon is alive (the health snapshot is * refreshed and the task keeps running; the silence episode * gets no second probe), a failure requeues the task with an agent * comment and marks the saddle offline (the normal unreachable * bookkeeping). The probe flag was taken by the caller; the * next push resets it. */ public async void probe_stale_saddle(Saddle saddle, TaskService tasks) throws Error { try { var status = yield request_saddle_json(saddle.host, "GET", "/status", null); yield record_status_probe(saddle, status); print(@"[ClankerSaddle] Stale saddle '$(saddle.name)' answered its exceptional status probe; task $(saddle.current_task_id) keeps running\n"); } catch (Error e) { warning(@"[ClankerSaddle] Saddle '$(saddle.name)' sent neither chunks nor results for $(stale_after_seconds())s and its status probe failed ($(e.message)); marking offline"); yield mark_saddle_unreachable(saddle.id, tasks, @"no chunks or results for $(stale_after_seconds())s; status probe failed ($(e.message))"); } } /** * Performs an HTTP request against a saddle daemon and parses the * response body as JSON; transport failures propagate to the * caller (the dispatcher marks the saddle unreachable). Transport * blips — typically a pooled keep-alive connection closed by the * daemon at the same instant the request reused it — are retried * once before giving up. Unlike {@link request_saddle_json} the * HTTP status is returned rather than thrown, and array bodies * parse too. */ private async int request_saddle(string host, string method, string path, owned Json.Builder? payload, out Json.Node? root) throws Error { root = null; GLib.Error? failure = null; for (var attempt = 0; attempt < 2; attempt++) { try { root = null; int status = yield request_saddle_once(host, method, path, payload, out root); return status; } catch (Error e) { failure = e; } if (((!)failure).code != 13 && ((!)failure).code != 14) { break; } GLib.Timeout.add(250, request_saddle.callback); yield; } throw (!)failure; } private async int request_saddle_once(string host, string method, string path, Json.Builder? payload, out Json.Node? root) throws Error { root = null; string base_url = host.has_prefix("http") ? host : @"http://$(host)"; var message = new Soup.Message(method, @"$(base_url)$(path)"); if (payload != null) { var generator = new Json.Generator(); generator.set_root(((!)payload).get_root()); message.set_request_body_from_bytes("application/json", new Bytes(generator.to_data(null).data)); } InputStream input; try { input = yield http.send_async(message, Priority.DEFAULT, null); } catch (Error e) { throw new Error(Quark.from_string("clanker-saddle"), 13, @"Saddle request to $(base_url)$(path) failed: $(e.message)"); } var output = new MemoryOutputStream.resizable(); try { yield output.splice_async(input, OutputStreamSpliceFlags.CLOSE_SOURCE, Priority.DEFAULT, null); yield output.close_async(Priority.DEFAULT, null); } catch (Error e) { throw new Error(Quark.from_string("clanker-saddle"), 14, @"Failed reading saddle response: $(e.message)"); } var written = output.get_data_size(); uint8[] data = output.steal_data(); // steal_data's array length is -1 and the buffer is sized to // the data — there is no NUL terminator to rely on — so slice // by the written byte count before reading it as a string. string response_text = written > 0 && data != null && data[0] != '\0' ? ((string) data).substring(0, (long) written) : "{}"; var parser = new Json.Parser(); try { parser.load_from_data(response_text); } catch (Error e) { throw new Error(Quark.from_string("clanker-saddle"), 16, @"Saddle returned invalid JSON: $(e.message)"); } root = parser.get_root(); return (int) message.status_code; } /** * Performs an HTTP request against a saddle daemon and returns the * parsed JSON object body (an empty object for non-object bodies); * HTTP statuses of 400 and above throw. */ private async Json.Object request_saddle_json(string host, string method, string path, owned Json.Builder? payload) throws Error { Json.Node? root = null; int status_code = yield request_saddle(host, method, path, (owned) payload, out root); if (status_code == 409) { // Machine-readable per the wire contract: the body (state, // running_task_id) rides in the message so the dispatch // path can distinguish zombie conflicts from transport // failures and recover. string body_text = "{}"; if (root != null && root.get_node_type() == Json.NodeType.OBJECT) { var generator = new Json.Generator(); generator.set_root(root); body_text = generator.to_data(null); } throw new Error(Quark.from_string("clanker-saddle"), 409, @"busy: $(body_text)"); } if (status_code >= 400) { throw new Error(Quark.from_string("clanker-saddle"), 15, @"Saddle request returned HTTP $(status_code)"); } if (root == null || root.get_node_type() != Json.NodeType.OBJECT) { return new Json.Object(); } return root.get_object(); } /** * A `{"name", "git_url", "path", "branch"}` project reference in * the saddle daemon's wire shape: `git_url` is the clone source, * and the checkout lives at * `SADDLE_PROJECTS_ROOT/task-/` inside the * saddle's container — one workspace per task, so concurrent * tasks never share a checkout. */ private static Json.Node project_node(Project project, int64 task_id, bool is_reference = false) { var root_var = (Environment.get_variable("SADDLE_PROJECTS_ROOT") ?? "").strip(); string projects_root = root_var.length > 0 ? root_var : "/workspace"; var builder = new Json.Builder(); builder.begin_object(); builder.set_member_name("name"); builder.add_string_value(project.name); builder.set_member_name("git_url"); builder.add_string_value(project.git_url); builder.set_member_name("path"); builder.add_string_value(is_reference ? @"$(projects_root)/task-$(task_id)/reference/$(project.name)" : @"$(projects_root)/task-$(task_id)/$(project.name)"); builder.set_member_name("branch"); builder.add_string_value(project.default_branch); builder.end_object(); return builder.get_root(); } /** `command` minus a leading `verb` and its separating whitespace. */ private static string strip_verb(string command, string verb) { var trimmed = command.strip(); if (trimmed == verb) { return ""; } return trimmed.has_prefix(verb + " ") ? trimmed.substring(verb.length + 1).strip() : trimmed; } /** * The answer as the saddle daemon expects it over /answer (or in a * resume payload): verbatim. The daemon applies its own * BlockRequest formatting, so this only strips the "Human * selected: …"/"Human answered (other): …" prefixes still stored * by answers recorded before the daemon took over formatting. */ private static string raw_answer_for_saddle(string answer) { if (answer.has_prefix("Human selected: ")) { return answer.substring("Human selected: ".length); } if (answer.has_prefix("Human answered (other): ")) { return answer.substring("Human answered (other): ".length); } return answer; } /** Strips one layer of matching double quotes, when present. */ private static string strip_quotes(string text) { if (text.length >= 2 && text.has_prefix("\"") && text.has_suffix("\"")) { return text.substring(1, text.length - 2); } return text; } private static string string_member(Json.Object obj, string name) { return obj.has_member(name) && obj.get_member(name).get_string() != null ? obj.get_string_member(name) : ""; } /** A nullable string member: null when absent or empty. */ private static string? optional_string_member(Json.Object obj, string name) { string value = string_member(obj, name); return value.length > 0 ? value : null; } /** * A string-array member: native JSON arrays are read element-wise; * a JSON-encoded string is parsed; anything else is empty. */ private static string[] string_array_member(Json.Object obj, string name) { if (!obj.has_member(name)) { return new string[0]; } var member = obj.get_member(name); if (member.get_node_type() == Json.NodeType.ARRAY) { var values = new Vector(); foreach (var element in member.get_array().get_elements()) { if (element.get_node_type() == Json.NodeType.VALUE && element.get_string() != null) { values.add(element.get_string()); } } return values.to_array(); } string raw = string_member(obj, name); if (raw.length == 0) { return new string[0]; } try { var parser = new Json.Parser(); parser.load_from_data(raw); var root = parser.get_root(); if (root != null && root.get_node_type() == Json.NodeType.ARRAY) { var values = new Vector(); foreach (var element in root.get_array().get_elements()) { if (element.get_node_type() == Json.NodeType.VALUE && element.get_string() != null) { values.add(element.get_string()); } } return values.to_array(); } } catch (Error e) { warning("[ClankerSaddle] Ignoring malformed options array: %s", e.message); } return new string[0]; } private static DateTime? parse_iso(Json.Object obj, string name) { if (!obj.has_member(name)) { return null; } var value = obj.get_member(name); if (value.get_node_type() != Json.NodeType.VALUE || value.get_string() == null) { return null; } return new DateTime.from_iso8601(value.get_string(), new TimeZone.utc()); } } }