From c96a8da4c357f7916deb9f9b15b33bf88805d564 Mon Sep 17 00:00:00 2001 From: Guy Date: Thu, 20 Aug 2026 19:11:48 +0300 Subject: [PATCH] feat(runs): GET /runs/{id}, GET /runs list, POST /runs/{id}/cancel - GET /runs/{id}: single run fetch incl usage_json + error_text; 404 unknown - GET /runs: agent_id/status filters, limit 1..1000 (default 50), newest-first - POST /runs/{id}/cancel: pending -> terminal 'cancelled' (ended_at_ms + error_text=note); running -> 'cancellation_requested' event appended, status unchanged (executor polls events, then transitions); terminal -> already_terminal; unknown -> 404. Admin token required (not lease-gated). - store: getRun/listRuns/cancelRun (tx-safe; not_found rolls back), CancelOutcome - openapi: /runs, /runs/{id}, /runs/{id}/cancel paths - tests: 5 store + 4 api unit tests; e2e section 7.1 --- src/api.zig | 231 ++++++++++++++++++++++++++++++++++++++++ src/openapi.json | 59 +++++++++++ src/store.zig | 264 ++++++++++++++++++++++++++++++++++++++++++++++ tests/test_e2e.sh | 92 ++++++++++++++++ 4 files changed, 646 insertions(+) diff --git a/src/api.zig b/src/api.zig index 161011e..e3803f1 100644 --- a/src/api.zig +++ b/src/api.zig @@ -244,8 +244,24 @@ pub fn handleRequest( } } + // Runs (collection) + if (eql(seg0, "runs") and seg1 == null) { + if (is_get) { + response = handleListRuns(ctx, path.query); + return response; + } + } + // Runs if (eql(seg0, "runs") and seg1 != null) { + if (is_get and seg2 == null) { + response = handleGetRun(ctx, seg1.?); + return response; + } + if (is_post and eql(seg2, "cancel")) { + response = handleCancelRun(ctx, seg1.?, body); + return finalizeWithIdempotency(ctx, method, path.path, idempotency, response); + } if (is_post and eql(seg2, "events")) { response = handleAddEvent(ctx, seg1.?, body, raw_request); return finalizeWithIdempotency(ctx, method, path.path, idempotency, response); @@ -918,6 +934,83 @@ fn handleListEvents(ctx: *Context, run_id: []const u8, query: ?[]const u8) HttpR return .{ .status = "200 OK", .body = out.written() }; } +fn handleGetRun(ctx: *Context, run_id: []const u8) HttpResponse { + const run = (ctx.store.getRun(run_id) catch return serverError(ctx.allocator)) orelse { + return respondError(ctx.allocator, 404, "not_found", "Run not found"); + }; + defer ctx.store.freeRunRow(run); + + var out: std.Io.Writer.Allocating = .init(ctx.allocator); + const w = &out.writer; + w.writeAll("{") catch return serverError(ctx.allocator); + writeRunFields(w, ctx.allocator, run) catch return serverError(ctx.allocator); + w.print(",\"usage\":{s}", .{run.usage_json}) catch return serverError(ctx.allocator); + if (run.error_text) |error_text| { + w.writeAll(",") catch return serverError(ctx.allocator); + writeStringField(w, ctx.allocator, "error_text", error_text) catch return serverError(ctx.allocator); + } else { + w.writeAll(",\"error_text\":null") catch return serverError(ctx.allocator); + } + w.writeAll("}") catch return serverError(ctx.allocator); + return .{ .status = "200 OK", .body = out.written() }; +} + +fn handleListRuns(ctx: *Context, query: ?[]const u8) HttpResponse { + const agent_id = parseQueryParam(query, "agent_id"); + const status = parseQueryParam(query, "status"); + const limit_str = parseQueryParam(query, "limit"); + const limit = if (limit_str) |ls| (std.fmt.parseInt(i64, ls, 10) catch 50) else 50; + if (limit <= 0 or limit > 1000) { + return respondError(ctx.allocator, 400, "invalid_limit", "limit must be between 1 and 1000"); + } + + const items = ctx.store.listRuns(agent_id, status, limit) catch return serverError(ctx.allocator); + defer ctx.store.freeRunRows(items); + + var out: std.Io.Writer.Allocating = .init(ctx.allocator); + const w = &out.writer; + w.writeAll("{\"items\":[") catch return serverError(ctx.allocator); + for (items, 0..) |run, i| { + if (i > 0) w.writeAll(",") catch return serverError(ctx.allocator); + w.writeAll("{") catch return serverError(ctx.allocator); + writeRunFields(w, ctx.allocator, run) catch return serverError(ctx.allocator); + w.print(",\"usage\":{s}", .{run.usage_json}) catch return serverError(ctx.allocator); + if (run.error_text) |error_text| { + w.writeAll(",") catch return serverError(ctx.allocator); + writeStringField(w, ctx.allocator, "error_text", error_text) catch return serverError(ctx.allocator); + } else { + w.writeAll(",\"error_text\":null") catch return serverError(ctx.allocator); + } + w.writeAll("}") catch return serverError(ctx.allocator); + } + w.writeAll("]}") catch return serverError(ctx.allocator); + return .{ .status = "200 OK", .body = out.written() }; +} + +fn handleCancelRun(ctx: *Context, run_id: []const u8, body: []const u8) HttpResponse { + var parsed = std.json.parseFromSlice(struct { + note: ?[]const u8 = null, + }, ctx.allocator, body, .{ .ignore_unknown_fields = true }) catch { + return respondError(ctx.allocator, 400, "invalid_json", "Invalid JSON body"); + }; + defer parsed.deinit(); + const note = parsed.value.note orelse "cancel requested"; + + var note_obj: std.json.ObjectMap = .empty; + note_obj.put(ctx.allocator, "note", .{ .string = note }) catch return serverError(ctx.allocator); + const data_json = jsonStringify(ctx.allocator, .{ .object = note_obj }) catch return serverError(ctx.allocator); + + const outcome = ctx.store.cancelRun(run_id, note, data_json) catch return serverError(ctx.allocator); + const outcome_str: []const u8 = switch (outcome) { + .cancelled_from_pending => "cancelled_from_pending", + .cancellation_requested_for_running => "cancellation_requested_for_running", + .already_terminal => "already_terminal", + .not_found => return respondError(ctx.allocator, 404, "not_found", "Run not found"), + }; + const resp = std.fmt.allocPrint(ctx.allocator, "{{\"outcome\":\"{s}\"}}", .{outcome_str}) catch return serverError(ctx.allocator); + return .{ .status = "200 OK", .body = resp }; +} + fn handleTransition(ctx: *Context, run_id: []const u8, body: []const u8, raw_request: []const u8) HttpResponse { const token = extractBearerToken(raw_request) orelse { return respondError(ctx.allocator, 401, "unauthorized", "Missing Authorization header"); @@ -1826,3 +1919,141 @@ test "store search rejects excessive limit" { try std.testing.expectEqualStrings("400 Bad Request", resp.status); try std.testing.expect(std.mem.indexOf(u8, resp.body, "\"invalid_limit\"") != null); } + +fn setupRunWithArena(arena: std.mem.Allocator, store: *Store, pipeline_name: []const u8, agent_id: []const u8) ![]const u8 { + const pipeline_def = + \\{"initial":"work","states":{"work":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"work","to":"done","trigger":"complete"}]} + ; + var name_buf: [64]u8 = undefined; + const name = try std.fmt.bufPrint(&name_buf, "{s}", .{pipeline_name}); + const pipeline_id = try store.createPipeline(name, pipeline_def); + defer store.freeOwnedString(pipeline_id); + const task_id = try store.createTask(pipeline_id, "Task", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(task_id); + const claim = (try store.claimTask(agent_id, "worker", 300_000, null)).?; + defer store.freeClaimResult(claim); + return arena.dupe(u8, claim.run.id); +} + +test "runs API: GET /runs/{id} returns run or 404" { + const allocator = std.testing.allocator; + var store = try Store.init(allocator, ":memory:"); + defer store.deinit(); + + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + + var ctx = Context{ + .store = &store, + .allocator = arena.allocator(), + }; + + const run_id = try setupRunWithArena(arena.allocator(), &store, "get-run-test", "worker-1"); + + const target = try std.fmt.allocPrint(arena.allocator(), "/runs/{s}", .{run_id}); + const raw = try std.fmt.allocPrint(arena.allocator(), "GET {s} HTTP/1.1\r\n\r\n", .{target}); + const resp = handleRequest(&ctx, "GET", target, "", raw); + try std.testing.expectEqualStrings("200 OK", resp.status); + try std.testing.expect(std.mem.indexOf(u8, resp.body, "\"status\":\"running\"") != null); + try std.testing.expect(std.mem.indexOf(u8, resp.body, "\"agent_id\":\"worker-1\"") != null); + try std.testing.expect(std.mem.indexOf(u8, resp.body, "\"usage\":{}") != null); + + const missing_resp = handleRequest(&ctx, "GET", "/runs/nope", "", "GET /runs/nope HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("404 Not Found", missing_resp.status); +} + +test "runs API: GET /runs filters by agent_id, status, limit" { + const allocator = std.testing.allocator; + var store = try Store.init(allocator, ":memory:"); + defer store.deinit(); + + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + + var ctx = Context{ + .store = &store, + .allocator = arena.allocator(), + }; + + _ = try setupRunWithArena(arena.allocator(), &store, "list-runs-test-a", "worker-a"); + _ = try setupRunWithArena(arena.allocator(), &store, "list-runs-test-b", "worker-b"); + + const all_resp = handleRequest(&ctx, "GET", "/runs", "", "GET /runs HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("200 OK", all_resp.status); + try std.testing.expect(std.mem.indexOf(u8, all_resp.body, "\"items\":[") != null); + try std.testing.expect(std.mem.indexOf(u8, all_resp.body, "\"agent_id\":\"worker-a\"") != null); + try std.testing.expect(std.mem.indexOf(u8, all_resp.body, "\"agent_id\":\"worker-b\"") != null); + + const filtered_resp = handleRequest(&ctx, "GET", "/runs?agent_id=worker-b", "", "GET /runs?agent_id=worker-b HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("200 OK", filtered_resp.status); + try std.testing.expect(std.mem.indexOf(u8, filtered_resp.body, "worker-a") == null); + + const status_resp = handleRequest(&ctx, "GET", "/runs?status=running", "", "GET /runs?status=running HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("200 OK", status_resp.status); + try std.testing.expect(std.mem.indexOf(u8, status_resp.body, "\"status\":\"running\"") != null); + + const bad_limit = handleRequest(&ctx, "GET", "/runs?limit=1001", "", "GET /runs?limit=1001 HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("400 Bad Request", bad_limit.status); +} + +test "runs API: POST /runs/{id}/cancel outcome mapping" { + const allocator = std.testing.allocator; + var store = try Store.init(allocator, ":memory:"); + defer store.deinit(); + + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + + var ctx = Context{ + .store = &store, + .allocator = arena.allocator(), + }; + + const run_id = try setupRunWithArena(arena.allocator(), &store, "cancel-api-test", "worker-1"); + + // running -> cancellation_requested_for_running + const cancel_body = "{\"note\":\"user asked to stop\"}"; + var target_buf: [128]u8 = undefined; + var raw_buf: [192]u8 = undefined; + const target = try std.fmt.bufPrint(&target_buf, "/runs/{s}/cancel", .{run_id}); + const raw = try std.fmt.bufPrint(&raw_buf, "POST {s} HTTP/1.1\r\n\r\n", .{target}); + const cancel_resp = handleRequest(&ctx, "POST", target, cancel_body, raw); + try std.testing.expectEqualStrings("200 OK", cancel_resp.status); + try std.testing.expect(std.mem.indexOf(u8, cancel_resp.body, "\"outcome\":\"cancellation_requested_for_running\"") != null); + + // run stays running; executor polls events for the note + const events_target = try std.fmt.allocPrint(arena.allocator(), "/runs/{s}/events", .{run_id}); + const events_raw = try std.fmt.allocPrint(arena.allocator(), "GET {s} HTTP/1.1\r\n\r\n", .{events_target}); + const events_resp = handleRequest(&ctx, "GET", events_target, "", events_raw); + try std.testing.expectEqualStrings("200 OK", events_resp.status); + try std.testing.expect(std.mem.indexOf(u8, events_resp.body, "\"kind\":\"cancellation_requested\"") != null); + try std.testing.expect(std.mem.indexOf(u8, events_resp.body, "user asked to stop") != null); + + // terminal -> already_terminal + try store.failRun(run_id, "boom", null); + const late_resp = handleRequest(&ctx, "POST", target, "{}", raw); + try std.testing.expectEqualStrings("200 OK", late_resp.status); + try std.testing.expect(std.mem.indexOf(u8, late_resp.body, "\"outcome\":\"already_terminal\"") != null); + + // unknown -> 404 + const missing_resp = handleRequest(&ctx, "POST", "/runs/nope/cancel", "{}", "POST /runs/nope/cancel HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("404 Not Found", missing_resp.status); +} + +test "runs API: GET /runs/{id} requires admin token when configured" { + const allocator = std.testing.allocator; + var store = try Store.init(allocator, ":memory:"); + defer store.deinit(); + + var ctx = Context{ + .store = &store, + .allocator = allocator, + .required_api_token = "secret", + }; + + const no_token = handleRequest(&ctx, "GET", "/runs/some-id", "", "GET /runs/some-id HTTP/1.1\r\n\r\n"); + try std.testing.expectEqualStrings("401 Unauthorized", no_token.status); + + const with_token = handleRequest(&ctx, "GET", "/runs/some-id", "", "GET /runs/some-id HTTP/1.1\r\nAuthorization: Bearer secret\r\n\r\n"); + try std.testing.expectEqualStrings("404 Not Found", with_token.status); +} diff --git a/src/openapi.json b/src/openapi.json index 5ced725..28598ab 100644 --- a/src/openapi.json +++ b/src/openapi.json @@ -370,6 +370,65 @@ } } }, + "/runs": { + "get": { + "summary": "List runs (newest-first)", + "security": [{ "bearerAuth": [] }], + "parameters": [ + { "name": "agent_id", "in": "query", "required": false, "schema": { "type": "string" }, "description": "Filter by the agent that claimed the run (worker)" }, + { "name": "status", "in": "query", "required": false, "schema": { "type": "string" }, "description": "Filter by run status (running, completed, failed, stale, cancelled, pending)" }, + { "name": "limit", "in": "query", "required": false, "schema": { "type": "integer", "minimum": 1, "maximum": 1000 } } + ], + "responses": { + "200": { + "description": "Run list", + "content": { "application/json": { "schema": { "type": "object", "properties": { "items": { "type": "array", "items": { "$ref": "#/components/schemas/Run" } } }, "required": ["items"] } } } + }, + "400": { "$ref": "#/components/responses/Error400" }, + "401": { "$ref": "#/components/responses/Error401" } + } + } + }, + "/runs/{id}": { + "get": { + "summary": "Get single run by id", + "security": [{ "bearerAuth": [] }], + "parameters": [ + { "name": "id", "in": "path", "required": true, "schema": { "type": "string" } } + ], + "responses": { + "200": { + "description": "Run", + "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Run" } } } + }, + "401": { "$ref": "#/components/responses/Error401" }, + "404": { "$ref": "#/components/responses/Error404" } + } + } + }, + "/runs/{id}/cancel": { + "post": { + "summary": "Request run cancellation", + "description": "Outcome-mapped cancel: pending runs transition directly to the terminal 'cancelled' status; running runs get a cancellation_requested event appended (the executor polls run events, observes it, and terminates the work); terminal runs report already_terminal.", + "security": [{ "bearerAuth": [] }], + "parameters": [ + { "name": "id", "in": "path", "required": true, "schema": { "type": "string" } } + ], + "requestBody": { + "required": false, + "content": { "application/json": { "schema": { "type": "object", "properties": { "note": { "type": "string" } } } } } + }, + "responses": { + "200": { + "description": "Cancel outcome", + "content": { "application/json": { "schema": { "type": "object", "properties": { "outcome": { "type": "string", "enum": ["cancelled_from_pending", "cancellation_requested_for_running", "already_terminal"] } }, "required": ["outcome"] } } } + }, + "400": { "$ref": "#/components/responses/Error400" }, + "401": { "$ref": "#/components/responses/Error401" }, + "404": { "$ref": "#/components/responses/Error404" } + } + } + }, "/artifacts": { "post": { "summary": "Add artifact", diff --git a/src/store.zig b/src/store.zig index 14bbe7d..b420497 100644 --- a/src/store.zig +++ b/src/store.zig @@ -66,6 +66,13 @@ pub const EventRow = struct { data_json: []const u8, }; +pub const CancelOutcome = enum { + cancelled_from_pending, + cancellation_requested_for_running, + already_terminal, + not_found, +}; + pub const ArtifactRow = struct { id: []const u8, task_id: ?[]const u8, @@ -800,6 +807,106 @@ pub const Store = struct { }; } + pub fn getRun(self: *Self, run_id: []const u8) !?RunRow { + const stmt = try self.prepare("SELECT id, task_id, attempt, status, agent_id, agent_role, started_at_ms, ended_at_ms, usage_json, error_text FROM runs WHERE id = ?;"); + defer _ = c.sqlite3_finalize(stmt); + self.bindText(stmt, 1, run_id); + if (c.sqlite3_step(stmt) != c.SQLITE_ROW) return null; + return self.readRunRow(stmt); + } + + pub fn listRuns(self: *Self, agent_id_filter: ?[]const u8, status_filter: ?[]const u8, limit: i64) ![]RunRow { + var sql_buf: [512]u8 = undefined; + var sql_len: usize = 0; + const base = "SELECT id, task_id, attempt, status, agent_id, agent_role, started_at_ms, ended_at_ms, usage_json, error_text FROM runs"; + @memcpy(sql_buf[0..base.len], base); + sql_len = base.len; + + var has_where = false; + if (agent_id_filter != null) { + const clause = " WHERE agent_id = ?"; + @memcpy(sql_buf[sql_len..][0..clause.len], clause); + sql_len += clause.len; + has_where = true; + } + if (status_filter != null) { + const clause = if (has_where) " AND status = ?" else " WHERE status = ?"; + @memcpy(sql_buf[sql_len..][0..clause.len], clause); + sql_len += clause.len; + has_where = true; + } + + const order = " ORDER BY started_at_ms DESC, id DESC LIMIT ?;"; + @memcpy(sql_buf[sql_len..][0..order.len], order); + sql_len += order.len; + sql_buf[sql_len] = 0; + const sql_z: [*:0]const u8 = @ptrCast(sql_buf[0..sql_len :0]); + + const stmt = try self.prepare(sql_z); + defer _ = c.sqlite3_finalize(stmt); + + var bind_idx: c_int = 1; + if (agent_id_filter) |af| { + self.bindText(stmt, bind_idx, af); + bind_idx += 1; + } + if (status_filter) |sf| { + self.bindText(stmt, bind_idx, sf); + bind_idx += 1; + } + _ = c.sqlite3_bind_int64(stmt, bind_idx, limit); + + var rows: std.ArrayListUnmanaged(RunRow) = .empty; + while (c.sqlite3_step(stmt) == c.SQLITE_ROW) { + try rows.append(self.allocator, self.readRunRow(stmt)); + } + return rows.toOwnedSlice(self.allocator); + } + + pub fn cancelRun(self: *Self, run_id: []const u8, note: []const u8, event_data_json: []const u8) !CancelOutcome { + try self.execSimple("BEGIN IMMEDIATE;"); + errdefer self.execSimple("ROLLBACK;") catch {}; + + const run_stmt = try self.prepare("SELECT status FROM runs WHERE id = ?;"); + defer _ = c.sqlite3_finalize(run_stmt); + self.bindText(run_stmt, 1, run_id); + if (c.sqlite3_step(run_stmt) != c.SQLITE_ROW) { + self.execSimple("ROLLBACK;") catch {}; + return CancelOutcome.not_found; + } + const status_view = self.colTextView(run_stmt, 0); + + if (std.mem.eql(u8, status_view, "running")) { + // Not a status change: the executor polls run events, observes the + // cancellation_requested note, SIGTERMs, then transitions/fails. + _ = try self.addEvent(run_id, "cancellation_requested", event_data_json); + try self.execSimple("COMMIT;"); + return CancelOutcome.cancellation_requested_for_running; + } + + if (std.mem.eql(u8, status_view, "pending")) { + const now_ms = ids.nowMs(); + const upd = try self.prepare("UPDATE runs SET status = 'cancelled', ended_at_ms = ?, error_text = ? WHERE id = ?;"); + defer _ = c.sqlite3_finalize(upd); + _ = c.sqlite3_bind_int64(upd, 1, now_ms); + self.bindText(upd, 2, note); + self.bindText(upd, 3, run_id); + _ = c.sqlite3_step(upd); + + const del = try self.prepare("DELETE FROM leases WHERE run_id = ?;"); + defer _ = c.sqlite3_finalize(del); + self.bindText(del, 1, run_id); + _ = c.sqlite3_step(del); + + try self.execSimple("COMMIT;"); + return CancelOutcome.cancelled_from_pending; + } + + // completed / failed / stale / cancelled + try self.execSimple("COMMIT;"); + return CancelOutcome.already_terminal; + } + pub fn listEventsPage(self: *Self, run_id: []const u8, cursor_id: ?i64, limit: i64) !EventPage { const page_limit: usize = @intCast(limit); const sql = if (cursor_id != null) @@ -1486,6 +1593,11 @@ pub const Store = struct { if (row.error_text) |error_text| self.allocator.free(error_text); } + pub fn freeRunRows(self: *Self, rows: []RunRow) void { + for (rows) |row| self.freeRunRow(row); + self.allocator.free(rows); + } + pub fn freeClaimResult(self: *Self, claim: ClaimResult) void { self.freeTaskRow(claim.task); self.freeRunRow(claim.run); @@ -2528,3 +2640,155 @@ test "fail run with retry policy" { try std.testing.expect(task_after_1.next_eligible_at_ms > 0); try std.testing.expect(task_after_1.dead_letter_reason == null); } + +test "getRun returns null for unknown run" { + var store = try Store.init(std.testing.allocator, ":memory:"); + defer store.deinit(); + const run = try store.getRun("nope"); + try std.testing.expect(run == null); +} + +test "getRun and listRuns: filters, newest-first" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"work","states":{"work":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"work","to":"done","trigger":"complete"}]} + ; + const pipeline_id = try store.createPipeline("runs-list-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + const t1 = try store.createTask(pipeline_id, "Task 1", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(t1); + const t2 = try store.createTask(pipeline_id, "Task 2", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(t2); + + const c1 = (try store.claimTask("worker-a", "worker", 300_000, null)).?; + defer store.freeClaimResult(c1); + const c2 = (try store.claimTask("worker-b", "worker", 300_000, null)).?; + defer store.freeClaimResult(c2); + + // getRun by id + const got = (try store.getRun(c1.run.id)).?; + defer store.freeRunRow(got); + try std.testing.expectEqualStrings(c1.run.id, got.id); + try std.testing.expectEqualStrings(t1, got.task_id); + try std.testing.expectEqualStrings("running", got.status); + try std.testing.expectEqualStrings("worker-a", got.agent_id.?); + + // listRuns unfiltered: 2 items + const all = try store.listRuns(null, null, 50); + defer store.freeRunRows(all); + try std.testing.expectEqual(@as(usize, 2), all.items.len); + + // filter by agent + const only_b = try store.listRuns("worker-b", null, 50); + defer store.freeRunRows(only_b); + try std.testing.expectEqual(@as(usize, 1), only_b.items.len); + try std.testing.expectEqualStrings(c2.run.id, only_b.items[0].id); + + // filter by status + const running = try store.listRuns(null, "running", 50); + defer store.freeRunRows(running); + try std.testing.expectEqual(@as(usize, 2), running.items.len); + + // limit + const limited = try store.listRuns(null, null, 1); + defer store.freeRunRows(limited); + try std.testing.expectEqual(@as(usize, 1), limited.items.len); + + // newest-first: second claim started later (or same ms; id DESC tiebreak) + const newest = try store.listRuns(null, null, 50); + defer store.freeRunRows(newest); + const ms0 = newest.items[0].started_at_ms.?; + const ms1 = newest.items[1].started_at_ms.?; + try std.testing.expect(ms0 >= ms1); +} + +test "cancelRun: running requests cancellation via event" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"work","states":{"work":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"work","to":"done","trigger":"complete"}]} + ; + const pipeline_id = try store.createPipeline("cancel-running-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + const task_id = try store.createTask(pipeline_id, "Cancel Task", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(task_id); + + const claim = (try store.claimTask("worker-a", "worker", 300_000, null)).?; + defer store.freeClaimResult(claim); + + const outcome = try store.cancelRun(claim.run.id, "user asked to stop", "{\"note\":\"user asked to stop\"}"); + try std.testing.expectEqual(CancelOutcome.cancellation_requested_for_running, outcome); + + // status unchanged + const run = (try store.getRun(claim.run.id)).?; + defer store.freeRunRow(run); + try std.testing.expectEqualStrings("running", run.status); + + // event visible to the executor poll + const events = try store.listEventsPage(claim.run.id, null, 10); + defer store.freeEventPage(events); + var found = false; + for (events.items) |e| { + if (std.mem.eql(u8, e.kind, "cancellation_requested")) found = true; + } + try std.testing.expect(found); +} + +test "cancelRun: pending cancels directly to terminal" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"work","states":{"work":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"work","to":"done","trigger":"complete"}]} + ; + const pipeline_id = try store.createPipeline("cancel-pending-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + const task_id = try store.createTask(pipeline_id, "Pending Task", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(task_id); + + // Runs are created at claim time as 'running'; a 'pending' run only exists if a + // producer creates one pre-claim. Simulate it directly. + try store.execSimple("INSERT INTO runs (id, task_id, attempt, status, started_at_ms) VALUES ('pending-run-1', 'does-not-exist-task', 1, 'pending', NULL);"); + const outcome = try store.cancelRun("pending-run-1", "cancel before start", "{}"); + try std.testing.expectEqual(CancelOutcome.cancelled_from_pending, outcome); + + const run = (try store.getRun("pending-run-1")).?; + defer store.freeRunRow(run); + try std.testing.expectEqualStrings("cancelled", run.status); + try std.testing.expect(run.ended_at_ms != null); + try std.testing.expectEqualStrings("cancel before start", run.error_text.?); +} + +test "cancelRun: terminal and unknown" { + const alloc = std.testing.allocator; + var store = try Store.init(alloc, ":memory:"); + defer store.deinit(); + + const pipeline_def = + \\{"initial":"work","states":{"work":{"agent_role":"worker"},"done":{"terminal":true}},"transitions":[{"from":"work","to":"done","trigger":"complete"}]} + ; + const pipeline_id = try store.createPipeline("cancel-terminal-test", pipeline_def); + defer store.freeOwnedString(pipeline_id); + + const task_id = try store.createTask(pipeline_id, "Terminal Task", "desc", 0, "{}", null, 0, null); + defer store.freeOwnedString(task_id); + + const claim = (try store.claimTask("worker-a", "worker", 300_000, null)).?; + defer store.freeClaimResult(claim); + try store.failRun(claim.run.id, "boom", null); + + const outcome = try store.cancelRun(claim.run.id, "late cancel", "{}"); + try std.testing.expectEqual(CancelOutcome.already_terminal, outcome); + + const unknown = try store.cancelRun("no-such-run", "x", "{}"); + try std.testing.expectEqual(CancelOutcome.not_found, unknown); +} diff --git a/tests/test_e2e.sh b/tests/test_e2e.sh index 58f7f5d..2321d66 100755 --- a/tests/test_e2e.sh +++ b/tests/test_e2e.sh @@ -415,6 +415,98 @@ RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" \ CODE=$(echo "$RESP" | tail -1) assert_status 200 "$CODE" "Complete after retry" +# ===== 7.1 Runs API: get / list / cancel ===== +echo "" +echo "=== 7.1 Runs API (get/list/cancel) ===" +RUNS_PIPELINE=$(cat <<'JSON' +{ + "name": "runs-api-e2e", + "definition": { + "initial": "work", + "states": { "work": { "agent_role": "worker" }, "done": { "terminal": true } }, + "transitions": [ { "from": "work", "to": "done", "trigger": "complete" } ] + } +} +JSON +) +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" -d "$RUNS_PIPELINE" "$BASE/pipelines") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 201 "$CODE" "POST /pipelines (runs-api-e2e)" +RUNS_PIPELINE_ID=$(echo "$BODY" | python3 -c "import sys,json; print(json.load(sys.stdin)['id'])") + +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" \ + -d "{\"pipeline_id\":\"$RUNS_PIPELINE_ID\",\"title\":\"Runs API Task\",\"description\":\"d\"}" \ + "$BASE/tasks") +CODE=$(echo "$RESP" | tail -1) +assert_status 201 "$CODE" "POST /tasks (runs-api-e2e)" + +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" \ + -d '{"agent_id":"worker-runs","agent_role":"worker"}' \ + "$BASE/leases/claim") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "POST /leases/claim (runs-api-e2e)" +RUNS_RUN_ID=$(echo "$BODY" | python3 -c "import sys,json; print(json.load(sys.stdin)['run']['id'])") + +# GET single run +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs/$RUNS_RUN_ID") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "GET /runs/{id}" +assert_json "$BODY" "data['status']" "running" "run status" +assert_json "$BODY" "data['agent_id']" "worker-runs" "run agent_id" +assert_json "$BODY" "data['usage']" "{}" "run usage object" + +# GET unknown run +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs/does-not-exist") +CODE=$(echo "$RESP" | tail -1) +assert_status 404 "$CODE" "GET /runs/{id} unknown -> 404" + +# GET list, filtered +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs?agent_id=worker-runs") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "GET /runs?agent_id=..." +assert_json "$BODY" "str(len(data['items']))" "1" "runs list filtered by agent" + +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs?status=running&limit=5") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "GET /runs?status=running" +assert_json "$BODY" "any(i['id'] == '$RUNS_RUN_ID' for i in data['items'])" "True" "runs list includes new run" + +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs?limit=0") +CODE=$(echo "$RESP" | tail -1) +assert_status 400 "$CODE" "GET /runs limit=0 -> 400" + +# Cancel running run -> cancellation_requested_for_running + event +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" \ + -d '{"note":"user pressed ctrl-c"}' \ + "$BASE/runs/$RUNS_RUN_ID/cancel") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "POST /runs/{id}/cancel (running)" +assert_json "$BODY" "data['outcome']" "cancellation_requested_for_running" "cancel outcome (running)" + +RESP=$(curl -s -w "\n%{http_code}" "$BASE/runs/$RUNS_RUN_ID/events") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "GET /runs/{id}/events after cancel" +assert_json "$BODY" "any(e['kind'] == 'cancellation_requested' for e in data['items'])" "True" "cancellation_requested event present" + +# Cancel unknown run -> 404 +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" -d '{}' "$BASE/runs/nope/cancel") +CODE=$(echo "$RESP" | tail -1) +assert_status 404 "$CODE" "POST /runs/{id}/cancel unknown -> 404" + +# Cancel terminal run -> already_terminal (RUN4 was failed earlier) +RESP=$(curl -s -w "\n%{http_code}" -X POST -H "Content-Type: application/json" -d '{}' "$BASE/runs/$RUN4_ID/cancel") +CODE=$(echo "$RESP" | tail -1) +BODY=$(echo "$RESP" | sed '$d') +assert_status 200 "$CODE" "POST /runs/{id}/cancel (terminal)" +assert_json "$BODY" "data['outcome']" "already_terminal" "cancel outcome (terminal)" + # ===== 8. Artifacts ===== echo "" echo "=== 8. Artifacts ==="