master_agent.ex

defmodule MasterAgent do
use GenServer


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :master_agent)
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/")
prompts_dir = homepath <> "threads/prompts"
prompt_file = prompts_dir <> "/prompt.txt"
cancel_file = prompts_dir <> "/cancel.txt"
active_conv_file = prompts_dir <> "/active_conversation.txt"

File.mkdir_p(prompts_dir <> "/.digest")
File.mkdir_p(homepath <> "threads/output")
File.mkdir_p(homepath <> "threads/output")
File.mkdir_p(homepath <> "autologs/digest")

# Event-driven: watch threads/prompts/ for prompt.txt and cancel.txt
# touches. Replaces the prior 5sec polling tick.
{:ok, watcher_pid} = FileSystem.start_link(dirs: [prompts_dir])
FileSystem.subscribe(watcher_pid)

state = %{
	settings: settings,
	homepath: homepath,
	prompts_dir: prompts_dir,
	prompt_file: prompt_file,
	cancel_file: cancel_file,
	active_conv_file: active_conv_file,
	watcher_pid: watcher_pid,
	task_pid: nil,
	task_ref: nil,
	task_meta: nil
}
# If a prompt was sitting in the file at startup (e.g., service restarted
# while a prompt was waiting), pick it up. Watcher events fire on changes
# only, so without this an initial-state prompt would be stuck.
{:ok, state, {:continue, :initial_check}}
end


def handle_continue(:initial_check, state) do
{:noreply, maybe_dispatch_pending_prompt(state)}
end


# ===== EVENT HANDLERS =====

# File watcher event. We only care about prompt_file (run a turn) and
# cancel_file (kill an active turn). Everything else is ignored.
def handle_info({:file_event, _watcher, {path, _events}}, state) do
cond do
	path == state.prompt_file ->
		new_state = maybe_dispatch_pending_prompt(state)
		{:noreply, new_state}

	path == state.cancel_file ->
		new_state = handle_cancel(state)
		{:noreply, new_state}

	true ->
		{:noreply, state}
end
end

# Watcher stopped (rare). No restart logic; if it happens we lose
# prompt event-driven dispatch. The task-completion path still
# re-checks the prompt file so prompts don't get permanently stuck.
def handle_info({:file_event, _watcher, :stop}, state) do
{:noreply, state}
end


# ===== DISPATCH HELPERS =====

# Read prompt_file (atomically via rename) and dispatch a turn if
# present. No-op if a task is already running. Called from file events
# AND after task completion, so a prompt arriving mid-task gets picked
# up exactly once after the prior turn ends.
defp maybe_dispatch_pending_prompt(state) do
if state.task_pid != nil and Process.alive?(state.task_pid) do
	state
else
	state = %{state | task_pid: nil, task_ref: nil, task_meta: nil}

	temp = state.prompt_file <> ".processing"
	prompt_content = case File.rename(state.prompt_file, temp) do
		:ok ->
			content = read_file(temp)
			File.rm(temp)
			content
		{:error, _} -> ""
	end

	if byte_size(prompt_content) > 0 do
		GenServer.cast(:impulse_node, {:master_message, %{command: "human_prompt"}})
		dispatch_turn(prompt_content, state)
	else
		state
	end
end
end

defp handle_cancel(state) do
File.rm(state.cancel_file)
if state.task_pid != nil and Process.alive?(state.task_pid) do
	kill_task(state)
	%{state | task_pid: nil, task_ref: nil, task_meta: nil}
else
	state
end
end


defp master_permission_mode do
content = read_file(Path.join(__DIR__, "settings.txt"))
settings = case Jason.decode(content) do
	{:ok, decoded} -> decoded
	_ -> %{}
end
Map.get(Map.get(settings, "master_pipeline", %{}), "permission_mode", "acceptEdits")
end


defp dispatch_turn(prompt_content, state) do
{model_raw, user_text} = parse_model_prefix(String.trim(prompt_content))
model_family = ConversationTurn.normalize_model_family(model_raw)

output_path = state.homepath <> "threads/output/response_" <> Integer.to_string(:os.system_time(:millisecond)) <> ".txt"

case read_active_conversation(state.active_conv_file) do
	{:error, reason} ->
		File.write(output_path, "> " <> user_text <> "\n\n")
		File.write(output_path, "The system needs an active conversation. " <> reason <> " Use the UI dropdown to select or create one, then try again.", [:append])
		state

	{:ok, scope, conv_name} ->
		master_memory = read_file(Path.join(state.homepath, "memory/core_master.md"))
		role_block = if byte_size(String.trim(master_memory)) > 0 do
			master_memory <> "\n\n"
		else
			""
		end

		ctx_path = Path.join([state.homepath, "memory_bank", "sessions", scope, conv_name, "context_settings.json"])
		ctx = case Jason.decode(read_file(ctx_path)) do
			{:ok, decoded} when is_map(decoded) -> decoded
			_ -> %{}
		end

		msg_count = if scope == "auto_ai" do
			raw = Map.get(state.settings, "starthook_messages_auto", 20)
			if is_binary(raw), do: String.to_integer(raw), else: raw
		else
			case Map.get(ctx, "message_count") do
				n when is_integer(n) -> n
				_ ->
					raw = Map.get(state.settings, "starthook_messages", 5)
					if is_binary(raw), do: String.to_integer(raw), else: raw
			end
		end
		conversation_context = ConversationTurn.fetch_recent_conversation(state.homepath, scope, conv_name, msg_count)

		extra_context = if scope == "human" do
			ConversationTurn.fetch_context_extras(state.homepath, ctx, user_text)
		else
			""
		end

		prompt = ConversationTurn.build_main_prompt(%{role_block: role_block, conversation_context: conversation_context, extra_context: extra_context, trigger_text: user_text})
		task = ConversationTurn.spawn_main_task(model_raw, prompt, output_path, user_text, homepath: state.homepath, permission_mode: master_permission_mode())

		meta = %{
			homepath: state.homepath,
			model_family: model_family,
			scope: scope,
			conversation: conv_name,
			user_text: user_text,
			output_path: output_path
		}
		%{state | task_pid: task.pid, task_ref: task.ref, task_meta: meta}
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)
		_ -> :skip_digest_on_nonzero_exit
	end
	%{state | task_pid: nil, task_ref: nil, task_meta: nil}
	|> maybe_dispatch_pending_prompt()
else
	state
end
{:noreply, state}
end

def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
state = if state.task_ref == ref do
	# the AI task crashed (non-normal exit); clear it and, like the normal
	# completion path, pick up any prompt that arrived while it was running
	%{state | task_pid: nil, task_ref: nil, task_meta: nil}
	|> maybe_dispatch_pending_prompt()
else
	state
end
{: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


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

defp parse_model_prefix(content) do
case Regex.run(~r/^\[model:([\w.\-]+)\](?:\[source:[\w\-]+\])?\n(.*)$/s, content) do
	[_, model, rest] -> {model, String.trim(rest)}
	_ -> {"claude", content}
end
end

defp read_active_conversation(path) do
case File.read(path) do
	{:ok, content} ->
		trimmed = String.trim(content)
		case String.split(trimmed, "/", parts: 2) do
			[scope, name] when scope in ["human", "auto_ai"] and byte_size(name) > 0 ->
				{:ok, scope, name}
			_ ->
				{:error, "The active_conversation.txt file is empty or malformed."}
		end
	{:error, _} ->
		{:error, "The active_conversation.txt file does not exist."}
end
end

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

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

end