Attach Brain material without blocking a schedule's status

Saving a topic or resuming sent one PUT per ready document (about 150),
and a topic change left the schedule in draft (shown as paused) until
all of them were done. Creating a schedule waited on the same loop with
no sign of progress.

A topic change now re-activates (or keeps paused) right after the PATCH,
and resuming is one activate; both then attach only the documents the
schedule does not have yet. Creating a schedule still attaches every
ready document, three at a time and paced under the proxy's rate limit,
retrying a busy answer instead of skipping it, and the Instruction chat
shows how many are done.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/api/client.ts

@@ -42,9 +42,11 @@ export interface RequestOptions extends Omit<RequestInit, 'body'> {
   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<T>(path: string, { json, headers, as = 'json', ...opts }: RequestOptions = {}): Promise<T> {
+export async function request<T>(path: string, { json, headers, as = 'json', paced, ...opts }: RequestOptions = {}): Promise<T> {
   const h = new Headers(headers)
   const token = await getToken()
   if (token) h.set('Authorization', `Bearer ${token}`)
@@ -54,7 +56,7 @@ export async function request<T>(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

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'])
+  })
+})

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<void> {
+  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<ExpressAgent> {
+export async function createSchedule(rule: Rule, chosen?: number[], onProgress?: AttachProgress): Promise<ExpressAgent> {
   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<ExpressAgent>(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<ExpressAgent>(`${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<ExpressAgent> {
   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<ExpressAgent>(`${AGENTS}/${schedule.agent.id}/${schedule.active ? 'activate' : 'pause'}`, { method: 'POST' })
+  const agent = await request<ExpressAgent>(`${AGENTS}/${schedule.agent.id}/${schedule.active ? 'activate' : 'pause'}`, { method: 'POST' })
+  await attachNewMaterial(schedule)
+  return agent
 }
 
 export const pauseSchedule = (id: number) => request<ExpressAgent>(`${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<ExpressAgent>(`${AGENTS}/${id}/activate`, { method: 'POST' })
+  const agent = await request<ExpressAgent>(`${AGENTS}/${schedule.agent.id}/activate`, { method: 'POST' })
+  await attachNewMaterial(schedule)
+  return agent
 }
 export const archiveSchedule = (id: number) => request<ExpressAgent>(`${AGENTS}/${id}/archive`, { method: 'POST' })
 

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<Key, string> = {
   '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',

src/pages/Instruction.tsx

@@ -36,6 +36,8 @@ export function InstructionPage() {
   const [agent, setAgent] = useState<AgentState>(EMPTY)
   const [messages, setMessages] = useState<Message[]>([])
   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<Phase>('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' && <Typing />}
             {idle && live.text && (
               <Bubble key={live.text} message={{ role: 'agent', text: live.text, typed: true }} lang={lang} busy={saving}
+                note={saving && progress?.total ? inLang(lang, () => t('agent.attaching', progress)) : undefined}
                 onTyped={() => setTypedLive(live.text)} onType={scroll} />
             )}
           </div>
@@ -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 }: {
       <div className="ai-bubble">
         {message.thought && <small className="ai-thought">{inLang(lang, () => t('agent.thought', { s: message.thought! }))}</small>}
         <p className={typing ? 'typing' : undefined}>{text}</p>
+        {note && <small className="ai-thought" role="status">{note}</small>}
         {message.outcomes && !typing && (
           <ul className="rule-outcomes appear">
             {message.outcomes.map((o, i) => (