Read only the schedules whose results changed, using the engine's agent status

Results polled one drafts endpoint per schedule every minute, archived ones included, so the Network tab filled with video-drafts reads. Each poll now reads GET /agents/status once (with If-None-Match, mostly a 304) and reads a schedule's drafts again only when its status changed, while a production plan is at work, or after 30 minutes so signed video URLs stay valid. Without the status read, every schedule is read as before. The poll is now every 30s. Also documents how Express uses the engine API in docs/engine-api.md. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-03 12:34:08 +07:00
parent 46730279cb
commit eb194df8e0
5 changed files with 262 additions and 24 deletions
--- a/docs/engine-api.md
+++ b/docs/engine-api.md
@@ -0,0 +1,145 @@
+# Alur API Express → Engine learning-lab
+
+Express tidak punya backend sendiri. Semua logika AI (perencanaan, pembuatan materi, video, riset) ada di engine **learning-lab** (FastAPI, `/api/v1`). Express hanya membuat *agent* (jadwal) beserta konfigurasinya, lalu membaca dan meninjau hasilnya.
+
+## Jalur request
+
+| Lingkungan | Jalur |
+|---|---|
+| Production | `express.bawana.com/api/*` → redirect Netlify ([netlify.toml](../netlify.toml)) → `https://learningstudio.bawana.com/api/*` |
+| Dev | Proxy Vite `/api` → `ENGINE_URL` (default `http://localhost:8000`, [vite.config.ts](../vite.config.ts)) |
+
+- Prefix `/api/v1` ditambahkan di [src/api/client.ts](../src/api/client.ts). Setiap request membawa `Authorization: Bearer <Clerk session token>`.
+- Request baca dan aksi diatur lajunya oleh `pace` di `client.ts`, supaya tidak terkena rate limit proxy (503/429).
+- Endpoint agent ada di `learning-lab/backend/app/api/v1/endpoints/agents.py` (semua di bawah `/organizations/current/agents`). Endpoint produksi konten ada di `agent_content.py` (`/organizations/current/agents/{id}/content`).
+
+Di bawah ini, `AGENTS` = `/organizations/current/agents`.
+
+## Jadwal = agent
+
+Setiap jadwal Express adalah satu agent engine. Namanya diawali prefix Express, lalu dibaca lewat `expressAgents()` di [schedules.ts](../src/express/schedules.ts).
+
+### Membuat jadwal (`createSchedule`)
+
+| Langkah | Endpoint |
+|---|---|
+| (khusus video) Ambil harga satu video sebagai budget run | `GET AGENTS/video-quote?scenes=…&duration_seconds=…` |
+| Buat agent dengan konfigurasi dari `configFor()` | `POST AGENTS` |
+| Pasang dokumen Brain (semua dokumen yang siap, atau hanya yang dipilih) | `PUT AGENTS/{id}/knowledge-sources` |
+| Aktifkan (menyetujui aturan produksi; sebelum aktif tidak ada yang berjalan) | `POST AGENTS/{id}/activate` |
+
+### Mengelola jadwal
+
+| Aksi | Endpoint |
+|---|---|
+| Ubah brief (engine mengembalikan agent ke draft, jadi langsung diaktifkan ulang atau dipause) | `PATCH AGENTS/{id}`, lalu `POST AGENTS/{id}/activate` atau `/pause` |
+| Pause / resume | `POST AGENTS/{id}/pause`, `POST AGENTS/{id}/activate` |
+| Arsipkan | `POST AGENTS/{id}/archive` |
+| Riwayat run | `GET AGENTS/{id}/runs` |
+| Kesiapan knowledge | `GET AGENTS/{id}/knowledge-readiness` |
+
+## 1. Konten materi (summary, quiz, flashcard, slide, audio, SCORM, …)
+
+**Konfigurasi agent** (`configFor`):
+
+```ts
+enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'compose.create_material']
+material_outputs: outputsFor(kind)
+content_goal: { enabled: true, weekly_target, execution_mode: 'automatic' }
+```
+
+Audio dan SCORM adalah jenis buatan Express sendiri, disusun dari output Compose. Audio adalah summary narasi; SCORM adalah pasangan summary dan quiz dari satu item rencana.
+
+**Produksi** (`AGENTS/{id}/content`):
+
+| Langkah | Endpoint |
+|---|---|
+| "Produce now", atau tick otomatis jam 09:00 WIB | `POST …/content/planning-quotes` `{ target }` |
+| Konfirmasi rencana | `POST …/content/plans/{plan}/confirm-planning` `{ revision }` |
+| Approve dan start | Dijalankan otomatis oleh tick engine (mode `automatic`) |
+| Retry satu item (maksimal 3 percobaan) | `POST …/plans/{plan}/items/{item}/retry-quote`, lalu `POST …/plans/{plan}/start` |
+| Status rencana dan periode minggu ini | `GET …/content` (di-cache 30 detik oleh `productionStatus`) |
+
+**Di sisi engine** (`backend/app/services/`):
+
+- `agent_content_planning.py`: model mengusulkan topik beserta bukti dari isi Brain.
+- `agent_content_execution.py`: membuat quote dan memesan kredit untuk setiap item.
+- `agent_materials.py`: menjalankan Compose (`compose_creation`) untuk membuat materinya.
+
+**Hasil**: `material-drafts`.
+
+## 2. Video
+
+**Konfigurasi agent**:
+
+```ts
+enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'studio.create_video']
+video_options: { max_scenes, max_duration_seconds, art_style: 'realistic', language }
+run_schedule: { cadence: 'weekly' | 'daily', hour: 9, minute: 0, weekday: 0 }
+```
+
+**Memicu run**: tick jadwal di engine, atau `runNow`, yaitu `POST AGENTS/{id}/runs` dengan header `Idempotency-Key`.
+
+**Di sisi engine**, `agent_worker.py` menjalankan:
+
+`agent_video_drafts.py` (membuat project Video Studio dan memesan kredit) → `agent_video_script.py` → `agent_video_storyboard.py` → `agent_video_clips.py` → `agent_video_render.py`
+
+**Hasil**: `video-drafts`. Satu draft tidak bisa di-retry sendiri; yang dilakukan adalah run baru dari jadwalnya, dan tombol retry hanya muncul di draft terbaru.
+
+## 3. Riset sumber web
+
+**Konfigurasi agent**:
+
+```ts
+enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'browse.search', 'library.add_url']
+research_options: { max_results: 5 }
+auto_approve_sources
+```
+
+**Memicu run**: jam 07:00 WIB, atau `POST AGENTS/{id}/runs`.
+
+**Hasil**: `source-drafts`.
+
+## Membaca dan meninjau hasil ([results.ts](../src/express/results.ts))
+
+| Jenis | Daftar | Detail | Aksi |
+|---|---|---|---|
+| Materi | `GET AGENTS/{id}/material-drafts` | `GET AGENTS/{id}/material-drafts/{draft}` | `POST …/material-drafts/{draft}/decision` `{ expected_revision, decision }` |
+| Video | `GET AGENTS/{id}/video-drafts` | (judul diambil dari daftar) | `POST …/video-drafts/{draft}/review` `{ decision }`, `POST …/video-drafts/{draft}/cancel` |
+| Sumber | `GET AGENTS/{id}/source-drafts` | — | `POST …/source-drafts/{draft}/review` `{ expected_revision, decision }` |
+
+Ekspor slide deck ke PPTX memakai `POST /slide-deck/{deckId}/export`.
+
+### Pola polling
+
+`listResults()` dipakai oleh halaman Results dan Production (query key `['results']`, `RESULTS_POLL` = setiap 30 detik):
+
+- Setiap poll membaca `GET AGENTS/status?archived=true` satu kali, dengan `If-None-Match` (opsi `revalidate` di `client.ts`). Kalau tidak ada perubahan, engine menjawab 304.
+- Dari respons itu dibuat tanda per agent. Endpoint per jadwal (`material-drafts`, `video-drafts`, `source-drafts`) hanya dibaca ulang untuk agent yang tandanya berubah.
+- Pembacaan ulang tetap dipaksa untuk jadwal konten dengan plan aktif (topik plan belum punya draft), setelah 30 menit (URL video bertanda tangan berlaku 1 jam), dan setelah setiap mutation (`forgetReads`, [main.tsx](../src/main.tsx)).
+- Kalau `/status` sibuk, tanda terakhir dipakai. Kalau endpoint tidak tersedia, semua jadwal dibaca seperti dulu; jadwal arsip tetap cukup dibaca sekali.
+- Paling banyak 4 request per jadwal berjalan bersamaan (`PER_SCHEDULE_LIMIT`). Konten dibaca dulu, baru riset.
+- Kalau engine membalas 503/429, request dicoba sekali lagi setelah 3 detik. Kalau tetap gagal, hasil poll sebelumnya yang dipakai.
+- Di engine, `_video_draft_out` menjalankan sekitar 5 query untuk setiap draft (project, run step, dokumen, scene, kredit), dengan batas 100 draft per agent.
+
+## Brain dan kredit ([src/api/engine.ts](../src/api/engine.ts))
+
+| Aksi | Endpoint |
+|---|---|
+| Daftar / status / hapus / unduh dokumen | `GET /upload/list`, `GET /upload/{id}`, `DELETE /upload/{id}`, `GET /upload/{id}/file` |
+| Upload | `POST /upload/` |
+| Format yang didukung | `GET /upload/supported-formats` |
+| Knowledge graph dokumen | `GET /upload/{id}/knowledge`, `POST /upload/{id}/knowledge/build` |
+| Proses ulang | `POST /upload/{id}/reprocess` |
+| Versi baru atau bukan | `POST /upload/{id}/versions/link` atau `/decline` |
+| Saldo kredit | `GET /organizations/me` |
+| Pemilik jadwal | `GET /auth/me`, `GET /organizations/current/dashboard/members` |
+
+## Kode lama yang tidak dipakai
+
+Fungsi berikut di `src/api/engine.ts` tidak dipanggil di mana pun di `src/`. Ini sisa alur versi awal, yang memakai satu agent tunggal bernama `Learning Lab Express` dan memanggil Compose langsung:
+
+- `generateText`, `generateQuiz`: `POST /compose/sets`, `/generate/summary`, `/generate/quiz`
+- `ensureAgent`, `attachSources`, `setSchedule`, `research`
+
+`resetAgentCache` masih dipakai oleh `AuthGate.tsx`.
--- a/src/api/client.test.ts
+++ b/src/api/client.test.ts
@@ -1,5 +1,5 @@
-import { describe, expect, it } from 'vitest'
-import { limiter, pacer } from './client'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { limiter, pacer, request } from './client'

 describe('limiter', () => {
  it('runs at most n calls at once, all of them, failures included', async () => {
@@ -37,3 +37,21 @@ describe('pacer', () => {
    expect(Date.now() - t0).toBeGreaterThanOrEqual(45)
  })
 })
+
+describe('request with revalidate', () => {
+  afterEach(() => vi.unstubAllGlobals())
+
+  it('sends the last ETag back and answers a 304 with the last body', async () => {
+    const sent: (string | null)[] = []
+    vi.stubGlobal('fetch', vi.fn(async (_url: string, init: RequestInit) => {
+      const tag = new Headers(init.headers).get('If-None-Match')
+      sent.push(tag)
+      return tag
+        ? new Response(null, { status: 304, headers: { ETag: 'W/"1"' } })
+        : new Response(JSON.stringify({ agents: [1] }), { status: 200, headers: { ETag: 'W/"1"' } })
+    }))
+    expect(await request('/etag-test', { revalidate: true })).toEqual({ agents: [1] })
+    expect(await request('/etag-test', { revalidate: true })).toEqual({ agents: [1] })
+    expect(sent).toEqual([null, 'W/"1"'])
+  })
+})
--- a/src/api/client.ts
+++ b/src/api/client.ts
@@ -44,12 +44,19 @@ export interface RequestOptions extends Omit<RequestInit, 'body'> {
  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
+  /** For a read the engine answers with an ETag: asks again with If-None-Match, and a 304 returns the last answer. */
+  revalidate?: boolean
 }

-export async function request<T>(path: string, { json, headers, as = 'json', paced, ...opts }: RequestOptions = {}): Promise<T> {
+// The last answer to each `revalidate` read, by path, with the ETag that identifies it.
+const validated = new Map<string, { etag: string; data: unknown }>()
+
+export async function request<T>(path: string, { json, headers, as = 'json', paced, revalidate, ...opts }: RequestOptions = {}): Promise<T> {
  const h = new Headers(headers)
  const token = await getToken()
  if (token) h.set('Authorization', `Bearer ${token}`)
+  const kept = revalidate ? validated.get(path) : undefined
+  if (kept) h.set('If-None-Match', kept.etag)
  let body = opts.body
  if (json !== undefined) {
    h.set('Content-Type', 'application/json')
@@ -66,6 +73,7 @@ export async function request<T>(path: string, { json, headers, as = 'json', pac
    throw new EngineError(t('err.unreachable'), 0)
  }
  if (res.status === 204) return null as T
+  if (res.status === 304 && kept) return kept.data as T
  if (as === 'blob' && res.ok) return (await res.blob()) as T
  const data = await res.json().catch(() => null)
  if (!res.ok) {
@@ -79,6 +87,8 @@ export async function request<T>(path: string, { json, headers, as = 'json', pac
    if (res.status === 402) err.message = creditMessage(err)
    throw err
  }
+  const etag = revalidate ? res.headers.get('ETag') : null
+  if (etag) validated.set(path, { etag, data })
  return data as T
 }

--- a/src/express/results.test.ts
+++ b/src/express/results.test.ts
@@ -147,6 +147,36 @@ describe('remembering an archived schedule', () => {
  })
 })

+describe('remembering with status marks', () => {
+  const active = { agent: { id: 9, status: 'active' } } as Schedule
+
+  it('reads a schedule again only when its mark changes', async () => {
+    const fn = vi.fn(async () => [1])
+    const read = remembering(fn, 0)
+    await read(active, 'a')
+    await read(active, 'a')
+    expect(fn).toHaveBeenCalledTimes(1)
+    await read(active, 'b')
+    expect(fn).toHaveBeenCalledTimes(2)
+  })
+
+  it('reads every time without a mark, or while the last read was not quiet', async () => {
+    const fn = vi.fn(async () => [1])
+    await remembering(fn, 0)(active)
+    await remembering(fn, 0)(active)
+    expect(fn).toHaveBeenCalledTimes(2)
+    let quiet = false
+    const read = remembering(fn, 0, () => quiet)
+    await read(active, 'a')
+    await read(active, 'a')
+    expect(fn).toHaveBeenCalledTimes(4)
+    quiet = true
+    await read(active, 'a')
+    await read(active, 'a')
+    expect(fn).toHaveBeenCalledTimes(5)
+  })
+})
+
 describe('videoStatus', () => {
  it('fails a video the engine stopped without a confirmed result', () => {
    expect(videoStatus({ status: 'clips', latest_error: 'clip_outcome_unknown' })).toBe('failed')
--- a/src/express/results.ts
+++ b/src/express/results.ts
@@ -88,6 +88,10 @@ const STOPPED = new Set<PlanItem['status']>(['failed', 'blocked', 'cancelled', '
 export const withPlanItem = (status: MaterialStatus, item: PlanItem['status'] | undefined): MaterialStatus =>
  item && STOPPED.has(item) && (status === 'preparing' || status === 'generating') ? 'failed' : status

+// Schedules whose last read had a production plan at work. A plan's topics show before they have
+// drafts, and the status read only sees drafts, so these are read on every poll.
+const planning = new Set<number>()
+
 async function materialsOf(schedule: Schedule): Promise<Material[]> {
  if (schedule.rule.type === 'content' && schedule.rule.kind === 'video') return videosOf(schedule)
  const { agent, rule } = schedule
@@ -98,6 +102,8 @@ async function materialsOf(schedule: Schedule): Promise<Material[]> {
    // the last complete one is shown rather than drafts missing their titles and planned pieces.
    productionStatus(agent.id).catch(e => { if (passing(e)) throw e; return null }),
  ])
+  if (status?.plans.some(p => PLAN_ACTIVE.has(p.status))) planning.add(agent.id)
+  else planning.delete(agent.id)
  const items = new Map<number, PlanItem & { planId: number }>()
  for (const plan of status?.plans ?? []) {
    for (const item of plan.items ?? []) if (item.run_id) items.set(item.run_id, { ...item, planId: plan.id })
@@ -216,59 +222,88 @@ const PER_SCHEDULE_LIMIT = 4
 // The proxy's rate limit (503/429), a 5xx or no answer at all passes; a 4xx (e.g. a deleted agent) does not.
 const passing = (e: unknown) => e instanceof EngineError && (e.status === 0 || e.status === 429 || e.status >= 500)

-const settledReads: Set<number>[] = []
+const settledReads: Map<number, unknown>[] = []

-/** After any action: an archived schedule's results (a decision on one) and production plans
- * (a started or retried piece) may have changed, so the next poll reads them again. */
+/** After any action: a schedule's results (a decision on one) and production plans (a started or
+ * retried piece) may have changed, so the next poll reads them again. */
 export function forgetReads() {
  for (const s of settledReads) s.clear()
  forgetProductionReads()
 }

+// A settled read is still read again after this long: a ready video's signed URL lasts an hour.
+const SETTLED_FOR = 30 * 60_000
+
 /** Wraps a per-schedule read so a passing failure does not change what the page shows: a busy
 * engine is asked once more after a pause, and if it still fails, the schedule's results from the
 * last poll that read them are used. Dropping them instead made every count on the page dip for
- * one poll and come back on the next. */
-export function remembering<T>(fn: (s: Schedule) => Promise<T[]>, pause = 3000) {
+ * one poll and come back on the next.
+ *
+ * `mark` is what the engine's status read says about the schedule now; while it is the same as at
+ * the last read, nothing the read shows has changed and that read is used again. A schedule whose
+ * last read was not `quiet` (e.g. a plan whose topics have no drafts yet) is read every time. */
+export function remembering<T>(fn: (s: Schedule) => Promise<T[]>, pause = 3000, quiet: (s: Schedule) => boolean = () => true) {
  const last = new Map<number, T[]>()
-  const settled = new Set<number>()
+  const settled = new Map<number, { mark: string; at: number }>()
  settledReads.push(settled)
-  return async (s: Schedule): Promise<T[]> => {
-    // An archived schedule makes nothing new, so after one read it costs nothing until an action.
-    const archived = s.agent.status === 'archived'
-    if (archived && settled.has(s.agent.id)) return last.get(s.agent.id)!
+  return async (s: Schedule, mark?: string): Promise<T[]> => {
+    const id = s.agent.id
+    // Without a status read, an archived schedule is the one known to make nothing new.
+    const key = mark ?? (s.agent.status === 'archived' ? 'archived' : undefined)
+    const kept = settled.get(id)
+    if (key && kept?.mark === key && Date.now() - kept.at < SETTLED_FOR) return last.get(id)!
    try {
      const read = await fn(s).catch(async e => {
        if (!(e instanceof EngineError && (e.status === 503 || e.status === 429))) throw e
        await sleep(pause)
        return fn(s)
      })
-      last.set(s.agent.id, read)
-      if (archived) settled.add(s.agent.id)
+      last.set(id, read)
+      if (key && quiet(s)) settled.set(id, { mark: key, at: Date.now() })
+      else settled.delete(id)
      return read
    } catch (e) {
-      const kept = last.get(s.agent.id)
+      const kept = last.get(id)
      if (kept && passing(e)) return kept
      throw e
    }
  }
 }

-const readMaterials = remembering(materialsOf)
+const readMaterials = remembering(materialsOf, undefined, s => !planning.has(s.agent.id))
 const readSources = remembering(sourcesOf)

-// Results cost one request per schedule (every video schedule's drafts), so they are read once a
-// minute, and a page opened or refocused within that minute reuses what was read. An action on a
-// result or schedule still refreshes them at once.
-export const RESULTS_POLL = { refetchInterval: 60_000, staleTime: 60_000 }
+interface AgentStatus { id: number }
+
+/** One read of every schedule's runs and drafts, as a mark per schedule that changes whenever
+ * anything in its results does (a draft's status or progress, a new run). The engine answers an
+ * unchanged one with 304. A busy engine keeps the last marks, rather than being sent a read of
+ * every schedule; without any (an engine that has no such read), every schedule is read. */
+let lastMarks: Map<number, string> | null = null
+async function changeMarks(): Promise<Map<number, string> | null> {
+  try {
+    const { agents } = await request<{ agents: AgentStatus[] }>(`${AGENTS}/status?archived=true`, { revalidate: true })
+    lastMarks = new Map(agents.map(a => [a.id, JSON.stringify(a)]))
+  } catch (e) {
+    if (!passing(e)) lastMarks = null
+    console.warn('Agent status could not be read', e)
+  }
+  return lastMarks
+}
+
+// One status read per poll, plus a read of each schedule whose results changed since the last one.
+// A page opened or refocused within the interval reuses what was read; an action on a result or
+// schedule still refreshes them at once.
+export const RESULTS_POLL = { refetchInterval: 30_000, staleTime: 30_000 }

 export async function listResults(): Promise<{ materials: Material[]; sources: FoundSource[] }> {
-  const schedules = await expressAgents({ archived: true })
+  const [schedules, marks] = await Promise.all([expressAgents({ archived: true }), changeMarks()])
  const content = schedules.filter(s => s.rule.type === 'content')
  const research = schedules.filter(s => s.rule.type === 'research')
+  const marked = <T,>(read: (s: Schedule, mark?: string) => Promise<T[]>) => (s: Schedule) => read(s, marks?.get(s.agent.id))
  // One shared budget: research runs after content rather than alongside it.
-  const materials = await settleLimited(content, PER_SCHEDULE_LIMIT, readMaterials)
-  const sources = await settleLimited(research, PER_SCHEDULE_LIMIT, readSources)
+  const materials = await settleLimited(content, PER_SCHEDULE_LIMIT, marked(readMaterials))
+  const sources = await settleLimited(research, PER_SCHEDULE_LIMIT, marked(readSources))
  const newest = <T extends { createdAt: string }>(a: T, b: T) => b.createdAt.localeCompare(a.createdAt)
  return { materials: keepSettled(materials).map(withDecisions).sort(newest), sources: keepSettled(sources).sort(newest) }
 }