diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/DownloadState.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/DownloadState.ets new file mode 100644 index 0000000..fda8d40 --- /dev/null +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/DownloadState.ets @@ -0,0 +1,78 @@ +// Type definitions and constants for the download subsystem. +// +// INVARIANTS (must hold at every observable moment): +// +// 1. bytesOnDisk MUST come from fs.statSync(tmpPath).size, never from +// dataPreferences / AppStorage / callback args. During active +// download we ACCEPT callback-reported bytes as a 500ms optimistic +// overlay, but every ≤5s an authoritative fs.statSync reconciles. +// +// 2. Derived fields (pct / speed / eta) are NEVER persisted. Each is +// recomputed from bytesOnDisk + bytesTotal + clock at publish time. +// +// 3. dataPreferences stores BOOLEAN HINTS only (e.g. "has_pending:"), +// never byte counts. Byte counts are always re-derived from fs on +// observation. +// +// 4. After any state transition (running → paused / completed / failed), +// the .tmp file MUST be fsync'd before AppStorage is updated. This +// bounds post-kill drift to the OS page cache, not to unflushed +// userspace buffers. +// +// 5. UI-displayed pct drift ≤ 5s vs actual fs bytes. Any larger drift +// is a bug. + +export type DownloadStatusKind = + | 'idle' + | 'running' + | 'paused' + | 'completed' + | 'failed'; + +export interface DownloadStatus { + kind: DownloadStatusKind; + /** + * Which model this download is for. Used by the page to distinguish + * "the model I'm viewing is downloading" (show progress) from + * "a different model is downloading" (show one-line notice). + */ + modelId?: string; + /** Currently-downloading file name, when known. */ + fileName?: string; + /** Race winner / single source label, when known. */ + source?: string; + /** Error message, only when kind === 'failed'. */ + error?: string; +} + +export interface DownloadSnapshot { + status: DownloadStatus; + /** INVARIANT 1: always reconciled from fs.statSync (.tmp file). */ + bytesOnDisk: number; + /** Total expected bytes; 0 means unknown. */ + bytesTotal: number; + /** INVARIANT 2: derived, never persisted. 0..1. */ + pct: number; + /** INVARIANT 2: derived, never persisted. bytes/sec. */ + speed: number; + /** INVARIANT 2: derived, never persisted. seconds. */ + eta: number; + /** Localized progress message for direct UI display. */ + statusText: string; + /** epoch ms of last publish. */ + lastUpdated: number; +} + +export const IDLE_SNAPSHOT: DownloadSnapshot = { + status: { kind: 'idle' }, + bytesOnDisk: 0, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: '', + lastUpdated: 0, +}; + +/** AppStorage key for the global download snapshot. */ +export const DOWNLOAD_STATE_KEY = 'minicpmv_download_state'; diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/ModelDownloadService.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/ModelDownloadService.ets new file mode 100644 index 0000000..9260cb4 --- /dev/null +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/ModelDownloadService.ets @@ -0,0 +1,754 @@ +// Singleton service that owns the model download lifecycle. +// +// Responsibilities: +// - Wraps DownloadManager.downloadModelsForSelected as the actual +// byte-pulling mechanism. +// - Translates (msg) callbacks into structured DownloadSnapshot published +// to AppStorage, so any @StorageLink observer re-renders automatically. +// - Cold-start / page-open reconciliation from filesystem (INVARIANT 1). +// +// The service holds an ability context attached from the page's +// aboutToAppear; the context survives across page navigations within the +// same ability, so downloads outlive any single page instance. + +import common from '@ohos.app.ability.common'; +import hilog from '@ohos.hilog'; +import fs from '@ohos.file.fs'; + +import { + downloadModelsForSelected, + pauseDownloadForFile, + resumeDownloadForFile, + cancelDownloadForFile, + reattachSystemTasks, + hasActiveTask, +} from '../utils/DownloadManager'; +import { LlamaEngine } from '../engine/LlamaEngine'; +import { ModelInfo } from '../engine/ModelInfo'; +import { + DownloadSnapshot, + DownloadStatus, + IDLE_SNAPSHOT, + DOWNLOAD_STATE_KEY, +} from './DownloadState'; +import { + scanModelDirForTmp, + parseProgressMsg, + expectedFilesForModel, + matchExpectedFile, +} from './Reconciler'; + +const DOMAIN = 0xC0DE; +const TAG = 'ModelDownloadService'; + +export class ModelDownloadService { + private static _shared: ModelDownloadService | null = null; + + static get shared(): ModelDownloadService { + if (ModelDownloadService._shared === null) { + ModelDownloadService._shared = new ModelDownloadService(); + } + return ModelDownloadService._shared as ModelDownloadService; + } + + private constructor() {} + + // ------------------------------------------------------------------- + // State + // ------------------------------------------------------------------- + + private _context: common.Context | null = null; + private _currentModel: ModelInfo | null = null; + private _currentFile: string | null = null; + private _bytesTotal: number = 0; + private _lastBytesOnDisk: number = 0; + private _lastTickAt: number = 0; + private _running: boolean = false; + /** + * In-flight download promise. Used to dedupe concurrent start() callers + * (e.g. Index page auto-download fires while the user is also looking + * at ModelManager — both want the same completion signal). + */ + private _currentRun: Promise | null = null; + /** + * Filesystem-authoritative reconcile timer. While a download is running, + * every 5 s we fs.statSync the current .tmp file and correct any drift + * between the optimistic callback value and the actual flushed bytes + * (INVARIANT 1 + INVARIANT 5). + */ + private _reconcileTimerId: number = -1; + /** + * Most recently published snapshot. The 5 s reconcile timer reads this + * to preserve UI fields (statusText, speed, eta) that it does NOT own — + * only bytesOnDisk / pct / lastUpdated are reconciled from fs, the rest + * are copied through so the user doesn't see the progress text flash + * back to "准备下载…" every time the timer fires. + */ + private _lastSnapshot: DownloadSnapshot = IDLE_SNAPSHOT; + /** + * One-shot guard: reattachSystemTasks should run at most once per app + * session. If the first call found nothing (no surviving system task), + * subsequent attach() calls (e.g. user navigates back to ModelManager) + * would otherwise re-run search + getTask and may pick up tasks we + * started ourselves in between, double-registering them. + */ + private _reattachDone: boolean = false; + + // ------------------------------------------------------------------- + // Public API + // ------------------------------------------------------------------- + + /** + * Bind the service to an ability context. Idempotent — the first call + * wins, subsequent calls with the same context are no-ops. Triggers + * cold-start reconciliation UNLESS a download is already in flight + * (in which case reconcile would clobber the running status). + * + * Called from ModelManager.aboutToAppear. + */ + async attach(context: common.Context): Promise { + if (this._context === null) { + this._context = context; + hilog.info(DOMAIN, TAG, 'attached to ability context'); + } + if (this._running) return; + if (!this._reattachDone) { + this._reattachDone = true; + // reattach re-hooks surviving tasks and publishes paused/running. If it + // found survivors, that publish is authoritative — return WITHOUT + // reconcileFromFs so a 0-byte mid-race partial can't be downgraded to + // 'idle' (which would let the user click Download and spawn a duplicate + // task on the same .tmp the survivor owns) — bug #1. + const found = await this.reattachAndPublish(); + if (found || this._running) return; + } + await this.reconcileFromFs(); + } + + /** + * Probe the system service for surviving download tasks owned by this app + * whose saveas matches the selected model, RE-ATTACH to them (keep alive + * for task.resume — the platform's HTTP-Range resume), and publish a + * 'paused' (held task) or 'running' (still transferring) snapshot. Returns + * true iff any matching survivor was found (so attach can skip + * reconcileFromFs and avoid the #1 idle-clobber). + * + * This is the ONLY path that truly resumes a partial: request.agent resume + * requires a live task + task.resume(); re-creating with Config.begins on a + * stale file fails on device. The user resumes the survivor via Resume → + * resume() → resumeDownloadForFile → task.resume(). + */ + private async reattachAndPublish(): Promise { + if (this._context === null) return false; + const model = LlamaEngine.getSelectedModel(this._context); + this._currentModel = model; + const dir = `${this._context.filesDir}/models/${model.id}`; + const expected = expectedFilesForModel(model); + const md5Map = new Map(); + if (model.ggufMd5) md5Map.set(model.ggufFileName, model.ggufMd5); + if (model.mmprojFileName && model.mmprojMd5) md5Map.set(model.mmprojFileName, model.mmprojMd5); + if (model.acousticFileName && model.acousticMd5) md5Map.set(model.acousticFileName, model.acousticMd5); + + try { + const result = await reattachSystemTasks(this._context, dir, expected, md5Map, + (msg: string) => { this.onProgressMsg(msg); }, + (fname: string) => { return this.onReattachedFileCompleted(fname); }); + if (result.matchedFileNames.length > 0) { + const firstName = result.matchedFileNames[0]; + this._currentFile = firstName; + this._lastBytesOnDisk = this.fsBytesForCurrentFile(); + if (this._lastBytesOnDisk < 0) this._lastBytesOnDisk = 0; + this._lastTickAt = Date.now(); + if (result.anyRunning) { + this._running = true; + this.startReconcileTimer(); + this.publish({ + status: { kind: 'running', modelId: model.id, fileName: firstName }, + bytesOnDisk: this._lastBytesOnDisk, + bytesTotal: 0, pct: 0, speed: 0, eta: 0, + statusText: this.prefixModelName(this.str('download_status_preparing')), + lastUpdated: Date.now(), + }); + hilog.info(DOMAIN, TAG, 'reattached to %{public}d live task(s), currentFile=%{public}s', + result.matchedFileNames.length, firstName); + } else { + this._running = false; + this.publish({ + status: { kind: 'paused', modelId: model.id, fileName: firstName }, + bytesOnDisk: this._lastBytesOnDisk, + bytesTotal: 0, pct: 0, speed: 0, eta: 0, + statusText: this.prefixModelName(this.str('download_status_paused_partial')), + lastUpdated: Date.now(), + }); + hilog.info(DOMAIN, TAG, 'reattached to %{public}d paused task(s), currentFile=%{public}s', + result.matchedFileNames.length, firstName); + } + return true; + } + } catch (e) { + hilog.warn(DOMAIN, TAG, 'reattachAndPublish threw: %{public}s', `${e}`); + } + return false; + } + + /** + * A file whose surviving task we re-attached has just completed. If the + * whole model is now present → publish completed; otherwise re-enter the + * job loop to fetch remaining files (the just-completed file is skipped via + * its MD5 fast-path). This is bug #3 — multi-file cold-start continuation. + */ + private async onReattachedFileCompleted(_fileName: string): Promise { + if (this._context === null) return; + if (LlamaEngine.modelsExist(this._context)) { + this._running = false; + this.stopReconcileTimer(); + this.onCompleted(this._context); + return; + } + if (this._currentRun === null) { + this.start(); + } + } + + /** + * Start downloading the currently-selected model. Returns a Promise + * that resolves when the download finishes (success or failure — failure + * is captured internally and surfaced via AppStorage, the promise itself + * does NOT reject). + * + * Behaviour depends on current state: + * - idle / completed / failed / cold-start paused → launch fresh download + * - running → no-op (dedupe, return same promise) + * - paused (user clicked Pause this session) → route to resume() so the + * existing system task picks up where it left off + * + * UI callers may ignore the returned promise — observing AppStorage + * via @StorageLink is the canonical way to track progress. + */ + start(context?: common.Context): Promise { + if (this._currentRun !== null) { + if (this._running) { + // Genuinely running — dedupe. + return this._currentRun; + } + // _currentRun is alive but _running is false → paused this session. + // Route to resume so we don't orphan the in-flight runDownload promise. + return this.resume(); + } + const ctx = context ?? this._context; + if (ctx === null) { + hilog.error(DOMAIN, TAG, 'start() called before attach() — returning resolved promise'); + return Promise.resolve(); + } + this._context = ctx; + + const model = LlamaEngine.getSelectedModel(ctx); + this._currentModel = model; + this._currentFile = null; + this._bytesTotal = 0; + this._lastBytesOnDisk = 0; + this._lastTickAt = Date.now(); + this._running = true; + + this.publish({ + status: { kind: 'running', modelId: model.id, fileName: model.ggufFileName }, + bytesOnDisk: 0, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName(this.str('download_status_preparing')), + lastUpdated: Date.now(), + }); + + const p = this.runDownload(ctx); + this._currentRun = p; + // Cleanup AFTER awaiters see the terminal status. Run on a separate + // chain so callers awaiting `p` are unaffected. + p.finally(() => { + this._currentRun = null; + this._running = false; + this.stopReconcileTimer(); + }).catch((): void => { + /* swallowed — runDownload catches and publishes onFailed internally */ + }); + this.startReconcileTimer(); + return p; + } + + /** + * Begin periodic fs-authoritative reconciliation while a download is + * active. INVARIANT 5: pct shown to + * the user must lag actual fs bytes by ≤ 5 s. The progress callback + * from request.agent reports optimistically (counts in-flight socket + * bytes); fs.statSync reports what's actually flushed to disk. On + * disagreement, fs wins. + */ + private startReconcileTimer(): void { + this.stopReconcileTimer(); + this._reconcileTimerId = setInterval((): void => { + this.reconcileFromFsDuringRun(); + }, 5000); + } + + private stopReconcileTimer(): void { + if (this._reconcileTimerId !== -1) { + clearInterval(this._reconcileTimerId); + this._reconcileTimerId = -1; + } + } + + /** + * Lightweight reconcile while running — only updates bytesOnDisk + pct + * from fs.statSync. Other fields (statusText, speed, eta, status.kind) + * are passed through from the previous snapshot so the UI doesn't + * flash back to "准备下载…" every time the timer ticks. + * + * Also detects completion: when the system task finishes it renames + * .tmp → final, fsBytesForCurrentFile returns -1 (no .tmp found), and + * modelsExist returns true → we publish 'completed'. + */ + private reconcileFromFsDuringRun(): void { + if (!this._running) return; + if (this._context === null || this._currentModel === null || this._currentFile === null) return; + const fsBytes = this.fsBytesForCurrentFile(); + + if (fsBytes < 0) { + // .tmp vanished. Two possibilities: + // 1) System task completed → renamed to final file → modelsExist true + // 2) Something went wrong (cancellation, fs error) + // Either way, stop the reconcile timer and re-evaluate from fs. + if (this._context !== null && LlamaEngine.modelsExist(this._context)) { + hilog.info(DOMAIN, TAG, 'reconcile: .tmp gone + final file exists → completed'); + this.stopReconcileTimer(); + this._running = false; + this._currentRun = null; + this.onCompleted(this._context); + } + return; + } + + // If the optimistic value is ahead of fs (callback reported bytes the + // OS hasn't flushed yet), pull it back to fs truth. If fs is ahead + // (callback lagged), advance to fs truth. Either way, fs wins. + if (fsBytes === this._lastBytesOnDisk) return; + this._lastBytesOnDisk = fsBytes; + const pct = this._bytesTotal > 0 ? Math.min(1, Math.max(0, fsBytes / this._bytesTotal)) : 0; + const prev = this._lastSnapshot; + this.publish({ + status: prev.status, + bytesOnDisk: fsBytes, + bytesTotal: this._bytesTotal, + pct, + speed: prev.speed, + eta: prev.eta, + statusText: prev.statusText, + lastUpdated: Date.now(), + }); + } + + /** + * Force a re-reconcile from filesystem. Called when the user re-enters + * the ModelManager page — picks up any drift if the page was destroyed + * mid-download. + */ + async reconcileNow(): Promise { + // During active download the optimistic callbacks + periodic reconcile + // are the authority; no need to clobber with cold-start logic. + if (this._running) return; + await this.reconcileFromFs(); + } + + /** + * Pause the currently-running download. Delegates to per-file task + * registry in DownloadManager. The .tmp file is preserved so a later + * resume can pick up where pause left off. + */ + async pause(): Promise { + // Deliberately do NOT early-return on `!this._running`. pause() + // flips _running=false unconditionally after pauseDownloadForFile, + // even if a task's pause() rejected or a trailing progress callback + // (which does NOT reset _running) re-published 'running' right after. + // Once _running is false while the task is still transferring, the old + // `if (!this._running) return` turned every later Pause click into a + // no-op — rapid clicking could leave Pause permanently dead. The Pause + // button is only rendered while the published status is 'running', so + // reaching here means the user genuinely wants to pause; always attempt + // it. pauseDownloadForFile is a no-op when there are no registered + // tasks, so this is safe to re-issue. + if (this._currentFile === null) return; + hilog.info(DOMAIN, TAG, 'pause requested for %{public}s (wasRunning=%{public}s)', + this._currentFile, `${this._running}`); + try { + await pauseDownloadForFile(this._currentFile); + this._running = false; + this.stopReconcileTimer(); + this.publish({ + status: { + kind: 'paused', + modelId: this._currentModel !== null ? this._currentModel.id : undefined, + fileName: this._currentFile ?? undefined, + }, + bytesOnDisk: this._lastBytesOnDisk, + bytesTotal: this._bytesTotal, + pct: this._bytesTotal > 0 ? this._lastBytesOnDisk / this._bytesTotal : 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName(this.str('download_status_paused_partial')), + lastUpdated: Date.now(), + }); + } catch (e) { + hilog.error(DOMAIN, TAG, 'pause failed: %{public}s', `${e}`); + } + } + + /** + * Resume a paused download via task.resume(); if the system task is still + * alive, the download picks up where it left off. + */ + async resume(): Promise { + if (this._running) return; + if (this._currentFile === null) return; + if (this._context === null) return; + // Cold-start / orphaned-.tmp case: a paused snapshot exists on disk but + // no system task survived (the system cleaned it after a long kill, or + // we stopped it during reattach). resumeDownloadForFile would be a no-op + // and we'd publish a bogus 'running'. Route through start() instead, + // which re-enters the job loop and resumes the .tmp via HTTP Range + // (Config.begins). The in-session pause→resume path (registry non-empty) + // still uses resumeDownloadForFile below. + if (!hasActiveTask(this._currentFile)) { + hilog.info(DOMAIN, TAG, 'resume: no live task for %{public}s — restarting via begins', this._currentFile); + return this.start(); + } + hilog.info(DOMAIN, TAG, 'resume requested for %{public}s', this._currentFile); + try { + await resumeDownloadForFile(this._currentFile); + this._running = true; + this._lastTickAt = Date.now(); + this.startReconcileTimer(); + this.publish({ + status: { + kind: 'running', + modelId: this._currentModel !== null ? this._currentModel.id : undefined, + fileName: this._currentFile ?? undefined, + }, + bytesOnDisk: this._lastBytesOnDisk, + bytesTotal: this._bytesTotal, + pct: this._bytesTotal > 0 ? this._lastBytesOnDisk / this._bytesTotal : 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName(this.str('download_status_preparing')), + lastUpdated: Date.now(), + }); + } catch (e) { + hilog.error(DOMAIN, TAG, 'resume failed: %{public}s', `${e}`); + } + } + + /** + * Cancel the current download. Stops system task, removes registration, + * AND deletes any .tmp file for the current model — "Cancel" semantics + * are "stop and discard progress", distinct from "Pause" which keeps + * the .tmp for later resume. The service transitions to idle. + */ + async cancel(): Promise { + if (this._currentFile === null) return; + if (this._context === null || this._currentModel === null) return; + hilog.info(DOMAIN, TAG, 'cancel requested for %{public}s', this._currentFile); + try { + await cancelDownloadForFile(this._currentFile); + } catch (e) { + hilog.warn(DOMAIN, TAG, 'cancel task cleanup failed (continuing): %{public}s', `${e}`); + } + // Wipe .tmp files for this model so the next start begins cleanly. + // Pause preserves .tmp; Cancel does not — that's the contract. + this.cleanupTmpFiles(this._context, this._currentModel); + + this._running = false; + this._currentRun = null; + this.stopReconcileTimer(); + this._currentFile = null; + this._bytesTotal = 0; + this._lastBytesOnDisk = 0; + + this.publish({ + status: { kind: 'idle' }, + bytesOnDisk: 0, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: '', + lastUpdated: Date.now(), + }); + } + + /** + * Remove every `.tmp` and `.r*.tmp` file in the model's directory whose + * name matches one of the model's expected output files. Used by cancel + * to implement "discard progress" semantics. + */ + private cleanupTmpFiles(context: common.Context, model: ModelInfo): void { + const dir = `${context.filesDir}/models/${model.id}`; + const expected = expectedFilesForModel(model); + let entries: string[]; + try { + entries = fs.listFileSync(dir); + } catch (_) { + return; + } + for (const entry of entries) { + if (matchExpectedFile(entry, expected) === null) continue; + try { fs.unlinkSync(`${dir}/${entry}`); } + catch (e) { hilog.warn(DOMAIN, TAG, 'unlink %{public}s failed: %{public}s', entry, `${e}`); } + } + } + + // ------------------------------------------------------------------- + // Internal: download loop + // ------------------------------------------------------------------- + + private async runDownload(context: common.Context): Promise { + try { + await downloadModelsForSelected(context, (msg: string) => { + this.onProgressMsg(msg); + }); + this.onCompleted(context); + } catch (e) { + this.onFailed(`${e}`); + } finally { + this.stopReconcileTimer(); + } + // NOTE: _running / _currentRun cleanup is done by start()'s outer + // finally chain so concurrent callers see consistent state. + } + + private onProgressMsg(msg: string): void { + const parsed = parseProgressMsg(msg); + + // Detect file transitions (gguf → mmproj → acoustic) and reset + // per-file accumulators so the snapshot reflects the CURRENT file, + // not a stale sum from the previous one. + if (parsed.fileName !== undefined && parsed.fileName !== this._currentFile) { + this._currentFile = parsed.fileName; + this._bytesTotal = 0; + this._lastBytesOnDisk = 0; + this._lastTickAt = Date.now(); + } + if (parsed.bytesTotal !== undefined && parsed.bytesTotal > 0) { + this._bytesTotal = parsed.bytesTotal; + } + + // INVARIANT 1: prefer fs bytes; fall back to parsed optimistic value. + const fsBytes = this.fsBytesForCurrentFile(); + const bytesOnDisk = fsBytes >= 0 + ? fsBytes + : (parsed.bytesOnDisk !== undefined ? parsed.bytesOnDisk : this._lastBytesOnDisk); + + const now = Date.now(); + const dtSec = (now - this._lastTickAt) / 1000; + let speed = 0; + if (dtSec > 0 && bytesOnDisk > this._lastBytesOnDisk) { + speed = (bytesOnDisk - this._lastBytesOnDisk) / dtSec; + } + const remaining = this._bytesTotal - bytesOnDisk; + const eta = speed > 0 ? remaining / speed : 0; + this._lastBytesOnDisk = bytesOnDisk; + this._lastTickAt = now; + + const pct = this._bytesTotal > 0 + ? Math.min(1, Math.max(0, bytesOnDisk / this._bytesTotal)) + : (parsed.pct !== undefined ? parsed.pct : 0); + + this.publish({ + status: { + kind: 'running', + modelId: this._currentModel !== null ? this._currentModel.id : undefined, + fileName: this._currentFile ?? undefined, + source: parsed.source, + }, + bytesOnDisk, + bytesTotal: this._bytesTotal, + pct, + speed, + eta, + statusText: this.prefixModelName(msg), + lastUpdated: now, + }); + } + + private onCompleted(context: common.Context): void { + // fs is the authority — verify final state from disk + const exists = LlamaEngine.modelsExist(context); + const modelId = this._currentModel !== null ? this._currentModel.id : undefined; + const status: DownloadStatus = exists + ? { kind: 'completed', modelId } + : { kind: 'failed', modelId, error: 'files missing after download' }; + + this.publish({ + status, + bytesOnDisk: exists ? this._bytesTotal : 0, + bytesTotal: this._bytesTotal, + pct: exists ? 1 : 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName( + exists + ? this.str('download_status_completed') + : this.str('download_status_failed_missing') + ), + lastUpdated: Date.now(), + }); + + hilog.info(DOMAIN, TAG, 'download completed, exists=%{public}s', `${exists}`); + } + + private onFailed(error: string): void { + const modelId = this._currentModel !== null ? this._currentModel.id : undefined; + this.publish({ + status: { kind: 'failed', modelId, error }, + bytesOnDisk: this._lastBytesOnDisk, + bytesTotal: this._bytesTotal, + pct: 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName(this.str('download_status_failed').replace('%s', error)), + lastUpdated: Date.now(), + }); + hilog.error(DOMAIN, TAG, 'download failed: %{public}s', error); + } + + // ------------------------------------------------------------------- + // Internal: filesystem reconciliation + // ------------------------------------------------------------------- + + /** + * Authoritative bytes-on-disk for the currently-downloading file, + * derived from fs scan. Returns -1 if we don't yet know which file to + * look at. + */ + private fsBytesForCurrentFile(): number { + if (this._context === null || this._currentModel === null || this._currentFile === null) { + return -1; + } + const expected = expectedFilesForModel(this._currentModel); + const sizes = scanModelDirForTmp(this._context, this._currentModel.id, expected); + const v = sizes.get(this._currentFile); + return v === undefined ? -1 : v; + } + + /** + * Cold-start / idle-state reconciliation. Looks at filesystem truth: + * - final file present + MD5 OK → completed + * - any tmp present → paused (incomplete) + * - otherwise → idle + * + * This is the ONLY path that should publish during app launch or page + * re-entry (when no download is actively running). + */ + private async reconcileFromFs(): Promise { + if (this._context === null) { + this.publish(IDLE_SNAPSHOT); + return; + } + const context = this._context; + const model = LlamaEngine.getSelectedModel(context); + this._currentModel = model; + + // Final files present? → completed (LlamaEngine.modelsExist checks + // gguf + mmproj + acoustic as configured for the model) + if (LlamaEngine.modelsExist(context)) { + this.publish({ + status: { kind: 'completed' }, + bytesOnDisk: 0, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: '', + lastUpdated: Date.now(), + }); + return; + } + + // Any in-flight tmp? → paused + const expected = expectedFilesForModel(model); + const sizes = scanModelDirForTmp(context, model.id, expected); + let totalBytes = 0; + let lastFileName: string | null = null; + sizes.forEach((bytes: number, fname: string) => { + totalBytes += bytes; + if (bytes > 0) lastFileName = fname; + }); + + if (totalBytes > 0 && lastFileName !== null) { + this._currentFile = lastFileName; + this._lastBytesOnDisk = totalBytes; + this.publish({ + status: { kind: 'paused', modelId: model.id, fileName: lastFileName }, + bytesOnDisk: totalBytes, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: this.prefixModelName(this.str('download_status_paused_partial')), + lastUpdated: Date.now(), + }); + hilog.info(DOMAIN, TAG, 'cold-start reconcile: paused at %{public}d bytes for %{public}s', + totalBytes, lastFileName); + return; + } + + // Clean state + this.publish({ + status: { kind: 'idle' }, + bytesOnDisk: 0, + bytesTotal: 0, + pct: 0, + speed: 0, + eta: 0, + statusText: '', + lastUpdated: Date.now(), + }); + } + + // ------------------------------------------------------------------- + // Internal: AppStorage bridge + // ------------------------------------------------------------------- + + private publish(snapshot: DownloadSnapshot): void { + this._lastSnapshot = snapshot; + try { + AppStorage.setOrCreate(DOWNLOAD_STATE_KEY, snapshot); + } catch (e) { + hilog.error(DOMAIN, TAG, 'publish failed: %{public}s', `${e}`); + } + } + + /** + * Prefix a status line with the current model's display name so the + * user sees "[MiniCPM-V 4.6] gguf-...: 42%" rather than just the + * file-name-only message that DownloadManager emits. + */ + private prefixModelName(text: string): string { + if (this._currentModel !== null) { + return `[${this._currentModel.displayName}] ${text}`; + } + return text; + } + + /** + * Look up a localised string from the ability's resourceManager. Falls + * back to the raw key if the service has not yet been attached (very + * unlikely — publish only happens after attach). + */ + private str(name: string): string { + if (this._context === null) return name; + try { + return this._context.resourceManager.getStringByNameSync(name); + } catch (_) { + return name; + } + } +} diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/Reconciler.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/Reconciler.ets new file mode 100644 index 0000000..1843086 --- /dev/null +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/download/Reconciler.ets @@ -0,0 +1,160 @@ +// Filesystem-authoritative reconciliation utilities. +// +// Two responsibilities: +// 1. scanModelDirForTmp — list every .tmp / .r*.tmp under models//, +// sum their sizes per expected file. This is the AUTHORITATIVE input +// for bytesOnDisk (INVARIANT 1). +// 2. parseProgressMsg — best-effort extraction of (fileName, pct, +// bytesOnDisk, bytesTotal) from a download message string. Used as +// the OPTIMISTIC overlay between fs reconcile ticks (every 500 ms). +// +// Rule of thumb: when fs and parseProgressMsg disagree, fs wins. + +import fs from '@ohos.file.fs'; +import common from '@ohos.app.ability.common'; + +import { ModelInfo } from '../engine/ModelInfo'; + +/** + * For each expected file name in `expectedFiles`, sum the sizes of every + * `*.tmp` and `*.r.tmp` file present under `models//`. The + * race downloader (DownloadManager.downloadFileWithSources) writes to + * `.r0.tmp` / `.r1.tmp` while racing, then renames the + * winner to `` on completion — only the loser cleanups are + * deleted, so the surviving tmp bytes ARE the in-flight download size. + * + * Returns a Map keyed by file name. Files with no tmp present have value 0. + */ +export function scanModelDirForTmp( + context: common.Context, + modelId: string, + expectedFiles: string[], +): Map { + const result = new Map(); + for (const fname of expectedFiles) { + result.set(fname, 0); + } + + const dir = `${context.filesDir}/models/${modelId}`; + let entries: string[]; + try { + entries = fs.listFileSync(dir); + } catch (_) { + return result; // dir does not exist yet — nothing in flight + } + + for (const entry of entries) { + if (!entry.endsWith('.tmp')) continue; + + const matchedFname = matchExpectedFile(entry, expectedFiles); + if (matchedFname === null) continue; + + try { + const stat = fs.statSync(`${dir}/${entry}`); + const prev = result.get(matchedFname) ?? 0; + result.set(matchedFname, prev + stat.size); + } catch (_) { + // file vanished between listSync and statSync — ignore + } + } + + return result; +} + +/** + * The list of file names that we expect to see on disk for a given model. + * Used by Reconciler to know which tmp files to scan. + */ +export function expectedFilesForModel(model: ModelInfo): string[] { + const files: string[] = [model.ggufFileName]; + if (model.mmprojFileName) files.push(model.mmprojFileName); + if (model.acousticFileName) files.push(model.acousticFileName); + return files; +} + +/** + * Match a disk entry / saveas basename to one of the expected model file + * names. Returns the matched file name, or null if it isn't one of ours. + * + * A match is either: + * - `.tmp` (single-source in-flight), or + * - `.r.tmp` (race in-flight, N a non-empty run of digits). + * + * Centralised because three call sites (scanModelDirForTmp, reattach's + * saveas matching, cleanupTmpFiles) previously inlined this logic and + * drifted — bug #2 was exactly such a drift: the reattach completed + * handler renamed `.tmp` while race tasks write `.rN.tmp`. + */ +export function matchExpectedFile(basename: string, expectedFiles: string[]): string | null { + const SUFFIX = '.tmp'; + for (const fname of expectedFiles) { + if (basename === `${fname}${SUFFIX}`) return fname; + const prefix = `${fname}.r`; + if (basename.startsWith(prefix) && basename.endsWith(SUFFIX)) { + const mid = basename.substring(prefix.length, basename.length - SUFFIX.length); + if (mid.length > 0 && /^\d+$/.test(mid)) return fname; + } + } + return null; +} + +/** + * Structured progress parsed from a download message string. All fields + * optional — non-progress messages (e.g. "verifying md5…") yield empty. + * + * Recognised formats (from DownloadManager.ets): + * ": % (/ MB) []" — race / single, with Content-Length + * ": MB []" — race / single, no Content-Length + * ": MB" — single source, no CL, no source label + */ +export interface ParsedProgress { + fileName?: string; + /** 0..1 */ + pct?: number; + /** bytes */ + bytesOnDisk?: number; + /** bytes; only set when the message includes a total */ + bytesTotal?: number; + /** race winner / single source label */ + source?: string; +} + +const RE_PCT_WITH_TOTAL = /^(.+?):\s*(\d+)%\s*\((\d+)\/(\d+)\s*MB\)(?:\s*\[(.+)\])?$/; +const RE_MB_ONLY = /^(.+?):\s*(\d+)\s*MB(?:\s*\[(.+)\])?$/; +const MB = 1024 * 1024; + +export function parseProgressMsg(msg: string): ParsedProgress { + const m1 = msg.match(RE_PCT_WITH_TOTAL); + if (m1 !== null && m1.length >= 5) { + const fileName = m1[1] ?? ''; + const pctStr = m1[2] ?? '0'; + const onDiskStr = m1[3] ?? '0'; + const totalStr = m1[4] ?? '0'; + const source = m1[5]; + const pctNum = parseInt(pctStr, 10); + const onDiskNum = parseInt(onDiskStr, 10); + const totalNum = parseInt(totalStr, 10); + return { + fileName: fileName.length > 0 ? fileName : undefined, + pct: isFinite(pctNum) ? pctNum / 100 : undefined, + bytesOnDisk: isFinite(onDiskNum) ? onDiskNum * MB : undefined, + bytesTotal: isFinite(totalNum) ? totalNum * MB : undefined, + source: source && source.length > 0 ? source : undefined, + }; + } + + const m2 = msg.match(RE_MB_ONLY); + if (m2 !== null && m2.length >= 3) { + const fileName = m2[1] ?? ''; + const mbStr = m2[2] ?? '0'; + const source = m2[3]; + const mbNum = parseInt(mbStr, 10); + return { + fileName: fileName.length > 0 ? fileName : undefined, + bytesOnDisk: isFinite(mbNum) ? mbNum * MB : undefined, + source: source && source.length > 0 ? source : undefined, + }; + } + + return {}; +} diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/LlamaEngine.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/LlamaEngine.ets index 73c3a51..eb2b7a2 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/LlamaEngine.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/LlamaEngine.ets @@ -33,7 +33,8 @@ import { STALE_MMPROJ_NAMES, findModelById, } from './ModelInfo'; -import { downloadModelsForSelected, ProgressListener } from '../utils/DownloadManager'; +import { ProgressListener } from '../utils/DownloadManager'; +import { ModelDownloadService } from '../download/ModelDownloadService'; const TAG = 'LlamaEngine'; const DOMAIN = 0xC0DE; @@ -106,11 +107,23 @@ export class LlamaEngine { private state: LlamaState = LlamaState.Uninitialized; private listeners: StateListener[] = []; + /** + * id of the model currently loaded into the native engine, or null when + * none is loaded. The UI uses this to tell "the selected model is the one + * that's ready" from "some other model is ready" — the engine is a + * singleton holding ONE model, so engine state alone is ambiguous on a + * page that browses multiple models. See ModelManager.syncStatusFromSelectedEngine. + */ + private loadedModelId: string | null = null; getCurrentState(): LlamaState { return this.state; } + getLoadedModelId(): string | null { + return this.loadedModelId; + } + // Mirror of Dispatchers.IO.limitedParallelism(1): every native invocation // is appended to this promise so we never have two LLM calls in flight at // the same time. @@ -225,6 +238,7 @@ export class LlamaEngine { hilog.info(DOMAIN, TAG, 'Auto-unloading before loading new model'); this.readyForSystemPrompt = false; this.mmprojLoaded = false; + this.loadedModelId = null; this.setState(LlamaState.UnloadingModel); nativeEngine.unload(); this.setState(LlamaState.Initialized); @@ -297,6 +311,10 @@ export class LlamaEngine { } hilog.info(DOMAIN, TAG, 'Model loaded!'); this.readyForSystemPrompt = true; + // The caller just selected this model then called loadModel, so the + // pref-backed selected id IS the loaded one. Record it so the UI can + // distinguish "selected == loaded (ready)" from "selected != loaded". + this.loadedModelId = LlamaEngine.getSelectedModel(this.context).id; this.setState(LlamaState.ModelReady); }); } @@ -533,6 +551,7 @@ export class LlamaEngine { hilog.info(DOMAIN, TAG, 'Unloading model...'); this.readyForSystemPrompt = false; this.mmprojLoaded = false; + this.loadedModelId = null; this.setState(LlamaState.UnloadingModel); nativeEngine.unload(); this.setState(LlamaState.Initialized); @@ -545,6 +564,7 @@ export class LlamaEngine { resetToInitialized(): void { this.mmprojLoaded = false; this.readyForSystemPrompt = false; + this.loadedModelId = null; this.setState(LlamaState.Initialized); } @@ -808,11 +828,21 @@ export class LlamaEngine { } // ------------------------------------------------------------------ - // Download wrapper (delegates to DownloadManager so the engine stays small) + // Download wrapper // ------------------------------------------------------------------ - - static downloadModels(context: common.Context, onProgress: ProgressListener): Promise { - return downloadModelsForSelected(context, onProgress); + // + // Routed through ModelDownloadService so that EVERY download — whether + // triggered from the Index page's first-launch auto-download or from + // ModelManager's manual click — surfaces through AppStorage and is + // visible to any @StorageLink observer. + // + // The legacy `onProgress` callback parameter is kept for source-level + // backward compatibility with the existing Index.ets call site, but is + // intentionally IGNORED. Callers should observe AppStorage via + // @StorageLink(DOWNLOAD_STATE_KEY) instead. + + static downloadModels(context: common.Context, _onProgress?: ProgressListener): Promise { + return ModelDownloadService.shared.start(context); } // ------------------------------------------------------------------ diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/ModelInfo.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/ModelInfo.ets index 4dd09b2..ab53c1d 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/ModelInfo.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/ModelInfo.ets @@ -119,7 +119,22 @@ export const AVAILABLE_MODELS: ModelInfo[] = [ 'mmproj-model-f16.gguf', '', 'openbmb/MiniCPM-V-4-gguf', - 'OpenBMB/MiniCPM-V-4-gguf' + 'OpenBMB/MiniCPM-V-4-gguf', + 'main', + 'master', + null, + null, + null, + null, + null, + null, + // MD5s cross-checked against iOS MiniCPMModelConst.swift:73 / :90. + // Android deliberately omits these (claiming git-LFS provides integrity), + // but our downloader hits HF/MS via direct HTTPS resolve URLs that bypass + // LFS pointer verification — partial / corrupt bytes would land on disk + // unchecked. Same reasoning applies to MiniCPM5-1B below. + '8fc4cc88e5ea73472ae795b57a0e7fdd', + 'fe15375bb4c579858df6054d2a8b639d' ), new ModelInfo( 'minicpm-v-4_6-instruct', @@ -157,7 +172,9 @@ export const AVAILABLE_MODELS: ModelInfo[] = [ null, null, null, - null + null, + // MD5 from iOS MiniCPMModelConst.swift:181 + 'a5f29552a5abcc0533f18066851fe8dc' ), new ModelInfo( 'voxcpm2', @@ -182,7 +199,10 @@ export const AVAILABLE_MODELS: ModelInfo[] = [ ), ]; -export const DEFAULT_MODEL: ModelInfo = AVAILABLE_MODELS[0]; +// Default is V-4.6 (newest instruct model with MD5 verification + video +// understanding support). The list still keeps V-4 at index 0 for ordering +// stability in the picker UI. +export const DEFAULT_MODEL: ModelInfo = AVAILABLE_MODELS[1]; // Maps any historical filename (left) to the current canonical name (right) // for files that may already exist in a user's per-model subdirectory. diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/TtsEngine.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/TtsEngine.ets index bdd4c0f..50e355b 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/TtsEngine.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/engine/TtsEngine.ets @@ -64,11 +64,17 @@ export class TtsEngine { private state: TtsState = TtsState.Uninitialized; private listeners: TtsStateListener[] = []; private chain: Promise = Promise.resolve(); + /** id of the TTS model currently loaded, or null. See LlamaEngine.loadedModelId. */ + private loadedModelId: string | null = null; getCurrentState(): TtsState { return this.state; } + getLoadedModelId(): string | null { + return this.loadedModelId; + } + private constructor(context: common.Context) { this.context = context; } @@ -129,6 +135,7 @@ export class TtsEngine { if (this.state.name === 'Ready' || this.state.name === 'Generating') { hilog.info(DOMAIN, TAG, 'Freeing existing VoxCPM2 runtime before reload'); nativeEngine.ttsFree(); + this.loadedModelId = null; this.setState(TtsState.Uninitialized); } if (this.state.name !== 'Uninitialized') { @@ -148,6 +155,7 @@ export class TtsEngine { hilog.info(DOMAIN, TAG, 'initOmni success: %{public}s', result); hilog.info(DOMAIN, TAG, 'VoxCPM2 loaded successfully'); + this.loadedModelId = LlamaEngine.getSelectedModel(this.context).id; this.setState(TtsState.Ready); }); } @@ -195,6 +203,7 @@ export class TtsEngine { if (this.state.name !== 'Uninitialized') { hilog.info(DOMAIN, TAG, 'Freeing TTS engine...'); nativeEngine.ttsFree(); + this.loadedModelId = null; this.setState(TtsState.Uninitialized); } }); diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/Index.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/Index.ets index d5f2a9e..51ab280 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/Index.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/Index.ets @@ -36,6 +36,8 @@ import { PickedVideo, } from '../utils/VideoFrameExtractor'; import { LocaleManager } from '../manager/LocaleManager'; +import { DownloadSnapshot, IDLE_SNAPSHOT } from '../download/DownloadState'; +import { findModelById } from '../engine/ModelInfo'; const DOMAIN = 0xC0DE; const TAG = 'Index'; @@ -68,6 +70,10 @@ struct Index { @State isVisionSupported: boolean = false; @State inputText: string = ''; @State messageList: ChatMessageList = new ChatMessageList(); + // Read-only observer of the global download state — Index needs to know + // whether a download is already in flight before deciding whether to + // auto-kick one on launch. + @StorageLink('minicpmv_download_state') dlState: DownloadSnapshot = IDLE_SNAPSHOT; private engine: LlamaEngine | null = null; private engineListener = (s: LlamaState) => this.onEngineState(s); @@ -190,22 +196,65 @@ struct Index { const needsDownload = !fs.accessSync(modelPath) || (!selectedModel.isTextOnly && !!mmprojPath && !fs.accessSync(mmprojPath)); if (needsDownload) { - hilog.info(DOMAIN, TAG, 'Default model missing, auto-downloading...'); - promptAction.showToast({ message: this.str('toast_first_use_downloading'), duration: 3000 }); + // Three-branch policy: + // 1) A download is already in flight → don't queue another; just + // tell the user to wait. + // 2) Fresh install (no models/ dir yet) → auto-download the + // currently-selected model with the "首次进入" prompt. + // 3) Non-first-time user with missing files → don't auto-download; + // point them at Model Manager so they can pick + download + // deliberately. The "首次进入" wording would be a lie here. + const isDownloading = this.dlState.status.kind === 'running'; + if (isDownloading) { + const dlId = this.dlState.status.modelId; + const dlName = dlId ? findModelById(dlId).displayName : ''; + promptAction.showToast({ + message: this.str('toast_another_downloading').replace('%s', dlName), + duration: 3000, + }); + this.hasAutoLoaded = false; + return; + } + + const modelsRoot = `${ctx.filesDir}/models`; + let isFirstLaunch = false; try { - await LlamaEngine.downloadModels(ctx, (msg: string) => { - hilog.info(DOMAIN, TAG, 'download: %{public}s', msg); + isFirstLaunch = !fs.accessSync(modelsRoot); + } catch (_) { + isFirstLaunch = true; + } + + if (isFirstLaunch) { + hilog.info(DOMAIN, TAG, 'First launch — auto-downloading default model'); + promptAction.showToast({ + message: this.str('toast_first_use_downloading').replace('%s', selectedModel.displayName), + duration: 3000, + }); + try { + await LlamaEngine.downloadModels(ctx); + } catch (e) { + hilog.error(DOMAIN, TAG, 'auto download failed: %{public}s', `${e}`); + this.engine.resetToInitialized(); + this.hasAutoLoaded = false; + promptAction.showToast({ + message: this.str('toast_download_failed').replace('%s', `${e}`), + duration: 5000, + }); + return; + } + // Path strings may have been refreshed by legacy-name migrations. + modelPath = LlamaEngine.modelPath(ctx); + mmprojPath = LlamaEngine.mmprojPath(ctx); + } else { + // Non-first-time + missing model — don't auto-download; prompt user + // to go to Model Manager and pick a model deliberately. + promptAction.showToast({ + message: this.str('toast_model_missing_go_download'), + duration: 4000, }); - } catch (e) { - hilog.error(DOMAIN, TAG, 'auto download failed: %{public}s', `${e}`); - this.engine.resetToInitialized(); this.hasAutoLoaded = false; - promptAction.showToast({ message: `${this.str('toast_download_failed').replace('%s', `${e}`)}`, duration: 5000 }); return; } - // Path strings may have been refreshed by legacy-name migrations. - modelPath = LlamaEngine.modelPath(ctx); - mmprojPath = LlamaEngine.mmprojPath(ctx); } if (!fs.accessSync(modelPath)) { diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/ModelManager.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/ModelManager.ets index 14bc16f..940eb0b 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/ModelManager.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/pages/ModelManager.ets @@ -18,6 +18,13 @@ import { } from '../engine/ModelInfo'; import { deleteSelectedModelFiles } from '../utils/DownloadManager'; import { LocaleManager, AppLanguage } from '../manager/LocaleManager'; +import { ModelDownloadService } from '../download/ModelDownloadService'; +import { + DownloadSnapshot, + DownloadStatusKind, + IDLE_SNAPSHOT, + DOWNLOAD_STATE_KEY, +} from '../download/DownloadState'; const DOMAIN = 0xC0DE; const TAG = 'ModelManager'; @@ -29,11 +36,16 @@ struct ModelManager { @State statusText: string = ''; @State selectedId: string = DEFAULT_MODEL.id; @State modelExists: boolean = false; - @State downloading: boolean = false; - @State downloadProgress: number = 0; @State loading: boolean = false; @State isModelReady: boolean = false; @State currentLang: AppLanguage = AppLanguage.ZH; + // Download state is owned by ModelDownloadService and broadcast through + // AppStorage; the page is a pure observer. @Watch bridges terminal + // status transitions back into local @State (modelExists) + toasts. + // Decorator argument must be a string literal (ArkTS restricts decorator + // expressions to literals); keep it in sync with DOWNLOAD_STATE_KEY. + @StorageLink('minicpmv_download_state') @Watch('onDlStateChange') dlState: DownloadSnapshot = IDLE_SNAPSHOT; + private prevDlStatusKind: DownloadStatusKind = 'idle'; private loadingLock: boolean = false; private engine: LlamaEngine | null = null; @@ -46,6 +58,68 @@ struct ModelManager { return this.resMgr!.getStringByNameSync(name); } + /** + * Time-based throttle for action buttons. Prevents the double-tap / + * accidental-triple-click scenario from kicking off concurrent state + * transitions (e.g. user taps Pause twice and the second tap races with + * the first tap's async publish, ending in an inconsistent snapshot). + * 500 ms is chosen as a balance: too short and a single human tap can + * span two windows (typical tap duration ~100 ms + travel time); too + * long and rapid intentional back-to-back actions feel sluggish. + */ + private lastActionAt: number = 0; + private shouldHandleAction(): boolean { + const now = Date.now(); + if (now - this.lastActionAt < 500) return false; + this.lastActionAt = now; + return true; + } + + /** + * True when the SELECTED model is the one currently downloading — used + * to decide whether to show a progress bar vs a one-line notice. + */ + private isThisModelDownloading(): boolean { + return this.dlState.status.kind === 'running' + && this.dlState.status.modelId === this.selectedId; + } + + /** + * True when the SELECTED model has a paused (partial) download. While + * paused, the canonical resume action is the Resume button in the status + * area — the bottom Download button must stay disabled so it cannot + * start a second, conflicting download task against the same .tmp file. + * This matters most after a cold-start reattach: start() would launch a + * fresh task instead of resuming the registered one, and two tasks + * writing the same .tmp corrupt the download. + */ + private isThisModelPaused(): boolean { + return this.dlState.status.kind === 'paused' + && this.dlState.status.modelId === this.selectedId; + } + + /** + * True when ANY model is downloading, regardless of selection. + */ + private isAnyModelDownloading(): boolean { + return this.dlState.status.kind === 'running'; + } + + /** + * Display name of the model currently being downloaded (for the + * "正在下载 X" notice shown when the user is browsing a different model). + * Returns empty string if no download is active. + */ + private downloadingModelName(): string { + const id = this.dlState.status.modelId; + if (id === undefined) return ''; + try { + return findModelById(id).displayName; + } catch (_) { + return id; + } + } + aboutToAppear(): void { this.resMgr = getContext(this).resourceManager; this.engine = LlamaEngine.getInstance(getContext(this)); @@ -56,6 +130,12 @@ struct ModelManager { this.engine.addListener(this.engineListener); this.ttsEngine.addListener(this.ttsEngineListener); this.currentLang = LocaleManager.currentLanguageSync(); + // Attach the download service to this ability's context (idempotent) + // and trigger a filesystem-authoritative reconcile so the snapshot + // reflects truth on page re-entry, not a stale callback value. + ModelDownloadService.shared.attach(getContext(this)); + // Sync local prev-status tracker to whatever the service already published + this.prevDlStatusKind = this.dlState.status.kind; } aboutToDisappear(): void { @@ -68,6 +148,16 @@ struct ModelManager { // ------------------------------------------------------------------ private onEngineState(state: LlamaState): void { + // Reflect LLM engine state only when (a) a non-TTS model is selected + // and (b) that model is the one actually loaded in the engine. The + // engine is a singleton holding ONE model; without the loadedModelId + // check the status line would mirror a global engine state that doesn't + // belong to the selected model (e.g. "已就绪" for a model that isn't + // loaded, after switching selection). addListener() also emits on + // subscribe, so this additionally stops the unused engine's state from + // clobbering on page re-entry. + if (findModelById(this.selectedId).isTts) return; + if (this.engine !== null && this.engine.getLoadedModelId() !== this.selectedId) return; this.stateName = state.name; switch (state.name) { case 'Uninitialized': this.statusText = this.str('status_uninitialized'); @@ -92,6 +182,10 @@ struct ModelManager { } private onTtsEngineState(state: TtsState): void { + // Reflect TTS engine state only when a TTS model is selected AND it is + // the one loaded. See onEngineState. + if (!findModelById(this.selectedId).isTts) return; + if (this.ttsEngine !== null && this.ttsEngine.getLoadedModelId() !== this.selectedId) return; switch (state.name) { case 'Uninitialized': this.statusText = this.str('status_uninitialized'); this.isModelReady = false; break; @@ -104,6 +198,30 @@ struct ModelManager { } } + /** + * Re-derive statusText / isModelReady for the SELECTED model. The status + * line should describe the selected model — which is "ready" only when + * the engine actually has THAT model loaded (loadedModelId === selectedId). + * When the user selects a different, not-loaded model we drop to a neutral + * status instead of mirroring the still-loaded model's state. + */ + private syncStatusFromSelectedEngine(): void { + const isTts = findModelById(this.selectedId).isTts; + const loadedId = isTts + ? (this.ttsEngine !== null ? this.ttsEngine.getLoadedModelId() : null) + : (this.engine !== null ? this.engine.getLoadedModelId() : null); + if (loadedId === this.selectedId) { + if (isTts) { + if (this.ttsEngine) this.onTtsEngineState(this.ttsEngine.getCurrentState()); + } else { + if (this.engine) this.onEngineState(this.engine.getCurrentState()); + } + } else { + this.statusText = ''; + this.isModelReady = false; + } + } + // ------------------------------------------------------------------ // Actions // ------------------------------------------------------------------ @@ -113,43 +231,103 @@ struct ModelManager { // setSelectedModel syncs cachedSelectedId immediately so the // following modelsExist() reads the *new* model, not the old one. LlamaEngine.setSelectedModel(getContext(this), modelId); + // dlState may reflect the previously-selected model — re-derive from fs + // so the snapshot matches the new selection. + ModelDownloadService.shared.reconcileNow(); this.modelExists = LlamaEngine.modelsExist(getContext(this)); + // Selection changed → the status line is now owned by a different + // engine. Re-derive it so we don't leave the previous model's engine + // state (e.g. an LLM 'Ready') showing for a now-selected TTS model. + this.syncStatusFromSelectedEngine(); const m = findModelById(modelId); promptAction.showToast({ message: this.str('toast_model_selected').replace('%s', m.displayName), duration: 1500 }); - if (this.modelExists) { - setTimeout(() => { this.onClickLoad(); }, 100); - } + // Loading is an explicit action via the Load button — selecting a model + // (even one already on disk) no longer auto-loads it. } - private async onClickDownload(): Promise { + private onClickDownload(): void { + if (!this.shouldHandleAction()) return; if (this.modelExists) { promptAction.showToast({ message: this.str('toast_already_downloaded'), duration: 1500 }); return; } - this.downloading = true; - this.downloadProgress = 0; - this.statusText = this.str('status_downloading'); - try { - await LlamaEngine.downloadModels(getContext(this), (msg) => { - this.statusText = msg; - const pctMatch = msg.match(/(\d+)%/); - if (pctMatch) { - this.downloadProgress = parseInt(pctMatch[1], 10); - } + if (this.isThisModelDownloading()) { + // Double-tap on the model that's already downloading — ignore. + return; + } + if (this.isThisModelPaused()) { + // A paused download is resumed via the status-area Resume button, + // not the Download button. Defensive guard (the button is also + // disabled) — without it a cold-start paused task would be hit by + // start(), spawning a SECOND task against the same .tmp file the + // reattached task already owns. + return; + } + if (this.isAnyModelDownloading()) { + // A different model is downloading — refuse; can't queue. + promptAction.showToast({ + message: this.str('toast_another_downloading').replace('%s', this.downloadingModelName()), + duration: 2500, + }); + return; + } + // A different model is paused — start() would route to resume() and + // silently keep downloading the OLD model while the user thinks + // they're starting the new one. Refuse and ask the user to resolve + // the paused task first. + if (this.dlState.status.kind === 'paused' && this.dlState.status.modelId !== this.selectedId) { + promptAction.showToast({ + message: this.str('toast_another_paused').replace('%s', this.downloadingModelName()), + duration: 3000, }); + return; + } + // Fire-and-forget — the service runs downloadModelsForSelected in the + // background and publishes progress / completion / failure through + // AppStorage. We observe via @StorageLink + @Watch. + ModelDownloadService.shared.start(); + } + + private onClickPause(): void { + if (!this.shouldHandleAction()) return; + ModelDownloadService.shared.pause(); + } + + private onClickResume(): void { + if (!this.shouldHandleAction()) return; + ModelDownloadService.shared.resume(); + } + + private onClickCancel(): void { + if (!this.shouldHandleAction()) return; + ModelDownloadService.shared.cancel(); + } + + /** + * Bridge download-status transitions back into local @State and toasts. + * ArkUI fires this on every change to dlState; we de-dupe by tracking + * the previous status kind and only reacting to actual transitions. + */ + private onDlStateChange(): void { + const kind = this.dlState.status.kind; + if (kind === this.prevDlStatusKind) return; + + if (kind === 'completed' && this.prevDlStatusKind !== 'completed') { this.modelExists = LlamaEngine.modelsExist(getContext(this)); - this.statusText = this.str('download_complete_status'); promptAction.showToast({ message: this.str('download_complete_toast'), duration: 1500 }); - } catch (e) { - hilog.error(DOMAIN, TAG, 'download failed: %{public}s', `${e}`); - this.statusText = this.str('toast_download_failed').replace('%s', `${e}`); - promptAction.showToast({ message: this.str('toast_download_failed').replace('%s', `${e}`), duration: 4000 }); - } finally { - this.downloading = false; + } else if (kind === 'failed' && this.prevDlStatusKind !== 'failed') { + this.modelExists = LlamaEngine.modelsExist(getContext(this)); + const err = this.dlState.status.error ?? ''; + promptAction.showToast({ + message: this.str('toast_download_failed').replace('%s', err), + duration: 4000, + }); } + this.prevDlStatusKind = kind; } private onClickLoad(): void { + if (!this.shouldHandleAction()) return; if (this.loadingLock) return; this.loadingLock = true; @@ -231,6 +409,7 @@ struct ModelManager { } private onClickDelete(): void { + if (!this.shouldHandleAction()) return; const m = findModelById(this.selectedId); const filesLine = m.mmprojFileName ? `\n• ${m.ggufFileName}\n• ${m.mmprojFileName}` @@ -248,6 +427,9 @@ struct ModelManager { try { const deleted = await deleteSelectedModelFiles(getContext(this)); this.modelExists = LlamaEngine.modelsExist(getContext(this)); + // After delete, the previous "completed" snapshot is stale — force a + // fs reconcile so dlState reflects the now-empty model dir. + ModelDownloadService.shared.reconcileNow(); if (deleted) { this.statusText = this.str('model_files_deleted'); promptAction.showToast({ message: this.str('model_files_deleted'), duration: 1500 }); @@ -361,15 +543,87 @@ struct ModelManager { @Builder StatusBar() { Column() { - Text(this.statusText) - .fontSize(14) - .fontColor($r('app.color.on_surface_variant')) - .width('100%') - .padding({ left: 16, right: 16, top: 12, bottom: 8 }); - if (this.downloading) { - Progress({ value: this.downloadProgress, total: 100, type: ProgressType.Linear }) + // Case 1: the selected model is the one downloading — full progress UI. + if (this.isThisModelDownloading()) { + Text(this.dlState.statusText) + .fontSize(14) + .fontColor($r('app.color.on_surface_variant')) + .width('100%') + .padding({ left: 16, right: 16, top: 12, bottom: 8 }); + Progress({ value: this.dlState.pct * 100, total: 100, type: ProgressType.Linear }) .width('100%').height(4) .margin({ left: 16, right: 16, bottom: 8 }); + Row({ space: 8 }) { + Button(this.str('btn_pause')) + .layoutWeight(1).height(36) + .backgroundColor($r('app.color.surface_variant')) + .fontColor($r('app.color.on_surface')) + .onClick(() => this.onClickPause()); + Button(this.str('btn_cancel')) + .layoutWeight(1).height(36) + .backgroundColor($r('app.color.surface_variant')) + .fontColor($r('app.color.on_surface')) + .onClick(() => this.onClickCancel()); + } + .width('100%') + .padding({ left: 16, right: 16, bottom: 8 }); + } + // Case 2: a DIFFERENT model is downloading — one-line notice so the + // user knows why their Download button is greyed out. + else if (this.isAnyModelDownloading()) { + Text( + this.str('download_status_other_running') + .replace('%s', this.downloadingModelName()) + .replace('%d', `${Math.floor(this.dlState.pct * 100)}`) + ) + .fontSize(12) + .fontColor($r('app.color.on_surface_variant')) + .width('100%') + .padding({ left: 16, right: 16, top: 12, bottom: 8 }); + } + // Case 3: selected model has partial .tmp on disk (cold-start state + // or after Pause). Offer Resume + Cancel buttons alongside the + // status text — without these, the user has no in-page way to + // resolve a paused task and would be stuck. + else if (this.dlState.status.kind === 'paused' && this.dlState.status.modelId === this.selectedId) { + Column() { + Text(this.dlState.statusText) + .fontSize(12) + .fontColor($r('app.color.on_surface_variant')) + .width('100%') + .padding({ left: 16, right: 16, top: 12, bottom: 4 }); + Row({ space: 8 }) { + Button(this.str('btn_resume')) + .layoutWeight(1).height(36) + .backgroundColor($r('app.color.surface_variant')) + .fontColor($r('app.color.on_surface')) + .onClick(() => this.onClickResume()); + Button(this.str('btn_cancel')) + .layoutWeight(1).height(36) + .backgroundColor($r('app.color.surface_variant')) + .fontColor($r('app.color.on_surface')) + .onClick(() => this.onClickCancel()); + } + .width('100%') + .padding({ left: 16, right: 16, bottom: 8 }); + } + } + // Case 3b: a DIFFERENT model is paused — surface it so the user + // knows why their Download button is going to be refused. + else if (this.dlState.status.kind === 'paused') { + Text(this.str('toast_another_paused').replace('%s', this.downloadingModelName())) + .fontSize(12) + .fontColor($r('app.color.on_surface_variant')) + .width('100%') + .padding({ left: 16, right: 16, top: 12, bottom: 8 }); + } + // Case 4: nothing in flight — show engine state (loading / ready / etc.) + else { + Text(this.statusText) + .fontSize(14) + .fontColor($r('app.color.on_surface_variant')) + .width('100%') + .padding({ left: 16, right: 16, top: 12, bottom: 8 }); } } .width('100%') @@ -422,21 +676,21 @@ struct ModelManager { .layoutWeight(1).height(44) .backgroundColor($r('app.color.surface_variant')) .fontColor($r('app.color.on_surface')) - .enabled(!this.modelExists && !this.downloading) + .enabled(!this.modelExists && !this.isThisModelDownloading() && !this.isThisModelPaused()) .onClick(() => this.onClickDownload()); Button(this.loading ? this.str('status_loading') : (this.isModelReady ? this.str('btn_reload') : this.str('btn_load'))) .layoutWeight(1).height(44) .backgroundColor(this.loading ? $r('app.color.surface_variant') : $r('app.color.primary')) .fontColor(this.loading ? $r('app.color.on_surface_variant') : $r('app.color.on_primary')) - .enabled(this.modelExists && !this.loading && !this.downloading) + .enabled(this.modelExists && !this.loading) .onClick(() => this.onClickLoad()); Button($r('app.string.btn_delete')) .layoutWeight(1).height(44) .backgroundColor($r('app.color.surface_variant')) .fontColor($r('app.color.on_surface')) - .enabled(this.modelExists && !this.loading && !this.downloading) + .enabled(this.modelExists && !this.loading) .onClick(() => this.onClickDelete()); } .width('100%') diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/utils/DownloadManager.ets b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/utils/DownloadManager.ets index 22a5e9d..a155924 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/utils/DownloadManager.ets +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/ets/utils/DownloadManager.ets @@ -33,6 +33,7 @@ import { AVAILABLE_MODELS, DEFAULT_MODEL, } from '../engine/ModelInfo'; +import { matchExpectedFile } from '../download/Reconciler'; const DOMAIN = 0xC0DE; const TAG = 'DownloadManager'; @@ -40,6 +41,444 @@ const PREFS_NAME = 'model_prefs'; export type ProgressListener = (msg: string) => void; +/** + * Active task registry, keyed by file name (e.g. "model.gguf"). + * Populated by downloadFileWithSources / downloadSingleSource; consumed + * by pauseDownloadForFile / resumeDownloadForFile / cancelDownloadForFile. + * + * The race downloader stores up to N tasks per file (one per source); the + * single-source path stores 1. After race settlement only the winner + * remains in the array (losers are removed via stopLoser). + */ +const activeTasks: Map = new Map(); + +/** Pause every active task for the given file. No-op if none registered. */ +export async function pauseDownloadForFile(fileName: string): Promise { + const arr = activeTasks.get(fileName); + if (!arr || arr.length === 0) return; + hilog.info(DOMAIN, TAG, 'pause %{public}s (%{public}d task(s))', fileName, arr.length); + await Promise.all(arr.map((t) => t.pause().catch((e: object) => { + hilog.warn(DOMAIN, TAG, 'pause task %{public}s failed: %{public}s', `${t.tid}`, `${e}`); + }))); +} + +/** Resume every active (paused) task for the given file. */ +export async function resumeDownloadForFile(fileName: string): Promise { + const arr = activeTasks.get(fileName); + if (!arr || arr.length === 0) return; + hilog.info(DOMAIN, TAG, 'resume %{public}s (%{public}d task(s))', fileName, arr.length); + await Promise.all(arr.map((t) => t.resume().catch((e: object) => { + hilog.warn(DOMAIN, TAG, 'resume task %{public}s failed: %{public}s', `${t.tid}`, `${e}`); + }))); +} + +/** True iff at least one active task is registered for the given file. */ +export function hasActiveTask(fileName: string): boolean { + const arr = activeTasks.get(fileName); + return !!arr && arr.length > 0; +} + +/** + * Diagnostic (read-only): enumerate every request.agent task this app owns + * and return a compact, human-readable summary (count + per-task tid · state · + * saveas). Used to observe duplicate / orphan tasks — e.g. call before and + * after a download cycle and confirm the count is stable. No stop / remove. + * + * Callable from a debug page / the test module; also logs the summary via + * hilog so it shows up in the device log during real-device testing. + */ +export async function debugListSystemTasks(): Promise { + const lines: string[] = []; + let tids: string[] = []; + try { + tids = await request.agent.search({ action: request.agent.Action.DOWNLOAD }); + } catch (e) { + const msg = `debugListSystemTasks: search failed: ${e}`; + hilog.warn(DOMAIN, TAG, '%{public}s', msg); + return msg; + } + lines.push(`request.agent task count = ${tids.length}`); + for (const tid of tids) { + try { + const info = await request.agent.show(tid); + const saveas = info.saveas ?? '(empty)'; + const state = info.progress.state; + lines.push(` tid=${tid} state=0x${state.toString(16)} saveas=${saveas}`); + } catch (e) { + lines.push(` tid=${tid} show failed: ${e}`); + } + } + const summary = lines.join('\n'); + hilog.info(DOMAIN, TAG, 'debugListSystemTasks:\n%{public}s', summary); + return summary; +} + +/** + * Stop + remove every active task for the given file. Does NOT delete + * the .tmp file — that's preserved for future resume. + */ +export async function cancelDownloadForFile(fileName: string): Promise { + const arr = activeTasks.get(fileName); + if (!arr || arr.length === 0) return; + hilog.info(DOMAIN, TAG, 'cancel %{public}s (%{public}d task(s))', fileName, arr.length); + await Promise.all(arr.map(async (t) => { + try { + await t.stop(); + } catch (_) { /* ignore */ } + try { + await request.agent.remove(t.tid); + } catch (_) { /* ignore */ } + })); + activeTasks.delete(fileName); +} + +/** + * Result of reattach: which expected file names had surviving system tasks + * that we re-attached to (kept alive for task.resume), whether any is + * actively transferring, and how many stray non-matching tasks we cleaned up. + */ +export interface ReattachResult { + matchedFileNames: string[]; + anyRunning: boolean; + cleanedUpStrayTids: number; +} + +/** + * On app launch, enumerate every request.agent task owned by this app and + * RE-ATTACH to any whose saveas matches one of `expectedFiles` (keeping the + * task alive so the user can resume it via task.resume — the platform's + * HTTP-Range resume mechanism). Stray / terminal tasks are removed. + * + * Why keep+resume instead of stop+remove+recreate: request.agent resume is + * designed around a LIVE task + task.resume() (the system issues the Range + * request internally). Re-creating a task with Config.begins on an existing + * partial FAILS on device ("GetFd File exists" with overwrite:false; silent + * truncation with overwrite:true) — verified on MLN-AL00. So the only way to + * truly resume a partial is to keep the surviving task and resume it. + * + * Only the race WINNER survives a kill (losers are stopped during race + * settlement, before any kill), so there is at most one task per file — no + * winner-gate is needed, and the completed handler renames the ACTUAL saveas + * basename (.tmp or .rN.tmp), fixing the old wrong-rename bug (#2). If two + * tasks for the same file somehow survive (kill during race, before + * settlement), the duplicate is stopped+removed (keep first). + * + * Strategy: + * - Matched, non-terminal task → getTask + register + re-hook callbacks + * - Duplicate survivor for same file → stop + remove (keep first) + * - Terminal task (COMPLETED/FAILED/REMOVED) → remove to free the record + * - Non-matching task → remove (leftover from a deleted model) + */ +export async function reattachSystemTasks( + context: common.Context, + dir: string, + expectedFiles: string[], + md5Map: Map, + onProgress: ProgressListener, + onCompletedFile: (fileName: string) => Promise, +): Promise { + const matchedFileNames: string[] = []; + const seen: Set = new Set(); + let anyRunning = false; + let cleanedUpStrayTids = 0; + + let tids: string[]; + try { + const filter: request.agent.Filter = { + action: request.agent.Action.DOWNLOAD, + }; + tids = await request.agent.search(filter); + } catch (e) { + hilog.warn(DOMAIN, TAG, 'reattach: search failed: %{public}s', `${e}`); + return { matchedFileNames, anyRunning, cleanedUpStrayTids }; + } + + hilog.info(DOMAIN, TAG, 'reattach: search returned %{public}d task(s): %{public}s', + tids.length, tids.join(',')); + + for (const tid of tids) { + let info: request.agent.TaskInfo; + try { + info = await request.agent.show(tid); + } catch (e) { + hilog.warn(DOMAIN, TAG, 'reattach: show(%{public}s) failed: %{public}s', tid, `${e}`); + continue; + } + + const saveas = info.saveas ?? ''; + if (saveas.length === 0) { + hilog.warn(DOMAIN, TAG, 'reattach: tid %{public}s has empty saveas, skipping', tid); + continue; + } + + // Match saveas to one of our expected file names. The basename is + // either `.tmp` (single-source) or `.r.tmp` (race). + const basename = saveas.indexOf('/') >= 0 + ? saveas.substring(saveas.lastIndexOf('/') + 1) + : saveas; + const matchedFname = matchExpectedFile(basename, expectedFiles); + + if (matchedFname === null) { + // Not our task (different URL / leftover from deleted model). + hilog.info(DOMAIN, TAG, 'reattach: tid %{public}s (saveas=%{public}s) does not match any expected file — removing', + tid, basename); + try { + await request.agent.remove(tid); + cleanedUpStrayTids++; + } catch (e) { + hilog.warn(DOMAIN, TAG, 'reattach: remove(%{public}s) failed: %{public}s', tid, `${e}`); + } + continue; + } + + const fileName = matchedFname; + const state = info.progress.state; + + // Terminal task (COMPLETED / FAILED / REMOVED): free the record. + if (state === request.agent.State.COMPLETED + || state === request.agent.State.FAILED + || state === request.agent.State.REMOVED) { + hilog.info(DOMAIN, TAG, 'reattach: %{public}s already terminal (state=%{public}d) — removing record', + fileName, state); + try { await request.agent.remove(tid); cleanedUpStrayTids++; } catch (_) { /* ignore */ } + continue; + } + + // Duplicate survivor for the same file (kill happened during the race, + // before settlement). We cannot know which source won, so keep the first + // and stop+remove the extra to avoid two completions racing on rename. + if (seen.has(fileName)) { + hilog.info(DOMAIN, TAG, 'reattach: duplicate survivor for %{public}s (tid=%{public}s) — stopping+removing', + fileName, tid); + try { + const dup = await request.agent.getTask(context, tid); + try { await dup.stop(); } catch (_) { /* ignore */ } + } catch (_) { /* ignore */ } + try { await request.agent.remove(tid); cleanedUpStrayTids++; } catch (_) { /* ignore */ } + continue; + } + seen.add(fileName); + + // Surviving in-flight task from a previous session — KEEP it and re-hook + // callbacks so the user can resume via task.resume() (the platform's + // Range-resume). Only the winner survives, so one task per file. + let task: request.agent.Task; + try { + task = await request.agent.getTask(context, tid); + } catch (e) { + hilog.warn(DOMAIN, TAG, 'reattach: getTask(%{public}s) failed: %{public}s', tid, `${e}`); + continue; + } + const actualTmpPath = `${dir}/${basename}`; + const targetPath = `${dir}/${fileName}`; + const md5 = md5Map.get(fileName) ?? null; + registerTask(fileName, task); + if (state === request.agent.State.RUNNING || state === request.agent.State.RETRYING) { + anyRunning = true; + } + hilog.info(DOMAIN, TAG, 'reattach: re-attached surviving %{public}s (state=%{public}d, tid=%{public}s, tmp=%{public}s)', + fileName, state, tid, basename); + + let lastProgressTime = 0; + task.on('progress', (p: request.agent.Progress) => { + const now = Date.now(); + if (now - lastProgressTime > 500) { + lastProgressTime = now; + const got = p.processed; + const totalArr = p.sizes; + const total = totalArr && totalArr.length > 0 ? totalArr[0] : -1; + if (total > 0) { + const pct = Math.floor(got * 100 / total); + const mb = Math.floor(got / (1024 * 1024)); + const totalMb = Math.floor(total / (1024 * 1024)); + onProgress(`${fileName}: ${pct}% (${mb}/${totalMb} MB)`); + } else { + const mb = Math.floor(got / (1024 * 1024)); + onProgress(`${fileName}: ${mb} MB`); + } + } + }); + + task.on('completed', async () => { + try { + if (fs.accessSync(targetPath)) fs.unlinkSync(targetPath); + // Rename the ACTUAL saveas file (.tmp or .rN.tmp), not a hardcoded + // .tmp — this was bug #2 (race tasks write .rN.tmp). + fs.renameSync(actualTmpPath, targetPath); + if (md5) { + onProgress(`verifying md5 ${fileName}`); + const actual = await computeMd5(targetPath); + if (actual.toLowerCase() !== md5.toLowerCase()) { + hilog.error(DOMAIN, TAG, 'reattach %{public}s MD5 mismatch: expected %{public}s, got %{public}s', + fileName, md5, actual); + try { fs.unlinkSync(targetPath); } catch (_) { /* ignore */ } + onProgress(`md5 mismatch ${fileName}`); + return; + } + hilog.info(DOMAIN, TAG, 'reattach %{public}s MD5 OK', fileName); + } + onProgress(`${fileName} complete`); + // Continue to any remaining files (#3): the service re-enters the + // job loop, which fast-paths this now-complete file and downloads rest. + await onCompletedFile(fileName); + } catch (e) { + hilog.error(DOMAIN, TAG, 'reattach %{public}s completed handler threw: %{public}s', + fileName, `${e}`); + } finally { + unregisterTask(fileName, task); + request.agent.remove(task.tid).catch(() => undefined); + } + }); + + task.on('failed', () => { + unregisterTask(fileName, task); + request.agent.show(task.tid).then((failedInfo: request.agent.TaskInfo) => { + onProgress(`${fileName} failed: ${JSON.stringify(failedInfo.faults)}`); + }).catch(() => { + onProgress(`${fileName} failed`); + }).finally(() => { + request.agent.remove(task.tid).catch(() => undefined); + }); + }); + + matchedFileNames.push(fileName); + } + + hilog.info(DOMAIN, TAG, 'reattach: re-attached %{public}d matched task(s) (anyRunning=%{public}s), cleaned up %{public}d stray', + matchedFileNames.length, `${anyRunning}`, cleanedUpStrayTids); + return { matchedFileNames, anyRunning, cleanedUpStrayTids }; +} + +/** Stop everything across all files. Used on app destruction. */ +export async function cancelAllDownloads(): Promise { + const fileNames = Array.from(activeTasks.keys()); + for (const f of fileNames) { + await cancelDownloadForFile(f); + } +} + +/** Internal: register a task under its file name. */ +function registerTask(fileName: string, task: request.agent.Task): void { + const arr = activeTasks.get(fileName); + if (arr) { + arr.push(task); + } else { + activeTasks.set(fileName, [task]); + } +} + +/** Internal: unregister a task. Safe to call multiple times. */ +function unregisterTask(fileName: string, task: request.agent.Task): void { + const arr = activeTasks.get(fileName); + if (!arr) return; + const i = arr.indexOf(task); + if (i >= 0) arr.splice(i, 1); + if (arr.length === 0) activeTasks.delete(fileName); +} + +/** Internal: a race-leftover .tmp file candidate for consolidation. */ +interface RaceTmpCandidate { + path: string; + size: number; +} + +/** + * Returns the byte size of `${dir}/${fileName}.tmp` if it exists, else 0. + * Used by the single-source path to decide whether to issue an HTTP Range + * request (Config.begins) for resumption. + */ +export function detectExistingTmpBytes(dir: string, fileName: string): number { + const tmpPath = `${dir}/${fileName}.tmp`; + try { + if (!fs.accessSync(tmpPath)) return 0; + return fs.statSync(tmpPath).size; + } catch (_) { + return 0; + } +} + +/** + * Remove `.tmp` and every `.r.tmp` under `dir`. Used + * when a leftover .tmp has no surviving live task and so cannot be resumed + * (begins-resume is unavailable on device) — discard it so the next download + * starts clean. + */ +export function cleanTmpsForFile(dir: string, fileName: string): void { + const prefix = `${fileName}.r`; + let entries: string[]; + try { entries = fs.listFileSync(dir); } catch (_) { return; } + for (const e of entries) { + if (e === `${fileName}.tmp` || (e.startsWith(prefix) && e.endsWith('.tmp'))) { + try { fs.unlinkSync(`${dir}/${e}`); } catch (_) { /* ignore */ } + } + } +} + +/** + * Inspect `${dir}` for race-leftover tmp files matching `${fileName}.r.tmp`. + * If any are found: + * - Pick the one with the largest size + * - Rename it to `${fileName}.tmp` (replacing any existing single .tmp) + * - Delete all other race tmps + * - Return the (now single-.tmp) byte count + * If none found, returns 0 — caller proceeds with a fresh race. + * + * Why: race + HTTP Range begin are incompatible (multiple sources writing + * to different files at the same offset → incoherent). When resuming, we + * must collapse to a single source. Picking the largest race tmp + * preserves the most progress; discarding the others avoids confusion. + */ +export function consolidateRaceTmps(dir: string, fileName: string, targetPath: string): number { + const singleTmpPath = `${targetPath}.tmp`; + let entries: string[]; + try { + entries = fs.listFileSync(dir); + } catch (_) { + return 0; + } + + const prefix = `${fileName}.r`; + const suffix = '.tmp'; + const raceTmps: RaceTmpCandidate[] = []; + + for (const entry of entries) { + if (!entry.startsWith(prefix) || !entry.endsWith(suffix)) continue; + const mid = entry.substring(prefix.length, entry.length - suffix.length); + if (mid.length === 0 || !/^\d+$/.test(mid)) continue; + const p = `${dir}/${entry}`; + try { + const s = fs.statSync(p); + const candidate: RaceTmpCandidate = { path: p, size: s.size }; + raceTmps.push(candidate); + } catch (_) { /* skip */ } + } + + if (raceTmps.length === 0) return 0; + + // Pick largest + raceTmps.sort((a, b) => b.size - a.size); + const winner = raceTmps[0]; + + // Replace single .tmp with the winner + try { + if (fs.accessSync(singleTmpPath)) fs.unlinkSync(singleTmpPath); + fs.renameSync(winner.path, singleTmpPath); + } catch (e) { + hilog.warn(DOMAIN, TAG, 'consolidate rename failed: %{public}s', `${e}`); + return 0; + } + + // Delete losers + for (const t of raceTmps) { + if (t.path === winner.path) continue; + try { fs.unlinkSync(t.path); } catch (_) { /* ignore */ } + } + + hilog.info(DOMAIN, TAG, 'consolidated %{public}d race tmp(s) → single .tmp (%{public}d bytes)', + raceTmps.length, winner.size); + return winner.size; +} + /** One physical source for a given file. */ interface RaceSource { label: string; @@ -164,10 +603,32 @@ async function downloadFileWithSources( // Single-source -> bypass race machinery for clarity / lower overhead. if (job.sources.length === 1) { const src = job.sources[0]; - await downloadSingleSource(context, dir, job.name, src, job.md5, onProgress); + // Detect existing .tmp for resume via HTTP Range (Config.begins). + const begins = detectExistingTmpBytes(dir, job.name); + await downloadSingleSource(context, dir, job.name, src, job.md5, onProgress, begins); return; } + // Multi-source: check for leftover .tmp from a previous run. If found, + // the race-tmp files are consolidated into a single .tmp (largest wins, + // others discarded) and we resume via single-source + begins instead of + // starting a fresh race. Race + begins would have multiple sources + // writing to the same offset of different files — incoherent. + // begins-resume of an existing partial is BROKEN on device (GetFd error / + // truncation — request.agent resume requires a LIVE task via task.resume, + // not Config.begins on a stale file). We reach this multi-source branch + // only when there is a leftover .tmp but NO surviving live task (system + // cleaned it, or the race never settled). Such a .tmp is unrecoverable — + // discard it and start a fresh race. (The common cold-start case keeps the + // live task and resumes it directly via reattach, never reaching here.) + const hasLeftoverTmp = consolidateRaceTmps(dir, job.name, targetPath) > 0 + || detectExistingTmpBytes(dir, job.name) > 0; + if (hasLeftoverTmp) { + hilog.info(DOMAIN, TAG, '%{public}s: discarding unrecoverable leftover .tmp (begins-resume unavailable) — fresh download', + job.name); + cleanTmpsForFile(dir, job.name); + } + hilog.info(DOMAIN, TAG, 'Race-downloading %{public}s from %{public}d sources', job.name, job.sources.length); return new Promise((resolve, reject) => { @@ -211,9 +672,6 @@ async function downloadFileWithSources( for (let i = 0; i < job.sources.length; i++) { const idx = i; const src = job.sources[idx]; - try { - if (fs.accessSync(tmpPaths[idx])) fs.unlinkSync(tmpPaths[idx]); - } catch (_) { /* ignore */ } hilog.info(DOMAIN, TAG, 'race[%{public}d] %{public}s start: %{public}s', idx, src.label, src.url); @@ -226,11 +684,11 @@ async function downloadFileWithSources( method: 'GET', mode: request.agent.Mode.FOREGROUND, network: request.agent.Network.ANY, - retry: false, }; request.agent.create(context, config).then((task: request.agent.Task) => { tasks[idx] = task; + registerTask(job.name, task); task.on('progress', (p: request.agent.Progress) => { // Race gate: first task to report non-empty bytes wins. @@ -265,6 +723,7 @@ async function downloadFileWithSources( if (settled) return; if (winnerIdx !== idx) { stopLoser(idx); + unregisterTask(job.name, task); return; } settled = true; @@ -294,6 +753,7 @@ async function downloadFileWithSources( } catch (e) { reject(e as object); } finally { + unregisterTask(job.name, task); request.agent.remove(task.tid).catch(() => undefined); cleanupLosers(idx); } @@ -303,6 +763,7 @@ async function downloadFileWithSources( if (settled) return; if (winnerIdx === idx) { settled = true; + unregisterTask(job.name, task); request.agent.show(task.tid).then(info => { reject(new Error( str(rm, 'download_source_failed_detail').replace('%s', src.label).replace('%s', JSON.stringify(info.faults)) @@ -311,17 +772,16 @@ async function downloadFileWithSources( reject(new Error(str(rm, 'download_source_failed').replace('%s', src.label))); }).finally(() => { request.agent.remove(task.tid).catch(() => undefined); - try { - if (fs.accessSync(tmpPaths[idx])) fs.unlinkSync(tmpPaths[idx]); - } catch (_) { /* ignore */ } }); return; } taskFailures.push(idx); hilog.warn(DOMAIN, TAG, '%{public}s race source %{public}s failed (%{public}d/%{public}d)', job.name, src.label, taskFailures.length + createFailures.length, job.sources.length); + unregisterTask(job.name, task); request.agent.remove(task.tid).catch(() => undefined); try { + // Loser tmp is safe to clean — only winner bytes are useful. if (fs.accessSync(tmpPaths[idx])) fs.unlinkSync(tmpPaths[idx]); } catch (_) { /* ignore */ } if (taskFailures.length + createFailures.length >= job.sources.length) { @@ -333,6 +793,7 @@ async function downloadFileWithSources( if (settled) return; hilog.warn(DOMAIN, TAG, '%{public}s race source %{public}s task.start() failed: %{public}s', job.name, src.label, `${e}`); + unregisterTask(job.name, task); taskFailures.push(idx); request.agent.remove(task.tid).catch(() => undefined); if (taskFailures.length + createFailures.length >= job.sources.length) { @@ -351,7 +812,10 @@ async function downloadFileWithSources( }); } -/** Simple wrapper for the rare 1-source case (e.g. legacy direct-mirror only). */ +/** Simple wrapper for the rare 1-source case (e.g. legacy direct-mirror only). + * Also used as the resume path: when `begins` > 0, the system issues an + * HTTP Range request and appends to the existing .tmp file. + */ async function downloadSingleSource( context: common.Context, dir: string, @@ -359,15 +823,21 @@ async function downloadSingleSource( src: RaceSource, md5: string | null, onProgress: ProgressListener, + begins: number = 0, ): Promise { const rm = context.resourceManager; const targetPath = `${dir}/${fileName}`; - onProgress(str(rm, 'download_fetching_file').replace('%s', src.label).replace('%s', fileName)); - hilog.info(DOMAIN, TAG, 'Downloading %{public}s from %{public}s: %{public}s', - fileName, src.label, src.url); + if (begins > 0) { + onProgress(str(rm, 'download_resuming').replace('%s', fileName).replace('%d', `${Math.floor(begins / (1024 * 1024))}`)); + hilog.info(DOMAIN, TAG, 'Resuming %{public}s from %{public}d bytes via %{public}s', + fileName, begins, src.label); + } else { + onProgress(str(rm, 'download_fetching_file').replace('%s', src.label).replace('%s', fileName)); + hilog.info(DOMAIN, TAG, 'Downloading %{public}s from %{public}s: %{public}s', + fileName, src.label, src.url); + } const tmpPath = `${targetPath}.tmp`; - if (fs.accessSync(tmpPath)) fs.unlinkSync(tmpPath); await new Promise((resolve, reject) => { let lastProgressTime = 0; @@ -376,13 +846,21 @@ async function downloadSingleSource( url: src.url, headers: { 'User-Agent': 'MiniCPMV-demo/1.0' }, saveas: tmpPath, + // NOTE: begins-resume of an EXISTING partial is broken on device: + // overwrite:true -> truncates .tmp, writes Range tail at offset 0 (corrupt) + // overwrite:false -> "GetFd File exists" error + // request.agent resume is designed around task.pause()/resume() on a LIVE + // task (HTTP Range handled internally), not Config.begins on a stale file. + // Kept overwrite:true so a stale .tmp is replaced; true resume needs the + // live-task reattach path or Route B (@ohos.net.http Range). See tmp/. overwrite: true, method: 'GET', mode: request.agent.Mode.FOREGROUND, network: request.agent.Network.ANY, - retry: false, + begins: begins, }; request.agent.create(context, config).then((task: request.agent.Task) => { + registerTask(fileName, task); task.on('progress', (p: request.agent.Progress) => { const now = Date.now(); if (now - lastProgressTime > 500) { @@ -409,16 +887,19 @@ async function downloadSingleSource( } catch (e) { reject(e as object); } finally { + unregisterTask(fileName, task); request.agent.remove(task.tid).catch(() => undefined); } }); task.on('failed', () => { + unregisterTask(fileName, task); request.agent.show(task.tid).then(info => { reject(new Error(`Download failed: ${JSON.stringify(info.faults)}`)); }).catch(() => { reject(new Error('Download failed')); }).finally(() => { request.agent.remove(task.tid).catch(() => undefined); + // NOTE: do NOT unlink tmp — preserved for resume. }); }); task.start(); diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/base/element/string.json b/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/base/element/string.json index 74f9b1d..6ac5dd5 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/base/element/string.json +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/base/element/string.json @@ -66,7 +66,19 @@ { "name": "toast_empty_input", "value": "请输入文字消息" }, { "name": "toast_model_not_ready", "value": "模型未就绪" }, { "name": "toast_vision_not_supported", "value": "当前模型不支持图像输入" }, - { "name": "toast_first_use_downloading", "value": "首次使用，正在下载模型..." }, + { "name": "toast_first_use_downloading", "value": "首次进入，正在下载 %s…" }, + { "name": "toast_model_missing_go_download", "value": "未找到模型文件，请前往模型管理下载" }, + { "name": "toast_another_downloading", "value": "正在下载 %s，请等待完成" }, + { "name": "toast_another_paused", "value": "%s 已暂停，请先取消或继续" }, + { "name": "download_status_preparing", "value": "准备下载…" }, + { "name": "download_status_paused_partial", "value": "已部分下载，点击重新下载" }, + { "name": "download_status_completed", "value": "下载完成" }, + { "name": "download_status_failed_missing", "value": "下载失败：文件缺失" }, + { "name": "download_status_failed", "value": "下载失败：%s" }, + { "name": "download_status_other_running", "value": "正在下载 %s (%d%%)" }, + { "name": "download_resuming", "value": "续传 %s（已下载 %d MB）…" }, + { "name": "btn_pause", "value": "暂停" }, + { "name": "btn_resume", "value": "继续" }, { "name": "toast_model_file_missing", "value": "模型文件不存在" }, { "name": "toast_model_load_failed", "value": "模型加载失败: %s" }, { "name": "toast_clear_chat_failed", "value": "清空失败: %s" }, diff --git a/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/en_US/element/string.json b/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/en_US/element/string.json index 1d34a74..3151213 100644 --- a/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/en_US/element/string.json +++ b/MiniCPM-V-demo-HarmonyOS/entry/src/main/resources/en_US/element/string.json @@ -61,7 +61,19 @@ { "name": "toast_empty_input", "value": "Please enter a message" }, { "name": "toast_model_not_ready", "value": "Model not ready" }, { "name": "toast_vision_not_supported", "value": "Current model does not support image input" }, - { "name": "toast_first_use_downloading", "value": "First launch — downloading model…" }, + { "name": "toast_first_use_downloading", "value": "First launch — downloading %s…" }, + { "name": "toast_model_missing_go_download", "value": "Model not found — please download it from Model Manager" }, + { "name": "toast_another_downloading", "value": "Downloading %s — please wait" }, + { "name": "toast_another_paused", "value": "%s is paused — cancel or resume it first" }, + { "name": "download_status_preparing", "value": "Preparing download…" }, + { "name": "download_status_paused_partial", "value": "Partially downloaded — click to re-download" }, + { "name": "download_status_completed", "value": "Download complete" }, + { "name": "download_status_failed_missing", "value": "Download failed: files missing" }, + { "name": "download_status_failed", "value": "Download failed: %s" }, + { "name": "download_status_other_running", "value": "Downloading %s (%d%%)" }, + { "name": "download_resuming", "value": "Resuming %s (%d MB already downloaded)…" }, + { "name": "btn_pause", "value": "Pause" }, + { "name": "btn_resume", "value": "Resume" }, { "name": "toast_model_file_missing", "value": "Model file not found" }, { "name": "toast_model_load_failed", "value": "Failed to load model: %s" }, { "name": "toast_clear_chat_failed", "value": "Failed to clear chat: %s" },