feat(jobs): read GitHub job logs on every surface

apps/cli/src/main.test.ts

@@ -3,6 +3,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { expect, test, vi } from "vitest";
 import { startService } from "../../server/src/http.js";
+import type { JobLogRequest, JobLogs } from "../../../packages/protocol/src/jobs.js";
 
 test("machine configure settles without --wait and status prints runner capacity", async () => {
   const home = await mkdtemp(join(tmpdir(), "vectis-cli-capacity-"));
@@ -225,3 +226,191 @@ test.each([
     }
   },
 );
+
+test.each([false, true])("job logs read every retained page, with follow=%s", async (follow) => {
+  const home = await mkdtemp(join(tmpdir(), "vectis-cli-job-logs-"));
+  const server = await startService({ home });
+  const argv = process.argv;
+  const exitCode = process.exitCode;
+  const output = vi.spyOn(process.stdout, "write").mockReturnValue(true);
+  const read = vi
+    .spyOn(server.service, "jobLogs")
+    .mockResolvedValueOnce({
+      text: "first\n",
+      next: 6,
+      status: "in_progress",
+      complete: false,
+      truncated: false,
+    })
+    .mockResolvedValueOnce({
+      text: "second\n",
+      next: 13,
+      status: "completed",
+      complete: true,
+      truncated: false,
+    });
+  const interrupts = process.listenerCount("SIGINT");
+  try {
+    vi.resetModules();
+    process.argv = [
+      process.execPath,
+      "vectis",
+      "--home",
+      home,
+      "job",
+      "logs",
+      "4",
+      "--binding",
+      "binding1",
+      ...(follow ? ["--follow"] : []),
+    ];
+    await import("./main.js");
+    await vi.waitFor(() => expect(output).toHaveBeenCalledWith("second\n"), { timeout: 6000 });
+    expect(output).toHaveBeenCalledWith("first\n");
+    expect(read.mock.calls.map(([input]) => input)).toEqual([
+      { bindingId: "binding1", jobId: 4 },
+      { bindingId: "binding1", jobId: 4, after: 6 },
+    ]);
+    expect(process.listenerCount("SIGINT")).toBe(interrupts);
+    expect(server.service.snapshot().operations).toHaveLength(0);
+  } finally {
+    process.argv = argv;
+    process.exitCode = exitCode;
+    output.mockRestore();
+    read.mockRestore();
+    await server.close();
+    await rm(home, { recursive: true, force: true });
+  }
+});
+
+test("job log follow stops on interrupt and releases its polling timer and listeners", async () => {
+  const home = await mkdtemp(join(tmpdir(), "vectis-cli-follow-"));
+  const server = await startService({ home });
+  const argv = process.argv;
+  const exitCode = process.exitCode;
+  const output = vi.spyOn(process.stdout, "write").mockReturnValue(true);
+  const read = vi.spyOn(server.service, "jobLogs").mockResolvedValue({
+    text: "waiting",
+    next: 7,
+    status: "in_progress",
+    complete: false,
+    truncated: false,
+  });
+  const interrupts = process.listenerCount("SIGINT");
+  try {
+    vi.resetModules();
+    process.argv = [
+      process.execPath,
+      "vectis",
+      "--home",
+      home,
+      "job",
+      "logs",
+      "4",
+      "--binding",
+      "binding1",
+      "--follow",
+    ];
+    await import("./main.js");
+    await vi.waitFor(() => expect(output).toHaveBeenCalledWith("waiting"));
+    process.emit("SIGINT");
+    await vi.waitFor(() => expect(process.listenerCount("SIGINT")).toBe(interrupts));
+    expect(read).toHaveBeenCalledOnce();
+    expect(process.exitCode).toBe(exitCode);
+  } finally {
+    process.argv = argv;
+    process.exitCode = exitCode;
+    output.mockRestore();
+    read.mockRestore();
+    await server.close();
+    await rm(home, { recursive: true, force: true });
+  }
+});
+
+test.each([false, true])(
+  "job log follow backs off empty chunks and resets on text, json=%s",
+  async (json) => {
+    const home = await mkdtemp(join(tmpdir(), "vectis-cli-backoff-"));
+    const argv = process.argv;
+    const exitCode = process.exitCode;
+    const output = vi.spyOn(process.stdout, "write").mockReturnValue(true);
+    const empty: JobLogs = {
+      text: "",
+      next: 0,
+      status: "in_progress",
+      complete: false,
+      truncated: false,
+    };
+    const read = vi
+      .fn<(input: JobLogRequest, signal?: AbortSignal) => Promise<JobLogs>>()
+      .mockResolvedValue(empty);
+    for (let index = 0; index < 6; index++) read.mockResolvedValueOnce(empty);
+    read.mockResolvedValueOnce({ ...empty, text: "new output", next: 10 });
+    read.mockResolvedValueOnce({ ...empty, next: 10 });
+    read.mockResolvedValueOnce({ ...empty, next: 10, status: "completed", complete: true });
+    const interrupts = process.listenerCount("SIGINT");
+    const terminations = process.listenerCount("SIGTERM");
+    vi.resetModules();
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+    vi.doMock("../../../packages/client/src/local.js", () => ({
+      localClient: async (selectedHome: string) => {
+        expect(selectedHome).toBe(home);
+        return { jobLogs: read };
+      },
+    }));
+    vi.doMock("node:timers/promises", () => ({
+      setTimeout: (milliseconds: number) =>
+        new Promise<void>((resolve) => setTimeout(resolve, milliseconds)),
+    }));
+    try {
+      process.argv = [
+        process.execPath,
+        "vectis",
+        "--home",
+        home,
+        "job",
+        "logs",
+        "4",
+        "--binding",
+        "binding1",
+        "--follow",
+        ...(json ? ["--json"] : []),
+      ];
+      await import("./main.js");
+      await vi.advanceTimersByTimeAsync(0);
+      expect(read).toHaveBeenCalledOnce();
+      const delays = [2000, 4000, 8000, 16000, 30000, 30000, 2000, 2000];
+      for (const [index, delay] of delays.entries()) {
+        await vi.advanceTimersByTimeAsync(delay - 1);
+        expect(read).toHaveBeenCalledTimes(index + 1);
+        await vi.advanceTimersByTimeAsync(1);
+        expect(read).toHaveBeenCalledTimes(index + 2);
+      }
+      expect(read.mock.calls.map(([input]) => input)).toEqual([
+        { bindingId: "binding1", jobId: 4 },
+        ...Array.from({ length: 6 }, () => ({ bindingId: "binding1", jobId: 4, after: 0 })),
+        ...Array.from({ length: 2 }, () => ({ bindingId: "binding1", jobId: 4, after: 10 })),
+      ]);
+      expect(output).toHaveBeenCalledWith(
+        json ? expect.stringContaining('"text":"new output"') : "new output",
+      );
+      expect(process.listenerCount("SIGINT")).toBe(interrupts);
+      expect(process.listenerCount("SIGTERM")).toBe(terminations);
+      expect(vi.getTimerCount()).toBe(0);
+      await vi.advanceTimersByTimeAsync(60000);
+      expect(read).toHaveBeenCalledTimes(9);
+      expect(process.exitCode).toBe(exitCode);
+    } finally {
+      if (process.listenerCount("SIGINT") > interrupts) process.emit("SIGINT");
+      vi.clearAllTimers();
+      vi.useRealTimers();
+      vi.doUnmock("node:timers/promises");
+      vi.doUnmock("../../../packages/client/src/local.js");
+      vi.resetModules();
+      process.argv = argv;
+      process.exitCode = exitCode;
+      output.mockRestore();
+      await rm(home, { recursive: true, force: true });
+    }
+  },
+);

apps/cli/src/main.ts

@@ -1,4 +1,6 @@
 #!/usr/bin/env node
+import { setTimeout as delay } from "node:timers/promises";
+import { JobLogRequest } from "../../../packages/protocol/src/jobs.js";
 import { parseArgs } from "node:util";
 import { readFile, mkdir, open } from "node:fs/promises";
 import { spawn } from "node:child_process";
@@ -74,6 +76,8 @@ async function main() {
       "memory-mib": { type: "string" },
       "storage-path": { type: "string" },
       map: { type: "string", multiple: true },
+      binding: { type: "string" },
+      follow: { type: "boolean" },
       lines: { type: "string" },
       limit: { type: "string" },
       kind: { type: "string" },
@@ -150,6 +154,7 @@ Repositories and jobs
   repository access <binding-id> --actors any|<id,id,...> --approval automatic|required
                                Actor IDs are numeric GitHub user IDs: gh api users/<login> --jq .id
   job list <binding-id>         Read recent GitHub job states for a connected repository
+  job logs <job-id> --binding <id> [--follow]  Read the retained log; follow backs off from 2 to 30 seconds when empty
   job scan <binding-id>         Discover missing jobs through the GitHub API
   job refresh <binding-id> <job-id>  Recover one job state directly from GitHub
 
@@ -465,6 +470,46 @@ Documentation: https://vectis.kerd.dev/docs
     output(await api.jobs(id));
     return;
   }
+  if (command === "job" && subcommand === "logs") {
+    const input = { bindingId: values.binding, jobId: Number(id) };
+    if (!Schema.is(JobLogRequest)(input))
+      throw new VectisError(
+        "invalid_argument",
+        "Use job logs <job-id> --binding <binding-id> with a positive whole job ID.",
+      );
+    const abort = new AbortController();
+    const stop = () => abort.abort();
+    process.once("SIGINT", stop);
+    process.once("SIGTERM", stop);
+    let after: number | undefined;
+    let pollDelay = 2000;
+    try {
+      while (!abort.signal.aborted) {
+        const log = await api.jobLogs(
+          { ...input, ...(after === undefined ? {} : { after }) },
+          abort.signal,
+        );
+        if (values.json) output(log);
+        else process.stdout.write(log.text);
+        if (log.truncated)
+          process.stderr.write("Earlier job log content was skipped (4 MiB retained window).\n");
+        after = log.next;
+        if (log.complete) break;
+        if (!log.text && !values.follow) break;
+        if (values.follow) {
+          if (log.text) pollDelay = 2000;
+          await delay(pollDelay, undefined, { signal: abort.signal });
+          if (!log.text) pollDelay = Math.min(pollDelay * 2, 30000);
+        }
+      }
+    } catch (error) {
+      if (!abort.signal.aborted) throw error;
+    } finally {
+      process.removeListener("SIGINT", stop);
+      process.removeListener("SIGTERM", stop);
+    }
+    return;
+  }
   if (command === "repository" && subcommand === "list") {
     output(await api.repositories());
     return;

apps/desktop/src/bridge.ts

@@ -11,6 +11,7 @@ export type DesktopAction =
   | "repositories"
   | "github.accounts"
   | "jobs"
+  | "job.logs"
   | "command"
   | "command.settle"
   | "service.install"

apps/desktop/src/main.ts

@@ -1,3 +1,4 @@
+import { JobLogRequest } from "../../../packages/protocol/src/jobs.js";
 import { existsSync } from "node:fs";
 import { rm, writeFile } from "node:fs/promises";
 import electronUpdater from "electron-updater";
@@ -197,6 +198,13 @@ else {
               data = await (await client()).jobs(bindingId);
               break;
             }
+            case "job.logs": {
+              const request = Schema.decodeUnknownSync(JobLogRequest, {
+                onExcessProperty: "error",
+              })(input);
+              data = await (await client()).jobLogs(request);
+              break;
+            }
             case "github.accounts":
               data = await (await client()).githubAccounts();
               break;

apps/desktop/src/renderer/components/activity-row.tsx

@@ -1,6 +1,8 @@
+import { useState } from "react";
 import { ExternalLinkIcon } from "lucide-react";
 import type { Activity, Operation } from "../../../../../packages/protocol/src/index.js";
 import { GitHubIcon } from "@/components/github-icon";
+import { JobLog } from "@/components/job-log";
 import { Details, ExpandableRow, Mono } from "@/components/layout";
 import { OsTile } from "@/components/os-icon";
 import { JobStatus, OperationStatus, RunnerStage, StatusText } from "@/components/status";
@@ -39,6 +41,7 @@ export function ActivityRow({
       ? steps.at(-1)
       : steps.find((step) => step.id === activity.rootOperationId);
   const job = activityJob(activity);
+  const [logOpen, setLogOpen] = useState(false);
   // A runner that did its part is not news once GitHub reports the job; the job outcome is.
   const runnerQuiet = activity.status === "succeeded" && job?.status !== undefined;
   const running = activity.status === "accepted" || activity.status === "running";
@@ -117,6 +120,23 @@ export function ActivityRow({
           </pre>
         </div>
       )}
+      {job && activity.bindingId && (
+        <div className="flex flex-col gap-3">
+          <div>
+            <Button
+              variant="secondary"
+              size="sm"
+              aria-expanded={logOpen}
+              onClick={() => setLogOpen((value) => !value)}
+            >
+              {logOpen ? "Hide job log" : "View job log"}
+            </Button>
+          </div>
+          {logOpen && (
+            <JobLog bindingId={activity.bindingId} jobId={job.jobId} htmlUrl={job.htmlUrl} />
+          )}
+        </div>
+      )}
       {job?.htmlUrl && (
         <div>
           <Button