Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions prompts/en/memory_persistence.md.j2
Original file line number Diff line number Diff line change
Expand Up @@ -69,3 +69,27 @@ This is an automatic process triggered periodically during conversation. You are
8. Verify stale memory before relying on it. If a recalled memory conflicts with
the newer conversation context, treat the older item as stale, avoid propagating
it as truth, and capture the latest truth via `updates` or `contradicts`.
{% if skill_reflection %}
## Skill Reflection

This session involved substantial work, so this pass also decides whether it produced a reusable procedure. Memory answers "who is the user and what is going on"; skills answer "how do we do this class of task here." A correction like "stop posting walls of text in Discord" is not a fact about the user — it's a standing procedure change, and it belongs in the skill governing that task class.

First decide whether anything is worth keeping. Ending with no skill writes is the common case and completely acceptable — but treat a session where the user corrected the agent's procedure as a strong write signal.

If something is worth keeping, follow this order strictly:

1. **Patch the skill that governed this task**, if one was used this session. Call `skills_list` to see what exists, `read_skill` before any modification.
2. **Patch a related existing skill** whose task class covers the lesson.
3. **Add a `references/` file** to an existing skill via `skill_manage`'s `write_file` action when the material is supporting detail, not procedure.
4. **Only then create a new skill** — named for the task class (`discord-formatting`), never the incident. Keep the description within its budget; it's loaded into every system prompt.
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

Never persist:

- Environment-dependent failures (a missing binary, a network hiccup, a full disk).
- Negative capability claims ("tool X doesn't work") — these harden into refusals that outlive the actual problem.
- Transient errors that resolved on retry.
- One-off task narratives — a skill is a procedure that will be right next month, not a log of what happened today.
- Unresolved failures dressed up as procedure. If the session didn't find a working approach, there is nothing to teach.

Skill writes happen through `skill_manage` and are subject to rails: you can only modify workspace skills you have read this session, and installed or pinned skills are off-limits. If a rail refuses a write, do not fight it — note the lesson as a memory instead.
{% endif %}
126 changes: 112 additions & 14 deletions src/agent/channel.rs
Original file line number Diff line number Diff line change
Expand Up @@ -715,6 +715,12 @@ pub struct Channel {
message_count: usize,
/// When the last memory persistence branch was triggered.
last_persistence_at: std::time::Instant,
/// Set when a turn or worker crossed the reflection work threshold.
/// Consumed by the next persistence branch, which then also reflects
/// on skills. Atomic because turn processing marks it through `&self`.
reflection_signal: std::sync::atomic::AtomicBool,
/// When the last skill-reflection pass was spawned, for cooldown.
last_reflection_at: Option<std::time::Instant>,
/// Branch IDs for silent memory persistence branches (results not injected into history).
memory_persistence_branches: HashSet<BranchId>,
/// Optional Discord reply target captured when each branch was started.
Expand Down Expand Up @@ -907,6 +913,8 @@ impl Channel {
message_count: 0,
last_persistence_at: std::time::Instant::now(),
memory_persistence_branches: HashSet::new(),
reflection_signal: std::sync::atomic::AtomicBool::new(false),
last_reflection_at: None,
branch_reply_targets: HashMap::new(),
coalesce_buffer: Vec::new(),
coalesce_deadline: None,
Expand Down Expand Up @@ -2889,6 +2897,25 @@ impl Channel {
.await;
}

// Count tool iterations from the pre-sanitization turn history.
// apply_history_after_turn strips tool-call messages on
// PromptCancelled (the normal reply-tool ending), so counting the
// applied guard would report zero for exactly the tool-heavy turns
// reflection exists to catch. PromptCancelled and MaxTurnsError carry
// the authoritative messages in the error; Ok turns carry them in
// `history`.
let turn_tool_calls = {
let source: &[rig::message::Message] = match &result {
Err(rig::completion::PromptError::PromptCancelled { chat_history, .. })
| Err(rig::completion::PromptError::MaxTurnsError { chat_history, .. }) => {
chat_history
}
_ => &history,
};
let appended_from = history_len_before.min(source.len());
crate::agent::channel_history::count_tool_call_messages(&source[appended_from..])
};

let applied_history = {
let mut guard = self.state.history.write().await;
apply_history_after_turn(
Expand All @@ -2901,6 +2928,13 @@ impl Channel {
)
};

{
let reflection = self.deps.runtime_config.skills_config.load().reflection;
if reflection.enabled && turn_tool_calls >= reflection.min_tool_iterations {
self.mark_reflection_signal("turn_tool_calls");
}
}

let remove_result = match self.resolved_settings.delegation {
DelegationMode::Direct => {
crate::tools::remove_direct_mode_tools(&self.tool_server, allow_direct_reply).await
Expand Down Expand Up @@ -3366,6 +3400,19 @@ impl Channel {

run_logger.log_worker_completed(*worker_id, result, *success);

// A worker finishing real work successfully is a reflection
// signal: the session likely produced a reusable lesson.
if *success {
let reflection = self.deps.runtime_config.skills_config.load().reflection;
if reflection.enabled {
self.mark_reflection_signal("worker_completed");
// A worker can finish after the last user turn; check
// now so reflection doesn't sit pending until the next
// inbound message.
self.check_memory_persistence().await;
}
}

Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
self.state.active_workers.write().await.remove(worker_id);
self.state.worker_inputs.write().await.remove(worker_id);
self.state.worker_injections.write().await.remove(worker_id);
Expand Down Expand Up @@ -3666,33 +3713,73 @@ impl Channel {
status.render_full(&current_time_line, &system_info)
}

/// Note that this conversation just did substantial work. The next
/// persistence pass will also reflect on skills, cooldown permitting.
///
/// Cron conversations never reflect: scheduled runs repeat the same
/// procedure on a timer and would grind out noise skills.
fn mark_reflection_signal(&self, source: &'static str) {
if self.id.starts_with("cron") {
return;
}
let was_set = self
.reflection_signal
.swap(true, std::sync::atomic::Ordering::Relaxed);
if !was_set {
tracing::debug!(channel_id = %self.id, source, "skill reflection signal set");
}
}

/// Whether the next persistence pass should reflect on skills.
fn reflection_due(&self) -> bool {
if !self
.reflection_signal
.load(std::sync::atomic::Ordering::Relaxed)
{
return false;
}
let config = self.deps.runtime_config.skills_config.load().reflection;
if !config.enabled {
return false;
}
match self.last_reflection_at {
Some(at) => at.elapsed().as_secs() >= config.cooldown_secs,
None => true,
}
}

/// Check if a memory persistence branch should be spawned.
///
/// Three triggers (any one fires):
/// 1. **Message count** — threshold reached (default 20, configurable)
/// 2. **Time-based** — elapsed since last persistence, if conversation is active
/// 3. **Event density** — working memory events from this channel since last persistence
/// Three memory triggers (any one fires): message count, time since last
/// persistence, and working-memory event density. A pending skill
/// reflection signal is a fourth trigger; the branch it spawns also
/// reflects on skills. Reflection rides the persistence pass, so every
/// trigger — reflection included — obeys the conversation's memory
/// persistence controls.
async fn check_memory_persistence(&mut self) {
let config = **self.deps.runtime_config.memory_persistence.load();
if !config.enabled
|| config.message_interval == 0
|| !self.resolved_settings.memory.persistence_enabled()
{
let persistence_enabled = config.enabled
&& config.message_interval != 0
&& self.resolved_settings.memory.persistence_enabled();
if !persistence_enabled {
return;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
}
let reflection_due = self.reflection_due();

let wm_config = **self.deps.runtime_config.working_memory.load();
let elapsed = self.last_persistence_at.elapsed();

// Trigger 1: Message count threshold.
let message_trigger = self.message_count >= wm_config.persistence_message_threshold;
let message_trigger =
persistence_enabled && self.message_count >= wm_config.persistence_message_threshold;

// Trigger 2: Time-based — only if conversation is active (message_count > 0).
let time_trigger = self.message_count > 0
let time_trigger = persistence_enabled
&& self.message_count > 0
&& elapsed.as_secs() >= wm_config.persistence_time_threshold_secs;

// Trigger 3: Event density — working memory events from this channel.
let density_trigger = if !message_trigger && !time_trigger {
let density_trigger = if persistence_enabled && !message_trigger && !time_trigger {
// Only check DB if the cheap triggers didn't fire.
let since = chrono::Utc::now() - chrono::Duration::seconds(elapsed.as_secs() as i64);
match self
Expand All @@ -3711,29 +3798,40 @@ impl Channel {
false
};

if !message_trigger && !time_trigger && !density_trigger {
if !message_trigger && !time_trigger && !density_trigger && !reflection_due {
return;
}

let trigger = if message_trigger {
"message_count"
} else if time_trigger {
"time"
} else {
} else if density_trigger {
"event_density"
} else {
"reflection"
};

// Reset counters before spawning so subsequent messages don't pile up.
self.message_count = 0;
self.last_persistence_at = std::time::Instant::now();

match spawn_memory_persistence_branch(&self.state, &self.deps).await {
match spawn_memory_persistence_branch(&self.state, &self.deps, reflection_due).await {
Ok(branch_id) => {
// Consume the reflection request only once the branch exists;
// a failed spawn leaves the signal set so the next check
// retries instead of losing the reflection for a cooldown.
if reflection_due {
self.last_reflection_at = Some(std::time::Instant::now());
self.reflection_signal
.store(false, std::sync::atomic::Ordering::Relaxed);
}
self.memory_persistence_branches.insert(branch_id);
tracing::info!(
channel_id = %self.id,
branch_id = %branch_id,
trigger,
skill_reflection = reflection_due,
"memory persistence branch spawned"
);
}
Expand Down
14 changes: 12 additions & 2 deletions src/agent/channel_dispatch.rs
Original file line number Diff line number Diff line change
Expand Up @@ -212,9 +212,14 @@ pub async fn spawn_branch_from_state(
/// Uses the same branching infrastructure as regular branches but with a
/// dedicated prompt focused on memory recall + save. The result is not injected
/// into channel history — the channel handles these branch IDs specially.
///
/// When `skill_reflection` is set, the same pass also reflects on skills:
/// the prompt gains the reflection section and the branch gets skill tools
/// under agent-origin rails.
pub(crate) async fn spawn_memory_persistence_branch(
state: &ChannelState,
deps: &AgentDeps,
skill_reflection: bool,
) -> std::result::Result<BranchId, AgentError> {
let contract_state = Arc::new(MemoryPersistenceContractState::default());

Expand All @@ -223,7 +228,7 @@ pub(crate) async fn spawn_memory_persistence_branch(
let model_name = routing.resolve(ProcessType::Branch, None).to_string();
let tool_use_enforcement = deps.runtime_config.tool_use_enforcement.load();
let system_prompt = prompt_engine
.render_static("memory_persistence")
.render_memory_persistence_prompt(skill_reflection)
.and_then(|prompt| {
prompt_engine.maybe_append_tool_use_enforcement(
prompt,
Expand All @@ -241,13 +246,18 @@ pub(crate) async fn spawn_memory_persistence_branch(
"memory persistence",
&prompt,
&system_prompt,
"persisting memories...",
if skill_reflection {
"persisting memories and reflecting on skills..."
} else {
"persisting memories..."
},
"memory_persistence_branch",
BranchSpawnOptions {
profile: BranchToolProfile::MemoryPersistence {
contract_state,
working_memory: Some(state.deps.working_memory.clone()),
channel_id: Some(state.channel_id.to_string()),
skill_reflection,
},
},
)
Expand Down
45 changes: 44 additions & 1 deletion src/agent/channel_history.rs
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,25 @@ pub(crate) struct AppliedHistory {
pub reply_text: Option<String>,
}

/// Count assistant messages carrying tool calls in a history slice.
///
/// Used as the per-turn work measure for skill reflection: each counted
/// message is one tool iteration of the agentic loop.
pub(crate) fn count_tool_call_messages(messages: &[rig::message::Message]) -> usize {
messages
.iter()
.filter(|message| {
if let rig::message::Message::Assistant { content, .. } = message {
content
.iter()
.any(|c| matches!(c, rig::message::AssistantContent::ToolCall(_)))
} else {
false
}
})
.count()
}

pub(crate) fn pop_retrigger_bridge_message(history: &mut Vec<rig::message::Message>) -> bool {
if history.last().is_some_and(is_retrigger_bridge_message) {
history.pop();
Expand Down Expand Up @@ -496,7 +515,7 @@ pub(crate) fn event_is_for_channel(event: &ProcessEvent, channel_id: &ChannelId)

#[cfg(test)]
mod tests {
use super::{apply_history_after_turn, event_is_for_channel};
use super::{apply_history_after_turn, count_tool_call_messages, event_is_for_channel};
use crate::{ChannelId, ProcessEvent, ProcessId};
use rig::completion::{CompletionError, PromptError};
use rig::message::Message;
Expand Down Expand Up @@ -800,6 +819,30 @@ mod tests {
);
}

#[test]
fn count_tool_call_messages_counts_only_tool_call_assistants() {
let tool_call_msg = Message::Assistant {
id: None,
content: rig::OneOrMany::one(rig::message::AssistantContent::tool_call(
"call_1",
"shell",
serde_json::json!({"command": "ls"}),
)),
};

let messages = vec![
user_msg("do the thing"),
tool_call_msg.clone(),
user_msg("(tool result)"),
tool_call_msg,
assistant_msg("done"),
];

assert_eq!(count_tool_call_messages(&messages), 2);
assert_eq!(count_tool_call_messages(&[]), 0);
assert_eq!(count_tool_call_messages(&make_history(&["hi", "hey"])), 0);
}

/// Rollback on empty history is a no-op and must not panic.
#[test]
fn rollback_on_empty_history_is_noop() {
Expand Down
1 change: 1 addition & 0 deletions src/agent/ingestion.rs
Original file line number Diff line number Diff line change
Expand Up @@ -505,6 +505,7 @@ async fn process_chunk(
contract_state: contract_state.clone(),
working_memory: Some(deps.working_memory.clone()),
channel_id: None,
skill_reflection: false,
},
deps.api_state.clone(),
deps.wiki_store.clone(),
Expand Down
1 change: 1 addition & 0 deletions src/api/agents.rs
Original file line number Diff line number Diff line change
Expand Up @@ -760,6 +760,7 @@ pub async fn create_agent_internal(
};

let raw_config = crate::config::AgentConfig {
skills: None,
id: agent_id.clone(),
default: false,
display_name: request.display_name.clone().filter(|s| !s.is_empty()),
Expand Down
Loading
Loading