diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 91b8473925..856ea0ae0a 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -386,56 +386,26 @@ func main() { result = append(result, gitStatusNotification) } } + + if contextWindow, ok := session.Metadata["context_window"]; ok { + notification, err := ws.NewNotification(ws.ActionSessionStateChanged, map[string]interface{}{ + "session_id": sessionID, + "task_id": session.TaskID, + "new_state": string(session.State), + "metadata": map[string]interface{}{ + "context_window": contextWindow, + }, + }) + if err == nil { + result = append(result, notification) + } + } } return result, nil }) log.Info("Session data provider configured for session subscriptions (git status)") - // NOTE: Pending permissions are now stored as messages and retrieved via the historical logs provider. - // The separate pending permissions provider is no longer needed. - - // NOTE: We no longer create comments for each ACP message chunk. - // Instead, the lifecycle manager accumulates message chunks and: - // 1. Flushes the buffer as a comment when a tool use starts (step boundary) - // 2. Flushes the buffer as a comment when the prompt completes (final response) - // This is handled via the prompt.complete and step.complete events. - - // Wire stream handler to broadcast agent stream events to WebSocket clients - // NOTE: Permission requests are now handled via message creation in the lifecycle manager - // and broadcast automatically via the session.message.added event - orchestratorSvc.RegisterStreamHandler(func(payload *lifecycle.AgentStreamEventPayload) { - if payload == nil || payload.Data == nil { - return - } - - sessionID := payload.SessionID - if sessionID == "" { - log.Warn("agent stream event missing session_id, skipping", - zap.String("task_id", payload.TaskID), - zap.String("type", payload.Data.Type)) - return - } - - // Broadcast agent stream event to session subscribers - action := "agent." + payload.Data.Type - notification, _ := ws.NewNotification(action, map[string]interface{}{ - "task_id": payload.TaskID, - "session_id": sessionID, - "type": payload.Data.Type, - "text": payload.Data.Text, - "tool_call_id": payload.Data.ToolCallID, - "tool_name": payload.Data.ToolName, - "tool_title": payload.Data.ToolTitle, - "tool_status": payload.Data.ToolStatus, - "tool_args": payload.Data.ToolArgs, - "tool_result": payload.Data.ToolResult, - "error": payload.Data.Error, - "timestamp": payload.Timestamp, - }) - gateway.Hub.BroadcastToSession(sessionID, notification) - }) - if err := orchestratorSvc.Start(ctx); err != nil { log.Fatal("Failed to start orchestrator", zap.Error(err)) } diff --git a/apps/backend/internal/agent/lifecycle/event_types.go b/apps/backend/internal/agent/lifecycle/event_types.go index 2a0934dd1f..84410bde59 100644 --- a/apps/backend/internal/agent/lifecycle/event_types.go +++ b/apps/backend/internal/agent/lifecycle/event_types.go @@ -137,3 +137,20 @@ type ShellExitEventPayload struct { func (p ShellExitEventPayload) GetSessionID() string { return p.SessionID } + +// ContextWindowEventPayload is the payload for context window update events. +type ContextWindowEventPayload struct { + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` + AgentID string `json:"agent_id"` + ContextWindowSize int64 `json:"context_window_size"` + ContextWindowUsed int64 `json:"context_window_used"` + ContextWindowRemaining int64 `json:"context_window_remaining"` + ContextEfficiency float64 `json:"context_efficiency"` + Timestamp string `json:"timestamp"` +} + +// GetSessionID returns the session ID for this event (used by event routing). +func (p ContextWindowEventPayload) GetSessionID() string { + return p.SessionID +} diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go index 96475c9db5..066871252d 100644 --- a/apps/backend/internal/agent/lifecycle/events.go +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -323,3 +323,34 @@ func (p *EventPublisher) PublishShellExit(execution *AgentExecution, exitCode in zap.Error(err)) } } + +// PublishContextWindow publishes a context window update event to the event bus. +func (p *EventPublisher) PublishContextWindow(execution *AgentExecution, size, used, remaining int64, efficiency float64) { + if p.eventBus == nil { + return + } + + sessionID := execution.SessionID + + payload := ContextWindowEventPayload{ + TaskID: execution.TaskID, + SessionID: sessionID, + AgentID: execution.ID, + ContextWindowSize: size, + ContextWindowUsed: used, + ContextWindowRemaining: remaining, + ContextEfficiency: efficiency, + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + } + + event := bus.NewEvent(events.ContextWindowUpdated, "agent-manager", payload) + subject := events.BuildContextWindowSubject(sessionID) + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish context window event", + zap.String("instance_id", execution.ID), + zap.String("task_id", execution.TaskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 4fe160051c..85530f4f8c 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -901,6 +901,24 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age // Handle permission request inline m.handlePermissionRequestEvent(execution, event) + + case "context_window": + m.logger.Debug("context window update received", + zap.String("execution_id", execution.ID), + zap.Int64("size", event.ContextWindowSize), + zap.Int64("used", event.ContextWindowUsed), + zap.Float64("efficiency", event.ContextEfficiency)) + + // Publish context window event to event bus for orchestrator to persist + m.eventPublisher.PublishContextWindow( + execution, + event.ContextWindowSize, + event.ContextWindowUsed, + event.ContextWindowRemaining, + event.ContextEfficiency, + ) + // Return early - context window events don't need to be published as stream events + return } // Publish agent stream event to event bus for WebSocket streaming diff --git a/apps/backend/internal/agentctl/server/adapter/adapter.go b/apps/backend/internal/agentctl/server/adapter/adapter.go index ab4225878f..094558faec 100644 --- a/apps/backend/internal/agentctl/server/adapter/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/adapter.go @@ -35,6 +35,7 @@ const ( EventTypeComplete = streams.EventTypeComplete EventTypeError = streams.EventTypeError EventTypePermissionRequest = streams.EventTypePermissionRequest + EventTypeContextWindow = streams.EventTypeContextWindow ) // AgentInfo contains information about the connected agent. diff --git a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go index 31dcba3799..76bc9fcffa 100644 --- a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go @@ -645,6 +645,55 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) ToolResult: p.Delta, }) + case codex.NotifyThreadTokenUsageUpdated: + var p codex.ThreadTokenUsageUpdatedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse thread token usage updated notification", zap.Error(err)) + return + } + // Extract context window information from the token usage update + if p.TokenUsage != nil && p.TokenUsage.ModelContextWindow > 0 { + contextWindowSize := p.TokenUsage.ModelContextWindow + contextWindowUsed := int64(p.TokenUsage.Last.TotalTokens) + + remaining := contextWindowSize - contextWindowUsed + if remaining < 0 { + remaining = 0 + } + efficiency := float64(contextWindowUsed) / float64(contextWindowSize) * 100 + + a.logger.Debug("emitting context window event", + zap.Int64("size", contextWindowSize), + zap.Int64("used", contextWindowUsed), + zap.Int64("remaining", remaining), + zap.Float64("efficiency", efficiency)) + + a.sendUpdate(AgentEvent{ + Type: EventTypeContextWindow, + SessionID: threadID, + OperationID: turnID, + ContextWindowSize: contextWindowSize, + ContextWindowUsed: contextWindowUsed, + ContextWindowRemaining: remaining, + ContextEfficiency: efficiency, + }) + } + + case codex.NotifyTokenCount: + // Legacy token_count notification - ignore as we now use thread/tokenUsage/updated + a.logger.Debug("ignoring legacy token_count notification") + + case codex.NotifyContextCompacted: + var p codex.ContextCompactedParams + if err := json.Unmarshal(params, &p); err != nil { + a.logger.Warn("failed to parse context compacted notification", zap.Error(err)) + return + } + a.logger.Info("context compacted", + zap.String("thread_id", p.ThreadID), + zap.String("turn_id", p.TurnID)) + // We could emit an event here if we want to notify the frontend about compaction + default: // Log unhandled notifications at debug level a.logger.Debug("unhandled notification", zap.String("method", method)) diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go index e3bda6db3f..c1d0acee5a 100644 --- a/apps/backend/internal/agentctl/types/streams/agent.go +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -28,6 +28,9 @@ const ( // EventTypeSessionStatus indicates a session status update (resumed or new). EventTypeSessionStatus = "session_status" + + // EventTypeContextWindow indicates a context window update. + EventTypeContextWindow = "context_window" ) // Session status constants for EventTypeSessionStatus events. @@ -129,6 +132,20 @@ type AgentEvent struct { // Data contains raw protocol-specific extensions. Data map[string]interface{} `json:"data,omitempty"` + + // --- Context window fields --- + + // ContextWindowSize is the total available tokens in the context window. + ContextWindowSize int64 `json:"context_window_size,omitempty"` + + // ContextWindowUsed is the number of tokens currently consumed. + ContextWindowUsed int64 `json:"context_window_used,omitempty"` + + // ContextWindowRemaining is the number of available tokens left. + ContextWindowRemaining int64 `json:"context_window_remaining,omitempty"` + + // ContextEfficiency is the percentage utilization (0-100). + ContextEfficiency float64 `json:"context_efficiency,omitempty"` } // PlanEntry represents an entry in the agent's execution plan. diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 3dff8c3abb..1f990a5ebe 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -110,6 +110,11 @@ const ( ShellExit = "shell.exit" // Shell process exited ) +// Event types for context window +const ( + ContextWindowUpdated = "context_window.updated" // Context window usage updated +) + // BuildShellOutputSubject creates a shell output subject for a specific session func BuildShellOutputSubject(sessionID string) string { return ShellOutput + "." + sessionID @@ -169,3 +174,13 @@ func BuildPermissionRequestSubject(sessionID string) string { func BuildPermissionRequestWildcardSubject() string { return PermissionRequestReceived + ".*" } + +// BuildContextWindowSubject creates a context window subject for a specific session +func BuildContextWindowSubject(sessionID string) string { + return ContextWindowUpdated + "." + sessionID +} + +// BuildContextWindowWildcardSubject creates a wildcard subscription for all context window events +func BuildContextWindowWildcardSubject() string { + return ContextWindowUpdated + ".*" +} diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index b74f4f75f8..158f20c910 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/apps/backend/internal/gateway/websocket/task_notifications.go @@ -88,6 +88,19 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action } else if data, ok := event.Data.(interface{ GetSessionID() string }); ok { sessionID = data.GetSessionID() } + + // Debug logging for session state changes with metadata + if action == ws.ActionSessionStateChanged { + if data, ok := event.Data.(map[string]interface{}); ok { + if metadata, ok := data["metadata"]; ok { + b.logger.Info("received session.state_changed with metadata", + zap.String("action", action), + zap.String("session_id", sessionID), + zap.Any("metadata", metadata)) + } + } + } + switch action { case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError, ws.ActionSessionStateChanged: if sessionID != "" { diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index c6f33d4282..26c26e9d88 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -138,14 +138,15 @@ func NewService( // Create the watcher with event handlers that wire everything together handlers := watcher.EventHandlers{ - OnTaskStateChanged: s.handleTaskStateChanged, - OnAgentReady: s.handleAgentReady, - OnAgentCompleted: s.handleAgentCompleted, - OnAgentFailed: s.handleAgentFailed, - OnAgentStreamEvent: s.handleAgentStreamEvent, - OnACPSessionCreated: s.handleACPSessionCreated, - OnPermissionRequest: s.handlePermissionRequest, - OnGitStatusUpdated: s.handleGitStatusUpdated, + OnTaskStateChanged: s.handleTaskStateChanged, + OnAgentReady: s.handleAgentReady, + OnAgentCompleted: s.handleAgentCompleted, + OnAgentFailed: s.handleAgentFailed, + OnAgentStreamEvent: s.handleAgentStreamEvent, + OnACPSessionCreated: s.handleACPSessionCreated, + OnPermissionRequest: s.handlePermissionRequest, + OnGitStatusUpdated: s.handleGitStatusUpdated, + OnContextWindowUpdated: s.handleContextWindowUpdated, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) @@ -1237,6 +1238,72 @@ func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitSt }() } +// handleContextWindowUpdated handles context window updates and persists them to session metadata +func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.ContextWindowData) { + s.logger.Debug("handling context window update", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.TaskSessionID), + zap.Int64("size", data.ContextWindowSize), + zap.Int64("used", data.ContextWindowUsed)) + + if data.TaskSessionID == "" { + s.logger.Debug("missing session_id for context window update", + zap.String("task_id", data.TaskID)) + return + } + + session, err := s.repo.GetTaskSession(ctx, data.TaskSessionID) + if err != nil { + s.logger.Debug("no task session for context window update", + zap.String("session_id", data.TaskSessionID), + zap.Error(err)) + return + } + + // Update session metadata with context window info + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + contextWindowData := map[string]interface{}{ + "size": data.ContextWindowSize, + "used": data.ContextWindowUsed, + "remaining": data.ContextWindowRemaining, + "efficiency": data.ContextEfficiency, + } + session.Metadata["context_window"] = contextWindowData + + // Persist to database asynchronously + go func() { + if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil { + s.logger.Error("failed to update session with context window", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID), + zap.Error(err)) + } else { + s.logger.Debug("persisted context window to session", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID)) + } + }() + + // Broadcast session state change with metadata so frontend can update + // This uses the existing session.state_changed event with metadata included + if s.eventBus != nil { + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( + events.TaskSessionStateChanged, + "orchestrator", + map[string]interface{}{ + "task_id": data.TaskID, + "session_id": session.ID, + "new_state": string(session.State), + "metadata": map[string]interface{}{ + "context_window": contextWindowData, + }, + }, + )) + } +} + // handlePermissionRequest handles permission request events and saves as message func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.PermissionRequestData) { s.logger.Info("handling permission request", diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 9cb08e14af..18b0649feb 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -70,6 +70,18 @@ type GitStatusData struct { Timestamp string `json:"timestamp"` } +// ContextWindowData contains data from context window events +type ContextWindowData struct { + TaskID string `json:"task_id"` + TaskSessionID string `json:"session_id"` + AgentID string `json:"agent_id"` + ContextWindowSize int64 `json:"context_window_size"` + ContextWindowUsed int64 `json:"context_window_used"` + ContextWindowRemaining int64 `json:"context_window_remaining"` + ContextEfficiency float64 `json:"context_efficiency"` + Timestamp string `json:"timestamp"` +} + // EventHandlers contains callbacks for different event types type EventHandlers struct { // Task events @@ -94,6 +106,9 @@ type EventHandlers struct { // Git status events OnGitStatusUpdated func(ctx context.Context, data GitStatusData) + + // Context window events + OnContextWindowUpdated func(ctx context.Context, data ContextWindowData) } // Watcher subscribes to events and dispatches to handlers @@ -166,6 +181,12 @@ func (w *Watcher) Start(ctx context.Context) error { return err } + // Subscribe to context window events + if err := w.subscribeToContextWindowEvents(); err != nil { + w.unsubscribeAll() + return err + } + w.running = true w.logger.Info("Event watcher started", zap.Int("subscriptions", len(w.subscriptions))) return nil @@ -484,6 +505,50 @@ func (w *Watcher) createGitStatusHandler() bus.EventHandler { } } +// subscribeToContextWindowEvents subscribes to context window events +func (w *Watcher) subscribeToContextWindowEvents() error { + if w.handlers.OnContextWindowUpdated == nil { + return nil + } + + // Use wildcard to subscribe to all context window events (context_window.updated.{session_id}) + subject := events.BuildContextWindowWildcardSubject() + + // Use regular subscription (each instance needs all messages) + sub, err := w.eventBus.Subscribe(subject, w.createContextWindowHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to context window events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + +// createContextWindowHandler creates a bus.EventHandler for context window events +func (w *Watcher) createContextWindowHandler() bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data ContextWindowData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse context window event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil // Don't return error to continue processing other events + } + + w.logger.Debug("Handling context window event", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.TaskSessionID), + zap.Int64("size", data.ContextWindowSize), + zap.Int64("used", data.ContextWindowUsed)) + + w.handlers.OnContextWindowUpdated(ctx, data) + return nil + } +} + // parseEventData converts event data (map or struct) to a typed struct func (w *Watcher) parseEventData(data interface{}, target interface{}) error { // Marshal to JSON and unmarshal to target type diff --git a/apps/backend/pkg/codex/types.go b/apps/backend/pkg/codex/types.go index 9c6978dea7..8d29fc2274 100644 --- a/apps/backend/pkg/codex/types.go +++ b/apps/backend/pkg/codex/types.go @@ -78,6 +78,9 @@ const ( NotifyAccountUpdated = "account/updated" NotifyAccountLoginCompleted = "account/login/completed" NotifyError = "error" + NotifyTokenCount = "token_count" + NotifyThreadTokenUsageUpdated = "thread/tokenUsage/updated" + NotifyContextCompacted = "context_compacted" ) // InitializeParams for initialize request @@ -340,3 +343,69 @@ type ErrorParams struct { Message string `json:"message"` Data any `json:"data,omitempty"` } + +// TokenCountParams for token_count notification +// This is emitted by Codex after each turn to report token usage. +type TokenCountParams struct { + Info *TokenUsageInfo `json:"info,omitempty"` + RateLimits *RateLimitSnapshot `json:"rateLimits,omitempty"` +} + +// TokenUsageInfo contains detailed token usage information. +type TokenUsageInfo struct { + TotalTokenUsage *TokenUsage `json:"totalTokenUsage,omitempty"` + LastTokenUsage *TokenUsage `json:"lastTokenUsage,omitempty"` + ModelContextWindow *int64 `json:"modelContextWindow,omitempty"` +} + +// TokenUsage contains token counts for a request/response cycle. +type TokenUsage struct { + InputTokens int32 `json:"inputTokens"` + CachedInputTokens int32 `json:"cachedInputTokens"` + OutputTokens int32 `json:"outputTokens"` + ReasoningOutputTokens int32 `json:"reasoningOutputTokens"` + TotalTokens int32 `json:"totalTokens"` +} + +// ThreadTokenUsageUpdatedParams for thread/tokenUsage/updated notification. +// This is emitted by Codex after each turn completes with token usage info. +type ThreadTokenUsageUpdatedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` + TokenUsage *ThreadTokenUsage `json:"tokenUsage"` +} + +// ThreadTokenUsage contains the token usage summary for a thread. +type ThreadTokenUsage struct { + Total *TokenUsage `json:"total,omitempty"` + Last *TokenUsage `json:"last,omitempty"` + ModelContextWindow int64 `json:"modelContextWindow"` +} + +// RateLimitSnapshot contains rate limit information. +type RateLimitSnapshot struct { + Primary *RateLimitWindow `json:"primary,omitempty"` + Secondary *RateLimitWindow `json:"secondary,omitempty"` + Credits *CreditsSnapshot `json:"credits,omitempty"` + PlanType *string `json:"planType,omitempty"` +} + +// RateLimitWindow contains rate limit window information. +type RateLimitWindow struct { + UsedPercent int32 `json:"usedPercent"` + WindowDurationMins *int64 `json:"windowDurationMins,omitempty"` + ResetsAt *int64 `json:"resetsAt,omitempty"` +} + +// CreditsSnapshot contains credits information (placeholder for future use). +type CreditsSnapshot struct { + Remaining *int64 `json:"remaining,omitempty"` + Used *int64 `json:"used,omitempty"` +} + +// ContextCompactedParams for context_compacted notification. +// Emitted when the context has been compacted due to reaching limits. +type ContextCompactedParams struct { + ThreadID string `json:"threadId"` + TurnID string `json:"turnId"` +} diff --git a/apps/web/components/task/chat/token-usage-display.tsx b/apps/web/components/task/chat/token-usage-display.tsx new file mode 100644 index 0000000000..d65deb2117 --- /dev/null +++ b/apps/web/components/task/chat/token-usage-display.tsx @@ -0,0 +1,60 @@ +'use client'; + +import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip'; +import { cn } from '@/lib/utils'; +import type { ContextWindowEntry } from '@/lib/state/store'; + +type TokenUsageDisplayProps = { + contextWindow: ContextWindowEntry; + className?: string; +}; + +function formatNumber(num: number): string { + if (num >= 1_000_000) { + return `${(num / 1_000_000).toFixed(1)}M`; + } + if (num >= 1_000) { + return `${(num / 1_000).toFixed(1)}K`; + } + return num.toLocaleString(); +} + +function getBarColor(efficiency: number): string { + if (efficiency >= 90) return 'bg-red-500'; + if (efficiency >= 75) return 'bg-orange-500'; + if (efficiency >= 50) return 'bg-yellow-500'; + return 'bg-emerald-500'; +} + +export function TokenUsageDisplay({ contextWindow, className }: TokenUsageDisplayProps) { + const { size, used, remaining, efficiency } = contextWindow; + + if (size === 0) return null; + + const usagePercent = Math.min(efficiency, 100); + + return ( + + +
+
+
+
+ + {efficiency.toFixed(0)}% + +
+ + +
+
Context: {formatNumber(used)} / {formatNumber(size)}
+
{formatNumber(remaining)} remaining
+
+
+ + ); +} + diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index 3cc5a6e46b..0fe686e9bd 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -27,8 +27,10 @@ import { useSessionMessages } from '@/hooks/use-session-messages'; import { useSettingsData } from '@/hooks/use-settings-data'; import { MessageRenderer } from '@/components/task/chat/message-renderer'; import { RunningIndicator } from '@/components/task/chat/messages/running-indicator'; +import { TokenUsageDisplay } from '@/components/task/chat/token-usage-display'; import { TodoSummary } from '@/components/task/chat/todo-summary'; import { SessionsDropdown } from '@/components/task/sessions-dropdown'; +import { useSessionContextWindow } from '@/hooks/use-session-context-window'; type AgentOption = { id: string; @@ -82,6 +84,9 @@ export function TaskChatPanel({ const sessionModel = sessionProfile?.model ?? null; + // Fetch context window usage for this session + const contextWindow = useSessionContextWindow(resolvedSessionId); + // Fetch messages for this session const { messages, isLoading: messagesLoading } = useSessionMessages(resolvedSessionId); const isInitialLoading = messagesLoading && messages.length === 0; @@ -338,7 +343,7 @@ export function TaskChatPanel({ {/* Session info - shows agent state and model */} {session?.state && ( -
+
{/* Model indicator */} {sessionModel && ( @@ -420,6 +425,8 @@ export function TaskChatPanel({ Toggle plan mode + {/* Context window usage indicator */} + {contextWindow && }
+ sessionId ? state.contextWindow.bySessionId[sessionId]?.size : undefined + ); + const used = useAppStore((state) => + sessionId ? state.contextWindow.bySessionId[sessionId]?.used : undefined + ); + const remaining = useAppStore((state) => + sessionId ? state.contextWindow.bySessionId[sessionId]?.remaining : undefined + ); + const efficiency = useAppStore((state) => + sessionId ? state.contextWindow.bySessionId[sessionId]?.efficiency : undefined + ); + const timestamp = useAppStore((state) => + sessionId ? state.contextWindow.bySessionId[sessionId]?.timestamp : undefined + ); + + // Memoize the combined object + const contextWindow = useMemo(() => { + if (size === undefined) return undefined; + return { size, used: used ?? 0, remaining: remaining ?? 0, efficiency: efficiency ?? 0, timestamp }; + }, [size, used, remaining, efficiency, timestamp]); + + const session = useAppStore((state) => + sessionId ? state.taskSessions.items[sessionId] : undefined + ); + const setContextWindow = useAppStore((state) => state.setContextWindow); + + // Populate context window from session metadata if not already in store + useEffect(() => { + if (!sessionId || contextWindow) return; + + // Try to extract context_window from session metadata + const metadata = session?.metadata; + if (!metadata || typeof metadata !== 'object') return; + + const storedContextWindow = (metadata as Record).context_window; + if (!storedContextWindow || typeof storedContextWindow !== 'object') return; + + // Map stored context window to ContextWindowEntry + const cw = storedContextWindow as Record; + const entry: ContextWindowEntry = { + size: (cw.size as number) ?? 0, + used: (cw.used as number) ?? 0, + remaining: (cw.remaining as number) ?? 0, + efficiency: (cw.efficiency as number) ?? 0, + timestamp: (cw.timestamp as string) ?? undefined, + }; + + setContextWindow(sessionId, entry); + }, [sessionId, contextWindow, session?.metadata, setContextWindow]); + + // Subscribe to session updates via WebSocket + useEffect(() => { + if (!sessionId) return; + const client = getWebSocketClient(); + if (client) { + const unsubscribe = client.subscribeSession(sessionId); + return () => { + unsubscribe(); + // Don't clear context window on cleanup - keep it cached + }; + } + }, [sessionId]); + + return contextWindow; +} + diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 7e4d582ce7..9aae81a379 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -189,6 +189,18 @@ export type GitStatusState = { bySessionId: Record; }; +export type ContextWindowEntry = { + size: number; + used: number; + remaining: number; + efficiency: number; + timestamp?: string; +}; + +export type ContextWindowState = { + bySessionId: Record; +}; + export type Worktree = { id: string; sessionId: string; @@ -270,6 +282,7 @@ export type AppState = { shell: ShellState; diffs: DiffState; gitStatus: GitStatusState; + contextWindow: ContextWindowState; connection: ConnectionState; messages: MessagesState; taskSessions: TaskSessionsState; @@ -337,6 +350,7 @@ export type AppState = { setSessionWorktrees: (sessionId: string, worktreeIds: string[]) => void; setGitStatus: (sessionId: string, gitStatus: GitStatusEntry) => void; clearGitStatus: (sessionId: string) => void; + setContextWindow: (sessionId: string, contextWindow: ContextWindowEntry) => void; bumpAgentProfilesVersion: () => void; }; @@ -380,6 +394,7 @@ const defaultState: AppState = { shell: { outputs: {}, statuses: {} }, diffs: { files: [] }, gitStatus: { bySessionId: {} }, + contextWindow: { bySessionId: {} }, connection: { status: 'disconnected', error: null }, messages: { bySession: {}, metaBySession: {} }, taskSessions: { items: {} }, @@ -433,6 +448,7 @@ const defaultState: AppState = { setSessionWorktrees: () => undefined, setGitStatus: () => undefined, clearGitStatus: () => undefined, + setContextWindow: () => undefined, bumpAgentProfilesVersion: () => undefined, }; @@ -486,6 +502,7 @@ function mergeInitialState( | 'setSessionWorktrees' | 'setGitStatus' | 'clearGitStatus' + | 'setContextWindow' | 'bumpAgentProfilesVersion' > { if (!initialState) return defaultState; @@ -515,6 +532,7 @@ function mergeInitialState( shell: { ...defaultState.shell, ...initialState.shell }, diffs: { ...defaultState.diffs, ...initialState.diffs }, gitStatus: { ...defaultState.gitStatus, ...initialState.gitStatus }, + contextWindow: { ...defaultState.contextWindow, ...initialState.contextWindow }, connection: { ...defaultState.connection, ...initialState.connection }, messages: { ...defaultState.messages, ...initialState.messages }, taskSessions: { ...defaultState.taskSessions, ...initialState.taskSessions }, @@ -883,6 +901,10 @@ export function createAppStore(initialState?: Partial) { set((draft) => { delete draft.gitStatus.bySessionId[sessionId]; }), + setContextWindow: (sessionId, contextWindow) => + set((draft) => { + draft.contextWindow.bySessionId[sessionId] = contextWindow; + }), bumpAgentProfilesVersion: () => set((draft) => { draft.agentProfiles.version += 1; diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index b6d046e238..92c88d0ae7 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -155,6 +155,7 @@ export type TaskSessionStateChangedPayload = { session_id: string; old_state?: string; new_state: string; + metadata?: Record; }; export type TaskSessionWaitingForInputPayload = { diff --git a/apps/web/lib/ws/handlers/agent-session.ts b/apps/web/lib/ws/handlers/agent-session.ts index eb73ad5116..a4d23cd0f5 100644 --- a/apps/web/lib/ws/handlers/agent-session.ts +++ b/apps/web/lib/ws/handlers/agent-session.ts @@ -54,6 +54,22 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle store.getState().setTaskSessionsForTask(taskId, nextSessions); } } + + // Extract context window data from metadata if present + const metadata = payload.metadata; + if (metadata && typeof metadata === 'object') { + const contextWindow = (metadata as Record).context_window; + if (contextWindow && typeof contextWindow === 'object') { + const cw = contextWindow as Record; + store.getState().setContextWindow(sessionId, { + size: (cw.size as number) ?? 0, + used: (cw.used as number) ?? 0, + remaining: (cw.remaining as number) ?? 0, + efficiency: (cw.efficiency as number) ?? 0, + timestamp: new Date().toISOString(), + }); + } + } } }, 'session.agentctl_starting': (message) => {