|
@@ -61,6 +61,10 @@ namespace ClankerSaddle.Services {
|
|
|
/** Set when the current task is a goal agent's run: the card links the goal. */
|
|
/** Set when the current task is a goal agent's run: the card links the goal. */
|
|
|
public int64 goal_id { get; set; }
|
|
public int64 goal_id { get; set; }
|
|
|
public string goal_title { get; set; default = ""; }
|
|
public string goal_title { get; set; default = ""; }
|
|
|
|
|
+ /** The running session the derivation keyed on (plan 16 item 23). */
|
|
|
|
|
+ public int64 session_id { get; set; }
|
|
|
|
|
+ /** When that session started — the "allocated" display stamp. */
|
|
|
|
|
+ public DateTime? started_at { get; set; default = null; }
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -90,13 +94,16 @@ namespace ClankerSaddle.Services {
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* Saddle registry and control-plane client (plan 10 phase 2): registers
|
|
* 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
|
|
|
|
|
|
|
+ * containers, derives idle capacity for the dispatcher's wake
|
|
|
|
|
+ * computation (plan 16 item 23: busy iff an open session on an
|
|
|
|
|
+ * assigned/working task exists — {@link current_work_view} is the ONE
|
|
|
|
|
+ * truth accessor; the saddle row keeps only the OFFLINE probe state),
|
|
|
|
|
+ * 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
|
|
* handed out exactly once (delivered ids are remembered in memory; a
|
|
|
* restart re-delivers and the daemon deduplicates).
|
|
* restart re-delivers and the daemon deduplicates).
|
|
|
*/
|
|
*/
|
|
@@ -233,20 +240,22 @@ namespace ClankerSaddle.Services {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * Removes a saddle registration; the task it was running
|
|
|
|
|
- * releases through the ONE path — back to the queue, its open
|
|
|
|
|
- * sessions closed stopped (no orphan open session outlives the
|
|
|
|
|
- * row). Revocation: the row dies here, and its per-saddle
|
|
|
|
|
- * token with it — nothing authorises `/api/saddle/*` for it
|
|
|
|
|
- * afterwards.
|
|
|
|
|
|
|
+ * Removes a saddle registration; the task it was running (the
|
|
|
|
|
+ * DERIVED running work — plan 16 item 23: sessions are the
|
|
|
|
|
+ * truth) releases through the ONE path — back to the queue,
|
|
|
|
|
+ * its open sessions closed stopped (no orphan open session
|
|
|
|
|
+ * outlives the row). Revocation: the row dies here, and its
|
|
|
|
|
+ * per-saddle token with it — nothing authorises
|
|
|
|
|
+ * `/api/saddle/*` for it afterwards.
|
|
|
*/
|
|
*/
|
|
|
public async void unregister_saddle(int64 id) throws Error {
|
|
public async void unregister_saddle(int64 id) throws Error {
|
|
|
var saddle = yield get_saddle(id);
|
|
var saddle = yield get_saddle(id);
|
|
|
if (saddle == null) {
|
|
if (saddle == null) {
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
|
- if (((!)saddle).current_task_id > 0) {
|
|
|
|
|
- yield task_service.release_task(((!)saddle).current_task_id, OUTCOME_STOPPED, true);
|
|
|
|
|
|
|
+ var work = yield current_work_view((!)saddle);
|
|
|
|
|
+ if (work != null) {
|
|
|
|
|
+ yield task_service.release_task(((!)work).task_id, OUTCOME_STOPPED, true);
|
|
|
}
|
|
}
|
|
|
yield db.delete_async<Saddle>((!)saddle);
|
|
yield db.delete_async<Saddle>((!)saddle);
|
|
|
yield pulse.bump(0, "saddle removed");
|
|
yield pulse.bump(0, "saddle removed");
|
|
@@ -267,20 +276,30 @@ namespace ClankerSaddle.Services {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * The saddle's current work resolved for display — the ONE
|
|
|
|
|
- * lookup the dashboard's saddle cards and the saddles page's
|
|
|
|
|
- * rows share: the current task's title, and the goal it
|
|
|
|
|
- * stands for when it is a goal agent's internal run (runs are
|
|
|
|
|
- * internal — never a task destination at a glance). Null when
|
|
|
|
|
- * the saddle holds no task.
|
|
|
|
|
|
|
+ * The saddle's current work resolved from the SESSION records —
|
|
|
|
|
+ * the ONE truth accessor every surface reads (plan 16 item 23:
|
|
|
|
|
+ * sessions own "who runs what"; the saddle row's cache columns
|
|
|
|
|
+ * are gone). The running session is the newest open session on
|
|
|
|
|
+ * the saddle whose task is still assigned/working — busy iff it
|
|
|
|
|
+ * exists, idle otherwise; a parked task's session closed
|
|
|
|
|
+ * 'parked' at its park (D4), so parks read as free exactly
|
|
|
|
|
+ * like the daemon's file-log brackets say. The task join also
|
|
|
|
|
+ * makes the derivation drift-tolerant: a stale open row on a
|
|
|
|
|
+ * task that has since moved on cannot wedge a saddle busy.
|
|
|
|
|
+ * Null when the saddle holds no running work.
|
|
|
*/
|
|
*/
|
|
|
public async CurrentWork? current_work_view(Saddle saddle) throws Error {
|
|
public async CurrentWork? current_work_view(Saddle saddle) throws Error {
|
|
|
- if (saddle.current_task_id <= 0) {
|
|
|
|
|
|
|
+ var session = yield running_session_on_saddle(saddle.id);
|
|
|
|
|
+ if (session == null) {
|
|
|
return null;
|
|
return null;
|
|
|
}
|
|
}
|
|
|
- var work = new CurrentWork() { task_id = saddle.current_task_id };
|
|
|
|
|
|
|
+ var work = new CurrentWork() {
|
|
|
|
|
+ task_id = ((!)session).task_id,
|
|
|
|
|
+ session_id = ((!)session).id,
|
|
|
|
|
+ started_at = ((!)session).started_at
|
|
|
|
|
+ };
|
|
|
var task = yield db.query<TaskEntity>()
|
|
var task = yield db.query<TaskEntity>()
|
|
|
- .where(expr("id == $0", elem<int64?>(saddle.current_task_id)))
|
|
|
|
|
|
|
+ .where(expr("id == $0", elem<int64?>(((!)session).task_id)))
|
|
|
.first_async();
|
|
.first_async();
|
|
|
if (task == null) {
|
|
if (task == null) {
|
|
|
return work;
|
|
return work;
|
|
@@ -299,13 +318,102 @@ namespace ClankerSaddle.Services {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * Lists saddles with status 'busy' (the dispatcher's poll
|
|
|
|
|
- * candidates).
|
|
|
|
|
|
|
+ * The derived busy/idle/offline status label every saddle view
|
|
|
|
|
+ * shares — 'busy' iff {@link current_work_view} holds running
|
|
|
|
|
+ * work, else the row's stored OFFLINE probe state ('offline' or
|
|
|
|
|
+ * 'idle').
|
|
|
|
|
+ */
|
|
|
|
|
+ public static string derived_status(Saddle saddle, CurrentWork? work) {
|
|
|
|
|
+ if (work != null) {
|
|
|
|
|
+ return SADDLE_BUSY;
|
|
|
|
|
+ }
|
|
|
|
|
+ return saddle.status == SADDLE_OFFLINE ? SADDLE_OFFLINE : SADDLE_IDLE;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * The saddle's RUNNING session — the newest open session whose
|
|
|
|
|
+ * task is assigned/working (the null check runs in SQL: the
|
|
|
|
|
+ * expression layer cannot compare columns against null). At
|
|
|
|
|
+ * most one exists per saddle by construction: every park closes
|
|
|
|
|
+ * its row (D4) and the claim's guard refuses a saddle that
|
|
|
|
|
+ * already holds one.
|
|
|
|
|
+ */
|
|
|
|
|
+ private async ClankerSaddle.Data.Session? running_session_on_saddle(int64 saddle_id)
|
|
|
|
|
+ throws Error {
|
|
|
|
|
+ var command = connection.create_command(
|
|
|
|
|
+ "SELECT s.id AS session_id, s.task_id AS task_id, s.started_at AS started"
|
|
|
|
|
+ + " FROM sessions s JOIN tasks t ON t.id = s.task_id"
|
|
|
|
|
+ + " WHERE s.saddle_id = :saddle AND s.ended_at IS NULL"
|
|
|
|
|
+ + " AND t.status IN ('assigned', 'working')"
|
|
|
|
|
+ + " ORDER BY s.id DESC LIMIT 1");
|
|
|
|
|
+ command.with_parameter<Invercargill.Element>("saddle",
|
|
|
|
|
+ elem<int64?>(saddle_id));
|
|
|
|
|
+ ClankerSaddle.Data.Session? found = null;
|
|
|
|
|
+ foreach (var row in yield command.execute_query_async()) {
|
|
|
|
|
+ int64? session_id = 0;
|
|
|
|
|
+ int64? task_id = 0;
|
|
|
|
|
+ ((!)row.get("session_id")).try_get_as<int64?>(out session_id);
|
|
|
|
|
+ ((!)row.get("task_id")).try_get_as<int64?>(out task_id);
|
|
|
|
|
+ found = new ClankerSaddle.Data.Session() {
|
|
|
|
|
+ id = session_id ?? 0,
|
|
|
|
|
+ task_id = task_id ?? 0,
|
|
|
|
|
+ saddle_id = saddle_id,
|
|
|
|
|
+ started_at = raw_date_time_of(row.get("started"))
|
|
|
|
|
+ };
|
|
|
|
|
+ }
|
|
|
|
|
+ return found;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /** A raw query column back as a DateTime (null when null/unreadable). */
|
|
|
|
|
+ private static DateTime? raw_date_time_of(Invercargill.Element? element) {
|
|
|
|
|
+ if (element == null) {
|
|
|
|
|
+ return null;
|
|
|
|
|
+ }
|
|
|
|
|
+ DateTime value;
|
|
|
|
|
+ return ((!)element).try_get_as<DateTime>(out value) ? value : null;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * The ids (decimal strings — the OR-chain batch convention) of
|
|
|
|
|
+ * every saddle holding a running session — the derived busy
|
|
|
|
|
+ * set the fleet lists and the offline stamp share.
|
|
|
|
|
+ */
|
|
|
|
|
+ private async Invercargill.DataStructures.HashSet<string> busy_saddle_ids()
|
|
|
|
|
+ throws Error {
|
|
|
|
|
+ var busy = new Invercargill.DataStructures.HashSet<string>();
|
|
|
|
|
+ var command = connection.create_command(
|
|
|
|
|
+ "SELECT DISTINCT s.saddle_id AS saddle_id FROM sessions s"
|
|
|
|
|
+ + " JOIN tasks t ON t.id = s.task_id"
|
|
|
|
|
+ + " WHERE s.ended_at IS NULL"
|
|
|
|
|
+ + " AND t.status IN ('assigned', 'working')");
|
|
|
|
|
+ foreach (var row in yield command.execute_query_async()) {
|
|
|
|
|
+ int64? saddle_id = 0;
|
|
|
|
|
+ ((!)row.get("saddle_id")).try_get_as<int64?>(out saddle_id);
|
|
|
|
|
+ if (saddle_id != null) {
|
|
|
|
|
+ busy.add(((!)saddle_id).to_string());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ return busy;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * Saddles holding a running session (the dispatcher's poll
|
|
|
|
|
+ * candidates) — derived from the session records, registration
|
|
|
|
|
+ * order.
|
|
|
*/
|
|
*/
|
|
|
public async ImmutableLot<Saddle> list_busy_saddles() throws Error {
|
|
public async ImmutableLot<Saddle> list_busy_saddles() throws Error {
|
|
|
- return yield db.query<Saddle>()
|
|
|
|
|
- .where(expr("status == $0", elem(SADDLE_BUSY)))
|
|
|
|
|
- .materialise_async();
|
|
|
|
|
|
|
+ var busy = yield busy_saddle_ids();
|
|
|
|
|
+ if (busy.count() == 0) {
|
|
|
|
|
+ return (new Vector<Saddle>()).to_immutable_buffer();
|
|
|
|
|
+ }
|
|
|
|
|
+ var rows = new Vector<Saddle>();
|
|
|
|
|
+ foreach (var saddle in yield db.query<Saddle>()
|
|
|
|
|
+ .where(TaskService.id_in("id", busy))
|
|
|
|
|
+ .order_by(expr("id"))
|
|
|
|
|
+ .materialise_async()) {
|
|
|
|
|
+ rows.add(saddle);
|
|
|
|
|
+ }
|
|
|
|
|
+ return rows.to_immutable_buffer();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -445,14 +553,15 @@ namespace ClankerSaddle.Services {
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* {@link abort_live_session} keyed by saddle instead of task —
|
|
* {@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).
|
|
|
|
|
|
|
+ * the unregister path. The cancel is delivered to the saddle's
|
|
|
|
|
+ * RUNNING session (the derived record; plan 16 item 23) — not
|
|
|
|
|
+ * re-resolved through a 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,
|
|
public async void abort_saddle_session(TaskService tasks, int64 saddle_id,
|
|
|
string verb) throws Error {
|
|
string verb) throws Error {
|
|
|
- var session = yield open_session_on_saddle(saddle_id);
|
|
|
|
|
|
|
+ var session = yield running_session_on_saddle(saddle_id);
|
|
|
if (session == null) {
|
|
if (session == null) {
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
@@ -499,11 +608,11 @@ namespace ClankerSaddle.Services {
|
|
|
* offline (a manual status refresh, a restart) — without this
|
|
* offline (a manual status refresh, a restart) — without this
|
|
|
* pass they would wait forever for a daemon that will not answer.
|
|
* pass they would wait forever for a daemon that will not answer.
|
|
|
* Only `assigned`/`working` tasks move: parked tasks
|
|
* Only `assigned`/`working` tasks move: parked tasks
|
|
|
- * (blocked/review/suspended) keep their sessions, which the
|
|
|
|
|
- * resume machinery needs. Each move is {@link TaskService.requeue_task}
|
|
|
|
|
- * — the release owner closes the session and frees the saddle —
|
|
|
|
|
- * plus a agent comment so the timeline shows why the task
|
|
|
|
|
- * restarted.
|
|
|
|
|
|
|
+ * (blocked/review/suspended, their sessions closed 'parked')
|
|
|
|
|
+ * keep their records, which the resume machinery needs. Each
|
|
|
|
|
+ * move is {@link TaskService.requeue_task} — the release owner
|
|
|
|
|
+ * closes the session (the close IS the free, item 23) — plus a
|
|
|
|
|
+ * system comment so the timeline shows why the task restarted.
|
|
|
*/
|
|
*/
|
|
|
public async void requeue_tasks_on_offline_saddles(TaskService tasks) throws Error {
|
|
public async void requeue_tasks_on_offline_saddles(TaskService tasks) throws Error {
|
|
|
foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
|
|
foreach (var status in new string[] { STATUS_ASSIGNED, STATUS_WORKING }) {
|
|
@@ -532,46 +641,39 @@ namespace ClankerSaddle.Services {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 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 = '" + SADDLE_IDLE
|
|
|
|
|
- + "', current_task_id = 0, allocated_at = NULL"
|
|
|
|
|
- + " WHERE id = :id AND current_task_id = :task");
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("id", elem<int64?>(saddle_id));
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("task", elem<int64?>(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.
|
|
|
|
|
|
|
+ * Every saddle free for work — derived (plan 16 item 23): not
|
|
|
|
|
+ * offline (the probe state) and holding no running session —
|
|
|
|
|
+ * registration order, the dispatcher's capability-matching
|
|
|
|
|
+ * candidate list. A parked task's closed session reads free
|
|
|
|
|
+ * exactly like the freed daemon it brackets.
|
|
|
*/
|
|
*/
|
|
|
public async ImmutableLot<Saddle> list_idle_saddles() throws Error {
|
|
public async ImmutableLot<Saddle> list_idle_saddles() throws Error {
|
|
|
- return yield db.query<Saddle>()
|
|
|
|
|
- .where(expr("status == $0", elem(SADDLE_IDLE)))
|
|
|
|
|
- .order_by(expr("id"))
|
|
|
|
|
- .materialise_async();
|
|
|
|
|
|
|
+ var busy = yield busy_saddle_ids();
|
|
|
|
|
+ var rows = new Vector<Saddle>();
|
|
|
|
|
+ foreach (var saddle in yield db.query<Saddle>()
|
|
|
|
|
+ .order_by(expr("id"))
|
|
|
|
|
+ .materialise_async()) {
|
|
|
|
|
+ if (saddle.status == SADDLE_OFFLINE
|
|
|
|
|
+ || busy.contains(((int64) saddle.id).to_string())) {
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+ rows.add(saddle);
|
|
|
|
|
+ }
|
|
|
|
|
+ return rows.to_immutable_buffer();
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* Marks a saddle offline when its control server cannot be
|
|
* 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' —
|
|
|
|
|
- * {@link TaskService.release_task} owns that bookkeeping; the
|
|
|
|
|
- * saddle row's own fate is the guarded offline write, never
|
|
|
|
|
- * the idle free). 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.
|
|
|
|
|
|
|
+ * reached. The task it was RUNNING (the derived running
|
|
|
|
|
+ * session — plan 16 item 23) releases back to 'queued' through
|
|
|
|
|
+ * {@link TaskService.release_task} (the ONE owner; it closes
|
|
|
|
|
+ * the session 'stopped'), with a system comment so the timeline
|
|
|
|
|
+ * shows the interruption; combined with the daemon's
|
|
|
|
|
+ * crash-resume, the next dispatch restores the workspace and
|
|
|
|
|
+ * reports `resumed` on /status. A task that parked mid-flight
|
|
|
|
|
+ * (its session already closed 'parked', the verdict awaiting
|
|
|
|
|
+ * the resume) keeps its park. The offline stamp itself is the
|
|
|
|
|
+ * ONE surviving guarded write — see {@link stamp_offline}.
|
|
|
*/
|
|
*/
|
|
|
public async void mark_saddle_unreachable(int64 id, TaskService tasks,
|
|
public async void mark_saddle_unreachable(int64 id, TaskService tasks,
|
|
|
string? reason = null) throws Error {
|
|
string? reason = null) throws Error {
|
|
@@ -579,61 +681,63 @@ namespace ClankerSaddle.Services {
|
|
|
if (saddle == null) {
|
|
if (saddle == null) {
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
|
- int64 held_task_at_entry = ((!)saddle).current_task_id;
|
|
|
|
|
|
|
+ var session = yield running_session_on_saddle(id);
|
|
|
|
|
+ int64 seen_session = session != null ? ((!)session).id : 0;
|
|
|
int64 held_task = 0;
|
|
int64 held_task = 0;
|
|
|
- bool release_held = false;
|
|
|
|
|
- if (held_task_at_entry > 0) {
|
|
|
|
|
- held_task = held_task_at_entry;
|
|
|
|
|
- bool keep_parked = false;
|
|
|
|
|
- var task = yield db.query<TaskEntity>()
|
|
|
|
|
- .where(expr("id == $0", elem<int64?>(held_task)))
|
|
|
|
|
- .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(held_task);
|
|
|
|
|
- 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 $(held_task) keeps its parked session: its answered block awaits delivery\n");
|
|
|
|
|
- } else {
|
|
|
|
|
- ((!)task).status = STATUS_QUEUED;
|
|
|
|
|
- yield db.update_async<TaskEntity>((!)task);
|
|
|
|
|
- string unreachable_text = reason != null && reason.length > 0
|
|
|
|
|
- ? @"Saddle became unreachable ($(reason)); task requeued"
|
|
|
|
|
- : "Saddle became unreachable; task requeued";
|
|
|
|
|
- yield tasks.add_system_comment(held_task, unreachable_text);
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ if (session != null) {
|
|
|
|
|
+ held_task = ((!)session).task_id;
|
|
|
|
|
+ // The requeue flip is guarded: a park (or terminal)
|
|
|
|
|
+ // interleaving this path keeps its state — the parked
|
|
|
|
|
+ // verdict must survive for the resume.
|
|
|
|
|
+ var flip = connection.create_command(
|
|
|
|
|
+ "UPDATE tasks SET status = 'queued' WHERE id = :id"
|
|
|
|
|
+ + " AND status IN ('assigned', 'working')");
|
|
|
|
|
+ flip.with_parameter<Invercargill.Element>("id", elem<int64?>(held_task));
|
|
|
|
|
+ if ((yield flip.execute_non_query_async()) > 0) {
|
|
|
|
|
+ string unreachable_text = reason != null && reason.length > 0
|
|
|
|
|
+ ? @"Saddle became unreachable ($(reason)); task requeued"
|
|
|
|
|
+ : "Saddle became unreachable; task requeued";
|
|
|
|
|
+ yield tasks.add_system_comment(held_task, unreachable_text);
|
|
|
|
|
+ // Closes the held session FIRST — the offline
|
|
|
|
|
+ // stamp's guard refuses to land over a running
|
|
|
|
|
+ // session, and by then there is none.
|
|
|
|
|
+ yield tasks.release_task(held_task, OUTCOME_STOPPED, false);
|
|
|
|
|
+ } else {
|
|
|
|
|
+ // Parked (or finished) mid-flight — the park closed
|
|
|
|
|
+ // its session and the verdict must survive for the
|
|
|
|
|
+ // resume; nothing to release.
|
|
|
|
|
+ held_task = 0;
|
|
|
}
|
|
}
|
|
|
- release_held = !keep_parked;
|
|
|
|
|
- }
|
|
|
|
|
- var offline_write = connection.create_command(
|
|
|
|
|
- "UPDATE saddles SET status = '" + SADDLE_OFFLINE
|
|
|
|
|
- + "', current_task_id = 0, allocated_at = NULL"
|
|
|
|
|
- + " WHERE id = :id AND current_task_id = :seen");
|
|
|
|
|
- offline_write.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
|
|
|
|
|
- offline_write.with_parameter<Invercargill.Element>("seen",
|
|
|
|
|
- elem<int64?>(held_task_at_entry));
|
|
|
|
|
- yield offline_write.execute_non_query_async();
|
|
|
|
|
|
|
+ }
|
|
|
|
|
+ yield stamp_offline(id, seen_session);
|
|
|
// An offline saddle is neither sleeping nor wake-awaiting;
|
|
// An offline saddle is neither sleeping nor wake-awaiting;
|
|
|
// a restarted daemon re-establishes both by asking.
|
|
// a restarted daemon re-establishes both by asking.
|
|
|
poll_state.clear_ask(id);
|
|
poll_state.clear_ask(id);
|
|
|
- if (release_held) {
|
|
|
|
|
- // The held task's session close runs through the ONE
|
|
|
|
|
- // release owner. Its idle-free no-ops against THIS row
|
|
|
|
|
- // (the guarded offline write above already cleared
|
|
|
|
|
- // current_task_id under the entry-seen guard); the
|
|
|
|
|
- // close applies whatever the task's state was —
|
|
|
|
|
- // flipped-and-requeued above, parked-suspended left
|
|
|
|
|
- // alone, or a row that no longer exists.
|
|
|
|
|
- yield tasks.release_task(held_task, OUTCOME_STOPPED, false);
|
|
|
|
|
- }
|
|
|
|
|
yield pulse.bump(0, "saddle unreachable");
|
|
yield pulse.bump(0, "saddle unreachable");
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * The ONE surviving guarded saddle write (plan 16 item 23):
|
|
|
|
|
+ * the OFFLINE probe stamp. The guard is the sessions-truth
|
|
|
|
|
+ * analogue of the old seen-assignment guard — the stamp lands
|
|
|
|
|
+ * unless a running session the probe did NOT see at entry
|
|
|
|
|
+ * appeared meanwhile (an ask claim is proof of life and must
|
|
|
|
|
+ * not be stomped offline). The session observed at entry
|
|
|
|
|
+ * (`seen_session`) never blocks its own stamp.
|
|
|
|
|
+ */
|
|
|
|
|
+ private async void stamp_offline(int64 saddle_id, int64 seen_session) throws Error {
|
|
|
|
|
+ var command = connection.create_command(
|
|
|
|
|
+ "UPDATE saddles SET status = '" + SADDLE_OFFLINE + "' WHERE id = :id"
|
|
|
|
|
+ + " AND NOT EXISTS (SELECT 1 FROM sessions s"
|
|
|
|
|
+ + " JOIN tasks t ON t.id = s.task_id"
|
|
|
|
|
+ + " WHERE s.saddle_id = :id AND s.ended_at IS NULL"
|
|
|
|
|
+ + " AND t.status IN ('assigned', 'working')"
|
|
|
|
|
+ + " AND s.id != :seen)");
|
|
|
|
|
+ command.with_parameter<Invercargill.Element>("id", elem<int64?>(saddle_id));
|
|
|
|
|
+ command.with_parameter<Invercargill.Element>("seen", elem<int64?>(seen_session));
|
|
|
|
|
+ yield command.execute_non_query_async();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
/**
|
|
/**
|
|
|
* Assembles the assignment payload shared by the /start push and
|
|
* Assembles the assignment payload shared by the /start push and
|
|
|
* the `POST /api/saddle/next-task` response — `{task, kind,
|
|
* the `POST /api/saddle/next-task` response — `{task, kind,
|
|
@@ -982,11 +1086,15 @@ namespace ClankerSaddle.Services {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
- * 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.
|
|
|
|
|
|
|
+ * Probes a saddle's `GET /status` and records what a probe can
|
|
|
|
|
+ * tell: the health snapshot, the daemon-reported ssh key and
|
|
|
|
|
+ * the goal-branch push flags. Busy/idle need NO write anymore —
|
|
|
|
|
+ * they are derived from the session records (plan 16 item 23).
|
|
|
|
|
+ * The row's only remaining write is the OFFLINE probe state: a
|
|
|
|
|
+ * successful probe is proof of life and clears it; a failed
|
|
|
|
|
+ * probe stamps it through the guarded write (an ask claim that
|
|
|
|
|
+ * landed while the probe was in flight is a NEWER proof of
|
|
|
|
|
+ * life and must not be stomped).
|
|
|
*/
|
|
*/
|
|
|
public async void refresh_saddle_status(int64 id) throws Error {
|
|
public async void refresh_saddle_status(int64 id) throws Error {
|
|
|
var saddle = yield get_saddle(id);
|
|
var saddle = yield get_saddle(id);
|
|
@@ -994,61 +1102,22 @@ namespace ClankerSaddle.Services {
|
|
|
throw new Error(Quark.from_string("clanker-saddle"), 17,
|
|
throw new Error(Quark.from_string("clanker-saddle"), 17,
|
|
|
@"No saddle exists with id $(id)");
|
|
@"No saddle exists with id $(id)");
|
|
|
}
|
|
}
|
|
|
- int64 seen_task = ((!)saddle).current_task_id;
|
|
|
|
|
- string next_status = SADDLE_IDLE;
|
|
|
|
|
- int64 next_task = 0;
|
|
|
|
|
- DateTime? next_allocated = null;
|
|
|
|
|
|
|
+ var session = yield running_session_on_saddle(id);
|
|
|
|
|
+ int64 seen_session = session != null ? ((!)session).id : 0;
|
|
|
try {
|
|
try {
|
|
|
var status = yield request_saddle_json(((!)saddle).host, "GET", "/status", null);
|
|
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);
|
|
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 = SADDLE_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.
|
|
|
|
|
|
|
+ var clear = connection.create_command(
|
|
|
|
|
+ "UPDATE saddles SET status = '" + SADDLE_IDLE
|
|
|
|
|
+ + "' WHERE id = :id AND status = '" + SADDLE_OFFLINE + "'");
|
|
|
|
|
+ clear.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
|
|
|
|
|
+ yield clear.execute_non_query_async();
|
|
|
} catch (Error e) {
|
|
} catch (Error e) {
|
|
|
- next_status = SADDLE_OFFLINE;
|
|
|
|
|
|
|
+ yield stamp_offline(id, seen_session);
|
|
|
}
|
|
}
|
|
|
- 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<Invercargill.Element>("status", elem(next_status));
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("task", elem<int64?>(next_task));
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("allocated", elem<DateTime?>(next_allocated));
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("id", elem<int64?>(id));
|
|
|
|
|
- command.with_parameter<Invercargill.Element>("seen", elem<int64?>(seen_task));
|
|
|
|
|
- yield command.execute_non_query_async();
|
|
|
|
|
yield pulse.bump(0, "saddle status refreshed");
|
|
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<ClankerSaddle.Data.Session>()
|
|
|
|
|
- .where(expr("saddle_id == $0", elem<int64?>(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 `task_id` not yet handed over anywhere;
|
|
* Answered blocks of `task_id` not yet handed over anywhere;
|
|
|
* marks them delivered — a resumed assignment carries its
|
|
* marks them delivered — a resumed assignment carries its
|
|
@@ -1493,18 +1562,20 @@ namespace ClankerSaddle.Services {
|
|
|
* plan 10 phase 2) — the same task/saddle bookkeeping the
|
|
* plan 10 phase 2) — the same task/saddle bookkeeping the
|
|
|
* deleted /status poll switch ran, driven by the daemon's push:
|
|
* deleted /status poll switch ran, driven by the daemon's push:
|
|
|
* `done`/`failed` complete or fail the task with the pushed
|
|
* `done`/`failed` complete or fail the task with the pushed
|
|
|
- * summary (freeing the saddle and closing the session) —
|
|
|
|
|
|
|
+ * summary (closing the session through the release owner) —
|
|
|
* except on a goal-run session, where they submit the
|
|
* except on a goal-run session, where they submit the
|
|
|
* completion report or fail the goal (plan 12); `blocked`
|
|
* completion report or fail the goal (plan 12); `blocked`
|
|
|
* raises the block payload as an answerable block on the
|
|
* raises the block payload as an answerable block on the
|
|
|
* goal run's OWN task row (plan 15 C1: the goal agent's
|
|
* goal run's OWN task row (plan 15 C1: the goal agent's
|
|
|
* blocks ride the run task — no mirrored second row exists)
|
|
* blocks ride the run task — no mirrored second row exists)
|
|
|
- * and parks the saddle; `suspended`
|
|
|
|
|
|
|
+ * and parks the round (the raise closes its session 'parked',
|
|
|
|
|
+ * D4); `suspended`
|
|
|
* parks the same way — for a goal agent the resume is driven
|
|
* parks the same way — for a goal agent the resume is driven
|
|
|
* by the goal's task-terminal events; and `working` flips an
|
|
* by the goal's task-terminal events; and `working` flips an
|
|
|
* ASSIGNED task to working (a parked task keeps its park —
|
|
* ASSIGNED task to working (a parked task keeps its park —
|
|
|
* the verdict arrives as a resuming allocation, never as a
|
|
* the verdict arrives as a resuming allocation, never as a
|
|
|
- * state push; plan 16 D4).
|
|
|
|
|
|
|
+ * state push; plan 16 D4). No saddle-row write remains
|
|
|
|
|
+ * (plan 16 item 23): busy/idle derive from the sessions.
|
|
|
*
|
|
*
|
|
|
* Plan 13 B2: a push against a task that is already terminal —
|
|
* Plan 13 B2: a push against a task that is already terminal —
|
|
|
* or a goal-context push against a finished goal — is an
|
|
* or a goal-context push against a finished goal — is an
|
|
@@ -1512,7 +1583,7 @@ namespace ClankerSaddle.Services {
|
|
|
* it carries; terminal transitions and their narrative fire
|
|
* it carries; terminal transitions and their narrative fire
|
|
|
* once, ever.
|
|
* once, ever.
|
|
|
*/
|
|
*/
|
|
|
- public async void apply_result_state(Saddle saddle, ClankerSaddle.Data.Session session,
|
|
|
|
|
|
|
+ public async void apply_result_state(ClankerSaddle.Data.Session session,
|
|
|
string state, Json.Object? block, string? summary, TaskService tasks) throws Error {
|
|
string state, Json.Object? block, string? summary, TaskService tasks) throws Error {
|
|
|
int64 task_id = session.task_id;
|
|
int64 task_id = session.task_id;
|
|
|
int64 goal_id = yield goal_id_of_run(task_id);
|
|
int64 goal_id = yield goal_id_of_run(task_id);
|
|
@@ -1536,7 +1607,6 @@ namespace ClankerSaddle.Services {
|
|
|
yield goal_service.submit_completion(goal_id,
|
|
yield goal_service.submit_completion(goal_id,
|
|
|
summary != null && summary.length > 0 ? summary
|
|
summary != null && summary.length > 0 ? summary
|
|
|
: "The goal agent submitted its completion report");
|
|
: "The goal agent submitted its completion report");
|
|
|
- yield park_if_current(saddle.id, task_id);
|
|
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
yield tasks.complete_task(task_id, summary, OUTCOME_DONE);
|
|
yield tasks.complete_task(task_id, summary, OUTCOME_DONE);
|
|
@@ -1565,7 +1635,6 @@ namespace ClankerSaddle.Services {
|
|
|
if (block == null) {
|
|
if (block == null) {
|
|
|
yield tasks.set_status(task_id, STATUS_BLOCKED);
|
|
yield tasks.set_status(task_id, STATUS_BLOCKED);
|
|
|
yield tasks.park_open_sessions(task_id);
|
|
yield tasks.park_open_sessions(task_id);
|
|
|
- yield park_if_current(saddle.id, task_id);
|
|
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
string question = block_question_from_payload((!)block);
|
|
string question = block_question_from_payload((!)block);
|
|
@@ -1583,7 +1652,6 @@ namespace ClankerSaddle.Services {
|
|
|
EntryJson.optional_string_member((!)block, "body"));
|
|
EntryJson.optional_string_member((!)block, "body"));
|
|
|
}
|
|
}
|
|
|
yield tasks.set_status(task_id, STATUS_BLOCKED);
|
|
yield tasks.set_status(task_id, STATUS_BLOCKED);
|
|
|
- yield park_if_current(saddle.id, task_id);
|
|
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
yield ensure_block(tasks, task_id, block != null ? (!)block : new Json.Object());
|
|
yield ensure_block(tasks, task_id, block != null ? (!)block : new Json.Object());
|
|
@@ -1596,20 +1664,21 @@ namespace ClankerSaddle.Services {
|
|
|
yield tasks.set_status(task_id,
|
|
yield tasks.set_status(task_id,
|
|
|
review ? STATUS_REVIEW_REQUESTED : STATUS_BLOCKED);
|
|
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);
|
|
|
|
|
|
|
+ // The daemon parks the task's workspace while
|
|
|
|
|
+ // blocked, freeing the saddle for other work
|
|
|
|
|
+ // (plan 16 item 23: the raise already closed the
|
|
|
|
|
+ // round's session 'parked' and the task's blocked
|
|
|
|
|
+ // status IS the pause — no saddle-row write
|
|
|
|
|
+ // remains; busy/idle derive from the sessions).
|
|
|
break;
|
|
break;
|
|
|
case "suspended":
|
|
case "suspended":
|
|
|
// `saddle suspend`: the same parking as a block, but
|
|
// `saddle suspend`: the same parking as a block, but
|
|
|
// the web drives resumption — a wake of the affine
|
|
// the web drives resumption — a wake of the affine
|
|
|
// saddle once a goal event lands, the resume
|
|
// saddle once a goal event lands, the resume
|
|
|
// assignment carrying the coalesced report.
|
|
// assignment carrying the coalesced report.
|
|
|
|
|
+ // (set_parked closed the round's session 'parked';
|
|
|
|
|
+ // the freed saddle is the derived view's reading.)
|
|
|
yield tasks.set_parked(task_id);
|
|
yield tasks.set_parked(task_id);
|
|
|
- yield park_if_current(saddle.id, task_id);
|
|
|
|
|
break;
|
|
break;
|
|
|
case "working":
|
|
case "working":
|
|
|
// A working push only advances a task that is
|
|
// A working push only advances a task that is
|
|
@@ -1681,11 +1750,12 @@ namespace ClankerSaddle.Services {
|
|
|
* bookkeeping). The probe flag was taken by the caller; the
|
|
* bookkeeping). The probe flag was taken by the caller; the
|
|
|
* next push resets it.
|
|
* next push resets it.
|
|
|
*/
|
|
*/
|
|
|
- public async void probe_stale_saddle(Saddle saddle, TaskService tasks) throws Error {
|
|
|
|
|
|
|
+ public async void probe_stale_saddle(Saddle saddle, TaskService tasks,
|
|
|
|
|
+ int64 running_task_id) throws Error {
|
|
|
try {
|
|
try {
|
|
|
var status = yield request_saddle_json(saddle.host, "GET", "/status", null);
|
|
var status = yield request_saddle_json(saddle.host, "GET", "/status", null);
|
|
|
yield record_status_probe(saddle, status);
|
|
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");
|
|
|
|
|
|
|
+ print(@"[ClankerSaddle] Stale saddle '$(saddle.name)' answered its exceptional status probe; task $(running_task_id) keeps running\n");
|
|
|
} catch (Error e) {
|
|
} 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");
|
|
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,
|
|
yield mark_saddle_unreachable(saddle.id, tasks,
|