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 `write_file` 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.

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 %}
109 changes: 92 additions & 17 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 @@ -2923,18 +2931,29 @@ impl Channel {
.await;
}

let applied_history = {
let (applied_history, turn_tool_calls) = {
let mut guard = self.state.history.write().await;
apply_history_after_turn(
let applied = apply_history_after_turn(
&result,
&mut guard,
history,
history_len_before,
&self.id,
is_retrigger,
)
);
let appended_from = history_len_before.min(guard.len());
let tool_calls =
crate::agent::channel_history::count_tool_call_messages(&guard[appended_from..]);
(applied, tool_calls)
};

{
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 @@ -3400,6 +3419,15 @@ 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");
}
}

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 @@ -3700,33 +3728,72 @@ 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 and can spawn the pass on its
/// own; the branch it spawns also reflects on skills.
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();
let reflection_due = self.reflection_due();

if !persistence_enabled && !reflection_due {
return;
}

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 @@ -3745,29 +3812,37 @@ 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();
if reflection_due {
self.last_reflection_at = Some(std::time::Instant::now());
self.reflection_signal
.store(false, std::sync::atomic::Ordering::Relaxed);
}

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) => {
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 @@ -759,6 +759,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