Vito
2cecfcf0ac
* feat(activity): scan new and unresolved files by default, all files on request
Scan for metadata now skips, before reading any metadata, the files the
Library already links with a resolved link (one generated_images row with
exactly one pair link), found with one streamed query that keeps only the
scanned paths and compares paths with os.path.normcase on both sides.
Skipped files count as Already known ("Skipped · already linked").
Unlinked, fallback-linked, ambiguous and unresolved files are read again.
POST /prompt_manager/jobs/scan takes {scope: "new" | "all"} (default
"new", anything else is a 400). The review dialog shows the concept's
"New and unresolved files" and adds an "All files" switch (D58). The
legacy /prompt_manager/scan route keeps scanning all files with unchanged
SSE payloads.
* feat(activity): download a job report
GET /prompt_manager/jobs/{id}/report returns the summary, counters, every
outcome group's listed rows with truncated flags, the scope and the log as a
JSON attachment (prompt-manager-<kind>-<YYYYMMDD-HHMM>.json), with no absolute
paths. The job view's Download report saves it through the api-client URL, and
the scan log opens at its newest line and follows new lines only while the
reader is at the bottom.
* fix(activity): re-read known files whose size changed, and review follow-ups
The new scope now skips a known, resolved file only while its stored
generated_images.file_size equals its size on disk; a NULL stored size is
re-read. A file replaced at the same path is read again and its conflict
shows under Needs review. The scan stores each file's size when it ingests
it, and both sides of the known-set match go through normalize_image_path.
A failed known-set read logs and falls back to reading every file, and an
empty request body asks for the default scope. The All files switch is
described by its hint, a lost job's Download report is disabled, the report
toast is skipped once the view is gone, and the scan log opens at its
newest line every time. Fixtures use the real scope shape.
* test(activity): wait for the job view's settled detail read, not a fixed delay
The 200-row cap test now starts counting detail reads once one read has
returned the full listing and none is in flight. The job view re-reads after
an in-flight read only when the summary still lists more than that read
returned, so no redundant read is left due after opening.
372 lines
19 KiB
JavaScript
372 lines
19 KiB
JavaScript
// Activity (#249 Activity-1): the session job store behind the top-bar button and pill and the Activity page.
|
|
const test = require('node:test');
|
|
const assert = require('node:assert/strict');
|
|
|
|
const source = import('../../web/js/workspace/activity-store.mjs');
|
|
|
|
const ID_A = 'a'.repeat(32);
|
|
const ID_B = 'b'.repeat(32);
|
|
const ID_C = 'c'.repeat(32);
|
|
|
|
function summary(overrides = {}) {
|
|
return { id: ID_A, kind: 'scan', title: 'Scan for metadata', state: 'running', started_at: 1000, finished_at: null,
|
|
progress: 0, status_line: '', processed: 0, total: 0,
|
|
counters: { added: 0, known: 0, review: 0, no_metadata: 0, errors: 0 }, scope: { files: 'new', output_folders: 2 },
|
|
...overrides };
|
|
}
|
|
|
|
function apiError(status, message, code) {
|
|
return Object.assign(new Error(message), { status, code });
|
|
}
|
|
|
|
/** A fake world: the API's answers, every opened stream, the pending timers and the toasts. */
|
|
function harness() {
|
|
const world = { list: [], listError: null, posts: [], postResult: null, postError: null, streams: [], timers: [],
|
|
toasts: [], gets: 0 };
|
|
const cleanups = [];
|
|
const controller = new AbortController();
|
|
const scope = { signal: controller.signal, cleanup: fn => cleanups.push(fn) };
|
|
const api = {
|
|
async get(url, options) {
|
|
assert.equal(url, '/prompt_manager/jobs');
|
|
assert.equal(options.cache, 'no-store');
|
|
world.gets += 1;
|
|
if (world.listError) throw world.listError;
|
|
return { success: true, jobs: world.list };
|
|
},
|
|
async post(url, body) {
|
|
world.posts.push([url, body]);
|
|
if (world.postError) throw world.postError;
|
|
return world.postResult;
|
|
},
|
|
};
|
|
const connect = url => {
|
|
const stream = { url, closed: false, onmessage: null, onerror: null, close() { this.closed = true; } };
|
|
world.streams.push(stream);
|
|
return stream;
|
|
};
|
|
const timers = {
|
|
setTimeout(fn, delay) { const timer = { fn, delay, cleared: false }; world.timers.push(timer); return timer; },
|
|
clearTimeout(timer) { timer.cleared = true; },
|
|
};
|
|
world.dispose = () => { controller.abort(); cleanups.forEach(fn => fn()); };
|
|
world.fire = async () => {
|
|
const timer = world.timers.find(item => !item.cleared && !item.fired);
|
|
timer.fired = true;
|
|
timer.fn();
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
return timer.delay;
|
|
};
|
|
world.open = () => world.streams.filter(stream => !stream.closed);
|
|
return { world, api, scope, connect, timers, pollMs: 0, notify: message => world.toasts.push(message) };
|
|
}
|
|
|
|
const send = (stream, job) => stream.onmessage({ data: JSON.stringify(job) });
|
|
|
|
test('store merges list and SSE updates and caps backoff', async () => {
|
|
const { createActivityStore, BACKOFF_MAX_MS, LOST_DETAIL } = await source;
|
|
const { world, ...deps } = harness();
|
|
world.list = [summary(), summary({ id: ID_B, state: 'finished', finished_at: 900, counters: { added: 3, errors: 0 } })];
|
|
const store = createActivityStore(deps);
|
|
let changes = 0;
|
|
store.subscribe(() => { changes += 1; });
|
|
await store.load();
|
|
assert.equal(store.status, 'ready');
|
|
assert.deepEqual(store.jobs().map(job => [job.id, job.state]), [[ID_A, 'running'], [ID_B, 'finished']]);
|
|
assert.deepEqual(store.running().map(job => job.id), [ID_A]);
|
|
// One stream per running job only; a finished job is not followed.
|
|
assert.deepEqual(world.streams.map(stream => stream.url), [`/prompt_manager/jobs/${ID_A}/events`]);
|
|
assert.ok(changes >= 1);
|
|
|
|
// An SSE summary replaces the listed one; garbage and another job's summary are ignored.
|
|
const [stream] = world.streams;
|
|
send(stream, summary({ progress: 42, processed: 8240, total: 19622 }));
|
|
stream.onmessage({ data: 'not json' });
|
|
send(stream, summary({ id: ID_B, progress: 99 }));
|
|
assert.equal(store.find(ID_A).progress, 42);
|
|
assert.equal(store.find(ID_B).state, 'finished');
|
|
assert.deepEqual(world.toasts, []);
|
|
|
|
// The end of the job: the finish toast, and the stream is closed.
|
|
send(stream, summary({ state: 'finished', progress: 100, finished_at: 2000, counters: { added: 1420, errors: 18 } }));
|
|
assert.deepEqual(world.toasts, ['Scan finished · 1,420 added · 18 errors']);
|
|
assert.equal(stream.closed, true);
|
|
assert.deepEqual(store.running(), []);
|
|
|
|
// A new running job from another tab, then the server goes away: every re-read waits longer, up to 30 s.
|
|
world.list = [summary({ id: ID_C, state: 'running' })];
|
|
await store.load();
|
|
const second = world.open()[0];
|
|
assert.equal(second.url, `/prompt_manager/jobs/${ID_C}/events`);
|
|
world.listError = new Error('connection refused');
|
|
second.onerror();
|
|
assert.equal(second.closed, true);
|
|
const delays = [];
|
|
for (let index = 0; index < 8; index += 1) delays.push(await world.fire());
|
|
assert.deepEqual(delays, [1000, 2000, 4000, 8000, 16000, 30000, 30000, 30000]);
|
|
assert.ok(delays.every(delay => delay <= BACKOFF_MAX_MS));
|
|
// While the server cannot be read the running job shows as ended, so no pill stays.
|
|
assert.equal(store.status, 'error');
|
|
assert.equal(store.find(ID_C).state, 'lost');
|
|
assert.deepEqual(store.running(), []);
|
|
assert.equal(typeof LOST_DETAIL, 'string');
|
|
|
|
// The server is back without the job (a restart): it stays ended; finished jobs the server no longer lists are
|
|
// dropped, so the list never outgrows the server's (review M3).
|
|
world.listError = null;
|
|
world.list = [];
|
|
await world.fire();
|
|
assert.equal(store.status, 'ready');
|
|
assert.deepEqual(store.jobs().map(job => [job.id, job.state]), [[ID_C, 'lost']]);
|
|
assert.equal(world.timers.filter(timer => !timer.cleared && !timer.fired).length, 0, 'no retry after a good read');
|
|
assert.equal(world.toasts.length, 1, 'an ended job has no finish toast');
|
|
world.dispose();
|
|
});
|
|
|
|
test('a dropped stream of a still running job reattaches after the backoff, then resets it', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, ...deps } = harness();
|
|
world.list = [summary()];
|
|
const store = createActivityStore(deps);
|
|
await store.load();
|
|
world.streams[0].onerror();
|
|
assert.equal(await world.fire(), 1000);
|
|
const again = world.open();
|
|
assert.equal(again.length, 1);
|
|
assert.equal(store.find(ID_A).state, 'running');
|
|
again[0].onerror();
|
|
assert.equal(await world.fire(), 2000);
|
|
send(world.open()[0], summary({ progress: 5 }));
|
|
world.open()[0].onerror();
|
|
assert.equal(await world.fire(), 1000, 'a received message resets the backoff');
|
|
// A second drop while a re-read is already scheduled does not schedule another.
|
|
world.open()[0].onerror();
|
|
store.load();
|
|
assert.equal(world.timers.filter(timer => !timer.cleared && !timer.fired).length, 1);
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
world.dispose();
|
|
assert.ok(world.streams.every(stream => stream.closed), 'dispose closes every stream');
|
|
assert.ok(world.timers.every(timer => timer.cleared || timer.fired), 'dispose clears the pending re-read');
|
|
});
|
|
|
|
test('start returns busy with the running job', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, ...deps } = harness();
|
|
const store = createActivityStore(deps);
|
|
world.postResult = { success: true, job: summary() };
|
|
const started = await store.start('scan');
|
|
assert.equal(started.job.id, ID_A);
|
|
assert.deepEqual(world.posts, [['/prompt_manager/jobs/scan', {}]]);
|
|
assert.deepEqual(world.toasts, ['Scan started.']);
|
|
assert.equal(world.open().length, 1, 'the started job is followed');
|
|
|
|
world.postError = apiError(409, 'Scan for metadata is already running', 'busy');
|
|
world.list = [summary()];
|
|
const busy = await store.start('scan');
|
|
assert.equal(busy.busy.id, ID_A);
|
|
assert.equal(busy.busy.title, 'Scan for metadata');
|
|
assert.equal(busy.message, 'Scan for metadata is already running');
|
|
assert.equal(world.toasts.length, 1, 'a refused start has no start toast');
|
|
|
|
world.postError = apiError(500, 'boom', null);
|
|
await assert.rejects(store.start('scan'), /boom/);
|
|
world.postError = null;
|
|
world.postResult = { success: true, job: { id: 'nope' } };
|
|
await assert.rejects(store.start('scan'), /Invalid job/);
|
|
|
|
world.postResult = { success: true, cancellation_requested: true };
|
|
assert.equal(await store.cancel(ID_A), true);
|
|
assert.deepEqual(world.posts.at(-1), [`/prompt_manager/jobs/${ID_A}/cancel`, {}]);
|
|
world.postResult = { success: true, cancellation_requested: false };
|
|
assert.equal(await store.cancel(ID_A), false);
|
|
world.dispose();
|
|
});
|
|
|
|
test('a list that is not a job list counts as a failed read', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, api, ...deps } = harness();
|
|
const store = createActivityStore({ ...deps, api: { ...api, get: async () => ({ success: true }) } });
|
|
await store.load();
|
|
assert.equal(store.status, 'error');
|
|
assert.equal(world.timers.length, 1);
|
|
world.dispose();
|
|
await store.load();
|
|
assert.equal(store.status, 'error', 'a disposed store reads nothing');
|
|
});
|
|
|
|
test('a visible tab reads the list again every 15 s; a hidden one skips its turn', async () => {
|
|
const { createActivityStore, POLL_MS } = await source;
|
|
const { world, ...deps } = harness();
|
|
let visible = true;
|
|
const store = createActivityStore({ ...deps, pollMs: POLL_MS, isVisible: () => visible });
|
|
await store.load();
|
|
assert.equal(world.gets, 1);
|
|
// A job started in another tab appears on the next read.
|
|
world.list = [summary()];
|
|
assert.equal(await world.fire(), 15000);
|
|
assert.equal(world.gets, 2);
|
|
assert.deepEqual(store.running().map(job => job.id), [ID_A]);
|
|
visible = false;
|
|
assert.equal(await world.fire(), 15000);
|
|
assert.equal(world.gets, 2, 'a hidden tab does not read');
|
|
visible = true;
|
|
await world.fire();
|
|
assert.equal(world.gets, 3);
|
|
// A manual read replaces the pending poll rather than adding one.
|
|
await store.load();
|
|
assert.equal(world.timers.filter(timer => !timer.cleared && !timer.fired).length, 1);
|
|
world.dispose();
|
|
assert.ok(world.timers.every(timer => timer.cleared || timer.fired));
|
|
});
|
|
|
|
test('a late or older answer never brings an ended job back, so it is announced once', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, api, ...deps } = harness();
|
|
const answers = [];
|
|
const slowApi = { ...api, get: () => new Promise(resolve => answers.push(resolve)) };
|
|
const store = createActivityStore({ ...deps, api: slowApi });
|
|
const first = store.load();
|
|
const second = store.load();
|
|
// The newer read answers first; the older one, answering later with stale data, is dropped.
|
|
answers[1]({ success: true, jobs: [summary()] });
|
|
await second;
|
|
answers[0]({ success: true, jobs: [] });
|
|
await first;
|
|
assert.deepEqual(store.jobs().map(job => job.state), ['running']);
|
|
send(world.open()[0], summary({ state: 'finished', counters: { added: 2, errors: 0 } }));
|
|
assert.deepEqual(world.toasts, ['Scan finished · 2 added · 0 errors']);
|
|
// A list read sent before the end still says running: the job stays finished and no stream reopens.
|
|
const third = store.load();
|
|
answers[2]({ success: true, jobs: [summary()] });
|
|
await third;
|
|
assert.equal(store.find(ID_A).state, 'finished');
|
|
assert.equal(world.open().length, 0);
|
|
assert.equal(world.toasts.length, 1);
|
|
// A stale failure is dropped too.
|
|
const failing = store.load();
|
|
const newest = store.load();
|
|
answers[4]({ success: true, jobs: [] });
|
|
await newest;
|
|
answers[3](Promise.reject(new Error('late failure')));
|
|
await failing;
|
|
assert.equal(store.status, 'ready');
|
|
world.dispose();
|
|
});
|
|
|
|
test('summaries are validated and the copy follows the spec', async () => {
|
|
const { parseJob, finishToast, startToast, pillParts, outcomeLine, backoffDelay } = await source;
|
|
for (const raw of [null, [], { id: 'x', state: 'running' }, summary({ state: 'lost' }), summary({ id: ID_A.toUpperCase() })]) {
|
|
assert.equal(parseJob(raw), null);
|
|
}
|
|
const loose = parseJob({ id: ID_A, state: 'running', progress: 140.4, processed: -1, total: 1.5, counters: { added: 'x', errors: 2 },
|
|
title: ' ', started_at: 'x', scope: [] });
|
|
assert.deepEqual(loose, { id: ID_A, kind: '', title: 'Background job', state: 'running', started_at: null, finished_at: null,
|
|
progress: 100, status_line: '', processed: 0, total: 0, counters: { added: 0, errors: 2 }, scope: {}, listed: {}, log_length: 0 });
|
|
assert.equal(parseJob(summary({ progress: -3 })).progress, 0);
|
|
|
|
const scan = parseJob(summary({ progress: 42, processed: 8240, total: 19622, counters: { added: 1, errors: 1 } }));
|
|
assert.deepEqual(pillParts(scan), { label: 'Scanning · 42%', files: ' · 8,240 / 19,622', text: 'Scanning · 42% · 8,240 / 19,622' });
|
|
assert.equal(pillParts({ ...scan, total: 0 }).text, 'Scanning · 42%');
|
|
assert.equal(pillParts({ ...scan, kind: 'thumbnails', title: 'Thumbnails' }).text, 'Thumbnails · 42% · 8,240 / 19,622');
|
|
assert.equal(finishToast({ ...scan, state: 'finished' }), 'Scan finished · 1 added · 1 error');
|
|
assert.equal(finishToast({ ...scan, state: 'cancelled' }), 'Scan cancelled · 1 added');
|
|
assert.equal(finishToast({ ...scan, state: 'failed' }), 'Scan failed · 1 added · 1 error');
|
|
assert.equal(finishToast({ ...scan, counters: {}, state: 'finished' }), 'Scan finished · 0 added · 0 errors');
|
|
const other = { ...scan, kind: 'backup', title: 'Backup' };
|
|
assert.equal(finishToast({ ...other, state: 'finished' }), 'Backup finished.');
|
|
assert.equal(finishToast({ ...other, state: 'cancelled' }), 'Backup was cancelled.');
|
|
assert.equal(finishToast({ ...other, state: 'failed' }), 'Backup failed.');
|
|
assert.equal(startToast(scan), 'Scan started.');
|
|
assert.equal(startToast(other), 'Backup started.');
|
|
assert.equal(outcomeLine(scan), '1 added · 1 error');
|
|
assert.equal(outcomeLine(other), '');
|
|
assert.deepEqual([-1, 0, 1, 4, 5, 6, 1e9, 0.5].map(backoffDelay), [1000, 1000, 2000, 16000, 30000, 30000, 30000, 1000]);
|
|
});
|
|
|
|
test('pagehide stops polling, closes streams and aborts the in-flight read; pageshow resumes', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, api, ...deps } = harness();
|
|
const lifecycle = new EventTarget();
|
|
const signals = [];
|
|
let answer = null;
|
|
const pendingApi = { ...api, get: (url, options) => {
|
|
signals.push(options.signal);
|
|
world.gets += 1;
|
|
if (world.gets === 1) return Promise.resolve({ success: true, jobs: [summary()] });
|
|
return new Promise((resolve, reject) => {
|
|
answer = resolve;
|
|
options.signal.addEventListener('abort', () => reject(Object.assign(new Error('aborted'), { name: 'AbortError' })));
|
|
});
|
|
} };
|
|
const store = createActivityStore({ ...deps, api: pendingApi, pollMs: 15000, lifecycle });
|
|
await store.load();
|
|
assert.equal(world.open().length, 1);
|
|
const poll = world.timers.find(timer => !timer.cleared && timer.delay === 15000);
|
|
assert.ok(poll, 'a visible tab polls');
|
|
// A read in flight when the page goes away (a closed tab, a test teardown) is aborted, nothing is scheduled again.
|
|
const read = store.load();
|
|
lifecycle.dispatchEvent(new Event('pagehide'));
|
|
await read;
|
|
assert.equal(signals.at(-1).aborted, true);
|
|
assert.equal(world.open().length, 0);
|
|
assert.ok(world.timers.every(timer => timer.cleared || timer.fired));
|
|
assert.equal(store.find(ID_A).state, 'running', 'a hidden page marks nothing lost');
|
|
await store.load();
|
|
assert.equal(world.gets, 2, 'a suspended store reads nothing');
|
|
// Back from the cache, it reads the list again and follows the job.
|
|
const resumed = Object.assign(new Event('pageshow'), { persisted: true });
|
|
lifecycle.dispatchEvent(resumed);
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
assert.equal(world.gets, 3);
|
|
answer({ success: true, jobs: [summary()] });
|
|
await new Promise(resolve => setImmediate(resolve));
|
|
assert.equal(world.open().length, 1);
|
|
world.dispose();
|
|
assert.equal(signals.at(-1).aborted, true, 'disposing aborts the store reads');
|
|
assert.equal(world.open().length, 0);
|
|
});
|
|
|
|
test('one failed read is not a restart; two in a row are, and a good read resets the backoff', async () => {
|
|
const { createActivityStore } = await source;
|
|
const { world, ...deps } = harness();
|
|
world.list = [summary(), summary({ id: ID_B, state: 'finished', finished_at: 900 })];
|
|
const store = createActivityStore(deps);
|
|
await store.load();
|
|
// A single failed read (a proxy's 502) keeps the job running and its stream open (review M1).
|
|
world.listError = apiError(502, 'Bad gateway', null);
|
|
await store.load();
|
|
assert.equal(store.status, 'error');
|
|
assert.equal(store.find(ID_A).state, 'running');
|
|
assert.equal(world.open().length, 1);
|
|
assert.equal(await world.fire(), 1000);
|
|
// The second failure in a row: the server is gone, so the running job ended with it.
|
|
assert.equal(store.find(ID_A).state, 'lost');
|
|
assert.equal(world.open().length, 0);
|
|
for (let index = 0; index < 3; index += 1) await world.fire();
|
|
// The server is back with a new running job: the backoff starts again at 1 s (review M2).
|
|
world.listError = null;
|
|
world.list = [summary({ id: ID_C })];
|
|
await world.fire();
|
|
assert.equal(store.status, 'ready');
|
|
assert.deepEqual(store.jobs().map(job => [job.id, job.state]), [[ID_C, 'running'], [ID_A, 'lost']]);
|
|
world.open()[0].onerror();
|
|
assert.equal(await world.fire(), 1000);
|
|
// A summary carries what its detail lists, for the job view's reads (review M12).
|
|
send(world.open()[0], summary({ id: ID_C, listed: { added: 3, errors: -1 }, log_length: 7 }));
|
|
assert.deepEqual([store.find(ID_C).listed, store.find(ID_C).log_length], [{ added: 3, errors: 0 }, 7]);
|
|
world.dispose();
|
|
});
|
|
|
|
test('numbers group the way the server writes them, whatever the browser locale', async () => {
|
|
const { pillParts, finishToast, parseJob } = await source;
|
|
const original = Number.prototype.toLocaleString;
|
|
Number.prototype.toLocaleString = function (locale, ...rest) { return original.call(this, locale ?? 'de-DE', ...rest); };
|
|
try {
|
|
const job = parseJob(summary({ progress: 41, processed: 8240, total: 19622, counters: { added: 1420, errors: 18 } }));
|
|
assert.equal(pillParts(job).text, 'Scanning · 41% · 8,240 / 19,622');
|
|
assert.equal(finishToast({ ...job, state: 'finished' }), 'Scan finished · 1,420 added · 18 errors');
|
|
} finally {
|
|
Number.prototype.toLocaleString = original;
|
|
}
|
|
});
|