import assert from 'node:assert/strict'
import { createServer, type ServerResponse } from 'node:http'
import { once } from 'node:events'
import { mkdtemp, writeFile } from 'node:fs/promises'
import { resolve, join, relative, isAbsolute } from 'node:path'
import { Gateway, GatewayError, type GatewayDelta, type WireMessage } from '../src/main/gateway'
import { runTurn } from '../src/main/agent/loop'
import { responsesInput } from '../src/main/gateway/responses'
import type { GatewayConfig, ChatEvent } from '../src/shared/types'

const directory = await mkdtemp(resolve('.tmp/gateway-responses-'))
const rel = relative(resolve('.tmp'), directory)
assert.ok(!isAbsolute(rel) && !rel.startsWith('..') && rel.startsWith('gateway-responses-'))
await writeFile(join(directory, 'proof.txt'), 'only controlled local data')
const secret = 'controlled-secret-must-not-appear-in-errors'
const requests: Array<{ path: string; body: any; authorizationPresent: boolean }> = []
const checks: string[] = []
const check = (name: string, actual: unknown, expected: unknown): void => { assert.deepEqual(actual, expected, name); checks.push(name) }
const payload = (type: string, values: Record<string, unknown> = {}): Record<string, unknown> => ({ type, ...values })
const frame = (data: unknown, event?: string): string => `${event ? 'event: ' + event + '\r\n' : ''}data: ${JSON.stringify(data)}\r\n\r\n`
const complete = (output: unknown[] = []): Record<string, unknown> => payload('response.completed', { response: { id: 'response-controlled', status: 'completed', output } })
const functionItem = { type: 'function_call', id: 'fc_internal', call_id: 'call_public', name: 'list_dir', arguments: '{"path":"."}', status: 'completed' }
const messageItem = (text: string, id = 'msg_1'): Record<string, unknown> => ({ type: 'message', id, role: 'assistant', status: 'completed', content: [{ type: 'output_text', text, annotations: [] }] })
const reasoningItem = { type: 'reasoning', id: 'rs_1', summary: [{ type: 'summary_text', text: '先看目录。' }] }
async function send(res: ServerResponse, events: unknown[], tail = ''): Promise<void> {
  res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache' })
  const data = Buffer.from(': keepalive\r\n\r\ndata: invalid-json-heartbeat\r\n\r\n' + events.map(item => frame(item)).join('') + tail)
  // Actual HTTP chunks split UTF-8 glyphs, CRLF boundaries and JSON tokens.
  for (let start = 0; start < data.length; start += 13) {
    if (res.destroyed) return
    if (!res.write(data.subarray(start, start + 13))) await once(res, 'drain')
    if (start % 143 === 0) await new Promise<void>(resolve => setImmediate(resolve))
  }
  res.end()
}
const server = createServer((req, res) => {
  void (async () => {
    let raw = ''; for await (const chunk of req) raw += chunk
    const body = JSON.parse(raw), url = new URL(req.url ?? '', 'http://127.0.0.1'), mode = url.searchParams.get('case')
    requests.push({ path: req.url ?? '', body, authorizationPresent: !!req.headers.authorization })
    if (url.pathname.endsWith('/chat/completions')) {
      await send(res, [{ choices: [{ delta: { content: '旧 Chat 保留' }, finish_reason: 'stop' }] }], 'data: [DONE]\n\n'); return
    }
    assert.equal(url.pathname, '/v1/responses')
    if (mode === 'http-error') { res.writeHead(429, { 'content-type': 'application/json' }); res.end(JSON.stringify({ error: { message: secret } })); return }
    if (mode === 'stream-error') { await send(res, [payload('error', { message: secret, code: 'provider_failure' })]); return }
    if (mode === 'failed') { await send(res, [payload('response.failed', { response: { status: 'failed', error: { message: secret } } })]); return }
    if (mode === 'incomplete') { await send(res, [payload('response.output_item.added', { output_index: 0, item: { ...functionItem, status: 'in_progress' } }), payload('response.incomplete', { response: { status: 'incomplete', incomplete_details: { reason: 'max_output_tokens' } } })]); return }
    if (mode === 'wrong-status') { await send(res, [payload('response.completed', { response: { status: 'failed', error: { message: secret } } })]); return }
    if (mode === 'missing-status' || mode === 'missing-output') {
      await send(res, [payload('response.output_item.done', { output_index: 0, item: functionItem }),
        payload('response.completed', { response: mode === 'missing-status' ? { output: [functionItem] } : { status: 'completed' } })]); return
    }
    if (mode === 'truncated') { await send(res, [payload('response.output_text.delta', { item_id: 'msg_1', output_index: 0, content_index: 0, delta: '部分文本' })], 'data: [DONE]\n\n'); return }
    if (mode === 'cancel') {
      res.writeHead(200, { 'content-type': 'text/event-stream' })
      res.write(frame(payload('response.output_text.delta', { item_id: 'msg_1', output_index: 0, content_index: 0, delta: '取消前文本' })))
      return
    }
    if (mode === 'tool') {
      if (body.input.some((item: any) => item.type === 'function_call_output')) {
        await send(res, [payload('response.output_text.delta', { item_id: 'msg_final', output_index: 0, content_index: 0, delta: '完成读取。' }), complete([messageItem('完成读取。', 'msg_final')])]); return
      }
      await send(res, [
        payload('response.created', { response: { status: 'in_progress' } }),
        payload('response.output_item.added', { output_index: 0, item: { type: 'reasoning', id: 'rs_1', summary: [] } }),
        payload('response.reasoning_summary_text.delta', { item_id: 'rs_1', output_index: 0, summary_index: 0, delta: '先看' }),
        payload('response.reasoning_summary_text.delta', { item_id: 'rs_1', output_index: 0, summary_index: 0, delta: '目录。' }),
        payload('response.reasoning_summary_text.done', { item_id: 'rs_1', output_index: 0, summary_index: 0, text: '先看目录。' }),
        payload('response.output_item.done', { output_index: 0, item: reasoningItem }),
        payload('response.output_text.delta', { item_id: 'msg_1', output_index: 1, content_index: 0, delta: '查看' }),
        payload('response.output_text.done', { item_id: 'msg_1', output_index: 1, content_index: 0, text: '查看目录。' }),
        payload('response.output_item.added', { output_index: 2, item: { ...functionItem, arguments: '', status: 'in_progress' } }),
        payload('response.function_call_arguments.delta', { item_id: 'fc_internal', output_index: 2, delta: '{"pa' }),
        payload('response.function_call_arguments.delta', { item_id: 'fc_internal', output_index: 2, delta: 'th":"."}' }),
        payload('response.function_call_arguments.done', { item_id: 'fc_internal', output_index: 2, arguments: '{"path":"."}' }),
        payload('response.output_item.done', { output_index: 2, item: functionItem }),
        complete([reasoningItem, messageItem('查看目录。'), functionItem])]); return
    }
    if (mode === 'snapshot') {
      await send(res, [complete([reasoningItem, messageItem('仅完成快照。')])]); return
    }
    if (mode === 'parallel') {
      await send(res, [payload('response.function_call_arguments.delta', { output_index: 1, item_id: 'fc_second', delta: '{"b":' }),
        payload('response.output_item.added', { output_index: 0, item: { type: 'function_call', id: 'fc_first', call_id: 'call_first', name: 'one', arguments: '', status: 'in_progress' } }),
        payload('response.function_call_arguments.done', { output_index: 0, item_id: 'fc_first', arguments: '{"a":1}' }),
        payload('response.output_item.done', { output_index: 0, item: { type: 'function_call', id: 'fc_first', call_id: 'call_first', name: 'one', arguments: '{"a":1}' } }),
        payload('response.output_item.done', { output_index: 1, item: { type: 'function_call', id: 'fc_second', call_id: 'call_second', name: 'two', arguments: '{"b":2}' } }), complete()]); return
    }
    if (mode === 'invalid-call') { await send(res, [complete([{ ...functionItem, arguments: '{' }])]); return }
    if (mode === 'missing-id') { await send(res, [complete([{ ...functionItem, call_id: '' }])]); return }
    await send(res, [payload('response.output_text.delta', { output_index: 0, content_index: 0, delta: '你好' }),
      payload('response.output_text.done', { item_id: 'msg_1', output_index: 0, content_index: 0, text: '你好，Responses。' }),
      payload('response.content_part.done', { item_id: 'msg_1', output_index: 0, content_index: 0, part: { type: 'output_text', text: '你好，Responses。' } }),
      complete([messageItem('你好，Responses。')])])
  })().catch(() => { res.destroy() })
})
server.listen(0, '127.0.0.1'); await once(server, 'listening')
const address = server.address(); assert.ok(address && typeof address === 'object')
const base = `http://127.0.0.1:${address.port}`
const config = (mode: string, suffix = '/v1/responses'): GatewayConfig => ({ baseUrl: `${base}${suffix}?case=${mode}`, model: 'gpt-6-luna', apiKey: secret, temperature: 0.3, maxToolRounds: 4, maxOutputTokens: 1024 })
const request = { messages: [{ role: 'user' as const, content: '受控测试' }] }
async function collect(gateway: Gateway, input = request): Promise<GatewayDelta[]> { const result: GatewayDelta[] = []; for await (const delta of gateway.stream(input)) result.push(delta); return result }
const joined = (deltas: GatewayDelta[], type: 'text' | 'reasoning'): string => deltas.filter(delta => delta.type === type).map(delta => 'text' in delta ? delta.text : '').join('')
try {
  const events: ChatEvent[] = []
  const result = await runTurn({ runId: 'controlled-responses', history: [], userText: '列出本地受控文件', signal: new AbortController().signal,
    emit: event => events.push(event), gateway: new Gateway(config('tool')), model: 'gpt-6-luna', maxToolRounds: 4, workspaceRoot: directory })
  check('real fallback loop executes function and completes next response', [result.content, result.tools.length, result.tools[0].status], ['查看目录。完成读取。', 1, 'ok'])
  const rounds = requests.filter(req => req.path.includes('case=tool'))
  check('complete endpoint has exactly two rounds and no repeated protocol path', rounds.map(req => req.path), ['/v1/responses?case=tool', '/v1/responses?case=tool'])
  check('function tools flatten and keep optional parameter semantics', [rounds[0].body.tools[0].type, rounds[0].body.tools[0].name, rounds[0].body.tools[0].strict, rounds[0].body.tools[0].function], ['function', 'list_dir', false, undefined])
  check('Responses token/temperature/model fields', [rounds[0].body.max_output_tokens, rounds[0].body.max_tokens, rounds[0].body.temperature, rounds[0].body.model, rounds[0].body.store], [1024, undefined, 0.3, 'gpt-6-luna', false])
  const call = rounds[1].body.input.find((item: any) => item.type === 'function_call'), output = rounds[1].body.input.find((item: any) => item.type === 'function_call_output')
  check('history replays public call_id, exact JSON and matching output', [call.call_id, call.arguments, output.call_id, output.output.includes('proof.txt')], ['call_public', '{"path":"."}', 'call_public', true])
  check('reasoning summary deltas do not duplicate snapshots', events.filter(event => event.type === 'part' && event.part.type === 'reasoning').at(-1)?.type === 'part' ? (events.filter(event => event.type === 'part' && event.part.type === 'reasoning').at(-1) as any).part.text : '', '先看目录。')
  check('fallback emits done exactly once and no error', [events.filter(event => event.type === 'done').length, events.filter(event => event.type === 'error').length], [1, 0])
  const multimodal = await collect(new Gateway(config('text')), { messages: [{ role: 'user', content: [{ type: 'text', text: '看图' }, { type: 'image_url', image_url: { url: 'data:image/png;base64,Y29udHJvbGxlZA==' } }] }] })
  const multimodalRequest = requests.at(-1)!
  check('image/text convert to Responses input content', multimodalRequest.body.input[0].content, [{ type: 'input_text', text: '看图' }, { type: 'input_image', image_url: 'data:image/png;base64,Y29udHJvbGxlZA==', detail: 'auto' }])
  check('no tools means no tools/tool_choice keys', [multimodalRequest.body.tools, multimodalRequest.body.tool_choice], [undefined, undefined])
  check('text delta/done/part/item/completed snapshots deduplicate even late item id', [joined(multimodal, 'text'), multimodal.at(-1)], ['你好，Responses。', { type: 'finish', reason: 'stop' }])
  const snapshots = await collect(new Gateway(config('snapshot')))
  check('snapshot-only response can deliver text/reasoning', [joined(snapshots, 'text'), joined(snapshots, 'reasoning')], ['仅完成快照。', '先看目录。'])
  const parallel = await collect(new Gateway(config('parallel'))), calls = parallel.find(delta => delta.type === 'tool_calls')
  check('parallel calls reconcile by item/index and order by output_index', calls?.type === 'tool_calls' ? calls.calls.map(call => [call.id, call.function.name, call.function.arguments]) : [], [['call_first', 'one', '{"a":1}'], ['call_second', 'two', '{"b":2}']])
  check('tool finish is emitted once', parallel.filter(delta => delta.type === 'finish'), [{ type: 'finish', reason: 'tool_calls' }])
  for (const mode of ['http-error', 'stream-error', 'failed', 'wrong-status', 'missing-status', 'missing-output', 'incomplete', 'truncated', 'invalid-call', 'missing-id']) {
    const deltas: GatewayDelta[] = []
    let error: unknown
    try { for await (const delta of new Gateway(config(mode)).stream(request)) deltas.push(delta) } catch (reason) { error = reason }
    check(`${mode}: throws GatewayError without successful finish or executable calls`, [error instanceof GatewayError, deltas.some(delta => delta.type === 'finish' || delta.type === 'tool_calls'), String(error).includes(secret), (error as GatewayError).body], [true, false, false, undefined])
    if (mode === 'http-error') check('HTTP failure retains safe status', (error as GatewayError).status, 429)
  }
  const failedEvents: ChatEvent[] = []
  await assert.rejects(runTurn({ runId: 'incomplete-turn', history: [], userText: '受控未完成', signal: new AbortController().signal,
    emit: event => failedEvents.push(event), gateway: new Gateway(config('incomplete')), model: 'gpt-6-luna', maxToolRounds: 2, workspaceRoot: directory }), GatewayError)
  check('incomplete fallback loop emits error rather than done/tool execution', [failedEvents.some(event => event.type === 'error'), failedEvents.some(event => event.type === 'done' || event.type === 'tool_call')], [true, false])
  const abort = new AbortController(), cancelled: GatewayDelta[] = []
  await assert.rejects(async () => { for await (const delta of new Gateway(config('cancel')).stream({ ...request, signal: abort.signal })) { cancelled.push(delta); abort.abort() } }, { name: 'AbortError' })
  check('abort preserves partial text and never finishes', [joined(cancelled, 'text'), cancelled.some(delta => delta.type === 'finish')], ['取消前文本', false])
  for (const suffix of ['/v1', '/v1/chat/completions/']) {
    const chat = await collect(new Gateway(config('chat', suffix)))
    check('legacy Chat endpoint and stream preserved: ' + suffix, [requests.at(-1)!.path, joined(chat, 'text'), requests.at(-1)!.body.max_tokens, requests.at(-1)!.body.max_output_tokens, requests.at(-1)!.body.messages], ['/v1/chat/completions?case=chat', '旧 Chat 保留', 1024, undefined, request.messages])
  }
  const history: WireMessage[] = [{ role: 'system', content: 'system' }, { role: 'assistant', content: null, tool_calls: [{ id: 'public', type: 'function', function: { name: 'f', arguments: '{}' } }] }, { role: 'tool', tool_call_id: 'public', content: 'ok' }, { role: 'assistant', content: 'reply' }]
  check('system/assistant/tool history has no null assistant tool-only message', responsesInput(history), [{ role: 'system', content: 'system' }, { type: 'function_call', call_id: 'public', name: 'f', arguments: '{}' }, { type: 'function_call_output', call_id: 'public', output: 'ok' }, { role: 'assistant', content: 'reply' }])
  assert.throws(() => responsesInput([{ role: 'tool', content: 'missing' }]), /调用 ID/); checks.push('missing historical tool call id fails locally')
  const report = { result: 'PASS', scope: 'Controlled loopback HTTP + actual Gateway/fallback runTurn/local list_dir; no real user endpoint or OMP', checks, requestCount: requests.length }
  await writeFile(join(directory, 'report.json'), JSON.stringify(report, null, 2))
  console.log(JSON.stringify({ ...report, directory }))
} finally { server.closeAllConnections(); await new Promise<void>(resolve => server.close(() => resolve())) }