Omar SobhandClaude Opus 5
d48bdbc9a7
The research scenario passed 4/4 and the log underneath it said: phase_summarizer: ... failed: anthropic 400 Bad Request: "Your credit balance is too low to access the Anthropic API" `phase_summarizer` and `mission_refiner` each built their own reqwest POST to the Messages API with `x-api-key: $ANTHROPIC_API_KEY`. No audit of `.complete(` call sites could have found them — they never touched a provider — so every phase summary and every mission-brief refinement on this deployment had been failing against an empty account while the phases themselves ran fine. The summarizer even persisted an error row per phase, which is why nothing ever retried loudly enough to notice. Both now go through `subscription::complete_with_fallback`, so they inherit the subscription-first credential choice, the 429 backoff, and the opus -> haiku -> glm chain. The summarizer records the model that ANSWERED in mission_phase_summaries.model rather than the one it asked for. The guard is a source WALK, not a file list: any .rs under cm-api/src that mentions the Messages API host or `x-api-key` fails the test. A hand-listed set of files is exactly what let these two hide. Co-Authored-By: Claude Opus 5 <[email protected]>
581 lines
20 KiB
Rust
581 lines
20 KiB
Rust
//! Post-phase summarization worker.
|
||
//!
|
||
//! Watches `mission_phases` for terminal-state transitions and, for
|
||
//! each one that hasn't been summarized yet, aggregates every
|
||
//! `topology_runs.checkpoint.outputs[]` bound to that phase plus the
|
||
//! phase's `mission_tasks` + `mission_artifacts` and asks Claude Opus
|
||
//! 4.8 to produce a structured completion card:
|
||
//!
|
||
//! { narrative, metrics, sources, tooling, next_actions }
|
||
//!
|
||
//! The output lands in `mission_phase_summaries` (one row per
|
||
//! phase_id, upserted). The mission phase card in the UI renders it
|
||
//! below the phase's other detail so the operator sees "what did this
|
||
//! phase actually accomplish, what did it produce, and what's next".
|
||
//!
|
||
//! Runs on a slow tick (30s) — summarization is cheap to defer, and
|
||
//! the LLM call is the expensive part.
|
||
|
||
use serde_json::{json, Value};
|
||
use sqlx::PgPool;
|
||
use sqlx::Row;
|
||
use std::time::Duration;
|
||
use uuid::Uuid;
|
||
|
||
const DEFAULT_MODEL: &str = "claude-opus-4-8";
|
||
const POLL_INTERVAL: Duration = Duration::from_secs(30);
|
||
/// Cap the raw material we send to the model. Missions can produce
|
||
/// hundreds of KB of agent output; we slice by turn and by phase
|
||
/// artifact but still bound the total prompt.
|
||
const MAX_OUTPUT_BYTES: usize = 60_000;
|
||
|
||
fn model_name() -> String {
|
||
std::env::var("CLAWMATES_SUMMARIZER_MODEL").unwrap_or_else(|_| DEFAULT_MODEL.to_string())
|
||
}
|
||
|
||
/// The runtime is carried purely so the summarizer can reach the SAME
|
||
/// providers as everything else. It used to hand-roll its own HTTPS POST with
|
||
/// `x-api-key: $ANTHROPIC_API_KEY`, which is why no audit of `.complete(` call
|
||
/// sites ever found it — and why every phase summary on this deployment died
|
||
/// with "credit balance is too low" while the phases themselves ran fine.
|
||
pub fn spawn(pool: PgPool, runtime: cm_runtime::Runtime) {
|
||
tokio::spawn(async move {
|
||
tokio::time::sleep(Duration::from_secs(45)).await;
|
||
let mut ticker = tokio::time::interval(POLL_INTERVAL);
|
||
ticker.tick().await;
|
||
loop {
|
||
ticker.tick().await;
|
||
if let Err(e) = sweep_once(&pool, &runtime).await {
|
||
eprintln!("phase_summarizer: sweep failed: {e}");
|
||
}
|
||
}
|
||
});
|
||
}
|
||
|
||
async fn sweep_once(pool: &PgPool, runtime: &cm_runtime::Runtime) -> Result<(), String> {
|
||
// Terminal phases with no summary yet.
|
||
let rows = sqlx::query(
|
||
"SELECT mp.id, mp.mission_id, mp.kind
|
||
FROM mission_phases mp
|
||
LEFT JOIN mission_phase_summaries mps ON mps.phase_id = mp.id
|
||
WHERE mp.status IN ('completed', 'failed')
|
||
AND mps.id IS NULL
|
||
LIMIT 10",
|
||
)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| format!("scan phases: {e}"))?;
|
||
for row in rows {
|
||
let phase_id: Uuid = row.get("id");
|
||
let mission_id: Uuid = row.get("mission_id");
|
||
let kind: String = row.get("kind");
|
||
if let Err(e) = summarize_one(pool, runtime, mission_id, phase_id, &kind).await {
|
||
// Persist an error row so we don't infinite-retry a broken
|
||
// phase — the UI can surface "summary unavailable: <e>".
|
||
eprintln!("phase_summarizer: {phase_id} ({kind}) failed: {e}");
|
||
let _ = record_error(pool, mission_id, phase_id, &kind, &e).await;
|
||
}
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
async fn summarize_one(
|
||
pool: &PgPool,
|
||
runtime: &cm_runtime::Runtime,
|
||
mission_id: Uuid,
|
||
phase_id: Uuid,
|
||
kind: &str,
|
||
) -> Result<(), String> {
|
||
let material = collect_material(pool, mission_id, phase_id).await?;
|
||
if material.outputs.is_empty() && material.tasks_created == 0 && material.artifacts.is_empty() {
|
||
// Nothing to summarize. Write a placeholder so we don't retry
|
||
// this phase every 30s.
|
||
return upsert_summary(
|
||
pool,
|
||
mission_id,
|
||
phase_id,
|
||
kind,
|
||
"claude-opus-4-8",
|
||
"This phase produced no recorded output. The agents may have failed \
|
||
to reach their working directory or found nothing to act on.",
|
||
&json!({
|
||
"outputs": 0,
|
||
"tasks": 0,
|
||
"artifacts": 0,
|
||
}),
|
||
&json!([]),
|
||
&json!([]),
|
||
&json!([]),
|
||
&json!([]),
|
||
)
|
||
.await;
|
||
}
|
||
let (narrative, structured, answered_by) = call_anthropic(runtime, kind, &material).await?;
|
||
let metrics = structured
|
||
.get("metrics")
|
||
.cloned()
|
||
.unwrap_or_else(|| json!({}));
|
||
let sources = structured
|
||
.get("sources")
|
||
.cloned()
|
||
.unwrap_or_else(|| json!([]));
|
||
let tooling = structured
|
||
.get("tooling")
|
||
.cloned()
|
||
.unwrap_or_else(|| json!([]));
|
||
let next_actions = structured
|
||
.get("next_actions")
|
||
.cloned()
|
||
.unwrap_or_else(|| json!([]));
|
||
let artifacts = serde_json::to_value(&material.artifacts).unwrap_or(json!([]));
|
||
upsert_summary(
|
||
pool,
|
||
mission_id,
|
||
phase_id,
|
||
kind,
|
||
&answered_by,
|
||
&narrative,
|
||
&metrics,
|
||
&sources,
|
||
&artifacts,
|
||
&tooling,
|
||
&next_actions,
|
||
)
|
||
.await
|
||
}
|
||
|
||
struct PhaseMaterial {
|
||
/// Concatenated per-turn outputs across every topology_run bound to
|
||
/// this phase, trimmed to `MAX_OUTPUT_BYTES`.
|
||
outputs: String,
|
||
/// Original count (pre-trim) — helps the LLM understand the scale
|
||
/// even when we truncated.
|
||
output_count: usize,
|
||
/// Total tokens across runs (from checkpoint.totals.tokens).
|
||
tokens: u64,
|
||
turns: u64,
|
||
tasks_created: usize,
|
||
tasks_completed: usize,
|
||
tasks_failed: usize,
|
||
artifacts: Vec<ArtifactRef>,
|
||
task_summaries: Vec<TaskRef>,
|
||
}
|
||
|
||
#[derive(serde::Serialize)]
|
||
struct ArtifactRef {
|
||
path: String,
|
||
kind: String,
|
||
title: Option<String>,
|
||
}
|
||
|
||
#[derive(serde::Serialize)]
|
||
struct TaskRef {
|
||
external_id: Option<String>,
|
||
title: String,
|
||
status: String,
|
||
}
|
||
|
||
/// Render this phase's material as plain evidence text.
|
||
///
|
||
/// Shared with the completion evaluator (`crate::evaluator`), which judges a
|
||
/// `done_when` condition against exactly the same material the summarizer
|
||
/// writes its card from — turn outputs, task counts, artifacts. Reusing this
|
||
/// keeps the two from disagreeing about what the phase actually produced, and
|
||
/// the truncation/aggregation logic only has to be right once.
|
||
pub async fn collect_evidence(
|
||
pool: &PgPool,
|
||
mission_id: Uuid,
|
||
phase_id: Uuid,
|
||
) -> Result<String, String> {
|
||
let m = collect_material(pool, mission_id, phase_id).await?;
|
||
let mut s = String::with_capacity(m.outputs.len() + 512);
|
||
s.push_str(&format!(
|
||
"turns: {}\ntokens: {}\nagent outputs: {}\ntasks: {} created, {} completed, {} failed\n",
|
||
m.turns, m.tokens, m.output_count, m.tasks_created, m.tasks_completed, m.tasks_failed,
|
||
));
|
||
if !m.artifacts.is_empty() {
|
||
s.push_str("\nartifacts written:\n");
|
||
for a in m.artifacts.iter().take(40) {
|
||
s.push_str(&format!("- {} ({})\n", a.path, a.kind));
|
||
}
|
||
}
|
||
if !m.task_summaries.is_empty() {
|
||
s.push_str("\ntask states:\n");
|
||
for t in m.task_summaries.iter().take(40) {
|
||
s.push_str(&format!(
|
||
"- {} [{}] {}\n",
|
||
t.external_id.as_deref().unwrap_or("-"),
|
||
t.status,
|
||
t.title
|
||
));
|
||
}
|
||
}
|
||
s.push_str("\nagent turn output:\n");
|
||
s.push_str(&m.outputs);
|
||
Ok(s)
|
||
}
|
||
|
||
async fn collect_material(
|
||
pool: &PgPool,
|
||
mission_id: Uuid,
|
||
phase_id: Uuid,
|
||
) -> Result<PhaseMaterial, String> {
|
||
// Runs — checkpoint.outputs + totals aggregated.
|
||
let run_rows = sqlx::query(
|
||
"SELECT checkpoint FROM topology_runs
|
||
WHERE mission_id = $1 AND mission_phase_id = $2",
|
||
)
|
||
.bind(mission_id)
|
||
.bind(phase_id)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| format!("load runs: {e}"))?;
|
||
let mut concat = String::new();
|
||
let mut output_count = 0usize;
|
||
let mut tokens = 0u64;
|
||
let mut turns = 0u64;
|
||
for row in &run_rows {
|
||
let cp: Option<Value> = row.get("checkpoint");
|
||
let Some(cp) = cp else { continue };
|
||
if let Some(t) = cp.get("totals") {
|
||
tokens += t.get("tokens").and_then(|v| v.as_u64()).unwrap_or(0);
|
||
turns += t.get("turns").and_then(|v| v.as_u64()).unwrap_or(0);
|
||
}
|
||
if let Some(arr) = cp.get("outputs").and_then(|v| v.as_array()) {
|
||
for (i, item) in arr.iter().enumerate() {
|
||
output_count += 1;
|
||
if concat.len() >= MAX_OUTPUT_BYTES {
|
||
continue;
|
||
}
|
||
let s = match item {
|
||
Value::String(s) => s.clone(),
|
||
other => other.to_string(),
|
||
};
|
||
concat.push_str(&format!("\n\n── turn {} ──\n", i + 1));
|
||
let remaining = MAX_OUTPUT_BYTES.saturating_sub(concat.len());
|
||
if s.len() > remaining {
|
||
concat.push_str(&s[..remaining]);
|
||
concat.push_str("\n… (truncated)");
|
||
} else {
|
||
concat.push_str(&s);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// Tasks — count by status, keep small summaries.
|
||
let task_rows = sqlx::query(
|
||
"SELECT external_id, title, status
|
||
FROM mission_tasks
|
||
WHERE mission_id = $1 AND phase_id = $2
|
||
ORDER BY created_at
|
||
LIMIT 40",
|
||
)
|
||
.bind(mission_id)
|
||
.bind(phase_id)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| format!("load tasks: {e}"))?;
|
||
let mut task_summaries = Vec::new();
|
||
let mut tasks_completed = 0usize;
|
||
let mut tasks_failed = 0usize;
|
||
for row in &task_rows {
|
||
let status: String = row.get("status");
|
||
match status.as_str() {
|
||
"complete" => tasks_completed += 1,
|
||
"failed" => tasks_failed += 1,
|
||
_ => {}
|
||
}
|
||
task_summaries.push(TaskRef {
|
||
external_id: row.get("external_id"),
|
||
title: row.get("title"),
|
||
status,
|
||
});
|
||
}
|
||
|
||
// Artifacts.
|
||
let artifact_rows = sqlx::query(
|
||
"SELECT path, kind, title
|
||
FROM mission_artifacts
|
||
WHERE mission_id = $1 AND phase_id = $2
|
||
ORDER BY created_at
|
||
LIMIT 40",
|
||
)
|
||
.bind(mission_id)
|
||
.bind(phase_id)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| format!("load artifacts: {e}"))?;
|
||
let artifacts: Vec<ArtifactRef> = artifact_rows
|
||
.into_iter()
|
||
.map(|r| ArtifactRef {
|
||
path: r.get("path"),
|
||
kind: r.get("kind"),
|
||
title: r.get("title"),
|
||
})
|
||
.collect();
|
||
|
||
Ok(PhaseMaterial {
|
||
outputs: concat,
|
||
output_count,
|
||
tokens,
|
||
turns,
|
||
tasks_created: task_rows.len(),
|
||
tasks_completed,
|
||
tasks_failed,
|
||
artifacts,
|
||
task_summaries,
|
||
})
|
||
}
|
||
|
||
/// Returns the narrative, the parsed object, and **the model that answered** —
|
||
/// which may be a fallback link rather than `model_name()`, and is recorded as
|
||
/// such.
|
||
async fn call_anthropic(
|
||
runtime: &cm_runtime::Runtime,
|
||
kind: &str,
|
||
material: &PhaseMaterial,
|
||
) -> Result<(String, Value, String), String> {
|
||
let model = model_name();
|
||
let system = system_prompt(kind);
|
||
let user = user_prompt(kind, material);
|
||
|
||
let (raw, answered_by) = crate::subscription::complete_with_fallback(
|
||
runtime, &system, &user, &model, 4096, false,
|
||
)
|
||
.await?;
|
||
let raw = raw.trim().to_string();
|
||
if raw.is_empty() {
|
||
return Err(format!("{answered_by} returned empty text"));
|
||
}
|
||
// Model returns a JSON object; extract narrative + rest.
|
||
let parsed: Value = serde_json::from_str(&strip_code_fence(&raw)).map_err(|e| {
|
||
format!(
|
||
"summarizer JSON parse failed: {e}. Raw head: {}",
|
||
&raw[..raw.len().min(400)]
|
||
)
|
||
})?;
|
||
let narrative = parsed
|
||
.get("narrative")
|
||
.and_then(|v| v.as_str())
|
||
.unwrap_or("")
|
||
.trim()
|
||
.to_string();
|
||
if narrative.is_empty() {
|
||
return Err("summarizer response missing narrative".into());
|
||
}
|
||
Ok((narrative, parsed, answered_by))
|
||
}
|
||
|
||
/// Trim a leading/trailing ```json … ``` fence the model sometimes wraps
|
||
/// around its output despite being asked for raw JSON.
|
||
fn strip_code_fence(s: &str) -> String {
|
||
let t = s.trim();
|
||
let stripped = t
|
||
.strip_prefix("```json")
|
||
.or_else(|| t.strip_prefix("```"))
|
||
.unwrap_or(t);
|
||
let stripped = stripped.trim_start_matches('\n');
|
||
stripped
|
||
.strip_suffix("```")
|
||
.map(|s| s.trim_end_matches('\n'))
|
||
.unwrap_or(stripped)
|
||
.to_string()
|
||
}
|
||
|
||
fn system_prompt(kind: &str) -> String {
|
||
let base = "You are the mission phase summarizer. Read the agent-produced \
|
||
material below and produce a compact JSON object that the operator \
|
||
UI will render as a completion card. Extract concrete facts from the \
|
||
outputs — never invent findings, PRs, files, or counts that the \
|
||
source material does not support.\n\
|
||
\n\
|
||
Return raw JSON (no code fence, no preamble). The shape MUST be:\n\
|
||
\n\
|
||
{\n \
|
||
\"narrative\": string, // 2–5 sentences: what this phase actually accomplished\n \
|
||
\"metrics\": { ... }, // kind-specific counts, see below\n \
|
||
\"sources\": [ ... ], // things the agents CONSULTED (URLs, files, docs)\n \
|
||
\"tooling\": [ ... ], // concrete recommendations — new skills / scripts / MCP tools worth wiring into the platform\n \
|
||
\"next_actions\": [ ... ] // what should happen next — cards to file, follow-ups for the next phase\n\
|
||
}\n\
|
||
\n\
|
||
Every array element is an object with at least a `title` and a short `note`. \
|
||
Sources also include a `url` or `path` when identifiable. Tooling \
|
||
entries include a `kind` ('skill' | 'script' | 'mcp' | 'workflow') and a \
|
||
`why` (what problem it solves that surfaced in the phase).\n\
|
||
\n\
|
||
Keep it tight — the card is small. If a section has nothing to say, \
|
||
return an empty array.";
|
||
let kind_hint = match kind {
|
||
"research" => "\n\nMetrics shape for RESEARCH:\n\
|
||
{ \"insights\": <int>, \"sources_gathered\": <int>, \"int_cards\": <int>, \
|
||
\"artifacts_saved\": <int>, \"handoffs_to_coding\": <int> }\n\
|
||
Focus the narrative on WHAT WAS LEARNED and WHAT THE CODING PHASE \
|
||
NEEDS TO DO next. INT-XX markers in the raw outputs are the count of \
|
||
concrete follow-up cards produced.",
|
||
"coding" => "\n\nMetrics shape for CODING:\n\
|
||
{ \"cards_picked_up\": <int>, \"cards_closed\": <int>, \"commits\": <int>, \
|
||
\"tests_added\": <int>, \"tests_passing\": <int>, \"tests_failing\": <int>, \
|
||
\"issues_found\": <int>, \"issues_fixed\": <int> }\n\
|
||
Focus the narrative on WHAT WAS BUILT, WHAT PASSED VALIDATION, and \
|
||
WHAT'S STILL OPEN. Commit hashes and PR/branch names are useful in \
|
||
`sources` when visible.",
|
||
"benchmark" => "\n\nMetrics shape for BENCHMARK:\n\
|
||
{ \"baselines\": <int>, \"comparisons\": <int>, \"regressions\": <int>, \
|
||
\"improvements\": <int> }\n\
|
||
Report the deltas the agents actually measured.",
|
||
"security_scan" => "\n\nMetrics shape for SECURITY:\n\
|
||
{ \"findings\": <int>, \"by_severity\": { \"crit\": <int>, \"high\": <int>, \"med\": <int>, \"low\": <int> }, \
|
||
\"patches_proposed\": <int> }",
|
||
_ => "",
|
||
};
|
||
format!("{base}{kind_hint}")
|
||
}
|
||
|
||
fn user_prompt(kind: &str, m: &PhaseMaterial) -> String {
|
||
let task_head: Vec<String> = m
|
||
.task_summaries
|
||
.iter()
|
||
.take(30)
|
||
.map(|t| {
|
||
format!(
|
||
"- [{}] {} — {}",
|
||
t.status,
|
||
t.external_id.as_deref().unwrap_or("--"),
|
||
t.title,
|
||
)
|
||
})
|
||
.collect();
|
||
let artifact_head: Vec<String> = m
|
||
.artifacts
|
||
.iter()
|
||
.take(30)
|
||
.map(|a| {
|
||
format!(
|
||
"- [{}] {}{}",
|
||
a.kind,
|
||
a.path,
|
||
a.title
|
||
.as_deref()
|
||
.map(|t| format!(" — {t}"))
|
||
.unwrap_or_default(),
|
||
)
|
||
})
|
||
.collect();
|
||
format!(
|
||
"Phase kind: {kind}\n\
|
||
Aggregate stats:\n\
|
||
- runs.checkpoint.outputs total (pre-truncate): {output_count}\n\
|
||
- total turns across runs: {turns}\n\
|
||
- total tokens across runs: {tokens}\n\
|
||
- mission_tasks in this phase: {tasks_created} (completed: {tasks_completed}, failed: {tasks_failed})\n\
|
||
- mission_artifacts in this phase: {artifact_count}\n\
|
||
\n\
|
||
Mission tasks in this phase (up to 30):\n\
|
||
{tasks}\n\
|
||
\n\
|
||
Mission artifacts in this phase (up to 30):\n\
|
||
{arts}\n\
|
||
\n\
|
||
Concatenated per-turn agent outputs (up to {max_bytes} bytes):\n\
|
||
{outputs}",
|
||
kind = kind,
|
||
output_count = m.output_count,
|
||
turns = m.turns,
|
||
tokens = m.tokens,
|
||
tasks_created = m.tasks_created,
|
||
tasks_completed = m.tasks_completed,
|
||
tasks_failed = m.tasks_failed,
|
||
artifact_count = m.artifacts.len(),
|
||
tasks = if task_head.is_empty() {
|
||
"(none)".to_string()
|
||
} else {
|
||
task_head.join("\n")
|
||
},
|
||
arts = if artifact_head.is_empty() {
|
||
"(none)".to_string()
|
||
} else {
|
||
artifact_head.join("\n")
|
||
},
|
||
max_bytes = MAX_OUTPUT_BYTES,
|
||
outputs = if m.outputs.is_empty() {
|
||
"(no outputs)".to_string()
|
||
} else {
|
||
m.outputs.clone()
|
||
},
|
||
)
|
||
}
|
||
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn upsert_summary(
|
||
pool: &PgPool,
|
||
mission_id: Uuid,
|
||
phase_id: Uuid,
|
||
kind: &str,
|
||
model: &str,
|
||
narrative: &str,
|
||
metrics: &Value,
|
||
sources: &Value,
|
||
artifacts: &Value,
|
||
tooling: &Value,
|
||
next_actions: &Value,
|
||
) -> Result<(), String> {
|
||
sqlx::query(
|
||
"INSERT INTO mission_phase_summaries
|
||
(mission_id, phase_id, kind, model, narrative, metrics,
|
||
sources, artifacts, tooling, next_actions, generated_at)
|
||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now())
|
||
ON CONFLICT (phase_id) DO UPDATE
|
||
SET kind = EXCLUDED.kind,
|
||
model = EXCLUDED.model,
|
||
narrative = EXCLUDED.narrative,
|
||
metrics = EXCLUDED.metrics,
|
||
sources = EXCLUDED.sources,
|
||
artifacts = EXCLUDED.artifacts,
|
||
tooling = EXCLUDED.tooling,
|
||
next_actions = EXCLUDED.next_actions,
|
||
generated_at = now(),
|
||
error = NULL",
|
||
)
|
||
.bind(mission_id)
|
||
.bind(phase_id)
|
||
.bind(kind)
|
||
.bind(model)
|
||
.bind(narrative)
|
||
.bind(metrics)
|
||
.bind(sources)
|
||
.bind(artifacts)
|
||
.bind(tooling)
|
||
.bind(next_actions)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| format!("upsert summary: {e}"))?;
|
||
Ok(())
|
||
}
|
||
|
||
async fn record_error(
|
||
pool: &PgPool,
|
||
mission_id: Uuid,
|
||
phase_id: Uuid,
|
||
kind: &str,
|
||
err: &str,
|
||
) -> Result<(), String> {
|
||
sqlx::query(
|
||
"INSERT INTO mission_phase_summaries
|
||
(mission_id, phase_id, kind, model, narrative, error)
|
||
VALUES ($1, $2, $3, 'error', 'Summary generation failed.', $4)
|
||
ON CONFLICT (phase_id) DO UPDATE
|
||
SET error = EXCLUDED.error,
|
||
generated_at = now()",
|
||
)
|
||
.bind(mission_id)
|
||
.bind(phase_id)
|
||
.bind(kind)
|
||
.bind(err)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| format!("record error: {e}"))?;
|
||
Ok(())
|
||
}
|