template-app/test/agent-events.test.ts

// SPDX-License-Identifier: LicenseRef-ConusAI-Template-1.0
// `agent.run.events` through the kernel's caller, the way the HTTP mount calls it: a project's
// reader gets the run's AG-UI events as Server-Sent Events, resumes strictly after an offset, and
// the stream closes after the run's terminal event; another organization's run is not found, as
// `agent.run.status` answers, and an offset of another run is refused.

import { AppError, type TransportCaller } from "@conusai-kit/core"
import { describe, expect, it, vi } from "vitest"
import { makeNotesApp, type NotesApp, type SeededProject } from "./support/notes-app.js"

const PLAN = [{ key: "read", label: "Read", tool: "documents.read", args: { fileId: "f-1" } }]

const refusal = async (promise: Promise<unknown>): Promise<AppError> => {
  try {
    await promise
  } catch (error) {
    if (error instanceof AppError) return error
    throw error
  }
  throw new Error("expected a refusal")
}

interface Frame {
  id: string | undefined
  data: { type: string; [field: string]: unknown }
}

const framesOf = (text: string): Frame[] =>
  text
    .split("\n\n")
    .filter((frame) => frame.trim() !== "" && !frame.startsWith(":"))
    .map((frame) => ({
      id: /^id: (.*)$/m.exec(frame)?.[1],
      data: JSON.parse(/^data: (.*)$/m.exec(frame)?.[1] ?? "null") as Frame["data"],
    }))

/** The first chunk of a stream that is still open (a run that has not ended), then the stream is let go. */
async function firstChunk(response: Response): Promise<string> {
  const reader = (response.body as ReadableStream<Uint8Array>).getReader()
  const { value } = await reader.read()
  await reader.cancel()
  return new TextDecoder().decode(value)
}

async function startRun(app: NotesApp, project: SeededProject) {
  const started = (await project.call.callByName("agent.run.start", {
    ...project.slugs,
    goal: "Review",
    plan: PLAN,
  })) as {
    runId: string
  }
  // The resume event names the run's project by id: what the engine would write under.
  const record = (await app.data.outbox.pending(50)).find(
    (entry) =>
      entry.name === "agent.runResumeRequested" && (entry.payload as { runId: string }).runId === started.runId,
  )
  const ids = record?.payload as { orgId: string; workspaceId: string; projectId: string }
  const scope = { level: "project" as const, actor: project.owner, ...ids }
  /** Ends the run as the engine would. */
  const finish = async () => {
    const entity = app.entities.find((candidate) => candidate.table === "agent_run")
    if (entity?.tenancy !== "project") throw new Error("the agent module declares no project run table")
    const scoped = app.data.forScope(scope)
    const table = scoped.of(entity as Parameters<typeof scoped.of>[0])
    const row = (await table.get(started.runId)) as { version: number } | null
    if (!row) throw new Error("the run is gone")
    await table.update(started.runId, { status: "cancelled" }, row.version)
  }
  return { runId: started.runId, finish }
}

const events = (call: TransportCaller, input: Record<string, unknown>) =>
  call.callByName("agent.run.events", input) as Promise<Response>

describe("agent.run.events", () => {
  it("streams a reader the run's events, resumes after an offset and closes after the terminal event", async () => {
    const app = makeNotesApp()
    const project = await app.seedProject("acme", "owner@acme.test")
    const run = await startRun(app, project)
    const viewer = await app.addMember(project, "viewer@acme.test", "viewer")

    const open = await events(viewer.call, { ...project.slugs, runId: run.runId })
    expect(open.headers.get("content-type")).toContain("text/event-stream")
    expect(open.headers.get("cache-control")).toBe("no-store")
    expect(framesOf(await firstChunk(open))).toEqual([
      {
        id: `agk1:${run.runId}:0`,
        data: { type: "RUN_STARTED", threadId: `agent-run:${run.runId}`, runId: run.runId },
      },
    ])

    await run.finish()
    const rest = framesOf(
      await (await events(viewer.call, { ...project.slugs, runId: run.runId, after: `agk1:${run.runId}:0` })).text(),
    )
    expect(rest).toEqual([
      {
        id: `agk1:${run.runId}:1`,
        data: {
          type: "RUN_FINISHED",
          threadId: `agent-run:${run.runId}`,
          runId: run.runId,
          outcome: { type: "success" },
          result: { status: "cancelled" },
        },
      },
    ])
    expect(await (await events(viewer.call, { ...project.slugs, runId: run.runId, after: "1" })).text()).toBe("")
  })

  it("answers another organization's run as not found, and refuses another run's offset", async () => {
    const app = makeNotesApp()
    const acme = await app.seedProject("acme", "owner@acme.test")
    const globex = await app.seedProject("globex", "owner@globex.test")
    const theirs = await startRun(app, acme)
    const ours = await startRun(app, globex)

    // Globex's member names Acme's run in Globex's own project: not found, as agent.run.status says.
    const asStatus = await refusal(globex.call.callByName("agent.run.status", { ...globex.slugs, runId: theirs.runId }))
    const asEvents = await refusal(events(globex.call, { ...globex.slugs, runId: theirs.runId }))
    expect(asEvents.status).toBe(404)
    expect(asEvents.code).toBe(asStatus.code)
    // ... and in Acme's project, of which they are no member: refused before anything is read.
    const outside = await refusal(events(globex.call, { ...acme.slugs, runId: theirs.runId }))
    expect([403, 404]).toContain(outside.status)

    const foreignOffset = await refusal(
      events(globex.call, { ...globex.slugs, runId: ours.runId, after: `agk1:${theirs.runId}:0` }),
    )
    expect(foreignOffset.status).toBe(422)
    const malformed = await refusal(events(globex.call, { ...globex.slugs, runId: ours.runId, after: "../0" }))
    expect(malformed.status).toBe(422)
  })
})

// This journey requires explicit optional-module admission.
vi.mock("../modules/inclusion.js", async (importOriginal) => ({
  ...(await importOriginal<typeof import("../modules/inclusion.js")>()),
  ...(await import("./support/native-readiness.js")).selectedModules(["agent"]),
}))