// DeepSeek Harness 第三方 API 缺失 message_stop/done 终止事件修复插件
//
// Cordis（DeepSeek Harness 的插件运行时）插件。第三方网关套了 Anthropic 流式
// 协议壳、却漏发 message_stop/done 时，dsh 的 agent-loop 会把它判为 TRANSPORT
// 错误并触发 llm-retry 最多 5 次指数退避重试（每次都把整轮重新问一遍）。
//
// 修复方式：监听 host 端 `llm/stream` waterfall，把下游流包一层、内容原样透传；
// 只有当流在"内容已到"后以"缺终止信号"结束（pi-ai 抛 STREAM_CLOSED，或
// in-band error finish / 尾部异常的失败信息含 "stream ended (before|without)"、
// "message_stop"、"done/error"）才吞掉错误、让流干净结束。BlockAssembler 会把
// "没有 finish chunk 的干净 EOF"当作正常 stop，回复正常提交，不触发重试。
//
// 守卫（贪安全）：只修"内容已到、所有未闭合 tool-call 块参数已是合法 JSON"的
// 流；真正中断/残缺照常抛错重试；abort、空响应不修。
//
// 作用范围（不再写死任何 provider）：
//   - 旧写法  config.providers: [a, b]            —— 这些 provider 的全部模型
//   - 新写法  config.rules:
//                - provider: xjhc      # 必填
//                  models: ['*']       # 选填，缺省 = 全部；支持 * 通配，如 deepseek-v4*
//                  enabled: true       # 选填，缺省 true
//   两者可混用，逐条累加；未配置任何规则 = 不干预任何流（仅提示一次）。
//
// 若以 web profile 插件形式安装且 `ctx.settings` 可用（@deepseek-ai/schemastery
// 可解析），会自动注册设置命名空间 `missing-stop-heal`：组合配置作为 base 层，
// 用户在"设置 → 插件"里改的文档覆盖其上，改动即时（live）生效于后续请求。

export const name = 'missing-stop-heal'

function isAsyncIterable(value) {
  return value !== null && value !== undefined && typeof value === 'object' &&
    typeof value[Symbol.asyncIterator] === 'function'
}

// 只匹配"流自然跑到尽头却没有终止事件"这一种特征；真网络错误/限流/超时/鉴权
// 失败/abort 均不匹配，原样放行。
function isMissingStopError(error) {
  if (error === null || typeof error !== 'object') return false
  if (error.code === 'STREAM_CLOSED') return true
  const message = typeof error.message === 'string' ? error.message : ''
  return /stream ended (?:before|without)\b/i.test(message) ||
    /message_stop|done\/error/i.test(message)
}

function escapeRegExp(text) {
  return String(text).replace(/[.+^${}()|[\]\\]/g, '\\$&')
}

// 简单通配：* 匹配任意字符序列；无 * 时按精确匹配。
function globToRegExp(pattern) {
  return new RegExp(`^${escapeRegExp(pattern).replace(/\*/g, '.*')}$`)
}

// models 缺省/为空/含 '*' = 放行全部模型。
function modelsMatch(models, model) {
  const list = Array.isArray(models) ? models : []
  if (list.length === 0) return true
  const actual = String(model ?? '')
  for (const entry of list) {
    const pattern = String(entry)
    if (pattern === '*' || pattern === actual || globToRegExp(pattern).test(actual)) return true
  }
  return false
}

// 把 (legacy providers + rules) 归一化成一条条 {provider, models, enabled}；
// 非法条目跳过并记 notes。
function normalizeRules(config, note) {
  const notes = []
  const out = []
  const push = (raw) => {
    if (raw === null || typeof raw !== 'object') return
    const provider = typeof raw.provider === 'string' && raw.provider.length > 0
      ? raw.provider
      : null
    if (provider === null) {
      notes.push('忽略一条缺少 provider 的规则')
      return
    }
    if (raw.models !== undefined && (!Array.isArray(raw.models) ||
      raw.models.some((m) => typeof m !== 'string'))) {
      notes.push(`provider "${provider}" 的 models 非法，按全部模型处理`)
      out.push({ provider, models: [], enabled: raw.enabled !== false })
      return
    }
    out.push({
      provider,
      models: Array.isArray(raw.models) ? [...raw.models] : [],
      enabled: raw.enabled !== false,
    })
  }
  const cfg = config !== null && typeof config === 'object' ? config : {}
  if (Array.isArray(cfg.providers)) {
    for (const p of cfg.providers) if (typeof p === 'string' && p.length > 0) push({ provider: p })
  }
  if (Array.isArray(cfg.rules)) for (const rule of cfg.rules) push(rule)
  if (notes.length > 0 && typeof note === 'function') note(notes.join('；'))
  return out
}

function ruleMatches(rules, provider, model) {
  if (rules.length === 0) return false
  if (typeof provider !== 'string') return false
  return rules.some((rule) =>
    rule.enabled && rule.provider === provider && modelsMatch(rule.models, model))
}

async function* healStream(source) {
  let stream = source
  while (stream !== null && typeof stream === 'object' && typeof stream.then === 'function') {
    stream = await stream
  }
  if (!isAsyncIterable(stream)) return

  let contentSeen = false
  const openTools = new Map()

  const healSafe = () => {
    for (const entry of openTools.values()) {
      if (!(entry.deltas > 0)) return false
      try {
        JSON.parse(entry.args)
      } catch {
        return false
      }
    }
    return true
  }

  try {
    for await (const chunk of stream) {
      if (chunk !== null && typeof chunk === 'object' && typeof chunk.type === 'string') {
        if (chunk.type === 'block-start') {
          contentSeen = true
          if (chunk.blockType === 'tool-call') openTools.set(chunk.index, { args: '', deltas: 0 })
          yield chunk
          continue
        }
        if (chunk.type === 'block-end') {
          contentSeen = true
          if (chunk.block !== null && typeof chunk.block === 'object' &&
            chunk.block.type === 'tool-call') openTools.delete(chunk.index)
          yield chunk
          continue
        }
        if (chunk.type === 'tool-call-delta') {
          contentSeen = true
          const entry = openTools.get(chunk.index)
          if (entry !== undefined) {
            entry.deltas += 1
            if (typeof chunk.argumentsDelta === 'string') entry.args += chunk.argumentsDelta
          }
          yield chunk
          continue
        }
        if (chunk.type === 'text-delta' || chunk.type === 'reasoning-delta') {
          contentSeen = true
          yield chunk
          continue
        }
        if (chunk.type === 'finish') {
          const reason = chunk.reason
          const failure = reason !== null && typeof reason === 'object' ? reason.failure : undefined
          const terminalMissing = (reason?.kind === 'error' || reason?.kind === 'aborted') &&
            failure !== undefined && typeof failure === 'object' && isMissingStopError(failure)
          if (terminalMissing && contentSeen && healSafe()) return
          yield chunk
          continue
        }
      }
      yield chunk
    }
  } catch (error) {
    if (contentSeen && healSafe() && isMissingStopError(error)) return
    throw error
  }
}

const NAMESPACE = 'missing-stop-heal'

export function apply(ctx, config = {}) {
  const logger = ctx !== null && typeof ctx === 'object' && ctx.logger &&
    typeof ctx.logger.warn === 'function'
    ? ctx.logger
    : null
  const warn = (message) => {
    if (logger) logger.warn(`[missing-stop-heal] ${message}`)
    else if (typeof console !== 'undefined' && console.warn) console.warn(`[missing-stop-heal] ${message}`)
  }

  const state = { rules: normalizeRules(config, warn) }
  let warnedEmpty = false

  // 可选接线：settings 服务存在时注册命名空间（组合配置为 base，用户文档覆盖），
  // 任何一步不可用都静默回退到纯行配置。
  void (async () => {
    try {
      if (typeof ctx?.get !== 'function') return
      const settings = ctx.get('settings')
      if (settings === undefined || settings === null || typeof settings.register !== 'function') return
      const mod = await import('@deepseek-ai/schemastery')
      const Schema = mod && mod.default ? mod.default : mod
      if (typeof Schema?.object !== 'function' || typeof Schema?.array !== 'function' ||
        typeof Schema?.string !== 'function' || typeof Schema?.boolean !== 'function') return
      const RuleSchema = Schema.object({
        provider: Schema.string(),
        models: Schema.array(Schema.string()).default(['*']),
        enabled: Schema.boolean().default(true),
      })
      const ConfigSchema = Schema.object({
        providers: Schema.array(Schema.string()).default([]),
        rules: Schema.array(RuleSchema).default([]),
      })
      const scope = settings.register(NAMESPACE, ConfigSchema, { base: config, applies: 'live' })
      const sync = () => {
        const resolved = typeof scope?.get === 'function' ? scope.get() : null
        state.rules = normalizeRules(resolved ?? {}, warn)
      }
      sync()
      if (typeof scope?.watch === 'function') scope.watch(() => sync())
      warn(`已注册设置命名空间 ${NAMESPACE}，规则可在设置中调整`)
    } catch (error) {
      warn(`设置命名空间不可用，回退到组合配置：${error instanceof Error ? error.message : String(error)}`)
    }
  })()

  ctx.on('llm/stream', (options, next) => {
    if (state.rules.length === 0) {
      if (!warnedEmpty) {
        warnedEmpty = true
        warn('未配置任何生效规则（providers/rules 为空），当前不干预任何流。' +
          '可在行配置 config 或 web 设置的插件配置中填写 rules / providers。')
      }
      return next()
    }
    if (typeof options !== 'object' || options === null ||
      !ruleMatches(state.rules, options.provider, options.model)) return next()
    const downstream = next()
    const thenable = downstream !== null && typeof downstream === 'object' &&
      typeof downstream.then === 'function'
    if (!isAsyncIterable(downstream) && !thenable) return downstream
    return healStream(downstream)
  })
}

export default { name, apply }