chatbot_node.ex

defmodule ChatbotNode do
use GenServer


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :chatbot_node)
end

def init(_opts) do
settings_path = Path.join(__DIR__, "settings.txt")
content = read_file(settings_path)
settings = case Jason.decode(content) do
	{:ok, decoded} -> decoded
	{:error, _} -> %{}
end

homepath = Map.get(settings, "homepath", "/app/")
incoming_path = homepath <> "threads/incoming/"
channels_path = homepath <> "threads/channels/"
config = Map.get(settings, "chatbot_pipeline", %{})
default_model = Map.get(config, "model", "claude-opus-high")
permission_mode = Map.get(config, "permission_mode", "")
tick_ms = compute_tick_ms(config, "tick_seconds", 5, 1000)

chatbot_config = load_chatbot_config(homepath)
mention_triggers = Map.get(chatbot_config, "mention_triggers", ["@claude"])
platforms = load_platforms(chatbot_config)

Enum.each(Map.keys(platforms), fn p ->
	File.mkdir_p(incoming_path <> p)
end)
File.mkdir_p(channels_path)
File.mkdir_p(homepath <> "autologs/chatbot")
File.mkdir_p(homepath <> "threads/audio")

{:ok, tref} = :timer.send_interval(tick_ms, :tick)

{:ok, %{
	settings: settings,
	homepath: homepath,
	incoming_path: incoming_path,
	channels_path: channels_path,
	default_model: default_model,
	permission_mode: permission_mode,
	mention_triggers: mention_triggers,
	platforms: platforms,
	task_pid: nil,
	task_ref: nil,
	task_meta: nil,
	timer_ref: tref,
	tick_ms: tick_ms
}}
end


# ===== THE MAIN LOOP =====

def handle_info(:tick, state) do

if state.task_pid != nil and Process.alive?(state.task_pid) do
	flush_ticks()
	{:noreply, state}
else
	state = %{state | task_pid: nil, task_ref: nil, task_meta: nil}
	state = reload_config(state)

	{state, mention} = process_all_incoming(state)

	case mention do
		nil ->
			flush_ticks()
			{:noreply, state}

		%{platform: platform, username: username, channel_id: channel_id, content: content, content_type: content_type, metadata: metadata} ->
			model = state.default_model
			GenServer.cast(:impulse_node, {:master_message, %{command: "human_prompt"}})

			final_content = if content_type == "voice" do
				handle_voice_incoming(content, metadata, platform, state)
			else
				content
			end

			conv_name = "chatbot_" <> platform
			memory_file = state.homepath <> "memory/chatbot_channel.md"
			log_file = state.channels_path <> platform <> ".jsonl"

			trigger_text = "Platform: " <> platform <>
				". Channel: " <> channel_id <>
				". User: " <> username <>
				".\nMessage: " <> final_content <>
				"\n\nChannel log: " <> log_file <>
				". Follow the instructions in your role context to respond."

			memory_text = read_file(memory_file)
			role_block = if byte_size(String.trim(memory_text)) > 0 do
				"== Agent instructions ==\n" <> memory_text <> "\n\n"
			else
				""
			end

			model_family = ConversationTurn.normalize_model_family(model)
			recent = ConversationTurn.fetch_recent_conversation(state.homepath, "auto_ai", conv_name, 20)
			prompt = ConversationTurn.build_main_prompt(%{role_block: role_block, conversation_context: recent, trigger_text: trigger_text})

			output_path = state.homepath <> "autologs/chatbot/tick_" <> Integer.to_string(:os.system_time(:millisecond)) <> ".log"
			task = ConversationTurn.spawn_main_task(model, prompt, output_path, trigger_text, homepath: state.homepath, conversation: "auto_ai/" <> conv_name, permission_mode: state.permission_mode)

			meta = %{
				homepath: state.homepath,
				model_family: model_family,
				scope: "auto_ai",
				conversation: conv_name,
				user_text: trigger_text,
				output_path: output_path,
				reply_platform: platform,
				reply_channel: channel_id,
				voice_response: Map.get(Map.get(state.platforms, platform, %{}), "voice_response", "text"),
				original_content_type: content_type
			}

			flush_ticks()
			{:noreply, %{state | task_pid: task.pid, task_ref: task.ref, task_meta: meta}}
	end
end
end


# ===== VOICE HANDLING =====

defp handle_voice_incoming(content, metadata, platform, state) do
audio_dir = state.homepath <> "threads/audio/"
File.mkdir_p(audio_dir)

adapter_script = state.homepath <> "tools/platform_adapter.py"
meta_json = Jason.encode!(metadata)
case System.cmd("python3", [adapter_script, "download_voice", platform, meta_json, audio_dir], stderr_to_stdout: true, cd: state.homepath) do
	{output, 0} ->
		case Jason.decode(String.trim(output)) do
			{:ok, %{"downloaded" => true, "path" => audio_path}} ->
				transcribe_script = state.homepath <> "tools/transcribe/transcribe.py"
				case System.cmd("python3", [transcribe_script, "transcribe", audio_path, "--json"], stderr_to_stdout: true, cd: state.homepath) do
					{t_output, 0} ->
						case Jason.decode(String.trim(t_output)) do
							{:ok, %{"text" => text}} when is_binary(text) and byte_size(text) > 0 -> text
							_ -> content
						end
					_ -> content
				end
			_ -> content
		end
	_ -> content
end
end


# ===== PROCESS ALL INCOMING MESSAGES =====

defp process_all_incoming(state) do
enabled = state.platforms
	|> Enum.filter(fn {_, cfg} -> Map.get(cfg, "enabled", false) end)
	|> Enum.map(fn {name, _} -> name end)

Enum.reduce(enabled, {state, nil}, fn platform, {acc_state, acc_mention} ->
	dir = acc_state.incoming_path <> platform <> "/"
	case File.ls(dir) do
		{:ok, files} ->
			json_files = files
				|> Enum.filter(fn f -> String.ends_with?(f, ".json") and not String.ends_with?(f, ".processing") end)
				|> Enum.sort()

			Enum.reduce(json_files, {acc_state, acc_mention}, fn filename, {s, m} ->
				filepath = dir <> filename
				case File.read(filepath) do
					{:ok, raw} ->
						case Jason.decode(raw) do
							{:ok, msg} ->
								username = Map.get(msg, "username", "unknown")
								content = Map.get(msg, "content", "")
								channel_id = Map.get(msg, "channelId", "")
								content_type = Map.get(msg, "content_type", "text")
								metadata = Map.get(msg, "metadata", %{})
								timestamp = :os.system_time(:second)

								log_entry = Jason.encode!(%{username: username, content: content, channelId: channel_id, timestamp: timestamp, is_bot: false, platform: platform})
								log_path = s.channels_path <> platform <> ".jsonl"
								File.write(log_path, log_entry <> "\n", [:append])

								File.rm(filepath)

								if m == nil do
									platform_cfg = Map.get(s.platforms, platform, %{})
									always_trigger = Map.get(platform_cfg, "always_trigger", false)

									if always_trigger do
										{s, %{platform: platform, username: username, channel_id: channel_id, content: content, content_type: content_type, metadata: metadata}}
									else
										detect_mention(content, platform, username, channel_id, content_type, metadata, s.mention_triggers)
										|> case do
											nil -> {s, nil}
											mention -> {s, mention}
										end
									end
								else
									{s, m}
								end

							_ ->
								File.rm(filepath)
								{s, m}
						end
					_ -> {s, m}
				end
			end)

		_ -> {acc_state, acc_mention}
	end
end)
end


# ===== MENTION DETECTION =====

defp detect_mention(content, platform, username, channel_id, content_type, metadata, triggers) do
content_lower = String.downcase(content)

match = Enum.find(triggers, fn trigger ->
	String.contains?(content_lower, String.downcase(trigger))
end)

cond do
	match != nil ->
		cleaned = String.replace(content, ~r/#{Regex.escape(match)}/i, "") |> String.trim()
		%{platform: platform, username: username, channel_id: channel_id, content: cleaned, content_type: content_type, metadata: metadata}

	String.contains?(content, "<@") ->
		cleaned = Regex.replace(~r/<@!?\d+>/, content, "") |> String.trim()
		%{platform: platform, username: username, channel_id: channel_id, content: cleaned, content_type: content_type, metadata: metadata}

	true -> nil
end
end


# ===== ASYNC RESULT HANDLERS =====

def handle_info({ref, result}, state) when is_reference(ref) do
Process.demonitor(ref, [:flush])
state = if state.task_ref == ref and state.task_meta != nil do
	case result do
		{_, 0} ->
			ConversationTurn.spawn_digest(state.task_meta)
			meta = state.task_meta
			homepath = state.homepath
			Task.Supervisor.start_child(Ala.TaskSupervisor, fn ->
				send_reply(meta, homepath)
			end)
		_ -> :skip_on_nonzero_exit
	end
	%{state | task_pid: nil, task_ref: nil, task_meta: nil}
else
	state
end
{:noreply, state}
end

def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
state = if state.task_ref == ref do
	%{state | task_pid: nil, task_ref: nil, task_meta: nil}
else
	state
end
{:noreply, state}
end


defp send_reply(meta, homepath) do
output_path = meta.output_path
prefix = "> " <> meta.user_text <> "\n\n"
response = case File.read(output_path) do
	{:ok, content} -> String.replace_prefix(content, prefix, "") |> String.trim()
	_ -> ""
end

if byte_size(response) > 0 do
	platform = meta.reply_platform
	channel = meta.reply_channel
	adapter_script = homepath <> "tools/platform_adapter.py"

	System.cmd("python3", [adapter_script, "send", platform, channel, response], stderr_to_stdout: true, cd: homepath)

	voice_mode = Map.get(meta, :voice_response, "text")
	if voice_mode in ["voice", "both"] and Map.get(meta, :original_content_type) == "voice" do
		audio_out = homepath <> "threads/audio/response_" <> Integer.to_string(:os.system_time(:millisecond)) <> ".wav"
		transcribe_script = homepath <> "tools/transcribe/transcribe.py"
		case System.cmd("python3", [transcribe_script, "speak", response, "-o", audio_out], stderr_to_stdout: true, cd: homepath) do
			{_, 0} -> System.cmd("python3", [adapter_script, "send_voice", platform, channel, audio_out], stderr_to_stdout: true, cd: homepath)
			_ -> :ok
		end
	end
end
end


# ===== RECEIVING FROM OTHER NODES =====

def handle_cast({:core_message, _from, _data}, state) do
{:noreply, state}
end

def handle_cast({:master_message, _data}, state) do
{:noreply, state}
end


# ===== CONFIG RELOAD =====

defp reload_config(state) do
content = read_file(Path.join(__DIR__, "settings.txt"))
settings = case Jason.decode(content) do
	{:ok, decoded} -> decoded
	{:error, _} -> state.settings
end

config = Map.get(settings, "chatbot_pipeline", %{})
default_model = Map.get(config, "model", "claude-opus-high")
permission_mode = Map.get(config, "permission_mode", "")
new_tick_ms = compute_tick_ms(config, "tick_seconds", 5, 1000)

chatbot_config = load_chatbot_config(state.homepath)
mention_triggers = Map.get(chatbot_config, "mention_triggers", ["@claude"])
platforms = load_platforms(chatbot_config)

updated = if new_tick_ms != state.tick_ms do
	:timer.cancel(state.timer_ref)
	{:ok, new_tref} = :timer.send_interval(new_tick_ms, :tick)
	%{state | timer_ref: new_tref, tick_ms: new_tick_ms}
else
	state
end

%{updated | settings: settings, mention_triggers: mention_triggers, default_model: default_model, permission_mode: permission_mode, platforms: platforms}
end


defp load_platforms(chatbot_config) do
Map.get(chatbot_config, "platforms", %{})
end

defp load_chatbot_config(homepath) do
path = Path.join([homepath, "config", "chatbot.json"])
content = read_file(path)
case Jason.decode(content) do
	{:ok, decoded} -> decoded
	{:error, _} -> %{}
end
end


# ===== UTILITY FUNCTIONS =====

# Clamp a configured tick to at least 1 unit, using the default for missing
# or non-numeric values, so a tick of 0, negative, or a bad type can never
# make send_interval flood the mailbox or crash the node on restart.
defp compute_tick_ms(config, key, default_units, ms_per_unit) do
	units = case Map.get(config, key) do
		n when is_number(n) -> max(n, 1)
		_ -> default_units
	end
	trunc(units * ms_per_unit)
end

defp read_file(path) do
case File.read(path) do
	{:ok, content} -> content
	{:error, _} -> ""
end
end

defp flush_ticks do
receive do
	:tick -> flush_ticks()
after
	0 -> :ok
end
end

end