impulse_node.ex

defmodule ImpulseNode do
use GenServer


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :impulse_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, "impulse_pipeline", %{})
model = Map.get(config, "model", "claude-sonnet")
permission_mode = Map.get(config, "permission_mode", "")
tick_ms = compute_tick_ms(config, "tick_minutes", 60, 60000)
status_file = homepath <> "threads/nodestatus.json"

# task agents ship disabled in config/agents.json, impulse ships paused here
File.mkdir_p(homepath <> "work")
pause_path = homepath <> "work/.pause_impulse"
if not File.exists?(pause_path) do
	File.write(pause_path, "paused (default, enable from UI)")
	File.chmod(pause_path, 0o777)
end

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

# the dashboard reads nodestatus.json, so it refreshes every minute on its
# own timer instead of waiting for the (much slower) main tick
{:ok, _} = :timer.send_interval(60_000, :write_status)

{:ok, %{settings: settings, homepath: homepath, model: model, permission_mode: permission_mode, impulse: 0, task_pid: nil, task_ref: nil, status_file: status_file, timer_ref: tref, tick_ms: tick_ms, pause_warning_logged: false}}
end


def handle_info(:write_status, state) do
write_node_status(state)
{:noreply, state}
end


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

def handle_info(:tick, state) do

# update node status file every tick
write_node_status(state)

# impulse > 0 means human is active, skip this tick
if state.impulse > 0 do
	flush_ticks()
	{:noreply, %{state | impulse: state.impulse - 1}}
else
	state = reload_config(state)

	paused_reason = cond do
		global_paused(state.homepath) != "" -> "global"
		File.exists?(state.homepath <> "work/.pause_impulse") -> "impulse"
		true -> ""
	end

	cond do
		paused_reason != "" ->
			new_state = if !state.pause_warning_logged do
				IO.puts("ImpulseNode: skipping tick, system paused (" <> paused_reason <> ")")
				%{state | pause_warning_logged: true}
			else
				state
			end
			flush_ticks()
			{:noreply, new_state}

		ConversationTurn.normalize_model_family(state.model) == "claude" and not ConversationTurn.claude_auth_ok?() ->
			IO.puts("ImpulseNode: skipping tick, claude not authenticated")
			flush_ticks()
			{:noreply, state}

		true ->
			kill_task(state)

			memory_file = state.homepath <> "memory/core_impulse.md"
			{context, _} = System.cmd("php", [state.homepath <> "tools/gather_context.php", "impulse_context"])
			recent = ConversationTurn.fetch_recent_conversation(state.homepath, "auto_ai", "impulse", 20)
			prompt = "Read your memory file at: " <> memory_file <> ". Here is pre-gathered context:\n\n" <> context <> "\n\n" <> recent <> "Follow your instructions for this turn. Respond with a short message to yourself about what you did."

			{cmd, args, opts} = cli_invocation(state, prompt)

			task = Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd(cmd, args, opts)
			end)

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


# ===== ASYNC RESULT HANDLERS =====
# old task results are harmless, just flush the monitor

def handle_info({ref, result}, state) when is_reference(ref) do
Process.demonitor(ref, [:flush])
{:noreply, state}
end

def handle_info({:DOWN, ref, :process, pid, reason}, state) do
{:noreply, state}
end


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

# synchronous can_work check for nodes that need a blocking answer
def handle_call(:can_work, from, state) do
{:reply, state.impulse == 0, state}
end

def handle_cast({:core_message, from, data}, state) do

state = case data do
	# nodes check if impulse allows their auto tick
	%{command: "can_work"} ->
		allowed = state.impulse == 0
		GenServer.cast(from, {:core_message, :impulse_node, %{command: "work_allowed", allowed: allowed}})
		state

	# any node can flag something for human review
	%{command: "flag_for_review"} ->
		flag_content = Map.get(data, :content, "")
		flag_path = state.homepath <> "threads/flags/flag_" <> Integer.to_string(:os.system_time(:second)) <> "_" <> Integer.to_string(:erlang.unique_integer([:positive])) <> ".txt"
		File.write(flag_path, flag_content)
		state

	_ -> state
end

{:noreply, state}
end

# master signals human activity
def handle_cast({:master_message, data}, state) do
case data[:command] do
	"human_prompt" ->
		# human is active, skip next 2 ticks
		{:noreply, %{state | impulse: 2}}

	_ -> {:noreply, state}
end
end




# ===== CLI HELPERS =====

# Uses the shared ConversationTurn builders so flags never drift again.
defp cli_invocation(state, prompt) do
env = [{"ALA_CONVERSATION", "auto_ai/impulse"}]
{cmd, args, base_opts} = ConversationTurn.build_main_cli_command(state.model, prompt, state.permission_mode)
{cmd, args, base_opts ++ [cd: state.homepath, env: env]}
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, "impulse_pipeline", %{})
model = Map.get(config, "model", "claude-sonnet")
permission_mode = Map.get(config, "permission_mode", "")
new_tick_ms = compute_tick_ms(config, "tick_minutes", 60, 60000)

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}
end


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

defp kill_task(state) do
if state.task_pid != nil do
	Process.demonitor(state.task_ref, [:flush])
	if Process.alive?(state.task_pid), do: Task.Supervisor.terminate_child(Ala.TaskSupervisor, state.task_pid)
end
end

# 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 write_node_status(state) do
	busy = state.task_pid != nil and Process.alive?(state.task_pid)
	impulse_status = if state.impulse > 0, do: "paused", else: (if busy, do: "busy", else: "idle")

	# the schedule tick is user-configurable, so read it from settings rather
	# than reporting the old hardcoded 60m
	sched_tick = case Map.get(Map.get(state.settings, "schedule_pipeline", %{}), "tick_minutes", 60) do
		n when is_integer(n) and n > 0 -> Integer.to_string(n) <> "m"
		_ -> "60m"
	end

	# core nodes with their display names and tick rates
	all_nodes = [
		{"master", :master_agent, "5s"},
		{"learning", :learning_node, "15m"},
		{"coding", :adaptive_coding_node, "15m"},
		{"research", :adaptive_research_node, "15m"},
		{"chatbot", :chatbot_node, "5s"},
		{"memory_bank", :memory_bank, "always"},
		{"schedule", :schedule_node, sched_tick}
	]

	# only include nodes that are actually registered (started by application.ex)
	dynamic_nodes = Enum.reduce(all_nodes, %{}, fn {name, registered, tick}, acc ->
		case Process.whereis(registered) do
			nil -> acc
			_pid -> Map.put(acc, name, %{status: node_alive_status(registered), tick: tick})
		end
	end)

	# user-defined task agents from config/agents.json
	agent_nodes = Enum.reduce(TaskAgentNode.list_ids(), %{}, fn id, acc ->
		s = TaskAgentNode.status(id)
		Map.put(acc, id, %{status: s.status, tick: s.tick})
	end)

	nodes = dynamic_nodes
		|> Map.merge(agent_nodes)
		|> Map.put("impulse", %{status: impulse_status, tick: Integer.to_string(div(state.tick_ms, 60000)) <> "m"})

	data = %{nodes: nodes, updated: :os.system_time(:second), impulse_skip: state.impulse}
	case Jason.encode(data) do
		{:ok, json} -> File.write(state.status_file, json)
		_ -> :ok
	end
end

defp node_alive_status(name) do
	case Process.whereis(name) do
		nil -> "down"
		pid -> if Process.alive?(pid), do: "active", else: "down"
	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