import { closeSync, readFileSync } from "node:fs";
import type { AgentSessionMessageDeliveryMode, AgentSessionMessageSender } from "../../core/agent-messages.js";
import type { IdleEvictionMinutes } from "../../core/session-action-store.js";

export { SESSION_LEASE_OWNER_ID_ENV, SESSION_LEASES_ENABLED_ENV } from "../../core/session-lease.js";

import type { WorkerRosterEntry } from "./agent-roster.js";
import type { DaemonClientCapability, DaemonCommand, DaemonOutbound } from "./daemon-protocol.js";

export const DAEMON_WORKER_ROLE_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER";
export const DAEMON_WORKER_TOKEN_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_TOKEN";
export const DAEMON_WORKER_INSTANCE_ID_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_INSTANCE_ID";
export const DAEMON_WORKER_ACTIVE_SESSION_ID_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_ACTIVE_SESSION_ID";
export const DAEMON_WORKER_SUPERVISOR_SOCKET_ENV = "PRIME_AGENT_INTERNAL_DAEMON_SUPERVISOR_SOCKET";
export const DAEMON_WORKER_RECOVERY_JOURNAL_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_RECOVERY_JOURNAL";
export const DAEMON_WORKER_STARTUP_GATE_FD_ENV = "PRIME_AGENT_INTERNAL_DAEMON_WORKER_STARTUP_GATE_FD";
export const DAEMON_WORKER_STARTUP_GATE_COMMIT = "start\n";
export type DaemonWorkerLifecycle = "starting" | "ready" | "recovering" | "stopping" | "failed";

// Worker->supervisor roster frames live outside the client-facing DaemonOutbound schema.
export type DaemonWorkerRosterOutbound =
	| {
			type: "roster_delta";
			entries: WorkerRosterEntry[];
			removedAgentIds?: string[];
			snapshot?: true;
	  }
	| { type: "roster_heartbeat" };

/** Advertised by new workers in the worker_auth response; absent on legacy workers. */
export const DAEMON_WORKER_ROSTER_CAPABILITY = "agent_roster";

/** Advertised in the worker_auth response by workers that accept peer transport grants. */
export const DAEMON_WORKER_PEER_TRANSPORT_CAPABILITY = "peer_transport";

/** Idle keepalive cadence for worker->supervisor roster frames; the supervisor staleness threshold derives from it. */
export const ROSTER_HEARTBEAT_INTERVAL_MS = 15_000;

export type DaemonWorkerFrameHeader =
	| {
			kind: "command";
			requestId: string;
			commandType: string;
	  }
	| {
			kind: "outbound";
			requestId?: string;
			outboundType: DaemonOutbound["type"] | DaemonWorkerRosterOutbound["type"];
			activeSessionId?: string;
			snapshotId?: string;
			sessionEventType?: string;
			payloadEncoding?: "jsonl" | "assistant-delta";
			snapshotPurpose?: "attach" | "replacement" | "catchup";
	  };

export type DaemonCreateCommand = Extract<DaemonCommand, { type: "create" }>;

export interface DurableDaemonCreateCommand {
	type: "create";
	sessionPath?: string;
	noSession?: boolean;
}

export function durableDaemonCreateCommand(command: DaemonCreateCommand): DurableDaemonCreateCommand {
	return {
		type: "create",
		...(command.sessionPath !== undefined ? { sessionPath: command.sessionPath } : {}),
		...(command.noSession !== undefined ? { noSession: command.noSession } : {}),
	};
}

/**
 * A single-use, worker-memory-only admission for one direct peer, scoped to one active-session
 * slot (stable across switch_session/new_session/fork) of one worker incarnation. The session pin
 * is a routing/accident guard, not a privilege boundary: ticket holders already hold supervisor
 * access, which can attach to, switch, or kill any session.
 */
export interface DaemonWorkerPeerGrant {
	grantId: string;
	token: string;
	expiresAt: string;
	purpose: "session_client";
	workerInstanceId: string;
	activeSessionId: string;
	issuerGeneration: string;
}

/** Commands a direct peer may send before it holds an authenticated session role. */
export type DaemonPeerCommand = {
	id?: string;
	type: "peer_auth";
	grantId: string;
	token: string;
	workerInstanceId: string;
	purpose: "session_client";
};

export type DaemonPeerCommandBody = Omit<DaemonPeerCommand, "id">;

export type DaemonWorkerCommand =
	| {
			id?: string;
			type: "worker_auth";
			token: string;
			workerInstanceId?: string;
			supervisorGeneration: string;
			supervisorPid: number;
			supervisorProcessStartId?: string;
			supervisorSocketPath: string;
	  }
	| {
			id?: string;
			type: "worker_subscribe";
			activeSessionId: string;
			capabilities?: readonly DaemonClientCapability[];
			supportsExtensionUi?: boolean;
	  }
	| { id?: string; type: "worker_unsubscribe"; activeSessionId: string }
	| { id?: string; type: "worker_register_peer_transport"; grant: DaemonWorkerPeerGrant }
	| { id?: string; type: "worker_archive_and_shutdown" }
	| {
			id?: string;
			type: "worker_passivate_idle_children";
			idleEvictionMinutes: IdleEvictionMinutes;
			now: number;
			limit: number;
	  }
	| {
			id?: string;
			type: "worker_deliver_message";
			targetActiveSessionId: string;
			message: string;
			sender: AgentSessionMessageSender;
			deliveryMode?: AgentSessionMessageDeliveryMode;
	  }
	| { id?: string; type: "worker_prepare_update" }
	| { id?: string; type: "worker_commit_update" }
	| { id?: string; type: "worker_cancel_update" };

export type DaemonWorkerCommandBody = DaemonWorkerCommand extends infer TCommand
	? TCommand extends { id?: string }
		? Omit<TCommand, "id">
		: never
	: never;

export interface DaemonWorkerDescriptor {
	version: 1 | 2;
	workerId: string;
	pid: number;
	processStartId?: string;
	socketPath: string;
	recoveryJournalPath: string;
	orphanProcessJournalPath?: string;
	supervisorSocketPath: string;
	authenticationToken: string;
	workerInstanceId?: string;
	rootActiveSessionId: string;
	/** Stable protocol client that owns this worker. Omitted for resident sessions. */
	ownerClientId?: string;
	rootSessionId?: string;
	sessionFile?: string;
	sessionDir?: string;
	telemetryDisabled?: true;
	createdAt: string;
	updatedAt: string;
	lifecycle: DaemonWorkerLifecycle;
	createCommand: DurableDaemonCreateCommand;
	consecutiveFailures: number;
	/** Durable intent written before root termination so replacement supervisors never recover it. */
	stopRequestedAt?: string;
	/** Complete the root's archived lifecycle state after its process has stopped. */
	archiveOnStop?: boolean;
	lastFailureAt?: string;
	lastError?: string;
}

export function durableDaemonWorkerDescriptor(descriptor: DaemonWorkerDescriptor): DaemonWorkerDescriptor {
	const versionOneCreateCommand = descriptor.createCommand as unknown as { config?: unknown };
	const versionOneConfig =
		descriptor.version === 1 &&
		typeof versionOneCreateCommand.config === "object" &&
		versionOneCreateCommand.config !== null
			? (versionOneCreateCommand.config as Record<string, unknown>)
			: undefined;
	const sessionDir =
		descriptor.sessionDir ??
		(typeof versionOneConfig?.sessionDir === "string" ? versionOneConfig.sessionDir : undefined);
	const telemetryDisabled = descriptor.telemetryDisabled === true || versionOneConfig?.telemetryDisabled === true;
	return {
		version: 2,
		workerId: descriptor.workerId,
		pid: descriptor.pid,
		...(descriptor.processStartId !== undefined ? { processStartId: descriptor.processStartId } : {}),
		socketPath: descriptor.socketPath,
		recoveryJournalPath: descriptor.recoveryJournalPath,
		...(descriptor.orphanProcessJournalPath !== undefined
			? { orphanProcessJournalPath: descriptor.orphanProcessJournalPath }
			: {}),
		supervisorSocketPath: descriptor.supervisorSocketPath,
		authenticationToken: descriptor.authenticationToken,
		...(descriptor.workerInstanceId !== undefined ? { workerInstanceId: descriptor.workerInstanceId } : {}),
		rootActiveSessionId: descriptor.rootActiveSessionId,
		...(descriptor.ownerClientId !== undefined ? { ownerClientId: descriptor.ownerClientId } : {}),
		...(descriptor.rootSessionId !== undefined ? { rootSessionId: descriptor.rootSessionId } : {}),
		...(descriptor.sessionFile !== undefined ? { sessionFile: descriptor.sessionFile } : {}),
		...(sessionDir !== undefined ? { sessionDir } : {}),
		...(telemetryDisabled ? { telemetryDisabled: true as const } : {}),
		createdAt: descriptor.createdAt,
		updatedAt: descriptor.updatedAt,
		lifecycle: descriptor.lifecycle,
		createCommand: durableDaemonCreateCommand(descriptor.createCommand),
		consecutiveFailures: descriptor.consecutiveFailures,
		...(descriptor.stopRequestedAt !== undefined ? { stopRequestedAt: descriptor.stopRequestedAt } : {}),
		...(descriptor.archiveOnStop !== undefined ? { archiveOnStop: descriptor.archiveOnStop } : {}),
		...(descriptor.lastFailureAt !== undefined ? { lastFailureAt: descriptor.lastFailureAt } : {}),
		...(descriptor.lifecycle === "failed" ? { lastError: "Waiting for a client with fresh runtime context" } : {}),
	};
}

export function isDaemonWorkerProcess(environment: NodeJS.ProcessEnv = process.env): boolean {
	return environment[DAEMON_WORKER_ROLE_ENV] === "1";
}

export function waitForDaemonWorkerStartupGate(environment: NodeJS.ProcessEnv = process.env): void {
	const rawFd = environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV];
	if (rawFd === undefined) {
		return;
	}
	delete environment[DAEMON_WORKER_STARTUP_GATE_FD_ENV];
	const fd = Number(rawFd);
	if (!Number.isInteger(fd) || fd < 3) {
		throw new Error("Daemon session worker has an invalid startup gate");
	}
	let marker: string;
	try {
		marker = readFileSync(fd, "utf8");
	} finally {
		closeSync(fd);
	}
	if (marker !== DAEMON_WORKER_STARTUP_GATE_COMMIT) {
		throw new Error("Daemon session worker startup was cancelled");
	}
}

export function daemonWorkerInstanceId(environment: NodeJS.ProcessEnv = process.env): string | undefined {
	return environment[DAEMON_WORKER_INSTANCE_ID_ENV] || undefined;
}

export function requireDaemonWorkerAuthenticationToken(environment: NodeJS.ProcessEnv = process.env): string {
	const token = environment[DAEMON_WORKER_TOKEN_ENV];
	if (!token) {
		throw new Error("Daemon session worker is missing its authentication token");
	}
	return token;
}

export function isDaemonWorkerFrameHeader(value: unknown): value is DaemonWorkerFrameHeader {
	if (!value || typeof value !== "object") {
		return false;
	}
	const candidate = value as Record<string, unknown>;
	if (candidate.kind === "command") {
		return typeof candidate.requestId === "string" && typeof candidate.commandType === "string";
	}
	return (
		candidate.kind === "outbound" &&
		typeof candidate.outboundType === "string" &&
		(candidate.requestId === undefined || typeof candidate.requestId === "string") &&
		(candidate.activeSessionId === undefined || typeof candidate.activeSessionId === "string") &&
		(candidate.snapshotId === undefined || typeof candidate.snapshotId === "string") &&
		(candidate.sessionEventType === undefined || typeof candidate.sessionEventType === "string") &&
		(candidate.snapshotPurpose === undefined ||
			candidate.snapshotPurpose === "attach" ||
			candidate.snapshotPurpose === "replacement" ||
			candidate.snapshotPurpose === "catchup") &&
		(candidate.payloadEncoding === undefined ||
			candidate.payloadEncoding === "jsonl" ||
			candidate.payloadEncoding === "assistant-delta")
	);
}