agenthub/tests/autowake.test.ts

/**
 * Regression tests for TSK-0230 (HOF-0086): realtime auto-wake / auto-claim.
 *
 * An agent sitting in the work-wait — CLI `agenthub work` AND the MCP
 * `agenthub_work` path (waitForTask) — must wake with NO manual poke when
 *   (a) the architect assigns an existing open task to it (task_assign), or
 *   (b) a message arrives for it,
 * over the SSE stream AND over the polling fallback alone (SSE down).
 * With default settings the auto-claim must land within ≤10s.
 *
 * Background: TSK-0230 was assigned + messaged to kimi-ah and the waiting
 * loop did not claim it — the CEO had to poke the agent manually. These
 * tests pin the wake path so that class of failure cannot regress.
 */
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
import { mkdtempSync, rmSync } from 'fs';
import { tmpdir } from 'os';
import { join } from 'path';
import { init } from '../src/cli/commands/init.js';
import { workAgent } from '../src/cli/commands/work.js';
import { findAddressedOpenTask, type AgentContext } from '../src/cli/commands/start.js';
import { remoteClient } from '../src/cli/remoteClient.js';
import { waitForTask } from '../src/mcp/server.js';
import { startServer } from '../src/server/index.js';
import type { Task } from '../src/core/schema.js';

const AGENT = 'kimi';

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

/** Create an open implementer task that is NOT addressed to AGENT. */
async function createUnaddressedTask(serverUrl: string): Promise<string> {
  const res = await fetch(`${serverUrl}/tasks`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ title: 'someone-else: not for kimi', role: 'implementer' }),
  });
  const task = (await res.json()) as Task;
  return task.id;
}

/** The architect's task_assign: address the open task to AGENT (no claim). */
async function assignTask(serverUrl: string, id: string): Promise<void> {
  await fetch(`${serverUrl}/tasks/${id}`, {
    method: 'PATCH',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ assignedTo: AGENT }),
  });
}

async function postMessage(serverUrl: string, text: string): Promise<void> {
  await fetch(`${serverUrl}/messages`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ from: 'claude', to: AGENT, text }),
  });
}

async function getTaskViaApi(serverUrl: string, id: string): Promise<Task> {
  // GET /tasks/:id returns { task, body }, not a flat Task.
  const res = (await fetch(`${serverUrl}/tasks/${id}`).then((r) => r.json())) as { task: Task };
  return res.task;
}

/**
 * Simulate "SSE down, REST fine": requests to /events hang open without ever
 * delivering a frame (the nasty case — a dropped event on an otherwise-open
 * stream), everything else passes through to the real fetch. The wait may
 * then only wake via the polling fallback (TSK-0224).
 */
function breakSseOnly(): void {
  const realFetch = globalThis.fetch;
  globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => {
    const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
    if (new URL(url).pathname === '/events') {
      return new Promise<Response>((_resolve, reject) => {
        init?.signal?.addEventListener('abort', () =>
          reject(new DOMException('The operation was aborted.', 'AbortError')),
        );
      });
    }
    return realFetch(input, init);
  }) as typeof fetch;
}

describe('auto-wake on task_assign / message (TSK-0230)', () => {
  let cwd: string;
  let server: Awaited<ReturnType<typeof startServer>>;
  const realFetch = globalThis.fetch;

  beforeEach(async () => {
    cwd = mkdtempSync(join(tmpdir(), 'ah-autowake-'));
    init(cwd, { projectName: 'autowake-test', yes: true });
    server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
  });

  afterEach(async () => {
    globalThis.fetch = realFetch;
    await server.app.close().catch(() => undefined);
    rmSync(cwd, { recursive: true, force: true });
  });

  // ── CLI work loop ─────────────────────────────────────────────────────────

  it('CLI: auto-claims an assigned task via SSE (no poke)', async () => {
    const taskId = await createUnaddressedTask(server.url);
    const workDone = workAgent({ serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer', timeoutSec: 8 });
    await sleep(200);

    const t0 = Date.now();
    await assignTask(server.url, taskId);
    await workDone;
    // SSE is live: the wake must be near-instant, far below the 10s bound.
    expect(Date.now() - t0).toBeLessThan(2_000);

    const task = await getTaskViaApi(server.url, taskId);
    expect(task.status).toBe('in_progress');
    expect(task.assignedTo).toBe(AGENT);
  }, 12_000);

  it('CLI: auto-claims an assigned task within ≤10s with SSE DOWN (polling fallback only, default interval)', async () => {
    breakSseOnly();
    const taskId = await createUnaddressedTask(server.url);
    // Default pollIntervalMs (4000) on purpose: this is the ≤10s acceptance proof.
    const workDone = workAgent({ serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer', timeoutSec: 10 });
    await sleep(300);

    const t0 = Date.now();
    await assignTask(server.url, taskId);
    await workDone;
    expect(Date.now() - t0).toBeLessThan(10_000);

    const task = await getTaskViaApi(server.url, taskId);
    expect(task.status).toBe('in_progress');
    expect(task.assignedTo).toBe(AGENT);
  }, 15_000);

  it('CLI: wakes on an incoming message with SSE DOWN (delivered, not read)', async () => {
    breakSseOnly();
    const workDone = workAgent({
      serverUrl: server.url,
      projectCwd: cwd,
      agent: AGENT,
      role: 'implementer',
      timeoutSec: 8,
      pollIntervalMs: 400,
    });
    await sleep(300);

    await postMessage(server.url, 'wake without poke (sse down)');
    await workDone; // resolves because the message woke the loop, not on timeout

    const inbox = (await fetch(`${server.url}/messages?agent=${AGENT}`).then((r) => r.json())) as Array<{
      text: string;
      status: string;
    }>;
    const m = inbox.find((x) => x.text.includes('wake without poke'));
    expect(m).toBeDefined();
    // Surfaced (unread → delivered) but NOT auto-read — a timeout would have left it unread.
    expect(m?.status).toBe('delivered');
  }, 12_000);

  // ── MCP agenthub_work path (waitForTask) ──────────────────────────────────

  /** Mimics the agenthub_work finder: addressed-open-task lookup + claim. */
  function mcpWorkFinder(ctx: AgentContext) {
    return async (url: string) => {
      ctx.serverUrl = url;
      const found = await findAddressedOpenTask(ctx);
      if (!found) return null;
      await remoteClient.claimTask(url, found.task.id, ctx.agent);
      return found.task;
    };
  }

  it('MCP: waitForTask auto-claims an assigned task via SSE (no poke)', async () => {
    const ctx: AgentContext = { serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer' };
    const taskId = await createUnaddressedTask(server.url);
    const waiting = waitForTask(server.url, mcpWorkFinder(ctx), 8);
    await sleep(200);

    const t0 = Date.now();
    await assignTask(server.url, taskId);
    const claimed = await waiting;
    // SSE is live: the wake must be near-instant, far below the 10s bound.
    expect(Date.now() - t0).toBeLessThan(2_000);

    expect(claimed?.id).toBe(taskId);
    const task = await getTaskViaApi(server.url, taskId);
    expect(task.status).toBe('in_progress');
    expect(task.assignedTo).toBe(AGENT);
  }, 12_000);

  it('MCP: waitForTask auto-claims an assigned task within ≤10s with SSE DOWN (polling fallback only)', async () => {
    breakSseOnly();
    const ctx: AgentContext = { serverUrl: server.url, projectCwd: cwd, agent: AGENT, role: 'implementer' };
    const taskId = await createUnaddressedTask(server.url);
    const waiting = waitForTask(server.url, mcpWorkFinder(ctx), 10);
    await sleep(300);

    const t0 = Date.now();
    await assignTask(server.url, taskId);
    const claimed = await waiting;
    expect(Date.now() - t0).toBeLessThan(10_000);

    expect(claimed?.id).toBe(taskId);
    const task = await getTaskViaApi(server.url, taskId);
    expect(task.status).toBe('in_progress');
  }, 15_000);

  it('MCP: waitForTask wakes on an incoming message with SSE DOWN', async () => {
    breakSseOnly();
    const finder = async (url: string) => {
      const msgs = await remoteClient.getInbox(url, AGENT, true);
      return msgs.length ? { messages: msgs } : null;
    };
    const waiting = waitForTask(server.url, finder, 8);
    await sleep(300);

    await postMessage(server.url, 'mcp message wake (sse down)');
    const hit = await waiting;

    expect(hit).not.toBeNull();
    expect(hit?.messages[0]?.text).toContain('mcp message wake');
    // listInbox read receipt: surfaced as delivered, never auto-read.
    expect(hit?.messages[0]?.status).toBe('delivered');
  }, 12_000);
});