Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions core/http/react-ui/src/App.css
Original file line number Diff line number Diff line change
Expand Up @@ -1936,6 +1936,56 @@
40% { transform: scale(1); opacity: 1; }
}

/* Staging progress indicator (replaces thinking dots during model transfer) */
.chat-staging-progress {
display: flex;
flex-direction: column;
gap: 6px;
min-width: 200px;
max-width: 320px;
}
.chat-staging-label {
font-size: 0.8rem;
color: var(--color-text-secondary);
display: flex;
align-items: center;
gap: 6px;
}
.chat-staging-label i {
color: var(--color-primary);
}
.chat-staging-detail {
display: flex;
align-items: center;
gap: 8px;
}
.chat-staging-bar-container {
flex: 1;
height: 4px;
background: var(--color-bg-tertiary);
border-radius: 2px;
overflow: hidden;
}
.chat-staging-bar {
height: 100%;
background: var(--color-primary);
border-radius: 2px;
transition: width 300ms ease;
}
.chat-staging-pct {
font-size: 0.75rem;
color: var(--color-text-muted);
min-width: 32px;
text-align: right;
}
.chat-staging-file {
font-size: 0.7rem;
color: var(--color-text-muted);
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}

/* Message completion flash */
.chat-message-bubble {
transition: border-color 300ms ease;
Expand Down
5 changes: 5 additions & 0 deletions core/http/react-ui/src/components/OperationsBar.jsx
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,11 @@ export default function OperationsBar() {
({op.error})
</span>
</>
) : op.taskType === 'staging' ? (
<>
<i className="fas fa-cloud-arrow-up" style={{ marginRight: 'var(--spacing-xs)' }} />
Staging model: {op.name}{op.nodeName ? ` → ${op.nodeName}` : ''}
</>
) : (
<>
{op.isDeletion ? 'Removing' : 'Installing'}{' '}
Expand Down
33 changes: 30 additions & 3 deletions core/http/react-ui/src/pages/Chat.jsx
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import UnifiedMCPDropdown from '../components/UnifiedMCPDropdown'
import { loadClientMCPServers } from '../utils/mcpClientStorage'
import ConfirmDialog from '../components/ConfirmDialog'
import { useAuth } from '../context/AuthContext'
import { useOperations } from '../hooks/useOperations'
import { relativeTime } from '../utils/format'

function getLastMessagePreview(chat) {
Expand Down Expand Up @@ -277,13 +278,20 @@ export default function Chat() {
const { addToast } = useOutletContext()
const navigate = useNavigate()
const { isAdmin } = useAuth()
const { operations } = useOperations()
const {
chats, activeChat, activeChatId, isStreaming, streamingChatId, streamingContent,
streamingReasoning, streamingToolCalls, tokensPerSecond, maxTokensPerSecond,
addChat, switchChat, deleteChat, deleteAllChats, renameChat, updateChatSettings,
sendMessage, stopGeneration, clearHistory, getContextUsagePercent, addMessage,
} = useChat(urlModel || '')

// Detect active staging operation for the current chat's model
const stagingOp = useMemo(() => {
if (!isStreaming || !activeChat?.model) return null
return operations.find(op => op.taskType === 'staging' && op.name === activeChat.model) || null
}, [operations, isStreaming, activeChat?.model])

const [input, setInput] = useState('')
const [files, setFiles] = useState([])
const [showSettings, setShowSettings] = useState(false)
Expand Down Expand Up @@ -1187,9 +1195,28 @@ export default function Chat() {
</div>
<div className="chat-message-bubble">
<div className="chat-message-content chat-thinking-indicator">
<span className="chat-thinking-dots">
<span /><span /><span />
</span>
{stagingOp ? (
<div className="chat-staging-progress">
<div className="chat-staging-label">
<i className="fas fa-cloud-arrow-up" /> Transferring model{stagingOp.nodeName ? ` to ${stagingOp.nodeName}` : ''}...
</div>
{stagingOp.progress > 0 && (
<div className="chat-staging-detail">
<div className="chat-staging-bar-container">
<div className="chat-staging-bar" style={{ width: `${stagingOp.progress}%` }} />
</div>
<span className="chat-staging-pct">{Math.round(stagingOp.progress)}%</span>
</div>
)}
{stagingOp.message && (
<div className="chat-staging-file">{stagingOp.message}</div>
)}
</div>
) : (
<span className="chat-thinking-dots">
<span /><span /><span />
</span>
)}
</div>
</div>
</div>
Expand Down
21 changes: 21 additions & 0 deletions core/http/routes/ui_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -153,6 +153,27 @@ func RegisterUIAPIRoutes(app *echo.Echo, cl *config.ModelConfigLoader, ml *model
operations = append(operations, opData)
}

// Append active file staging operations (distributed mode only)
if d := applicationInstance.Distributed(); d != nil && d.Router != nil {
for modelID, status := range d.Router.StagingTracker().GetAll() {
operations = append(operations, map[string]any{
"id": "staging:" + modelID,
"name": modelID,
"fullName": modelID,
"jobID": "staging:" + modelID,
"progress": int(status.Progress),
"taskType": "staging",
"isDeletion": false,
"isBackend": false,
"isQueued": false,
"isCancelled": false,
"cancellable": false,
"message": status.Message,
"nodeName": status.NodeName,
})
}
}

// Sort operations by progress (ascending), then by ID for stable display order
slices.SortFunc(operations, func(a, b map[string]any) int {
progressA := a["progress"].(int)
Expand Down
61 changes: 41 additions & 20 deletions core/services/nodes/file_stager_http.go
Original file line number Diff line number Diff line change
Expand Up @@ -146,10 +146,11 @@ func (h *HTTPFileStager) doUpload(ctx context.Context, addr, nodeID, localPath,
defer f.Close()

var body io.Reader = f
// For files > 100MB, wrap with progress logging
cb := StagingProgressFromContext(ctx)
// For files > 100MB or when a progress callback is set, wrap with progress reporting
const progressThreshold = 100 << 20
if fileSize > progressThreshold {
body = newProgressReader(f, fileSize, filepath.Base(localPath), nodeID)
if fileSize > progressThreshold || cb != nil {
body = newProgressReader(f, fileSize, filepath.Base(localPath), nodeID, cb)
}

req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, body)
Expand Down Expand Up @@ -268,26 +269,30 @@ func (h *HTTPFileStager) probeExisting(ctx context.Context, addr, localPath, key
}

// progressReader wraps an io.Reader and logs upload progress periodically.
// If a StagingProgressCallback is present in the context, it also calls it
// for UI-visible progress updates.
type progressReader struct {
reader io.Reader
total int64
read int64
file string
node string
lastLog time.Time
lastPct int
start time.Time
mu sync.Mutex
reader io.Reader
total int64
read int64
file string
node string
lastLog time.Time
lastPct int
start time.Time
mu sync.Mutex
progressCb StagingProgressCallback
}

func newProgressReader(r io.Reader, total int64, file, node string) *progressReader {
func newProgressReader(r io.Reader, total int64, file, node string, cb StagingProgressCallback) *progressReader {
return &progressReader{
reader: r,
total: total,
file: file,
node: node,
start: time.Now(),
lastLog: time.Now(),
reader: r,
total: total,
file: file,
node: node,
start: time.Now(),
lastLog: time.Now(),
progressCb: cb,
}
}

Expand All @@ -313,6 +318,10 @@ func (pr *progressReader) Read(p []byte) (int, error) {
pr.lastLog = now
pr.lastPct = pct
}
// Call external progress callback for UI visibility
if pr.progressCb != nil {
pr.progressCb(pr.file, pr.read, pr.total)
}
pr.mu.Unlock()
}
return n, err
Expand Down Expand Up @@ -385,7 +394,19 @@ func (h *HTTPFileStager) FetchRemoteByKey(ctx context.Context, nodeID, key, loca
}
defer f.Close()

written, err := io.Copy(f, resp.Body)
// Wrap response body with progress reporting if callback is set or file is large
var src io.Reader = resp.Body
cb := StagingProgressFromContext(ctx)
totalSize := resp.ContentLength
const progressThreshold = 100 << 20
if totalSize > progressThreshold || cb != nil {
if totalSize <= 0 {
totalSize = 0 // unknown size — progress reader will still report bytes
}
src = newProgressReader(resp.Body, totalSize, filepath.Base(key), nodeID, cb)
}

written, err := io.Copy(f, src)
if err != nil {
os.Remove(localDst)
return fmt.Errorf("writing to %s: %w", localDst, err)
Expand Down
9 changes: 8 additions & 1 deletion core/services/nodes/file_stager_s3.go
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,14 @@ func (s *S3NATSFileStager) EnsureRemote(ctx context.Context, nodeID, localPath,
// Upload to S3 if not already present
exists, _ := s.fm.Exists(ctx, key)
if !exists {
if err := s.fm.Upload(ctx, key, localPath); err != nil {
// Wrap with progress reporting if a staging callback is available
var progressFn storage.UploadProgressFunc
if cb := StagingProgressFromContext(ctx); cb != nil {
progressFn = func(fileName string, bytesWritten, totalBytes int64) {
cb(fileName, bytesWritten, totalBytes)
}
}
if err := s.fm.UploadWithProgress(ctx, key, localPath, progressFn); err != nil {
return "", fmt.Errorf("uploading %s to S3: %w", localPath, err)
}
}
Expand Down
Loading
Loading