import assert from 'node:assert/strict'
import { setTimeout as delay } from 'node:timers/promises'
import test from 'node:test'
import { Context } from '@deepseek-ai/cordis'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRuntime, { defineTool } from '@deepseek-ai/dsh-tools'
import { Client } from '@modelcontextprotocol/sdk/client/index.js'
import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js'
import { ToolListChangedNotificationSchema } from '@modelcontextprotocol/sdk/types.js'
import plugin, { createMcpServer } from '../src/index.mjs'

async function runtime(t) {
  const ctx = new Context()
  await ctx.plugin(SystemPrompt, {})
  await ctx.plugin(ToolRuntime, { mode: 'native' })
  t.after(() => ctx.fiber.dispose())
  return ctx
}

function register(ctx, name = 'echo', options = {}) {
  return ctx.tools.register(defineTool({
    name,
    description: '测试文本工具',
    parameters: { text: { type: 'string', required: true } },
    output: {
      schema: { type: 'string' },
      render: (_args, value) => [{ type: 'text', text: value }],
    },
    async execute({ text }) { return text },
    ...options,
  }))
}

async function connect(ctx, t) {
  const exported = createMcpServer(ctx)
  const client = new Client({ name: 'message-mcp-test', version: '0.1.0' })
  const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair()
  await exported.server.connect(serverTransport)
  await client.connect(clientTransport)
  t.after(async () => {
    await exported.close()
    await client.close()
  })
  return { ...exported, client }
}

test('从真实工具注册表列出 schema 并通知新增与移除', async (t) => {
  const ctx = await runtime(t)
  register(ctx)
  const { client } = await connect(ctx, t)
  assert.deepEqual((await client.listTools()).tools, [{
    name: 'echo',
    description: '测试文本工具',
    inputSchema: {
      type: 'object',
      properties: { text: { type: 'string' } },
      required: ['text'],
    },
  }])
  const changed = Promise.withResolvers()
  client.setNotificationHandler(ToolListChangedNotificationSchema, () => changed.resolve())
  const unregister = register(ctx, 'second')
  await changed.promise
  assert.equal((await client.listTools()).tools.length, 2)
  unregister()
  assert.deepEqual((await client.listTools()).tools.map((tool) => tool.name), ['echo'])
})

test('调用通过真实前置、执行包装、后置与最终通知策略且不泄漏原始值', async (t) => {
  const ctx = await runtime(t)
  const order = []
  register(ctx, 'echo', {
    async execute() { order.push('body'); return '原始私有值' },
  })
  ctx.on('tools/pre-execute', async (_exec, next) => { order.push('pre'); return next() })
  ctx.tools.guard(() => { order.push('guard') })
  ctx.on('tools/execute', async (_exec, next) => { order.push('around'); return next() })
  ctx.on('tools/post-execute', async () => {
    order.push('post')
    return { kind: 'accept', content: [{ type: 'text', text: '已脱敏' }] }
  })
  ctx.on('tools/result', () => { order.push('result') })
  const { client } = await connect(ctx, t)
  const result = await client.callTool({ name: 'echo', arguments: { text: '输入' } })
  assert.deepEqual(order, ['pre', 'guard', 'around', 'body', 'post', 'result'])
  assert.deepEqual(result, { isError: false, content: [{ type: 'text', text: '已脱敏' }] })
  assert.equal(JSON.stringify(result).includes('原始私有值'), false)
})

test('guard 拒绝与无 agent 审批不会执行工具', async (t) => {
  const ctx = await runtime(t)
  let executed = 0
  register(ctx, 'denied', { async execute() { executed++; return '不应执行' } })
  register(ctx, 'ask', { async execute() { executed++; return '不应执行' } })
  ctx.tools.guard((exec) => exec.name === 'denied' ? '测试禁止' : undefined)
  ctx.on('tools/pre-execute', async (exec, next) => exec.name === 'ask'
    ? { kind: 'ask', reason: '需要人工批准' }
    : next())
  const { client } = await connect(ctx, t)
  for (const name of ['denied', 'ask']) {
    const result = await client.callTool({ name, arguments: { text: 'x' } })
    assert.equal(result.isError, true)
  }
  assert.equal(executed, 0)
})

test('参数错误和未知工具保留工具错误响应，之后仍可正常调用', async (t) => {
  const ctx = await runtime(t)
  register(ctx)
  const { client } = await connect(ctx, t)
  assert.equal((await client.callTool({ name: 'echo', arguments: { text: 12 } })).isError, true)
  assert.equal((await client.callTool({ name: 'missing', arguments: {} })).isError, true)
  assert.equal((await client.callTool({ name: 'echo', arguments: { text: '正常' } })).isError, false)
})

test('非文本结果仅给占位，不序列化内部内容', async (t) => {
  const ctx = await runtime(t)
  register(ctx, 'echo', {
    output: { schema: { type: 'string' }, render: () => [{ type: 'reasoning', text: '内部信息' }] },
  })
  const { client } = await connect(ctx, t)
  const result = await client.callTool({ name: 'echo', arguments: { text: 'x' } })
  assert.equal(JSON.stringify(result).includes('内部信息'), false)
  assert.match(result.content[0].text, /非文本/)
})

test('并发 MCP 请求串行执行', async (t) => {
  const ctx = await runtime(t)
  let active = 0
  let maximum = 0
  register(ctx, 'echo', {
    async execute({ text }) {
      maximum = Math.max(maximum, ++active)
      await delay(10)
      active--
      return text
    },
  })
  const { client } = await connect(ctx, t)
  const results = await Promise.all(['a', 'b', 'c'].map((text) => client.callTool({ name: 'echo', arguments: { text } })))
  assert.equal(maximum, 1)
  assert.deepEqual(results.map((result) => result.content[0].text), ['a', 'b', 'c'])
})

test('同一 Cordis 根下不同上下文的 MCP Server 共用独占队列', async (t) => {
  const ctx = await runtime(t)
  let active = 0
  let maximum = 0
  register(ctx, 'echo', {
    async execute({ text }) {
      maximum = Math.max(maximum, ++active)
      await delay(10)
      active--
      return text
    },
  })
  const first = await connect(ctx, t)
  const second = await connect(ctx.extend(), t)
  await Promise.all([first, second].map(({ client }) => client.callTool({ name: 'echo', arguments: { text: '独占' } })))
  assert.equal(maximum, 1)
})

test('客户端取消传入真实工具信号且关闭等待工具清理', async (t) => {
  const ctx = await runtime(t)
  const started = Promise.withResolvers()
  const cleaned = Promise.withResolvers()
  let finalized = false
  register(ctx, 'wait', {
    async execute(_args, exec) {
      started.resolve()
      await new Promise((resolve) => exec.signal.addEventListener('abort', resolve, { once: true }))
      await delay(10)
      finalized = true
      cleaned.resolve()
      return '已清理'
    },
  })
  const { client, close } = await connect(ctx, t)
  const controller = new AbortController()
  const request = client.callTool({ name: 'wait', arguments: { text: 'x' } }, undefined, { signal: controller.signal })
  const rejected = assert.rejects(request)
  await started.promise
  controller.abort()
  await rejected
  await close()
  await cleaned.promise
  assert.equal(finalized, true)
})

test('关闭会取消执行和排队请求，且可重复关闭', async (t) => {
  const ctx = await runtime(t)
  const started = Promise.withResolvers()
  let bodies = 0
  let finalized = false
  register(ctx, 'wait', {
    async execute(_args, exec) {
      bodies++
      started.resolve()
      await new Promise((resolve) => exec.signal.addEventListener('abort', resolve, { once: true }))
      await delay(10)
      finalized = true
      return '已清理'
    },
  })
  const { client, close } = await connect(ctx, t)
  const first = client.callTool({ name: 'wait', arguments: { text: 'a' } })
  const second = client.callTool({ name: 'wait', arguments: { text: 'b' } })
  const requests = Promise.allSettled([first, second])
  await started.promise
  await close()
  await requests
  await close()
  assert.equal(bodies, 1)
  assert.equal(finalized, true)
})

test('Cordis 默认不接管 stdin，错误传输配置不能启动', async (t) => {
  const ctx = await runtime(t)
  const before = process.stdin.listenerCount('data')
  const fiber = await ctx.plugin(plugin, {})
  assert.equal(process.stdin.listenerCount('data'), before)
  await fiber.dispose()
  // 使用标准 schema 验证，与 Loader 的实际启动路径一致。
  const result = await plugin.Config['~standard'].validate({ transport: 'unsupported' })
  assert.ok(result.issues)
})

test('Cordis 卸载清理 stdio 监听且允许再次挂载', async (t) => {
  const ctx = await runtime(t)
  const beforeData = process.stdin.listenerCount('data')
  const beforeEnd = process.stdin.listenerCount('end')
  for (let index = 0; index < 2; index++) {
    const fiber = await ctx.plugin(plugin, { transport: 'stdio' })
    assert.deepEqual(ctx.get('messageMcpExport'), { transport: 'stdio' })
    assert.equal(process.stdin.listenerCount('data'), beforeData + 1)
    assert.equal(process.stdin.listenerCount('end'), beforeEnd + 1)
    await fiber.dispose()
    assert.equal(ctx.get('messageMcpExport'), undefined)
    assert.equal(process.stdin.listenerCount('data'), beforeData)
    assert.equal(process.stdin.listenerCount('end'), beforeEnd)
  }
})

test('执行与schema异常经过最终秘密脱敏边界', async (t) => {
  const secret = 'fixture-only-error-secret'
  let failList = false
  const ctx = {
    get: key => key === 'messageSecretRedactor' ? text => text.split(secret).join('[REDACTED]') : undefined,
    on: () => () => {},
    tools: {
      schemas: () => { if (failList) throw new Error(secret); return [{ name: 'failure', description: 'error fixture', parameters: { type: 'object', properties: {} } }] },
      execute: async () => { throw new Error(secret) },
    },
  }
  const { client } = await connect(ctx, t)
  await assert.rejects(client.callTool({ name: 'failure', arguments: {} }), error => error.message.includes('[REDACTED]') && !error.message.includes(secret))
  failList = true
  await assert.rejects(client.listTools(), error => error.message.includes('[REDACTED]') && !error.message.includes(secret))
});

test('短环境值脱敏不会破坏JSON Schema的数值约束', async (t) => {
  const ctx = {
    get: () => text => text.split('1').join('[REDACTED]'), on: () => () => {},
    tools: {
      schemas: () => [{ name: 'short', description: 'text 1', parameters: { type: 'object', properties: { number: { type: 'number', minimum: 1 } } } }],
      execute: async () => ({ isError: false, content: [{ type: 'text', text: '1' }] }),
    },
  }
  const { client } = await connect(ctx, t)
  assert.equal((await client.listTools()).tools[0].inputSchema.properties.number.minimum, 1)
  assert.equal((await client.callTool({ name: 'short' })).content[0].text, '[REDACTED]')
});