import { isAbsolute } from 'node:path'

import {
	ACP_CLIENT_NOTIFICATIONS,
	ACP_CLIENT_REQUESTS,
	ACP_ERROR_CODES,
	ACP_FILESYSTEM_CAPABILITY,
	ACP_METHODS,
	ACP_PERMISSION_CAPABILITY,
	ACP_PROTOCOL_VERSION,
} from '../../constants/acp/index.js'
import type { HostCommandRegistry } from '../../registry/command/index.js'
import type { ToolPresenter } from '../../registry/tool/presentation.js'
import type {
	AcpFsReadResult,
	AcpInitializeParams,
	AcpInitializeResult,
	AcpRequestPermissionResult,
	AcpSessionCancelParams,
	AcpSessionLoadParams,
	AcpSessionNewParams,
	AcpSessionNewResult,
	AcpSessionPromptParams,
	AcpSessionPromptResult,
	AcpSessionUpdate,
} from '../../types/acp/index.js'
import type { MCPJsonRpcMessage, MCPTransport } from '../../types/connector/mcp.js'
import type { SessionEvent } from '../../types/session/events.js'
import { isTurnInProgressError } from '../../types/session/turn.js'
import { generateSessionId } from '../../utils/id.js'
import { type Logger, resolveLogger } from '../../utils/logger.js'
import type { AcpClientFilesystem } from './filesystem.js'
import type {
	AcpPermissionAsker,
	AcpPermissionOutcome,
	AcpPermissionRequest,
} from './permission.js'
import { toAcpSessionUpdate, toAcpStopReason } from './update.js'

/**
 * An agent-client protocol server over stdio.
 *
 * An editor extension or a CI orchestrator could previously do two things:
 * shell out to the CLI and scrape its output, or embed the SDK in its own
 * process. This is the third — a wire surface a peer written in any language
 * can drive.
 *
 * **The precedent in this tree is a warning, and this change answers it.**
 * `MCPServer` and `ServerStdioTransport` are both exported, and nothing
 * anywhere constructs an `MCPServer`: a complete protocol server with no
 * driver. So `packages/cli/src/commands/acp.ts` ships in the same change,
 * and a subprocess test spawns the real binary — the wire half alone is not
 * a deliverable.
 *
 * **stdout belongs to the protocol.** `ServerStdioTransport`'s own header
 * says it and this is the surface that pays for it: one stray `console.log`
 * anywhere in the process and a client reports malformed JSON with nothing
 * naming the culprit. This repository's logger writes to stderr, and a test
 * asserts zero non-JSON bytes on the child's stdout under info-level
 * logging.
 */

/** What the bridge needs from the runtime, taken as an interface. */
export interface AcpAgentGateway {
	/**
	 * Run one prompt as one turn of the session, streaming its events, and
	 * resolve with the stop reason.
	 *
	 * A session has one active turn at a time. A gateway whose session log
	 * refuses the turn because another is active (in this process or another)
	 * throws the `TurnInProgressError`; the bridge answers the prompt with
	 * `INVALID_REQUEST` naming the active turn.
	 *
	 * Deliberately not `AgentManagerContract` itself. This bridge needs one
	 * verb, and a session front end holding the whole manager could cancel
	 * somebody else's task, spawn children, or drain a queue it does not own
	 * — none of which a peer asked for. The CLI passes an adapter over
	 * `sendMessage`/`cancel`.
	 */
	prompt(request: {
		readonly sessionId: string
		readonly prompt: string
		readonly cwd: string
		readonly onEvent: (event: SessionEvent) => void
		readonly signal: AbortSignal
		/**
		 * Ask the human in front of the client about a tool batch.
		 *
		 * Handed to the gateway rather than installed by it, so the ONE place
		 * that knows how to reach the client is this bridge. A gateway that
		 * built its own asker would be a second path to the same human, and
		 * the one that forgot to latch `approve_all` would be it.
		 */
		readonly ask: AcpPermissionAsker
		/**
		 * The client's buffers, when it declared the capability. `undefined`
		 * means disk — which is correct, and is what every non-editor peer
		 * wants.
		 */
		readonly filesystem: AcpClientFilesystem | undefined
		/** Turns to resume from, oldest first. Empty for a fresh session. */
		readonly history: readonly unknown[]
	}): Promise<{
		readonly stopReason?: string
		/**
		 * The exact conversation to use for the next prompt, after the turn has
		 * settled. Omit it when the gateway does not own durable history.
		 */
		readonly history?: readonly unknown[]
	}>

	/**
	 * The turns a prior session left behind, for `session/load`.
	 *
	 * Optional: a gateway with no session store cannot resume, and saying so
	 * by not implementing this is better than returning an empty history that
	 * a client cannot tell apart from a session that really had no turns.
	 *
	 * Resolves `undefined` when the store has no session by that id, which
	 * the bridge answers with `INVALID_PARAMS` naming the id. The id is the
	 * client's and may be any string: a namzu session id, or one the gateway
	 * maps to a session through the index's `acp` / `session` refs.
	 */
	load?(sessionId: string): Promise<readonly unknown[] | undefined>
}

export interface AcpServerOptions {
	readonly transport: MCPTransport
	readonly gateway: AcpAgentGateway
	/**
	 * The command surface, verbatim from the registry.
	 *
	 * Passed as the registry rather than as a list so `describe()` is called
	 * per initialize: a host that registers a command after construction
	 * still has it appear, and this module has no place to hard-code one.
	 */
	readonly commands: HostCommandRegistry
	readonly presenter: ToolPresenter
	readonly agentInfo: { readonly name: string; readonly version: string }
	/**
	 * The id `session/new` answers with. Defaults to a new namzu session id
	 * (UUIDv7). A host may return any string; the bridge treats it as opaque,
	 * and injecting it also keeps a test off a random id.
	 */
	readonly newSessionId?: () => string
	readonly log?: Logger
}

interface Session {
	readonly cwd: string
	controller: AbortController
	/** One live prompt per protocol session. Other sessions remain independent. */
	promptInFlight: boolean
	/**
	 * Whether the human said "approve all" for THIS session.
	 *
	 * Per session, on the session record, and that placement is the whole
	 * property: hoisting it to the server — or to a module-level variable —
	 * would make one person's "stop asking me" silently cover the next
	 * session this process serves, which may be a different repository, a
	 * different editor window, or a different human.
	 */
	approveAll: boolean
	/** Prior turns, when this session was loaded rather than created. */
	history: readonly unknown[]
}

type Handler = (params: Record<string, unknown>) => Promise<unknown> | unknown

/** A JSON-RPC error this bridge answers with, rather than throwing out. */
class AcpError extends Error {
	constructor(
		readonly code: number,
		message: string,
	) {
		super(message)
		this.name = 'AcpError'
	}
}

export class ACPServer {
	private readonly log: Logger
	private readonly sessions = new Map<string, Session>()
	/** IDs being loaded or created but not yet published. */
	private readonly reservedSessionIds = new Map<string, symbol>()
	private initialized = false
	private stopped = false
	private clientCapabilities: readonly string[] = []

	/**
	 * The method table, authored INDEPENDENTLY of `ACP_METHODS`.
	 *
	 * Deriving it from the constant would make the drift test a tautology.
	 * Two hand-written sets compared in both directions is the only shape
	 * where "a handler nobody advertises" and "an advertised method with no
	 * handler" are both catchable.
	 */
	private readonly handlers: Readonly<Record<string, Handler>> = {
		initialize: (p) => this.onInitialize(p as unknown as AcpInitializeParams),
		'session/new': (p) => this.onSessionNew(p as unknown as AcpSessionNewParams),
		'session/prompt': (p) => this.onSessionPrompt(p as unknown as AcpSessionPromptParams),
		'session/cancel': (p) => this.onSessionCancel(p as unknown as AcpSessionCancelParams),
		'session/load': (p) => this.onSessionLoad(p as unknown as AcpSessionLoadParams),
	}

	/** Requests this side has out to the client, keyed by their id. */
	private readonly pending = new Map<
		string | number,
		{ resolve: (value: unknown) => void; reject: (err: Error) => void }
	>()

	private requestSeq = 0

	constructor(private readonly options: AcpServerOptions) {
		this.log = resolveLogger(options.log).child({ 'namzu.log.scope': 'bridge/acp' })
	}

	/** The method names this server answers. For the drift test. */
	methodNames(): readonly string[] {
		return Object.keys(this.handlers).sort()
	}

	async start(): Promise<void> {
		if (this.stopped) throw new Error('An ACP server cannot be restarted after it has stopped.')
		this.options.transport.onMessage((message) => {
			void this.dispatch(message)
		})
		await this.options.transport.connect()
	}

	async stop(): Promise<void> {
		this.stopped = true
		// Anything still waiting on the client is rejected rather than left
		// pending: the transport is about to close, so no answer is coming, and
		// a promise nobody will ever settle keeps whatever awaited it alive.
		for (const waiting of this.pending.values()) {
			waiting.reject(new Error('The client connection closed before it answered.'))
		}
		this.pending.clear()
		for (const session of this.sessions.values()) session.controller.abort()
		this.sessions.clear()
		this.reservedSessionIds.clear()
		await this.options.transport.close()
	}

	/**
	 * Ask the client something and wait for its answer.
	 *
	 * The direction this bridge did not have. A notification is fire and
	 * forget; a permission prompt is a question the turn cannot proceed past,
	 * so it needs an id, a place to park the promise, and a `dispatch` that
	 * recognises a response frame.
	 */
	private request<T>(
		method: string,
		params: Record<string, unknown>,
		signal?: AbortSignal,
	): Promise<T> {
		this.requestSeq += 1
		const id = `agent_${this.requestSeq}`
		return new Promise<T>((resolve, reject) => {
			const finish = (): boolean => {
				if (this.pending.get(id) !== entry) return false
				this.pending.delete(id)
				signal?.removeEventListener('abort', onAbort)
				return true
			}
			const onAbort = (): void => {
				const reason =
					signal?.reason instanceof Error
						? signal.reason
						: new DOMException('The permission request was cancelled.', 'AbortError')
				entry.reject(reason)
			}
			const entry = {
				resolve: (value: unknown) => {
					if (finish()) resolve(value as T)
				},
				reject: (error: Error) => {
					if (finish()) reject(error)
				},
			}
			this.pending.set(id, entry)
			signal?.addEventListener('abort', onAbort, { once: true })
			void this.send({ jsonrpc: '2.0', id, method, params }).catch((err: unknown) => {
				entry.reject(err instanceof Error ? err : new Error(String(err)))
			})
		})
	}

	private async dispatch(message: MCPJsonRpcMessage): Promise<void> {
		if (!message.method) {
			// A frame with no method is the client ANSWERING something this side
			// asked. Before permission requests existed there was nothing out on
			// the wire, so ignoring it was right; now dropping it would leave the
			// asker waiting forever and the turn parked with nobody coming.
			if (message.id !== undefined && this.pending.has(message.id)) {
				const waiting = this.pending.get(message.id)
				if (message.error) {
					waiting?.reject(new Error(message.error.message))
				} else {
					waiting?.resolve(message.result)
				}
			}
			return
		}

		const handler = this.handlers[message.method]
		if (!handler) {
			// Answered, and the CONNECTION STAYS OPEN. A client probing for a
			// feature must not lose its session because this agent does not have
			// it yet.
			await this.respondError(
				message.id,
				ACP_ERROR_CODES.METHOD_NOT_FOUND,
				`Unknown method "${message.method}". This agent implements: ${this.methodNames().join(', ')}.`,
			)
			return
		}

		try {
			const result = await handler(message.params ?? {})
			if (message.id !== undefined) {
				await this.send({ jsonrpc: '2.0', id: message.id, result })
			}
		} catch (err) {
			const code = err instanceof AcpError ? err.code : ACP_ERROR_CODES.INTERNAL_ERROR
			await this.respondError(message.id, code, err instanceof Error ? err.message : String(err))
		}
	}

	private async respondError(
		id: MCPJsonRpcMessage['id'],
		code: number,
		message: string,
	): Promise<void> {
		// A notification (no id) that failed has nowhere to send an error, so
		// it is logged instead of dropped silently — on stderr, where it does
		// not corrupt the protocol stream.
		if (id === undefined) {
			this.log.warn('an acp notification failed', {
				'namzu.acp.error_code': code,
				'namzu.acp.error_message': message,
			})
			return
		}
		await this.send({ jsonrpc: '2.0', id, error: { code, message } })
	}

	private async send(message: MCPJsonRpcMessage): Promise<void> {
		try {
			await this.options.transport.send(message)
		} catch (err) {
			// The client hung up mid-write. Nothing to recover; a throw here
			// would escape into whichever handler happened to be running.
			this.log.warn('acp send failed', {
				'namzu.acp.error_message': err instanceof Error ? err.message : String(err),
			})
		}
	}

	private async notifyUpdate(sessionId: string, update: AcpSessionUpdate): Promise<void> {
		await this.send({
			jsonrpc: '2.0',
			method: ACP_CLIENT_NOTIFICATIONS.SESSION_UPDATE,
			params: { sessionId, update } as unknown as Record<string, unknown>,
		})
	}

	private onInitialize(params: AcpInitializeParams): AcpInitializeResult {
		this.initialized = true
		this.clientCapabilities = params.capabilities ?? []
		return {
			protocolVersion: ACP_PROTOCOL_VERSION,
			agentInfo: this.options.agentInfo,
			// From the registry, per call. A hard-coded list here would be a
			// second definition of the command surface.
			commands: this.options.commands.describe(),
			requiredClientCapabilities: [ACP_PERMISSION_CAPABILITY],
			// Advertised so a client author can see the option exists. Separate
			// from `required` on purpose: a peer that is not an editor has no
			// buffers, and demanding this of it would refuse a session that is
			// perfectly able to run.
			optionalClientCapabilities: [ACP_FILESYSTEM_CAPABILITY],
		}
	}

	private onSessionNew(params: AcpSessionNewParams): AcpSessionNewResult {
		this.requireInitialized()
		this.requirePermissionCapability()
		const cwd = this.requireAbsoluteCwd(params.cwd)
		const { sessionId, token } = this.reserveGeneratedSessionId()
		// No await separates reserve from publish, so another operation cannot
		// take or close this namespace slot between the two statements. Load has
		// an async store read and therefore keeps its explicit failure cleanup.
		this.publishSession(sessionId, token, cwd, [])
		return { sessionId }
	}

	/**
	 * Resume a session the store already has.
	 *
	 * The prior turns come from the gateway, never from this bridge: the
	 * history lives in whatever session store the host wired up, and a bridge
	 * that kept its own copy would answer a resume with the turns THIS process
	 * happened to see rather than the ones the session actually had.
	 */
	private async onSessionLoad(params: AcpSessionLoadParams): Promise<AcpSessionNewResult> {
		this.requireInitialized()
		this.requirePermissionCapability()
		if (!this.options.gateway.load) {
			throw new AcpError(
				ACP_ERROR_CODES.METHOD_NOT_FOUND,
				'This agent has no session store, so there is nothing to resume. Create a new session instead.',
			)
		}
		const cwd = this.requireAbsoluteCwd(params.cwd)
		const token = this.reserveSessionId(params.sessionId)
		try {
			const history = await this.options.gateway.load(params.sessionId)
			if (history === undefined) {
				throw new AcpError(
					ACP_ERROR_CODES.INVALID_PARAMS,
					`There is no session "${params.sessionId}" to load. Create a new session instead.`,
				)
			}
			if (!Array.isArray(history)) {
				throw new AcpError(
					ACP_ERROR_CODES.INTERNAL_ERROR,
					`The session store returned an invalid history for "${params.sessionId}".`,
				)
			}
			// The SAME id, not a new one. A client that asked to resume `ses_x` and
			// got `ses_y` back has to rewrite whatever it had keyed by the old one.
			this.publishSession(params.sessionId, token, cwd, history)
			return { sessionId: params.sessionId }
		} catch (error) {
			this.releaseSessionId(params.sessionId, token)
			throw error
		}
	}

	private requireAbsoluteCwd(cwd: string | undefined): string {
		const resolved = cwd ?? process.cwd()
		if (!isAbsolute(resolved)) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_PARAMS,
				`ACP session cwd must be an absolute path; received ${JSON.stringify(resolved)}.`,
			)
		}
		return resolved
	}

	private reserveGeneratedSessionId(): { readonly sessionId: string; readonly token: symbol } {
		if (this.options.newSessionId) {
			const sessionId = this.options.newSessionId()
			return { sessionId, token: this.reserveSessionId(sessionId) }
		}

		// A namzu session id, so the gateway can open the session log under the
		// very id the client holds. A collision with an open or reserved id is
		// astronomically unlikely, and minting again is cheaper than reasoning
		// about it.
		for (;;) {
			const sessionId: string = generateSessionId()
			if (this.sessions.has(sessionId) || this.reservedSessionIds.has(sessionId)) continue
			return { sessionId, token: this.reserveSessionId(sessionId) }
		}
	}

	private reserveSessionId(sessionId: string): symbol {
		if (this.stopped) {
			throw new AcpError(ACP_ERROR_CODES.INVALID_REQUEST, 'The ACP connection has closed.')
		}
		if (!sessionId || this.sessions.has(sessionId) || this.reservedSessionIds.has(sessionId)) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_PARAMS,
				`Session id "${sessionId}" is already open or being loaded on this connection.`,
			)
		}
		const token = Symbol(sessionId)
		this.reservedSessionIds.set(sessionId, token)
		return token
	}

	private releaseSessionId(sessionId: string, token: symbol): void {
		if (this.reservedSessionIds.get(sessionId) === token) this.reservedSessionIds.delete(sessionId)
	}

	private publishSession(
		sessionId: string,
		token: symbol,
		cwd: string,
		history: readonly unknown[],
	): void {
		if (this.stopped || this.reservedSessionIds.get(sessionId) !== token) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_REQUEST,
				`Session "${sessionId}" could not be opened because the connection closed.`,
			)
		}
		this.sessions.set(sessionId, {
			cwd,
			controller: new AbortController(),
			promptInFlight: false,
			// Fresh per session. See `Session.approveAll`.
			approveAll: false,
			history: [...history],
		})
		this.releaseSessionId(sessionId, token)
	}

	private requireInitialized(): void {
		if (!this.initialized) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_REQUEST,
				`Call "${ACP_METHODS.INITIALIZE}" before creating a session.`,
			)
		}
	}

	private requirePermissionCapability(): void {
		if (!this.clientCapabilities.includes(ACP_PERMISSION_CAPABILITY)) {
			// REFUSED, not auto-approved. A session that cannot ask a human
			// anything and runs every tool regardless is not a degraded version
			// of asking — it is the opposite of it, arrived at by omission.
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_REQUEST,
				`This agent will not create a session for a client that did not declare the "${ACP_PERMISSION_CAPABILITY}" capability. Tool calls need somewhere to ask, and a session without one would approve everything silently.`,
			)
		}
	}

	private async onSessionPrompt(params: AcpSessionPromptParams): Promise<AcpSessionPromptResult> {
		const session = this.requireSession(params.sessionId)
		if (session.promptInFlight) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_REQUEST,
				`Session "${params.sessionId}" already has a prompt in flight. Wait for it to settle or cancel it before sending another.`,
			)
		}
		session.promptInFlight = true
		// A fresh controller per turn: the previous one may already be aborted
		// by a cancel of the turn before, and reusing it would start this one
		// already cancelled.
		const controller = new AbortController()
		session.controller = controller

		try {
			const outcome = await this.options.gateway.prompt({
				sessionId: params.sessionId,
				prompt: params.prompt,
				cwd: session.cwd,
				signal: controller.signal,
				onEvent: (event) => {
					const update = toAcpSessionUpdate(event, this.options.presenter)
					if (update) void this.notifyUpdate(params.sessionId, update)
				},
				ask: (request) => this.askPermission(params.sessionId, session, request, controller.signal),
				filesystem: this.clientFilesystem(),
				history: session.history,
			})
			if (outcome.history !== undefined) {
				if (!Array.isArray(outcome.history)) {
					throw new AcpError(
						ACP_ERROR_CODES.INTERNAL_ERROR,
						`The gateway returned an invalid history for session "${params.sessionId}".`,
					)
				}
				session.history = [...outcome.history]
			}

			return {
				stopReason: controller.signal.aborted ? 'cancelled' : toAcpStopReason(outcome.stopReason),
			}
		} catch (error) {
			if (controller.signal.aborted) return { stopReason: 'cancelled' }
			// The session log refused the turn: another one is active, perhaps
			// started by another process on the same session. A client error, not
			// an internal one, and it says what to do.
			if (isTurnInProgressError(error)) {
				throw new AcpError(
					ACP_ERROR_CODES.INVALID_REQUEST,
					`Session "${params.sessionId}" already has an active turn (${error.activeTurnId}, ${error.state}). Wait for it to settle, or resume or abandon it, before sending another prompt.`,
				)
			}
			throw error
		} finally {
			session.promptInFlight = false
		}
	}

	private onSessionCancel(params: AcpSessionCancelParams): null {
		this.requireSession(params.sessionId).controller.abort()
		// `null` rather than `{}`: a JSON-RPC result has to be present, and an
		// empty object invites a client to look for a field.
		return null
	}

	/**
	 * Put a tool batch in front of the human, unless they already said yes to
	 * everything for this session.
	 */
	private async askPermission(
		sessionId: string,
		session: Session,
		request: AcpPermissionRequest,
		signal: AbortSignal,
	): Promise<AcpPermissionOutcome> {
		signal.throwIfAborted()
		// The latch, read from the SESSION. Checked here rather than in the
		// gateway so no gateway can forget it.
		if (session.approveAll) return { kind: 'approve_all' }

		const answer = await this.request<AcpRequestPermissionResult>(
			ACP_CLIENT_REQUESTS.REQUEST_PERMISSION,
			{
				// The wire session owns this identity. A gateway is allowed to
				// describe the tool batch, but it cannot relabel A's consent as B's.
				sessionId,
				toolCalls: request.toolCalls,
			},
			signal,
		)
		signal.throwIfAborted()

		switch (answer?.outcome) {
			case 'approve':
				return { kind: 'approve' }
			case 'approve_all':
				session.approveAll = true
				return { kind: 'approve_all' }
			case 'reject':
				return { kind: 'reject', ...(answer.feedback ? { feedback: answer.feedback } : {}) }
			default:
				// An answer this side cannot read is not an approval. A client that
				// sent something unrecognised has not said yes, and treating
				// "unparseable" as consent is the failure this whole exchange
				// exists to prevent.
				return {
					kind: 'reject',
					feedback: `The client answered the permission request with an outcome this agent does not recognise (${JSON.stringify(answer?.outcome)}), so the calls were not run.`,
				}
		}
	}

	/**
	 * The client's buffers, when it declared the capability.
	 *
	 * `undefined` otherwise, and that is the ordinary case: a peer that is not
	 * an editor has no buffers, and the agent should read the disk.
	 */
	private clientFilesystem(): AcpClientFilesystem | undefined {
		if (!this.clientCapabilities.includes(ACP_FILESYSTEM_CAPABILITY)) return undefined
		return {
			readTextFile: async (path) => {
				const result = await this.request<AcpFsReadResult>(ACP_CLIENT_REQUESTS.FS_READ, { path })
				return result?.content ?? ''
			},
			writeTextFile: async (path, content) => {
				await this.request(ACP_CLIENT_REQUESTS.FS_WRITE, { path, content })
			},
		}
	}

	private requireSession(sessionId: string): Session {
		const session = this.sessions.get(sessionId)
		if (!session) {
			throw new AcpError(
				ACP_ERROR_CODES.INVALID_PARAMS,
				`No session "${sessionId}". Create one with "${ACP_METHODS.SESSION_NEW}" first.`,
			)
		}
		return session
	}
}