reflection: catch-up cadence so a missed boundary cannot be lost
is_multiple_of fired only at the exact boundary turn; an interrupt, a failed pass, or a restart rehydrating past it lost the cadence forever and the N+25 pass never ran. The reflection tree now holds a marker of the last successful and last attempted turns; a due-but-missed pass stays due, and a failed one backs off five turns instead of burning a call per turn. Outcomes are recorded from on_response.
This commit is contained in:
parent
b75137689f
commit
727c496949
2 changed files with 74 additions and 6 deletions
|
|
@ -23,6 +23,7 @@
|
||||||
//! command both go through this seam).
|
//! command both go through this seam).
|
||||||
//!
|
//!
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
|
@ -87,6 +88,20 @@ pub struct ReflectionEngine {
|
||||||
max_tokens: Option<u32>,
|
max_tokens: Option<u32>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The cadence's state on disk: which turn a pass last succeeded at, and
|
||||||
|
/// which turn one last failed at. Written to the reflection tree after every
|
||||||
|
/// due check, so a boundary missed by an interrupt or a failed pass is caught
|
||||||
|
/// up on later, and the fact of the pass survives restarts.
|
||||||
|
#[derive(Clone, Default, Serialize, Deserialize)]
|
||||||
|
pub struct ReflectionMarker {
|
||||||
|
pub last_succeeded_turn: u32,
|
||||||
|
pub last_attempted_turn: u32,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Turns to wait after a failed pass before trying again. A model error at
|
||||||
|
/// the boundary must not burn a call on every turn until it clears.
|
||||||
|
pub const REFLECTION_RETRY_BACKOFF_TURNS: u32 = 5;
|
||||||
|
|
||||||
impl ReflectionEngine {
|
impl ReflectionEngine {
|
||||||
pub fn new(
|
pub fn new(
|
||||||
agents: Arc<AgentInventory>,
|
agents: Arc<AgentInventory>,
|
||||||
|
|
@ -104,6 +119,43 @@ impl ReflectionEngine {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The cadence's own tree is where a pass's state belongs: it exists
|
||||||
|
/// because the pass exists, and a marker there is legible in `git log`
|
||||||
|
/// alongside her conclusions.
|
||||||
|
fn marker_path(&self, primary_id: &str) -> PathBuf {
|
||||||
|
self.agents
|
||||||
|
.cadence_memory_root(primary_id, Cadence::Reflection)
|
||||||
|
.join("system/reflection-last.json")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read the cadence state. A missing or unreadable marker reads as a
|
||||||
|
/// fresh cadence: the first pass is due at the configured interval.
|
||||||
|
pub fn marker(&self, primary_id: &str) -> ReflectionMarker {
|
||||||
|
std::fs::read_to_string(self.marker_path(primary_id))
|
||||||
|
.ok()
|
||||||
|
.and_then(|raw| serde_json::from_str(&raw).ok())
|
||||||
|
.unwrap_or_default()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Record a due check's outcome. `succeeded` updates the last-successful
|
||||||
|
/// turn (resetting the cadence); a failure records the attempt so the
|
||||||
|
/// retry backs off rather than firing every turn.
|
||||||
|
pub fn record_marker(&self, primary_id: &str, turn: u32, succeeded: bool) {
|
||||||
|
let path = self.marker_path(primary_id);
|
||||||
|
let mut marker = self.marker(primary_id);
|
||||||
|
if succeeded {
|
||||||
|
marker.last_succeeded_turn = turn;
|
||||||
|
} else {
|
||||||
|
marker.last_attempted_turn = turn;
|
||||||
|
}
|
||||||
|
if let Some(parent) = path.parent() {
|
||||||
|
let _ = std::fs::create_dir_all(parent);
|
||||||
|
}
|
||||||
|
if let Ok(raw) = serde_json::to_string_pretty(&marker) {
|
||||||
|
let _ = std::fs::write(&path, raw);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Run a reflection pass over `messages` for `agent_id`. Returns a
|
/// Run a reflection pass over `messages` for `agent_id`. Returns a
|
||||||
/// report; any ledger/memory writes have already been committed by
|
/// report; any ledger/memory writes have already been committed by
|
||||||
/// the memory sensor.
|
/// the memory sensor.
|
||||||
|
|
|
||||||
|
|
@ -235,6 +235,7 @@ impl ConsciousnessEngine {
|
||||||
/// to reflect on a different rhythm silently kept everyone else's.
|
/// to reflect on a different rhythm silently kept everyone else's.
|
||||||
async fn reflection_due(&self, agent_id: &str, turn_count: u32) -> bool {
|
async fn reflection_due(&self, agent_id: &str, turn_count: u32) -> bool {
|
||||||
use crate::core::config::ReflectionTrigger;
|
use crate::core::config::ReflectionTrigger;
|
||||||
|
use crate::core::reflection::REFLECTION_RETRY_BACKOFF_TURNS;
|
||||||
|
|
||||||
if turn_count == 0 {
|
if turn_count == 0 {
|
||||||
return false;
|
return false;
|
||||||
|
|
@ -249,10 +250,20 @@ impl ConsciousnessEngine {
|
||||||
}
|
}
|
||||||
match trigger {
|
match trigger {
|
||||||
ReflectionTrigger::Off => false,
|
ReflectionTrigger::Off => false,
|
||||||
ReflectionTrigger::StepCount => turn_count.is_multiple_of(interval as u32),
|
// Catch-up, not one-shot. `is_multiple_of` fired only at the
|
||||||
// Compaction-driven reflection has no signal wired to it yet; the
|
// exact boundary: an interrupted turn, a failed pass, or a
|
||||||
// step rhythm is the honest fallback rather than never firing.
|
// restart that rehydrated past it lost the cadence forever —
|
||||||
ReflectionTrigger::CompactionEvent => turn_count.is_multiple_of(interval as u32),
|
// the N+25 pass never ran while the surfaces counted dozens of
|
||||||
|
// turns. A due-but-missed pass stays due; a failed one retries
|
||||||
|
// after a short backoff so a sick model cannot burn a call on
|
||||||
|
// every turn.
|
||||||
|
ReflectionTrigger::StepCount | ReflectionTrigger::CompactionEvent => {
|
||||||
|
let marker = self.reflection.marker(agent_id);
|
||||||
|
turn_count >= interval as u32
|
||||||
|
&& turn_count >= marker.last_succeeded_turn + interval as u32
|
||||||
|
&& turn_count.saturating_sub(marker.last_attempted_turn)
|
||||||
|
>= REFLECTION_RETRY_BACKOFF_TURNS
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -330,14 +341,17 @@ impl ConsciousnessEngine {
|
||||||
let pressure = self.pressure_for(agent_id, messages).await;
|
let pressure = self.pressure_for(agent_id, messages).await;
|
||||||
|
|
||||||
// ── N+25 reflection ──
|
// ── N+25 reflection ──
|
||||||
// Fires at every Nth turn, and the N now comes from config. It was
|
// Due at the configured interval, with catch-up semantics and the
|
||||||
// hardcoded `25` while `ReflectionConfig` carried `enabled`,
|
// outcome recorded in the reflection tree (reflection-last.json).
|
||||||
|
// It was hardcoded `25` while `ReflectionConfig` carried `enabled`,
|
||||||
// `message_interval`, `trigger` and `per_agent` — all four editable in
|
// `message_interval`, `trigger` and `per_agent` — all four editable in
|
||||||
// the settings UI, all four printed by the CLI, and only `.model` ever
|
// the settings UI, all four printed by the CLI, and only `.model` ever
|
||||||
// read. Turning it off did not turn it off.
|
// read. Turning it off did not turn it off.
|
||||||
if self.reflection_due(agent_id, turn_count).await {
|
if self.reflection_due(agent_id, turn_count).await {
|
||||||
match self.reflection.reflect_now(agent_id, messages).await {
|
match self.reflection.reflect_now(agent_id, messages).await {
|
||||||
Ok(report) => {
|
Ok(report) => {
|
||||||
|
self.reflection
|
||||||
|
.record_marker(agent_id, turn_count, true);
|
||||||
let header = if report.exited_cleanly {
|
let header = if report.exited_cleanly {
|
||||||
format!("N+25 reflection ({} turns reviewed)", report.turns_reviewed)
|
format!("N+25 reflection ({} turns reviewed)", report.turns_reviewed)
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -351,6 +365,8 @@ impl ConsciousnessEngine {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
self.reflection
|
||||||
|
.record_marker(agent_id, turn_count, false);
|
||||||
tracing::warn!("N+25 reflection failed: {}", e);
|
tracing::warn!("N+25 reflection failed: {}", e);
|
||||||
events.push(ConsciousnessEvent::Reflection {
|
events.push(ConsciousnessEvent::Reflection {
|
||||||
content: format!(
|
content: format!(
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue