conversation_turn.ex

defmodule ConversationTurn do
# Shared helper for the conversation flow. Nodes compose one turn by calling:
#   prompt = ConversationTurn.build_main_prompt(opts)
#   task   = ConversationTurn.spawn_main_task(model_key, prompt, output_path, trigger)
# On task completion:
#   ConversationTurn.spawn_digest(meta)

@auth_cache_ttl_ms 60_000

# Permission handling is per-agent / per-node now. An unset mode passes no
# permission flag at all, so the CLI falls back to its own default. The logic
# is: you did not set permissions, so we do not set any. See the
# permission_args helpers below for the mode -> flag mapping.

# Claude CLI session pattern (proven by prior automation scripts and verified against docs):
# first call in a session:  claude --session-id <uuid v4 we mint> --effort <level> --permission-mode acceptEdits --allowedTools <list> --model <m> -p <prompt>
# follow-up calls:          claude --resume <same uuid> <same flags> -p <prompt>
# We always mint the uuid ourselves so multi-turn sessions never need to parse json output for an id.

# ===== MAIN PROMPT =====

def build_main_prompt(opts) do
role_block = Map.get(opts, :role_block, "")
conversation_context = Map.get(opts, :conversation_context, "")
extra_context = Map.get(opts, :extra_context, "")
trigger_text = Map.get(opts, :trigger_text, "")
role_block <> conversation_context <> extra_context <> trigger_text
end

def fetch_recent_conversation(homepath, scope, conversation, n) do
query_script = Path.join(homepath, "tools/conversation_query.php")

case System.cmd("php", [query_script, "recent", scope, conversation, Integer.to_string(n)], [stderr_to_stdout: true]) do
	{output, 0} ->
		case Jason.decode(String.trim(output)) do
			{:ok, %{"turns" => turns}} when is_list(turns) and length(turns) > 0 ->
				reversed = Enum.reverse(turns)
				lines = Enum.map(reversed, fn t ->
					role = Map.get(t, "role", "unknown")
					text = Map.get(t, "text", "")
					role <> ": " <> text
				end)
				"[Recent conversation - last " <> Integer.to_string(length(reversed)) <> " messages]\n\n" <> Enum.join(lines, "\n\n") <> "\n\n"
			_ -> ""
		end
	_ -> ""
end
end


def fetch_context_extras(homepath, ctx, user_text) do
parts = []

parts = if Map.get(ctx, "include_flags") == true do
	parts ++ [fetch_flags_context(homepath)]
else
	parts
end

parts = if Map.get(ctx, "auto_query") == true do
	char_limit = case Map.get(ctx, "auto_query_char_limit") do
		n when is_integer(n) and n > 0 -> n
		_ -> 2000
	end
	parts ++ [fetch_auto_query_context(homepath, user_text, char_limit)]
else
	parts
end

parts = if Map.get(ctx, "auto_query") == true do
	parts ++ [fetch_skills_context(homepath, user_text)]
else
	parts
end

goal_ids = Map.get(ctx, "selected_goals", [])
parts = if is_list(goal_ids) and length(goal_ids) > 0 do
	parts ++ [fetch_goals_context(homepath, goal_ids)]
else
	parts
end

topic_ids = Map.get(ctx, "selected_topics", [])
parts = if is_list(topic_ids) and length(topic_ids) > 0 do
	parts ++ [fetch_topic_digests_context(homepath, topic_ids)]
else
	parts
end

sys_prompt = Map.get(ctx, "system_prompt", "")
parts = if is_binary(sys_prompt) and byte_size(String.trim(sys_prompt)) > 0 do
	parts ++ ["[User instructions]\n" <> String.trim(sys_prompt) <> "\n\n"]
else
	parts
end

parts |> Enum.filter(fn s -> byte_size(s) > 0 end) |> Enum.join("")
end


def fetch_flags_context(homepath) do
flags_dir = Path.join(homepath, "threads/flags")
case File.ls(flags_dir) do
	{:ok, files} ->
		flag_files = files |> Enum.filter(fn f -> not String.starts_with?(f, ".") end) |> Enum.sort() |> Enum.take(20)
		if length(flag_files) == 0 do
			""
		else
			lines = Enum.map(flag_files, fn f ->
				content = case File.read(Path.join(flags_dir, f)) do
					{:ok, c} -> String.slice(c, 0, 200)
					_ -> ""
				end
				"- " <> String.trim(content)
			end)
			"[Active flags - " <> Integer.to_string(length(flag_files)) <> " pending]\n" <> Enum.join(lines, "\n") <> "\n\n"
		end
	_ -> ""
end
end


def fetch_auto_query_context(homepath, user_text, char_limit) do
query_script = Path.join(homepath, "tools/memory_query.php")
safe_text = String.slice(user_text, 0, 500)
case System.cmd("php", [query_script, "query", "all", safe_text], [stderr_to_stdout: true, cd: homepath]) do
	{output, 0} ->
		case Jason.decode(String.trim(output)) do
			{:ok, %{"results" => results}} when is_list(results) ->
				high = results
					|> Enum.filter(fn r ->
						score = case Map.get(r, "score", 0) do
							s when is_binary(s) -> case Float.parse(s) do {f, _} -> f; _ -> 0.0 end
							n when is_number(n) -> n
							_ -> 0.0
						end
						score >= 0.7
					end)
					|> Enum.sort_by(fn r ->
						case Map.get(r, "score", 0) do
							s when is_binary(s) -> case Float.parse(s) do {f, _} -> f; _ -> 0.0 end
							n when is_number(n) -> n
							_ -> 0.0
						end
					end, :desc)
				if length(high) == 0 do
					""
				else
					{lines, _used} = Enum.reduce_while(high, {[], 0}, fn r, {acc, used} ->
						summary = Map.get(r, "summary", "")
						entry_type = Map.get(r, "type", "")
						line = "- [" <> entry_type <> "] " <> summary
						new_used = used + byte_size(line) + 1
						if new_used > char_limit do
							{:halt, {acc, used}}
						else
							{:cont, {acc ++ [line], new_used}}
						end
					end)
					if length(lines) == 0 do
						""
					else
						"<memory-context source=\"auto_query\">\n" <> Enum.join(lines, "\n") <> "\n</memory-context>\n\n"
					end
				end
			_ -> ""
		end
	_ -> ""
end
end


def fetch_skills_context(homepath, user_text) do
query_script = Path.join(homepath, "tools/memory_query.php")
safe_text = String.slice(user_text, 0, 300)
case System.cmd("php", [query_script, "query", "learned_procedure", safe_text], [stderr_to_stdout: true, cd: homepath]) do
	{output, 0} ->
		case Jason.decode(String.trim(output)) do
			{:ok, %{"results" => results}} when is_list(results) ->
				matched = results
					|> Enum.filter(fn r ->
						score = case Map.get(r, "score", 0) do
							s when is_binary(s) -> case Float.parse(s) do {f, _} -> f; _ -> 0.0 end
							n when is_number(n) -> n
							_ -> 0.0
						end
						score >= 0.6
					end)
					|> Enum.take(2)
				if length(matched) == 0 do
					""
				else
					lines = Enum.map(matched, fn r ->
						name = Map.get(r, "name", "")
						summary = Map.get(r, "summary", "")
						"- [procedure] " <> name <> ": " <> summary
					end)
					"<memory-context source=\"learned_procedures\">\n" <> Enum.join(lines, "\n") <> "\n</memory-context>\n\n"
				end
			_ -> ""
		end
	_ -> ""
end
end


def fetch_topic_digests_context(homepath, topic_ids) when is_list(topic_ids) do
query_script = Path.join(homepath, "tools/topic_query.php")
digests = Enum.map(topic_ids, fn tid ->
	case System.cmd("php", [query_script, "get", tid, "1"], [stderr_to_stdout: true]) do
		{output, 0} ->
			case Jason.decode(String.trim(output)) do
				{:ok, %{"digest" => digest}} when is_binary(digest) and byte_size(digest) > 0 -> digest
				_ -> ""
			end
		_ -> ""
	end
end) |> Enum.filter(fn d -> byte_size(d) > 0 end)

if length(digests) == 0 do
	""
else
	"<memory-context source=\"pinned_topics\">\n" <> Enum.join(digests, "\n---\n") <> "\n</memory-context>\n\n"
end
end
def fetch_topic_digests_context(_, _), do: ""


def fetch_goals_context(homepath, goal_ids) when is_list(goal_ids) do
goals_path = Path.join(homepath, "threads/goals/goals.json")
case File.read(goals_path) do
	{:ok, content} ->
		case Jason.decode(content) do
			{:ok, %{"goals" => goals}} when is_list(goals) ->
				id_set = MapSet.new(goal_ids)
				matched = Enum.filter(goals, fn g -> MapSet.member?(id_set, Map.get(g, "id", "")) end)
				if length(matched) == 0 do
					""
				else
					lines = Enum.map(matched, fn g ->
						name = Map.get(g, "name", "")
						status = Map.get(g, "status", "active")
						desc = Map.get(g, "description", "")
						"- " <> name <> " (" <> status <> "): " <> desc
					end)
					"<memory-context source=\"pinned_goals\">\n" <> Enum.join(lines, "\n") <> "\n</memory-context>\n\n"
				end
			_ -> ""
		end
	_ -> ""
end
end
def fetch_goals_context(_, _), do: ""


# ===== MAIN CALL (spawn) =====

def spawn_main_task(model_family, prompt, output_path, trigger_text, opts \\ []) do
homepath = Keyword.get(opts, :homepath, __DIR__)
conversation = Keyword.get(opts, :conversation)
env = if is_binary(conversation), do: [{"ALA_CONVERSATION", conversation}], else: []
File.write(output_path, "> " <> trigger_text <> "\n\n")

# Multi-turn callers (coding/research pipelines) pass session_id and set
# resume: true after the first step. One-shot callers pass nothing and get
# a fresh session id per call.
session_id = Keyword.get(opts, :session_id)
resume = Keyword.get(opts, :resume, false) == true
mode = Keyword.get(opts, :permission_mode)
session_args = cond do
	is_binary(session_id) and resume -> ["--resume", session_id]
	is_binary(session_id) -> ["--session-id", session_id]
	true -> ["--session-id", generate_session_id()]
end

family = normalize_model_family(model_family)
cond do
	family == "claude" and not claude_auth_ok?() ->
		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.]", [:append])
		Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn -> {"", 1} end)

	true ->
		{cmd, args, opts2} = build_main_cli_command(model_family, prompt, session_args, mode)
		Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
			result = System.cmd("/usr/bin/stdbuf", ["-oL" | [cmd | args]], opts2 ++ [cd: homepath, env: env, into: File.stream!(output_path, [:append, :utf8])])
			case result do
				{_, 0} -> :ok
				{_, _} -> File.write(output_path, "\n\n[error: AI call exited non-zero. Check the service log for details.]", [:append])
			end
			result
		end)
end
end


# ===== AUTH PROBE =====

# Returns true if a claude OAuth credential file exists with a refresh token.
# We check the file directly rather than invoking 'claude -p', because the
# claude CLI hangs when called without a TTY (which is the situation under
# systemd / System.cmd). Result is cached per calling process for
# @auth_cache_ttl_ms so concurrent ticks do not re-stat the file repeatedly.
def claude_auth_ok? do
case Process.get(:cns_claude_auth_check) do
	{ts, ok} when is_integer(ts) ->
		now = System.monotonic_time(:millisecond)
		if now - ts < @auth_cache_ttl_ms, do: ok, else: probe_and_cache()
	_ -> probe_and_cache()
end
end

defp probe_and_cache do
home = System.get_env("HOME") || "/home/ec2-user"
creds_path = Path.join([home, ".claude", ".credentials.json"])
ok = 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
# Only cache positive results, so a fresh `claude login` is picked up on the
# next tick instead of being masked by a cached negative for the full TTL.
if ok, do: Process.put(:cns_claude_auth_check, {System.monotonic_time(:millisecond), ok})
ok
end


# ===== DIGEST SPAWN =====

def spawn_digest(meta) do
# meta: %{homepath, scope, conversation, user_text, output_path, model_family}
prefix = "> " <> meta.user_text <> "\n\n"
assistant_text = case File.read(meta.output_path) do
	{:ok, content} -> String.replace_prefix(content, prefix, "")
	{:error, _} -> ""
end

if byte_size(String.trim(assistant_text)) == 0 do
	:skip_empty_digest
else
	ts = Integer.to_string(:os.system_time(:millisecond))
	digest_dir = Path.join(meta.homepath, "threads/prompts/.digest")
	File.mkdir_p(digest_dir)
	user_path = Path.join(digest_dir, ts <> "_user.txt")
	asst_path = Path.join(digest_dir, ts <> "_asst.txt")
	File.write(user_path, meta.user_text)
	File.write(asst_path, assistant_text)

	log_path = Path.join(meta.homepath, "autologs/digest/digest_" <> ts <> ".log")
	digest_script = Path.join(meta.homepath, "tools/digest_turn.py")

	cmd = "nohup python3 " <> shell_escape(digest_script) <> " " <>
		shell_escape(meta.scope) <> " " <>
		shell_escape(meta.conversation) <> " " <>
		shell_escape(user_path) <> " " <>
		shell_escape(asst_path) <> " " <>
		shell_escape(meta.model_family) <>
		" > " <> shell_escape(log_path) <> " 2>&1 &"
	System.cmd("/bin/sh", ["-c", cmd])
end
end


# ===== CLI COMMAND BUILDERS =====

# Public entry for nodes that run the CLI themselves (e.g. impulse).
# Mints a fresh session id. spawn_main_task passes explicit session args.
def build_main_cli_command(model_family, prompt, mode) do
build_main_cli_command(model_family, prompt, ["--session-id", generate_session_id()], mode)
end

# Claude variants. The permission mode is per-agent / per-node config; an
# unset mode passes no flag at all. Sessions persist on disk, which is what
# makes --resume work for multi-turn pipelines.
defp build_main_cli_command("claude-opus-high", prompt, session_args, mode) do
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | claude_args("opus", "high", mode, session_args, prompt)], []}
end
defp build_main_cli_command("claude-opus-xhigh", prompt, session_args, mode) do
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | claude_args("opus", "xhigh", mode, session_args, prompt)], []}
end
defp build_main_cli_command("claude-opus-max", prompt, session_args, mode) do
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | claude_args("opus", "max", mode, session_args, prompt)], []}
end
defp build_main_cli_command("claude-sonnet", prompt, session_args, mode) do
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | claude_args("sonnet", nil, mode, session_args, prompt)], []}
end

# Codex variants. The sandbox mode is the codex permission lever, mapped from
# the same per-agent / per-node config. Codex cannot take a caller-chosen
# session id, so sessions stay ephemeral.
defp build_main_cli_command("codex-5.5", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.5", nil, mode, prompt)], []}
end
defp build_main_cli_command("codex-5.5-high", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.5", "high", mode, prompt)], []}
end
defp build_main_cli_command("codex-5.5-xhigh", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.5", "xhigh", mode, prompt)], []}
end
# small model option; codex-4o kept as an alias for configs that predate the rename
defp build_main_cli_command("codex-mini", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.4-mini", nil, mode, prompt)], []}
end
defp build_main_cli_command("codex-4o", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.4-mini", nil, mode, prompt)], []}
end

# Ollama local models. No permission concept, it is a local text model.
defp build_main_cli_command("ollama-" <> model_name, prompt, _session_args, _mode) do
{"ollama", ["run", model_name, prompt], []}
end

# Bare provider defaults (from pipeline nodes)
defp build_main_cli_command("codex", prompt, _session_args, mode) do
{"sh", ["-c", "exec codex exec \"$@\" </dev/null", "_" | codex_args("gpt-5.5", nil, mode, prompt)], []}
end
defp build_main_cli_command(_, prompt, session_args, mode) do
{"sh", ["-c", "exec claude \"$@\" </dev/null", "_" | session_args ++ claude_permission_args(mode) ++ ["-p", prompt]], []}
end

defp claude_args(model, effort, mode, session_args, prompt) do
effort_args = if effort == nil, do: [], else: ["--effort", effort]
session_args ++ effort_args ++ claude_permission_args(mode) ++ ["--model", model, "-p", prompt]
end

# codex exec never prompts, the sandbox mode alone decides what is allowed
defp codex_args(model, effort, mode, prompt) do
effort_args = if effort == nil, do: [], else: ["-c", "model_reasoning_effort=" <> effort]
codex_permission_args(mode) ++ ["--skip-git-repo-check", "--ephemeral", "--model", model] ++ effort_args ++ [prompt]
end


# ===== PERMISSION FLAGS (per agent / per node) =====

# An unset or unrecognized mode returns no flags, so the CLI uses its own
# default. dontAsk takes no gated action, plan is read-only, acceptEdits is
# file edits only, full skips all checks.
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"]
# Cross-provider: a codex-labelled value chosen for a claude agent maps to the
# nearest claude mode instead of being silently dropped to no flag.
defp claude_permission_args("read-only"), do: ["--permission-mode", "plan"]
defp claude_permission_args("workspace-write"), do: ["--permission-mode", "acceptEdits"]
defp claude_permission_args(_), do: []

# workspace-write keeps the network-access override the tools layer relied on.
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"]
# Cross-provider: a claude-labelled value chosen for a codex agent maps to the
# nearest codex sandbox instead of being silently dropped to no flag.
defp codex_permission_args("plan"), do: ["--sandbox", "read-only"]
defp codex_permission_args("dontAsk"), do: ["--sandbox", "workspace-write", "-c", "sandbox_workspace_write.network_access=true"]
defp codex_permission_args("acceptEdits"), do: ["--sandbox", "workspace-write", "-c", "sandbox_workspace_write.network_access=true"]
defp codex_permission_args(_), do: []


# ===== UTILITIES =====

# Valid UUID v4, required format for claude --session-id.
def 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

def normalize_model_family("codex" <> _), do: "codex"
def normalize_model_family("ollama-" <> _), do: "ollama"
def normalize_model_family(_), do: "claude"

defp shell_escape(s), do: "'" <> String.replace(s, "'", "'\\''") <> "'"

end