SDK contract for browser conversation
apps/packages/plugin-sdk/src/index.ts ↗The SDK defines the public, runtime-free types that plugins import and the Host implements.
Public types
export interface PluginConversationMessage {
id: string;
taskId: string | null;
sessionId: string;
turnId?: string;
authorType: PluginConversationAuthor;
type: string;
content: string;
createdAt: string;
updatedAt: string;
promptIndex?: number;
senderTaskId?: string;
}
export interface PluginConversationTurn {
id: string;
taskId: string | null;
sessionId: string;
startedAt: string;
completedAt?: string;
updatedAt: string;
}
export interface PluginConversationApi {
useSessionMessages(query: PluginSessionMessagesQuery): PluginSessionMessagesState;
useSessionTurns(sessionId: string | null, taskId?: string | null): PluginSessionTurnsState;
useMessageFavorite(sessionId: string | null, messageId: string): boolean;
}
export interface PluginTaskPanelConversationCapability {
openMessage(messageId: string): PluginOpenMessageResult;
history: PluginConversationApi;
}
Capability-gated conversation routes
apps/backend/internal/plugins/conversation_handlers.go ↗The backend exposes authenticated, plugin-scoped reads that enforce api_read:messages and session authorization.
Route registration
func registerConversationRoutes(api *gin.RouterGroup, ctrl *Controller) {
api.GET("/:id/conversation/binding", ctrl.conversationBinding)
api.GET("/:id/conversation/task-sessions/:sessionId/messages", ctrl.conversationMessages)
api.GET("/:id/conversation/task-sessions/:sessionId/turns", ctrl.conversationTurns)
api.POST("/:id/conversation/continuation/renew", ctrl.conversationContinuationRenew)
}
Message DTO mapping
func conversationMessageModelToDTO(message *taskmodels.Message) conversationMessageDTO {
updatedAt := message.UpdatedAt
if updatedAt.IsZero() {
updatedAt = message.CreatedAt
}
dto := conversationMessageDTO{
ID: message.ID,
TaskID: nonEmptyStringPointer(message.TaskID),
SessionID: message.TaskSessionID,
AuthorType: string(message.AuthorType),
Content: sysprompt.StripSystemContent(message.Content),
CreatedAt: message.CreatedAt.UTC().Format(time.RFC3339Nano),
UpdatedAt: updatedAt.UTC().Format(time.RFC3339Nano),
}
if senderTaskID, ok := message.Metadata["sender_task_id"].(string); ok && senderTaskID != "" {
dto.SenderTaskID = &senderTaskID
}
return dto
}
Journal snapshot at cutoff
apps/backend/internal/plugins/conversation_journal.go ↗The journal serves deterministic pages from immutable version rows at the committed cutoff, not from live tables.
Snapshot query
func (s *Service) conversationMessagesAt(ctx context.Context, sessionID string, cutoff uint64, taskID *string, authors []string, sortOrder string, cursorID string, limit int) ([]*taskmodels.Message, bool, error) {
where := []string{"version_rank = 1", "tombstone = FALSE"}
if taskID != nil {
where = append(where, "task_id = ?")
}
query := fmt.Sprintf(`
WITH ranked AS (
SELECT message_id, task_id, author_type, created_at, tombstone, payload,
ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY row_sequence DESC) AS version_rank
FROM conversation_message_versions
WHERE session_id = ? AND row_sequence <= ?
), live AS (
SELECT * FROM ranked WHERE version_rank = 1 AND tombstone = FALSE
)
SELECT payload FROM live
WHERE %s
ORDER BY created_at %s, message_id %s
LIMIT ?`, strings.Join(where, " AND "), orderDirection, orderDirection)
rows, err := s.conversationJournal.QueryxContext(ctx, query, args...)
return decodeConversationSnapshotRows(rows, sessionID, limit)
}
Durable ordered event stream
apps/backend/internal/plugins/conversation_stream.go ↗The stream allocates a per-session sequence, persists the event, and tracks poison delivery with lease and retry.
Append with poison detection
func (l *SessionEventLog) Append(sessionID string, taskID *string, eventType string, payload json.RawMessage) (SessionEvent, error) {
partition := l.state.Sessions[sessionID]
if partition == nil {
partition = &sessionEventPartition{}
l.state.Sessions[sessionID] = partition
}
if partition.Terminal {
return SessionEvent{}, ErrSessionRemoved
}
event := SessionEvent{
ProtocolVersion: SessionEventProtocolVersion,
EventType: eventType,
SessionID: sessionID,
TaskID: taskID,
Sequence: partition.Watermark + 1,
ID: uuid.NewString(),
Payload: append(json.RawMessage(nil), payload...),
CreatedAt: l.now().UTC(),
}
partition.Events = append(partition.Events, event)
partition.Watermark = event.Sequence
if _, err := ProjectSessionEvent(event); err != nil {
l.state.Poison[poisonKey(sessionID, event.ID)] = &SessionPoisonRecord{
SessionID: sessionID, EventID: event.ID, Sequence: event.Sequence,
State: SessionPoisonPending, OwnerEpoch: 1,
}
}
return event, l.persistAppendLocked(event, poison)
}
Hub fanout
func (h *Hub) appendAndBroadcastOrderedSessionEvent(sessionID string, message *ws.Message) {
h.orderedSessionMu.Lock()
defer h.orderedSessionMu.Unlock()
service := h.pluginConversationService
eventType := orderedEventTypeByAction[message.Action]
if service.HasConversationJournal() {
events, _ := service.SyncCommittedSessionEvents(context.Background(), sessionID)
for _, event := range events {
h.broadcastCommittedOrderedSessionEvent(service, event)
}
return
}
payload := sanitizedOrderedSessionPayload(eventType, source)
event, _ := service.SessionEvents().Append(sessionID, taskID, eventType, encoded)
h.deliverOrderedSessionEvent(service, event, h.orderedSessionRecipients(sessionID))
}
Ordered conversation scope
apps/web/lib/plugins/conversation-scope.tsx ↗The scope owns binding, watermark, buffering, and reconnect so hooks never handle transport directly.
Scope state
class OrderedConversationScope implements ConversationScope {
private bindingPromise: Promise<Binding> | null = null;
private readyPromise: Promise<OrderedReady> | null = null;
private acknowledgedSequence = 0;
private nextSequence = 1;
private sequenceBlocked = false;
private terminal = false;
private readonly buffered: RawSessionEvent[] = [];
private readonly pendingBySequence = new Map<number, RawSessionEvent>();
private consumerId = generateUUID();
async ready(): Promise<OrderedReady> {
const pending = (this.readyPromise ??= this.initializeReady());
return pending.then((current) => this.refreshBindingIfNeeded(current));
}
}
Event acceptance
accept(event: RawSessionEvent) {
if (!isRawSessionEvent(event)) {
this.blockForPoison();
return;
}
if (event.sequence < this.nextSequence) return;
this.pendingBySequence.set(event.sequence, event);
this.drainPending(sessionId);
}
private drainPending(sessionId: string) {
while (!this.sequenceBlocked && !this.terminal) {
const event = this.pendingBySequence.get(this.nextSequence);
if (!event) return;
if (!isCompatibleConversationEvent(event, sessionId)) {
this.blockForPoison();
return;
}
this.pendingBySequence.delete(this.nextSequence);
this.nextSequence += 1;
if (this.shouldBuffer(event)) {
this.buffered.push(event);
continue;
}
if (!this.project(event)) {
this.sequenceBlocked = true;
return;
}
}
}
Host conversation hooks
apps/web/lib/plugins/conversation-host.tsx ↗The hooks merge paginated snapshots with live events, handle loadMore joining, and fence stale generations.
Message hook
function useSessionMessages(query: PluginSessionMessagesQuery): PluginSessionMessagesState {
const scope = React.useContext(ConversationScopeContext);
const [state, setState] = React.useState({ ...EMPTY_MESSAGES });
const cursorRef = React.useRef<string | null>(null);
const resolved = React.useMemo(() => scope ? resolveTaskId(scope, query.taskId) : { taskId: null, error: null }, [query.taskId, scope]);
const snapshotKey = JSON.stringify([query.sessionId, resolved.taskId, authorsKey, sort, limit]);
const loadPage = useMessagePageLoader({ scope, sessionId: query.sessionId, taskId: resolved.taskId, snapshotKey, setState, cursorRef });
useOrderedMessageEvents({ scope, sessionId: query.sessionId, taskId: resolved.taskId, snapshotKey, setState });
useInitialMessagePage({ scope, sessionId: query.sessionId, revision, loadPage, setState });
return useMessageControls({ scope, sessionId: query.sessionId, state, setState, loadPage, cursorRef });
}
Page merge
function mergeMessagePage(current: readonly PluginConversationMessage[], page: readonly PluginConversationMessage[], append: boolean, sort: "asc" | "desc") {
const existingIds = new Set(current.map((m) => m.id));
const messagesById = new Map<string, PluginConversationMessage>();
if (append) for (const m of current) messagesById.set(m.id, m);
for (const m of page) {
const prev = messagesById.get(m.id);
if (!prev || prev.updatedAt.localeCompare(m.updatedAt) < 0) messagesById.set(m.id, m);
}
const messages = Array.from(messagesById.values());
messages.sort((a, b) => compareConversationMessages(a, b, sort));
return { messages, additionCount: page.filter((m) => !existingIds.has(m.id)).length };
}
