Files
learning-lab-express/src/express/schedules.ts
Iqbal 677937eb98 Follow a running production every few seconds
A started production kept showing "AI picks new topics" for a minute
or more: each schedule's status was read once a minute, and the refresh
right after starting got the read kept from before the start. That read
is now dropped when a production starts, and while the latest production
is under way its status is read every five seconds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-30 13:50:05 +07:00

461 lines
22 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
// 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 { EngineError, limiter, opKey, request } from '../api/client'
import { listDocuments, type Agent, type RunSchedule } from '../api/engine'
import { getLang, t } from '../i18n'
import {
KINDS, LANGUAGES, SCHEDULABLE, videoCadence, weeklyTarget,
type ContentKind, type ContentRule, type OutputLanguage, type Per, type ResearchRule, type Rule,
} from './parse'
const AGENTS = '/organizations/current/agents'
export const PREFIX = 'Express · '
const TIMEZONE = 'Asia/Jakarta'
export interface ContentGoal {
enabled: boolean
weekly_target: number
execution_mode: 'automatic' | 'review'
production_cadence: 'spread' | 'weekly' | 'manual'
production_hour: number
production_minute: number
production_weekday: number
weekly_budget: number
}
export interface VideoOptions {
max_scenes: number
max_duration_seconds: number
art_style: 'realistic' | 'cartoon' | 'anime' | 'watercolor' | '3d'
language: OutputLanguage
}
export interface ExpressAgent extends Agent {
next_run_at: string | null
enabled_tools: string[]
material_outputs: { kind: ContentKind; options: Record<string, unknown> }[]
content_goal: ContentGoal | null
video_options: VideoOptions | null
auto_approve_sources: boolean
run_budget: number
monthly_budget: number
created_at: string
created_by_user_id: number | null
}
export interface Schedule {
agent: ExpressAgent
rule: Rule
active: boolean
}
// The per-day/per-week wording is not part of the engine config, so it travels in the agent name.
// Names stay in English whatever the UI language, so they parse the same for every viewer.
function scheduleName(rule: Rule): string {
const topic = rule.topic ? ` · ${rule.topic}` : ''
if (rule.type === 'research') return `${PREFIX}Research ${rule.cadence}${topic}`.slice(0, 120)
return `${PREFIX}${rule.count} ${rule.kind} per ${rule.per}${topic}`.slice(0, 120)
}
const isKind = (k: string | undefined): k is ContentKind => KINDS.includes(k as ContentKind)
/** A rule with no language chosen is written in the UI language, as before languages could be chosen. */
export const defaultLanguage = (): OutputLanguage => (getLang() === 'id' ? 'Bahasa Indonesia' : 'English')
// How the agent's English brief names each language; read back by languageOfAgent.
const LANGUAGE_TEXT: Record<OutputLanguage, string> = {
English: 'English', 'Bahasa Indonesia': 'Bahasa Indonesia', 中文: 'Simplified Chinese (中文)',
}
const languageSentence = (l: OutputLanguage) => `Write all content in ${LANGUAGE_TEXT[l]}.`
function languageOfAgent(agent: ExpressAgent): OutputLanguage | undefined {
return agent.video_options?.language ?? LANGUAGES.find(l => agent.instructions?.includes(languageSentence(l)))
}
export function ruleOf(agent: ExpressAgent): Rule {
const rule = baseRuleOf(agent)
return rule.type === 'content' && usesChosen(agent) ? { ...rule, chosen: true } : rule
}
function baseRuleOf(agent: ExpressAgent): Rule {
const topic = agent.name.split(' · ').slice(2).join(' · ')
if (agent.enabled_tools.includes('studio.create_video')) {
const daily = agent.run_schedule?.cadence === 'daily'
return { type: 'content', kind: 'video', count: 1, per: daily ? 'day' : 'week', topic, language: languageOfAgent(agent) }
}
if (agent.content_goal?.enabled && agent.material_outputs[0]) {
// " per hari" comes from schedules created before names were in English.
const per: Per = / per (day|hari)\b/.test(agent.name) ? 'day' : 'week'
const target = agent.content_goal.weekly_target
// Audio and SCORM are Express kinds made of Compose outputs, so the name carries the kind.
const named = agent.name.match(/^Express · \d+ (\w+) per /)?.[1]
const kind = isKind(named) ? named : agent.material_outputs[0].kind
return { type: 'content', kind, per, count: per === 'day' ? Math.max(1, Math.round(target / 7)) : target, topic, language: languageOfAgent(agent) }
}
const s = agent.run_schedule ?? { cadence: 'off' }
return { type: 'research', cadence: s.cadence === 'weekly' ? 'weekly' : 'daily', weekday: s.weekday ?? 0, topic, autoApprove: agent.auto_approve_sources }
}
// Archived schedules are gone from the Schedule page, but what they produced still belongs on Results.
export async function expressAgents({ archived = false } = {}): Promise<Schedule[]> {
const agents = await request<ExpressAgent[]>(AGENTS)
return agents
.filter(a => a.name.startsWith(PREFIX) && (archived || a.status !== 'archived'))
.sort((a, b) => b.created_at.localeCompare(a.created_at))
.map(agent => ({ agent, rule: ruleOf(agent), active: agent.status === 'active' }))
}
export const listSchedules = () => expressAgents()
// An organization admin sees every schedule in the organization, a member only their own (and any
// shared with them), so the Schedule page says who made each one and starts with the viewer's own.
export interface Owners { me: number; names: Map<number, string> }
/** The viewer's engine user id and the organization roster's names. The roster is admin-only;
* a member gets no names, which is fine: a member's list holds their own schedules. */
export async function scheduleOwners(): Promise<Owners> {
const [me, roster] = await Promise.all([
request<{ id: number }>('/auth/me'),
request<{ members: { user_id: number; name: string | null; email: string | null }[] }>('/organizations/current/dashboard/members')
.catch(() => ({ members: [] })),
])
return { me: me.id, names: new Map(roster.members.map(m => [m.user_id, m.name || m.email || `#${m.user_id}`])) }
}
export const isMine = (s: Schedule, owners: Owners) => s.agent.created_by_user_id === owners.me
/** Schedules made of one engine run per slot (research, video), rather than weekly production plans. */
export const isRunSchedule = (rule: Rule) => rule.type === 'research' || rule.kind === 'video'
// The engine plans each production piece from a pair of documents, so a content schedule
// with fewer ready sources than this cannot plan anything. Video and research runs need one.
export const MIN_PRODUCTION_SOURCES = 2
export const needsTwoSources = (rule: Rule) => !isRunSchedule(rule)
const TOO_FEW_SOURCES = 'At least two ready sources with complementary evidence are required'
/** The engine's planner refusal, said in the UI language; any other message as it came. */
export const productionError = (message: string) =>
(message.includes(TOO_FEW_SOURCES) ? t('sch.tooFewSources', { n: MIN_PRODUCTION_SOURCES }) : message)
// Video length is fixed per schedule: 4 scenes, 40 seconds in all (the engine allows 4–15 s per scene).
export const VIDEO = { scenes: 4, seconds: 40 }
export interface VideoQuote { max_scenes: number; max_total_seconds: number; max_credits: number }
/** The current price ceiling of one video; the engine pins and rechecks it when a run starts. */
export const videoQuote = () =>
request<VideoQuote>(`${AGENTS}/video-quote?scenes=${VIDEO.scenes}&duration_seconds=${VIDEO.seconds}`)
// Credit limits. The engine requires run ≤ weekly ≤ monthly; a run is one planned item, one research
// pass, or one video. A SCORM item is two Compose outputs (lesson + quiz), so it gets twice the room.
export function budgetFor(rule: Rule, videoCredits = 0) {
if (rule.type === 'research') {
const run = 50
return { run, weekly: null, monthly: rule.cadence === 'daily' ? run * 31 : run * 5 }
}
if (rule.kind === 'video') {
// A little headroom over today's quote, so a small price change does not stop the schedule.
const run = Math.max(1, Math.ceil((videoCredits * 11) / 10))
return { run, weekly: null, monthly: videoCadence(rule) === 'daily' ? run * 31 : run * 5 }
}
const perItem = rule.kind === 'scorm' ? 40 : 20
const weekly = Math.max(rule.kind === 'scorm' ? 120 : 60, weeklyTarget(rule) * perItem)
return { run: Math.min(rule.kind === 'scorm' ? 60 : 30, weekly), weekly, monthly: weekly * 5 }
}
const KIND_TEXT: Record<ContentKind, string> = {
summary: 'summaries', quiz: 'quizzes', flashcards: 'flashcard sets', roleplay: 'roleplay scenarios',
simulator: 'simulations', slide_deck: 'slide decks', video: 'short explainer videos', audio: 'audio lessons',
scorm: 'e-learning modules (a lesson with a quiz that checks it)',
}
// What each Express kind asks of the engine's Compose outputs.
const KIND_BRIEF: Partial<Record<ContentKind, string>> = {
audio: 'Each summary is the narration script of an audio lesson of about 3 to 5 minutes. Write it to be read aloud: '
+ 'conversational, short sentences, a friendly opening and a short recap at the end. No headings, bullet points, tables or Markdown.',
scorm: 'Each lesson summary and its quiz form one e-learning module: the quiz checks exactly what the summary teaches.',
video: 'Each video explains one practical idea from the Brain in plain language for employees.',
}
function outputsFor(kind: ContentKind): ExpressAgent['material_outputs'] {
if (kind === 'audio') return [{ kind: 'summary', options: {} }]
if (kind === 'scorm') return [{ kind: 'summary', options: {} }, { kind: 'quiz', options: { question_count: 10, question_type: 'multiple_choice_only' } }]
return [{ kind, options: {} }]
}
// The agent's brief is in English; the content language is the one chosen in the Instruction chat.
// The engine's content production has no language field, so it is part of the instructions.
function goalText(rule: Rule): string {
const about = rule.topic ? ` about ${rule.topic}` : ''
if (rule.type === 'research') {
return `Research the public internet${about || ' for learning topics relevant to the L&D team'} and add new, relevant public sources to the team's Brain.`
}
return `Create new learning ${KIND_TEXT[rule.kind]}${about} from the L&D team's Brain, without repeating content that already exists.`
}
// A schedule made from chosen material says so in its brief. That keeps it to those documents:
// editing or resuming it attaches nothing else, while a whole-Brain schedule picks up new material.
const CHOSEN_SENTENCE = 'Use only the Brain material chosen for this schedule.'
const usesChosen = (agent: ExpressAgent) => !!agent.instructions?.includes(CHOSEN_SENTENCE)
function instructionsFor(rule: Rule, chosen = false): string {
const language = languageSentence((rule.type === 'content' && rule.language) || defaultLanguage())
const brief = rule.type === 'content' ? KIND_BRIEF[rule.kind] : undefined
return [rule.topic && `Focus on the topic: ${rule.topic}.`, brief, chosen && CHOSEN_SENTENCE, language].filter(Boolean).join(' ')
}
export function configFor(rule: Rule, videoCredits = 0, chosen = false) {
const budget = budgetFor(rule, videoCredits)
const common = {
name: scheduleName(rule),
goal: goalText(rule),
instructions: instructionsFor(rule, chosen),
timezone: TIMEZONE,
run_budget: budget.run,
monthly_budget: budget.monthly,
}
if (rule.type === 'research') {
return {
...common,
enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'browse.search', 'library.add_url'],
research_options: { max_results: 5 },
// Express is hands-off: accepted sources go straight into the Brain unless the user unticks this.
auto_approve_sources: rule.autoApprove,
run_schedule: researchSchedule(rule),
}
}
if (rule.kind === 'video') {
return {
...common,
enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'studio.create_video'],
video_options: {
max_scenes: VIDEO.scenes,
max_duration_seconds: VIDEO.seconds,
art_style: 'realistic',
language: rule.language ?? defaultLanguage(),
} satisfies VideoOptions,
run_schedule: { cadence: videoCadence(rule) ?? 'weekly', hour: 9, minute: 0, weekday: 0 } satisfies RunSchedule,
}
}
return {
...common,
enabled_tools: ['knowledge.retrieve', 'compose.graph.retrieve', 'compose.create_material'],
material_outputs: outputsFor(rule.kind),
content_goal: {
enabled: true,
weekly_target: weeklyTarget(rule),
execution_mode: 'automatic',
// "spread": the engine picks delivery days; a daily rule (target ≥ 7) produces every day.
production_cadence: 'spread',
production_hour: 9,
production_minute: 0,
weekly_budget: budget.weekly!,
} satisfies Partial<ContentGoal>,
}
}
const researchSchedule = (rule: ResearchRule): RunSchedule =>
({ cadence: rule.cadence, hour: 7, minute: 0, weekday: rule.weekday })
export function canSchedule(rule: Rule): rule is ResearchRule | ContentRule {
if (rule.type === 'research') return true
if (rule.count < 1) return false
if (rule.kind === 'video') return videoCadence(rule) !== null
return SCHEDULABLE.has(rule.kind) && weeklyTarget(rule) <= 50
}
// The production corpus is the agent's own sources: the chosen documents, or every ready Brain document.
// 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)
}
}
/** With `chosen` document ids the schedule uses only those; without, the whole Brain. */
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 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 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 { name, goal, instructions } = configFor(rule, 0, usesChosen(schedule.agent))
await request(`${AGENTS}/${schedule.agent.id}`, { method: 'PATCH', json: { name, goal, instructions } })
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 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' })
/** Starts one run of a research or video schedule now, outside its daily or weekly slot. */
export const runNow = (id: number) =>
request<{ id: number; status: string }>(`${AGENTS}/${id}/runs`, { method: 'POST', headers: { 'Idempotency-Key': opKey(`run-${id}`) }, json: {} })
// ── Status and history ──
export interface ProductionStatus {
enabled: boolean
goal_status: string | null
planning_available: boolean
period: { week_start: string; target: number; ready: number; claimed: number; failed: number; shortfall: number; spent_credits: number } | null
plans: ContentPlan[]
}
export interface ContentPlan {
id: number
revision: number
status: 'quoted' | 'planning' | 'review' | 'approved' | 'running' | 'completed' | 'paused' | 'failed' | 'cancelled'
target: number
error_code: string | null
planning_quote: { credits: number | null }
items: PlanItem[]
}
export interface PlanItem {
id: number
status: 'proposed' | 'approved' | 'queued' | 'generating' | 'ready' | 'blocked' | 'failed' | 'skipped' | 'cancelled' | 'rejected'
run_id: number | null
attempt_count: number
title: string
required_outputs: { kind: ContentKind }[]
}
const DROPPED = new Set<PlanItem['status']>(['skipped', 'cancelled', 'rejected'])
// One plan's progress, for the Schedule row and the Results banner.
export function planProgress(plan: ContentPlan) {
const items = plan.items.filter(i => !DROPPED.has(i.status))
return {
total: items.length,
ready: items.filter(i => i.status === 'ready').length,
failed: items.filter(i => i.status === 'failed' || i.status === 'blocked').length,
// The engine may find fewer new topics than asked for; the rest is a shortfall, not an error.
shortfall: plan.status === 'planning' ? 0 : Math.max(0, plan.target - items.length),
}
}
// The Schedule page (one query per row) and Results (one read per schedule) both poll this; a read
// from the last half minute, or one still on its way, answers both. forgetReads drops them after an action.
const PRODUCTION_FRESH = 30_000
const productionReads = new Map<number, { at: number; read: Promise<ProductionStatus> }>()
export function productionStatus(id: number, maxAge = PRODUCTION_FRESH): Promise<ProductionStatus> {
const kept = productionReads.get(id)
if (kept && Date.now() - kept.at < maxAge) return kept.read
const read = request<ProductionStatus>(`${AGENTS}/${id}/content`)
productionReads.set(id, { at: Date.now(), read })
read.catch(() => { if (productionReads.get(id)?.read === read) productionReads.delete(id) })
return read
}
export const forgetProductionReads = () => productionReads.clear()
// Produce now = the same plan the 09:00 tick would create, just earlier. The engine's tick then
// approves and starts it within a minute because Express schedules run in automatic mode.
export const quoteProduction = (id: number, target: number) =>
request<ContentPlan>(`${AGENTS}/${id}/content/planning-quotes`, { method: 'POST', headers: { 'Idempotency-Key': opKey(`produce-${id}`) }, json: { target } })
export const confirmProduction = (id: number, plan: ContentPlan) =>
request<ContentPlan>(`${AGENTS}/${id}/content/plans/${plan.id}/confirm-planning`, { method: 'POST', json: { revision: plan.revision } })
// Retrying a failed item = a fresh generation quote for that item alone, then starting it. The engine
// allows three attempts per item and only by the person who started the first one.
export const MAX_ATTEMPTS = 3
export async function retryItem(agentId: number, planId: number, itemId: number) {
const plan = (await productionStatus(agentId)).plans.find(p => p.id === planId)
if (!plan) throw new Error('The production plan is no longer available')
const base = `${AGENTS}/${agentId}/content/plans/${planId}`
const quote = await request<{ revision: number; quote_signature: string }>(`${base}/items/${itemId}/retry-quote`, { method: 'POST', json: { revision: plan.revision } })
return request<ContentPlan>(`${base}/start`, {
method: 'POST', headers: { 'Idempotency-Key': opKey(`retry-${itemId}`) },
json: { revision: quote.revision, quote_signature: quote.quote_signature },
})
}
export const PLAN_ACTIVE = new Set<ContentPlan['status']>(['planning', 'review', 'approved', 'running'])
export interface RunSummary {
id: number
status: string
trigger_kind: string
safe_error_code: string | null
budget_spent: number | null
has_readiness_issues: boolean
/** Research runs: what came of the sources found, e.g. no_new_sources when nothing was kept. */
source_outcome?: string | null
created_at: string
finished_at: string | null
}
export const listRuns = (id: number) => request<RunSummary[]>(`${AGENTS}/${id}/runs`)
// What stops a run with knowledge_not_ready: the schedule's own material that is not ready (the whole
// Library only for an agent with no material). This is the current state, not the run's.
export interface ReadinessIssue {
kind: 'document' | 'compose_set'
resource_id: number
title: string
reason: string
action_url: string
}
export const knowledgeReadiness = (id: number) =>
request<{ ready: boolean; issues: ReadinessIssue[] }>(`${AGENTS}/${id}/knowledge-readiness`)