defmodule OpenAgents.Forge.WAL do
@moduledoc """
Write-ahead log for OpenAgents.Forge git repositories.
The WAL in object storage is the source of truth for git refs; node disks
are only a cache. Every push appends one immutable entry object (the packed
objects plus ref updates) and then advances a single per-repo index document
through a compare-and-swap on the storage generation. Because the index CAS
serializes all pushes, two nodes can never both believe they advanced the
same ref — the loser sees `:cas_conflict`, refetches, and retries.
The index is a JSON document:
%{
"version" => 1,
"entries" => [
%{
"seq" => 0,
"object" => "entries/00000000-<sha256 prefix>",
"refs" => %{"refs/heads/main" => "<sha>"},
"principal" => "...",
"pushed_at" => "2026-08-18T00:00:00Z",
"link" => "<sha256 of the previous link and this entry>"
}
],
"refs" => %{"refs/heads/main" => "<sha>"}
}
`"link"` chains each entry to the one before it (`chain_link/2`), so a
rewritten entry invalidates every link after it. `EXIT-005` states what that
does and does not prove; the chain is not published anywhere outside this
storage yet, so it does not by itself detect an operator who recomputes the
whole chain.
Adapters implement the storage behaviour. The configured adapter comes from
`Application.get_env(:openagents, :forge_wal_adapter)` and defaults to
`OpenAgents.Forge.WAL.Local`; production uses `OpenAgents.Forge.WAL.Gcs`.
"""
@typedoc "Repository name: `[a-z0-9][a-z0-9_-]*`, no `.git` suffix, no slashes."
@type repo :: String.t()
@typedoc "Opaque adapter-specific generation (integer for Local, GCS generation string for GCS)."
@type generation :: term()
@typedoc "Decoded JSON index document."
@type index :: map()
@callback read_index(repo) :: {:ok, generation, index} | {:error, :not_found} | {:error, term}
@callback cas_index(repo, expected :: generation | :none, index) ::
{:ok, generation} | {:error, :cas_conflict} | {:error, term}
@callback put_entry(repo, seq :: non_neg_integer(), payload :: binary()) ::
{:ok, object_key :: String.t()} | {:error, term}
@callback put_entry_file(repo, seq :: non_neg_integer(), path :: String.t()) ::
{:ok, object_key :: String.t()} | {:error, term}
@callback get_entry(repo, object_key :: String.t()) :: {:ok, binary()} | {:error, term}
@callback get_entry_file(repo, object_key :: String.t(), path :: String.t()) ::
:ok | {:error, term}
@callback put_object(repo, object_key :: String.t(), payload :: binary()) ::
{:ok, String.t()} | {:error, term}
@callback delete_repo(repo) :: :ok | {:error, term}
@chain_field "link"
@chain_domain "openagents.forge.wal.link.v1"
@repo_pattern ~r/^[a-z0-9](?:[a-z0-9_-]|\.(?=[a-z0-9]))*$/
@entry_key_pattern ~r/^entries\/[0-9]{8}-[0-9a-f]{12}$/
@artifact_key_pattern ~r/^artifacts\/[0-9a-f]{64}\.tar$/
## Dispatcher
@doc """
Read the current index for `repo` through the configured adapter.
Returns `{:ok, generation, index}` where `generation` is the opaque token
`cas_index/3` must be given to advance the index.
"""
@spec read_index(repo) :: {:ok, generation, index} | {:error, :not_found} | {:error, term}
def read_index(repo) do
with :ok <- validate_repo(repo), do: adapter().read_index(repo)
end
@doc """
Compare-and-swap the index for `repo`.
`expected` is the generation returned by the last `read_index/1`, or `:none`
to create the index only if it does not exist yet (first push). Returns the
new generation on success and `{:error, :cas_conflict}` when the stored
index moved under the caller.
"""
@spec cas_index(repo, generation | :none, index) ::
{:ok, generation} | {:error, :cas_conflict} | {:error, term}
def cas_index(repo, expected, index) when is_map(index) do
with :ok <- validate_repo(repo), do: adapter().cas_index(repo, expected, index)
end
@doc """
Store one immutable WAL entry payload for `repo` at sequence `seq`.
Returns the object key to record in the index entry. Idempotent: the key is
derived from the sequence number and the payload hash, so re-putting the
same payload yields the same key.
"""
@spec put_entry(repo, non_neg_integer(), binary()) :: {:ok, String.t()} | {:error, term}
def put_entry(repo, seq, payload) when is_integer(seq) and seq >= 0 and is_binary(payload) do
with :ok <- validate_repo(repo), do: adapter().put_entry(repo, seq, payload)
end
@doc """
Stream one immutable WAL entry from `path` through the configured adapter.
The adapter derives the same content-addressed key as `put_entry/3` without
loading the complete file into the BEAM heap.
"""
@spec put_entry_file(repo, non_neg_integer(), String.t()) ::
{:ok, String.t()} | {:error, term}
def put_entry_file(repo, seq, path)
when is_integer(seq) and seq >= 0 and is_binary(path) do
with :ok <- validate_repo(repo),
{:ok, %File.Stat{type: :regular}} <- File.stat(path) do
adapter().put_entry_file(repo, seq, path)
else
{:ok, _not_regular} -> {:error, :invalid_entry_file}
{:error, reason} -> {:error, reason}
end
end
@doc """
Fetch a previously stored WAL entry payload for `repo` by its object key.
"""
@spec get_entry(repo, String.t()) :: {:ok, binary()} | {:error, term}
def get_entry(repo, object_key) when is_binary(object_key) do
with :ok <- validate_repo(repo),
:ok <- validate_entry_key(object_key) do
adapter().get_entry(repo, object_key)
end
end
@doc "Stream a WAL entry into `path` without retaining its complete body in memory."
@spec get_entry_file(repo, String.t(), String.t()) :: :ok | {:error, term}
def get_entry_file(repo, object_key, path)
when is_binary(object_key) and is_binary(path) do
with :ok <- validate_repo(repo),
:ok <- validate_entry_key(object_key) do
adapter().get_entry_file(repo, object_key, path)
end
end
@doc "Delete every WAL object for one repository. This operation is idempotent."
@spec delete_repo(repo) :: :ok | {:error, term}
def delete_repo(repo) do
with :ok <- validate_repo(repo), do: adapter().delete_repo(repo)
end
@doc """
Store a named artifact blob alongside the WAL (P6, #123): built beam tars
land here so a replaced node — whose local artifact cache is empty — can
still boot-converge to the promoted target. Cache, not authority: the
same content is re-buildable from the pushed commit.
"""
@spec put_artifact(repo, String.t(), binary()) :: {:ok, String.t()} | {:error, term}
def put_artifact(repo, digest, payload) when is_binary(digest) and is_binary(payload) do
key = artifact_key(digest)
with :ok <- validate_repo(repo),
:ok <- validate_artifact_key(key),
true <- artifact_digest(payload) == digest or {:error, :artifact_digest_mismatch} do
adapter().put_object(repo, key, payload)
end
end
@doc "Fetch an artifact blob by SHA-256 digest (see `put_artifact/3`)."
@spec get_artifact(repo, String.t()) :: {:ok, binary()} | {:error, term}
def get_artifact(repo, digest) when is_binary(digest) do
key = artifact_key(digest)
with :ok <- validate_repo(repo),
:ok <- validate_artifact_key(key) do
adapter().get_entry(repo, key)
end
end
@doc false
def artifact_key(digest), do: "artifacts/" <> digest <> ".tar"
## Pure helpers
@doc """
A fresh, empty index document.
"""
@spec new_index() :: index
def new_index do
%{"version" => 1, "entries" => [], "refs" => %{}}
end
@doc """
Append `entry` to the index and replace the top-level `"refs"` with the
entry's post-state refs.
The entry must carry `"seq"`, `"object"`, `"refs"`, `"principal"`, and
`"pushed_at"`. Raises `ArgumentError` if `"seq"` is not the next sequence
number (`length(entries)`) — an out-of-order append is a caller bug, never
something to write into the log.
The appended entry also gains a `"link"` field: `chain_link/2` over the
previous entry's link and this entry's own contents. Every writer reaches
the log through this function, so the chain covers imports, stack ref
batches, and pushes alike, and a push that retries after a CAS conflict
links against the predecessor it actually lands behind rather than the one
it first read.
"""
@spec append_entry(index, map()) :: index
def append_entry(%{"entries" => entries} = index, %{"seq" => seq} = entry)
when is_list(entries) do
expected = length(entries)
unless seq == expected do
raise ArgumentError,
"WAL entry seq #{inspect(seq)} does not match next seq #{expected}"
end
unless is_map(entry["refs"]) do
raise ArgumentError, "WAL entry must carry a \"refs\" map, got: #{inspect(entry["refs"])}"
end
index
|> Map.put("entries", entries ++ [chain(entries, entry)])
|> Map.put("refs", entry["refs"])
end
@doc """
The chain link an entry carries, or `nil` when it carries none.
"""
@spec entry_link(map()) :: String.t() | nil
def entry_link(%{@chain_field => link}) when is_binary(link), do: link
def entry_link(entry) when is_map(entry), do: nil
@doc """
The link of the last entry in `entries`, or `""` when the list is empty or
its last entry carries no link.
`""` is the chain start. An entry written before this contract existed
carries no link, so the first entry that does carry one binds to `""`
rather than to a predecessor it cannot name.
"""
@spec previous_link([map()]) :: String.t()
def previous_link(entries) when is_list(entries) do
case List.last(entries) do
nil -> ""
entry -> entry_link(entry) || ""
end
end
@doc """
The chain link for `entry` following `previous_link`.
The link is `sha256` over a domain tag, the previous link, and a canonical
encoding of every field of the entry except the link itself, so an entry
commits to its own contents and to the whole prefix of the log before it.
Rewriting one accepted entry therefore invalidates the link of every entry
after it, which is what makes a rewrite non-local. It does not make a
rewrite impossible: an operator who recomputes every link produces a
self-consistent chain. See `INVARIANTS.md`, `EXIT-005`.
Returns `:error` rather than raising, because this runs on the push path
and no push may fail on it.
"""
@spec chain_link(String.t(), map()) :: {:ok, String.t()} | :error
def chain_link(previous_link, entry) when is_binary(previous_link) and is_map(entry) do
payload =
@chain_domain <>
"\n" <> previous_link <> "\n" <> canonical(Map.delete(entry, @chain_field))
{:ok, :sha256 |> :crypto.hash(payload) |> Base.encode16(case: :lower)}
rescue
_uncanonical -> :error
catch
_kind, _reason -> :error
end
def chain_link(_previous_link, _entry), do: :error
# Nothing here may fail a push. A link that cannot be derived is omitted,
# and the entry is written unchained: `OpenAgents.Forge.Verification`
# reports the gap as `chain_link_missing`, which is a thing to find out
# about rather than a reason to refuse a push the forge can accept.
defp chain(entries, entry) do
case chain_link(previous_link(entries), entry) do
{:ok, link} -> Map.put(entry, @chain_field, link)
:error -> entry
end
end
# A deterministic, unambiguous encoding of the JSON values a WAL entry
# holds. The link cannot be taken over encoded JSON, because encoders do not
# agree on key order and this digest has to survive being written by one
# release and recomputed by another. Every value carries its own length or
# terminator, so no two distinct entries encode alike, and total by
# construction — the last clause accepts anything.
#
# Sorting the keys is defensive rather than proven: equal maps iterate
# identically in this runtime, so no test here can distinguish sorted from
# unsorted. Map order is an implementation detail and not a contract, and
# every link ever written depends on this encoding, so the encoding is
# pinned by a golden vector in `test/openagents/forge/wal_test.exs` instead.
defp canonical(nil), do: "n;"
defp canonical(true), do: "b1;"
defp canonical(false), do: "b0;"
defp canonical(value) when is_binary(value),
do: "s" <> Integer.to_string(byte_size(value)) <> ":" <> value
defp canonical(value) when is_integer(value), do: "i" <> Integer.to_string(value) <> ";"
defp canonical(value) when is_float(value), do: "d" <> Float.to_string(value) <> ";"
defp canonical(value) when is_atom(value), do: "a" <> canonical(Atom.to_string(value))
defp canonical(value) when is_list(value), do: "l" <> Enum.map_join(value, &canonical/1) <> "e"
defp canonical(%_struct{} = value), do: "x" <> canonical(inspect(value))
defp canonical(value) when is_map(value) do
encoded =
value
|> Enum.map(fn {key, item} -> {canonical(key), canonical(item)} end)
|> Enum.sort()
|> Enum.map_join(fn {key, item} -> key <> item end)
"m" <> encoded <> "e"
end
defp canonical(value), do: "x" <> canonical(inspect(value))
@doc """
The next sequence number to append to `index`.
"""
@spec next_seq(index) :: non_neg_integer()
def next_seq(%{"entries" => entries}) when is_list(entries), do: length(entries)
@doc """
The current ref map (`ref name => sha`) of `index`.
"""
@spec refs(index) :: map()
def refs(index), do: Map.get(index, "refs", %{})
@doc """
The ordered entry list of `index`.
"""
@spec entries(index) :: [map()]
def entries(index), do: Map.get(index, "entries", [])
@doc """
The canonical object key for the entry at `seq` with `payload`:
`entries/<zero-padded seq>-<first 12 hex chars of sha256(payload)>`.
"""
@spec entry_key(non_neg_integer(), binary()) :: String.t()
def entry_key(seq, payload) when is_integer(seq) and seq >= 0 and is_binary(payload) do
digest =
:crypto.hash(:sha256, payload)
|> Base.encode16(case: :lower)
|> binary_part(0, 12)
"entries/" <> String.pad_leading(Integer.to_string(seq), 8, "0") <> "-" <> digest
end
@doc "Derive the immutable entry key for a file without loading the file into memory."
@spec entry_key_file(non_neg_integer(), String.t()) :: {:ok, String.t()} | {:error, term}
def entry_key_file(seq, path) when is_integer(seq) and seq >= 0 and is_binary(path) do
with {:ok, digest} <- file_digest(path) do
{:ok, entry_key_from_digest(seq, digest)}
end
end
defp file_digest(path) do
try do
digest =
path
|> File.stream!(1_048_576, [])
|> Enum.reduce(:crypto.hash_init(:sha256), &:crypto.hash_update(&2, &1))
|> :crypto.hash_final()
{:ok, digest}
rescue
File.Error -> {:error, :entry_file_unavailable}
end
end
defp entry_key_from_digest(seq, digest) do
prefix =
digest
|> Base.encode16(case: :lower)
|> binary_part(0, 12)
"entries/" <> String.pad_leading(Integer.to_string(seq), 8, "0") <> "-" <> prefix
end
@doc """
Validate a repository name (`[a-z0-9][a-z0-9_-]*`).
"""
@spec validate_repo(term()) :: :ok | {:error, :invalid_repo}
def validate_repo(repo) when is_binary(repo) do
if Regex.match?(@repo_pattern, repo), do: :ok, else: {:error, :invalid_repo}
end
def validate_repo(_repo), do: {:error, :invalid_repo}
defp validate_entry_key(object_key) do
if Regex.match?(@entry_key_pattern, object_key) or
Regex.match?(@artifact_key_pattern, object_key) do
:ok
else
{:error, :invalid_object_key}
end
end
defp validate_artifact_key(object_key) do
if Regex.match?(@artifact_key_pattern, object_key) do
:ok
else
{:error, :invalid_object_key}
end
end
defp artifact_digest(payload) do
:sha256
|> :crypto.hash(payload)
|> Base.encode16(case: :lower)
end
defp adapter do
Application.get_env(:openagents, :forge_wal_adapter, OpenAgents.Forge.WAL.Local)
end
end