pikasTech-agentrun/src/selftest/cases/75-queue-q2-dispatch.ts

import assert from "node:assert/strict";
import { chmod, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
import { startManagerServer } from "../../mgr/server.js";
import { MemoryAgentRunStore } from "../../mgr/store.js";
import { ManagerClient } from "../../mgr/client.js";
import type { JsonRecord, QueueDispatchResult, QueueTaskRecord } from "../../common/types.js";
import { assertNoSecretLeak, type SelfTestCase } from "../harness.js";

const selfTest: SelfTestCase = async (context) => {
  const fakeKubectl = path.join(context.tmp, "fake-kubectl-queue-q2.js");
  const createdManifest = path.join(context.tmp, "created-queue-q2-runner-job.json");
  await writeFile(fakeKubectl, `#!/usr/bin/env bun
const chunks = [];
for await (const chunk of Bun.stdin.stream()) chunks.push(chunk);
const text = Buffer.concat(chunks.map((chunk) => Buffer.from(chunk))).toString("utf8");
await Bun.write(${JSON.stringify(createdManifest)}, text);
const manifest = JSON.parse(text);
console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kind, metadata: { uid: "job-uid-queue-q2", resourceVersion: "1", name: manifest.metadata.name, namespace: manifest.metadata.namespace } }));
`);
  await chmod(fakeKubectl, 0o755);
  const store = new MemoryAgentRunStore();
  const server = await startManagerServer({
    port: 0,
    host: "127.0.0.1",
    sourceCommit: "self-test",
    store,
    runnerJobDefaults: {
      namespace: "agentrun-v01",
      managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080",
      image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111",
      kubectlCommand: fakeKubectl,
    },
  });
  try {
    const client = new ManagerClient(server.baseUrl);
    const created = await client.post("/api/v1/queue/tasks", {
      tenantId: "unidesk",
      projectId: "pikasTech/unidesk",
      queue: "dev",
      lane: "q2",
      title: "Q2 queue dispatch task",
      priority: 20,
      backendProfile: "codex",
      providerId: "G14",
      workspaceRef: { kind: "host-path", path: context.workspace },
      executionPolicy: {
        sandbox: "workspace-write",
        approval: "never",
        timeoutMs: 15_000,
        network: "default",
        secretScope: { allowCredentialEcho: false, providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v01-provider-codex", keys: ["auth.json", "config.toml"], mountPath: context.codexHome } }] },
      },
      resourceBundleRef: null,
      payload: { prompt: "queue dispatch hello" },
      references: [{ kind: "issue", url: "https://github.com/pikasTech/agentrun/issues/39" }],
      metadata: { source: "queue-q2-self-test" },
      idempotencyKey: "queue-q2-dispatch-self-test",
    }) as QueueTaskRecord;
    const dispatched = await client.post(`/api/v1/queue/tasks/${created.id}/dispatch`, { attemptId: "attempt_queue_q2_selftest" }) as QueueDispatchResult;
    assert.equal(dispatched.action, "queue-dispatch");
    assert.equal(dispatched.mutation, true);
    assert.equal(dispatched.latestAttempt.attemptId, "attempt_queue_q2_selftest");
    assert.equal(dispatched.latestAttempt.runId, dispatched.run.id);
    assert.equal(dispatched.latestAttempt.commandId, dispatched.command.id);
    assert.ok(dispatched.latestAttempt.runnerJobId);
    assert.equal(dispatched.task.state, "running");
    assert.equal(dispatched.task.latestAttempt?.attemptId, "attempt_queue_q2_selftest");
    assert.equal(dispatched.task.sessionPath, null);

    const shown = await client.get(`/api/v1/queue/tasks/${created.id}`) as QueueTaskRecord;
    assert.equal(shown.state, "running");
    assert.equal(shown.latestAttempt?.runId, dispatched.run.id);
    assert.equal(shown.latestAttempt?.commandId, dispatched.command.id);
    assert.equal(shown.latestAttempt?.runnerJobId, dispatched.latestAttempt.runnerJobId);

    const jobs = await client.get(`/api/v1/runs/${dispatched.run.id}/runner-jobs?commandId=${dispatched.command.id}`) as { items?: JsonRecord[]; count?: number };
    assert.equal(jobs.count, 1);
    assert.equal(jobs.items?.[0]?.attemptId, "attempt_queue_q2_selftest");
    const events = await client.get(`/api/v1/runs/${dispatched.run.id}/events?afterSeq=0&limit=100`) as { items?: JsonRecord[] };
    assert.ok(events.items?.some((item) => ((item.payload as JsonRecord).phase) === "queue-dispatched"));
    await assert.rejects(
      () => client.post(`/api/v1/queue/tasks/${created.id}/dispatch`, { attemptId: "attempt_queue_q2_duplicate" }),
      (error) => error instanceof Error && error.message.includes("not pending"),
    );
    await client.patch(`/api/v1/commands/${dispatched.command.id}/status`, { terminalStatus: "completed", failureKind: null, failureMessage: null });
    const refreshed = await client.post(`/api/v1/queue/tasks/${created.id}/refresh`, {}) as QueueTaskRecord;
    assert.equal(refreshed.state, "completed");
    assert.equal(refreshed.latestAttempt?.state, "completed");
    assert.equal(refreshed.latestAttempt?.runId, dispatched.run.id);
    const manifest = JSON.parse(await readFile(createdManifest, "utf8")) as JsonRecord;
    assert.ok(JSON.stringify(manifest).includes(dispatched.run.id));
    assertNoSecretLeak(dispatched);
    return { name: "queue-q2-dispatch", tests: ["queue-dispatch-run-command-runner-job", "queue-refresh-from-core-status", "queue-dispatch-no-repeat"] };
  } finally {
    await new Promise<void>((resolve) => server.server.close(() => resolve()));
  }
};

export default selfTest;