https://jayreddinjr-jr-odysseus-hf.hf.space/static/js/research/jobs.js

1/**
2 * Research job queue — add, start, monitor, cancel research jobs.
3 */
4
5let _jobs = [];
6let _apiBase = '';
7let _renderCb = null;
8let _idCounter = 0;
9
10// Dismissed-from-panel IDs persist across reloads so Clear actually sticks.
11// (Items still live on disk and in the Library; this just hides them here.)
12const _DISMISSED_KEY = 'odysseus-research-dismissed';
13function _loadDismissed() {
14  try {
15    const raw = localStorage.getItem(_DISMISSED_KEY);
16    return new Set(raw ? JSON.parse(raw) : []);
17  } catch { return new Set(); }
18}
19function _saveDismissed(set) {
20  try { localStorage.setItem(_DISMISSED_KEY, JSON.stringify([...set])); } catch {}
21}
22function _isDismissed(id) { return _loadDismissed().has(id); }
23function _markDismissed(ids) {
24  const set = _loadDismissed();
25  for (const id of ids) set.add(id);
26  _saveDismissed(set);
27}
28
29let _activePollInterval = null;
30
31export function init(apiBase) {
32  _apiBase = apiBase;
33  _reconnectActive();
34  // Poll for active sessions periodically so research started elsewhere
35  // (e.g. by the agent via trigger_research) gets adopted into the
36  // sidebar — _reconnectActive only ran once at load before, so
37  // agent-started jobs never appeared until a page reload.
38  if (_activePollInterval) clearInterval(_activePollInterval);
39  _activePollInterval = setInterval(() => { _reconnectActive(); }, 12000);
40}
41
42// Allow an immediate adopt when the chat stream signals a new research
43// session (research_started ui_event) — faster than the 12s poll.
44export function adoptSession(sessionId) {
45  if (!sessionId || _jobs.some(j => j.id === sessionId)) return;
46  _reconnectActive();
47}
48
49async function _reconnectActive() {
50  try {
51    // Reconnect to running tasks
52    const res = await fetch(`${_apiBase}/api/research/active`, { credentials: 'same-origin' });
53    if (res.ok) {
54      const data = await res.json();
55      for (const task of (data.active || [])) {
56        if (_jobs.some(j => j.id === task.session_id)) continue;
57        const job = {
58          id: task.session_id, query: task.query, status: 'running',
59          progress: task.progress || {},
60          startedAt: task.started_at ? task.started_at * 1000 : Date.now(),
61          elapsed: task.started_at ? Date.now() - task.started_at * 1000 : 0,
62          result: null, sources: null, findings: null,
63          errorMsg: null, avgDuration: null, modelName: null,
64          settings: {}, _es: null, _timerInterval: null,
65        };
66        _jobs.push(job);
67        _connectStream(job);
68      }
69    }
70
71    // Load recent completed research from disk
72    const libRes = await fetch(`${_apiBase}/api/research/library?sort=recent&limit=20`, { credentials: 'same-origin' });
73    if (libRes.ok) {
74      const libData = await libRes.json();
75      const dismissed = _loadDismissed();
76      for (const item of (libData.research || [])) {
77        if (item.status !== 'done') continue;
78        if (dismissed.has(item.id)) continue;
79        if (_jobs.some(j => j.id === item.id)) continue;
80        const elapsed = item.duration ? _parseDuration(item.duration) : 0;
81        _jobs.push({
82          id: item.id, query: item.query, status: 'done',
83          progress: {}, startedAt: (item.started_at || 0) * 1000,
84          elapsed, result: null, sources: null, findings: null,
85          sourceCount: item.source_count || 0,
86          category: item.category || '',
87          errorMsg: null, avgDuration: null, modelName: null,
88          settings: { max_rounds: item.rounds || 8 },
89          _es: null, _timerInterval: null, _fromLibrary: true,
90        });
91      }
92    }
93
94    _notify();
95  } catch {}
96}
97
98function _parseDuration(s) {
99  if (!s) return 0;
100  const m = s.match(/(\d+)/);
101  return m ? parseInt(m[1], 10) * 1000 : 0;
102}
103export function setRenderCallback(cb) { _renderCb = cb; }
104export function getJobs() { return _jobs; }
105
106export function addToQueue(query, settings) {
107  const job = _makeJob(query, settings);
108  _jobs.push(job);
109  _notify();
110  return job;
111}
112
113export async function startJob(query, settings) {
114  const job = addToQueue(query, settings);
115  await _launchJob(job);
116  return job;
117}
118
119export async function startQueued(jobId) {
120  const job = _jobs.find(j => j.id === jobId);

121  if (!job || job.status !== 'queued') return;
122  await _launchJob(job);
123}
124
125export async function startAllQueued() {
126  const queued = _jobs.filter(j => j.status === 'queued');
127  await Promise.all(queued.map(j => _launchJob(j)));
128}
129
130/** Run queued jobs one at a time — waits for each to finish before launching
131 *  the next. Useful when you want to avoid hammering the same model server. */
132export async function startAllQueuedSequential() {
133  const queued = _jobs.filter(j => j.status === 'queued');
134  for (const job of queued) {
135    await _launchJob(job);
136    // Wait until this specific job is no longer running
137    await new Promise(resolve => {
138      const tick = setInterval(() => {
139        if (job.status !== 'running') { clearInterval(tick); resolve(); }
140      }, 1000);
141    });
142  }
143}
144
145export async function retryJob(jobId) {
146  const job = _jobs.find(j => j.id === jobId);
147  if (!job) return;
148  job.status = 'queued';
149  job.progress = {};
150  job.errorMsg = null;
151  job.result = null;
152  job.sources = null;
153  job.findings = null;
154  job.elapsed = 0;
155  job.avgDuration = null;
156  _notify();
157  await _launchJob(job);
158}
159
160export async function cancelJob(id) {
161  const job = _jobs.find(j => j.id === id);
162  if (!job) return;
163  if (job.status === 'queued') { job.status = 'cancelled'; _notify(); return; }
164  try { await fetch(`${_apiBase}/api/research/cancel/${id}`, { method: 'POST', credentials: 'same-origin' }); } catch {}
165  _finishJob(job, 'cancelled');
166}
167
168export function removeJob(id) {
169  const idx = _jobs.findIndex(j => j.id === id);
170  if (idx >= 0) {
171    const job = _jobs[idx];
172    // Persist dismissal so it doesn't reappear from the library on reload.
173    if (job.status === 'done') _markDismissed([id]);
174    _jobs.splice(idx, 1);
175  }
176  _notify();
177}
178
179export function clearAll() {
180  // Mark all completed jobs as dismissed so they don't reappear on reload.
181  const doneIds = _jobs.filter(j => j.status === 'done').map(j => j.id);
182  if (doneIds.length) _markDismissed(doneIds);
183  for (const job of _jobs) {
184    if (job._es) { job._es.close(); job._es = null; }
185    if (job._timerInterval) { clearInterval(job._timerInterval); job._timerInterval = null; }
186  }
187  _jobs = [];
188  _notify();
189}
190
191export function formatElapsed(ms) {
192  if (!ms) return '0:00';
193  const s = Math.floor(ms / 1000);
194  return `${Math.floor(s / 60)}:${String(s % 60).padStart(2, '0')}`;
195}
196
197export function formatPhase(progress, maxRounds) {
198  if (!progress || !progress.phase) return 'Starting...';
199  const p = progress;
200  const rn = p.round ? (maxRounds ? `Round ${p.round}/${maxRounds}: ` : `Round ${p.round}: `) : '';
201  switch (p.phase) {
202    case 'probing': return 'Probing model...';
203    case 'planning': return 'Planning research strategy...';
204    case 'searching': return `${rn}Searching (${p.queries || 0} queries)`;
205    case 'reading': return `${rn}Reading ${p.total_sources || 0} sources`;
206    case 'analyzing': return `${rn}Analyzing ${p.total_findings || 0} findings`;
207    case 'writing': return `Writing report -- ${p.total_sources || 0} sources`;
208    default: return p.phase;
209  }
210}
211
212function _makeJob(query, settings) {
213  return {
214    id: `pending-${++_idCounter}`,
215    query, settings, status: 'queued',
216    progress: {}, startedAt: null, elapsed: 0,
217    result: null, sources: null, findings: null,
218    category: settings?.category || '',
219    errorMsg: null, avgDuration: null,
220    modelName: null, endpointName: null,
221    _es: null, _timerInterval: null,
222  };
223}
224
225async function _launchJob(job) {
226  const body = { query: job.query, ...job.settings };
227  let data;
228  try {
229    const res = await fetch(`${_apiBase}/api/research/start`, {
230      method: 'POST', credentials: 'same-origin',
231      headers: { 'Content-Type': 'application/json' },
232      body: JSON.stringify(body),
233    });
234    if (!res.ok) {
235      const txt = await res.text();
236      try { job.errorMsg = JSON.parse(txt).detail || txt; } catch { job.errorMsg = txt; }
237      job.status = 'error';
238      _notify();
239      return;
240    }
241    data = await res.json();
242  } catch (e) {
243    job.errorMsg = e.message;
244    job.status = 'error';
245    _notify();
246    return;
247  }
248  job.id = data.session_id;
249  job.status = 'running';
250  job.startedAt = Date.now();
251  _connectStream(job);
252  _notify();
253}
254
255function _connectStream(job) {
256  job._timerInterval = setInterval(() => {

257    job.elapsed = Date.now() - job.startedAt;
258    _notify();
259  }, 1000);
260
261  const es = new EventSource(`${_apiBase}/api/research/stream/${job.id}`);
262  job._es = es;
263
264  es.onmessage = (evt) => {
265    try {
266      const d = JSON.parse(evt.data);
267      if (d.status === 'not_found') { _finishJob(job, 'error'); return; }
268      job.progress = d;
269      if (d.model && !job.modelName) job.modelName = d.model;
270      if (d.final) {
271        if (d.error) job.errorMsg = d.error;
272        _finishJob(job, d.status === 'done' ? 'done' : d.status === 'cancelled' ? 'cancelled' : 'error');
273        if (d.status === 'done') _fetchResult(job);
274        return;
275      }
276      _notify();
277    } catch {}
278  };
279
280  es.onerror = () => {
281    es.close();
282    if (job.status === 'running') setTimeout(() => _pollFallback(job), 3000);
283  };
284}
285
286async function _pollFallback(job) {
287  if (job.status !== 'running') return;
288  try {
289    const res = await fetch(`${_apiBase}/api/research/status/${job.id}`, { credentials: 'same-origin' });
290    if (!res.ok) { _finishJob(job, 'error'); return; }
291    const d = await res.json();
292    job.progress = d.progress || {};
293    if (d.avg_duration) job.avgDuration = d.avg_duration;
294    if (d.status !== 'running') {
295      _finishJob(job, d.status === 'done' ? 'done' : 'error');
296      if (d.status === 'done') _fetchResult(job);
297      return;
298    }
299    setTimeout(() => _pollFallback(job), 2000);
300  } catch { _finishJob(job, 'error'); }
301}
302
303function _finishJob(job, status) {
304  job.status = status;
305  if (job._es) { job._es.close(); job._es = null; }
306  if (job._timerInterval) { clearInterval(job._timerInterval); job._timerInterval = null; }
307  job.elapsed = Date.now() - (job.startedAt || Date.now());
308  if (status === 'done') {
309    if ('Notification' in window && Notification.permission === 'granted') {
310      try { new Notification('Research Complete', { body: job.query.slice(0, 80) }); } catch {}
311    }
312    if (_onCompleteCb) _onCompleteCb(job);
313  }
314  _notify();
315}
316
317let _onCompleteCb = null;
318export function onComplete(cb) { _onCompleteCb = cb; }
319
320async function _fetchResult(job) {
321  try {
322    const res = await fetch(`${_apiBase}/api/research/result-peek/${job.id}`, {
323      method: 'POST', credentials: 'same-origin',
324    });
325    if (!res.ok) return;
326    const d = await res.json();
327    job.result = d.result;
328    job.sources = d.sources;
329    job.findings = d.raw_findings;
330    if (d.category && !job.category) job.category = d.category;
331    _notify();
332  } catch {}
333}
334
335function _notify() { if (_renderCb) _renderCb(); }