it; a counter that read `thread_events` without that join would republish, in

aggregate, transcripts their owners kept `dark`.

Amended 2026-08-25 (issue #218): the tier now gates a second consumer, the

benchmark workbench's trace corpus. `OpenAgents.Threads.WekaExport` turns a

consenting thread's transcript into a WEKA v1 trace — the block-hash format

AgentX replays (`docs/2026-08-24-benchmark-workbench-agentx.md`, section 5) —

and refuses a `dark` thread with `:consent_required`. The gate is applied per

thread rather than per subtree, because a consenting parent may spawn a

narrower child (THREAD-003): a `dark` child produces no sub-agent entry, no

request, no block count, and no record that it existed. `corpus/2` takes a

recorded thread-id set rather than running a query — a query is how a consent

gate gets widened by accident — and records every refusal by id and reason

beside the code revision that built the document, so a corpus is reproducible

and honest about what it does not contain. `mix openagents.weka.export` is the

only surface that writes one; no route publishes a corpus, and publication

stays a separate, explicit decision. What leaves is structure and timing:

prompts are cut into 64-token blocks replaced by session-salted chained

hashes, remapped to session-local integers, so prefix reuse survives and

content does not.

Evidence: `OpenAgents.Threads.fetch_readable/2`,

`OpenAgents.Threads.Thread.visibilities/0`, `OpenAgentsWeb.ThreadController`,

`OpenAgentsWeb.ThreadShowLive`, `OpenAgents.DataRights.AccountExport`,

`OpenAgents.Threads.WekaExport`,

`priv/repo/migrations/20260824210500_add_visibility_to_threads.exs`,

`test/openagents/threads/visibility_test.exs`,

`test/openagents_web/thread_visibility_test.exs`, and

`test/openagents_web/thread_visibility_test.exs`,

`test/openagents/threads/weka_export_test.exs`,

`test/mix/tasks/openagents_weka_export_test.exs`, and

`test/openagents/threads/grant_token_reach_test.exs`.

### THREAD-003 — Child threads are nested, budgeted, and typed

  users, and — if published — the corpus is a contribution back to the

  AgentX ecosystem under our own consent rules.

**Shipped (issue #218).** `OpenAgents.Threads.WekaExport` is the exporter.

`export/2` turns one consenting thread into a WEKA v1 trace: an ordered list

of model calls, each carrying its wall-clock offset, call duration, think

time, input and output token counts, and the ids of the 64-token blocks its

prompt was made of, under `block_size: 64` and `hash_id_scope: "local"` — the

shape `proxy_to_weka.py` writes and AIPerf replays. A call closes at each

`tool.*` and each `turn.assistant`; its prompt is everything recorded before

the model-authored run began. Blocks are session-salted chained SHA-256 —

chained because a repeated block at a different context position is not

cache-equivalent, and only chaining tells the two apart — then remapped to

session-local integers. Only whole blocks are hashed, so `in` is exactly what

a replay would send. Consenting child threads become `subagent` entries with

their own fresh context in the shared id space; a `dark` child is absent

entirely.

`corpus/2` builds a corpus from a recorded thread-id set, never from a query,

and records the set, the refusals with their reasons, and the code revision,

so the same set and revision rebuild the same document. `prefix_reuse/1`

reports the measurement the axis-1 comparison needs — reused leading blocks

over all blocks, per agent — and

`test/openagents/threads/weka_export_test.exs` holds that number equal to the

same statistic computed on the raw transcript, so the anonymized trace carries

the source session's prefix-reuse characteristics rather than merely claiming

to. `mix openagents.weka.export --threads SET_FILE --out CORPUS_FILE` is the

operator surface. No route publishes a corpus; publication stays a separate

decision, as this section always said.

## 6. Non-goals

- No 2MW GPU fleet, no hardware SKU comparisons — that is InferenceX's

defmodule Mix.Tasks.Openagents.Weka.Export do

  @shortdoc "Export consenting thread transcripts as a WEKA trace corpus"

  @moduledoc """

  Builds the WEKA v1 trace corpus the benchmark workbench replays

  (`docs/2026-08-24-benchmark-workbench-agentx.md`, section 5; issue #218).

  The thread-id set is an input, not a query. Nothing here selects threads on

  the operator's behalf, because a corpus is only reproducible if the set that

  built it was recorded, and because a query is how a consent gate gets widened

  by accident. Name the ids, or name a file of them:

      mix openagents.weka.export --threads corpus-set.txt --out corpus.json

      mix openagents.weka.export --thread THREAD_ID --thread THREAD_ID \\

        --out /tmp/corpus.json --salt CORPUS_SALT

  A thread whose owner has not widened it is refused, recorded in the corpus by

  id and reason, and contributes nothing. The written document carries the

  requested set and the code revision that produced it, so the same file can be

  rebuilt and compared byte for byte.

  Publication is a separate, explicit decision. This task writes a file.

  """

  use Mix.Task

  alias OpenAgents.Threads.WekaExport

  @requirements ["app.config"]

  @switches [

    thread: :keep,

    threads: :string,

    out: :string,

    salt: :string,

    revision: :string,

    database_url: :string

  ]

  @impl Mix.Task

  def run(arguments) do

    {options, remaining, invalid} = OptionParser.parse(arguments, strict: @switches)

    if remaining != [] or invalid != [], do: Mix.raise("invalid WEKA export options")

    thread_ids = thread_ids(options)

    if thread_ids == [], do: Mix.raise("--thread THREAD_ID or --threads FILE is required")

    ensure_repo(Keyword.get(options, :database_url))

    build_options =

      options

      |> Keyword.take([:salt, :revision])

      |> Keyword.new()

    {:ok, corpus} = WekaExport.corpus(thread_ids, build_options)

    output = Keyword.get(options, :out, "openagents-weka-corpus.json")

    File.write!(output, Jason.encode!(corpus, pretty: true))

    reuse = corpus["prefix_reuse"]

    Mix.shell().info(

      "wrote #{output}: #{length(corpus["traces"])} trace(s), " <>

        "#{length(corpus["refused"])} refused, #{reuse["requests"]} request(s), " <>

        "#{reuse["blocks"]} block(s), prefix reuse " <>

        "#{Float.round(reuse["rate"] * 100, 1)}%, revision #{corpus["code_revision"]}"

    )

    for refusal <- corpus["refused"] do

      Mix.shell().info("  refused #{refusal["thread_id"]}: #{refusal["reason"]}")

    end

  end

  defp thread_ids(options) do

    named = for {:thread, id} <- options, do: String.trim(id)

    from_file =

      case Keyword.get(options, :threads) do

        nil ->

          []

        path ->

          path

          |> File.read!()

          |> String.split(~r/\r?\n/, trim: true)

          |> Enum.map(&String.trim/1)

          |> Enum.reject(&(&1 == "" or String.starts_with?(&1, "#")))

      end

    Enum.uniq(from_file ++ named)

  end

  # Inside the application — a test, a release console — the repository is

  # already up and starting a second one would take a connection outside the

  # caller's ownership. Outside it, start only the repository: this task has no

  # use for the supervision tree's recovery workers.

  defp ensure_repo(database_url) do

    if Process.whereis(OpenAgents.Repo) do

      :ok

    else

      start_repo_only(database_url)

    end

  end

  defp start_repo_only(database_url) do

    {:ok, _apps} = Application.ensure_all_started(:ssl)

    {:ok, _apps} = Application.ensure_all_started(:postgrex)

    {:ok, _apps} = Application.ensure_all_started(:ecto_sql)

    repo_options =

      case database_url do

        nil -> [pool_size: 2]

        url when is_binary(url) -> [url: url, pool_size: 2, ssl: false]

      end

    case OpenAgents.Repo.start_link(repo_options) do

      {:ok, _repo} -> :ok

      {:error, {:already_started, _repo}} -> :ok

      {:error, reason} -> Mix.raise("repo start failed: #{inspect(reason)}")

    end

  end

end

defmodule OpenAgents.Threads.WekaExport do

  @moduledoc """

  Exports a ledger-visible thread transcript into a WEKA-trace v1 document.

  Exports consenting thread transcripts as WEKA v1 traces, and a recorded set

  of them as a corpus (issue #218).

  The output is a content-anonymized, chain-hashed trajectory that preserves

  event metadata (role, type, timestamps, block counts) but replaces every

  raw text payload with deterministic SHA-256 block hashes. Export is

  consent-gated by the thread's visibility tier (THREAD-002).

  WEKA is the trace format AgentX replays under AIPerf

  (`docs/2026-08-24-benchmark-workbench-agentx.md`, section 5). A trace is a

  session: an ordered list of model calls, each carrying the wall-clock offset

  it happened at, how long the call took, how many tokens went in and came

  out, and the identity of the 64-token blocks its prompt was made of. Content

  never appears. What survives is the shape of the traffic — multi-turn

  accumulation, context growth, prefix reuse, and sub-agent bursts — which is

  the thing a serving stack is measured against.

  ## What a request is

  The transcript vocabulary is `turn.user`, `turn.reasoning`, `tool.ran`,

  `turn.assistant` (`docs/2026-08-24-coder-account-integration-audit.md`). One

  model call produces a run of model-authored events that ends either at a tool

  call or at an answer, so a request closes at each `tool.*` and each

  `turn.assistant`. Its prompt is everything recorded before that run began;

  its output is the run. Anything else — the user's turn, the thread's own

  lifecycle records — accumulates into the context the next call carries.

  Because a call's prompt is a prefix of the next call's prompt, the block ids

  of one request are a prefix of the next request's. That is not asserted here;

  it falls out of how contexts grow, which is exactly why the measurement in

  `prefix_reuse/1` says something about the source session.

  ## How blocks are anonymized

  A prompt is cut into 64-token blocks and each block is replaced by a

  session-scoped chained hash: `sha256(previous_hash <> block_text)`, seeded

  from a salt derived from the thread id and the caller's salt. The chain is

  load-bearing rather than decorative. A block of text that appears twice in

  one context at different positions is not cache-equivalent — only a matching

  *prefix* hits a warm KV cache — and chaining is what tells the two apart.

  The salt makes block identity session-scoped, so nothing can be matched

  across sessions or attacked with a dictionary of likely blocks.

  The emitted ids are then remapped to session-local integers in first-seen

  order, `hash_id_scope: "local"`, which is what the reference converter

  publishes and what the replayer consumes. Only whole blocks are hashed: the

  trailing partial block is dropped and `in` counts the tokens actually

  covered, so a replayed prompt is exactly what was hashed and nothing more. A

  prompt with no whole block reports its true token count instead, the same

  fallback the reference converter uses when no hash coverage exists.

  Tokens here are whitespace-separated words, not a model's byte-pair tokens.

  The measure that matters is block *structure* — which blocks repeat, and

  where — and that is preserved under any consistent tokenizer.

  ## Consent

  Export is gated by the thread's visibility tier (THREAD-002). A `dark`

  thread refuses with `:consent_required` and contributes nothing: not a

  request, not a block count, not a token. The gate applies to sub-agents

  independently, because a consenting parent may spawn a narrower child

  (THREAD-003) — a `dark` child produces no sub-agent entry and no trace of

  having existed. `corpus/2` records every refusal by thread id and reason, so

  a corpus is honest about what it does not contain.

  Publication of a corpus is a separate, explicit decision. This module builds

  documents; nothing here publishes one.

  """

  import Ecto.Query

  alias OpenAgents.BuildInfo

  alias OpenAgents.Repo

  alias OpenAgents.Threads.Event

  alias OpenAgents.Threads.Thread

  @weka_format "weka-trace-v1"

  @chunk_size 64

  @hash_display 16

  @block_size 64

  @hash_id_scope "local"

  @corpus_format "openagents-weka-corpus-v1"

  @subagent_type "child_thread"

  @typedoc "A WEKA v1 trace document."

  @type trace :: %{String.t() => term()}

  @typedoc "A corpus of WEKA v1 traces with the set and revision that built it."

  @type corpus :: %{String.t() => term()}

  @doc "The KV-cache block size the WEKA format is defined against."

  @spec block_size() :: pos_integer()

  def block_size, do: @block_size

  @doc """

  Exports a thread to a WEKA-trace v1 document.

  Exports one consenting thread as a WEKA v1 trace.

  Accepts a `Thread` struct or a thread id. Options:

    * `:salt` — caller-supplied salt folded into the session's block identity.

      Defaults to `""`. It changes nothing an exported document shows, because

      ids are remapped session-locally; it changes what the hashes behind them

      are, which is what keeps block identity from being guessable.

  Accepts a `Thread` struct or a thread id (UUID string). The optional `salt`

  is caller-supplied and defaults to `""`. Returns `{:ok, document}` for a

  ledger-visible thread, or `{:error, :consent_required}` /

  `{:error, :thread_not_found}` otherwise.

  Returns `{:error, :consent_required}` for a thread its owner left `dark`,

  and `{:error, :thread_not_found}` for an id that resolves to nothing.

  """

  @spec export(Thread.t() | String.t(), String.t()) :: {:ok, map()} | {:error, atom()}

  def export(thread_or_id, salt \\ "")

  @spec export(Thread.t() | String.t(), keyword()) ::

          {:ok, trace()} | {:error, :consent_required | :thread_not_found}

  def export(thread_or_id, options \\ [])

  def export(%Thread{} = thread, salt) when is_binary(salt) do

    do_export(thread, salt)

  def export(%Thread{} = thread, options) when is_list(options), do: gate(thread, options)

  def export(thread_id, options) when is_binary(thread_id) and is_list(options) do

    with {:ok, id} <- Ecto.UUID.cast(thread_id),

         %Thread{} = thread <- Repo.get(Thread, id) do

      gate(thread, options)

    else

      _unresolved -> {:error, :thread_not_found}

    end

  end

  def export(thread_id, salt) when is_binary(thread_id) and is_binary(salt) do

    case Ecto.UUID.cast(thread_id) do

      {:ok, id} ->

        case Repo.get(Thread, id) do

          %Thread{} = thread -> do_export(thread, salt)

          nil -> {:error, :thread_not_found}

  @doc """

  Builds a corpus from a recorded set of thread ids.

  The document records the ids it was asked for, the ids it included, every

  refusal with its reason, and the code revision that built it, so the same set

  and the same revision rebuild the same corpus. Options are `:salt` (see

  `export/2`) and `:revision`, which defaults to this build's own.

  """

  @spec corpus([String.t()], keyword()) :: {:ok, corpus()}

  def corpus(thread_ids, options \\ []) when is_list(thread_ids) do

    requested = Enum.map(thread_ids, &to_string/1)

    salt = Keyword.get(options, :salt, "")

    revision = Keyword.get(options, :revision, build_revision())

    {traces, refused} =

      Enum.reduce(requested, {[], []}, fn thread_id, {traces, refused} ->

        case export(thread_id, salt: salt) do

          {:ok, trace} ->

            {[trace | traces], refused}

          {:error, reason} ->

            {traces, [%{"thread_id" => thread_id, "reason" => to_string(reason)} | refused]}

        end

      end)

      :error ->

        {:error, :thread_not_found}

    end

    traces = Enum.reverse(traces)

    {:ok,

     %{

       "format" => @corpus_format,

       "block_size" => @block_size,

       "hash_id_scope" => @hash_id_scope,

       "code_revision" => revision,

       "salt_digest" => digest(salt),

       "requested_thread_ids" => requested,

       "included_thread_ids" => Enum.map(traces, & &1["id"]),

       "refused" => Enum.reverse(refused),

       "traces" => traces,

       "prefix_reuse" => prefix_reuse(%{"traces" => traces})

     }}

  end

  @doc """

  Measures prefix reuse over an exported trace or corpus.

  For each agent — the main thread and each sub-agent entry — every call after

  the agent's first contributes the number of leading blocks it shares with the

  call before it. `rate` is those reused blocks over all blocks, which is the

  KV-cache hit rate a replay of this traffic would see with a perfect cache.

  The number is computed from the anonymized block ids, and it is the same

  number the raw transcript gives, because the ids carry the source's block

  structure. `test/openagents/threads/weka_export_test.exs` computes both and

  holds them equal.

  """

  @spec prefix_reuse(trace() | corpus()) :: %{String.t() => number()}

  def prefix_reuse(%{"traces" => traces}) when is_list(traces) do

    traces

    |> Enum.flat_map(&agent_sequences/1)

    |> tally()

  end

  def prefix_reuse(%{"requests" => _requests} = trace) do

    trace |> agent_sequences() |> tally()

  end

  defp do_export(%Thread{} = thread, salt) do

  # ── the consent gate ───────────────────────────────────────────────────────

  defp gate(%Thread{} = thread, options) do

    if Thread.wide?(thread) do

      events = load_events(thread)

      session_salt = derive_session_salt(thread, salt)

      {event_docs, total_blocks} = build_trace(events, session_salt)

      document = %{

        "format" => @weka_format,

        "thread_id" => thread.id,

        "generation" => thread.generation,

        "visibility" => thread.visibility,

        "started_at" => format_dt(thread.started_at),

        "completed_at" => format_dt(thread.completed_at),

        "event_count" => length(events),

        "total_blocks" => total_blocks,

        "events" => event_docs

      {:ok, build(thread, options)}

    else

      {:error, :consent_required}

    end

  end

  # ── the trace ──────────────────────────────────────────────────────────────

  defp build(%Thread{} = thread, options) do

    salt = Keyword.get(options, :salt, "")

    hashes = %{salt: session_salt(thread, salt), assigned: %{}, next: 0}

    events = load_events(thread.id)

    origin = origin(thread, events)

    {entries, _hashes} =

      walk(events,

        hashes: hashes,

        model: thread.model,

        children: load_consenting_children(thread.id),

        origin: origin,

        call_start: origin

      )

    %{

      "id" => thread.id,

      "models" => models(thread, entries),

      "block_size" => @block_size,

      "hash_id_scope" => @hash_id_scope,

      "requests" => entries

    }

  end

  defp models(%Thread{model: model}, entries) do

    entries

    |> Enum.flat_map(&Map.get(&1, "models", []))

    |> List.insert_at(0, model)

    |> Enum.reject(&is_nil/1)

    |> Enum.uniq()

    |> Enum.sort()

  end

  # One agent's events become one list of entries. The parent walks with the

  # spawn children it may expand; a sub-agent walks with none, because the WEKA

  # format nests one level and a grandchild's spawn is only context.

  defp walk(events, options) do

    accumulator = %{

      context: new_context(Keyword.fetch!(options, :hashes).salt),

      hashes: Keyword.fetch!(options, :hashes),

      model: Keyword.fetch!(options, :model),

      children: Keyword.get(options, :children, %{}),

      origin: Keyword.fetch!(options, :origin),

      call_start: Keyword.fetch!(options, :call_start),

      previous_end: nil,

      entries: [],

      run: []

    }

    final = Enum.reduce(events, accumulator, &step/2)

    {Enum.reverse(final.entries), final.hashes}

  end

  defp step(%Event{} = event, accumulator) do

    tokens = tokenize(extract_text(event.payload))

    cond do

      terminating?(event.event_type) -> close_request(accumulator, event, tokens)

      model_authored?(event.event_type) -> %{accumulator | run: accumulator.run ++ tokens}

      true -> absorb_input(accumulator, event, tokens)

    end

  end

  defp absorb_input(accumulator, %Event{} = event, tokens) do

    {context, hashes} =

      absorb(accumulator.context, tokens, accumulator.hashes)

    accumulator

    |> maybe_subagent(event)

    |> Map.merge(%{context: context, hashes: hashes, call_start: event.emitted_at})

  end

  defp close_request(accumulator, %Event{} = event, tokens) do

    run = accumulator.run ++ tokens

    {ids, input_tokens} = prompt(accumulator.context)

    call_start = accumulator.call_start

    request =

      %{

        "t" => elapsed(accumulator.origin, call_start),

        "type" => "n",

        "model" => accumulator.model,

        "in" => input_tokens,

        "out" => length(run),

        "hash_ids" => ids,

        "api_time" => elapsed(call_start, event.emitted_at)

      }

      |> put_think_time(accumulator.previous_end, call_start)

      {:ok, document}

    {context, hashes} = absorb(accumulator.context, run, accumulator.hashes)

    %{

      accumulator

      | entries: [request | accumulator.entries],

        context: context,

        hashes: hashes,

        run: [],

        call_start: event.emitted_at,

        previous_end: event.emitted_at

    }

  end

  defp put_think_time(request, nil, _call_start), do: request

  defp put_think_time(request, previous_end, call_start) do

    Map.put(request, "think_time", elapsed(previous_end, call_start))

  end

  # ── sub-agents ─────────────────────────────────────────────────────────────

  defp maybe_subagent(accumulator, %Event{event_type: "thread.spawn", payload: payload}) do

    with child_id when is_binary(child_id) <- spawned_id(payload),

         %{thread: child, events: events} <- Map.get(accumulator.children, child_id) do

      {inner, hashes} =

        walk(events,

          hashes: accumulator.hashes,

          model: child.model,

          origin: accumulator.origin,

          call_start: child.started_at || accumulator.call_start

        )

      case inner do

        [] -> %{accumulator | hashes: hashes}

        requests -> add_subagent(accumulator, hashes, child, events, requests)

      end

    else

      {:error, :consent_required}

      _absent -> accumulator

    end

  end

  defp load_events(%Thread{id: thread_id}) do

    from(e in Event,

      where: e.thread_id == ^thread_id,

      order_by: [asc: e.id]

  defp maybe_subagent(accumulator, %Event{}), do: accumulator

  defp add_subagent(accumulator, hashes, %Thread{} = child, events, requests) do

    first = List.first(requests)

    last = List.last(requests)

    finished = last["t"] + last["api_time"]

    entry = %{

      "t" => first["t"],

      "type" => "subagent",

      "agent_id" => "thread_" <> child.id,

      "subagent_type" => @subagent_type,

      "duration_ms" => round(max(finished - first["t"], 0.0) * 1000),

      "total_tokens" => Enum.reduce(requests, 0, &(&2 + &1["in"] + &1["out"])),

      "tool_use_count" => Enum.count(events, &String.starts_with?(&1.event_type, "tool.")),

      "status" => subagent_status(child.status),

      "requests" => requests,

      "models" => child.model |> List.wrap() |> Enum.sort()

    }

    %{accumulator | entries: [entry | accumulator.entries], hashes: hashes}

  end

  # The reference converter writes "completed" because its source has no other

  # word. Ours does: a thread that is still open did not complete, and saying

  # so costs nothing.

  defp subagent_status("succeeded"), do: "completed"

  defp subagent_status("open"), do: "running"

  defp subagent_status(status) when is_binary(status), do: status

  defp subagent_status(_status), do: "running"

  defp spawned_id(payload) when is_map(payload) do

    case payload["child_thread_id"] do

      id when is_binary(id) -> id

      _other -> nil

    end

  end

  defp spawned_id(_payload), do: nil

  # ── the block chain ────────────────────────────────────────────────────────

  defp new_context(salt), do: %{ids: [], chain: salt, tail: [], seen: 0}

  defp absorb(context, tokens, hashes) do

    commit(

      %{context | tail: context.tail ++ tokens, seen: context.seen + length(tokens)},

      hashes

    )

    |> Repo.all()

  end

  defp derive_session_salt(%Thread{id: thread_id}, salt) do

    :crypto.hash(:sha256, "#{thread_id}:#{salt}")

    |> Base.encode16(case: :lower)

  defp commit(%{tail: tail} = context, hashes) when length(tail) >= @block_size do

    {block, rest} = Enum.split(tail, @block_size)

    chain = :crypto.hash(:sha256, context.chain <> "\n" <> Enum.join(block, " "))

    {id, hashes} = intern(hashes, Base.encode16(chain, case: :lower))

    commit(%{context | tail: rest, chain: chain, ids: [id | context.ids]}, hashes)

  end

  defp build_trace(events, session_salt) do

    events

    |> Enum.reduce({[], 0, session_salt}, fn event, {docs, total, prev_hash} ->

      text = extract_text(event.payload)

      tokens = String.split(text, ~r/\s+/, trim: true)

      chunks = Enum.chunk_every(tokens, @chunk_size)

      {blocks, next_hash} = hash_chunks(chunks, session_salt, prev_hash)

  defp commit(context, hashes), do: {context, hashes}

      event_doc = %{

        "id" => event.id,

        "event_type" => event.event_type,

        "emitted_at" => format_dt(event.emitted_at),

        "role" => extract_role(event),

        "block_count" => length(blocks),

        "blocks" => blocks

      }

  defp intern(%{assigned: assigned, next: next} = hashes, hash) do

    case Map.fetch(assigned, hash) do

      {:ok, id} -> {id, hashes}

      :error -> {next, %{hashes | assigned: Map.put(assigned, hash, next), next: next + 1}}

    end

  end

      {[event_doc | docs], total + length(blocks), next_hash}

    end)

    |> then(fn {docs, total, _hash} -> {Enum.reverse(docs), total} end)

  # Whole blocks only, the way the reference converter emits them: the replayed

  # prompt is exactly what was hashed. A prompt with no whole block has no hash

  # coverage at all, so it reports its true token count instead of zero.

  defp prompt(%{ids: []} = context), do: {[], context.seen}

  defp prompt(context) do

    ids = Enum.reverse(context.ids)

    {ids, length(ids) * @block_size}

  end

  defp session_salt(%Thread{id: thread_id}, salt) do

    :crypto.hash(:sha256, thread_id <> "\n" <> salt)

  end

  defp digest(value) do

    :crypto.hash(:sha256, value) |> Base.encode16(case: :lower)

  end

  defp hash_chunks(chunks, session_salt, initial_hash) do

    {blocks, final_hash} =

      Enum.reduce(chunks, {[], initial_hash}, fn chunk, {blocks, prev} ->

        chunk_text = Enum.join(chunk, " ")

  # ── the measurement ────────────────────────────────────────────────────────

  defp agent_sequences(%{"requests" => requests}) do

    main = requests |> Enum.reject(&(&1["type"] == "subagent")) |> Enum.map(& &1["hash_ids"])

        hash =

          :crypto.hash(:sha256, session_salt <> prev <> chunk_text)

          |> Base.encode16(case: :lower)

          |> String.slice(0, @hash_display)

    inner =

      requests

      |> Enum.filter(&(&1["type"] == "subagent"))

      |> Enum.map(fn entry -> Enum.map(entry["requests"], & &1["hash_ids"]) end)

    Enum.reject([main | inner], &(&1 == []))

  end

        {[hash | blocks], hash}

  defp tally(sequences) do

    {requests, blocks, reused} =

      Enum.reduce(sequences, {0, 0, 0}, fn sequence, {requests, blocks, reused} ->

        {sequence_blocks, sequence_reused} = sequence_reuse(sequence)

        {requests + length(sequence), blocks + sequence_blocks, reused + sequence_reused}

      end)

    {Enum.reverse(blocks), final_hash}

    %{

      "requests" => requests,

      "blocks" => blocks,

      "reused_blocks" => reused,

      "rate" => if(blocks == 0, do: 0.0, else: reused / blocks)

    }

  end

  defp sequence_reuse(sequence) do

    sequence

    |> Enum.reduce({0, 0, nil}, fn ids, {blocks, reused, previous} ->

      shared = if previous, do: common_prefix(previous, ids), else: 0

      {blocks + length(ids), reused + shared, ids}

    end)

    |> then(fn {blocks, reused, _previous} -> {blocks, reused} end)

  end

  defp common_prefix(earlier, later) do

    earlier |> Enum.zip(later) |> Enum.take_while(fn {a, b} -> a == b end) |> length()

  end

  # ── the transcript ─────────────────────────────────────────────────────────

  defp load_events(thread_id) do

    Repo.all(

      from(event in Event, where: event.thread_id == ^thread_id, order_by: [asc: event.id])

    )

  end

  defp load_consenting_children(thread_id) do

    wide = Thread.wide_visibilities()

    children =

      Repo.all(

        from(thread in Thread,

          where: thread.parent_thread_id == ^thread_id and thread.visibility in ^wide,

          order_by: [asc: thread.inserted_at]

        )

      )

    grouped =

      children

      |> Enum.map(& &1.id)

      |> child_events()

      |> Enum.group_by(& &1.thread_id)

    Map.new(children, fn child ->

      {child.id, %{thread: child, events: Map.get(grouped, child.id, [])}}

    end)

  end

  defp child_events([]), do: []

  defp child_events(ids) do

    Repo.all(from(event in Event, where: event.thread_id in ^ids, order_by: [asc: event.id]))

  end

  defp origin(%Thread{started_at: %DateTime{} = started_at}, _events), do: started_at

  defp origin(%Thread{}, [%Event{emitted_at: emitted_at} | _rest]), do: emitted_at

  defp origin(%Thread{inserted_at: inserted_at}, []), do: inserted_at

  defp elapsed(from, to) do

    max(DateTime.diff(to, from, :microsecond) / 1_000_000, 0.0)

  end

  defp tokenize(text), do: String.split(text, ~r/\s+/, trim: true)

  # A payload is a JSON object carrying whatever happened. Where it names its

  # text, that is the text; where it does not, the encoded object stands in, so

  # a structured record still contributes its shape rather than being skipped.

  # Either way what leaves is a hash of it.

  defp extract_text(payload) when is_map(payload) do

    case payload["content"] || payload["text"] || payload["message"] || payload["output"] do

      value when is_binary(value) -> value

  defp extract_text(payload) when is_binary(payload), do: payload

  defp extract_text(_payload), do: ""

  defp extract_role(%Event{event_type: type, payload: payload}) do

    case payload do

      %{"role" => role} when is_binary(role) -> role

      _ -> role_from_type(type)

    end

  end

  # One model call ends when the model either asks for a tool or answers.

  defp terminating?("turn.assistant"), do: true

  defp terminating?("tool." <> _rest), do: true

  defp terminating?(_type), do: false

  defp role_from_type("turn." <> rest), do: rest

  defp role_from_type("tool." <> _), do: "tool"

  defp role_from_type("thread." <> _), do: "system"

  defp role_from_type(_), do: "unknown"

  defp model_authored?("turn.reasoning"), do: true

  defp model_authored?(type), do: terminating?(type)

  defp format_dt(%DateTime{} = dt), do: DateTime.to_iso8601(dt)

  defp format_dt(nil), do: nil

  defp build_revision do

    version = to_string(Application.spec(:openagents, :vsn) || "unknown")

    version <> "+" <> to_string(BuildInfo.revision())

  end

end

defmodule Mix.Tasks.Openagents.Weka.ExportTest do

  @moduledoc """

  The operator surface for the WEKA corpus. The consent gate is proven again

  here rather than only at `OpenAgents.Threads.WekaExport`, because this is the

  path a corpus actually leaves by, and a gate that holds in the context and

  not at the surface is not a gate.

  """

  use OpenAgents.DataCase

  import ExUnit.CaptureIO

  import OpenAgentsWeb.ConnCase, only: [github_user: 1]

  alias OpenAgents.Repo

  alias OpenAgents.Threads

  alias OpenAgents.Threads.Event

  test "writes a corpus from a recorded id file and refuses a dark thread" do

    consenting = session("weka-task-open", "ledger")

    dark = session("weka-task-dark", "dark")

    directory = tmp_dir()

    set = Path.join(directory, "corpus-set.txt")

    out = Path.join(directory, "corpus.json")

    File.write!(set, """

    # the recorded thread-id set

    #{consenting.id}

    #{dark.id}

    """)

    output =

      capture_io(fn ->

        Mix.Tasks.Openagents.Weka.Export.run(["--threads", set, "--out", out])

      end)

    assert output =~ "1 trace(s)"

    assert output =~ "1 refused"

    assert output =~ "refused #{dark.id}: consent_required"

    corpus = out |> File.read!() |> Jason.decode!()

    assert corpus["requested_thread_ids"] == [consenting.id, dark.id]

    assert corpus["included_thread_ids"] == [consenting.id]

    assert corpus["refused"] == [%{"thread_id" => dark.id, "reason" => "consent_required"}]

    assert is_binary(corpus["code_revision"])

    refute String.contains?(File.read!(out), "zqdark")

  after

    Mix.Task.reenable("openagents.weka.export")

  end

  test "refuses to run without a thread-id set" do

    assert_raise Mix.Error, fn -> Mix.Tasks.Openagents.Weka.Export.run([]) end

  after

    Mix.Task.reenable("openagents.weka.export")

  end

  defp session(handle, visibility) do

    user = github_user(handle)

    marker = if visibility == "dark", do: "zqdark", else: "zqopen"

    {:ok, thread} = Threads.open(user, "Objective", visibility: visibility)

    insert_event(thread, "turn.user", %{"content" => words(marker <> "u", 200)})

    insert_event(thread, "turn.assistant", %{"output" => words(marker <> "a", 120)})

    thread

  end

  defp insert_event(thread, event_type, payload) do

    %Event{}

    |> Event.changeset(%{

      thread_id: thread.id,

      event_type: event_type,

      payload: payload,

      emitted_at: DateTime.utc_now()

    })

    |> Repo.insert!()

  end

  defp words(prefix, count), do: Enum.map_join(1..count, " ", &"#{prefix}#{&1}")

  defp tmp_dir do

    directory =

      Path.join(System.tmp_dir!(), "weka-export-#{System.unique_integer([:positive])}")

    File.mkdir_p!(directory)

    on_exit(fn -> File.rm_rf!(directory) end)

    directory

  end

end

defmodule OpenAgents.Threads.WekaExportTest do

  @moduledoc """

  Tests for `OpenAgents.Threads.WekaExport`.

  The proof behind `OpenAgents.Threads.WekaExport` and the export half of

  THREAD-002.

  Two claims carry the weight. The first is the consent gate: a thread its

  owner left `dark` — and a `dark` child of a consenting parent — contributes

  nothing to a trace or a corpus, not a request, not a block count, not a

  token. The second is fidelity: the exported block ids carry the source

  session's prefix structure, so prefix reuse measured on the anonymized trace

  is the same number as prefix reuse measured on the raw transcript. Both are

  falsifiable — an exporter that chunked per event, salted per request, or

  copied text through would fail one of them.

  """

  use OpenAgents.DataCase, async: true

  alias OpenAgents.Threads.Event

  alias OpenAgents.Threads.WekaExport

  describe "export/2" do

    test "a ledger-visible thread exports a deterministic, content-free document" do

      user = github_user("weka-export-consent")

      {:ok, thread} = Threads.open(user, "Test objective", visibility: "ledger")

      fixed_dt = DateTime.from_naive!(~N[2026-08-24 12:00:00.000000], "Etc/UTC")

      thread = thread |> change(started_at: fixed_dt) |> Repo.update!()

      insert_event(thread, "turn.user", %{"content" => "secret user prompt"}, fixed_dt)

      insert_event(

        thread,

        "turn.assistant",

        %{"output" => long_text_with("confidential code")},

        fixed_dt

      )

      assert {:ok, doc} = WekaExport.export(thread, "fixed-salt")

      assert doc["format"] == "weka-trace-v1"

      assert doc["thread_id"] == thread.id

      assert doc["generation"] == thread.generation

      assert doc["visibility"] == "ledger"

      assert doc["started_at"] == DateTime.to_iso8601(fixed_dt)

      assert doc["completed_at"] == nil

      assert doc["event_count"] == 4

      json = Jason.encode!(doc)

      refute String.contains?(json, "secret user prompt")

      refute String.contains?(json, "confidential code")

      for event <- doc["events"] do

        assert is_binary(event["emitted_at"])

        assert is_integer(event["block_count"])

        assert is_list(event["blocks"])

        assert length(event["blocks"]) == event["block_count"]

  @block 64

  @base ~U[2026-08-25 10:00:00.000000Z]

  describe "export/2 — the WEKA v1 document" do

    test "a consenting thread exports a replayable weka trace" do

      thread = coding_session("weka-shape")

      assert {:ok, trace} = WekaExport.export(thread)

      assert trace["id"] == thread.id

      assert trace["block_size"] == @block

      assert trace["hash_id_scope"] == "local"

      assert is_list(trace["models"])

      assert thread.model in trace["models"]

      requests = trace["requests"]

      assert length(requests) == 3

      for request <- requests do

        assert request["type"] == "n"

        assert is_binary(request["model"])

        assert is_integer(request["in"]) and request["in"] >= 0

        assert is_integer(request["out"]) and request["out"] > 0

        assert is_list(request["hash_ids"])

        assert Enum.all?(request["hash_ids"], &is_integer/1)

        assert is_float(request["t"]) and request["t"] >= 0.0

        assert is_float(request["api_time"]) and request["api_time"] > 0.0

      end

      user_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.user"))

      assistant_event = Enum.find(doc["events"], &(&1["event_type"] == "turn.assistant"))

      # The hashed prompt is whole blocks, exactly as the reference converter

      # emits it, so `in` is what the replayer will actually send.