|
|
@@ -0,0 +1,734 @@
|
|
|
+using Invercargill;
|
|
|
+using Invercargill.DataStructures;
|
|
|
+using Invercargill.Expressions;
|
|
|
+using InvercargillSql.Orm;
|
|
|
+using Inversion;
|
|
|
+using Soup;
|
|
|
+using ClankerSaddle.Data;
|
|
|
+
|
|
|
+namespace ClankerSaddle.Services {
|
|
|
+
|
|
|
+ /** 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 = ""; }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 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: registers containers,
|
|
|
+ * finds idle capacity for the dispatcher, pushes task assignments to a
|
|
|
+ * daemon's `/start` endpoint, and polls `/status` + `/output` to keep
|
|
|
+ * the local audit trail in sync. Block answers are handed out exactly
|
|
|
+ * once per saddle poll (delivered ids are remembered in memory; a
|
|
|
+ * restart re-delivers and the daemon deduplicates).
|
|
|
+ */
|
|
|
+ public class SaddleService : Object {
|
|
|
+
|
|
|
+ private OrmSession db = inject<OrmSession>();
|
|
|
+ private Soup.Session http = new Soup.Session();
|
|
|
+ private SaddlePollState poll_state = inject<SaddlePollState>();
|
|
|
+
|
|
|
+ /** Registers a saddle container; `host` is host:port of its control server. */
|
|
|
+ public async Saddle register_saddle(string name, string host) throws Error {
|
|
|
+ var saddle = new Saddle() {
|
|
|
+ name = name,
|
|
|
+ host = host,
|
|
|
+ status = "idle"
|
|
|
+ };
|
|
|
+ yield db.insert_async<Saddle>(saddle);
|
|
|
+ return saddle;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Removes a saddle registration; tasks it was running are re-queued. */
|
|
|
+ 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<TaskEntity>()
|
|
|
+ .where(expr("id == $0", elem<int64?>(((!)saddle).current_task_id)))
|
|
|
+ .first_async();
|
|
|
+ if (task != null && (((!)task).status == STATUS_ASSIGNED
|
|
|
+ || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED)) {
|
|
|
+ ((!)task).status = STATUS_QUEUED;
|
|
|
+ yield db.update_async<TaskEntity>((!)task);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ yield db.delete_async<Saddle>((!)saddle);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Loads the saddle identified by `id`, or null. */
|
|
|
+ public async Saddle? get_saddle(int64 id) throws Error {
|
|
|
+ return yield db.query<Saddle>()
|
|
|
+ .where(expr("id == $0", elem<int64?>(id)))
|
|
|
+ .first_async();
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Lists every saddle ordered by name. */
|
|
|
+ public async ImmutableLot<Saddle> list_saddles() throws Error {
|
|
|
+ return yield db.query<Saddle>()
|
|
|
+ .order_by(expr("name"))
|
|
|
+ .materialise_async();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Lists saddles with status 'busy' (the dispatcher's poll
|
|
|
+ * candidates).
|
|
|
+ */
|
|
|
+ public async ImmutableLot<Saddle> list_busy_saddles() throws Error {
|
|
|
+ return yield db.query<Saddle>()
|
|
|
+ .where(expr("status == $0", elem("busy")))
|
|
|
+ .materialise_async();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Updates a saddle's status and current task. A null
|
|
|
+ * `current_task_id` clears the assignment (and `allocated_at`).
|
|
|
+ */
|
|
|
+ public async void update_saddle_status(int64 id, string status, int64? current_task_id) throws Error {
|
|
|
+ var saddle = yield get_saddle(id);
|
|
|
+ if (saddle == null) {
|
|
|
+ throw new Error(Quark.from_string("clanker-saddle"), 10,
|
|
|
+ @"No saddle exists with id $(id)");
|
|
|
+ }
|
|
|
+ ((!)saddle).status = status;
|
|
|
+ ((!)saddle).current_task_id = current_task_id != null ? ((!)current_task_id) : 0;
|
|
|
+ ((!)saddle).allocated_at = current_task_id != null ? new DateTime.now_local() : null;
|
|
|
+ yield db.update_async<Saddle>((!)saddle);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** The first saddle with status 'idle', or null when none is free. */
|
|
|
+ public async Saddle? get_idle_saddle() throws Error {
|
|
|
+ return yield db.query<Saddle>()
|
|
|
+ .where(expr("status == $0", elem("idle")))
|
|
|
+ .order_by(expr("id"))
|
|
|
+ .first_async();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Marks a saddle offline when its control server cannot be
|
|
|
+ * reached, releasing any task it was holding back to 'queued' and
|
|
|
+ * closing that task's open session (outcome 'stopped').
|
|
|
+ */
|
|
|
+ public async void mark_saddle_unreachable(int64 id) throws Error {
|
|
|
+ var saddle = yield get_saddle(id);
|
|
|
+ if (saddle == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (((!)saddle).current_task_id > 0) {
|
|
|
+ int64 task_id = ((!)saddle).current_task_id;
|
|
|
+ var task = yield db.query<TaskEntity>()
|
|
|
+ .where(expr("id == $0", elem<int64?>(task_id)))
|
|
|
+ .first_async();
|
|
|
+ if (task != null && (((!)task).status == STATUS_ASSIGNED
|
|
|
+ || ((!)task).status == STATUS_WORKING || ((!)task).status == STATUS_BLOCKED)) {
|
|
|
+ ((!)task).status = STATUS_QUEUED;
|
|
|
+ yield db.update_async<TaskEntity>((!)task);
|
|
|
+ }
|
|
|
+
|
|
|
+ var sessions = yield db.query<ClankerSaddle.Data.Session>()
|
|
|
+ .where(expr("task_id == $0", elem<int64?>(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<ClankerSaddle.Data.Session>(session);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ ((!)saddle).status = "offline";
|
|
|
+ ((!)saddle).current_task_id = 0;
|
|
|
+ ((!)saddle).allocated_at = null;
|
|
|
+ yield db.update_async<Saddle>((!)saddle);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Pushes a task assignment to the saddle daemon's `POST /start`
|
|
|
+ * endpoint in its wire shape — `{task, llm_config}` — with the
|
|
|
+ * project, co-developed/reference links and recent comments inside
|
|
|
+ * the task object. Project checkouts are addressed by the
|
|
|
+ * `SADDLE_PROJECTS_ROOT` convention (`root/name`), the LLM endpoint
|
|
|
+ * by the {@link LlmEndpoint} environment. Returns the parsed JSON
|
|
|
+ * response body.
|
|
|
+ */
|
|
|
+ public async Json.Object push_task_to_saddle(int64 saddle_id, TaskEntity task, int64 session_id,
|
|
|
+ ProjectService projects) throws Error {
|
|
|
+ var saddle = yield get_saddle(saddle_id);
|
|
|
+ if (saddle == null) {
|
|
|
+ throw new Error(Quark.from_string("clanker-saddle"), 11,
|
|
|
+ @"No saddle exists with id $(saddle_id)");
|
|
|
+ }
|
|
|
+ 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<Project>()
|
|
|
+ .where(expr("id == $0", elem<int64?>(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)");
|
|
|
+ }
|
|
|
+
|
|
|
+ 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("is_design_task");
|
|
|
+ payload.add_boolean_value(task.is_design_task);
|
|
|
+ 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);
|
|
|
+
|
|
|
+ var co_developed = new Vector<Project>();
|
|
|
+ var references = new Vector<Project>();
|
|
|
+ foreach (var link in yield projects.list_links(task.project_id)) {
|
|
|
+ var linked = yield db.query<Project>()
|
|
|
+ .where(expr("id == $0", elem<int64?>(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);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ payload.set_member_name("co_developed");
|
|
|
+ payload.begin_array();
|
|
|
+ payload.add_value(project_node(((!)project)));
|
|
|
+ foreach (var linked in co_developed) {
|
|
|
+ payload.add_value(project_node(linked));
|
|
|
+ }
|
|
|
+ payload.end_array();
|
|
|
+ payload.set_member_name("reference");
|
|
|
+ payload.begin_array();
|
|
|
+ foreach (var linked in references) {
|
|
|
+ payload.add_value(project_node(linked));
|
|
|
+ }
|
|
|
+ payload.end_array();
|
|
|
+
|
|
|
+ payload.set_member_name("comments");
|
|
|
+ payload.begin_array();
|
|
|
+ var comments = yield db.query<TaskComment>()
|
|
|
+ .where(expr("task_id == $0", elem<int64?>(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);
|
|
|
+ payload.end_object();
|
|
|
+ }
|
|
|
+ payload.end_array();
|
|
|
+ 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();
|
|
|
+ payload.end_object();
|
|
|
+
|
|
|
+ return yield request_saddle_json(((!)saddle).host, "POST", "/start",
|
|
|
+ (owned) payload);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Seeds the output cursor of a not-yet-pushed session from the
|
|
|
+ * daemon's current `latest_entry_id`. The daemon's entry log is
|
|
|
+ * global across its sessions, so this baseline — read before the
|
|
|
+ * `POST /start` push, when every logged entry belongs to earlier
|
|
|
+ * sessions — keeps prior sessions' entries out of the new
|
|
|
+ * session's audit trail and its domain events.
|
|
|
+ */
|
|
|
+ public async void seed_session_cursor(int64 saddle_id, int64 session_id) throws Error {
|
|
|
+ var saddle = yield get_saddle(saddle_id);
|
|
|
+ if (saddle == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null);
|
|
|
+ int64 baseline = status.has_member("latest_entry_id")
|
|
|
+ && status.get_member("latest_entry_id").get_value_type() == typeof(int64)
|
|
|
+ ? status.get_int_member("latest_entry_id") : 0;
|
|
|
+ poll_state.store_since(session_id, baseline);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Polls a busy saddle: stores new entries reported by the daemon's
|
|
|
+ * `GET /output` as `session_commands` rows of the open session and
|
|
|
+ * projects the agent's `saddle comment/subtask/done` events into
|
|
|
+ * their domain tables, then applies status changes from
|
|
|
+ * `GET /status` — done/failed complete or fail the task, blocked
|
|
|
+ * raises the daemon's question as an answerable block and delivers
|
|
|
+ * any human answers to the daemon's `POST /answer`. Returns true
|
|
|
+ * when the saddle reported the task finished; the status pass is
|
|
|
+ * skipped once the entries already finished the task.
|
|
|
+ */
|
|
|
+ public async bool poll_saddle_updates(int64 saddle_id, TaskService tasks) throws Error {
|
|
|
+ var saddle = yield get_saddle(saddle_id);
|
|
|
+ if (saddle == null || ((!)saddle).current_task_id == 0) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ int64 task_id = ((!)saddle).current_task_id;
|
|
|
+
|
|
|
+ var session = yield tasks.open_session_for_task(task_id);
|
|
|
+ if (session != null) {
|
|
|
+ var fresh = yield store_new_output(saddle_id, ((!)session).id);
|
|
|
+ if (yield apply_entry_events(task_id, fresh, tasks)) {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null);
|
|
|
+ string state = status.has_member("state") ? status.get_string_member("state") : "";
|
|
|
+ switch (state) {
|
|
|
+ case "done":
|
|
|
+ yield tasks.complete_task(task_id,
|
|
|
+ status.has_member("summary") ? status.get_string_member("summary") : null,
|
|
|
+ null, "done");
|
|
|
+ return true;
|
|
|
+ case "failed":
|
|
|
+ yield tasks.fail_task(task_id, null);
|
|
|
+ return true;
|
|
|
+ case "blocked":
|
|
|
+ string question = string_member(status, "block_question");
|
|
|
+ if (question.length > 0) {
|
|
|
+ yield ensure_block(tasks, task_id, question);
|
|
|
+ yield deliver_answered_blocks((!)saddle, task_id, question);
|
|
|
+ }
|
|
|
+ yield tasks.update_task(task_id, null, null, null, STATUS_BLOCKED);
|
|
|
+ break;
|
|
|
+ case "working":
|
|
|
+ var task = yield tasks.update_task(task_id, null, null, null, STATUS_WORKING);
|
|
|
+ if (task != null && ((!)task).started_at == null) {
|
|
|
+ ((!)task).started_at = new DateTime.now_local();
|
|
|
+ yield db.update_async<TaskEntity>((!)task);
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ case "idle":
|
|
|
+ // The daemon stopped without reporting done: release the
|
|
|
+ // saddle and re-queue the task for another attempt.
|
|
|
+ yield tasks.requeue_task(task_id);
|
|
|
+ break;
|
|
|
+ default:
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 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 current task assignment is
|
|
|
+ * left untouched.
|
|
|
+ */
|
|
|
+ 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)");
|
|
|
+ }
|
|
|
+ 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", "") : "";
|
|
|
+ ((!)saddle).status = state == "idle" ? "idle" : "busy";
|
|
|
+ } catch (Error e) {
|
|
|
+ ((!)saddle).status = "offline";
|
|
|
+ }
|
|
|
+ yield db.update_async<Saddle>((!)saddle);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Answered blocks of the saddle's current task not yet handed over
|
|
|
+ * the poll endpoint; marks them delivered.
|
|
|
+ */
|
|
|
+ public async ImmutableLot<PendingAnswer> 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<PendingAnswer>().to_immutable_buffer();
|
|
|
+ }
|
|
|
+
|
|
|
+ var blocks = yield db.query<TaskBlock>()
|
|
|
+ .where(expr("task_id == $0", elem<int64?>(((!)saddle).current_task_id)))
|
|
|
+ .materialise_async();
|
|
|
+ var rows = new Vector<PendingAnswer>();
|
|
|
+ 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 = ((!)block).answer ?? ""
|
|
|
+ });
|
|
|
+ }
|
|
|
+ return rows.to_immutable_buffer();
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Stores entries newly reported by the daemon's `GET /output`
|
|
|
+ * (a bare JSON array of `{id, command, output, exit_code, type,
|
|
|
+ * timestamp}` objects) as `session_commands` rows, advancing the
|
|
|
+ * session's cursor in {@link SaddlePollState}. Returns the entries
|
|
|
+ * stored, oldest first, for {@link apply_entry_events}.
|
|
|
+ */
|
|
|
+ private async Vector<Json.Object> store_new_output(int64 saddle_id, int64 session_id) throws Error {
|
|
|
+ var stored = new Vector<Json.Object>();
|
|
|
+ var saddle = yield get_saddle(saddle_id);
|
|
|
+ if (saddle == null) {
|
|
|
+ return stored;
|
|
|
+ }
|
|
|
+
|
|
|
+ int64 since = poll_state.since_for(session_id);
|
|
|
+ Json.Node? root = null;
|
|
|
+ int status_code = yield request_saddle(((!)saddle).host, "GET",
|
|
|
+ @"/output?since=$(since)", null, out root);
|
|
|
+ if (status_code >= 400 || root == null || root.get_node_type() != Json.NodeType.ARRAY) {
|
|
|
+ throw new Error(Quark.from_string("clanker-saddle"), 15,
|
|
|
+ @"Saddle output request returned HTTP $(status_code)");
|
|
|
+ }
|
|
|
+
|
|
|
+ foreach (var element in root.get_array().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_int_member("id") : 0;
|
|
|
+ if (entry_id <= since) {
|
|
|
+ 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,
|
|
|
+ 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<SessionCommand>(command);
|
|
|
+ stored.add(entry);
|
|
|
+ since = entry_id;
|
|
|
+ }
|
|
|
+ poll_state.store_since(session_id, since);
|
|
|
+ return stored;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Projects the agent's conversation-level events onto the task:
|
|
|
+ * `saddle comment` lines become bot comments, `saddle subtask`
|
|
|
+ * lines create child tasks, and the `saddle done` entry completes
|
|
|
+ * the task with the summary the agent gave. Returns true once a
|
|
|
+ * done event finished the task.
|
|
|
+ */
|
|
|
+ private async bool apply_entry_events(int64 task_id, Vector<Json.Object> entries,
|
|
|
+ TaskService tasks) throws Error {
|
|
|
+ foreach (var entry in entries) {
|
|
|
+ string command = string_member(entry, "command");
|
|
|
+ switch (string_member(entry, "type")) {
|
|
|
+ case "comment":
|
|
|
+ string body = strip_quotes(strip_verb(command, "saddle comment"));
|
|
|
+ if (body.length > 0) {
|
|
|
+ yield tasks.add_comment(task_id, "bot", body);
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ case "subtask":
|
|
|
+ string[] operands = split_operands(strip_verb(command, "saddle subtask"));
|
|
|
+ if (operands.length >= 1 && operands[0].length > 0) {
|
|
|
+ yield tasks.create_subtask(task_id, operands[0],
|
|
|
+ operands.length >= 2 ? operands[1] : null);
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ case "done":
|
|
|
+ yield tasks.complete_task(task_id,
|
|
|
+ strip_quotes(strip_verb(command, "saddle done")), null, "done");
|
|
|
+ return true;
|
|
|
+ default:
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Raises the daemon's current blocking question 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
|
|
|
+ * question is already represented.
|
|
|
+ */
|
|
|
+ private async void ensure_block(TaskService tasks, int64 task_id, string question) throws Error {
|
|
|
+ foreach (var block in yield tasks.list_blocks(task_id)) {
|
|
|
+ if (block.question != question) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (block.answer == null || !poll_state.is_delivered(block.id)) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ yield tasks.add_block(task_id, question);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * Delivers human answers of the daemon's current question to its
|
|
|
+ * `POST /answer` endpoint. Delivery is marked in
|
|
|
+ * {@link SaddlePollState} once the daemon accepts it (200) or has
|
|
|
+ * nobody waiting (409 — the session moved on), so each answer is
|
|
|
+ * pushed exactly once.
|
|
|
+ */
|
|
|
+ private async void deliver_answered_blocks(Saddle saddle, int64 task_id, string question) throws Error {
|
|
|
+ foreach (var block in yield db.query<TaskBlock>()
|
|
|
+ .where(expr("task_id == $0", elem<int64?>(task_id)))
|
|
|
+ .order_by(expr("id"))
|
|
|
+ .materialise_async()) {
|
|
|
+ if (block.answer == null || block.question != question
|
|
|
+ || poll_state.is_delivered(block.id)) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ var payload = new Json.Builder();
|
|
|
+ payload.begin_object();
|
|
|
+ payload.set_member_name("answer");
|
|
|
+ payload.add_string_value(((!)block).answer ?? "");
|
|
|
+ payload.end_object();
|
|
|
+ Json.Node? root = null;
|
|
|
+ int status_code = yield request_saddle(saddle.host, "POST", "/answer",
|
|
|
+ (owned) payload, out root);
|
|
|
+ if (status_code != 200 && status_code != 409) {
|
|
|
+ throw new Error(Quark.from_string("clanker-saddle"), 15,
|
|
|
+ @"Saddle answer delivery returned HTTP $(status_code)");
|
|
|
+ }
|
|
|
+ poll_state.mark_delivered(block.id);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 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)");
|
|
|
+ }
|
|
|
+
|
|
|
+ uint8[] data = output.steal_data();
|
|
|
+ // steal_data's array length is -1; the buffer itself is
|
|
|
+ // NUL-terminated, so read it as a string directly.
|
|
|
+ string response_text = data != null && data[0] != '\0' ? (string) data : "{}";
|
|
|
+
|
|
|
+ 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 >= 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", "path", "branch"}` project reference in the saddle
|
|
|
+ * daemon's wire shape: the checkout lives at
|
|
|
+ * `SADDLE_PROJECTS_ROOT/name` inside the saddle's container.
|
|
|
+ */
|
|
|
+ private static Json.Node project_node(Project project) {
|
|
|
+ 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("path");
|
|
|
+ builder.add_string_value(@"$(projects_root)/$(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;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** Splits operands on whitespace honouring double quotes. */
|
|
|
+ private static string[] split_operands(string args) {
|
|
|
+ var operands = new Vector<string>();
|
|
|
+ var current = new StringBuilder();
|
|
|
+ bool in_operand = false;
|
|
|
+ bool quoted = false;
|
|
|
+ for (var index = 0; index < args.length; index++) {
|
|
|
+ char character = args[index];
|
|
|
+ if (character == '"') {
|
|
|
+ quoted = true;
|
|
|
+ in_operand = true;
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ if (!quoted && character.isspace()) {
|
|
|
+ if (in_operand) {
|
|
|
+ operands.add(current.str);
|
|
|
+ current = new StringBuilder();
|
|
|
+ in_operand = false;
|
|
|
+ quoted = false;
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ current.append_c(character);
|
|
|
+ in_operand = true;
|
|
|
+ }
|
|
|
+ if (in_operand) {
|
|
|
+ operands.add(current.str);
|
|
|
+ }
|
|
|
+ return operands.to_array();
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 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) : "";
|
|
|
+ }
|
|
|
+
|
|
|
+ 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());
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|