Revert "Merge pull request #2562 from pikasTech/fix/agent-observer-progressive-pagination"
This reverts commitd568f9d126, reversing changes made toc3f6b0a08a.
This commit is contained in:
root
@@ -2293,26 +2293,20 @@ async function agentObserverCommand(context: any) {
|
||||
const apiKey = explicitApiKey(context.parsed, context.env);
|
||||
if (!apiKey) throw cliError("hwlab_api_key_required", "client agent-observer list requires HWLAB_API_KEY.", { credentialSource: "HWLAB_API_KEY" });
|
||||
const timeoutMs = numberOption(context.parsed.timeoutMs) ?? DEFAULT_TIMEOUT_MS;
|
||||
const page = numberOption(context.parsed.page) ?? 1;
|
||||
const pageSize = numberOption(context.parsed.pageSize) ?? numberOption(context.parsed.limit) ?? 25;
|
||||
if (!Number.isInteger(page) || page <= 0) throw cliError("invalid_page", "--page must be a positive integer.", { page });
|
||||
if (!Number.isInteger(pageSize) || pageSize <= 0 || pageSize > 100) throw cliError("invalid_page_size", "--page-size must be between 1 and 100.", { pageSize });
|
||||
const query = new URLSearchParams({ page: String(page), pageSize: String(pageSize) });
|
||||
for (const [option, parameter] of [["status", "status"], ["projectId", "projectId"], ["workspace", "workspace"], ["backendProfile", "backendProfile"], ["providerId", "providerId"], ["search", "search"], ["updatedAfter", "updatedAfter"], ["sort", "sort"]] as const) {
|
||||
const value = text(context.parsed[option]);
|
||||
if (value) query.set(parameter, value);
|
||||
}
|
||||
const limit = numberOption(context.parsed.limit) ?? 100;
|
||||
if (!Number.isInteger(limit) || limit <= 0) throw cliError("invalid_limit", "--limit must be a positive integer.", { limit });
|
||||
const observation = await collectAgentObserverRuns({
|
||||
fetchImpl: context.fetchImpl,
|
||||
url: `${endpoint.baseUrl}/v1/agent-observer/runs?${query.toString()}`,
|
||||
url: `${endpoint.baseUrl}/v1/agent-observer/events`,
|
||||
apiKey,
|
||||
timeoutMs,
|
||||
limit
|
||||
});
|
||||
return ok("client.agent-observer.list", {
|
||||
baseUrl: endpoint.baseUrl,
|
||||
runtimeEndpoint: runtimeEndpointVisibility(endpoint),
|
||||
route: route("GET", "/v1/agent-observer/runs"),
|
||||
authority: "agentrun-manager-durable-store",
|
||||
route: route("GET", "/v1/agent-observer/events"),
|
||||
authority: "hwlab.event.v1",
|
||||
...observation,
|
||||
valuesPrinted: false
|
||||
});
|
||||
@@ -2322,9 +2316,9 @@ function agentObserverHelp() {
|
||||
return ok("client.agent-observer.help", {
|
||||
serviceRuntime: false,
|
||||
imagePublished: false,
|
||||
commands: ["list [--page 1] [--page-size 25] [--status <status[,status]>] [--project-id <id>] [--workspace <path>] [--backend-profile <profile>] [--provider-id <id>] [--search <text>] [--updated-after <iso>] [--sort updated-desc|updated-asc|status-asc|id-asc] [--timeout-ms 30000]"],
|
||||
commands: ["list [--limit 100] [--timeout-ms 30000]"],
|
||||
auth: "HWLAB_API_KEY",
|
||||
authority: "agentrun-manager-durable-store"
|
||||
authority: "hwlab.event.v1"
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -1,54 +1,151 @@
|
||||
type FetchLike = typeof fetch;
|
||||
|
||||
export async function collectAgentObserverRuns({ fetchImpl, url, apiKey, timeoutMs }: {
|
||||
type ObserverRun = {
|
||||
runId: string;
|
||||
taskId: string | null;
|
||||
commandId: string | null;
|
||||
sessionId: string | null;
|
||||
projectId: string | null;
|
||||
workspace: string | null;
|
||||
lane: string | null;
|
||||
backend: string | null;
|
||||
providerId: string | null;
|
||||
status: string | null;
|
||||
lastEventAt: string | null;
|
||||
};
|
||||
|
||||
export async function collectAgentObserverRuns({ fetchImpl, url, apiKey, timeoutMs, limit }: {
|
||||
fetchImpl: FetchLike;
|
||||
url: string;
|
||||
apiKey: string;
|
||||
timeoutMs: number;
|
||||
limit: number;
|
||||
}) {
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), timeoutMs);
|
||||
const startedAt = Date.now();
|
||||
const runs = new Map<string, ObserverRun>();
|
||||
let replay: unknown = null;
|
||||
let phase = "replay";
|
||||
let eventCount = 0;
|
||||
try {
|
||||
const response = await fetchImpl(url, {
|
||||
method: "GET",
|
||||
headers: { accept: "application/json", authorization: `Bearer ${apiKey}` },
|
||||
signal: controller.signal,
|
||||
headers: { accept: "text/event-stream", authorization: `Bearer ${apiKey}` },
|
||||
signal: controller.signal
|
||||
});
|
||||
const body = await response.text();
|
||||
let payload: any = null;
|
||||
try { payload = body ? JSON.parse(body) : null; } catch { /* handled below */ }
|
||||
if (!response.ok || payload?.ok === false) {
|
||||
throw observerError(
|
||||
text(payload?.error?.code) ?? "agent_observer_request_failed",
|
||||
text(payload?.error?.message) ?? `GET /v1/agent-observer/runs returned HTTP ${response.status}.`,
|
||||
{ httpStatus: response.status, bodyBytes: Buffer.byteLength(body) },
|
||||
);
|
||||
if (!response.ok) {
|
||||
const body = await response.text();
|
||||
throw observerError("agent_observer_request_failed", `GET /v1/agent-observer/events returned HTTP ${response.status}.`, { httpStatus: response.status, bodyBytes: Buffer.byteLength(body) });
|
||||
}
|
||||
if (!payload || !Array.isArray(payload.items)) {
|
||||
throw observerError("agent_observer_response_invalid", "Agent observer run collection did not return an items array.");
|
||||
const reader = response.body?.getReader();
|
||||
if (!reader) throw observerError("agent_observer_stream_missing", "Agent observer response did not provide an SSE body.");
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
let connected = false;
|
||||
while (!connected) {
|
||||
const chunk = await reader.read();
|
||||
if (chunk.done) break;
|
||||
buffer += decoder.decode(chunk.value, { stream: true });
|
||||
const frames = splitSseFrames(buffer);
|
||||
buffer = frames.remainder;
|
||||
for (const frame of frames.complete) {
|
||||
const parsed = parseSseFrame(frame);
|
||||
if (!parsed) continue;
|
||||
if (parsed.event === "hwlab.event.v1") {
|
||||
eventCount += 1;
|
||||
mergeRun(runs, parsed.data);
|
||||
} else if (parsed.event === "agent-observer.handoff") {
|
||||
phase = "handoff";
|
||||
replay = objectValue(parsed.data)?.replay ?? replay;
|
||||
} else if (parsed.event === "agent-observer.connected") {
|
||||
phase = "live";
|
||||
replay = objectValue(parsed.data)?.replay ?? replay;
|
||||
connected = true;
|
||||
break;
|
||||
} else if (parsed.event === "agent-observer.error") {
|
||||
const error = objectValue(parsed.data);
|
||||
throw observerError(text(error?.code) || "agent_observer_stream_failed", text(error?.message) || "Agent observer SSE reported an error.");
|
||||
}
|
||||
}
|
||||
}
|
||||
return {
|
||||
...payload,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
returnedRunCount: payload.items.length,
|
||||
truncated: payload.hasNextPage === true,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
await reader.cancel().catch(() => undefined);
|
||||
if (!connected) throw observerError("agent_observer_handoff_incomplete", "Agent observer SSE ended before retention-to-live handoff completed.");
|
||||
} catch (error: any) {
|
||||
if (controller.signal.aborted && error?.code === undefined) {
|
||||
throw observerError("agent_observer_timeout", `Agent observer run collection exceeded ${timeoutMs}ms.`, { timeoutMs });
|
||||
}
|
||||
if (controller.signal.aborted && error?.code === undefined) throw observerError("agent_observer_timeout", `Agent observer did not complete replay within ${timeoutMs}ms.`, { timeoutMs });
|
||||
throw error;
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
controller.abort();
|
||||
}
|
||||
|
||||
const allRuns = [...runs.values()].sort((left, right) => String(right.lastEventAt ?? "").localeCompare(String(left.lastEventAt ?? "")));
|
||||
return {
|
||||
phase,
|
||||
replay,
|
||||
eventCount,
|
||||
runCount: allRuns.length,
|
||||
returnedRunCount: Math.min(allRuns.length, limit),
|
||||
truncated: allRuns.length > limit,
|
||||
runs: allRuns.slice(0, limit),
|
||||
valuesPrinted: false
|
||||
};
|
||||
}
|
||||
|
||||
function text(value: unknown): string | null {
|
||||
function mergeRun(runs: Map<string, ObserverRun>, value: unknown) {
|
||||
const envelope = objectValue(value);
|
||||
const event = objectValue(envelope?.event);
|
||||
const context = objectValue(envelope?.context);
|
||||
const runId = firstText(envelope?.runId, context?.runId, event?.runId);
|
||||
if (!runId) return;
|
||||
const previous = runs.get(runId);
|
||||
runs.set(runId, {
|
||||
runId,
|
||||
taskId: firstText(event?.taskId, previous?.taskId),
|
||||
commandId: firstText(envelope?.commandId, context?.commandId, event?.commandId, previous?.commandId),
|
||||
sessionId: firstText(envelope?.sessionId, envelope?.hwlabSessionId, event?.sessionId, previous?.sessionId),
|
||||
projectId: firstText(context?.projectId, event?.projectId, previous?.projectId),
|
||||
workspace: firstText(event?.workspace, previous?.workspace),
|
||||
lane: firstText(event?.lane, previous?.lane),
|
||||
backend: firstText(event?.backend, previous?.backend),
|
||||
providerId: firstText(event?.providerId, previous?.providerId),
|
||||
status: firstText(event?.runStatus, event?.status, previous?.status),
|
||||
lastEventAt: firstText(event?.createdAt, envelope?.producedAt, previous?.lastEventAt)
|
||||
});
|
||||
}
|
||||
|
||||
function splitSseFrames(value: string) {
|
||||
const normalized = value.replace(/\r\n/gu, "\n");
|
||||
const parts = normalized.split("\n\n");
|
||||
return { complete: parts.slice(0, -1), remainder: parts.at(-1) ?? "" };
|
||||
}
|
||||
|
||||
function parseSseFrame(frame: string) {
|
||||
let event = "message";
|
||||
const data = [];
|
||||
for (const line of frame.split("\n")) {
|
||||
if (line.startsWith("event:")) event = line.slice(6).trim();
|
||||
if (line.startsWith("data:")) data.push(line.slice(5).trimStart());
|
||||
}
|
||||
if (data.length === 0) return null;
|
||||
try { return { event, data: JSON.parse(data.join("\n")) }; } catch { return null; }
|
||||
}
|
||||
|
||||
function objectValue(value: any): any {
|
||||
return value && typeof value === "object" && !Array.isArray(value) ? value : null;
|
||||
}
|
||||
|
||||
function text(value: unknown) {
|
||||
return typeof value === "string" && value.trim() ? value.trim() : null;
|
||||
}
|
||||
|
||||
function firstText(...values: unknown[]) {
|
||||
for (const value of values) {
|
||||
const result = text(value);
|
||||
if (result) return result;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function observerError(code: string, message: string, details: Record<string, unknown> = {}) {
|
||||
return Object.assign(new Error(message), { code, details: { ...details, valuesPrinted: false } });
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user