diff --git a/config/aipods/artificer.yaml b/config/aipods/artificer.yaml index 48720b5..be5d554 100644 --- a/config/aipods/artificer.yaml +++ b/config/aipods/artificer.yaml @@ -147,7 +147,7 @@ spec: - name: git-spec - name: unidesk-trans - name: unidesk-gh - - name: unidesk-code-queue + - name: unidesk-agentrun - name: unidesk-cicd - name: unidesk-subagent - name: unidesk-decision diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 62c5aa1..b8f194a 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -5,8 +5,8 @@ import { AgentRunError } from "../common/errors.js"; import { redactJson } from "../common/redaction.js"; import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; -import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, SessionSteerAdmission, SessionSteerAdmissionInput, SessionTurnAdmission, SessionTurnAdmissionInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; -import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionSteerAdmissionContract, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, receivableActiveSessionTurn, runCreatedEventPayload, runnerDispatchFailureKind, runnerDispatchFailureReason, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, staleActiveSessionTurn, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; +import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListRunsInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunCollectionResult, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, SessionSteerAdmission, SessionSteerAdmissionInput, SessionTurnAdmission, SessionTurnAdmissionInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; +import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionSteerAdmissionContract, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, receivableActiveSessionTurn, runCollectionItem, runCollectionResult, runCreatedEventPayload, runnerDispatchFailureKind, runnerDispatchFailureReason, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, staleActiveSessionTurn, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js"; import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js"; import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; @@ -600,6 +600,80 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( return runFromRow(row); } + async listRuns(input: ListRunsInput): Promise { + const parameters: unknown[] = []; + const conditions: string[] = []; + const equalFilter = (column: string, value: string | undefined) => { + if (!value) return; + parameters.push(value); + conditions.push(`${column} = $${parameters.length}`); + }; + if (input.statuses?.length) { + parameters.push(input.statuses); + conditions.push(`r.status = ANY($${parameters.length}::text[])`); + } + equalFilter("r.project_id", input.projectId); + equalFilter("r.workspace_ref->>'path'", input.workspace); + equalFilter("r.backend_profile", input.backendProfile); + equalFilter("r.provider_id", input.providerId); + if (input.updatedAfter) { + parameters.push(input.updatedAfter); + conditions.push(`r.updated_at >= $${parameters.length}::timestamptz`); + } + if (input.search) { + parameters.push(`%${input.search.toLowerCase()}%`); + const index = parameters.length; + conditions.push(`LOWER(CONCAT_WS(E'\\n', r.id, r.project_id, r.workspace_ref->>'path', r.backend_profile, r.provider_id, r.session_ref->>'sessionId', r.session_ref->'metadata'->>'title', r.trace_sink->>'taskId', r.trace_sink->>'lane', r.failure_kind, r.failure_message)) LIKE $${index}`); + } + parameters.push(input.pageSize, (input.page - 1) * input.pageSize); + const limitIndex = parameters.length - 1; + const offsetIndex = parameters.length; + const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; + const orderBy = input.sort === "updated-asc" ? "updated_at ASC, id ASC" + : input.sort === "status-asc" ? "status ASC, updated_at DESC, id DESC" + : input.sort === "id-asc" ? "id ASC" + : "updated_at DESC, id DESC"; + const qualifiedOrderBy = orderBy.replaceAll("updated_at", "p.updated_at").replaceAll("status", "p.status").replaceAll("id", "p.id"); + const result = await this.pool.query(` + WITH filtered AS ( + SELECT r.* + FROM agentrun_runs r + ${where} + ), total AS ( + SELECT COUNT(*)::text AS total_count FROM filtered + ), facets AS ( + SELECT jsonb_build_object( + 'projects', COALESCE((SELECT jsonb_agg(project_id ORDER BY project_id) FROM (SELECT DISTINCT project_id FROM agentrun_runs) values_by_project), '[]'::jsonb), + 'workspaces', COALESCE((SELECT jsonb_agg(workspace ORDER BY workspace) FROM (SELECT DISTINCT workspace_ref->>'path' AS workspace FROM agentrun_runs WHERE workspace_ref->>'path' IS NOT NULL) values_by_workspace), '[]'::jsonb), + 'backends', COALESCE((SELECT jsonb_agg(backend_profile ORDER BY backend_profile) FROM (SELECT DISTINCT backend_profile FROM agentrun_runs) values_by_backend), '[]'::jsonb), + 'providers', COALESCE((SELECT jsonb_agg(provider_id ORDER BY provider_id) FROM (SELECT DISTINCT provider_id FROM agentrun_runs) values_by_provider), '[]'::jsonb), + 'statusCounts', COALESCE((SELECT jsonb_object_agg(status, count) FROM (SELECT status, COUNT(*)::int AS count FROM agentrun_runs GROUP BY status) values_by_status), '{}'::jsonb) + ) AS value + ), paged AS ( + SELECT * FROM filtered ORDER BY ${orderBy} LIMIT $${limitIndex} OFFSET $${offsetIndex} + ) + SELECT p.*, total.total_count, facets.value AS facets, row_to_json(c.*) AS latest_command, row_to_json(j.*) AS latest_runner_job + FROM total CROSS JOIN facets + LEFT JOIN paged p ON TRUE + LEFT JOIN LATERAL ( + SELECT * FROM agentrun_commands command_row WHERE command_row.run_id = p.id ORDER BY command_row.seq DESC LIMIT 1 + ) c ON TRUE + LEFT JOIN LATERAL ( + SELECT * FROM agentrun_runner_jobs job_row WHERE job_row.run_id = p.id ORDER BY job_row.created_at DESC, job_row.id DESC LIMIT 1 + ) j ON TRUE + ORDER BY ${qualifiedOrderBy} NULLS LAST + `, parameters); + const total = Number(result.rows[0]?.total_count ?? 0); + const items = result.rows + .filter((row) => row.id !== null && row.id !== undefined) + .map((row) => runCollectionItem( + runFromRow(row), + row.latest_command ? commandFromRow(jsonRecord(row.latest_command)) : null, + row.latest_runner_job ? runnerJobFromRow(jsonRecord(row.latest_runner_job)) : null, + )); + return runCollectionResult(input, items, total, jsonRecord(result.rows[0]?.facets)); + } + async listEvents(runId: string, afterSeq: number, limit: number): Promise { await this.getRun(runId); const result = await this.pool.query("SELECT * FROM agentrun_events WHERE run_id = $1 AND seq > $2 ORDER BY seq ASC LIMIT $3", [runId, afterSeq, clamp(limit, 1, 500)]); diff --git a/src/mgr/server.ts b/src/mgr/server.ts index 5ff70dc..faa5f4a 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -1,7 +1,7 @@ import type { Server } from "node:http"; import { createServer } from "node:http"; import type { AddressInfo } from "node:net"; -import type { AgentRunStore, ListQueueTasksInput, ListSessionsInput, SessionEventPageInput } from "./store.js"; +import type { AgentRunStore, ListQueueTasksInput, ListRunsInput, ListSessionsInput, SessionEventPageInput } from "./store.js"; import { assertSessionBoundary, isTerminalCommandState, isTerminalRunStatus, openAgentRunStoreFromEnv, receivableActiveSessionTurn } from "./store.js"; import { agentRunKafkaConfig, type AgentRunKafkaConfig } from "../common/kafka-events.js"; import { AgentRunError, errorToJson } from "../common/errors.js"; @@ -922,6 +922,31 @@ async function route({ method, url, body, signal, store, sourceCommit, authSumma void emitAgentRunOtelSpan("run_created", run, process.env, { startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/runs", "http.status_code": 200, backendProfile: run.backendProfile, providerId: run.providerId } }); return run as unknown as JsonValue; } + if (method === "GET" && path === "/api/v1/runs") { + const page = positiveIntegerQuery(url, "page", 1); + const pageSize = positiveIntegerQuery(url, "pageSize", 25, 100); + const input: ListRunsInput = { page, pageSize }; + const status = url.searchParams.get("status"); + if (status) input.statuses = status.split(",").map((item) => validateRunCollectionStatus(item.trim())); + const projectId = optionalTrimmedQuery(url, "projectId"); + const workspace = optionalTrimmedQuery(url, "workspace"); + const backendProfile = optionalTrimmedQuery(url, "backendProfile"); + const providerId = optionalTrimmedQuery(url, "providerId"); + const search = optionalTrimmedQuery(url, "search"); + const updatedAfter = optionalTrimmedQuery(url, "updatedAfter"); + const sort = optionalTrimmedQuery(url, "sort"); + if (projectId) input.projectId = projectId; + if (workspace) input.workspace = workspace; + if (backendProfile) input.backendProfile = backendProfile; + if (providerId) input.providerId = providerId; + if (search) input.search = search; + if (updatedAfter) { + if (!Number.isFinite(Date.parse(updatedAfter))) throw new AgentRunError("schema-invalid", "updatedAfter must be an ISO timestamp", { httpStatus: 400 }); + input.updatedAfter = updatedAfter; + } + if (sort) input.sort = validateRunCollectionSort(sort); + return await store.listRuns(input) as unknown as JsonValue; + } const runMatch = path.match(/^\/api\/v1\/runs\/([^/]+)$/u); if (method === "GET" && runMatch) return await store.getRun(runMatch[1] ?? "") as unknown as JsonValue; const eventMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/events$/u); @@ -1476,6 +1501,21 @@ function positiveIntegerQuery(url: URL, key: string, fallback?: number, maximum return value; } +function optionalTrimmedQuery(url: URL, key: string): string | undefined { + const value = url.searchParams.get(key)?.trim(); + return value ? value : undefined; +} + +function validateRunCollectionStatus(value: string): NonNullable[number] { + if (["pending", "claimed", "running", "completed", "failed", "blocked", "cancelled"].includes(value)) return value as NonNullable[number]; + throw new AgentRunError("schema-invalid", `unsupported run status: ${value}`, { httpStatus: 400 }); +} + +function validateRunCollectionSort(value: string): NonNullable { + if (["updated-desc", "updated-asc", "status-asc", "id-asc"].includes(value)) return value as NonNullable; + throw new AgentRunError("schema-invalid", `unsupported run collection sort: ${value}`, { httpStatus: 400 }); +} + async function waitForEventsPage(input: { store: AgentRunStore; runId: string; afterSeq: number; expect: number; timeoutMs: number; limit: number; signal: AbortSignal }): Promise<{ items: RunEvent[]; timedOut: boolean; observed: number; nextAfterSeq: number; hasMore: boolean; completionReason: "expect-reached" | "run-terminal" | "command-terminal" | "timeout" }> { const deadline = Date.now() + input.timeoutMs; while (true) { diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 2ee6790..bb6f361 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -78,6 +78,7 @@ export interface AgentRunStore { health(): MaybePromise; createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise; getRun(runId: string): MaybePromise; + listRuns(input: ListRunsInput): MaybePromise; listEvents(runId: string, afterSeq: number, limit: number): MaybePromise; countEvents(runId: string, afterSeq: number): MaybePromise; waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted">; @@ -144,6 +145,62 @@ export interface CreateRunIdentity { id: string; } +export interface ListRunsInput { + page: number; + pageSize: number; + statuses?: RunRecord["status"][]; + projectId?: string; + workspace?: string; + backendProfile?: string; + providerId?: string; + search?: string; + updatedAfter?: string; + sort?: "updated-desc" | "updated-asc" | "status-asc" | "id-asc"; +} + +export interface RunCollectionItem extends JsonRecord { + id: string; + taskId: string | null; + taskStatus: string; + title: string | null; + commandId: string | null; + commandState: string | null; + commandType: string | null; + attemptId: string | null; + attemptStatus: string; + runnerId: string | null; + runnerJobId: string | null; + sessionId: string | null; + traceId: string | null; + projectId: string; + workspace: string | null; + backend: string; + providerId: string; + nodeId: string; + lane: string | null; + runStatus: string; + terminalStatus: string | null; + phase: string; + result: string | null; + errorCode: string | null; + lastMessage: string | null; + startedAt: string; + finishedAt: string | null; + updatedAt: string; +} + +export interface RunCollectionResult extends JsonRecord { + items: RunCollectionItem[]; + page: number; + pageSize: number; + total: number; + totalPages: number; + hasNextPage: boolean; + facets: JsonRecord; + filters: JsonRecord; + valuesPrinted: false; +} + export interface ListQueueTasksInput { queue?: string; state?: QueueTaskState; @@ -177,6 +234,119 @@ export interface RecordQueueRetryAttemptFailureInput { failureMessage: string; } +export function runCollectionResult(input: ListRunsInput, items: RunCollectionItem[], total: number, facets: JsonRecord = {}): RunCollectionResult { + const totalPages = Math.ceil(total / input.pageSize); + return { + items, + page: input.page, + pageSize: input.pageSize, + total, + totalPages, + hasNextPage: input.page < totalPages, + facets, + filters: { + statuses: input.statuses ?? null, + projectId: input.projectId ?? null, + workspace: input.workspace ?? null, + backendProfile: input.backendProfile ?? null, + providerId: input.providerId ?? null, + search: input.search ?? null, + updatedAfter: input.updatedAfter ?? null, + sort: input.sort ?? "updated-desc", + }, + valuesPrinted: false, + }; +} + +export function runCollectionItem(run: RunRecord, command: CommandRecord | null, runnerJob: RunnerJobRecord | null): RunCollectionItem { + const traceSink = jsonObject(run.traceSink); + const commandPayload = command?.payload ?? null; + const runnerResult = runnerJob?.result ?? null; + const terminal = isTerminalRunStatus(run.status); + const terminalStatus = run.terminalStatus ?? (terminal ? run.status as TerminalStatus : null); + return { + id: run.id, + taskId: textField(traceSink, "taskId"), + taskStatus: run.status, + title: textField(run.sessionRef?.metadata ?? null, "title"), + commandId: command?.id ?? null, + commandState: command?.state ?? null, + commandType: command?.type ?? null, + attemptId: runnerJob?.attemptId ?? null, + attemptStatus: textField(runnerResult, "status") ?? textField(runnerResult, "terminalStatus") ?? run.status, + runnerId: runnerJob?.runnerId ?? run.claimedBy, + runnerJobId: runnerJob?.id ?? null, + sessionId: run.sessionRef?.sessionId ?? null, + traceId: textField(commandPayload, "traceId") ?? textField(traceSink, "traceId"), + projectId: run.projectId, + workspace: run.workspaceRef.path ?? null, + backend: run.backendProfile, + providerId: run.providerId, + nodeId: run.providerId, + lane: textField(traceSink, "lane"), + runStatus: run.status, + terminalStatus, + phase: textField(runnerResult, "phase") ?? (terminal ? `terminal:${terminalStatus}` : run.status), + result: terminalStatus, + errorCode: run.failureKind ?? textField(runnerResult, "failureKind"), + lastMessage: run.failureMessage ?? textField(runnerResult, "failureMessage") ?? textField(runnerResult, "message"), + startedAt: run.createdAt, + finishedAt: terminal ? run.updatedAt : null, + updatedAt: latestIso(run.updatedAt, command?.updatedAt, runnerJob?.updatedAt), + }; +} + +function runCollectionSearchText(run: RunRecord): string { + const traceSink = jsonObject(run.traceSink); + return [ + run.id, + run.projectId, + run.workspaceRef.path, + run.backendProfile, + run.providerId, + run.sessionRef?.sessionId, + textField(run.sessionRef?.metadata ?? null, "title"), + textField(traceSink, "taskId"), + textField(traceSink, "lane"), + run.failureKind, + run.failureMessage, + ].filter((value): value is string => typeof value === "string").join("\n").toLowerCase(); +} + +function jsonObject(value: unknown): JsonRecord | null { + return value !== null && typeof value === "object" && !Array.isArray(value) ? value as JsonRecord : null; +} + +function textField(value: unknown, key: string): string | null { + const object = jsonObject(value); + const field = object?.[key]; + return typeof field === "string" && field.trim().length > 0 ? field.trim() : null; +} + +function latestIso(...values: Array): string { + return values.filter((value): value is string => Boolean(value)).sort().at(-1) ?? nowIso(); +} + +function compareRunCollection(left: RunRecord, right: RunRecord, sort: NonNullable): number { + if (sort === "updated-asc") return left.updatedAt.localeCompare(right.updatedAt) || left.id.localeCompare(right.id); + if (sort === "status-asc") return left.status.localeCompare(right.status) || right.updatedAt.localeCompare(left.updatedAt) || right.id.localeCompare(left.id); + if (sort === "id-asc") return left.id.localeCompare(right.id); + return right.updatedAt.localeCompare(left.updatedAt) || right.id.localeCompare(left.id); +} + +function runCollectionFacets(runs: RunRecord[]): JsonRecord { + const values = (items: Array) => [...new Set(items.filter((item): item is string => Boolean(item)))].sort(); + const statusCounts: JsonRecord = {}; + for (const run of runs) statusCounts[run.status] = Number(statusCounts[run.status] ?? 0) + 1; + return { + projects: values(runs.map((run) => run.projectId)), + workspaces: values(runs.map((run) => run.workspaceRef.path)), + backends: values(runs.map((run) => run.backendProfile)), + providers: values(runs.map((run) => run.providerId)), + statusCounts, + }; +} + export interface ListSessionsInput { state?: SessionListState; backendProfile?: BackendProfile; @@ -300,6 +470,31 @@ export class MemoryAgentRunStore implements AgentRunStore { return run; } + listRuns(input: ListRunsInput): RunCollectionResult { + const needle = input.search?.trim().toLowerCase() ?? ""; + const allRuns = Array.from(this.runs.values()); + const matching = allRuns + .filter((run) => !input.statuses || input.statuses.includes(run.status)) + .filter((run) => !input.projectId || run.projectId === input.projectId) + .filter((run) => !input.workspace || run.workspaceRef.path === input.workspace) + .filter((run) => !input.backendProfile || run.backendProfile === input.backendProfile) + .filter((run) => !input.providerId || run.providerId === input.providerId) + .filter((run) => !needle || runCollectionSearchText(run).includes(needle)) + .filter((run) => !input.updatedAfter || run.updatedAt >= input.updatedAfter) + .sort((left, right) => compareRunCollection(left, right, input.sort ?? "updated-desc")); + const offset = (input.page - 1) * input.pageSize; + const items = matching.slice(offset, offset + input.pageSize).map((run) => { + const command = Array.from(this.commands.values()) + .filter((item) => item.runId === run.id) + .sort((left, right) => right.seq - left.seq)[0] ?? null; + const runnerJob = Array.from(this.runnerJobs.values()) + .filter((item) => item.runId === run.id) + .sort((left, right) => right.createdAt.localeCompare(left.createdAt))[0] ?? null; + return runCollectionItem(run, command, runnerJob); + }); + return runCollectionResult(input, items, matching.length, runCollectionFacets(allRuns)); + } + listEvents(runId: string, afterSeq: number, limit: number): RunEvent[] { this.getRun(runId); return (this.eventsByRun.get(runId) ?? []).filter((event) => event.seq > afterSeq).slice(0, Math.max(1, Math.min(limit, 500))); diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index cd66c10..763b28d 100644 --- a/src/runner/run-once.ts +++ b/src/runner/run-once.ts @@ -205,7 +205,7 @@ function requiresBundledWorkReadyTools(run: RunRecord): boolean { const toolCredentials = run.executionPolicy.secretScope.toolCredentials ?? []; if (toolCredentials.some((item) => item.tool === "unidesk-ssh" || item.tool === "github")) return true; const requiredSkills = run.resourceBundleRef?.requiredSkills ?? []; - return requiredSkills.some((item) => item.name === "unidesk-trans" || item.name === "unidesk-gh" || item.name === "unidesk-code-queue" || item.name === "unidesk-cicd" || item.name === "dad-dev"); + return requiredSkills.some((item) => item.name === "unidesk-trans" || item.name === "unidesk-gh" || item.name === "unidesk-agentrun" || item.name === "unidesk-cicd" || item.name === "dad-dev"); } function withResourceAssembly(options: RunnerOnceOptions, resourceEnv: NodeJS.ProcessEnv | undefined, initialPrompt: InitialPromptAssembly | undefined): RunnerOnceOptions { diff --git a/src/selftest/cases/10-manager-memory.ts b/src/selftest/cases/10-manager-memory.ts index 0a8e086..7fd0685 100644 --- a/src/selftest/cases/10-manager-memory.ts +++ b/src/selftest/cases/10-manager-memory.ts @@ -19,14 +19,78 @@ const selfTest: SelfTestCase = async () => { assert.ok((((health.runnerWorkReady as { requiredImageTools?: string[] } | undefined)?.requiredImageTools) ?? []).includes("npm")); assert.equal(health.secretRefs?.valuesPrinted, false); await assertManagerAuthBoundary(); + await assertRunCollectionPagination(client, store); await assertEventsLongPolling(client, store); await assertLongResultUsesTerminalAssistant(client, store); - return { name: "manager-memory", tests: ["manager-memory-lifecycle", "manager-auth-boundary", "manager-events-long-poll", "manager-result-long-trace"] }; + return { name: "manager-memory", tests: ["manager-memory-lifecycle", "manager-auth-boundary", "manager-run-collection-pagination", "manager-events-long-poll", "manager-result-long-trace"] }; } finally { await new Promise((resolve) => server.server.close(() => resolve())); } }; +async function assertRunCollectionPagination(client: ManagerClient, store: MemoryAgentRunStore): Promise { + for (const suffix of ["a", "b", "c"]) { + const run = store.createRun({ + tenantId: "unidesk", + projectId: suffix === "a" ? "pikasTech/HWLAB" : "pikasTech/agentrun", + workspaceRef: { kind: "git-worktree", path: `/tmp/agentrun-collection-${suffix}` }, + sessionRef: { sessionId: `sess_artificer_collection_${suffix}`, metadata: { title: `分页任务${suffix}` } }, + providerId: "NC01", + backendProfile: suffix === "c" ? "gpt-pika" : "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 1000, + network: "none", + secretScope: { allowCredentialEcho: false, providerCredentials: [] }, + }, + traceSink: { taskId: `qt_collection_${suffix}`, lane: "nc01-release", traceId: `trc_collection_${suffix}` }, + }, { id: `run_collection_${suffix}` }); + const command = store.createCommand(run.id, { type: "turn", payload: { prompt: suffix }, idempotencyKey: `collection-${suffix}` }); + if (suffix === "c") { + store.saveRunnerJob({ + runId: run.id, + commandId: command.id, + payloadHash: "collection-c", + attemptId: "attempt_collection_c", + runnerId: "runner_collection_c", + namespace: "agentrun-release", + jobName: "collection-c", + managerUrl: "http://127.0.0.1", + image: "runner-image", + sourceCommit: "self-test", + serviceAccountName: null, + result: { status: "running", phase: "runner:active" }, + }); + } + } + + const first = await client.get("/api/v1/runs?page=1&pageSize=2") as JsonRecord; + assert.equal(first.total, 3); + assert.equal(first.totalPages, 2); + assert.equal(first.hasNextPage, true); + assert.equal((first.items as JsonRecord[]).length, 2); + const second = await client.get("/api/v1/runs?page=2&pageSize=2") as JsonRecord; + assert.equal((second.items as JsonRecord[]).length, 1); + assert.equal(second.hasNextPage, false); + + const searched = await client.get(`/api/v1/runs?search=${encodeURIComponent("分页任务c")}`) as JsonRecord; + const item = (searched.items as JsonRecord[])[0]; + assert.equal(searched.total, 1); + assert.equal(item?.taskId, "qt_collection_c"); + assert.equal(item?.title, "分页任务c"); + assert.equal(item?.workspace, "/tmp/agentrun-collection-c"); + assert.equal(item?.lane, "nc01-release"); + assert.equal(item?.backend, "gpt-pika"); + assert.equal(item?.runnerId, "runner_collection_c"); + assert.equal(item?.attemptId, "attempt_collection_c"); + assert.equal(item?.commandState, "pending"); + + const filtered = await client.get("/api/v1/runs?projectId=pikasTech%2FHWLAB&backendProfile=codex") as JsonRecord; + assert.equal(filtered.total, 1); + assert.equal(((filtered.items as JsonRecord[])[0])?.id, "run_collection_a"); +} + async function assertManagerAuthBoundary(): Promise { const store = new MemoryAgentRunStore(); const server = await startManagerServer({ port: 0, host: "127.0.0.1", sourceCommit: "self-test", store, auth: { apiKey: "self-test-secret", source: "self-test" } });