learning_node.ex

defmodule LearningNode do
use GenServer


def start_link(_opts) do
GenServer.start_link(__MODULE__, nil, name: :learning_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/")
config = Map.get(settings, "learning_pipeline", %{})
model = Map.get(config, "model", "claude-opus-high")
permission_mode = Map.get(config, "permission_mode", "")
tick_ms = compute_tick_ms(config, "tick_minutes", 10, 60000)
mb_port = Map.get(settings, "memory_bank_port", 9500)
analysis_port = Map.get(settings, "analysis_port", 9510)

has_claude = System.find_executable("claude") != nil

File.mkdir_p(homepath <> "autologs/learning")

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

now = :os.system_time(:second)

{:ok, %{
	settings: settings,
	homepath: homepath,
	model: model,
	permission_mode: permission_mode,
	has_claude: has_claude,
	mb_port: mb_port,
	analysis_port: analysis_port,
	task_pid: nil,
	task_ref: nil,
	task_meta: nil,
	timer_ref: tref,
	tick_ms: tick_ms,
	last_quality_check: now,
	last_curation: now,
	last_suggestion_review: now,
	prev_anomaly_count: 0,
	prev_training_precision: nil,
	accuracy_dropped: false,
	pause_warning_logged: false
}}
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}

	# reload settings every tick so live edits (model, permission_mode, ports,
	# tick_minutes) take effect without a full OTP restart
	state = reload_config(state)

	state = if state.has_claude and not ConversationTurn.claude_auth_ok?() do
		%{state | has_claude: false}
	else
		state
	end

	paused_reason = global_paused(state.homepath)

	if paused_reason == "" do
		state = %{state | pause_warning_logged: false}
		now = :os.system_time(:second)

		state = run_maintenance(state)

		state = if now - state.last_quality_check >= 1800 do
			new_state = check_training_accuracy(state)
			new_state = check_anomaly_rate(new_state)
			%{new_state | last_quality_check: now}
		else
			state
		end

		state = if now - state.last_curation >= 21600 do
			curate_training_data(state.homepath, state.mb_port)
			%{state | last_curation: now}
		else
			state
		end

		state = if now - state.last_suggestion_review >= 86400 do
			track_suggestion_quality(state.homepath)
			%{state | last_suggestion_review: now}
		else
			state
		end

		state = if ml_signals_pending(state) do
			spawn_analysis_task(state)
		else
			state
		end

		write_status(state)
		flush_ticks()
		{:noreply, state}
	else
		new_state = if !state.pause_warning_logged do
			log_path = state.homepath <> "autologs/learning/log_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
			File.write(log_path, "Skipped: system paused (" <> paused_reason <> ").\n")
			%{state | pause_warning_logged: true}
		else
			state
		end
		flush_ticks()
		{:noreply, new_state}
	end
end
end


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

def handle_info({ref, result}, state) when is_reference(ref) do
Process.demonitor(ref, [:flush])
if state.task_ref == ref and state.task_meta != nil do
	case result do
		{_, 0} -> ConversationTurn.spawn_digest(state.task_meta)
		_ -> :skip_digest_on_nonzero_exit
	end
	{:noreply, %{state | task_pid: nil, task_ref: nil, task_meta: nil}}
else
	{:noreply, state}
end
end

def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
{:noreply, state}
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


# ===== MAINTENANCE (every tick, mechanical, no AI) =====

defp run_maintenance(state) do
state
end


# ===== PERIODIC CHECKS =====

defp check_training_accuracy(state) do
hp = state.homepath
log_dir = hp <> "MLModels/logs/"

case File.ls(log_dir) do
	{:ok, files} ->
		ft_logs = files
			|> Enum.filter(&String.starts_with?(&1, "fasttext_classifier_"))
			|> Enum.sort()

		case ft_logs do
			[] -> state
			logs ->
				latest_path = log_dir <> List.last(logs)
				case File.read(latest_path) do
					{:ok, content} ->
						case Jason.decode(content) do
							{:ok, data} ->
								precision = Map.get(data, "precision", 0)
								prev = state.prev_training_precision

								dropped = prev != nil and is_number(prev) and prev - precision >= 0.10

								if dropped do
									log_path = hp <> "autologs/learning/accuracy_drop_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
									File.write(log_path, "FastText precision dropped from " <> Float.to_string(prev * 1.0) <> " to " <> Float.to_string(precision * 1.0) <> ". Training data may need curation.\n")
								end

								%{state | prev_training_precision: precision, accuracy_dropped: dropped}
							_ -> state
						end
					_ -> state
				end
		end
	_ -> state
end
end


defp check_anomaly_rate(state) do
hp = state.homepath
flags_dir = hp <> "threads/flags/"

count = case File.ls(flags_dir) do
	{:ok, files} -> Enum.count(files, &String.starts_with?(&1, "anomaly_"))
	_ -> 0
end

prev = state.prev_anomaly_count
diff = count - prev

if abs(diff) >= 5 do
	log_path = hp <> "autologs/learning/anomaly_rate_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
	File.write(log_path, "Anomaly flag count changed from " <> Integer.to_string(prev) <> " to " <> Integer.to_string(count) <> " (delta: " <> Integer.to_string(diff) <> ").\n")
end

%{state | prev_anomaly_count: count}
end


defp curate_training_data(homepath, mb_port) do
data_path = homepath <> "work/topic_training_data.txt"

case File.read(data_path) do
	{:ok, content} ->
		lines = String.split(content, "\n", trim: true)
		original_count = length(lines)

		valid_ids = case mb_call(mb_port, %{"action" => "topic_list"}) do
			{:ok, %{"topics" => topics}} when is_list(topics) ->
				MapSet.new(Enum.map(topics, fn t -> Map.get(t, "topic_id", "") end))
			_ -> nil
		end

		filtered = if valid_ids != nil do
			Enum.filter(lines, fn line ->
				case String.split(line, " ", parts: 2) do
					[label | _] ->
						topic_id = String.replace_prefix(label, "__label__", "")
						MapSet.member?(valid_ids, topic_id)
					_ -> false
				end
			end)
		else
			lines
		end

		filtered = Enum.filter(filtered, fn line ->
			words = String.split(line, " ")
			length(words) >= 6
		end)

		filtered = Enum.uniq(filtered)

		removed = original_count - length(filtered)
		if removed > 0 do
			File.write(data_path, Enum.join(filtered, "\n") <> "\n")
			log_path = homepath <> "autologs/learning/curation_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
			File.write(log_path, "Training data curated: removed " <> Integer.to_string(removed) <> " lines (" <> Integer.to_string(length(filtered)) <> " remaining).\n")
		end
	_ -> :ok
end
end


defp track_suggestion_quality(homepath) do
types = [
	{"marketing", "php", [homepath <> "tools/marketing.php", "list_suggestions"]},
	{"customerservice", "php", [homepath <> "tools/customerservice.php", "list_suggestions"]},
	{"socialmedia", "python3", [homepath <> "tools/socialmedia.py", "list_suggestions"]},
	{"devteam", "php", [homepath <> "tools/devteam.php", "list_suggestions"]}
]

results = Enum.map(types, fn {type_name, cmd, args} ->
	counts = try do
		Enum.reduce(["approved", "rejected", "implemented"], %{}, fn status, acc ->
			case System.cmd(cmd, args ++ [status], stderr_to_stdout: true) do
				{output, 0} ->
					case Jason.decode(String.trim(output)) do
						{:ok, %{"suggestions" => list}} when is_list(list) -> Map.put(acc, status, length(list))
						_ -> Map.put(acc, status, 0)
					end
				_ -> Map.put(acc, status, 0)
			end
		end)
	catch
		_, _ -> %{"approved" => 0, "rejected" => 0, "implemented" => 0}
	end

	total_decided = Map.get(counts, "approved", 0) + Map.get(counts, "rejected", 0) + Map.get(counts, "implemented", 0)
	approval_rate = if total_decided > 0 do
		(Map.get(counts, "approved", 0) + Map.get(counts, "implemented", 0)) / total_decided
	else
		nil
	end

	{type_name, counts, approval_rate}
end)

log_lines = Enum.reduce(results, [], fn {type_name, counts, rate}, acc ->
	if rate != nil do
		line = type_name <> ": " <> Integer.to_string(Map.get(counts, "approved", 0)) <> " approved, " <>
			Integer.to_string(Map.get(counts, "rejected", 0)) <> " rejected, " <>
			Integer.to_string(Map.get(counts, "implemented", 0)) <> " implemented" <>
			" (approval rate: " <> Integer.to_string(round(rate * 100)) <> "%)"
		acc ++ [line]
	else
		acc
	end
end)

if length(log_lines) > 0 do
	log_path = homepath <> "autologs/learning/suggestion_quality_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
	File.write(log_path, Enum.join(log_lines, "\n") <> "\n")
end
end


# ===== ML SIGNAL ANALYSIS (the AI call) =====

defp ml_signals_pending(state) do
hp = state.homepath

anomaly_flags = case File.ls(hp <> "threads/flags/") do
	{:ok, files} -> Enum.count(files, &String.starts_with?(&1, "anomaly_"))
	_ -> 0
end

consolidation_logs = case File.ls(hp <> "autologs/consolidation/") do
	{:ok, files} -> length(files)
	_ -> 0
end

anomaly_flags > 3 or state.accuracy_dropped or consolidation_logs > 0
end


defp spawn_analysis_task(state) do
hp = state.homepath
signals = gather_ml_signals(state)

if signals == "" do
	state
else
	memory_file = hp <> "memory/system/learning_analysis.md"
	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

	trigger_text = "ML signal analysis tick. Here are the signals detected:\n\n" <> signals <>
		"\n\nAnalyze these signals. Create experience entries for validated patterns. Write a summary of findings."

	prompt = ConversationTurn.build_main_prompt(%{role_block: role_block, trigger_text: trigger_text})

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

	meta = %{
		homepath: hp,
		model_family: "claude",
		scope: "auto_ai",
		conversation: "learning",
		user_text: trigger_text,
		output_path: output_path
	}

	%{state | task_pid: task.pid, task_ref: task.ref, task_meta: meta, accuracy_dropped: false}
end
end


defp gather_ml_signals(state) do
homepath = state.homepath
parts = []

flags_dir = homepath <> "threads/flags/"
anomaly_parts = case File.ls(flags_dir) do
	{:ok, files} ->
		files
		|> Enum.filter(&String.starts_with?(&1, "anomaly_"))
		|> Enum.sort()
		|> Enum.take(10)
		|> Enum.map(fn f ->
			case File.read(flags_dir <> f) do
				{:ok, content} -> "Flag " <> f <> ":\n" <> String.slice(content, 0, 500)
				_ -> ""
			end
		end)
		|> Enum.filter(&(&1 != ""))
	_ -> []
end

parts = if length(anomaly_parts) > 0 do
	parts ++ ["=== Anomaly Flags (" <> Integer.to_string(length(anomaly_parts)) <> ") ===\n" <> Enum.join(anomaly_parts, "\n\n")]
else
	parts
end

cons_dir = homepath <> "autologs/consolidation/"
cons_parts = case File.ls(cons_dir) do
	{:ok, files} ->
		files
		|> Enum.sort()
		|> Enum.take(-3)
		|> Enum.map(fn f ->
			case File.read(cons_dir <> f) do
				{:ok, content} -> String.slice(content, 0, 500)
				_ -> ""
			end
		end)
		|> Enum.filter(&(&1 != ""))
	_ -> []
end

parts = if length(cons_parts) > 0 do
	parts ++ ["=== Recent Consolidation Logs ===\n" <> Enum.join(cons_parts, "\n")]
else
	parts
end

topic_sample = case mb_call(state.mb_port, %{"action" => "topic_list"}) do
	{:ok, %{"topics" => topics}} when is_list(topics) and length(topics) > 0 ->
		sampled = topics |> Enum.shuffle() |> Enum.take(3)
		lines = Enum.map(sampled, fn t ->
			tid = Map.get(t, "topic_id", "")
			case mb_call(state.mb_port, %{"action" => "topic_get", "topic_id" => tid, "include_digest" => true}) do
				{:ok, %{"topic" => topic_data, "digest" => digest}} ->
					member_count = length(Map.get(topic_data, "members", []))
					digest_text = if is_binary(digest) and byte_size(digest) > 0, do: String.slice(digest, 0, 300), else: "(no digest)"
					"Topic " <> tid <> " (" <> Integer.to_string(member_count) <> " members):\n  Digest: " <> digest_text
				_ ->
					"Topic " <> tid <> " (could not fetch details)"
			end
		end)
		"=== Topic Assignment Samples ===\n" <> Enum.join(lines, "\n\n")
	_ -> ""
end

parts = if topic_sample != "", do: parts ++ [topic_sample], else: parts

analysis_stats = case mb_call(state.analysis_port, %{"action" => "stats"}) do
	{:ok, stats} when is_map(stats) ->
		ft = if Map.get(stats, "fasttext_loaded", false), do: "loaded", else: "not loaded"
		ra = if Map.get(stats, "river_anomaly_loaded", false), do: "loaded", else: "not loaded"
		items = Map.get(stats, "items_processed_total", 0)
		classified = Map.get(stats, "turns_classified", 0)
		uptime = Map.get(stats, "uptime_seconds", 0)
		"=== Analysis Service Health ===\nUptime: " <> Integer.to_string(uptime) <> "s, items processed: " <> Integer.to_string(items) <> ", turns classified: " <> Integer.to_string(classified) <> "\nFastText: " <> ft <> ", River anomaly: " <> ra
	_ -> ""
end

parts = if analysis_stats != "", do: parts ++ [analysis_stats], else: parts

Enum.join(parts, "\n\n")
end


# ===== STATUS =====

defp write_status(state) do
data = %{
	activity: "maintenance",
	last_quality_check: state.last_quality_check,
	last_curation: state.last_curation,
	last_suggestion_review: state.last_suggestion_review,
	anomaly_count: state.prev_anomaly_count,
	ai_task_running: state.task_pid != nil,
	time: :os.system_time(:second)
}
case Jason.encode(data) do
	{:ok, json} -> File.write(state.homepath <> "work/learning/status.json", json)
	_ -> :ok
end
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, "learning_pipeline", %{})
model = Map.get(config, "model", "claude-opus-high")
permission_mode = Map.get(config, "permission_mode", "")
new_tick_ms = compute_tick_ms(config, "tick_minutes", 10, 60000)
mb_port = Map.get(settings, "memory_bank_port", 9500)
analysis_port = Map.get(settings, "analysis_port", 9510)

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, model: model, permission_mode: permission_mode, mb_port: mb_port, analysis_port: analysis_port}
end


# ===== SOCKET COMMUNICATION =====

defp mb_call(port, data) do
case :gen_tcp.connect({127, 0, 0, 1}, port, [:binary, active: false, packet: :line], 5000) do
	{:ok, sock} ->
		:gen_tcp.send(sock, Jason.encode!(data) <> "\n")
		result = case :gen_tcp.recv(sock, 0, 10000) do
			{:ok, line} -> Jason.decode(String.trim(line))
			{:error, _} -> {:error, :recv_failed}
		end
		:gen_tcp.close(sock)
		result
	{:error, _} -> {:error, :connect_failed}
end
end


# ===== UTILITY =====

# 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

defp global_paused(homepath) do
	if File.exists?(homepath <> "work/.global_pause"), do: "global", else: ""
end

end