osobh/clawmates
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(continuous-research): the digest reaches the vault, and the renderer stops racing a reaper

Browse Source 
Two halves of one defect found while reviewing template maturity.

THE DIGEST NEVER LANDED. Paper notes auto-merge into the vault from the
day the library shipped — library.rs calls auto_merge::try_merge. The
digest that analyses them did not, because nothing on the mission path
ever called it: a mission branch waits for an operator merge
(routes::missions::merge_branch) which, measured on 2026-09-22, had not
happened since 2026-08-18. Every continuous_research run in that month
produced analysis.md, script.md and episode.json onto a branch nobody
merged. Papers flowed; the thinking about them did not.

mission_delivery now accrues such a branch into the repo's default
branch after a successful push, behind three independent limits, none of
which trusts the mission type alone: AdditiveOnly (try_merge re-reads the
diff against the REMOTE base and refuses any modify/delete/rename — a
digest is a new dated folder, so all adds); the phase's own judge verdict
read from mission_phase_evaluations rather than inferred from its status,
because a phase with no condition completes unjudged; and
accrues_automatically(), a pure predicate listing exactly one recipe so
adding another is a reviewed edit rather than a condition buried in a
query. The outcome — including a refusal, which is the interesting half —
lands on the artifact as merged/merge_reason.

Placement checked rather than assumed: capture selects on phase status
'completed', and a phase reaches that only after the judge rules, so the
verdict exists by delivery time.

THE RENDERER RACED A REAPER. podcast::render_pending read script.md from
the mission checkout, deleted 30 minutes after a terminal state; the
2-minute sweep was a mitigation and record_unrenderable the loss, whose
own message pointed at the vault as manual recovery. It now takes the
vault instead of mentioning it: default branch first, the delivery branch
second, shallow and cleaned up. A script on the vault is re-renderable
next week; a script in a reaped checkout is gone.

Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01WZb5A2kfVfjpdwSochkuHz
This commit is contained in:
Omar Sobh
2026-09-22 10:43:09 -05:00
co-authored by Claude Opus 5
parent 594cd99e54
commit 6a9ec2b74b
2 changed files with 270 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-api/src/mission_delivery.rs
+142 -1
View File
@@ -380,6 +380,9 @@ pub async fn capture_phase_diff_at(
// [`untrusted_empty_reason`].
let mut outcome: Option<TestOutcome> = None;
let mut published: Option<Publish> = None;
// Set only for a recipe whose output accrues into its repository; `None`
// means "not that kind of mission", which is different from "refused".
let mut merged: Option<crate::auto_merge::MergeOutcome> = None;
let mut publish_error: Option<String> = untrusted_empty_reason(empty, diff_error.as_deref());
if let Some(c) = committed.as_ref() {
if !empty {
@@ -424,7 +427,15 @@ pub async fn capture_phase_diff_at(
Ok(Some(url)) => {
let verified = outcome.as_ref().and_then(TestOutcome::verified);
match publish_phase_branch(&repo, &url, &c.branch, gate, verified).await {
Ok(p) => published = Some(p),
Ok(p) => {
if p.pushed {
merged = try_accrue_to_default_branch(
pool, mission_id, phase_id, &repo, &url, &p.branch,
)
.await;
}
published = Some(p);
}
// `publish_phase_branch` only returns Err for a local
// git failure; a rejected push is Ok with an error
// inside. Both must reach the artifact.
@@ -484,6 +495,11 @@ pub async fn capture_phase_diff_at(
"tests_status": outcome.as_ref().map(TestOutcome::status),
"tests_detail": outcome.as_ref().and_then(TestOutcome::detail),
"pushed": published.as_ref().map(|p| p.pushed),
// Whether the branch was accrued into the repo's default branch, and
// why not when it was not. Null for every recipe that is reviewed by
// a human, which is all of them but continuous_research.
"merged": merged.as_ref().map(|m| m.merged),
"merge_reason": merged.as_ref().map(|m| m.reason.clone()),
"push_error": published
.as_ref()
.and_then(|p| p.error.clone())
@@ -867,6 +883,113 @@ pub fn discover_test_command(repo: &Path) -> Option<Vec<String>> {
/// after clone because agents run as root in a container that mounts the
/// checkout. Building it here also means a rotated token takes effect
/// immediately instead of at the next clone.
/// Merge a delivered branch into the repository's default branch, when the
/// mission is one whose output is meant to accrue rather than be reviewed.
///
/// **Why this exists.** Paper notes reached the vault automatically from the
/// day the library shipped (`library.rs` calls `auto_merge::try_merge`); the
/// DIGEST that analyses them did not, because nothing on the mission path
/// ever called it. A mission branch waited for an operator merge
/// (`routes::missions::merge_branch`) that, measured on 2026-09-22, had not
/// happened since 2026-08-18: every `continuous_research` run in that month
/// produced `analysis.md`, `script.md` and `episode.json` onto a branch
/// nobody merged. The pipeline was half-continuous — papers flowed, the
/// thinking about them did not.
///
/// **Why it is safe to do automatically here.** Three independent limits,
/// none of which trusts the mission type on its own:
/// * `MergePolicy::AdditiveOnly` — `try_merge` re-reads the diff against
/// the REMOTE base and refuses on any modify, delete or rename. A digest
/// writes a new `ContinuousResearch/<date>/` folder, so it is all adds;
/// the day that stops being true the merge stops, loudly.
/// * `verified` — the phase's own judge said `met`. Read from
/// `mission_phase_evaluations` rather than inferred from the phase's
/// status, the same way `library.rs` measures `healthy() && shelved`
/// instead of assuming a clean run.
/// * the recipe — only `continuous_research`, whose whole point is an
/// unattended loop into the operator's own vault. Every other recipe's
/// branch is left exactly as it was, for a human.
///
/// Returns `None` when this mission is not one of those; `Some` otherwise,
/// including when the merge was refused, because a refusal is the interesting
/// half and belongs in the artifact beside the branch.
/// Which recipes deliver into a repository that ACCRUES rather than one a
/// human reviews. Pure, so the list is readable and testable without a
/// database — and so adding one is a deliberate edit here rather than a
/// condition buried in a query.
///
/// Only `continuous_research`: its output is a dated folder in the
/// operator's own vault, produced on a schedule, and a human merge gate in
/// front of it means the digest is never read (measured: a month of them).
/// Every other recipe writes into a code repository where a review gate is
/// the point.
fn accrues_automatically(template_kind: &str) -> bool {
template_kind == crate::continuous_research::TEMPLATE_KIND
}
async fn try_accrue_to_default_branch(
pool: &sqlx::PgPool,
mission_id: Uuid,
phase_id: Uuid,
repo: &std::path::Path,
push_url: &str,
branch: &str,
) -> Option<crate::auto_merge::MergeOutcome> {
let row: (String, Option<String>) = sqlx::query_as::<_, (String, Option<String>)>(
"SELECT m.template_kind, r.default_branch
FROM missions m LEFT JOIN repos r ON r.id = m.repo_id
WHERE m.id = $1",
)
.bind(mission_id)
.fetch_optional(pool)
.await
.map_err(|e| eprintln!("mission_delivery: accrue lookup for {mission_id} failed: {e}"))
.ok()
.flatten()?;
let (template_kind, default_branch) = row;
if !accrues_automatically(&template_kind) {
return None;
}
// The judge's own verdict for THIS phase, not the phase status. A phase
// with no completion condition completes without ever being judged, and
// merging that into a knowledge base on the strength of "it finished"
// is the kind of inference this codebase keeps paying for.
let met: Option<bool> = sqlx::query_scalar(
"SELECT met FROM mission_phase_evaluations
WHERE phase_id = $1 ORDER BY iteration DESC LIMIT 1",
)
.bind(phase_id)
.fetch_optional(pool)
.await
.unwrap_or(None);
let verified = met == Some(true);
let base = default_branch.unwrap_or_else(|| "main".to_string());
let outcome = crate::auto_merge::try_merge(
repo,
push_url,
branch,
&base,
crate::auto_merge::MergePolicy::AdditiveOnly,
verified,
)
.await
.unwrap_or_else(|e| crate::auto_merge::MergeOutcome {
merged: false,
reason: format!("merge attempt failed: {e}"),
});
eprintln!(
"mission_delivery: {branch} -> {base} — {} ({})",
outcome.reason,
match verified {
true => "judge met",
false => "not judged met",
}
);
Some(outcome)
}
async fn push_url_for(pool: &sqlx::PgPool, mission_id: Uuid) -> Result<Option<String>, String> {
let url: Option<String> = sqlx::query_scalar(
"SELECT r.clone_url FROM missions m JOIN repos r ON r.id = m.repo_id WHERE m.id = $1",
@@ -1314,6 +1437,24 @@ mod changed_path_capture_tests {
#[cfg(test)]
mod tests {
/// Exactly one recipe accrues without a human. The others deliver into
/// code repositories where the review gate is the point, and a recipe
/// added to that list should be an edit somebody reviewed.
#[test]
fn only_continuous_research_accrues_automatically() {
assert!(accrues_automatically(crate::continuous_research::TEMPLATE_KIND));
for kind in [
"research_and_code",
"research_only",
"security_hardening",
"benchmark",
"refactor",
"",
] {
assert!(!accrues_automatically(kind), "{kind} must not auto-merge");
}
}
use super::*;
/// Git says "your history diverged" several ways, and the one production
crates/cm-api/src/podcast.rs
+128 -17
View File
@@ -720,33 +720,54 @@ pub async fn render_pending(
let date = crate::continuous_research::today();
let checkout = crate::mission_workspace::checkout_path(mission_id);
let mut path = checkout.join(script_path(&date));
// Set when the checkout is gone and the vault supplied the script.
let mut vault_script: Option<String> = None;
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 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), and
// this sweep used to lose that race permanently — the
// comment here said as much and pointed at the vault as
// the manual recovery. So take the vault instead of
// saying it: `mission_delivery` pushed the script and,
// for this recipe, merged it into the default branch.
//
// 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;
// A script on the vault is also re-renderable next week;
// a script in a reaped checkout is gone. The 2-minute
// sweep was a mitigation for a source that should never
// have been temporary.
match script_from_vault(pool, mission_id, &date).await {
Some((p, text)) => {
eprintln!(
"podcast: mission {mission_id} — checkout is gone; \
rendering from the vault ({p})"
);
vault_script = Some(text);
}
None => {
// NEVER silent: not here and not in the vault
// means this day has no episode.
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 md = match vault_script {
Some(text) => text,
None => 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() {
@@ -817,6 +838,96 @@ pub async fn render_pending(
/// 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.
/// Read a mission's script out of its repository when the checkout is gone.
///
/// Tries the repository's default branch first — where `mission_delivery`
/// accrues a continuous-research digest — then the phase's own delivery
/// branch, which exists whether or not the merge was taken. A shallow clone
/// into a scratch directory, removed afterwards; the vault is markdown, so
/// this is cheap.
///
/// Returns the path it found and the contents, or `None` when neither ref
/// has a script — which is the genuinely unrenderable case.
async fn script_from_vault(
pool: &sqlx::PgPool,
mission_id: uuid::Uuid,
date: &str,
) -> Option<(String, String)> {
use sqlx::Row;
let row = sqlx::query(
"SELECT r.clone_url, r.default_branch,
(SELECT a.metadata->>'branch' FROM mission_artifacts a
WHERE a.mission_id = m.id AND a.kind = 'code_diff'
AND a.metadata->>'branch' IS NOT NULL
ORDER BY a.created_at DESC LIMIT 1) AS delivered
FROM missions m JOIN repos r ON r.id = m.repo_id
WHERE m.id = $1",
)
.bind(mission_id)
.fetch_optional(pool)
.await
.ok()
.flatten()?;
let clone_url: String = row.try_get("clone_url").ok()?;
let default_branch: Option<String> = row.try_get("default_branch").ok();
let delivered: Option<String> = row.try_get("delivered").ok();
let auth = crate::mission_workspace::with_ambient_auth(&clone_url);
let workdir = crate::mission_workspace::missions_root()
.join("_episode")
.join(mission_id.to_string());
let _ = tokio::fs::remove_dir_all(&workdir).await;
if let Some(parent) = workdir.parent() {
let _ = tokio::fs::create_dir_all(parent).await;
}
let base = default_branch.unwrap_or_else(|| "main".to_string());
let clone = tokio::process::Command::new("git")
.args(["clone", "--quiet", "--no-single-branch", "--depth", "1", &auth.url])
.arg(&workdir)
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.await
.ok()?;
if !clone.status.success() {
eprintln!(
"podcast: could not clone the vault for {mission_id}: {}",
String::from_utf8_lossy(&clone.stderr).trim()
);
let _ = tokio::fs::remove_dir_all(&workdir).await;
return None;
}
let mut found = None;
for reference in [Some(base), delivered].into_iter().flatten() {
let co = tokio::process::Command::new("git")
.args(["-C"])
.arg(&workdir)
.args(["fetch", "--quiet", "--depth", "1", "origin", &reference])
.env("GIT_TERMINAL_PROMPT", "0")
.output()
.await;
if co.map(|o| o.status.success()).unwrap_or(false) {
let show = tokio::process::Command::new("git")
.args(["-C"])
.arg(&workdir)
.args(["show", &format!("FETCH_HEAD:{}", script_path(date))])
.output()
.await;
if let Ok(out) = show {
if out.status.success() {
let text = String::from_utf8_lossy(&out.stdout).to_string();
if !text.trim().is_empty() {
found = Some((format!("{reference}:{}", script_path(date)), text));
break;
}
}
}
}
}
let _ = tokio::fs::remove_dir_all(&workdir).await;
found
}
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 \