task_agent_node.ex

defmodule TaskAgentNode do
use GenServer

# Generic tick agent. One instance runs per enabled record in
# config/agents.json, started and stopped at runtime by AgentManager under
# Ala.NodeSupervisor. Replaces the old fixed marketing, customerservice
# and socialmedia nodes.
#
# The agent map is fully resolved by AgentManager before start:
#   id, type, homepath, model, tick_minutes, conversation,
#   instructions_file, trigger_label, pre_command (argv list or nil),
#   gate_command (argv list or nil), gate_keys (list of strings)
#
# Config edits and removals always let a running task finish first.


def child_spec(agent) do
%{
	id: {:task_agent, agent.id},
	start: {__MODULE__, :start_link, [agent]},
	restart: :transient
}
end

def start_link(agent) do
GenServer.start_link(__MODULE__, agent, name: via(agent.id))
end

def via(id), do: {:via, Registry, {Ala.AgentRegistry, id}}

def list_ids do
Registry.select(Ala.AgentRegistry, [{{:"$1", :"$2", :"$3"}, [], [:"$1"]}])
end

def status(id) do
try do
	GenServer.call(via(id), :status, 1000)
catch
	:exit, _ -> %{status: "down", tick: "", type: ""}
end
end

def update_config(id, agent) do
GenServer.cast(via(id), {:update_config, agent})
end

def remove(id) do
GenServer.cast(via(id), :remove)
end

# Immediately terminates the agent's running task (the agent itself keeps
# ticking). Wired to the UI kill button via AgentManager's command watcher.
def kill_now(id) do
GenServer.cast(via(id), :kill_now)
end

def init(agent) do
File.mkdir_p(agent.homepath <> "autologs/" <> agent.id)
tick_ms = agent.tick_minutes * 60000
{:ok, tref} = :timer.send_interval(tick_ms, :tick)

{:ok, %{
	agent: agent,
	pending_agent: nil,
	removing: false,
	task_pid: nil,
	task_ref: nil,
	timer_ref: tref,
	tick_ms: tick_ms,
	gate_warning_logged: 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
	# Everything that can block (pre_command, gate_command, the AI prompt
	# sequence) runs inside one supervised task, NOT on this GenServer, so a
	# hung tool can never wedge status, kill_now, update_config, or remove.
	# The task reports the warning-throttle flags, applied when it finishes.
	state = %{state | task_pid: nil, task_ref: nil}
	state = apply_pending(state)
	task = spawn_tick_task(state)
	flush_ticks()
	{:noreply, %{state | task_pid: task.pid, task_ref: task.ref}}
end
end


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

def handle_info({ref, result}, state) when is_reference(ref) do
Process.demonitor(ref, [:flush])
# The whole tick (intake, gate, and any AI session) ran inside this one task.
# Its output already streamed to the tick log, so there is nothing to persist
# here; we only pick up the warning-throttle flags the task computed.
state = if state.task_ref == ref do
	flags = case result do
		{:tick_result, f} -> f
		_ -> %{}
	end
	%{state |
		task_pid: nil,
		task_ref: nil,
		gate_warning_logged: Map.get(flags, :gate_warning_logged, state.gate_warning_logged),
		pause_warning_logged: Map.get(flags, :pause_warning_logged, state.pause_warning_logged)
	}
else
	state
end
if state.removing do
	{:stop, :normal, state}
else
	{:noreply, apply_pending(state)}
end
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}
else
	state
end
if state.removing and state.task_pid == nil do
	{:stop, :normal, state}
else
	{:noreply, state}
end
end


# ===== THE TICK TASK (all blocking work off the GenServer) =====

# One supervised task per tick runs the full sequence: intake (pre_command),
# the gate check, and, if the gate is open, the AI prompt session. None of it
# runs on the GenServer, so a slow or hung tool cannot block status, kill_now,
# update_config, or remove. The task returns {:tick_result, flags} with the
# warning-throttle booleans, which the result handler applies to state.
defp spawn_tick_task(state) do
agent = state.agent
flags = %{gate_warning_logged: state.gate_warning_logged, pause_warning_logged: state.pause_warning_logged}
Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn -> run_tick(agent, flags) end)
end

defp run_tick(agent, flags) do
# intake runs every tick, even while paused, so nothing piles up unprocessed
checkin_summary = run_pre_command(agent)

globally_paused = File.exists?(agent.homepath <> "work/.global_pause")
node_paused = File.exists?(agent.homepath <> "work/.pause_" <> agent.id)

cond do
	globally_paused or node_paused ->
		pause_warning_logged = if node_paused and not flags.pause_warning_logged do
			write_log(agent, "Skipped: " <> agent.id <> " paused. This warning will not repeat until pause is lifted.")
			true
		else
			flags.pause_warning_logged
		end
		{:tick_result, %{gate_warning_logged: flags.gate_warning_logged, pause_warning_logged: pause_warning_logged}}

	true ->
		{gate_open, config_info} = check_gate(agent)
		if gate_open do
			run_agent_turn(agent, config_info, checkin_summary)
			{:tick_result, %{gate_warning_logged: false, pause_warning_logged: false}}
		else
			gate_warning_logged = if not flags.gate_warning_logged do
				write_log(agent, "Gate check failed (none of " <> Enum.join(agent.gate_keys, ", ") <> " are enabled). Tick skipped. This warning will not repeat until configuration changes.")
				true
			else
				flags.gate_warning_logged
			end
			{:tick_result, %{gate_warning_logged: gate_warning_logged, pause_warning_logged: false}}
		end
end
end

# Each tick is its own claude session. Prompt 1 tells the model to read the
# instructions file by path (never inlined) and runs the user's first prompt;
# any further prompts continue the SAME session via --resume, so the session
# itself carries the within-tick history. Cross-tick memory is the instructions
# file the agent maintains, not a re-injected transcript. Runs synchronously
# inside the tick task (see run_tick).
defp run_agent_turn(agent, config_info, checkin_summary) do
hp = agent.homepath

read_line = "Read the file at " <> hp <> agent.instructions_file <> " for your instructions.\n\n"
status_part = if config_info != "", do: "Current system status:\n" <> config_info <> "\n\n", else: ""
checkin_part = if checkin_summary != "", do: "Incoming check summary: " <> checkin_summary <> "\n\n", else: ""

[first | rest] = agent.prompts
first_prompt = read_line <> status_part <> checkin_part <> first

session_id = generate_session_id()
output_path = hp <> "autologs/" <> agent.id <> "/tick_" <> Integer.to_string(:os.system_time(:millisecond)) <> ".log"

family = model_family(agent.model)
if family == "claude" and not claude_auth_ok?() do
	File.write(output_path, "[error: claude is not logged in on this server. Run 'claude login' as the service user (ec2-user) to enable AI responses.]\n")
else
	run_prompts(agent, session_id, [first_prompt | rest], output_path, 0)
end
end

# Runs each prompt in order against the one session. Prompt 1 opens the session
# with --session-id, the rest --resume it. A non-zero exit aborts the chain so
# later prompts are not fired into a broken session.
defp run_prompts(_agent, _session_id, [], _output_path, _idx), do: {"", 0}
defp run_prompts(agent, session_id, [prompt | rest], output_path, idx) do
{cmd, args} = build_cli_command(agent.model, prompt, session_id, idx > 0, agent.permission_mode)
File.write(output_path, "\n> [prompt " <> Integer.to_string(idx + 1) <> "]\n" <> prompt <> "\n\n", [:append, :utf8])
result = System.cmd("/usr/bin/stdbuf", ["-oL", cmd | args], [cd: agent.homepath, into: File.stream!(output_path, [:append, :utf8])])
case result do
	{_, 0} -> run_prompts(agent, session_id, rest, output_path, idx + 1)
	{_, code} ->
		File.write(output_path, "\n\n[error: prompt " <> Integer.to_string(idx + 1) <> " exited non-zero (" <> Integer.to_string(code) <> "). Remaining prompts skipped.]\n", [:append])
		{"", code}
end
end


# ===== CLI COMMAND BUILDER (self-contained, no shared module) =====

# Permission mode is per-agent config (agents.json). An unset mode passes no
# permission flag, so the CLI uses its own default. Sessions persist on disk,
# which is what makes --resume work across the prompts in one tick.

defp model_family("codex" <> _), do: "codex"
defp model_family(_), do: "claude"

defp build_cli_command("claude-opus-high", prompt, sid, resume, mode), do: claude_cmd("opus", "high", sid, resume, mode, prompt)
defp build_cli_command("claude-opus-xhigh", prompt, sid, resume, mode), do: claude_cmd("opus", "xhigh", sid, resume, mode, prompt)
defp build_cli_command("claude-opus-max", prompt, sid, resume, mode), do: claude_cmd("opus", "max", sid, resume, mode, prompt)
defp build_cli_command("claude-sonnet", prompt, sid, resume, mode), do: claude_cmd("sonnet", nil, sid, resume, mode, prompt)
# Codex cannot take a caller-chosen session id (it runs --ephemeral), so a codex
# agent's follow-up prompts run as fresh, disconnected sessions. claude is the
# model that gives true multi-prompt continuity.
defp build_cli_command("codex-5.5-high", prompt, _sid, _resume, mode), do: codex_cmd("gpt-5.5", "high", mode, prompt)
defp build_cli_command("codex-5.5-xhigh", prompt, _sid, _resume, mode), do: codex_cmd("gpt-5.5", "xhigh", mode, prompt)
defp build_cli_command("codex-5.5", prompt, _sid, _resume, mode), do: codex_cmd("gpt-5.5", nil, mode, prompt)
defp build_cli_command("codex-mini", prompt, _sid, _resume, mode), do: codex_cmd("gpt-5.4-mini", nil, mode, prompt)
# Anything unrecognized falls back to claude sonnet.
defp build_cli_command(_, prompt, sid, resume, mode), do: claude_cmd("sonnet", nil, sid, resume, mode, prompt)

defp claude_cmd(model, effort, session_id, resume, mode, prompt) do
session_args = if resume, do: ["--resume", session_id], else: ["--session-id", session_id]
effort_args = if effort == nil, do: [], else: ["--effort", effort]
flags = session_args ++ effort_args ++ claude_permission_args(mode) ++ ["--model", model, "-p", prompt]
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | flags]}
end

defp codex_cmd(model, effort, mode, prompt) do
effort_args = if effort == nil, do: [], else: ["-c", "model_reasoning_effort=" <> effort]
flags = codex_permission_args(mode) ++ ["--skip-git-repo-check", "--ephemeral", "--model", model] ++ effort_args ++ [prompt]
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | flags]}
end

# An unset or unrecognized mode returns no flags, so the CLI uses its own
# default. Kept local to match this node's self-contained builder.
defp claude_permission_args("dontAsk"), do: ["--permission-mode", "dontAsk"]
defp claude_permission_args("plan"), do: ["--permission-mode", "plan"]
defp claude_permission_args("acceptEdits"), do: ["--permission-mode", "acceptEdits"]
defp claude_permission_args("full"), do: ["--dangerously-skip-permissions"]
defp claude_permission_args(_), do: []

defp codex_permission_args("read-only"), do: ["--sandbox", "read-only"]
defp codex_permission_args("workspace-write"), do: ["--sandbox", "workspace-write", "-c", "sandbox_workspace_write.network_access=true"]
defp codex_permission_args("full"), do: ["--sandbox", "danger-full-access"]
defp codex_permission_args(_), do: []


# ===== SESSION ID + AUTH (self-contained) =====

# Valid UUID v4, required format for claude --session-id.
defp generate_session_id do
<<u0::48, _::4, u1::12, _::2, u2::62>> = :crypto.strong_rand_bytes(16)
hex = Base.encode16(<<u0::48, 4::4, u1::12, 2::2, u2::62>>, case: :lower)
<<a::binary-size(8), b::binary-size(4), c::binary-size(4), d::binary-size(4), e::binary-size(12)>> = hex
a <> "-" <> b <> "-" <> c <> "-" <> d <> "-" <> e
end

# True if a claude OAuth credential file exists with a refresh token. We check
# the file directly because the claude CLI hangs when called without a TTY.
defp claude_auth_ok? do
home = System.get_env("HOME") || "/home/ec2-user"
creds_path = Path.join([home, ".claude", ".credentials.json"])
case File.read(creds_path) do
	{:ok, content} ->
		case Jason.decode(content) do
			{:ok, %{"claudeAiOauth" => %{"refreshToken" => rt}}} when is_binary(rt) and byte_size(rt) > 0 -> true
			_ -> false
		end
	{:error, _} -> false
end
end


# ===== PRE COMMAND AND GATE =====

defp run_pre_command(agent) do
case agent.pre_command do
	[cmd | args] when is_binary(cmd) ->
		{output, _} = System.cmd(cmd, args, [cd: agent.homepath, stderr_to_stdout: true])
		summary = String.trim(output)
		if has_activity?(summary), do: write_log(agent, "checkIncoming: " <> summary)
		summary
	_ -> ""
end
end

# any positive count in the checkin json means something came in worth logging
defp has_activity?(summary) do
case Jason.decode(summary) do
	{:ok, map} when is_map(map) ->
		Enum.any?(map, fn {_k, v} -> is_number(v) and v > 0 end)
	_ -> false
end
end

defp check_gate(agent) do
case agent.gate_command do
	[cmd | args] when is_binary(cmd) ->
		{output, _} = System.cmd(cmd, args, [cd: agent.homepath, stderr_to_stdout: true])
		info = String.trim(output)
		open = case Jason.decode(info) do
			{:ok, map} when is_map(map) ->
				agent.gate_keys == [] or Enum.any?(agent.gate_keys, fn k -> Map.get(map, k, false) == true end)
			_ -> false
		end
		{open, info}
	_ -> {true, ""}
end
end


# ===== MANAGER COMMANDS =====

def handle_cast({:update_config, agent}, state) do
state = %{state | removing: false}
if state.task_pid != nil and Process.alive?(state.task_pid) do
	{:noreply, %{state | pending_agent: agent}}
else
	{:noreply, apply_config(state, agent)}
end
end

def handle_cast(:remove, state) do
if state.task_pid != nil and Process.alive?(state.task_pid) do
	{:noreply, %{state | removing: true}}
else
	{:stop, :normal, state}
end
end

def handle_cast(:kill_now, state) do
if state.task_pid != nil and Process.alive?(state.task_pid) do
	Process.demonitor(state.task_ref, [:flush])
	Task.Supervisor.terminate_child(Ala.TaskSupervisor, state.task_pid)
	write_log(state.agent, "Current task killed by owner request. The agent keeps running and ticks again normally.")
	{:noreply, %{state | task_pid: nil, task_ref: 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


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

def handle_call(:status, _from, state) do
busy = state.task_pid != nil and Process.alive?(state.task_pid)
paused = File.exists?(state.agent.homepath <> "work/.pause_" <> state.agent.id)
status = cond do
	paused -> "paused"
	busy -> "busy"
	true -> "idle"
end
{:reply, %{status: status, tick: Integer.to_string(div(state.tick_ms, 60000)) <> "m", type: state.agent.type}, state}
end


# ===== CONFIG APPLY =====

defp apply_pending(state) do
case state.pending_agent do
	nil -> state
	agent -> apply_config(%{state | pending_agent: nil}, agent)
end
end

defp apply_config(state, agent) do
new_tick_ms = agent.tick_minutes * 60000
state = 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
%{state | agent: agent}
end


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

defp write_log(agent, message) do
path = agent.homepath <> "autologs/" <> agent.id <> "/log_" <> Integer.to_string(:os.system_time(:second)) <> ".txt"
File.write(path, message <> "\n")
end

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

end