processor.ts

import { MessageV2 } from "./message-v2"
import { Log } from "@/util/log"
import { Session } from "."
import { Agent } from "@/agent/agent"
import { Snapshot } from "@/snapshot"
import { SessionSummary } from "./summary"
import { Bus } from "@/bus"
import { SessionRetry } from "./retry"
import { SessionStatus } from "./status"
import { Plugin } from "@/plugin"
import type { Provider } from "@/provider/provider"
import { LLM } from "./llm"
import { Config } from "@/config/config"
import { SessionCompaction } from "./compaction"
import { PermissionNext } from "@/permission/next"
import { Question } from "@/question"
import { PartID } from "./schema"
import type { SessionID, MessageID } from "./schema"
// altimate_change start — import Telemetry for per-generation token tracking
import { Telemetry } from "@/altimate/telemetry"
// altimate_change end

export namespace SessionProcessor {
  const DOOM_LOOP_THRESHOLD = 3
  // altimate_change start — per-tool repeat threshold to catch varied-input loops (e.g. todowrite 2,080x)
  // Legitimate tool use rarely exceeds 20-25 calls per tool per session.
  // 30 catches pathological patterns while avoiding false positives for power users.
  const TOOL_REPEAT_THRESHOLD = 30
  // altimate_change end
  // altimate_change start — escalating circuit breaker for doom loops
  // When the repeat threshold is hit and auto-accepted (headless, config allow), the
  // counter resets and the loop continues indefinitely. Escalation levels:
  //   1st hit (30 calls):  ask permission (existing behavior)
  //   2nd hit (60 calls):  ask + inject synthetic warning telling model to change approach
  //   3rd hit (90 calls):  force-stop the session — the model is stuck
  const DOOM_LOOP_WARN_ESCALATION = 2 // hits before injecting warning
  const DOOM_LOOP_STOP_ESCALATION = 3 // hits before force-stopping
  // altimate_change end
  const log = Log.create({ service: "session.processor" })

  export type Info = Awaited<ReturnType<typeof create>>
  export type Result = Awaited<ReturnType<Info["process"]>>

  export function create(input: {
    assistantMessage: MessageV2.Assistant
    sessionID: SessionID
    model: Provider.Model
    abort: AbortSignal
  }) {
    const toolcalls: Record<string, MessageV2.ToolPart> = {}
    // altimate_change start — per-tool call counter for varied-input loop detection
    const toolCallCounts: Record<string, number> = {}
    // altimate_change end
    // altimate_change start — escalation counter: how many times each tool has hit TOOL_REPEAT_THRESHOLD
    const toolLoopHits: Record<string, number> = {}
    // altimate_change end
    let snapshot: string | undefined
    let blocked = false
    let attempt = 0
    let needsCompaction = false
    // altimate_change start — per-step generation telemetry
    let stepStartTime = Date.now()
    // altimate_change end
    // altimate_change start — plan-agent tool-call-refusal detection
    // Some models (observed: qwen3-coder-next, occasionally gpt-5.4) end plan-agent
    // steps with finish_reason=stop and never emit tool calls. User abandons the
    // session thinking it's stuck. Track whether the session has ever produced a
    // tool call; if plan agent finishes its first step with stop-no-tools, warn.
    let sessionToolCallsMade = 0
    let planNoToolWarningEmitted = false
    // altimate_change end

    const result = {
      get message() {
        return input.assistantMessage
      },
      partFromToolCall(toolCallID: string) {
        return toolcalls[toolCallID]
      },
      async process(streamInput: LLM.StreamInput) {
        log.info("process")
        needsCompaction = false
        const shouldBreak = (await Config.get()).experimental?.continue_loop_on_deny !== true
        while (true) {
          try {
            let currentText: MessageV2.TextPart | undefined
            let reasoningMap: Record<string, MessageV2.ReasoningPart> = {}
            const stream = await LLM.stream(streamInput)

            for await (const value of stream.fullStream) {
              input.abort.throwIfAborted()
              switch (value.type) {
                case "start":
                  SessionStatus.set(input.sessionID, { type: "busy" })
                  break

                case "reasoning-start":
                  if (value.id in reasoningMap) {
                    continue
                  }
                  const reasoningPart = {
                    id: PartID.ascending(),
                    messageID: input.assistantMessage.id,
                    sessionID: input.assistantMessage.sessionID,
                    type: "reasoning" as const,
                    text: "",
                    time: {
                      start: Date.now(),
                    },
                    metadata: value.providerMetadata,
                  }
                  reasoningMap[value.id] = reasoningPart
                  await Session.updatePart(reasoningPart)
                  break

                case "reasoning-delta":
                  if (value.id in reasoningMap) {
                    const part = reasoningMap[value.id]
                    part.text += value.text
                    if (value.providerMetadata) part.metadata = value.providerMetadata
                    await Session.updatePartDelta({
                      sessionID: part.sessionID,
                      messageID: part.messageID,
                      partID: part.id,
                      field: "text",
                      delta: value.text,
                    })
                  }
                  break

                case "reasoning-end":
                  if (value.id in reasoningMap) {
                    const part = reasoningMap[value.id]
                    part.text = part.text.trimEnd()

                    part.time = {
                      ...part.time,
                      end: Date.now(),
                    }
                    if (value.providerMetadata) part.metadata = value.providerMetadata
                    await Session.updatePart(part)
                    delete reasoningMap[value.id]
                  }
                  break

                case "tool-input-start":
                  const part = await Session.updatePart({
                    id: toolcalls[value.id]?.id ?? PartID.ascending(),
                    messageID: input.assistantMessage.id,
                    sessionID: input.assistantMessage.sessionID,
                    type: "tool",
                    tool: value.toolName,
                    callID: value.id,
                    state: {
                      status: "pending",
                      input: {},
                      raw: "",
                    },
                  })
                  toolcalls[value.id] = part as MessageV2.ToolPart
                  break

                case "tool-input-delta":
                  break

                case "tool-input-end":
                  break

                case "tool-call": {
                  const match = toolcalls[value.toolCallId]
                  if (match) {
                    const part = await Session.updatePart({
                      ...match,
                      tool: value.toolName,
                      state: {
                        status: "running",
                        input: value.input,
                        time: {
                          start: Date.now(),
                        },
                      },
                      metadata: value.providerMetadata,
                    })
                    toolcalls[value.toolCallId] = part as MessageV2.ToolPart
                    // altimate_change start — session has now tool-called; suppresses plan refusal warning
                    sessionToolCallsMade++
                    // altimate_change end

                    const parts = await MessageV2.parts(input.assistantMessage.id)
                    const lastThree = parts.slice(-DOOM_LOOP_THRESHOLD)

                    if (
                      lastThree.length === DOOM_LOOP_THRESHOLD &&
                      lastThree.every(
                        (p) =>
                          p.type === "tool" &&
                          p.tool === value.toolName &&
                          p.state.status !== "pending" &&
                          JSON.stringify(p.state.input) === JSON.stringify(value.input),
                      )
                    ) {
                      const agent = await Agent.get(input.assistantMessage.agent)
                      await PermissionNext.ask({
                        permission: "doom_loop",
                        patterns: [value.toolName],
                        sessionID: input.assistantMessage.sessionID,
                        metadata: {
                          tool: value.toolName,
                          input: value.input,
                        },
                        always: [value.toolName],
                        ruleset: agent.permission,
                      })
                    }

                    // altimate_change start — per-tool repeat counter with escalating circuit breaker
                    // Counter is scoped to the processor lifetime (create() call), so it accumulates
                    // across multiple process() invocations within a session. This is intentional:
                    // cross-turn accumulation catches slow-burn loops that stay under the threshold
                    // per-turn but add up over the session.
                    toolCallCounts[value.toolName] = (toolCallCounts[value.toolName] ?? 0) + 1
                    if (toolCallCounts[value.toolName] >= TOOL_REPEAT_THRESHOLD) {
                      toolLoopHits[value.toolName] = (toolLoopHits[value.toolName] ?? 0) + 1
                      const hits = toolLoopHits[value.toolName]
                      const totalCalls = hits * TOOL_REPEAT_THRESHOLD

                      Telemetry.track({
                        type: "doom_loop_detected",
                        timestamp: Date.now(),
                        session_id: input.sessionID,
                        tool_name: value.toolName,
                        repeat_count: totalCalls,
                        escalation_level: hits,
                      })

                      // Escalation level 3+: force-stop — the model is irretrievably stuck
                      if (hits >= DOOM_LOOP_STOP_ESCALATION) {
                        log.warn("doom loop circuit breaker: force-stopping session", {
                          tool: value.toolName,
                          totalCalls,
                          hits,
                          sessionID: input.sessionID,
                        })
                        await Session.updatePart({
                          id: PartID.ascending(),
                          messageID: input.assistantMessage.id,
                          sessionID: input.assistantMessage.sessionID,
                          type: "text",
                          synthetic: true,
                          text:
                            `⚠️ altimate-code: session stopped — \`${value.toolName}\` was called ${totalCalls} times, ` +
                            `indicating the agent is stuck in a loop. Please start a new session with a revised prompt.`,
                          time: { start: Date.now(), end: Date.now() },
                        })
                        blocked = true
                        toolCallCounts[value.toolName] = 0
                        toolLoopHits[value.toolName] = 0
                        break
                      }

                      // Escalation level 2: warn the model via synthetic message
                      if (hits >= DOOM_LOOP_WARN_ESCALATION) {
                        log.warn("doom loop escalation: injecting warning", {
                          tool: value.toolName,
                          totalCalls,
                          hits,
                          sessionID: input.sessionID,
                        })
                        await Session.updatePart({
                          id: PartID.ascending(),
                          messageID: input.assistantMessage.id,
                          sessionID: input.assistantMessage.sessionID,
                          type: "text",
                          synthetic: true,
                          text:
                            `⚠️ altimate-code: \`${value.toolName}\` has been called ${totalCalls} times this session. ` +
                            `You appear to be stuck in a loop. Stop repeating the same approach. ` +
                            `Either try a fundamentally different strategy or explain to the user what is blocking you. ` +
                            `The session will be force-stopped if this continues.`,
                          time: { start: Date.now(), end: Date.now() },
                        })
                      }

                      // Escalation level 1: ask permission (existing behavior)
                      // Reset before ask so denial/exception doesn't leave count at threshold
                      toolCallCounts[value.toolName] = 0
                      const agent = await Agent.get(input.assistantMessage.agent)
                      await PermissionNext.ask({
                        permission: "doom_loop",
                        patterns: [value.toolName],
                        sessionID: input.assistantMessage.sessionID,
                        metadata: {
                          tool: value.toolName,
                          input: value.input,
                          repeat_count: totalCalls,
                        },
                        always: [value.toolName],
                        ruleset: agent.permission,
                      })
                    }
                    // altimate_change end
                  }
                  break
                }
                case "tool-result": {
                  const match = toolcalls[value.toolCallId]
                  if (match && match.state.status === "running") {
                    await Session.updatePart({
                      ...match,
                      state: {
                        status: "completed",
                        input: value.input ?? match.state.input,
                        output: value.output.output,
                        metadata: value.output.metadata,
                        title: value.output.title,
                        time: {
                          start: match.state.time.start,
                          end: Date.now(),
                        },
                        attachments: value.output.attachments,
                      },
                    })

                    delete toolcalls[value.toolCallId]
                  }
                  break
                }

                case "tool-error": {
                  const match = toolcalls[value.toolCallId]
                  if (match && match.state.status === "running") {
                    await Session.updatePart({
                      ...match,
                      state: {
                        status: "error",
                        input: value.input ?? match.state.input,
                        error: (value.error as any).toString(),
                        time: {
                          start: match.state.time.start,
                          end: Date.now(),
                        },
                      },
                    })

                    if (
                      value.error instanceof PermissionNext.RejectedError ||
                      value.error instanceof Question.RejectedError
                    ) {
                      blocked = shouldBreak
                    }
                    delete toolcalls[value.toolCallId]
                  }
                  break
                }
                case "error":
                  throw value.error

                case "start-step":
                  snapshot = await Snapshot.track()
                  // altimate_change start — record step start time for generation telemetry duration
                  stepStartTime = Date.now()
                  // altimate_change end
                  await Session.updatePart({
                    id: PartID.ascending(),
                    messageID: input.assistantMessage.id,
                    sessionID: input.sessionID,
                    snapshot,
                    type: "step-start",
                  })
                  break

                case "finish-step":
                  const usage = Session.getUsage({
                    model: input.model,
                    usage: value.usage,
                    metadata: value.providerMetadata,
                  })
                  input.assistantMessage.finish = value.finishReason
                  input.assistantMessage.cost += usage.cost
                  input.assistantMessage.tokens = usage.tokens
                  // altimate_change start — emit per-generation telemetry with token breakdown
                  // Optional fields are only included when the provider actually returns them.
                  Telemetry.track({
                    type: "generation",
                    timestamp: Date.now(),
                    session_id: input.sessionID,
                    message_id: input.assistantMessage.id,
                    model_id: input.model.id,
                    provider_id: input.model.providerID,
                    agent: input.assistantMessage.agent,
                    finish_reason: value.finishReason ?? "unknown",
                    cost: usage.cost,
                    duration_ms: Date.now() - stepStartTime,
                    tokens_input: usage.tokens.input,
                    tokens_output: usage.tokens.output,
                    // altimate_change start — include total input tokens (with cache) when they differ from tokens_input
                    ...(usage.tokens.inputTotal !== usage.tokens.input && { tokens_input_total: usage.tokens.inputTotal }),
                    // altimate_change end
                    ...(value.usage.reasoningTokens !== undefined && { tokens_reasoning: usage.tokens.reasoning }),
                    ...(value.usage.cachedInputTokens !== undefined && { tokens_cache_read: usage.tokens.cache.read }),
                    ...(usage.tokens.cache.write > 0 && { tokens_cache_write: usage.tokens.cache.write }),
                  })
                  // altimate_change end
                  // altimate_change start — detect plan-agent tool-call refusal
                  // A plan-agent step that ends with finish=stop and NO tool calls
                  // (ever) in the session means the model wrote text and gave up.
                  // Users read the text, see no progress, and abandon. Surface a
                  // warning + telemetry so the pattern is measurable and the user
                  // knows to try a different model.
                  if (
                    input.assistantMessage.agent === "plan" &&
                    value.finishReason === "stop" &&
                    sessionToolCallsMade === 0 &&
                    !planNoToolWarningEmitted
                  ) {
                    planNoToolWarningEmitted = true
                    Telemetry.track({
                      type: "plan_no_tool_generation",
                      timestamp: Date.now(),
                      session_id: input.sessionID,
                      message_id: input.assistantMessage.id,
                      model_id: input.model.id,
                      provider_id: input.model.providerID,
                      finish_reason: value.finishReason,
                      tokens_output: usage.tokens.output,
                    })
                    log.warn("plan agent stopped without tool calls — model may not be tool-calling properly", {
                      sessionID: input.sessionID,
                      modelID: input.model.id,
                      providerID: input.model.providerID,
                      tokensOutput: usage.tokens.output,
                    })
                    // synthetic: true so this warning is shown in the TUI but
                    // excluded when the transcript is replayed to the LLM next turn
                    // (prompt.ts filters synthetic text parts — see lines 648, 795).
                    await Session.updatePart({
                      id: PartID.ascending(),
                      messageID: input.assistantMessage.id,
                      sessionID: input.assistantMessage.sessionID,
                      type: "text",
                      synthetic: true,
                      text:
                        `⚠️ altimate-code: the \`plan\` agent is running on \`${input.model.providerID}/${input.model.id}\`, ` +
                        `which returned text without calling any tools. If you expected the plan agent to explore the ` +
                        `codebase, try switching to a model with stronger tool-use via \`/model\`.`,
                      time: { start: Date.now(), end: Date.now() },
                    })
                  }
                  // altimate_change end
                  await Session.updatePart({
                    id: PartID.ascending(),
                    reason: value.finishReason,
                    snapshot: await Snapshot.track(),
                    messageID: input.assistantMessage.id,
                    sessionID: input.assistantMessage.sessionID,
                    type: "step-finish",
                    tokens: usage.tokens,
                    cost: usage.cost,
                  })
                  await Session.updateMessage(input.assistantMessage)
                  if (snapshot) {
                    const patch = await Snapshot.patch(snapshot)
                    if (patch.files.length) {
                      await Session.updatePart({
                        id: PartID.ascending(),
                        messageID: input.assistantMessage.id,
                        sessionID: input.sessionID,
                        type: "patch",
                        hash: patch.hash,
                        files: patch.files,
                      })
                    }
                    snapshot = undefined
                  }
                  SessionSummary.summarize({
                    sessionID: input.sessionID,
                    messageID: input.assistantMessage.parentID,
                  })
                  if (
                    !input.assistantMessage.summary &&
                    (await SessionCompaction.isOverflow({ tokens: usage.tokens, model: input.model }))
                  ) {
                    needsCompaction = true
                  }
                  break

                case "text-start":
                  currentText = {
                    id: PartID.ascending(),
                    messageID: input.assistantMessage.id,
                    sessionID: input.assistantMessage.sessionID,
                    type: "text",
                    text: "",
                    time: {
                      start: Date.now(),
                    },
                    metadata: value.providerMetadata,
                  }
                  await Session.updatePart(currentText)
                  break

                case "text-delta":
                  if (currentText) {
                    currentText.text += value.text
                    if (value.providerMetadata) currentText.metadata = value.providerMetadata
                    await Session.updatePartDelta({
                      sessionID: currentText.sessionID,
                      messageID: currentText.messageID,
                      partID: currentText.id,
                      field: "text",
                      delta: value.text,
                    })
                  }
                  break

                case "text-end":
                  if (currentText) {
                    currentText.text = currentText.text.trimEnd()
                    const textOutput = await Plugin.trigger(
                      "experimental.text.complete",
                      {
                        sessionID: input.sessionID,
                        messageID: input.assistantMessage.id,
                        partID: currentText.id,
                      },
                      { text: currentText.text },
                    )
                    currentText.text = textOutput.text
                    currentText.time = {
                      start: currentText.time?.start ?? Date.now(),
                      end: Date.now(),
                    }
                    if (value.providerMetadata) currentText.metadata = value.providerMetadata
                    await Session.updatePart(currentText)
                  }
                  currentText = undefined
                  break

                case "finish":
                  break

                default:
                  log.info("unhandled", {
                    ...value,
                  })
                  continue
              }
              if (needsCompaction) break
              // altimate_change start — exit stream loop immediately on doom loop force-stop
              if (blocked) break
              // altimate_change end
            }
          } catch (e: any) {
            log.error("process", {
              error: e,
              stack: JSON.stringify(e.stack),
            })
            const error = MessageV2.fromError(e, { providerID: input.model.providerID })
            if (MessageV2.ContextOverflowError.isInstance(error)) {
              needsCompaction = true
              Bus.publish(Session.Event.Error, {
                sessionID: input.sessionID,
                error,
              })
            } else {
              const retry = SessionRetry.retryable(error)
              // altimate_change start — cap retries to avoid infinite loops, log on exhaustion
              if (retry !== undefined && attempt < SessionRetry.RETRY_MAX_ATTEMPTS) {
              // altimate_change end
                attempt++
                const delay = SessionRetry.delay(attempt, error.name === "APIError" ? error : undefined)
                SessionStatus.set(input.sessionID, {
                  type: "retry",
                  attempt,
                  message: retry,
                  next: Date.now() + delay,
                })
                await SessionRetry.sleep(delay, input.abort).catch(() => {})
                continue
              }
              // altimate_change start — log when retries exhausted for debugging
              if (retry !== undefined) {
                log.warn("max retry attempts reached, giving up", {
                  attempt,
                  message: retry,
                  providerID: input.model.providerID,
                  modelID: input.model.id,
                })
              }
              // altimate_change end
              input.assistantMessage.error = error
              Bus.publish(Session.Event.Error, {
                sessionID: input.assistantMessage.sessionID,
                error: input.assistantMessage.error,
              })
              SessionStatus.set(input.sessionID, { type: "idle" })
            }
          }
          if (snapshot) {
            const patch = await Snapshot.patch(snapshot)
            if (patch.files.length) {
              await Session.updatePart({
                id: PartID.ascending(),
                messageID: input.assistantMessage.id,
                sessionID: input.sessionID,
                type: "patch",
                hash: patch.hash,
                files: patch.files,
              })
            }
            snapshot = undefined
          }
          const p = await MessageV2.parts(input.assistantMessage.id)
          for (const part of p) {
            if (part.type === "tool" && part.state.status !== "completed" && part.state.status !== "error") {
              await Session.updatePart({
                ...part,
                state: {
                  ...part.state,
                  status: "error",
                  error: "Tool execution aborted",
                  time: {
                    start: Date.now(),
                    end: Date.now(),
                  },
                },
              })
            }
          }
          input.assistantMessage.time.completed = Date.now()
          await Session.updateMessage(input.assistantMessage)
          if (needsCompaction) return "compact"
          if (blocked) return "stop"
          if (input.assistantMessage.error) return "stop"
          return "continue"
        }
      },
    }
    return result
  }
}