Plugin-scoped conversation read routes
apps/backend/internal/plugins/conversation_handlers.go ↗The handler checks binding and snapshot tokens, validates query, and returns sanitized paginated messages or turns.
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 handler core
func (c *Controller) conversationMessages(ctx *gin.Context) {
ctx.Header("Cache-Control", conversationNoStore)
record, identity, ok := c.authorizeConversationRequest(ctx)
if !ok || !c.validBinding(ctx, record, identity.UserID) {
return
}
query, ok := parseConversationMessageQuery(ctx)
if !ok {
return
}
snapshot, ok := c.validSnapshot(ctx, record, identity.UserID, sessionID)
if !ok {
return
}
messages, hasMore, err = c.svc.conversationMessagesAt(ctx.Request.Context(), sessionID, uint64(snapshot.Cutoff), query.taskID, query.authors, query.sort, cursorID, query.limit)
response, ok := c.buildConversationMessagesResponse(ctx, record, identity.UserID, sessionID, query, snapshot, messages, hasMore)
ctx.JSON(http.StatusOK, response)
}
Snapshot journal with sanitized DTOs
apps/backend/internal/plugins/conversation_journal.go ↗The journal reads message and turn versions at a committed cutoff and strips system content and private metadata.
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 = ?")
}
orderDirection := "ASC"
comparison := ">"
if sortOrder == "desc" {
orderDirection = "DESC"
comparison = "<"
}
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)
return decodeConversationSnapshotRows(rows, sessionID, limit)
}
Sanitization
func sanitizeConversationMessagePayload(target, source map[string]any) {
if content, ok := source["content"].(string); ok {
target["content"] = sysprompt.StripSystemContent(content)
}
sanitizedMetadata := SanitizeConversationMessageMetadata(metadata)
if senderTaskID, ok := source["sender_task_id"].(string); ok && senderTaskID != "" {
target["sender_task_id"] = senderTaskID
}
}
Durable ordered session event log
apps/backend/internal/plugins/conversation_stream.go ↗The log assigns a monotonic sequence, detects poison events, and persists the partition atomically.
Append with poison check
func (l *SessionEventLog) Append(sessionID string, taskID *string, eventType string, payload json.RawMessage) (SessionEvent, error) {
if !json.Valid(payload) {
return SessionEvent{}, fmt.Errorf("%w: malformed payload", ErrPoisonEvent)
}
l.mu.Lock()
defer l.mu.Unlock()
partition := l.state.Sessions[sessionID]
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 _, projectionErr := ProjectSessionEvent(event); projectionErr != nil {
poison = &SessionPoisonRecord{SessionID: sessionID, EventID: event.ID, Sequence: event.Sequence, State: SessionPoisonPending}
l.state.Poison[poisonKey(sessionID, event.ID)] = poison
}
return event, l.persistAppendLocked(event, poison)
}
WS ordered event fanout
apps/backend/internal/gateway/websocket/ordered_session_events.go ↗The hub mirrors committed journal events or appends live events, then delivers through the poison-aware dispatcher.
Hub bridge
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, err := service.SyncCommittedSessionEvents(context.Background(), sessionID)
for _, event := range events {
h.broadcastCommittedOrderedSessionEvent(service, event)
}
return
}
payload := sanitizedOrderedSessionPayload(eventType, source)
encoded, _ := json.Marshal(payload)
event, err := service.SessionEvents().Append(sessionID, taskID, eventType, encoded)
h.deliverOrderedSessionEvent(service, event, h.orderedSessionRecipients(sessionID))
}
Browser conversation scope
apps/web/lib/plugins/conversation-scope.tsx ↗The scope owns binding, snapshot tokens, ordered WS subscription, gap and poison recovery, and snapshot buffering.
Scope interface
export type ConversationScope = {
pluginId: string;
taskId: string;
sessionId: string | null;
signal: AbortSignal;
ready(): Promise<OrderedReady>;
renewContinuation(cursor: string, queryIdentity?: string): Promise<{ cursor: string; binding: OrderedReady }>;
subscribe(listener: ConversationEventListener, kind: SnapshotKind, snapshotKey: SnapshotKey): () => void;
subscribeRebind(listener: () => void): () => void;
commitSnapshot(kind: SnapshotKind, snapshotKey?: SnapshotKey): void;
invalidateSnapshot(kind: SnapshotKind, snapshotKey?: SnapshotKey): void;
accept(event: RawSessionEvent): void;
close(): void;
};
Ordered accept and buffering
accept(event: RawSessionEvent) {
if (!isRawSessionEvent(event)) { this.blockForPoison(); return; }
if (!this.acceptsScope(event, sessionId) || event.sequence < this.nextSequence) return;
this.pendingBySequence.set(event.sequence, event);
this.drainPending(sessionId);
}
private shouldBuffer(event: RawSessionEvent) {
if (this.listeners.size === 0) return true;
const eventKind = snapshotKindForEvent(event.event_type);
for (const listener of this.listeners) {
if (this.listenerSnapshotKinds.get(listener) !== eventKind) continue;
const key = this.listenerSnapshotKeys.get(listener) ?? "";
if (!this.committedSnapshots.has(`${eventKind}:${key}`)) return true;
}
return false;
}
Host facade hooks
apps/web/lib/plugins/conversation-host.tsx ↗The hooks fetch pages with binding and snapshot headers, merge pages, and project live ordered events.
Message page loader
async function fetchMessagePage({ scope, sessionId, taskId, authorsKey, sort, limit, cursor, binding }): Promise<MessagePage> {
const params = new URLSearchParams({ sort, limit: String(limit) });
if (taskId !== null) params.set("task_id", taskId);
for (const author of authorsKey ? authorsKey.split(",") : []) params.append("author_type", author);
if (cursor) params.set("cursor", cursor);
const response = await fetch(pluginConversationUrl(scope.pluginId, `/conversation/task-sessions/${encodeURIComponent(sessionId)}/messages?${params}`), {
credentials: "include",
cache: "no-store",
headers: { "X-Kandev-Plugin-Binding": binding.bindingToken, "X-Kandev-Snapshot-Token": binding.snapshotToken },
signal: scope.signal,
});
return parseConversationResponse<MessagePage>(response);
}
Live projection
React.useEffect(() => {
if (!scope || !sessionId) return;
return scope.subscribe((event) => {
if (event.event_type === "session.removed") {
setState((current) => ({ ...current, removed: true, hasMore: false }));
return true;
}
if (!event.event_type.startsWith("message.") || !eventMatchesTask(event, taskId)) return true;
const message = messageFromEvent(event);
if (!message) return false;
setState((current) => {
const messages = current.messages.filter((item) => item.id !== message.id);
messages.push(message);
messages.sort((left, right) => compareConversationMessages(left, right, sort));
return { ...current, messages };
});
return true;
}, "messages", snapshotKey);
}, [scope, sessionId, taskId, sort, snapshotKey]);
Public SDK and task-panel contract
apps/packages/plugin-sdk/src/index.ts ↗The SDK adds conversation types and extends task-panel registration with title key, visibility, and scoped navigation.
SDK 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 PluginConversationApi {
useSessionMessages(query: PluginSessionMessagesQuery): PluginSessionMessagesState;
useSessionTurns(sessionId: string | null, taskId?: string | null): PluginSessionTurnsState;
useMessageFavorite(sessionId: string | null, messageId: string): boolean;
}
export interface PluginTaskPanelProps extends PluginTaskPanelContext {
panelId: string;
conversation: PluginTaskPanelConversationCapability;
}
Host wiring
export function buildHostApi(pluginId: string, storeApi: StoreApi<AppState>): PluginHostApi {
return {
pluginId,
React,
jsx: React.createElement,
i18n: buildPluginI18nApi(pluginId),
conversation: pluginConversationApi,
ui: createPluginUIApi(pluginId),
// ...
};
}
