adaptive_coding_node.ex

defmodule AdaptiveCodingNode do
use GenServer

@max_step_retries 2


def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: :adaptive_coding_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, "coding_pipeline", %{})
pipeline = build_pipeline(config)
permission_mode = Map.get(config, "permission_mode", "")
tick_ms = compute_tick_ms(config, "tick_minutes", 15, 60000)

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

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

queue_dir = homepath <> "work/coding/queue/"
File.mkdir_p(queue_dir)
File.chmod(queue_dir, 0o777)

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

{:ok, %{
	settings: settings,
	homepath: homepath,
	permission_mode: permission_mode,
	task_type: "",
	pipeline: pipeline,
	step: 0,
	has_claude: has_claude,
	tasks: [],
	single_step_meta: nil,
		session_started: false,
		step_failures: 0,
	current_queue_file: nil,
	session_id: nil,
	timer_ref: tref,
	tick_ms: tick_ms,
	pause_warning_logged: false
}}
end


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

def handle_info(:tick, state) do

	cond do
		Enum.any?(state.tasks, fn t -> Process.alive?(t.pid) end) ->
			flush_ticks()
			{:noreply, state}

		state.single_step_meta != nil ->
			# a step finished but its result/crash is not reconciled yet; wait so
			# we advance or retry exactly once (avoids a tick-boundary re-run)
			flush_ticks()
			{:noreply, state}

		true ->
			kill_tasks(state.tasks)
			state = %{state | tasks: []}

			# a task is in progress from pick until it finishes or is failed out;
			# only pick a new one when idle at step 0 with no current queue file
			is_fresh = state.step == 0 and state.current_queue_file == nil

			state = if is_fresh do
				reload_config(state)
			else
				state
			end

			state = if state.has_claude and not ConversationTurn.claude_auth_ok?() do
				IO.puts("AdaptiveCodingNode: skipping claude calls this cycle, claude not authenticated")
				%{state | has_claude: false}
			else
				state
			end

			{task_type, task_content, queue_file} = if is_fresh do
				pick_next_task(state)
			else
				{state.task_type, "continuing", state.current_queue_file}
			end
			has_task = task_content != ""

			paused_reason = if has_task, do: global_paused(state.homepath), else: ""

			if has_task and length(state.pipeline) > 0 and paused_reason == "" and not claude_down(state) do
				# A fresh task gets one claude session for its whole pipeline and a
				# clean retry counter; later steps and retries resume that session.
				state = if is_fresh do
					%{state | pause_warning_logged: false, task_type: task_type, current_queue_file: queue_file, session_id: ConversationTurn.generate_session_id(), session_started: false, step_failures: 0}
				else
					state
				end
				{step_type, model} = Enum.at(state.pipeline, state.step)

				{tasks, meta} = run_single_step(step_type, model, state)

				write_step_status(state.homepath, step_type, model, state.step, length(state.pipeline))

				# Advancement, retry, and queue removal are decided when the step
				# result or crash comes back, see advance_step / handle_step_failure.
				flush_ticks()
				{:noreply, %{state | tasks: tasks, single_step_meta: meta, session_started: true}}
			else
				new_state = cond do
					paused_reason != "" and !state.pause_warning_logged ->
						log_path = state.homepath <> "autologs/coding/log_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
						File.write(log_path, "Skipped: coding paused (reason: " <> paused_reason <> "). This warning will not repeat until pause is lifted.\n")
						%{state | pause_warning_logged: true}

					true -> 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])
	matched = Enum.any?(state.tasks, fn t -> t.ref == ref end) and state.single_step_meta != nil
	if matched do
		meta = state.single_step_meta
		state = %{state | single_step_meta: nil}
		case result do
			{_, 0} ->
				if Map.get(meta, :persist, true), do: ConversationTurn.spawn_digest(meta)
				{:noreply, advance_step(state)}
			_ ->
				{:noreply, handle_step_failure(state)}
		end
	else
		{:noreply, state}
	end
end

def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
	# a crash (no normal result) counts as a failed step attempt
	matched = Enum.any?(state.tasks, fn t -> t.ref == ref end) and state.single_step_meta != nil
	if matched do
		{:noreply, handle_step_failure(%{state | single_step_meta: nil})}
	else
		{:noreply, state}
	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


# ===== SINGLE STEP =====

defp run_single_step(step_type, model, state) do
hp = state.homepath
memory_file = step_memory_file(step_type, hp, state.task_type)
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

instruction = step_prompt(step_type, hp, state.task_type)
recent = ConversationTurn.fetch_recent_conversation(hp, "auto_ai", "coding", 20)
trigger_text = "Coding pipeline step: " <> Atom.to_string(step_type) <> "\n\n" <> instruction

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

output_path = hp <> "autologs/coding/step_" <> Atom.to_string(step_type) <> "_" <> Integer.to_string(:os.system_time(:millisecond)) <> ".log"
task = ConversationTurn.spawn_main_task(model, prompt, output_path, trigger_text, homepath: hp, conversation: "auto_ai/coding", session_id: state.session_id, resume: state.session_started, permission_mode: state.permission_mode)

meta = %{
	homepath: hp,
	model_family: ConversationTurn.normalize_model_family(model),
	scope: "auto_ai",
	conversation: "coding",
	user_text: trigger_text,
	output_path: output_path,
	persist: step_type == :plan_and_build
}

{[task], meta}
end


# ===== PIPELINE CONFIGURATION =====

defp build_pipeline(config) do
primary = Map.get(config, "primary_model", "claude-opus-high")
passes = Map.get(config, "review_passes", 2)

core = [{:plan_and_build, primary}]
reviews = List.duplicate({:review_fix, primary}, max(passes, 2))
final = [{:final_review_fix, primary}]

core ++ reviews ++ final
end

# Role files live in memory/agent_systems/. Untyped tasks and types
# without their own coding files use devteam, the generic development type.
defp step_memory_file(step_type, homepath, task_type) do
name = case step_type do
	t when t in [:plan_and_build, :final_review_fix] -> "coding_prototype.md"
	:review_fix -> "coding_skeptical.md"
end
type = if task_type != "", do: task_type, else: "devteam"
typed = homepath <> "memory/agent_systems/" <> type <> "_" <> name
if File.exists?(typed) do
	typed
else
	homepath <> "memory/agent_systems/devteam_" <> name
end
end

defp step_prompt(step_type, homepath, _task_type) do
w = homepath <> "work/coding/"
task = w <> "task.md"
projects = homepath <> "memory/projects/"

case step_type do
	:plan_and_build -> "/plan this task based on the task spec at " <> task <> ". Break it into concrete steps. Then implement the plan. Write code following codebase conventions."
	:review_fix -> "ok now /review and fix all errors, bugs, gaps or missing pieces in the code for this task. Read " <> task <> " to understand intent. Read every file changed. Check for: bugs, logic errors, security issues, missing handling, convention violations. List every issue with file paths and specific details first, then fix all known problems in this turn."
	:final_review_fix -> "ok /review and fix all errors, bugs, gaps or missing pieces. Make the code clean and correct. When done, update project memory files in " <> projects <> " with what was built and current code state. Write a log to " <> homepath <> "autologs/coding/log_" <> Integer.to_string(:os.system_time(:second)) <> ".txt summarizing this cycle."
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, "coding_pipeline", %{})
pipeline = build_pipeline(config)
permission_mode = Map.get(config, "permission_mode", "")
new_tick_ms = compute_tick_ms(config, "tick_minutes", 15, 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, pipeline: pipeline, permission_mode: permission_mode}
end

defp write_step_status(homepath, step_type, model, step_idx, total) do
step_names = %{plan_and_build: "Building", review_fix: "Review", final_review_fix: "Review"}
step_name = Map.get(step_names, step_type, Atom.to_string(step_type))
model_name = ConversationTurn.normalize_model_family(model)
data = %{step: Atom.to_string(step_type), stepName: step_name, model: model_name, progress: Integer.to_string(step_idx + 1) <> "/" <> Integer.to_string(total), time: :os.system_time(:second)}
case Jason.encode(data) do
	{:ok, json} -> File.write(homepath <> "work/coding/status.json", json)
	_ -> :ok
end
end

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

defp kill_tasks(tasks) do
Enum.each(tasks, fn t ->
	Process.demonitor(t.ref, [:flush])
	if Process.alive?(t.pid), do: Task.Supervisor.terminate_child(Ala.TaskSupervisor, t.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 flush_ticks do
receive do
	:tick -> flush_ticks()
after
	0 -> :ok
end
end


defp pick_next_task(state) do
	queue_dir = state.homepath <> "work/coding/queue/"
	task_path = state.homepath <> "work/coding/task.md"
	case File.ls(queue_dir) do
		{:ok, files} ->
			md_files = files
				|> Enum.filter(&String.ends_with?(&1, ".md"))
				|> Enum.sort()
			case md_files do
				[first | _] ->
					src = queue_dir <> first
					case File.read(src) do
						{:ok, content} ->
							File.write(task_path, content)
							File.chmod(task_path, 0o777)
							trimmed = String.trim(content)
							lines = String.split(trimmed, "\n", parts: 2)
							first_line = hd(lines)
							if String.starts_with?(first_line, "type:") do
								task_type = String.trim(String.replace_prefix(first_line, "type:", ""))
								task_type = if Regex.match?(~r/^[a-z0-9_]+$/, task_type), do: task_type, else: ""
								task_body = if length(lines) > 1, do: Enum.at(lines, 1), else: ""
								{task_type, String.trim(task_body), src}
							else
								{"", trimmed, src}
							end
						{:error, _} -> {"", "", nil}
					end
				[] -> {"", "", nil}
			end
		{:error, _} -> {"", "", nil}
	end
end

defp claude_down(state) do
	{_step, model} = if length(state.pipeline) > 0, do: Enum.at(state.pipeline, state.step), else: {nil, ""}
	ConversationTurn.normalize_model_family(model) == "claude" and not ConversationTurn.claude_auth_ok?()
end

# ===== STEP OUTCOME (advance / retry / fail) =====

defp advance_step(state) do
	next = state.step + 1
	if next >= length(state.pipeline) do
		finish_current_task(%{state | step: 0, step_failures: 0, session_started: false})
	else
		%{state | step: next, step_failures: 0}
	end
end

# A step that exits non-zero (or crashes) is retried on the same session up to
# @max_step_retries times; after that the task is moved out of the queue so it
# does not loop forever.
defp handle_step_failure(state) do
	failures = state.step_failures + 1
	{step_type, _model} = Enum.at(state.pipeline, state.step)
	log_path = state.homepath <> "autologs/coding/log_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
	if failures > @max_step_retries do
		File.write(log_path, "Step " <> Atom.to_string(step_type) <> " failed " <> Integer.to_string(failures) <> " times, giving up. Task moved to work/coding/failed/.\n")
		fail_current_task(state, step_type)
	else
		File.write(log_path, "Step " <> Atom.to_string(step_type) <> " exited non-zero (attempt " <> Integer.to_string(failures) <> " of " <> Integer.to_string(@max_step_retries + 1) <> "). Retrying next tick.\n")
		%{state | step_failures: failures}
	end
end

defp fail_current_task(state, step_type) do
	failed_dir = state.homepath <> "work/coding/failed/"
	File.mkdir_p(failed_dir)
	if state.current_queue_file != nil do
		File.rename(state.current_queue_file, failed_dir <> Path.basename(state.current_queue_file))
	end
	File.rm(state.homepath <> "work/coding/task.md")
	write_failure_status(state.homepath, state.step, step_type)
	%{state | current_queue_file: nil, task_type: "", step: 0, step_failures: 0, session_started: false}
end

defp write_failure_status(homepath, step_idx, step_type) do
	data = %{status: "failed", failed_step: Atom.to_string(step_type), step: step_idx, time: :os.system_time(:second)}
	case Jason.encode(data) do
		{:ok, json} -> File.write(homepath <> "work/coding/status.json", json)
		_ -> :ok
	end
end

defp finish_current_task(state) do
	if state.current_queue_file != nil do
		File.rm(state.current_queue_file)
	end
	File.rm(state.homepath <> "work/coding/task.md")
	%{state | current_queue_file: nil, task_type: ""}
end

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

end