OpenAgents

Commit

Merge the modalities into one timeline

bcf589630a71 · AtlantisPleb · · parent 316d7d39b920 

Merge the modalities into one timeline

Voice, the coder's threads, and web chat each record their own events
in their own shape, and nothing read them together — so context did
not carry when someone spoke, then coded, then reviewed on the web.

OpenAgents.Timeline is the read side of that: one ordered sequence for
an account across the three modalities, each entry carrying the
modality it came from, a timestamp, a kind, the identifier of the
record it projects, and a display summary. The sort key is documented
and total, so the order is deterministic rather than incidental.

It reads and projects; it copies nothing. There is no new table and no
change to how any modality writes, so each record keeps one source of
truth. Scope is the account's own records throughout, which is what
keeps THREAD-002 honest here: a thread's events reach the timeline
only for the account that opened it.

This is the read side only — no UI, and no carry-over of live context
between surfaces yet.

Built by a Devin child through the openagents coder's delegate tool;
57 thread, voice, and timeline tests re-run before landing.

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 346 · 2026-08-25T06:03:19.758084Z

Changed files

  • added lib/openagents/timeline.ex
  • added test/openagents/timeline_test.exs

Diff

2 files changed, +325 -0

lib/openagents/timeline.ex added +242

@@ -0,0 +1,242 @@

1 
defmodule OpenAgents.Timeline do
2 
  @moduledoc """
3 
  Owner-scoped, read-only merge of timeline entries across an account's
4 
  coder/threads, voice sessions, and web chat.
5
6 
  Every entry is rooted in the account's visitor, so a caller can see only
7 
  records the account already owns. Thread transcripts are included only for
8 
  threads the account opened; `dark` threads therefore stay visible only to
9 
  their owner (THREAD-002).
10 
  """
11
12 
  import Ecto.Query
13
14 
  alias OpenAgents.Accounts.User
15 
  alias OpenAgents.Conversations.{Conversation, ToolStep, Turn, Visitor}
16 
  alias OpenAgents.Repo
17 
  alias OpenAgents.Threads.{Event, Thread}
18 
  alias OpenAgents.Voice.{Session, TranscriptItem}
19 
  alias OpenAgents.Voice.ToolStep, as: VoiceToolStep
20
21 
  @enforce_keys [:modality, :timestamp, :kind, :record_id, :summary]
22 
  defstruct [:modality, :timestamp, :kind, :record_id, :summary, metadata: %{}]
23
24 
  @type t :: %__MODULE__{
25 
          modality: :voice | :coder | :chat,
26 
          timestamp: DateTime.t() | nil,
27 
          kind: :turn | :tool_step | :decision | :system,
28 
          record_id: term(),
29 
          summary: String.t(),
30 
          metadata: map()
31 
        }
32
33 
  @doc """
34 
  Returns the account's timeline, oldest first, with a documented tie-break.
35
36 
  The sort key is `{timestamp, modality rank, record_id, kind rank}`. Rows
37 
  without an event timestamp are ordered as though they occurred at the Unix
38 
  epoch start, which is deterministic and monotonic.
39 
  """
40 
  @spec for_user(User.t()) :: [t()]
41 
  def for_user(%User{id: user_id}) do
42 
    case Repo.get_by(Visitor, user_id: user_id) do
43 
      %Visitor{id: visitor_id} ->
44 
        [thread_entries(visitor_id), voice_entries(visitor_id), chat_entries(visitor_id)]
45 
        |> Enum.concat()
46 
        |> Enum.sort_by(&entry_sort_key/1)
47
48 
      nil ->
49 
        []
50 
    end
51 
  end
52
53 
  defp entry_sort_key(%__MODULE__{timestamp: nil} = entry) do
54 
    {sentinel_timestamp(), modality_rank(entry.modality), entry.record_id, kind_rank(entry.kind)}
55 
  end
56
57 
  defp entry_sort_key(%__MODULE__{timestamp: ts} = entry) do
58 
    {ts, modality_rank(entry.modality), entry.record_id, kind_rank(entry.kind)}
59 
  end
60
61 
  defp modality_rank(:coder), do: 0
62 
  defp modality_rank(:voice), do: 1
63 
  defp modality_rank(:chat), do: 2
64
65 
  defp kind_rank(:turn), do: 0
66 
  defp kind_rank(:tool_step), do: 1
67 
  defp kind_rank(:decision), do: 2
68 
  defp kind_rank(:system), do: 3
69
70 
  defp sentinel_timestamp do
71 
    DateTime.from_naive!(~N[1970-01-01 00:00:00], "Etc/UTC")
72 
  end
73
74 
  # Coder/threads
75 
  defp thread_entries(visitor_id) do
76 
    thread_ids =
77 
      from(t in Thread, where: t.owner_visitor_id == ^visitor_id, select: t.id)
78 
      |> Repo.all()
79
80 
    from(e in Event,
81 
      where: e.thread_id in ^thread_ids,
82 
      order_by: [asc: e.emitted_at, asc: e.id]
83 
    )
84 
    |> Repo.all()
85 
    |> Enum.map(&thread_entry/1)
86 
  end
87
88 
  defp thread_entry(%Event{} = event) do
89 
    %__MODULE__{
90 
      modality: :coder,
91 
      timestamp: event.emitted_at,
92 
      kind: thread_event_kind(event.event_type),
93 
      record_id: event.id,
94 
      summary: thread_summary(event),
95 
      metadata: %{event_type: event.event_type, thread_id: event.thread_id}
96 
    }
97 
  end
98
99 
  defp thread_event_kind("thread.opened"), do: :system
100 
  defp thread_event_kind("thread.visibility_set"), do: :system
101 
  defp thread_event_kind("thread.turn." <> _), do: :turn
102 
  defp thread_event_kind("tool." <> _), do: :tool_step
103 
  defp thread_event_kind(_), do: :decision
104
105 
  defp thread_summary(%Event{event_type: event_type, payload: payload}) do
106 
    case event_type do
107 
      "tool.ran" ->
108 
        "Tool ran: #{payload["tool"] || event_type}"
109
110 
      "thread.opened" ->
111 
        "Thread opened"
112
113 
      "thread.visibility_set" ->
114 
        "Thread visibility changed"
115
116 
      _ ->
117 
        humanize_event_type(event_type)
118 
    end
119 
  end
120
121 
  defp humanize_event_type(event_type) do
122 
    event_type
123 
    |> String.replace(".", " ")
124 
    |> String.capitalize()
125 
  end
126
127 
  # Voice
128 
  defp voice_entries(visitor_id) do
129 
    conversation_ids =
130 
      from(c in Conversation, where: c.visitor_id == ^visitor_id, select: c.id)
131 
      |> Repo.all()
132
133 
    session_ids =
134 
      from(s in Session, where: s.conversation_id in ^conversation_ids, select: s.id)
135 
      |> Repo.all()
136
137 
    transcript_items =
138 
      from(ti in TranscriptItem, where: ti.voice_session_id in ^session_ids)
139 
      |> Repo.all()
140 
      |> Enum.map(&voice_transcript_entry/1)
141
142 
    tool_steps =
143 
      from(ts in VoiceToolStep, where: ts.voice_session_id in ^session_ids)
144 
      |> Repo.all()
145 
      |> Enum.map(&voice_tool_step_entry/1)
146
147 
    transcript_items ++ tool_steps
148 
  end
149
150 
  defp voice_transcript_entry(%TranscriptItem{} = item) do
151 
    %__MODULE__{
152 
      modality: :voice,
153 
      timestamp: item.observed_at,
154 
      kind: :turn,
155 
      record_id: item.id,
156 
      summary: "Voice #{item.role}: #{shorten(item.content)}",
157 
      metadata: %{
158 
        role: item.role,
159 
        status: item.status,
160 
        voice_session_id: item.voice_session_id
161 
      }
162 
    }
163 
  end
164
165 
  defp voice_tool_step_entry(%VoiceToolStep{} = step) do
166 
    %__MODULE__{
167 
      modality: :voice,
168 
      timestamp: step.requested_at,
169 
      kind: :tool_step,
170 
      record_id: step.id,
171 
      summary: "Voice tool #{step.tool_name} (#{step.status})",
172 
      metadata: %{
173 
        tool_name: step.tool_name,
174 
        status: step.status,
175 
        voice_session_id: step.voice_session_id
176 
      }
177 
    }
178 
  end
179
180 
  # Web chat
181 
  defp chat_entries(visitor_id) do
182 
    conversation_ids =
183 
      from(c in Conversation, where: c.visitor_id == ^visitor_id, select: c.id)
184 
      |> Repo.all()
185
186 
    turns =
187 
      from(t in Turn, where: t.conversation_id in ^conversation_ids, preload: [:user_message])
188 
      |> Repo.all()
189 
      |> Enum.map(&chat_turn_entry/1)
190
191 
    tool_steps =
192 
      from(ts in ToolStep,
193 
        join: t in Turn,
194 
        on: t.id == ts.turn_id,
195 
        where: t.conversation_id in ^conversation_ids
196 
      )
197 
      |> Repo.all()
198 
      |> Enum.map(&chat_tool_step_entry/1)
199
200 
    turns ++ tool_steps
201 
  end
202
203 
  defp chat_turn_entry(%Turn{} = turn) do
204 
    content = if turn.user_message, do: turn.user_message.content, else: ""
205 
    timestamp = turn.started_at || turn.inserted_at
206
207 
    %__MODULE__{
208 
      modality: :chat,
209 
      timestamp: timestamp,
210 
      kind: :turn,
211 
      record_id: turn.id,
212 
      summary: "Chat: #{shorten(content)}",
213 
      metadata: %{
214 
        status: turn.status,
215 
        conversation_id: turn.conversation_id
216 
      }
217 
    }
218 
  end
219
220 
  defp chat_tool_step_entry(%ToolStep{} = step) do
221 
    %__MODULE__{
222 
      modality: :chat,
223 
      timestamp: step.requested_at,
224 
      kind: :tool_step,
225 
      record_id: step.id,
226 
      summary: "Tool #{step.tool_name} (#{step.status})",
227 
      metadata: %{
228 
        tool_name: step.tool_name,
229 
        status: step.status,
230 
        side_effect_class: step.side_effect_class
231 
      }
232 
    }
233 
  end
234
235 
  defp shorten(content) when is_binary(content) do
236 
    if String.length(content) > 80,
237 
      do: String.slice(content, 0, 80) <> "…",
238 
      else: content
239 
  end
240
241 
  defp shorten(_content), do: ""
242 
end
test/openagents/timeline_test.exs added +83

@@ -0,0 +1,83 @@

1 
defmodule OpenAgents.TimelineTest do
2 
  @moduledoc """
3 
  Issue #228: an owner-scoped, ordered timeline across coder/threads,
4 
  voice, and web chat.
5 
  """
6
7 
  use OpenAgents.DataCase, async: false
8
9 
  import OpenAgentsWeb.ConnCase, only: [github_user: 1]
10
11 
  alias OpenAgents.{Conversations, Threads, Timeline}
12
13 
  defp owner(key), do: github_user("timeline-#{key}")
14
15 
  describe "for_user/1" do
16 
    test "merges thread events and web chat turns in timestamp order" do
17 
      user = owner("merge")
18 
      {:ok, thread} = Threads.open(user, "Build the timeline")
19
20 
      {:ok, _thread} =
21 
        Threads.record_event(thread, "thread.turn.started", %{"turn" => 1})
22
23 
      {:ok, conversation} = Conversations.ensure_conversation(user)
24
25 
      {:ok, %{turn: turn}} =
26 
        Conversations.create_turn(conversation, "Hello from web chat")
27
28 
      entries = Timeline.for_user(user)
29
30 
      assert length(entries) == 3
31
32 
      coder = Enum.filter(entries, &(&1.modality == :coder))
33 
      chat = Enum.filter(entries, &(&1.modality == :chat))
34
35 
      assert length(coder) == 2
36 
      assert length(chat) == 1
37
38 
      opened = Enum.find(coder, &(&1.summary == "Thread opened"))
39 
      started = Enum.find(coder, &(&1.summary == "Thread turn started"))
40
41 
      assert opened.modality == :coder
42 
      assert opened.kind == :system
43 
      assert opened.record_id == Threads.list_events(thread) |> hd() |> Map.fetch!(:id)
44
45 
      assert started.modality == :coder
46 
      assert started.kind == :turn
47
48 
      [chat_turn] = chat
49 
      assert chat_turn.modality == :chat
50 
      assert chat_turn.kind == :turn
51 
      assert chat_turn.record_id == turn.id
52 
      assert chat_turn.summary =~ "Hello from web chat"
53
54 
      timestamps = Enum.map(entries, & &1.timestamp)
55 
      assert timestamps == Enum.sort(timestamps)
56 
    end
57
58 
    test "excludes records owned by another account" do
59 
      me = owner("me")
60 
      other = owner("other")
61
62 
      {:ok, _my_thread} = Threads.open(me, "My private work")
63 
      {:ok, _other_thread} = Threads.open(other, "Their private work")
64
65 
      my_entries = Timeline.for_user(me)
66 
      other_entries = Timeline.for_user(other)
67
68 
      assert length(my_entries) == 1
69 
      assert length(other_entries) == 1
70 
      refute List.first(my_entries).record_id == List.first(other_entries).record_id
71 
    end
72
73 
    test "returns an empty list for an account with no visitor" do
74 
      user = owner("no-visitor")
75
76 
      # The account has never opened a thread, chat, or voice session,
77 
      # so no visitor row exists yet and there is nothing to merge.
78 
      entries = Timeline.for_user(user)
79
80 
      assert entries == []
81 
    end
82 
  end
83 
end

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