OpenAgents

Commit

Export a consenting thread as a WEKA trace

d808cf179a62 · AtlantisPleb · · parent fdab08c09463 

Export a consenting thread as a WEKA trace

The exporter core of issue #218: OpenAgents.Threads.WekaExport turns a
wider-than-dark thread's events into a weka-trace-v1 document — event
metadata, ordering, timestamps, and 64-token block counts preserved;
every text payload replaced with session-salted chained SHA-256 block
hashes, so context growth and prefix-reuse shape survive with no
recoverable content. The same thread and salt reproduce a
byte-identical corpus; a dark thread refuses with consent_required.
No HTTP surface yet; corpus publication stays a separate decision.

Built by a Devin child through the openagents coder's delegate tool;
review added the structured-payload guard in extract_text.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GoYpb8FEmdxVErsv7ABCYi
Co-Authored-By
Claude Fable 5 <noreply@anthropic.com>

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

pushed
by user · WAL seq 334 · 2026-08-25T03:03:11.754165Z

Changed files

  • added lib/openagents/threads/weka_export.ex
  • added test/openagents/threads/weka_export_test.exs

Diff

2 files changed, +291 -0

lib/openagents/threads/weka_export.ex added +149

@@ -0,0 +1,149 @@

1 
defmodule OpenAgents.Threads.WekaExport do
2 
  @moduledoc """
3 
  Exports a ledger-visible thread transcript into a WEKA-trace v1 document.
4
5 
  The output is a content-anonymized, chain-hashed trajectory that preserves
6 
  event metadata (role, type, timestamps, block counts) but replaces every
7 
  raw text payload with deterministic SHA-256 block hashes. Export is
8 
  consent-gated by the thread's visibility tier (THREAD-002).
9 
  """
10
11 
  import Ecto.Query
12
13 
  alias OpenAgents.Repo
14 
  alias OpenAgents.Threads.Event
15 
  alias OpenAgents.Threads.Thread
16
17 
  @weka_format "weka-trace-v1"
18 
  @chunk_size 64
19 
  @hash_display 16
20
21 
  @doc """
22 
  Exports a thread to a WEKA-trace v1 document.
23
24 
  Accepts a `Thread` struct or a thread id (UUID string). The optional `salt`
25 
  is caller-supplied and defaults to `""`. Returns `{:ok, document}` for a
26 
  ledger-visible thread, or `{:error, :consent_required}` /
27 
  `{:error, :thread_not_found}` otherwise.
28 
  """
29 
  @spec export(Thread.t() | String.t(), String.t()) :: {:ok, map()} | {:error, atom()}
30 
  def export(thread_or_id, salt \\ "")
31
32 
  def export(%Thread{} = thread, salt) when is_binary(salt) do
33 
    do_export(thread, salt)
34 
  end
35
36 
  def export(thread_id, salt) when is_binary(thread_id) and is_binary(salt) do
37 
    case Ecto.UUID.cast(thread_id) do
38 
      {:ok, id} ->
39 
        case Repo.get(Thread, id) do
40 
          %Thread{} = thread -> do_export(thread, salt)
41 
          nil -> {:error, :thread_not_found}
42 
        end
43
44 
      :error ->
45 
        {:error, :thread_not_found}
46 
    end
47 
  end
48
49 
  defp do_export(%Thread{} = thread, salt) do
50 
    if Thread.wide?(thread) do
51 
      events = load_events(thread)
52 
      session_salt = derive_session_salt(thread, salt)
53 
      {event_docs, total_blocks} = build_trace(events, session_salt)
54
55 
      document = %{
56 
        "format" => @weka_format,
57 
        "thread_id" => thread.id,
58 
        "generation" => thread.generation,
59 
        "visibility" => thread.visibility,
60 
        "started_at" => format_dt(thread.started_at),
61 
        "completed_at" => format_dt(thread.completed_at),
62 
        "event_count" => length(events),
63 
        "total_blocks" => total_blocks,
64 
        "events" => event_docs
65 
      }
66
67 
      {:ok, document}
68 
    else
69 
      {:error, :consent_required}
70 
    end
71 
  end
72
73 
  defp load_events(%Thread{id: thread_id}) do
74 
    from(e in Event,
75 
      where: e.thread_id == ^thread_id,
76 
      order_by: [asc: e.id]
77 
    )
78 
    |> Repo.all()
79 
  end
80
81 
  defp derive_session_salt(%Thread{id: thread_id}, salt) do
82 
    :crypto.hash(:sha256, "#{thread_id}:#{salt}")
83 
    |> Base.encode16(case: :lower)
84 
  end
85
86 
  defp build_trace(events, session_salt) do
87 
    events
88 
    |> Enum.reduce({[], 0, session_salt}, fn event, {docs, total, prev_hash} ->
89 
      text = extract_text(event.payload)
90 
      tokens = String.split(text, ~r/\s+/, trim: true)
91 
      chunks = Enum.chunk_every(tokens, @chunk_size)
92 
      {blocks, next_hash} = hash_chunks(chunks, session_salt, prev_hash)
93
94 
      event_doc = %{
95 
        "id" => event.id,
96 
        "event_type" => event.event_type,
97 
        "emitted_at" => format_dt(event.emitted_at),
98 
        "role" => extract_role(event),
99 
        "block_count" => length(blocks),
100 
        "blocks" => blocks
101 
      }
102
103 
      {[event_doc | docs], total + length(blocks), next_hash}
104 
    end)
105 
    |> then(fn {docs, total, _hash} -> {Enum.reverse(docs), total} end)
106 
  end
107
108 
  defp hash_chunks(chunks, session_salt, initial_hash) do
109 
    {blocks, final_hash} =
110 
      Enum.reduce(chunks, {[], initial_hash}, fn chunk, {blocks, prev} ->
111 
        chunk_text = Enum.join(chunk, " ")
112
113 
        hash =
114 
          :crypto.hash(:sha256, session_salt <> prev <> chunk_text)
115 
          |> Base.encode16(case: :lower)
116 
          |> String.slice(0, @hash_display)
117
118 
        {[hash | blocks], hash}
119 
      end)
120
121 
    {Enum.reverse(blocks), final_hash}
122 
  end
123
124 
  defp extract_text(payload) when is_map(payload) do
125 
    case payload["content"] || payload["text"] || payload["message"] || payload["output"] do
126 
      value when is_binary(value) -> value
127 
      nil -> Jason.encode!(payload)
128 
      value -> Jason.encode!(value)
129 
    end
130 
  end
131
132 
  defp extract_text(payload) when is_binary(payload), do: payload
133 
  defp extract_text(_payload), do: ""
134
135 
  defp extract_role(%Event{event_type: type, payload: payload}) do
136 
    case payload do
137 
      %{"role" => role} when is_binary(role) -> role
138 
      _ -> role_from_type(type)
139 
    end
140 
  end
141
142 
  defp role_from_type("turn." <> rest), do: rest
143 
  defp role_from_type("tool." <> _), do: "tool"
144 
  defp role_from_type("thread." <> _), do: "system"
145 
  defp role_from_type(_), do: "unknown"
146
147 
  defp format_dt(%DateTime{} = dt), do: DateTime.to_iso8601(dt)
148 
  defp format_dt(nil), do: nil
149 
end
test/openagents/threads/weka_export_test.exs added +142

@@ -0,0 +1,142 @@

1 
defmodule OpenAgents.Threads.WekaExportTest do
2 
  @moduledoc """
3 
  Tests for `OpenAgents.Threads.WekaExport`.
4 
  """
5
6 
  use OpenAgents.DataCase, async: true
7
8 
  import OpenAgentsWeb.ConnCase, only: [github_user: 1]
9
10 
  alias OpenAgents.Repo
11 
  alias OpenAgents.Threads
12 
  alias OpenAgents.Threads.Event
13 
  alias OpenAgents.Threads.WekaExport
14
15 
  describe "export/2" do
16 
    test "a ledger-visible thread exports a deterministic, content-free document" do
17 
      user = github_user("weka-export-consent")
18 
      {:ok, thread} = Threads.open(user, "Test objective", visibility: "ledger")
19
20 
      fixed_dt = DateTime.from_naive!(~N[2026-08-24 12:00:00.000000], "Etc/UTC")
21 
      thread = thread |> change(started_at: fixed_dt) |> Repo.update!()
22
23 
      insert_event(thread, "turn.user", %{"content" => "secret user prompt"}, fixed_dt)
24
25 
      insert_event(
26 
        thread,
27 
        "turn.assistant",
28 
        %{"output" => long_text_with("confidential code")},
29 
        fixed_dt
30 
      )
31
32 
      assert {:ok, doc} = WekaExport.export(thread, "fixed-salt")
33
34 
      assert doc["format"] == "weka-trace-v1"
35 
      assert doc["thread_id"] == thread.id
36 
      assert doc["generation"] == thread.generation
37 
      assert doc["visibility"] == "ledger"
38 
      assert doc["started_at"] == DateTime.to_iso8601(fixed_dt)
39 
      assert doc["completed_at"] == nil
40 
      assert doc["event_count"] == 4
41
42 
      json = Jason.encode!(doc)
43 
      refute String.contains?(json, "secret user prompt")
44 
      refute String.contains?(json, "confidential code")
45
46 
      for event <- doc["events"] do
47 
        assert is_binary(event["emitted_at"])
48 
        assert is_integer(event["block_count"])
49 
        assert is_list(event["blocks"])
50 
        assert length(event["blocks"]) == event["block_count"]
51 
      end
52
53 
      user_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.user"))
54 
      assistant_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.assistant"))
55
56 
      assert user_event["role"] == "user"
57 
      assert user_event["block_count"] == 1
58
59 
      assert user_event["emitted_at"] ==
60 
               DateTime.to_iso8601(DateTime.truncate(fixed_dt, :microsecond))
61
62 
      assert assistant_event["role"] == "assistant"
63 
      assert assistant_event["block_count"] == 3
64
65 
      total = Enum.reduce(doc["events"], 0, &(&2 + &1["block_count"]))
66 
      assert doc["total_blocks"] == total
67 
    end
68
69 
    test "a dark thread refuses export" do
70 
      user = github_user("weka-export-dark")
71 
      {:ok, thread} = Threads.open(user, "Dark work")
72
73 
      assert WekaExport.export(thread) == {:error, :consent_required}
74 
      assert WekaExport.export(thread.id) == {:error, :consent_required}
75 
    end
76
77 
    test "an unknown or invalid thread id refuses export" do
78 
      assert WekaExport.export(Ecto.UUID.generate()) == {:error, :thread_not_found}
79 
      assert WekaExport.export("not-a-uuid") == {:error, :thread_not_found}
80 
    end
81
82 
    test "the same thread and salt export to a byte-identical document" do
83 
      user = github_user("weka-export-repro")
84 
      {:ok, thread} = Threads.open(user, "Repro", visibility: "ledger")
85 
      insert_event(thread, "turn.user", %{"content" => "hello"}, DateTime.utc_now())
86
87 
      salt = "same-salt"
88 
      assert WekaExport.export(thread, salt) == WekaExport.export(thread, salt)
89 
    end
90
91 
    test "different salts produce different block hashes" do
92 
      user = github_user("weka-export-salt")
93 
      {:ok, thread} = Threads.open(user, "Salt", visibility: "ledger")
94 
      insert_event(thread, "turn.user", %{"content" => "salted"}, DateTime.utc_now())
95
96 
      {:ok, doc1} = WekaExport.export(thread, "salt-a")
97 
      {:ok, doc2} = WekaExport.export(thread, "salt-b")
98
99 
      blocks1 = doc1 |> Map.get("events") |> Enum.flat_map(& &1["blocks"])
100 
      blocks2 = doc2 |> Map.get("events") |> Enum.flat_map(& &1["blocks"])
101
102 
      refute blocks1 == blocks2
103 
    end
104
105 
    test "multi-turn context grows and block hashes chain" do
106 
      user = github_user("weka-export-chain")
107 
      {:ok, thread} = Threads.open(user, "Chain", visibility: "ledger")
108
109 
      fixed_dt = DateTime.from_naive!(~N[2026-08-24 12:00:00.000000], "Etc/UTC")
110 
      insert_event(thread, "turn.user", %{"content" => "word"}, fixed_dt)
111 
      insert_event(thread, "turn.assistant", %{"output" => long_text_with("end")}, fixed_dt)
112
113 
      assert {:ok, doc} = WekaExport.export(thread, "chain-salt")
114
115 
      user_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.user"))
116 
      assistant_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.assistant"))
117
118 
      assert user_event["block_count"] == 1
119 
      assert assistant_event["block_count"] == 3
120
121 
      all_blocks = doc["events"] |> Enum.flat_map(& &1["blocks"])
122 
      assert length(all_blocks) == length(Enum.uniq(all_blocks))
123 
      assert doc["total_blocks"] == length(all_blocks)
124 
    end
125 
  end
126
127 
  defp insert_event(thread, event_type, payload, emitted_at) do
128 
    %Event{}
129 
    |> Event.changeset(%{
130 
      thread_id: thread.id,
131 
      event_type: event_type,
132 
      payload: payload,
133 
      emitted_at: emitted_at
134 
    })
135 
    |> Repo.insert!()
136 
  end
137
138 
  defp long_text_with(suffix) do
139 
    words = List.duplicate("word", 130) ++ [suffix]
140 
    Enum.join(words, " ")
141 
  end
142 
end

This page updates live while a promote is in flight · changelog