feat(fleet): port observe-agent fleet commands from Go CLI

AGENTS.md

@@ -42,6 +42,7 @@ src/
 │   │   └── tracing/    # Trace Explorer (install, view)
 │   ├── dataset/        # Dataset commands (list, view)
 │   ├── datastream/     # Datastream commands (create, list, view, update)
+│   ├── fleet/          # observe-agent fleet commands (status, host, versions, auth)
 │   ├── ingest-token/   # Ingest token commands (create, list, view, update)
 │   ├── metric/         # Metric commands (list, view)
 │   ├── skill/          # AI agent skill commands (list, view)

README.md

@@ -74,6 +74,10 @@ To update installed skills after edits in this repo, run `npx skills update`.
 | `observe datastream view`               | View a datastream by ID                                 |
 | `observe datastream update`             | Update a datastream                                     |
 | `observe datastream-token check-status` | Poll a datastream token until ingest data arrives       |
+| `observe fleet status`                  | Current observe-agent inventory (newest first)          |
+| `observe fleet host`                    | Event history for one host, incl. agent start time      |
+| `observe fleet versions`                | Agent version distribution across the fleet             |
+| `observe fleet auth`                    | Agent auth-check status (failures first)                |
 | `observe cli install`                   | Configure shell integration (PATH, completions)         |
 | `observe cli uninstall`                 | Remove shell integration                                |
 | `observe cli upgrade`                   | Upgrade to the latest version                           |

src/app.ts

@@ -9,6 +9,7 @@ import { cliRoutes } from "./commands/cli/index.js";
 import { contentRoutes } from "./commands/content/index.js";
 import { datasetRoutes } from "./commands/dataset/index.js";
 import { datastreamRoutes } from "./commands/datastream/index.js";
+import { fleetRoutes } from "./commands/fleet/index.js";
 import { helpCommand } from "./commands/help.js";
 import { ingestTokenRoutes } from "./commands/ingest-token/index.js";
 import { metricRoutes } from "./commands/metric/index.js";
@@ -36,6 +37,7 @@ export const routes = buildRouteMap({
     "data-connection": dataConnectionRoutes,
     datastream: datastreamRoutes,
     "datastream-token": datastreamTokenRoutes,
+    fleet: fleetRoutes,
     cli: cliRoutes,
   },
   defaultCommand: "help",

src/commands/fleet/auth.ts

@@ -0,0 +1,70 @@
+import { buildCommand } from "@stricli/core";
+import type { LocalContext } from "../../context";
+import { loadConfig } from "../../lib/config";
+import { formatFleetError } from "./format-error";
+import { muteStatusWriter } from "../../lib/writer";
+import { runFleetQuery, type RunFleetQueryDeps } from "./run-query";
+import { writeFleetResult, type FleetOutputFormat } from "./format-output";
+import { FLEET_AUTH_PIPELINE } from "./pipelines";
+
+const DEFAULT_WINDOW = "20m";
+
+interface FleetAuthFlags {
+  window: string;
+  format?: Exclude<FleetOutputFormat, "table">;
+}
+
+export interface FleetAuthDeps extends RunFleetQueryDeps {
+  loadConfig?: typeof loadConfig;
+}
+
+export async function auth(
+  this: LocalContext,
+  flags: FleetAuthFlags,
+  deps: FleetAuthDeps = {},
+): Promise<void> {
+  const { loadConfig: loadConfigImpl = loadConfig, datasetQueryOutput } = deps;
+  const format: FleetOutputFormat = flags.format ?? "table";
+  const { process, writer: rawWriter } = this;
+  const writer = muteStatusWriter(rawWriter, {
+    muted: format !== "table",
+  });
+
+  try {
+    const config = loadConfigImpl();
+    writer.info("Querying fleet auth status...");
+    const result = await runFleetQuery(
+      { config, pipeline: FLEET_AUTH_PIPELINE, window: flags.window },
+      { datasetQueryOutput },
+    );
+    writeFleetResult(writer, result, format);
+  } catch (error) {
+    writer.error(`fleet auth failed: ${formatFleetError(error)}`);
+    process.exit(1);
+  }
+}
+
+export const authCommand = buildCommand({
+  loader: async () => auth,
+  parameters: {
+    positional: { kind: "tuple", parameters: [] },
+    flags: {
+      window: {
+        kind: "parsed",
+        parse: String,
+        brief: "Time window for the query (Go duration, e.g. 20m, 24h, 168h)",
+        default: DEFAULT_WINDOW,
+      },
+      format: {
+        kind: "enum",
+        values: ["json", "csv"],
+        brief: "Output format (json, csv) (default: table)",
+        optional: true,
+      },
+    },
+    aliases: { w: "window" },
+  },
+  docs: {
+    brief: "Auth-check status across the fleet (failures first)",
+  },
+});

src/commands/fleet/format-error.ts

@@ -0,0 +1,18 @@
+import { GqlApiError } from "../../gql/gql-request";
+
+/**
+ * Format an error thrown from the fleet query path into a display string.
+ * Fleet talks only to the GraphQL query service, so we surface `GqlApiError`
+ * with its HTTP status and fall back to the message for everything else.
+ * (Unlike the shared `formatApiError`, this avoids the REST client so the
+ * GraphQL-only fleet commands don't pull in generated REST runtime types.)
+ */
+export function formatFleetError(error: unknown): string {
+  if (error instanceof GqlApiError) {
+    return `API Error (${error.statusCode}): ${error.message}`;
+  }
+  if (error instanceof Error) {
+    return error.message;
+  }
+  return String(error);
+}

src/commands/fleet/format-output.ts

@@ -0,0 +1,91 @@
+import type { Writer } from "../../lib/writer";
+import type { FleetQueryResult, FleetRow } from "./run-query";
+import type { GqlDatasetQueryField } from "../../gql/dataset/dataset-query-output";
+import { DataType } from "../../gql/generated/graphql";
+import { formatTable, type ColumnDef } from "../../lib/formatters/table";
+import { valueToString } from "../../lib/formatters/value";
+import { renderAsCSV } from "../../lib/formatters/csv";
+import { cyan, green, muted, red, yellow } from "../../lib/formatters/colors";
+
+export type FleetOutputFormat = "table" | "json" | "csv";
+
+/**
+ * Render a parsed fleet query result in the requested format. Column order is
+ * preserved from the OPAL `pick_col`/output schema (matching the Go CLI's
+ * table columns), and row order from the OPAL `sort`. `table` is the default
+ * human-readable view; `json` and `csv` emit the raw row records.
+ */
+export function writeFleetResult(
+  writer: Writer,
+  result: FleetQueryResult,
+  format: FleetOutputFormat,
+): void {
+  const { headers, fields, rows } = result;
+
+  if (format === "csv") {
+    writer.write(renderAsCSV(rows));
+    return;
+  }
+
+  if (format === "json") {
+    writer.write(JSON.stringify(rows, null, 2));
+    return;
+  }
+
+  if (rows.length === 0) {
+    writer.info("No results");
+    return;
+  }
+
+  const fieldMap = new Map<string, GqlDatasetQueryField>();
+  for (const field of fields) {
+    if (field.name) fieldMap.set(field.name, field);
+  }
+
+  const columns: ColumnDef<FleetRow>[] = headers.map((h) => ({
+    header: h,
+    accessorFn: (row) => row[h],
+    format: getFieldFormatter(fieldMap.get(h)),
+    maxLines: 3,
+  }));
+
+  writer.write("\n" + formatTable(rows, columns));
+  writer.info(`\n${rows.length} row(s)`);
+}
+
+/**
+ * Get a formatter for a field based on its result type, mirroring the
+ * coloring used by the `query` command so fleet tables look consistent.
+ */
+function getFieldFormatter(
+  field?: GqlDatasetQueryField,
+): ((value: unknown) => string) | undefined {
+  if (!field) return undefined;
+
+  switch (field.type.tag) {
+    case DataType.Int64:
+    case DataType.Float64:
+      return (v) => cyan(valueToString(v));
+    case DataType.Bool:
+      return (v) => (isTruthyBool(v) ? green("true") : red("false"));
+    case DataType.Timestamp:
+      return (v) => muted(valueToString(v));
+    case DataType.Object:
+    case DataType.Variant:
+    case DataType.Array:
+      return (v) => yellow(valueToString(v));
+    default:
+      return undefined;
+  }
+}
+
+/**
+ * The server encodes scalar values as strings in PaginatedResults, so "false"
+ * arrives as a truthy string. Treat the string "true" (or a real boolean) as
+ * true; everything else is false.
+ */
+function isTruthyBool(value: unknown): boolean {
+  if (typeof value === "boolean") return value;
+  if (typeof value === "string") return value.toLowerCase() === "true";
+  return Boolean(value);
+}

src/commands/fleet/host.ts

@@ -0,0 +1,79 @@
+import { buildCommand } from "@stricli/core";
+import type { LocalContext } from "../../context";
+import { loadConfig } from "../../lib/config";
+import { formatFleetError } from "./format-error";
+import { muteStatusWriter } from "../../lib/writer";
+import { runFleetQuery, type RunFleetQueryDeps } from "./run-query";
+import { writeFleetResult, type FleetOutputFormat } from "./format-output";
+import { fleetHostPipeline } from "./pipelines";
+
+const DEFAULT_WINDOW = "20m";
+
+interface FleetHostFlags {
+  window: string;
+  format?: Exclude<FleetOutputFormat, "table">;
+}
+
+export interface FleetHostDeps extends RunFleetQueryDeps {
+  loadConfig?: typeof loadConfig;
+}
+
+export async function host(
+  this: LocalContext,
+  flags: FleetHostFlags,
+  hostname: string,
+  deps: FleetHostDeps = {},
+): Promise<void> {
+  const { loadConfig: loadConfigImpl = loadConfig, datasetQueryOutput } = deps;
+  const format: FleetOutputFormat = flags.format ?? "table";
+  const { process, writer: rawWriter } = this;
+  const writer = muteStatusWriter(rawWriter, {
+    muted: format !== "table",
+  });
+
+  try {
+    const config = loadConfigImpl();
+    writer.info(`Querying fleet history for host ${hostname}...`);
+    const result = await runFleetQuery(
+      { config, pipeline: fleetHostPipeline(hostname), window: flags.window },
+      { datasetQueryOutput },
+    );
+    writeFleetResult(writer, result, format);
+  } catch (error) {
+    writer.error(`fleet host failed: ${formatFleetError(error)}`);
+    process.exit(1);
+  }
+}
+
+export const hostCommand = buildCommand({
+  loader: async () => host,
+  parameters: {
+    positional: {
+      kind: "tuple",
+      parameters: [
+        {
+          brief: "Hostname (identifiers[\"host.name\"]) to show event history for",
+          parse: String,
+        },
+      ],
+    },
+    flags: {
+      window: {
+        kind: "parsed",
+        parse: String,
+        brief: "Time window for the query (Go duration, e.g. 20m, 24h, 168h)",
+        default: DEFAULT_WINDOW,
+      },
+      format: {
+        kind: "enum",
+        values: ["json", "csv"],
+        brief: "Output format (json, csv) (default: table)",
+        optional: true,
+      },
+    },
+    aliases: { w: "window" },
+  },
+  docs: {
+    brief: "Event history for one host, including agent start time",
+  },
+});

src/commands/fleet/index.ts

@@ -0,0 +1,29 @@
+import { buildRouteMap } from "@stricli/core";
+import { statusCommand } from "./status";
+import { hostCommand } from "./host";
+import { versionsCommand } from "./versions";
+import { authCommand } from "./auth";
+
+export const fleetRoutes = buildRouteMap({
+  routes: {
+    status: statusCommand,
+    host: hostCommand,
+    versions: versionsCommand,
+    auth: authCommand,
+  },
+  docs: {
+    brief: "Query fleet status of observe-agent instances",
+    fullDescription: [
+      "Query the observe-agent fleet from the Default.Observe Agent/Events",
+      "dataset. Each command runs an OPAL query over the AgentLifecycleEvent",
+      "stream; use --window to set the lookback (Go duration, e.g. 20m, 24h,",
+      "168h; default 20m).",
+      "",
+      "Commands:",
+      "  status     Current agent inventory (newest first)",
+      "  host       Event history for one host, incl. agent start time",
+      "  versions   Version distribution across the fleet",
+      "  auth       Auth-check status (failures first)",
+    ].join("\n"),
+  },
+});

src/commands/fleet/pipelines.ts

@@ -0,0 +1,38 @@
+/**
+ * OPAL pipelines for the fleet commands, ported verbatim from the Go CLI
+ * (`cmd_fleet.go`). Each runs against the Observe Agent Events dataset and
+ * filters to `AgentLifecycleEvent` rows. The `pick_col` order defines the
+ * output columns and the trailing `sort` defines row order — both are
+ * preserved by the query path, so output matches the Go CLI.
+ */
+
+/** `fleet status` — current agent inventory, newest first. */
+export const FLEET_STATUS_PIPELINE =
+  'filter kind = "AgentLifecycleEvent" | make_col host:string(identifiers["host.name"]), env:string(identifiers["observe.agent.environment"]), version:string(facets["observe.agent.version"]), instance_id:string(identifiers["observe.agent.instance.id"]), data_obj:parse_json(data) | make_col auth_ok:bool(data_obj.authCheck.passed) | pick_col valid_from, host, env, version, auth_ok, instance_id | sort desc(valid_from)';
+
+/** `fleet versions` — version distribution across the fleet. */
+export const FLEET_VERSIONS_PIPELINE =
+  'filter kind = "AgentLifecycleEvent" | make_col host:string(identifiers["host.name"]), env:string(identifiers["observe.agent.environment"]), version:string(facets["observe.agent.version"]) | pick_col valid_from, host, env, version | sort asc(version), asc(host)';
+
+/** `fleet auth` — auth-check status, failures first. */
+export const FLEET_AUTH_PIPELINE =
+  'filter kind = "AgentLifecycleEvent" | make_col host:string(identifiers["host.name"]), env:string(identifiers["observe.agent.environment"]), version:string(facets["observe.agent.version"]), data_obj:parse_json(data) | make_col auth_ok:bool(data_obj.authCheck.passed), auth_code:int64(data_obj.authCheck.responseCode), auth_url:string(data_obj.authCheck.url) | pick_col valid_from, host, env, version, auth_ok, auth_code, auth_url | sort asc(auth_ok), desc(valid_from)';
+
+/**
+ * `fleet host <hostname>` — event history for one host, newest first,
+ * including agent start time. The hostname is interpolated into a quoted OPAL
+ * string literal, matching the Go CLI's `%q` formatting.
+ */
+export function fleetHostPipeline(hostname: string): string {
+  return `filter kind = "AgentLifecycleEvent" | filter string(identifiers["host.name"]) = ${quoteOpalString(hostname)} | make_col host:string(identifiers["host.name"]), env:string(identifiers["observe.agent.environment"]), version:string(facets["observe.agent.version"]), data_obj:parse_json(data) | make_col auth_ok:bool(data_obj.authCheck.passed), start_time:from_nanoseconds(int64(data_obj.agentStartTime)*1000000000) | pick_col valid_from, host, env, version, auth_ok, start_time | sort desc(valid_from)`;
+}
+
+/**
+ * Quote a value as an OPAL double-quoted string literal, escaping characters
+ * the way Go's `%q` does for the cases that matter here (backslash and
+ * double-quote). Prevents a hostname from breaking out of the filter.
+ */
+function quoteOpalString(value: string): string {
+  const escaped = value.replace(/\\/g, "\\\\").replace(/"/g, '\\"');
+  return `"${escaped}"`;
+}