diff --git a/cmd/gateway.go b/cmd/gateway.go index f72d996c..f337ff34 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -242,6 +242,9 @@ func runGateway() { if meta.OriginLocalKey != "" { batchMeta["origin_local_key"] = meta.OriginLocalKey } + if meta.OriginSessionKey != "" { + batchMeta["origin_session_key"] = meta.OriginSessionKey + } msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: senderID, @@ -617,6 +620,9 @@ func runGateway() { server.SetBuiltinToolsHandler(builtinToolsH) } + // Workspace file serving endpoint + server.SetFilesHandler(httpapi.NewFilesHandler(workspace, cfg.Gateway.Token)) + // Seed + apply builtin tool disables if pgStores.BuiltinTools != nil { seedBuiltinTools(context.Background(), pgStores.BuiltinTools) diff --git a/cmd/gateway_consumer.go b/cmd/gateway_consumer.go index 0e960c31..7ebbf735 100644 --- a/cmd/gateway_consumer.go +++ b/cmd/gateway_consumer.go @@ -5,6 +5,8 @@ import ( "errors" "fmt" "log/slog" + "mime" + "path/filepath" "strings" "sync" "time" @@ -390,11 +392,13 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents continue } - // Use SAME session as user's original chat so agent has context. - sessionKey := sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) - - // Override session key for forum topics / DM threads (same logic as main lane). - sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + // Use exact origin session key if available (WS uses non-standard format). + sessionKey := msg.Metadata["origin_session_key"] + if sessionKey == "" { + // Fallback: rebuild session key from origin metadata (works for Telegram, Discord, etc.) + sessionKey = sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + } slog.Info("subagent announce → scheduler (subagent lane)", "subagent", msg.SenderID, @@ -421,10 +425,25 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents outMeta := buildAnnounceOutMeta(origLocalKey) // Build request before goroutine to capture msg fields. + // WS: convert media to markdown URLs inline (no outbound channel handler). + // Must happen BEFORE the run so the agent loop saves images in the session + // before run.completed fires — avoids race with frontend loadHistory(). + fwdMedia := msg.Media + announceMessage := msg.Content + wsMediaSuffix := "" + if origChannel == "ws" && len(msg.Media) > 0 { + mdMedia := mediaToMarkdownFromPaths(msg.Media, cfg.WorkspacePath(), cfg) + if mdMedia != "" { + announceMessage += "\n\n[Image URLs for web display — include these EXACTLY as-is at the end of your response]" + mdMedia + wsMediaSuffix = mdMedia + } + fwdMedia = nil // already embedded as markdown + } + announceReq := agent.RunRequest{ SessionKey: sessionKey, - Message: msg.Content, - ForwardMedia: msg.Media, + Message: announceMessage, + ForwardMedia: fwdMedia, Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, @@ -432,15 +451,15 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents UserID: announceUserID, RunID: fmt.Sprintf("announce-%s", msg.SenderID), RunKind: "announce", + HideInput: true, // don't persist raw system message in chat history Stream: false, ParentTraceID: parentTraceID, ParentRootSpanID: parentRootSpanID, } - // Handle announce asynchronously with per-session serialization. // The mutex ensures concurrent announces for the same session wait for // each other, so each reads up-to-date session history. - go func(sessionKey, origCh, chatID, senderID, label string, meta map[string]string, req agent.RunRequest) { + go func(sessionKey, origCh, chatID, senderID, label string, meta map[string]string, req agent.RunRequest, mediaSuffix string) { mu := getAnnounceMu(sessionKey) mu.Lock() defer mu.Unlock() @@ -477,6 +496,12 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if isSilent { announceContent = "" // suppress NO_REPLY text but still send media } + + // Fallback: if LLM stripped the image markdown URLs, append them. + if mediaSuffix != "" && !strings.Contains(announceContent, mediaSuffix) { + announceContent += mediaSuffix + } + outMsg := bus.OutboundMessage{ Channel: origCh, ChatID: chatID, @@ -490,7 +515,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(sessionKey, origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"], outMeta, announceReq) + }(sessionKey, origChannel, msg.ChatID, msg.SenderID, msg.Metadata["subagent_label"], outMeta, announceReq, wsMediaSuffix) continue } @@ -513,10 +538,13 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents continue } - sessionKey := sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) - - // Override session key for forum topics / DM threads. - sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + // Use exact origin session key if available (WS uses non-standard format). + sessionKey := msg.Metadata["origin_session_key"] + if sessionKey == "" { + // Fallback: rebuild session key from origin metadata (works for Telegram, Discord, etc.) + sessionKey = sessions.BuildScopedSessionKey(parentAgent, origChannel, sessions.PeerKind(origPeerKind), msg.ChatID, cfg.Sessions.Scope, cfg.Sessions.DmScope, cfg.Sessions.MainKey) + sessionKey = overrideSessionKeyFromLocalKey(sessionKey, origLocalKey, parentAgent, origChannel, msg.ChatID, origPeerKind) + } slog.Info("delegate announce → scheduler (delegate lane)", "delegation", msg.SenderID, @@ -541,10 +569,23 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents // Build outbound metadata for topic/thread routing. outMeta := buildAnnounceOutMeta(origLocalKey) + // WS: convert media to markdown URLs inline (same as subagent announce). + fwdMedia := msg.Media + announceMessage := msg.Content + wsMediaSuffix := "" + if origChannel == "ws" && len(msg.Media) > 0 { + mdMedia := mediaToMarkdownFromPaths(msg.Media, cfg.WorkspacePath(), cfg) + if mdMedia != "" { + announceMessage += "\n\n[Image URLs for web display — include these EXACTLY as-is at the end of your response]" + mdMedia + wsMediaSuffix = mdMedia + } + fwdMedia = nil + } + announceReq := agent.RunRequest{ SessionKey: sessionKey, - Message: msg.Content, - ForwardMedia: msg.Media, + Message: announceMessage, + ForwardMedia: fwdMedia, Channel: origChannel, ChatID: msg.ChatID, PeerKind: origPeerKind, @@ -552,13 +593,14 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents UserID: announceUserID, RunID: fmt.Sprintf("delegate-announce-%s", msg.Metadata["delegation_id"]), RunKind: "announce", + HideInput: true, // don't persist raw system message in chat history Stream: false, ParentTraceID: parentTraceID, ParentRootSpanID: parentRootSpanID, } // Same per-session serialization as subagent announce above. - go func(sessionKey, origCh, chatID, senderID string, meta map[string]string, req agent.RunRequest) { + go func(sessionKey, origCh, chatID, senderID string, meta map[string]string, req agent.RunRequest, mediaSuffix string) { mu := getAnnounceMu(sessionKey) mu.Lock() defer mu.Unlock() @@ -588,6 +630,12 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents if isSilent { announceContent = "" // suppress NO_REPLY text but still send media } + + // Fallback: if LLM stripped the image markdown URLs, append them. + if mediaSuffix != "" && !strings.Contains(announceContent, mediaSuffix) { + announceContent += mediaSuffix + } + outMsg := bus.OutboundMessage{ Channel: origCh, ChatID: chatID, @@ -601,7 +649,7 @@ func consumeInboundMessages(ctx context.Context, msgBus *bus.MessageBus, agents }) } msgBus.PublishOutbound(outMsg) - }(sessionKey, origChannel, msg.ChatID, msg.SenderID, outMeta, announceReq) + }(sessionKey, origChannel, msg.ChatID, msg.SenderID, outMeta, announceReq, wsMediaSuffix) continue } @@ -862,6 +910,69 @@ func overrideSessionKeyFromLocalKey(sessionKey, localKey, agentID, channel, chat // buildAnnounceOutMeta builds outbound metadata for announce messages so that // Send() can route replies to the correct forum topic or DM thread. +// mediaToMarkdown converts media results to markdown image/link syntax using the +// /v1/files/ HTTP endpoint. Used for WS channel where outbound media attachments +// are not supported (no channel handler). Returns empty string if no media. +func mediaToMarkdown(media []agent.MediaResult, workspace string, cfg *config.Config) string { + if len(media) == 0 { + return "" + } + + scheme := "http" + if cfg.Tailscale.EnableTLS { + scheme = "https" + } + host := cfg.Gateway.Host + if host == "" || host == "0.0.0.0" { + host = "localhost" + } + baseURL := fmt.Sprintf("%s://%s:%d/v1/files", scheme, host, cfg.Gateway.Port) + tokenQuery := "" + if cfg.Gateway.Token != "" { + tokenQuery = "?token=" + cfg.Gateway.Token + } + + cleanRoot := filepath.Clean(workspace) + var parts []string + for _, mr := range media { + cleanPath := filepath.Clean(mr.Path) + relPath, err := filepath.Rel(cleanRoot, cleanPath) + if err != nil || strings.HasPrefix(relPath, "..") { + continue + } + fileURL := baseURL + "/" + relPath + tokenQuery + if strings.HasPrefix(mr.ContentType, "image/") { + parts = append(parts, fmt.Sprintf("![image](%s)", fileURL)) + } else { + parts = append(parts, fmt.Sprintf("[%s](%s)", filepath.Base(mr.Path), fileURL)) + } + } + if len(parts) == 0 { + return "" + } + return "\n\n" + strings.Join(parts, "\n") +} + +// mediaToMarkdownFromPaths is like mediaToMarkdown but accepts raw file paths +// ([]string from bus.InboundMessage.Media) instead of []agent.MediaResult. +func mediaToMarkdownFromPaths(paths []string, workspace string, cfg *config.Config) string { + if len(paths) == 0 { + return "" + } + media := make([]agent.MediaResult, 0, len(paths)) + for _, p := range paths { + ct := mime.TypeByExtension(filepath.Ext(p)) + if ct == "" { + ct = "application/octet-stream" + } + media = append(media, agent.MediaResult{ + Path: p, + ContentType: ct, + }) + } + return mediaToMarkdown(media, workspace, cfg) +} + func buildAnnounceOutMeta(localKey string) map[string]string { if localKey == "" { return nil diff --git a/internal/agent/loop.go b/internal/agent/loop.go index 9e986582..868386eb 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -355,10 +355,12 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) // This prevents concurrent runs from seeing each other's in-progress messages. // NOTE: pendingMsgs stores TEXT ONLY (no images) to avoid bloating session storage. var pendingMsgs []providers.Message - pendingMsgs = append(pendingMsgs, providers.Message{ - Role: "user", - Content: req.Message, - }) + if !req.HideInput { + pendingMsgs = append(pendingMsgs, providers.Message{ + Role: "user", + Content: req.Message, + }) + } // 4. Run LLM iteration loop var loopDetector toolLoopState // detects repeated no-progress tool calls @@ -668,15 +670,19 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } + toolResultPayload := map[string]interface{}{ + "name": tc.Name, + "id": tc.ID, + "is_error": result.IsError, + } + if result.IsError && result.ForLLM != "" { + toolResultPayload["content"] = result.ForLLM + } emitRun(AgentEvent{ Type: protocol.AgentEventToolResult, AgentID: l.id, RunID: req.RunID, - Payload: map[string]interface{}{ - "name": tc.Name, - "id": tc.ID, - "is_error": result.IsError, - }, + Payload: toolResultPayload, }) l.scanWebToolResult(tc.Name, result) @@ -797,15 +803,19 @@ func (l *Loop) runLoop(ctx context.Context, req RunRequest) (*RunResult, error) } } + parToolResultPayload := map[string]interface{}{ + "name": r.tc.Name, + "id": r.tc.ID, + "is_error": r.result.IsError, + } + if r.result.IsError && r.result.ForLLM != "" { + parToolResultPayload["content"] = r.result.ForLLM + } emitRun(AgentEvent{ Type: protocol.AgentEventToolResult, AgentID: l.id, RunID: req.RunID, - Payload: map[string]interface{}{ - "name": r.tc.Name, - "id": r.tc.ID, - "is_error": r.result.IsError, - }, + Payload: parToolResultPayload, }) l.scanWebToolResult(r.tc.Name, r.result) diff --git a/internal/agent/loop_types.go b/internal/agent/loop_types.go index 65991b2d..14718d87 100644 --- a/internal/agent/loop_types.go +++ b/internal/agent/loop_types.go @@ -276,7 +276,8 @@ type RunRequest struct { MaxIterations int // per-request override (0 = use agent default, must be lower) // Run classification - RunKind string // "delegation", "announce" — empty for user-initiated runs + RunKind string // "delegation", "announce" — empty for user-initiated runs + HideInput bool // don't persist input message in session history (announce runs) // Delegation context (set when running as a delegate agent) DelegationID string // delegation ID for event correlation diff --git a/internal/gateway/server.go b/internal/gateway/server.go index d169a09f..858b37d3 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -45,6 +45,7 @@ type Server struct { delegationsHandler *httpapi.DelegationsHandler // delegation history API builtinToolsHandler *httpapi.BuiltinToolsHandler // builtin tool management API oauthHandler *httpapi.OAuthHandler // OAuth endpoints + filesHandler *httpapi.FilesHandler // workspace file serving agentStore store.AgentStore // for context injection in tools_invoke upgrader websocket.Upgrader @@ -195,6 +196,11 @@ func (s *Server) BuildMux() *http.ServeMux { s.builtinToolsHandler.RegisterRoutes(mux) } + // Workspace file serving (available in all modes) + if s.filesHandler != nil { + s.filesHandler.RegisterRoutes(mux) + } + // OAuth endpoints (available in all modes) if s.oauthHandler != nil { s.oauthHandler.RegisterRoutes(mux) @@ -338,6 +344,9 @@ func (s *Server) SetBuiltinToolsHandler(h *httpapi.BuiltinToolsHandler) { // SetOAuthHandler sets the OAuth handler (available in all modes). func (s *Server) SetOAuthHandler(h *httpapi.OAuthHandler) { s.oauthHandler = h } +// SetFilesHandler sets the workspace file serving handler. +func (s *Server) SetFilesHandler(h *httpapi.FilesHandler) { s.filesHandler = h } + // SetAgentStore sets the agent store for context injection in tools_invoke. func (s *Server) SetAgentStore(as store.AgentStore) { s.agentStore = as } diff --git a/internal/http/files.go b/internal/http/files.go new file mode 100644 index 00000000..4f84b781 --- /dev/null +++ b/internal/http/files.go @@ -0,0 +1,84 @@ +package http + +import ( + "log/slog" + "mime" + "net/http" + "os" + "path/filepath" + "strings" +) + +// FilesHandler serves workspace files over HTTP with Bearer token auth. +type FilesHandler struct { + workspaceRoot string + token string +} + +// NewFilesHandler creates a handler that serves files from the workspace root. +func NewFilesHandler(workspaceRoot, token string) *FilesHandler { + return &FilesHandler{workspaceRoot: workspaceRoot, token: token} +} + +// RegisterRoutes registers the file serving route. +func (h *FilesHandler) RegisterRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /v1/files/{path...}", h.auth(h.handleServe)) +} + +func (h *FilesHandler) auth(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if h.token != "" { + // Accept token via Bearer header or ?token= query param (for ). + provided := extractBearerToken(r) + if provided == "" { + provided = r.URL.Query().Get("token") + } + if provided != h.token { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + } + next(w, r) + } +} + +func (h *FilesHandler) handleServe(w http.ResponseWriter, r *http.Request) { + relPath := r.PathValue("path") + if relPath == "" { + http.Error(w, "path is required", http.StatusBadRequest) + return + } + + // Prevent path traversal + if strings.Contains(relPath, "..") { + slog.Warn("security.files_traversal", "path", relPath) + http.Error(w, "invalid path", http.StatusBadRequest) + return + } + + absPath := filepath.Join(h.workspaceRoot, relPath) + + // Double-check the resolved path is within the workspace + cleanAbs := filepath.Clean(absPath) + cleanRoot := filepath.Clean(h.workspaceRoot) + if !strings.HasPrefix(cleanAbs, cleanRoot+string(filepath.Separator)) && cleanAbs != cleanRoot { + slog.Warn("security.files_escape", "resolved", cleanAbs, "root", cleanRoot) + http.Error(w, "invalid path", http.StatusBadRequest) + return + } + + info, err := os.Stat(cleanAbs) + if err != nil || info.IsDir() { + http.NotFound(w, r) + return + } + + // Set Content-Type from extension + ext := filepath.Ext(cleanAbs) + ct := mime.TypeByExtension(ext) + if ct != "" { + w.Header().Set("Content-Type", ct) + } + + http.ServeFile(w, r, cleanAbs) +} diff --git a/internal/tools/announce_queue.go b/internal/tools/announce_queue.go index cc455b09..5bc95704 100644 --- a/internal/tools/announce_queue.go +++ b/internal/tools/announce_queue.go @@ -24,6 +24,7 @@ type AnnounceMetadata struct { OriginPeerKind string OriginLocalKey string // composite key with topic/thread suffix for routing OriginUserID string + OriginSessionKey string // exact parent session key (WS uses non-standard format) ParentAgent string OriginTraceID string // parent trace UUID for announce linking OriginRootSpanID string // parent agent's root span UUID diff --git a/internal/tools/context_keys.go b/internal/tools/context_keys.go index 5a76b882..d8f0d7b7 100644 --- a/internal/tools/context_keys.go +++ b/internal/tools/context_keys.go @@ -21,7 +21,8 @@ const ( ctxSandboxKey toolContextKey = "tool_sandbox_key" ctxAsyncCB toolContextKey = "tool_async_cb" ctxWorkspace toolContextKey = "tool_workspace" - ctxAgentKey toolContextKey = "tool_agent_key" + ctxAgentKey toolContextKey = "tool_agent_key" + ctxSessionKey toolContextKey = "tool_session_key" // origin session key for announce routing ) func WithToolChannel(ctx context.Context, channel string) context.Context { @@ -101,6 +102,18 @@ func ToolAgentKeyFromCtx(ctx context.Context) string { return v } +// WithToolSessionKey injects the parent's session key so subagent announce +// can route results back to the exact same session (required for WS where +// session keys don't follow BuildScopedSessionKey format). +func WithToolSessionKey(ctx context.Context, key string) context.Context { + return context.WithValue(ctx, ctxSessionKey, key) +} + +func ToolSessionKeyFromCtx(ctx context.Context) string { + v, _ := ctx.Value(ctxSessionKey).(string) + return v +} + // --- Vision / ImageGen config (per-agent overrides) --- const ( diff --git a/internal/tools/delegate.go b/internal/tools/delegate.go index 74b015cb..ecc5c460 100644 --- a/internal/tools/delegate.go +++ b/internal/tools/delegate.go @@ -33,10 +33,11 @@ type DelegationTask struct { CompletedAt *time.Time `json:"completed_at,omitempty"` // Origin metadata for async announce routing - OriginChannel string `json:"-"` - OriginChatID string `json:"-"` - OriginPeerKind string `json:"-"` - OriginLocalKey string `json:"-"` // composite key with topic/thread suffix for routing + OriginChannel string `json:"-"` + OriginChatID string `json:"-"` + OriginPeerKind string `json:"-"` + OriginLocalKey string `json:"-"` // composite key with topic/thread suffix for routing + OriginSessionKey string `json:"-"` // exact parent session key for announce routing (WS uses non-standard format) // Trace context for announce linking (same pattern as SubagentTask) OriginTraceID uuid.UUID `json:"-"` diff --git a/internal/tools/delegate_async.go b/internal/tools/delegate_async.go index 6a53d55d..e430b8d2 100644 --- a/internal/tools/delegate_async.go +++ b/internal/tools/delegate_async.go @@ -161,6 +161,9 @@ func (dm *DelegateManager) DelegateAsync(ctx context.Context, opts DelegateOpts) if task.OriginLocalKey != "" { announceMeta["origin_local_key"] = task.OriginLocalKey } + if task.OriginSessionKey != "" { + announceMeta["origin_session_key"] = task.OriginSessionKey + } // Emit announce event so WS clients know all results are being sent to lead. hasMedia := len(artifacts.Media) > 0 var announceSummaries []protocol.DelegationAnnounceResultSummary diff --git a/internal/tools/delegate_prep.go b/internal/tools/delegate_prep.go index 4484a93c..ff4cf6c2 100644 --- a/internal/tools/delegate_prep.go +++ b/internal/tools/delegate_prep.go @@ -174,6 +174,7 @@ func (dm *DelegateManager) prepareDelegation(ctx context.Context, opts DelegateO OriginChatID: chatID, OriginPeerKind: peerKind, OriginLocalKey: localKey, + OriginSessionKey: ToolSessionKeyFromCtx(ctx), OriginTraceID: tracing.TraceIDFromContext(ctx), OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), TeamTaskID: opts.TeamTaskID, diff --git a/internal/tools/registry.go b/internal/tools/registry.go index 3cadd993..09f4bc10 100644 --- a/internal/tools/registry.go +++ b/internal/tools/registry.go @@ -88,6 +88,7 @@ func (r *Registry) ExecuteWithContext(ctx context.Context, name string, args map } if sessionKey != "" { ctx = WithToolSandboxKey(ctx, sessionKey) + ctx = WithToolSessionKey(ctx, sessionKey) } if asyncCB != nil { ctx = WithToolAsyncCB(ctx, asyncCB) diff --git a/internal/tools/subagent.go b/internal/tools/subagent.go index cdcd2549..3b691cec 100644 --- a/internal/tools/subagent.go +++ b/internal/tools/subagent.go @@ -51,11 +51,12 @@ type SubagentTask struct { Result string `json:"result,omitempty"` Depth int `json:"depth"` Model string `json:"model,omitempty"` // model override for this subagent - OriginChannel string `json:"originChannel,omitempty"` - OriginChatID string `json:"originChatId,omitempty"` - OriginPeerKind string `json:"originPeerKind,omitempty"` // "direct" or "group" (for session key building) - OriginLocalKey string `json:"originLocalKey,omitempty"` // composite key with topic/thread suffix for routing - OriginUserID string `json:"originUserId,omitempty"` // parent's userID for per-user scoping propagation + OriginChannel string `json:"originChannel,omitempty"` + OriginChatID string `json:"originChatId,omitempty"` + OriginPeerKind string `json:"originPeerKind,omitempty"` // "direct" or "group" (for session key building) + OriginLocalKey string `json:"originLocalKey,omitempty"` // composite key with topic/thread suffix for routing + OriginUserID string `json:"originUserId,omitempty"` // parent's userID for per-user scoping propagation + OriginSessionKey string `json:"originSessionKey,omitempty"` // exact parent session key for announce routing (WS uses non-standard format) CreatedAt int64 `json:"createdAt"` CompletedAt int64 `json:"completedAt,omitempty"` OriginTraceID uuid.UUID `json:"-"` // parent trace for announce linking @@ -193,11 +194,12 @@ func (sm *SubagentManager) Spawn( OriginChannel: channel, OriginChatID: chatID, OriginPeerKind: peerKind, - OriginLocalKey: ToolLocalKeyFromCtx(ctx), - OriginUserID: store.UserIDFromContext(ctx), - OriginTraceID: tracing.TraceIDFromContext(ctx), - OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), - CreatedAt: time.Now().UnixMilli(), + OriginLocalKey: ToolLocalKeyFromCtx(ctx), + OriginUserID: store.UserIDFromContext(ctx), + OriginSessionKey: ToolSessionKeyFromCtx(ctx), + OriginTraceID: tracing.TraceIDFromContext(ctx), + OriginRootSpanID: tracing.ParentSpanIDFromContext(ctx), + CreatedAt: time.Now().UnixMilli(), } // Create per-task context for real goroutine cancellation taskCtx, taskCancel := context.WithCancel(ctx) diff --git a/internal/tools/subagent_exec.go b/internal/tools/subagent_exec.go index 5911ebaf..2c4eba9a 100644 --- a/internal/tools/subagent_exec.go +++ b/internal/tools/subagent_exec.go @@ -37,6 +37,7 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call OriginPeerKind: task.OriginPeerKind, OriginLocalKey: task.OriginLocalKey, OriginUserID: task.OriginUserID, + OriginSessionKey: task.OriginSessionKey, ParentAgent: task.ParentID, OriginTraceID: task.OriginTraceID.String(), OriginRootSpanID: task.OriginRootSpanID.String(), @@ -63,6 +64,9 @@ func (sm *SubagentManager) runTask(ctx context.Context, task *SubagentTask, call if task.OriginLocalKey != "" { announceMeta["origin_local_key"] = task.OriginLocalKey } + if task.OriginSessionKey != "" { + announceMeta["origin_session_key"] = task.OriginSessionKey + } sm.msgBus.PublishInbound(bus.InboundMessage{ Channel: "system", SenderID: fmt.Sprintf("subagent:%s", task.ID), diff --git a/ui/web/src/components/chat/message-bubble.tsx b/ui/web/src/components/chat/message-bubble.tsx index 749b877a..eb3bdb6c 100644 --- a/ui/web/src/components/chat/message-bubble.tsx +++ b/ui/web/src/components/chat/message-bubble.tsx @@ -1,14 +1,15 @@ import { useState } from "react"; -import { Bot, User, Wrench, ChevronRight } from "lucide-react"; +import { Bot, User, ChevronRight, Check, AlertTriangle } from "lucide-react"; import { MessageContent } from "./message-content"; import type { ChatMessage } from "@/types/chat"; import type { ToolCall } from "@/types/session"; interface MessageBubbleProps { message: ChatMessage; + toolCallErrors?: Map; } -export function MessageBubble({ message }: MessageBubbleProps) { +export function MessageBubble({ message, toolCallErrors }: MessageBubbleProps) { const isUser = message.role === "user"; const isTool = message.role === "tool"; @@ -17,7 +18,7 @@ export function MessageBubble({ message }: MessageBubbleProps) { } const hasContent = !!message.content?.trim(); - const hasToolCalls = message.toolCalls && message.toolCalls.length > 0; + const hasToolCalls = message.tool_calls && message.tool_calls.length > 0; // Skip assistant messages with neither text nor tool calls if (!isUser && !hasContent && !hasToolCalls) { @@ -28,8 +29,8 @@ export function MessageBubble({ message }: MessageBubbleProps) { if (!isUser && !hasContent && hasToolCalls) { return (
- {message.toolCalls!.map((tc) => ( - + {message.tool_calls!.map((tc) => ( + ))}
); @@ -54,8 +55,8 @@ export function MessageBubble({ message }: MessageBubbleProps) { > {hasToolCalls && (
- {message.toolCalls!.map((tc) => ( - + {message.tool_calls!.map((tc) => ( + ))}
)} @@ -73,37 +74,44 @@ export function MessageBubble({ message }: MessageBubbleProps) { ); } -function ToolCallItem({ toolCall, compact }: { toolCall: ToolCall; compact?: boolean }) { +function ToolCallItem({ toolCall, compact, isError, errorContent }: { toolCall: ToolCall; compact?: boolean; isError?: boolean; errorContent?: string }) { const [expanded, setExpanded] = useState(false); const hasArgs = toolCall.arguments && Object.keys(toolCall.arguments).length > 0; + const canExpand = hasArgs || (isError && !!errorContent); + const iconSize = compact ? "h-3 w-3" : "h-3.5 w-3.5"; + + const StatusIcon = isError + ? + : ; return (
- {expanded && hasArgs && ( - 
-          {JSON.stringify(toolCall.arguments, null, 2)}
-
+ {isError && errorContent && ( + 
{errorContent}
+ )} + {hasArgs && ( + 
{JSON.stringify(toolCall.arguments, null, 2)}
+ )} +
)} ); diff --git a/ui/web/src/components/chat/tool-call-card.tsx b/ui/web/src/components/chat/tool-call-card.tsx index 347ddd38..7d3f2ca3 100644 --- a/ui/web/src/components/chat/tool-call-card.tsx +++ b/ui/web/src/components/chat/tool-call-card.tsx @@ -1,4 +1,5 @@ -import { Wrench, Check, AlertTriangle, Loader2 } from "lucide-react"; +import { useState } from "react"; +import { Wrench, Check, AlertTriangle, Loader2, ChevronRight } from "lucide-react"; import type { ToolStreamEntry } from "@/types/chat"; interface ToolCallCardProps { @@ -6,11 +7,41 @@ interface ToolCallCardProps { } export function ToolCallCard({ entry }: ToolCallCardProps) { + const [expanded, setExpanded] = useState(false); + const hasArgs = entry.arguments && Object.keys(entry.arguments).length > 0; + const hasError = entry.phase === "error" && !!entry.errorContent; + const canExpand = hasArgs || hasError; + return ( -
- - {entry.name} - +
+ + {expanded && canExpand && ( +
+ {hasError && ( + 
{entry.errorContent}
+ )} + {hasArgs && ( + 
{JSON.stringify(entry.arguments, null, 2)}
+ )} +
+ )}
); } diff --git a/ui/web/src/components/shared/markdown-renderer.tsx b/ui/web/src/components/shared/markdown-renderer.tsx index 50a3b98d..ed9c7f0e 100644 --- a/ui/web/src/components/shared/markdown-renderer.tsx +++ b/ui/web/src/components/shared/markdown-renderer.tsx @@ -65,6 +65,19 @@ export function MarkdownRenderer({ content, className }: MarkdownRendererProps) ); }, + img({ src, alt, ...props }) { + return ( + + {alt + + ); + }, table({ children, ...props }) { return (
diff --git a/ui/web/src/pages/chat/chat-thread.tsx b/ui/web/src/pages/chat/chat-thread.tsx index d0c58cfe..e6ef9e29 100644 --- a/ui/web/src/pages/chat/chat-thread.tsx +++ b/ui/web/src/pages/chat/chat-thread.tsx @@ -1,3 +1,4 @@ +import { useMemo } from "react"; import { Circle } from "lucide-react"; import { MessageBubble } from "@/components/chat/message-bubble"; import { StreamingText } from "@/components/chat/streaming-text"; @@ -28,6 +29,19 @@ export function ChatThread({ [messages.length, streamText, thinkingText, toolStream.length], ); + // Build map of tool_call_id → error content for tool results that indicate errors + const toolCallErrors = useMemo(() => { + const errors = new Map(); + for (const msg of messages) { + if (msg.role !== "tool" || !msg.tool_call_id || !msg.content) continue; + const c = msg.content.trimStart(); + if (c.startsWith("Error") || c.startsWith("error:") || c.includes("failed:") || c.includes("Failed:")) { + errors.set(msg.tool_call_id, msg.content); + } + } + return errors; + }, [messages]); + // Show spinner while loading history for a different session if (loading) { return ( @@ -54,7 +68,7 @@ export function ChatThread({ >
{messages.map((msg, i) => ( - + ))} {/* Tool stream during active run */} diff --git a/ui/web/src/pages/chat/hooks/use-chat-messages.ts b/ui/web/src/pages/chat/hooks/use-chat-messages.ts index df78b360..cd3ba07b 100644 --- a/ui/web/src/pages/chat/hooks/use-chat-messages.ts +++ b/ui/web/src/pages/chat/hooks/use-chat-messages.ts @@ -26,6 +26,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { const expectingRunRef = useRef(false); const streamRef = useRef(""); const thinkingRef = useRef(""); + const toolStreamRef = useRef([]); const agentIdRef = useRef(agentId); agentIdRef.current = agentId; @@ -44,6 +45,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { expectingRunRef.current = false; streamRef.current = ""; thinkingRef.current = ""; + toolStreamRef.current = []; } // Load history (no loading spinner — the empty state placeholder is shown instead) @@ -87,6 +89,13 @@ export function useChatMessages(sessionKey: string, agentId: string) { const event = payload as AgentEventPayload; if (!event) return; + // Announce run.completed: reload history when a subagent/delegate announce finishes + // for the current agent (these runs aren't tracked by runIdRef). + if (event.type === "run.completed" && event.runKind === "announce" && event.agentId === agentIdRef.current) { + loadHistory(); + return; + } + // Capture run.started when we are expecting a run for this agent if (event.type === "run.started") { if (expectingRunRef.current && event.agentId === agentIdRef.current) { @@ -98,6 +107,7 @@ export function useChatMessages(sessionKey: string, agentId: string) { setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + toolStreamRef.current = []; } return; } @@ -123,20 +133,24 @@ export function useChatMessages(sessionKey: string, agentId: string) { toolCallId: event.payload?.id ?? "", runId: event.runId, name: event.payload?.name ?? "tool", + arguments: event.payload?.arguments, phase: "calling", startedAt: Date.now(), updatedAt: Date.now(), }; - setToolStream((prev) => [...prev, entry]); + toolStreamRef.current = [...toolStreamRef.current, entry]; + setToolStream(toolStreamRef.current); break; } case "tool.result": { + const isError = event.payload?.is_error; setToolStream((prev) => prev.map((t) => t.toolCallId === event.payload?.id ? { ...t, - phase: event.payload?.is_error ? "error" : "completed", + phase: isError ? "error" : "completed", + errorContent: isError ? event.payload?.content : undefined, updatedAt: Date.now(), } : t, @@ -147,12 +161,28 @@ export function useChatMessages(sessionKey: string, agentId: string) { case "run.completed": { setIsRunning(false); runIdRef.current = null; - loadHistory(); + + // If we have streamed text and no tool calls, promote it directly + // to avoid a flash caused by loadHistory() replacing the DOM. + // When tools were used, reload to get structured tool_calls data. + const hadTools = toolStreamRef.current.length > 0; + const streamed = streamRef.current; + setStreamText(null); setThinkingText(null); setToolStream([]); streamRef.current = ""; thinkingRef.current = ""; + toolStreamRef.current = []; + + if (streamed && !hadTools) { + setMessages((prev) => [ + ...prev, + { role: "assistant", content: streamed, timestamp: Date.now() }, + ]); + } else { + loadHistory(); + } break; } case "run.failed": { diff --git a/ui/web/src/types/chat.ts b/ui/web/src/types/chat.ts index d775bd7d..28684fbb 100644 --- a/ui/web/src/types/chat.ts +++ b/ui/web/src/types/chat.ts @@ -13,12 +13,14 @@ export interface AgentEventPayload { type: string; // "run.started" | "run.completed" | "run.failed" | "chunk" | "tool.call" | "tool.result" agentId: string; runId: string; + runKind?: string; // "delegation" | "announce" — omitted for user-initiated runs payload?: { content?: string; name?: string; id?: string; is_error?: boolean; error?: string; + arguments?: Record; }; } @@ -28,6 +30,8 @@ export interface ToolStreamEntry { runId: string; name: string; phase: "calling" | "completed" | "error"; + arguments?: Record; + errorContent?: string; startedAt: number; updatedAt: number; } diff --git a/ui/web/src/types/session.ts b/ui/web/src/types/session.ts index ca179223..0bcc140a 100644 --- a/ui/web/src/types/session.ts +++ b/ui/web/src/types/session.ts @@ -23,8 +23,8 @@ export interface SessionPreview { export interface Message { role: "user" | "assistant" | "tool"; content: string; - toolCalls?: ToolCall[]; - toolCallId?: string; + tool_calls?: ToolCall[]; + tool_call_id?: string; } export interface ToolCall {