diff --git a/src/api/client.ts b/src/api/client.ts index 5a5424c..1dea1b1 100644 --- a/src/api/client.ts +++ b/src/api/client.ts @@ -42,9 +42,11 @@ export interface RequestOptions extends Omit { body?: BodyInit /** 'blob' for file downloads; errors still come back as JSON. */ as?: 'json' | 'blob' + /** Waits its turn like a read: for actions sent many at a time, which would otherwise outrun the rate limit. */ + paced?: boolean } -export async function request(path: string, { json, headers, as = 'json', ...opts }: RequestOptions = {}): Promise { +export async function request(path: string, { json, headers, as = 'json', paced, ...opts }: RequestOptions = {}): Promise { const h = new Headers(headers) const token = await getToken() if (token) h.set('Authorization', `Bearer ${token}`) @@ -54,7 +56,7 @@ export async function request(path: string, { json, headers, as = 'json', ... body = JSON.stringify(json) } - if (!opts.method || opts.method === 'GET') await pace.wait() + if (paced || !opts.method || opts.method === 'GET') await pace.wait() else pace.take() let res: Response diff --git a/src/express/schedules.test.ts b/src/express/schedules.test.ts index 2df68de..34ac74f 100644 --- a/src/express/schedules.test.ts +++ b/src/express/schedules.test.ts @@ -1,6 +1,9 @@ -import { describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import type { ContentRule, ResearchRule } from './parse' -import { budgetFor, canSchedule, configFor, isMine, needsTwoSources, productionError, ruleOf, VIDEO, type ExpressAgent } from './schedules' +import { + budgetFor, canSchedule, configFor, createSchedule, isMine, missingDocuments, needsTwoSources, productionError, resumeSchedule, ruleOf, updateTopic, VIDEO, + type ExpressAgent, type Schedule, +} from './schedules' const rule = (kind: ContentRule['kind'], count = 2, per: ContentRule['per'] = 'week', topic = 'onboarding', language: ContentRule['language'] = 'English'): ContentRule => ({ type: 'content', kind, count, per, topic, language }) @@ -139,3 +142,77 @@ describe('isMine', () => { expect(isMine(schedule(null), owners)).toBe(false) }) }) + +describe('missingDocuments', () => { + it('leaves out the documents a schedule already has', () => { + expect(missingDocuments([1, 2, 3], [{ kind: 'document', id: 2 }, { kind: 'compose_set', id: 3 }])).toEqual([1, 3]) + }) +}) + +// Answers the engine's calls by method and path, and records them in order. +function fakeEngine(answer: (method: string, path: string) => { status?: number; body?: unknown }) { + const calls: string[] = [] + vi.stubGlobal('fetch', async (url: string, init: RequestInit = {}) => { + const method = init.method ?? 'GET' + const path = url.replace('/api/v1', '') + calls.push(`${method} ${path}${init.body ? ` ${init.body}` : ''}`) + const { status = 200, body = {} } = answer(method, path) + return new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/json' } }) + }) + return calls +} + +const doc = (id: number) => ({ id, filename: `d${id}.pdf`, status: 'completed', knowledge_status: 'ready', created_at: '2026-10-01T00:00:00Z' }) +const put = (id: number) => `PUT /organizations/current/agents/1/knowledge-sources {"kind":"document","source_id":${id}}` + +describe('updateTopic and resumeSchedule', () => { + afterEach(() => vi.unstubAllGlobals()) + const schedule = (active: boolean): Schedule => { + const agent = { ...agentFrom(configFor(rule('summary'))), status: active ? 'active' : 'paused' } + return { agent, rule: ruleOf(agent), active } + } + const engine = () => fakeEngine((method, path) => { + if (path === '/upload/list') return { body: [doc(1), doc(2), doc(3)] } + if (path.endsWith('/knowledge-sources') && method === 'GET') return { body: [{ kind: 'document', id: 1 }, { kind: 'document', id: 2 }] } + return { body: { id: 1, status: 'active' } } + }) + + it('re-activates right after the topic changes, then attaches only new material', async () => { + const calls = engine() + await updateTopic(schedule(true), 'leadership') + expect(calls[0]).toMatch(/^PATCH \/organizations\/current\/agents\/1 /) + expect(calls[1]).toBe('POST /organizations/current/agents/1/activate') + expect(calls.filter(c => c.startsWith('PUT'))).toEqual([put(3)]) + }) + + it('keeps a paused schedule paused', async () => { + const calls = engine() + await updateTopic(schedule(false), 'leadership') + expect(calls[1]).toBe('POST /organizations/current/agents/1/pause') + }) + + it('resumes with one activate and attaches only new material', async () => { + const calls = engine() + await resumeSchedule(schedule(false)) + expect(calls[0]).toBe('POST /organizations/current/agents/1/activate') + expect(calls.filter(c => c.includes('/activate'))).toHaveLength(1) + expect(calls.filter(c => c.startsWith('PUT'))).toEqual([put(3)]) + }) +}) + +describe('createSchedule', () => { + afterEach(() => vi.unstubAllGlobals()) + + it('attaches every ready document, skips one it cannot share, reports progress, then activates', async () => { + const calls = fakeEngine((method, path) => { + if (path === '/upload/list') return { body: [doc(1), doc(2), doc(3)] } + if (method === 'PUT') return calls.at(-1)!.includes('"source_id":2') ? { status: 404, body: { detail: 'Ready workspace source not found' } } : { body: {} } + return { body: { ...agentFrom(configFor(rule('summary'))), status: 'active' } } + }) + const progress: string[] = [] + await createSchedule(rule('summary'), undefined, (done, total) => progress.push(`${done}/${total}`)) + expect(calls.filter(c => c.startsWith('PUT')).sort()).toEqual([put(1), put(2), put(3)]) + expect(calls.at(-1)).toBe('POST /organizations/current/agents/1/activate') + expect(progress).toEqual(['0/3', '1/3', '2/3', '3/3']) + }) +}) diff --git a/src/express/schedules.ts b/src/express/schedules.ts index a474b29..fe12470 100644 --- a/src/express/schedules.ts +++ b/src/express/schedules.ts @@ -1,7 +1,7 @@ // Each Express schedule is one engine agent. Content rules use the engine's recurring production // (content_goal: weekly plans with novelty checks, started automatically); research and video rules // use run_schedule (one run per day or week; a video run renders one video). -import { opKey, request } from '../api/client' +import { EngineError, limiter, opKey, request } from '../api/client' import { listDocuments, type Agent, type RunSchedule } from '../api/engine' import { getLang, t } from '../i18n' import { @@ -265,41 +265,84 @@ export function canSchedule(rule: Rule): rule is ResearchRule | ContentRule { } // The production corpus is the agent's own sources: the chosen documents, or every ready Brain document. -async function attachBrain(agentId: number, chosen?: number[]) { - const ids = chosen ?? (await listDocuments()).filter(d => d.knowledge === 'ready').map(d => d.docId) - for (const id of ids) { - await request(`${AGENTS}/${agentId}/knowledge-sources`, { method: 'PUT', json: { kind: 'document', source_id: id } }) - .catch(() => undefined) // a document someone else cannot share is skipped, not fatal +// The engine adds them one request at a time and checks the whole workspace on each, so a Brain of +// 150 documents is 150 slow requests: they go a few at a time, paced under the proxy's rate limit. +const ATTACH_AT_ONCE = 3 +const ATTACH_TRIES = 3 + +export type AttachProgress = (done: number, total: number) => void + +async function attachOne(agentId: number, id: number, tries = ATTACH_TRIES): Promise { + try { + await request(`${AGENTS}/${agentId}/knowledge-sources`, { method: 'PUT', paced: true, json: { kind: 'document', source_id: id } }) + } catch (e) { + // A busy proxy is asked again once the pace has waited for it; a document someone else cannot + // share (404) is skipped, not fatal. + const busy = e instanceof EngineError && (e.status === 503 || e.status === 429) + if (busy && tries > 1) return attachOne(agentId, id, tries - 1) + } +} + +async function attachDocuments(agentId: number, ids: number[], onProgress?: AttachProgress) { + const one = limiter(ATTACH_AT_ONCE) + let done = 0 + onProgress?.(0, ids.length) + await Promise.all(ids.map(id => one(async () => { + await attachOne(agentId, id) + onProgress?.(++done, ids.length) + }))) +} + +/** The ready documents a whole-Brain schedule does not have yet. */ +export const missingDocuments = (ready: number[], attached: { kind: string; id: number }[]) => { + const have = new Set(attached.filter(s => s.kind === 'document').map(s => s.id)) + return ready.filter(id => !have.has(id)) +} + +const readyDocuments = async () => (await listDocuments()).filter(d => d.knowledge === 'ready').map(d => d.docId) + +// A whole-Brain schedule picks up material added since it was made; a schedule of chosen material keeps to it. +// The schedule's own change has gone through by then, so a failure here does not fail it: the next +// edit or resume attaches what is still missing. +async function attachNewMaterial(schedule: Schedule) { + if (usesChosen(schedule.agent)) return + const id = schedule.agent.id + try { + const [ready, attached] = await Promise.all([readyDocuments(), request<{ kind: string; id: number }[]>(`${AGENTS}/${id}/knowledge-sources`)]) + await attachDocuments(id, missingDocuments(ready, attached)) + } catch (e) { + console.warn(`New Brain material was not attached to schedule ${id}`, e) } - return ids.length } /** With `chosen` document ids the schedule uses only those; without, the whole Brain. */ -export async function createSchedule(rule: Rule, chosen?: number[]): Promise { +export async function createSchedule(rule: Rule, chosen?: number[], onProgress?: AttachProgress): Promise { const only = chosen?.length ? chosen : undefined // A video schedule's run budget is today's price of one video, so it is fetched first. const credits = rule.type === 'content' && rule.kind === 'video' ? (await videoQuote()).max_credits : 0 const agent = await request(AGENTS, { method: 'POST', json: configFor(rule, credits, !!only) }) - await attachBrain(agent.id, only) + await attachDocuments(agent.id, only ?? (await readyDocuments()), onProgress) // Activating approves the production rules as they stand; until then nothing runs. return request(`${AGENTS}/${agent.id}/activate`, { method: 'POST' }) } -/** Changing a rule changes what was approved, so the agent is re-activated right after. */ +/** Changing the brief changes what was approved, so the engine sets the agent back to draft; it is + * re-activated (or kept paused) at once. Sources are not part of the approval, so new material + * is attached after that, without leaving the schedule paused meanwhile. */ export async function updateTopic(schedule: Schedule, topic: string): Promise { const rule = { ...schedule.rule, topic: topic.trim() } - const chosen = usesChosen(schedule.agent) - const { name, goal, instructions } = configFor(rule, 0, chosen) + const { name, goal, instructions } = configFor(rule, 0, usesChosen(schedule.agent)) await request(`${AGENTS}/${schedule.agent.id}`, { method: 'PATCH', json: { name, goal, instructions } }) - if (!chosen) await attachBrain(schedule.agent.id) - return request(`${AGENTS}/${schedule.agent.id}/${schedule.active ? 'activate' : 'pause'}`, { method: 'POST' }) + const agent = await request(`${AGENTS}/${schedule.agent.id}/${schedule.active ? 'activate' : 'pause'}`, { method: 'POST' }) + await attachNewMaterial(schedule) + return agent } export const pauseSchedule = (id: number) => request(`${AGENTS}/${id}/pause`, { method: 'POST' }) export async function resumeSchedule(schedule: Schedule) { - const id = schedule.agent.id - if (!usesChosen(schedule.agent)) await attachBrain(id) - return request(`${AGENTS}/${id}/activate`, { method: 'POST' }) + const agent = await request(`${AGENTS}/${schedule.agent.id}/activate`, { method: 'POST' }) + await attachNewMaterial(schedule) + return agent } export const archiveSchedule = (id: number) => request(`${AGENTS}/${id}/archive`, { method: 'POST' }) diff --git a/src/i18n.ts b/src/i18n.ts index 0ae85d6..f82a688 100644 --- a/src/i18n.ts +++ b/src/i18n.ts @@ -403,6 +403,7 @@ const en = { 'agent.wholeBrain': 'OK, the whole Brain it is.', 'agent.reset': 'Starting over. What should I create?', 'agent.approving': 'Creating the schedules…', + 'agent.attaching': 'Connecting the Brain to schedule {n} of {count}: {done} of {total} documents', 'agent.done': 'Done: {n} schedules are active and start right away.', 'agent.partial': '{ok} of {n} schedules were created. The ones that failed are still in the plan: fix them, then approve again.', 'agent.optSkip': 'Skip', @@ -905,6 +906,7 @@ const id: Record = { 'agent.wholeBrain': 'Oke, pakai seluruh Otak.', 'agent.reset': 'Mulai dari awal. Apa yang perlu dibuat?', 'agent.approving': 'Membuat jadwal…', + 'agent.attaching': 'Menghubungkan Brain ke jadwal {n} dari {count}: {done} dari {total} dokumen', 'agent.done': 'Selesai: {n} jadwal aktif dan langsung berjalan.', 'agent.partial': '{ok} dari {n} jadwal berhasil dibuat. Yang gagal masih ada di rencana: perbaiki, lalu setujui lagi.', 'agent.optSkip': 'Lewati', diff --git a/src/pages/Instruction.tsx b/src/pages/Instruction.tsx index f66f32a..40986ae 100644 --- a/src/pages/Instruction.tsx +++ b/src/pages/Instruction.tsx @@ -36,6 +36,8 @@ export function InstructionPage() { const [agent, setAgent] = useState(EMPTY) const [messages, setMessages] = useState([]) const [saving, setSaving] = useState(false) + // While schedules are created: which one, and how many Brain documents it has been given so far. + const [progress, setProgress] = useState<{ n: number; count: number; done: number; total: number } | null>(null) const docs = useDocuments() const [phase, setPhase] = useState('idle') // The live question whose typing has finished, so its reply options can show. @@ -108,9 +110,10 @@ export function InstructionPage() { // Approving from the plan before the agent asked about the language uses the conversation's language. const language: OutputLanguage = state.language ?? (replyLang === 'id' ? 'Bahasa Indonesia' : 'English') // One at a time so a failure (e.g. not an organization owner) is reported per schedule. - for (const { id, rule } of items) { + for (const [i, { id, rule }] of items.entries()) { try { - await createSchedule(rule.type === 'content' ? { ...rule, language } : rule) + await createSchedule(rule.type === 'content' ? { ...rule, language } : rule, undefined, + (d, total) => setProgress({ n: i + 1, count: items.length, done: d, total })) done.push({ rule, ok: true }) created.add(id) } catch (e) { @@ -118,6 +121,7 @@ export function InstructionPage() { } } setSaving(false) + setProgress(null) // A new schedule may start producing right away; the dashboard shows it under content in progress. await Promise.all([queryClient.invalidateQueries({ queryKey: ['schedules'] }), queryClient.invalidateQueries({ queryKey: ['results'] })]) const ok = done.filter(o => o.ok).length @@ -148,6 +152,7 @@ export function InstructionPage() { {phase === 'pausing' && } {idle && live.text && ( t('agent.attaching', progress)) : undefined} onTyped={() => setTypedLive(live.text)} onType={scroll} /> )} @@ -221,8 +226,8 @@ function useTyped(text: string, enabled: boolean, onDone?: () => void, onType?: return text.slice(0, shown) } -function Bubble({ message, lang, busy, onTyped, onType }: { - message: Message; lang: Lang; busy?: boolean; onTyped?: () => void; onType?: () => void +function Bubble({ message, lang, busy, note, onTyped, onType }: { + message: Message; lang: Lang; busy?: boolean; note?: string; onTyped?: () => void; onType?: () => void }) { const text = useTyped(message.text, !!message.typed, onTyped, onType) const typing = text.length < message.text.length @@ -232,6 +237,7 @@ function Bubble({ message, lang, busy, onTyped, onType }: {
{message.thought && {inLang(lang, () => t('agent.thought', { s: message.thought! }))}}

{text}

+ {note && {note}} {message.outcomes && !typing && (
    {message.outcomes.map((o, i) => (