diff --git a/docs/engine-api.md b/docs/engine-api.md new file mode 100644 index 0000000..fd2136b --- /dev/null +++ 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 `. +- 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`. diff --git a/src/api/client.test.ts b/src/api/client.test.ts index 54683df..25cd9dd 100644 --- 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"']) + }) +}) diff --git a/src/api/client.ts b/src/api/client.ts index 1dea1b1..159c02d 100644 --- a/src/api/client.ts +++ b/src/api/client.ts @@ -44,12 +44,19 @@ export interface RequestOptions extends Omit { 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(path: string, { json, headers, as = 'json', paced, ...opts }: RequestOptions = {}): Promise { +// The last answer to each `revalidate` read, by path, with the ETag that identifies it. +const validated = new Map() + +export async function request(path: string, { json, headers, as = 'json', paced, revalidate, ...opts }: RequestOptions = {}): Promise { 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(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(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 } diff --git a/src/express/results.test.ts b/src/express/results.test.ts index 03edc07..9087608 100644 --- 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') diff --git a/src/express/results.ts b/src/express/results.ts index 6b3a952..44464c8 100644 --- a/src/express/results.ts +++ b/src/express/results.ts @@ -88,6 +88,10 @@ const STOPPED = new Set(['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() + async function materialsOf(schedule: Schedule): Promise { 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 { // 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() 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[] = [] +const settledReads: Map[] = [] -/** 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(fn: (s: Schedule) => Promise, 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(fn: (s: Schedule) => Promise, pause = 3000, quiet: (s: Schedule) => boolean = () => true) { const last = new Map() - const settled = new Set() + const settled = new Map() settledReads.push(settled) - return async (s: Schedule): Promise => { - // 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 => { + 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 | null = null +async function changeMarks(): Promise | 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 = (read: (s: Schedule, mark?: string) => Promise) => (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 = (a: T, b: T) => b.createdAt.localeCompare(a.createdAt) return { materials: keepSettled(materials).map(withDecisions).sort(newest), sources: keepSettled(sources).sort(newest) } }