import assert from 'node:assert/strict'
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import test from 'node:test'
import { fileURLToPath } from 'node:url'

const __dirname = path.dirname(fileURLToPath(import.meta.url))
const workerRoot = path.resolve(__dirname, '../..')

function execMigrationOnce(db, file) {
  try {
    db.exec(fs.readFileSync(path.join(workerRoot, file), 'utf8'))
  } catch (error) {
    if (!/duplicate column|already exists/i.test(error?.message || '')) throw error
  }
}

test('local runtime events reach SSE before the child process closes', async (t) => {
  const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), 'aipros-route-local-stream-'))
  const marker = path.join(tempRoot, 'child-closed.txt')
  process.env.DB_PATH = path.join(tempRoot, 'test.db')
  process.env.OUTPUTS_DIR = path.join(tempRoot, 'outputs')

  t.after(() => fs.rmSync(tempRoot, { recursive: true, force: true }))

  const { default: app } = await import('../index.js')
  const { rawDB } = await import('../db.js')
  execMigrationOnce(rawDB, 'migrations/0010_runtime_engines.sql')
  execMigrationOnce(rawDB, 'migrations/0014_local_runtime_agents.sql')

  const script = [
    "process.stdout.write('first')",
    `setTimeout(() => { process.stdout.write('second'); require('node:fs').writeFileSync(${JSON.stringify(marker)}, 'done') }, 180)`,
  ].join(';')

  rawDB.prepare('UPDATE runtime_engines SET enabled = 0, is_active = 0').run()
  rawDB.prepare(`
    UPDATE runtime_engines
    SET local_command = ?, local_args_json = ?, enabled = 1, is_active = 1,
        purpose_scopes_json = '["run"]'
    WHERE code = 'local-workbuddy'
  `).run(process.execPath, JSON.stringify(['-e', script]))
  rawDB.prepare(`
    UPDATE runtime_agents
    SET runtime_code = 'local-workbuddy', applies_to_json = '["run"]', enabled = 1
    WHERE id = 'agent-skill-runner'
  `).run()

  const response = await app.request('/api/skill-run/stream', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-User-Email': '494161546@qq.com',
    },
    body: JSON.stringify({
      mode: 'run',
      systemPrompt: '# runner',
      messages: [{ role: 'user', content: '请执行流式测试' }],
    }),
  })

  assert.equal(response.status, 200)
  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  let text = ''
  let eventBuffer = ''
  let sawFirstDelta = false
  while (!sawFirstDelta) {
    const { value, done } = await reader.read()
    assert.equal(done, false, '应在流结束前收到 first')
    const chunk = decoder.decode(value, { stream: true })
    text += chunk
    eventBuffer += chunk
    const parts = eventBuffer.split('\n\n')
    eventBuffer = parts.pop() || ''
    for (const part of parts) {
      const data = part.split(/\r?\n/)
        .filter(line => line.startsWith('data:'))
        .map(line => line.slice(5).trim())
        .join('\n')
      try {
        const event = JSON.parse(data)
        if (event?.t === 'delta' && event?.v === 'first') sawFirstDelta = true
      } catch {}
    }
  }

  assert.equal(fs.existsSync(marker), false, '首个 delta 必须早于子进程结束')

  while (true) {
    const { value, done } = await reader.read()
    if (done) break
    text += decoder.decode(value, { stream: true })
  }
  assert.match(text, /second/)
  assert.match(text, /"state":"completed"/)
})
