// The run job over the real tools adapter, MCP client and gateway adapter; only the platform is
// answered, at `fetch`. What it pins is the line `StepOutcome` draws: a call a retry cannot fix ends
// the run with a typed failure and the job completes, so the queue does not run it again, while a
// transient failure rejects, so the queue does, and the step resumes on its idempotency key. A tool
// that answers an error is never a committed step: the model would otherwise read the error as the
// tool's result and the run would end `completed`.

import { AGENT_MODEL, AGENT_TOOLS, dataRunStore } from "@conusai-kit/agent"
import type { JobContext } from "@conusai-kit/core"
import { createMemoryDataSource, DATA_SOURCE } from "@conusai-kit/data"
import { afterEach, describe, expect, it, vi } from "vitest"
import { gatewayAgentModel, registryAgentTools, runJob } from "../adapters/procedures.js"
import { AGENT_SCOPE } from "../core/ports.js"
import { cancelRun, startRun } from "../core/runs.js"
import { claimedDataSource, claimedDelivery } from "./claimed-job.js"
import { workerFixture } from "./worker-fixture.js"

const API_URL = "https://cloud.example.test/api"
const MCP_URL = "https://cloud.example.test/v1/orgs/7/mcp"
const GATEWAY_URL = `${API_URL}/ai/v1/chat/completions`
const TOKEN = "dt_synthetic_fixture"
const TOOL_TEXT = "the backing service said something only its tenant may read"

interface ToolCall {
  readonly name: string
  readonly idempotencyKey: string
  readonly authorization: string | null
}
type Mcp = (id: number) => Response

const rpc =
  (result: unknown): Mcp =>
  (id) =>
    Response.json({ jsonrpc: "2.0", id, result })
const answered = rpc({
  resultType: "complete",
  isError: false,
  structuredContent: { rows: 3 },
  content: [{ type: "text", text: '{"rows":3}' }],
})
const denied = (reason: string) =>
  rpc({
    resultType: "complete",
    isError: true,
    structuredContent: { status: "denied", reason },
    content: [{ type: "text", text: `refused: ${reason}` }],
  })
const status =
  (code: number): Mcp =>
  () =>
    new Response(null, { status: code })

/** The gateway as a model that follows the tool choice: it calls the step offered, then answers. */
function gatewayReply(body: { tool_choice?: unknown }): Response {
  const choice = body.tool_choice
  const step =
    typeof choice === "object" && choice !== null
      ? (choice as { function?: { name?: string } }).function?.name
      : undefined
  const delta =
    step === undefined
      ? { role: "assistant", content: "Done." }
      : {
          role: "assistant",
          tool_calls: [{ index: 0, id: `call-${step}`, type: "function", function: { name: step, arguments: "{}" } }],
        }
  const chunk = {
    id: "chatcmpl-1",
    object: "chat.completion.chunk",
    created: 1,
    model: "gateway-model",
    choices: [{ index: 0, delta, finish_reason: step === undefined ? "stop" : "tool_calls" }],
  }
  return new Response(`data: ${JSON.stringify(chunk)}\n\ndata: [DONE]\n\n`, {
    status: 200,
    headers: { "content-type": "text/event-stream" },
  })
}

/** A queued run of one step, and a platform whose tool endpoint answers with `tool.mcp`. */
async function harness(toolName = "records__read") {
  const tool: { mcp: Mcp; calls: ToolCall[] } = { mcp: answered, calls: [] }
  vi.stubGlobal(
    "fetch",
    vi.fn(async (input: string | URL | Request, init?: RequestInit) => {
      const url = String(input)
      const body = JSON.parse(String(init?.body ?? "{}"))
      if (url === GATEWAY_URL) return gatewayReply(body)
      if (url !== MCP_URL) throw new Error(`no such stand-in: ${url}`)
      tool.calls.push({
        name: body.params.name,
        idempotencyKey: body.params._meta["com.conusai/idempotencyKey"],
        authorization: new Headers(init?.headers).get("authorization"),
      })
      return tool.mcp(body.id)
    }),
  )
  const data = await claimedDataSource()
  const { container, scope } = await workerFixture(data)
  container.value(AGENT_SCOPE, scope)
  container.value(AGENT_MODEL, gatewayAgentModel({ apiUrl: API_URL, token: TOKEN, model: "gateway-model" }))
  container.value(
    AGENT_TOOLS,
    registryAgentTools({
      apiUrl: API_URL,
      token: TOKEN,
      orgId: "7",
      scope: { orgSlug: "acme", workspaceSlug: "main", projectSlug: "p" },
    }),
  )
  const started = await startRun({ data }, scope, {
    goal: "Read the records",
    plan: [{ key: "read", label: "Read the records", tool: toolName, args: { limit: 1 } }],
  })
  const store = dataRunStore(data, scope)
  return {
    tool,
    data,
    scope,
    container,
    runId: started.runId,
    /** One delivery of the job: a rejection is what makes the queue deliver it again. */
    deliver: (extra: Partial<JobContext> = {}) => claimedDelivery(data, container, scope, started.runId, extra),
    cancel: () => cancelRun({ data }, scope, started.runId),
    run: () => store.get(started.runId),
    step: async () => (await store.listSteps(started.runId))[0],
  }
}

afterEach(() => {
  vi.unstubAllGlobals()
})

describe("the agent run job's tool calls", () => {
  it("calls the tool with the deployment token and completes the run", async () => {
    const job = await harness()
    await job.deliver()
    expect(job.tool.calls).toEqual([
      { name: "records__read", idempotencyKey: `${job.runId}:read`, authorization: `Bearer ${TOKEN}` },
    ])
    expect(await job.run()).toMatchObject({ status: "completed", failureCode: null })
    expect(await job.step()).toMatchObject({ status: "committed", attempts: 1, result: { rows: 3 } })
  })

  it.each<[string, Mcp, string]>([
    ["a token without the grant (403)", status(403), "forbidden"],
    ["a token the endpoint does not accept (401)", status(401), "unauthenticated"],
    ["a policy denial", denied("forbidden"), "forbidden"],
    ["a plan denial", denied("plan-required"), "plan-required"],
  ])("ends the run failed/tool_error on %s, and the queue does not run it again", async (_case, mcp, reason) => {
    const job = await harness()
    job.tool.mcp = mcp
    await expect(job.deliver()).resolves.toBeUndefined()
    const run = await job.run()
    expect(run).toMatchObject({ status: "failed", failureCode: "tool_error" })
    expect(run?.failureDetail).toContain(reason)
    expect(run?.finishedAt).not.toBeNull()
    expect(await job.step()).toMatchObject({ status: "failed", attempts: 1, result: null })
    // A redelivery reaches a terminal run: no second call.
    await job.deliver()
    expect(job.tool.calls).toHaveLength(1)
  })

  it("does not resume a refused run when the grant is approved later", async () => {
    const job = await harness()
    job.tool.mcp = status(403)
    await job.deliver()
    job.tool.mcp = answered
    await job.deliver()
    expect(await job.run()).toMatchObject({ status: "failed", failureCode: "tool_error" })
    expect(job.tool.calls).toHaveLength(1)
  })

  it("does not resume a cancelled run: a run stuck on a transient failure can be ended by a person", async () => {
    const job = await harness()
    job.tool.mcp = status(503)
    await expect(job.deliver()).rejects.toThrow()
    await job.cancel()
    job.tool.mcp = answered
    await job.deliver()
    expect(await job.run()).toMatchObject({ status: "cancelled", failureCode: null })
    await job.deliver()
    expect(job.tool.calls).toHaveLength(1)
  })

  it("does not commit a step whose tool answered an error, and keeps the tool's text out of the run", async () => {
    const job = await harness()
    job.tool.mcp = rpc({ resultType: "complete", isError: true, content: [{ type: "text", text: TOOL_TEXT }] })
    await expect(job.deliver()).resolves.toBeUndefined()
    const run = await job.run()
    expect(run).toMatchObject({ status: "failed", failureCode: "tool_error", usageSteps: 0 })
    expect(run?.failureDetail).toContain("records__read")
    expect(run?.failureDetail).not.toContain(TOOL_TEXT)
    const step = await job.step()
    expect(step).toMatchObject({ status: "failed", result: null })
    expect(step?.error).not.toContain(TOOL_TEXT)
    await job.deliver()
    expect(job.tool.calls).toHaveLength(1)
  })

  it.each<[string, Mcp, string?]>([
    ["a 503", status(503)],
    ["a 429", status(429)],
    ["a 500", status(500)],
    [
      "busy local compute",
      rpc({
        resultType: "complete",
        isError: true,
        structuredContent: { status: "compute_busy", retryAfterSeconds: 5 },
      }),
    ],
    [
      "a catalog that is still indexing",
      rpc({ resultType: "complete", isError: true, structuredContent: { status: "indexing", retryAfterSeconds: 5 } }),
      "find_tools",
    ],
    ["a runtime the platform could not reach", denied("runtime-unavailable")],
  ])("leaves the run to the queue on %s, and the next delivery resumes the step", async (_case, mcp, toolName) => {
    const job = await harness(toolName)
    job.tool.mcp = mcp
    await expect(job.deliver()).rejects.toThrow()
    expect(await job.run()).toMatchObject({ status: "running", failureCode: null })
    expect(await job.step()).toMatchObject({ status: "pending", attempts: 1 })

    job.tool.mcp = answered
    await job.deliver()
    expect(await job.run()).toMatchObject({ status: "completed", failureCode: null })
    expect(await job.step()).toMatchObject({ status: "committed", attempts: 2 })
    // The same key both times: an effect the first call made is not made twice.
    expect(new Set(job.tool.calls.map((call) => call.idempotencyKey))).toEqual(new Set([`${job.runId}:read`]))
  })

  it("leaves the run to the queue when the network fails", async () => {
    const job = await harness()
    job.tool.mcp = () => {
      throw new TypeError("fetch failed")
    }
    await expect(job.deliver()).rejects.toThrow("fetch failed")
    expect(await job.run()).toMatchObject({ status: "running", failureCode: null })
  })
})

it("persists sanitized exhaustion on the actual final attempt, with no replay calls", async () => {
  const job = await harness()
  job.tool.mcp = status(503)
  await job.deliver({ finalAttempt: true })
  expect(await job.run()).toMatchObject({ status: "failed", failureCode: "internal" })
  expect((await job.run())?.failureDetail).toBe("the durable job exhausted its retry attempts")
  await job.deliver({ finalAttempt: true })
  expect(job.tool.calls).toHaveLength(1)
})

it("a revoked initiating member is refused before any provider call despite the warm memo", async () => {
  const job = await harness()
  const { orgMemberships } = await import("@conusai-kit/tenancy")
  const repos = job.data.forScope(job.scope)
  const membership = (await repos.of(orgMemberships).list({ limit: 10 })).items[0]
  if (!membership) throw new Error("expected membership")
  await repos.of(orgMemberships).delete(membership.id)
  await job.deliver()
  expect(await job.run()).toMatchObject({ status: "failed", failureCode: "internal" })
  expect(vi.mocked(fetch)).not.toHaveBeenCalled()
})

it("revocation after a tool prevents the next model call", async () => {
  const job = await harness()
  const { orgMemberships } = await import("@conusai-kit/tenancy")
  const original = job.container.resolve(AGENT_TOOLS)
  const child = job.container.child()
  child.value(AGENT_TOOLS, {
    async call(...args) {
      const result = await original.call(...args)
      const repos = job.data.forScope(job.scope)
      const member = (await repos.of(orgMemberships).list({ limit: 10 })).items[0]
      if (!member) throw new Error("expected membership")
      await repos.of(orgMemberships).delete(member.id)
      return result
    },
  })
  await job.deliver({ container: child })
  expect(await job.run()).toMatchObject({ status: "failed", usageSteps: 1 })
  expect(job.tool.calls).toHaveLength(1)
  expect(vi.mocked(fetch).mock.calls.filter(([url]) => String(url) === GATEWAY_URL)).toHaveLength(1)
})

it("missing adapter execution refuses before durable or provider work even at exhaustion", async () => {
  const job = await harness()
  await expect(
    runJob.handler(
      { runId: job.runId },
      {
        container: job.container,
        correlationId: "unsupported-adapter",
        attempt: 1,
        finalAttempt: true,
      },
    ),
  ).rejects.toMatchObject({ code: "job_claim_lost" })
  expect((await job.run())?.status).toBe("queued")
  expect(job.tool.calls).toHaveLength(0)
  expect(vi.mocked(fetch)).not.toHaveBeenCalled()
})

it("an actual claim with an unsupported data source hook refuses before Agent effects", async () => {
  const job = await harness()
  const child = job.container.child()
  child.value(DATA_SOURCE, createMemoryDataSource({ entities: [] }))
  await expect(job.deliver({ container: child, finalAttempt: true })).rejects.toMatchObject({ code: "job_claim_lost" })
  expect((await job.run())?.status).toBe("queued")
  expect(job.tool.calls).toHaveLength(0)
  expect(vi.mocked(fetch)).not.toHaveBeenCalled()
})