[codex] harden familiar mnesia recovery

.github/workflows/verify.yml

@@ -47,11 +47,12 @@ jobs:
   # Live integration tests against a real provider. Several real bugs in v1
   # prep (streaming tool calls dropped, multi-send losing assistant history)
   # shipped past unit tests because the mocks didn't match real provider
-  # behavior. This costs API tokens, so PRs run unit verification only; main,
-  # release branch, and tag pushes must have the Anthropic secret configured.
+  # behavior. Runs on PRs as well as main/release/tag pushes: every PR here
+  # is a same-repo branch (secrets available, no fork risk), the suite costs
+  # about a minute of Haiku, and a gate belongs before the merge, not after.
+  # The Anthropic secret must be configured; the job hard-fails without it.
   live:
     runs-on: ubuntu-latest
-    if: github.event_name == 'push'
 
     steps:
       - name: Checkout

lib/cantrip.ex

@@ -1513,7 +1513,13 @@ defmodule Cantrip do
   defp validate_mnesia_storage_opts(opts) when is_map(opts) or is_list(opts) do
     opts = opts |> normalize_input_map() |> prefer_atom_keys()
 
-    case NimbleOptions.validate(Map.to_list(opts), table: [type: :atom], mnesia: [type: :atom]) do
+    schema = [
+      table: [type: :atom],
+      mnesia: [type: :atom],
+      on_corrupt: [type: {:in, [:quarantine]}]
+    ]
+
+    case NimbleOptions.validate(Map.to_list(opts), schema) do
       {:ok, validated} -> {:ok, Map.new(validated)}
       {:error, %NimbleOptions.ValidationError{message: msg}} -> {:error, msg}
     end

lib/cantrip/loom.ex

@@ -113,6 +113,10 @@ defmodule Cantrip.Loom do
           * The storage backend's prerequisites aren't met (e.g.
             disk path is unwritable, Mnesia schema not created on
             this node)
+          * A prior BEAM died while Mnesia was writing and left a
+            corrupt transaction log such as LATEST.LOG. Preserve the
+            old store for forensics and start fresh by passing
+            `on_corrupt: :quarantine` in the Mnesia loom options.
 
         If you want to allow falling back to in-memory loom, do not
         pass `:loom_storage` (or pass `nil`) when constructing the

lib/cantrip/loom/storage/mnesia.ex

@@ -16,7 +16,8 @@ defmodule Cantrip.Loom.Storage.Mnesia do
       mnesia = Map.get(opts, :mnesia, :mnesia)
 
       case with_schema_lock(fn ->
-             with :ok <- ensure_mnesia_started(mnesia),
+             with :ok <- preflight_store(opts, mnesia),
+                  :ok <- ensure_mnesia_started(mnesia),
                   :ok <- ensure_table(table, mnesia) do
                {:ok, %{table: table, mnesia: mnesia}}
              end
@@ -33,7 +34,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do
     key = System.unique_integer([:positive, :monotonic])
     event = storage_event(%{type: :turn, turn: turn})
 
-    case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do
+    case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do
       {:atomic, :ok} -> {:ok, state}
       {:aborted, reason} -> {:error, reason}
       other -> {:error, other}
@@ -46,7 +47,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do
     key = System.unique_integer([:positive, :monotonic])
     event = storage_event(%{type: :reward, index: index, reward: reward})
 
-    case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do
+    case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do
       {:atomic, :ok} -> {:ok, state}
       {:aborted, reason} -> {:error, reason}
       other -> {:error, other}
@@ -59,7 +60,7 @@ defmodule Cantrip.Loom.Storage.Mnesia do
     key = System.unique_integer([:positive, :monotonic])
     event = storage_event(event)
 
-    case call(mnesia, :transaction, [fn -> call(mnesia, :write, [{table, key, event}]) end]) do
+    case durable_transaction(mnesia, fn -> call(mnesia, :write, [{table, key, event}]) end) do
       {:atomic, :ok} -> {:ok, state}
       {:aborted, reason} -> {:error, reason}
       other -> {:error, other}
@@ -128,6 +129,14 @@ defmodule Cantrip.Loom.Storage.Mnesia do
     end
   end
 
+  defp durable_transaction(mnesia, fun) when is_function(fun, 0) do
+    if function_exported?(mnesia, :sync_transaction, 1) do
+      call(mnesia, :sync_transaction, [fun])
+    else
+      call(mnesia, :transaction, [fun])
+    end
+  end
+
   defp ensure_mnesia_started(mnesia) do
     case call(mnesia, :system_info, [:is_running]) do
       :yes ->
@@ -145,6 +154,127 @@ defmodule Cantrip.Loom.Storage.Mnesia do
     end
   end
 
+  defp preflight_store(opts, mnesia) do
+    case call(mnesia, :system_info, [:is_running]) do
+      :yes ->
+        :ok
+
+      _ ->
+        dir = mnesia_dir()
+
+        case corrupt_log_files(dir) do
+          [] ->
+            :ok
+
+          corrupt ->
+            maybe_quarantine_corrupt_store(dir, corrupt, opts)
+        end
+    end
+  rescue
+    _ -> :ok
+  catch
+    :exit, _ -> :ok
+  end
+
+  defp mnesia_dir do
+    case Application.get_env(:mnesia, :dir) do
+      nil -> ~c"Mnesia.#{node()}"
+      dir -> dir
+    end
+    |> to_string()
+  end
+
+  defp corrupt_log_files(dir) do
+    ["LATEST.LOG", "PREVIOUS.LOG"]
+    |> Enum.map(&Path.join(dir, &1))
+    |> Enum.filter(&File.regular?/1)
+    |> Enum.flat_map(fn path ->
+      case probe_log(path) do
+        :ok -> []
+        {:error, reason} -> [{path, reason}]
+      end
+    end)
+  end
+
+  defp probe_log(path) do
+    name = :"#{__MODULE__}.probe.#{System.unique_integer([:positive])}"
+
+    case :disk_log.open(name: name, file: String.to_charlist(path), mode: :read_only) do
+      {:ok, ^name} ->
+        :disk_log.close(name)
+        :ok
+
+      {:repaired, ^name, _recovered, _badbytes} ->
+        :disk_log.close(name)
+        :ok
+
+      {:error, {:need_repair, ^name}} ->
+        :disk_log.close(name)
+        :ok
+
+      {:error, {:not_a_log_file, _} = reason} ->
+        {:error, reason}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  defp maybe_quarantine_corrupt_store(dir, corrupt, opts) do
+    case Map.get(opts, :on_corrupt) || Map.get(opts, "on_corrupt") do
+      :quarantine ->
+        quarantine_corrupt_store(dir, corrupt)
+
+      "quarantine" ->
+        quarantine_corrupt_store(dir, corrupt)
+
+      _ ->
+        {:error,
+         {:corrupt_mnesia_store,
+          %{
+            dir: dir,
+            corrupt_logs: corrupt,
+            repair: "pass on_corrupt: :quarantine to move this store aside and start a fresh loom"
+          }}}
+    end
+  end
+
+  defp quarantine_corrupt_store(dir, corrupt) do
+    quarantine_dir = quarantine_dir(dir)
+    do_quarantine_corrupt_store(dir, quarantine_dir, corrupt)
+  end
+
+  defp do_quarantine_corrupt_store(dir, quarantine_dir, corrupt) do
+    with :ok <- File.mkdir_p!(Path.dirname(quarantine_dir)),
+         :ok <- File.rename(dir, quarantine_dir),
+         :ok <- File.mkdir_p(dir) do
+      :ok
+    else
+      {:error, reason} ->
+        {:error,
+         {:corrupt_mnesia_store_quarantine_failed,
+          %{dir: dir, quarantine_dir: quarantine_dir, corrupt_logs: corrupt, reason: reason}}}
+    end
+  rescue
+    e ->
+      {:error,
+       {:corrupt_mnesia_store_quarantine_failed,
+        %{
+          dir: dir,
+          quarantine_dir: quarantine_dir,
+          corrupt_logs: corrupt,
+          reason: Exception.message(e)
+        }}}
+  end
+
+  defp quarantine_dir(dir) do
+    stamp =
+      DateTime.utc_now()
+      |> Calendar.strftime("%Y%m%dT%H%M%SZ")
+
+    "#{dir}-corrupt-#{stamp}-#{System.unique_integer([:positive])}"
+  end
+
   defp ensure_schema(mnesia) do
     case call(mnesia, :create_schema, [[node()]]) do
       :ok -> :ok

lib/mix/tasks/cantrip.familiar.ex

@@ -165,7 +165,7 @@ defmodule Mix.Tasks.Cantrip.Familiar do
         case :net_kernel.start([name, :longnames]) do
           {:ok, _} ->
             :erlang.set_cookie(node(), cookie)
-            configure_mnesia_dir!(workspace_root)
+            configure_mnesia_workspace!(workspace_root)
 
           {:error, {:already_started, _}} ->
             :ok
@@ -190,15 +190,17 @@ defmodule Mix.Tasks.Cantrip.Familiar do
       _named ->
         # Already named (someone launched with --sname/--name). Trust
         # their setup; just relocate Mnesia under .cantrip/.
-        configure_mnesia_dir!(workspace_root)
+        configure_mnesia_workspace!(workspace_root)
     end
   end
 
   # Point Mnesia at `.cantrip/mnesia/` for this workspace. Mnesia is
   # in `included_applications` (not `extra_applications`), so it's
-  # loaded but not yet started. Setting `:dir` before the adapter's
-  # lazy `:mnesia.start/0` is enough — no stop/restart cycle, no
-  # orphaned `Mnesia.<node>/` dir at cwd from a premature auto-start.
+  # loaded but not yet started. Setting `:dir` and related durability
+  # knobs before the adapter's lazy `:mnesia.start/0` is enough — no
+  # stop/restart cycle, no orphaned `Mnesia.<node>/` dir at cwd from a
+  # premature auto-start, and no repo-root `MnesiaCore.*` files if
+  # Mnesia fatals while recovering a transaction log.
   #
   # Verified empirically: after `mix run`, `Application.started_applications/0`
   # does not include `:mnesia`, and `:mnesia.system_info(:tables)`
@@ -207,13 +209,91 @@ defmodule Mix.Tasks.Cantrip.Familiar do
   # started with the parent" concern doesn't apply here because
   # `Cantrip.Application.start/2` never calls `Application.ensure_*`
   # on Mnesia.
-  defp configure_mnesia_dir!(workspace_root) do
-    desired = Path.join([workspace_root, ".cantrip", "mnesia"]) |> String.to_charlist()
-    File.mkdir_p!(to_string(desired))
-    Application.put_env(:mnesia, :dir, desired)
+  @doc false
+  @spec configure_mnesia_workspace!(String.t()) :: :ok
+  def configure_mnesia_workspace!(workspace_root) when is_binary(workspace_root) do
+    env = mnesia_env_for_workspace(workspace_root)
+
+    env
+    |> Keyword.fetch!(:dir)
+    |> to_string()
+    |> File.mkdir_p!()
+
+    env
+    |> Keyword.fetch!(:core_dir)
+    |> to_string()
+    |> File.mkdir_p!()
+
+    ensure_mnesia_configurable!(env)
+
+    for {key, value} <- env do
+      Application.put_env(:mnesia, key, value)
+    end
+
     :ok
   end
 
+  @doc false
+  @spec mnesia_env_for_workspace(String.t()) :: keyword()
+  def mnesia_env_for_workspace(workspace_root) when is_binary(workspace_root) do
+    mnesia_dir = Path.join([workspace_root, ".cantrip", "mnesia"])
+
+    [
+      dir: String.to_charlist(mnesia_dir),
+      core_dir: String.to_charlist(Path.join(mnesia_dir, "cores")),
+      auto_repair: true,
+      dc_dump_limit: 16,
+      dump_log_write_threshold: 100,
+      dump_log_time_threshold: 30_000
+    ]
+  end
+
+  defp ensure_mnesia_configurable!(env) do
+    if mnesia_running?() do
+      mismatches =
+        env
+        |> Enum.flat_map(fn {key, expected} ->
+          info_key = if key == :dir, do: :directory, else: key
+          actual = :mnesia.system_info(info_key)
+
+          if actual == expected do
+            []
+          else
+            [{key, expected, actual}]
+          end
+        end)
+
+      if mismatches != [] do
+        raise """
+        Mnesia is already running with workspace settings that do not match this Familiar.
+
+        #{format_mnesia_mismatches(mismatches)}
+
+        Restart the BEAM before launching the workspace Familiar, or opt out of
+        Mnesia with:
+
+          mix cantrip.familiar --loom-path .cantrip/familiar.jsonl
+        """
+      end
+    end
+  end
+
+  defp mnesia_running? do
+    Code.ensure_loaded?(:mnesia) and :mnesia.system_info(:is_running) == :yes
+  rescue
+    _ -> false
+  catch
+    :exit, _ -> false
+  end
+
+  defp format_mnesia_mismatches(mismatches) do
+    mismatches
+    |> Enum.map(fn {key, expected, actual} ->
+      "  * #{key}: expected #{inspect(expected)}, got #{inspect(actual)}"
+    end)
+    |> Enum.join("\n")
+  end
+
   # `System.cmd("epmd", ["-daemon"], ...)` raises `ErlangError` when
   # epmd is not on PATH. Catching here keeps the actionable
   # `--loom-path` error message in `ensure_named_node!` reachable