feat(podcast): render finished missions into episodes, and serve them as a feed
The renderer existed but nothing called it. This wires it to the missions and puts the result somewhere a phone can reach. **A sweep, not a phase step.** Rendering is not the agents' work and must not be able to fail a phase that succeeded; a transient API error simply retries next tick, and a mission already rendered is skipped because its episode row exists. `podcast_episodes` is that record — without it the sweep would re-render on every pass and re-bill for it, the same lesson `corpus_items` taught for papers. **It is racing a reaper.** script.md lives in the mission checkout, and `mission_runtime`'s sweeper deletes that tree 30 minutes after the mission reaches a terminal state. So the sweep runs every 2 minutes, leaving ~15 attempts inside the window. When it does lose — as it did for three missions that had completed hours before this shipped — it now SAYS so and records a marker rather than skipping in silence, which is how a feed ends up quietly missing a day. The feed filters those markers out: a zero-byte enclosure shows a broken episode in a podcast app, where showing nothing is honest. **Duration is read from the audio, not estimated from the script.** The feed advertises a length and that length should be the real one — and it is the check that catches a 6 MB file playing for six seconds. **The feed authenticates by query-string token**, because no podcast app can set headers. That is a real trade: the token lands in the app's database and any proxy log. It reuses `AuthService::authenticate`, so revoking the session revokes the feed with it rather than creating a second secret to forget to rotate. Titles are XML-escaped — one raw ampersand makes a client reject the WHOLE feed, not one episode. 363 tests pass. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
co-authored by
Claude Opus 5
Claude Opus 5
parent
55b16f25c8
commit
1f39f642a3
@@ -438,3 +438,251 @@ mod live {
|
||||
assert!(bytes.len() > 100_000, "suspiciously small: {} bytes", bytes.len());
|
||||
}
|
||||
}
|
||||
|
||||
// ── Rendering a finished mission into an episode ──────────────────────
|
||||
|
||||
/// Where a mission's script lives inside its checkout.
|
||||
pub fn script_path(date: &str) -> String {
|
||||
format!("ContinuousResearch/{date}/script.md")
|
||||
}
|
||||
|
||||
/// Blob key for an episode's audio.
|
||||
pub fn blob_key(mission_id: uuid::Uuid, date: &str) -> String {
|
||||
format!("podcast/{date}/{mission_id}.mp3")
|
||||
}
|
||||
|
||||
/// Duration of a joined CBR stream, from its frame headers.
|
||||
///
|
||||
/// Read from the audio rather than estimated from the script, because the
|
||||
/// estimate is what a listener is NOT owed: the feed advertises a length and
|
||||
/// that length should be the real one. Also the check that caught a six-second
|
||||
/// "episode" — a file can be 6 MB and still play for seconds.
|
||||
pub fn duration_secs(mp3: &[u8]) -> u32 {
|
||||
let mut i = 0usize;
|
||||
let mut seconds = 0f64;
|
||||
while i + 4 <= mp3.len() {
|
||||
if mp3[i] == 0xFF && mp3[i + 1] & 0xE0 == 0xE0 {
|
||||
let br = MP3_BITRATES[((mp3[i + 2] >> 4) & 0x0F) as usize];
|
||||
let sr = MP3_RATES[((mp3[i + 2] >> 2) & 0x03) as usize];
|
||||
if br > 0 && sr > 0 {
|
||||
let pad = ((mp3[i + 2] >> 1) & 1) as usize;
|
||||
let len = (144 * br as usize * 1000 / sr as usize) + pad;
|
||||
seconds += 1152.0 / sr as f64;
|
||||
i += len.max(4);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
i += 1;
|
||||
}
|
||||
seconds.round() as u32
|
||||
}
|
||||
|
||||
fn sha_hex(bytes: &[u8]) -> String {
|
||||
use sha2::{Digest, Sha256};
|
||||
format!("{:x}", Sha256::digest(bytes))
|
||||
}
|
||||
|
||||
/// Render every finished Continuous Research mission that has a script and no
|
||||
/// episode yet.
|
||||
///
|
||||
/// Driven from a sweep rather than the phase itself: rendering is not the
|
||||
/// agents' work and must not be able to fail a phase that succeeded. It is also
|
||||
/// retryable by construction — a run that fails on a transient API error is
|
||||
/// simply picked up next tick, and one that succeeded is skipped because the
|
||||
/// episode row exists.
|
||||
pub async fn render_pending(
|
||||
pool: &sqlx::PgPool,
|
||||
blobs: &std::sync::Arc<dyn cm_files::BlobStore>,
|
||||
backend: &dyn AudioBackend,
|
||||
) -> Result<usize, String> {
|
||||
use sqlx::Row;
|
||||
let rows = sqlx::query(
|
||||
"SELECT m.id, m.workspace_id, m.title
|
||||
FROM missions m
|
||||
WHERE m.template_kind = $1
|
||||
AND m.status IN ('completed', 'failed')
|
||||
AND NOT EXISTS (SELECT 1 FROM podcast_episodes e WHERE e.mission_id = m.id)
|
||||
ORDER BY m.completed_at DESC NULLS LAST
|
||||
LIMIT 3",
|
||||
)
|
||||
.bind(crate::continuous_research::TEMPLATE_KIND)
|
||||
.fetch_all(pool)
|
||||
.await
|
||||
.map_err(|e| format!("select missions to render: {e}"))?;
|
||||
|
||||
let mut made = 0usize;
|
||||
for row in rows {
|
||||
let mission_id: uuid::Uuid = row.get("id");
|
||||
let workspace_id: uuid::Uuid = row.get("workspace_id");
|
||||
let mission_title: String = row.get("title");
|
||||
|
||||
// A `failed` mission is included on purpose: the script phase may have
|
||||
// written a perfectly good script and failed its judge. The audio is
|
||||
// worth having either way, and the mission record still says it failed.
|
||||
let date = crate::continuous_research::today();
|
||||
let checkout = crate::mission_workspace::checkout_path(mission_id);
|
||||
let mut path = checkout.join(script_path(&date));
|
||||
if !path.is_file() {
|
||||
// The mission may have run yesterday; take the newest script it has
|
||||
// rather than assuming the render happens on the same UTC day.
|
||||
match newest_script(&checkout) {
|
||||
Some(p) => path = p,
|
||||
None => {
|
||||
// NEVER silent. The checkout is deleted 30 minutes after a
|
||||
// mission reaches a terminal state (`mission_runtime`'s
|
||||
// sweeper tears down the container and the tree with it), so
|
||||
// a script that is not here is not late — it is gone, and
|
||||
// this mission will never produce an episode. Saying so is
|
||||
// the difference between a known gap and a feed that is
|
||||
// quietly missing a day.
|
||||
//
|
||||
// The audio is recoverable by hand: the script was pushed to
|
||||
// the phase's own vault branch by `mission_delivery`.
|
||||
record_unrenderable(pool, mission_id, &checkout).await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
}
|
||||
let md = match std::fs::read_to_string(&path) {
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
eprintln!("podcast: cannot read {}: {e}", path.display());
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let script = parse_script(&md);
|
||||
if script.turns.is_empty() {
|
||||
eprintln!("podcast: {} has no spoken turns — skipping", path.display());
|
||||
continue;
|
||||
}
|
||||
|
||||
let audio = match backend.render(&script).await {
|
||||
Ok(a) => a,
|
||||
Err(e) => {
|
||||
// Loud, and NOT fatal to the sweep: one mission's transient API
|
||||
// failure must not stop the others being rendered.
|
||||
eprintln!("podcast: render failed for mission {mission_id}: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let secs = duration_secs(&audio);
|
||||
let key = blob_key(mission_id, &date);
|
||||
if let Err(e) = blobs.put(&key, &audio).await {
|
||||
eprintln!("podcast: shelving {key} failed: {e}");
|
||||
continue;
|
||||
}
|
||||
|
||||
let title = if script.title.trim().is_empty() {
|
||||
mission_title
|
||||
} else {
|
||||
script.title.clone()
|
||||
};
|
||||
if let Err(e) = sqlx::query(
|
||||
"INSERT INTO podcast_episodes
|
||||
(id, workspace_id, mission_id, episode_date, title, blob_key,
|
||||
bytes, duration_secs, rendered_by, script_sha)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
||||
ON CONFLICT (mission_id) DO UPDATE
|
||||
SET title = EXCLUDED.title, blob_key = EXCLUDED.blob_key,
|
||||
bytes = EXCLUDED.bytes, duration_secs = EXCLUDED.duration_secs,
|
||||
rendered_by = EXCLUDED.rendered_by, script_sha = EXCLUDED.script_sha",
|
||||
)
|
||||
.bind(uuid::Uuid::now_v7())
|
||||
.bind(workspace_id)
|
||||
.bind(mission_id)
|
||||
.bind(&date)
|
||||
.bind(&title)
|
||||
.bind(&key)
|
||||
.bind(audio.len() as i64)
|
||||
.bind(secs as i32)
|
||||
.bind(backend.describe())
|
||||
.bind(sha_hex(md.as_bytes()))
|
||||
.execute(pool)
|
||||
.await
|
||||
{
|
||||
eprintln!("podcast: recording episode for {mission_id} failed: {e}");
|
||||
continue;
|
||||
}
|
||||
eprintln!(
|
||||
"podcast: episode for mission {mission_id} — {} turns, {}s, {} bytes at {key}",
|
||||
script.turns.len(),
|
||||
secs,
|
||||
audio.len()
|
||||
);
|
||||
made += 1;
|
||||
}
|
||||
Ok(made)
|
||||
}
|
||||
|
||||
/// Say — once — that a mission can never be rendered.
|
||||
///
|
||||
/// Once, not every tick: the sweep revisits the same missions forever, and a
|
||||
/// line per mission per five minutes would bury everything else in the log. The
|
||||
/// episode row is the marker, with a zero-length blob key that the feed skips.
|
||||
async fn record_unrenderable(pool: &sqlx::PgPool, mission_id: uuid::Uuid, checkout: &std::path::Path) {
|
||||
eprintln!(
|
||||
"podcast: mission {mission_id} has no script at {} — the checkout was reaped before the \
|
||||
render sweep reached it, so this day has no episode. The script is still on the phase's \
|
||||
vault branch if it is wanted.",
|
||||
checkout.display()
|
||||
);
|
||||
let _ = sqlx::query(
|
||||
"INSERT INTO podcast_episodes
|
||||
(id, workspace_id, mission_id, episode_date, title, blob_key, bytes,
|
||||
duration_secs, rendered_by, script_sha)
|
||||
SELECT $1, m.workspace_id, m.id, '', m.title, '', 0, 0, 'unrenderable', ''
|
||||
FROM missions m WHERE m.id = $2
|
||||
ON CONFLICT (mission_id) DO NOTHING",
|
||||
)
|
||||
.bind(uuid::Uuid::now_v7())
|
||||
.bind(mission_id)
|
||||
.execute(pool)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// The most recent `ContinuousResearch/<date>/script.md` in a checkout.
|
||||
fn newest_script(checkout: &std::path::Path) -> Option<std::path::PathBuf> {
|
||||
let root = checkout.join("ContinuousResearch");
|
||||
let mut dates: Vec<String> = std::fs::read_dir(root)
|
||||
.ok()?
|
||||
.filter_map(Result::ok)
|
||||
.filter(|e| e.path().is_dir())
|
||||
.map(|e| e.file_name().to_string_lossy().to_string())
|
||||
.collect();
|
||||
// ISO dates sort lexicographically, which is the whole reason for the format.
|
||||
dates.sort();
|
||||
dates.iter().rev().find_map(|d| {
|
||||
let p = checkout.join(script_path(d));
|
||||
p.is_file().then_some(p)
|
||||
})
|
||||
}
|
||||
|
||||
/// Spawn the render sweep.
|
||||
pub fn spawn(
|
||||
pool: sqlx::PgPool,
|
||||
blobs: Option<std::sync::Arc<dyn cm_files::BlobStore>>,
|
||||
interval: std::time::Duration,
|
||||
) {
|
||||
let Some(blobs) = blobs else {
|
||||
eprintln!("podcast: no blob storage configured — episodes will not be rendered");
|
||||
return;
|
||||
};
|
||||
let Some(backend) = ElevenLabs::from_env() else {
|
||||
// Not an error. A deployment without a key simply produces no audio,
|
||||
// and every other part of the mission still works.
|
||||
eprintln!("podcast: ELEVENLABS_API_KEY not set — episodes will not be rendered");
|
||||
return;
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let mut tick = tokio::time::interval(interval);
|
||||
tick.tick().await;
|
||||
loop {
|
||||
tick.tick().await;
|
||||
match render_pending(&pool, &blobs, &backend).await {
|
||||
Ok(n) if n > 0 => eprintln!("podcast: rendered {n} episode(s)"),
|
||||
Ok(_) => {}
|
||||
Err(e) => eprintln!("podcast: sweep failed: {e}"),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user