Merge pull request #218 from ScriptedAlchemy/codex/activity-coupled-triggering

feat(automation): couple scheduler triggering to LCM session activity

Author: ScriptedAlchemy

docs/SELF-IMPROVING-LOOPS-CONTRACTS.md

@@ -27,6 +27,12 @@ TraceDecay standalone automation is time-scheduled by the daemon, not by Codex n
 | `session_reflector` | Every 15 minutes, with a 5-minute cooldown | Validated accepted session facts auto-apply under the same memory auto-apply policy; otherwise they stay as dashboard fact proposals. |
 | `skill_writer` | Every 60 minutes, after a 15-minute idle window, with a 5-minute cooldown | Creates or updates managed skill drafts; skills are not auto-enabled while `auto_enable_skills=false`. |
 
+Scheduling is activity-coupled, not purely wall-clock. The scheduler reads the newest LCM session-message timestamp for the project store as its session-activity signal on every tick:
+
+- `min_idle_secs` is a true idle window: the task only runs after the project has been quiet (no LCM session ingest) for at least that long. A missing session store counts as idle.
+- `session_reflector` and `skill_writer` additionally require new session activity since their last successful run; when nothing new landed they skip with `no_new_session_activity` instead of re-reviewing the same transcripts. Because skips do not consume the interval clock, the task fires on the first tick after fresh activity lands (once the idle window is satisfied).
+- `memory_curator` reviews the fact store rather than session transcripts, so it keeps the plain interval/cooldown cadence.
+
 The daemon loop is the host for these jobs. It should not create Codex top-level chats for scheduler work, and it should not rely on Codex native recurring automations for liveness. Host backends provide the model call; TraceDecay owns evidence collection, validation, ledgers, apply policy, and scheduler state.
 
 ## Standalone And Delegated Modes

src/automation/lifecycle.rs

@@ -16,7 +16,8 @@ use super::run_ledger::{
     AutomationTrigger,
 };
 use super::scheduler::{
-    schedule_decision, stale_lock_secs, AutomationScheduleDecision, AutomationTaskLock,
+    load_session_activity, schedule_decision, stale_lock_secs, AutomationScheduleDecision,
+    AutomationTaskLock,
 };
 use crate::errors::{Result, TraceDecayError};
 use crate::tracedecay::current_timestamp;
@@ -37,6 +38,9 @@ pub(crate) struct AgentTaskRunContext<'a> {
     pub(crate) run_id: String,
     pub(crate) trigger: AutomationTrigger,
     pub(crate) dashboard_root: PathBuf,
+    /// LCM sessions database for the project store; the scheduler gate reads
+    /// its newest message timestamp as the session-activity signal.
+    sessions_db_path: PathBuf,
     config: &'a AutomationConfig,
     task: AgentTaskKind,
     started_at: String,
@@ -50,6 +54,7 @@ pub(crate) struct AgentTaskRunContext<'a> {
 impl<'a> AgentTaskRunContext<'a> {
     pub(crate) fn new(
         dashboard_root: PathBuf,
+        sessions_db_path: PathBuf,
         run_id: Option<String>,
         run_id_prefix: &'static str,
         trigger: AutomationTrigger,
@@ -60,6 +65,7 @@ impl<'a> AgentTaskRunContext<'a> {
             run_id: run_id.unwrap_or_else(|| generated_run_id(run_id_prefix)),
             trigger,
             dashboard_root,
+            sessions_db_path,
             config,
             task,
             started_at: current_timestamp().to_string(),
@@ -72,8 +78,14 @@ impl<'a> AgentTaskRunContext<'a> {
     }
 
     pub(crate) async fn gate(&mut self) -> Result<SchedulerGate> {
-        let (gate, records) =
-            task_run_gate(self.config, &self.dashboard_root, self.task, self.trigger).await?;
+        let (gate, records) = task_run_gate(
+            self.config,
+            &self.dashboard_root,
+            &self.sessions_db_path,
+            self.task,
+            self.trigger,
+        )
+        .await?;
         self.ledger_records = records;
         Ok(gate)
     }
@@ -149,6 +161,7 @@ pub(crate) fn task_skip_reason(
 pub(crate) async fn scheduler_gate(
     config: &AutomationConfig,
     dashboard_root: &Path,
+    sessions_db_path: &Path,
     task: AgentTaskKind,
     trigger: AutomationTrigger,
 ) -> Result<(SchedulerGate, Option<Vec<AutomationRunLedgerRecord>>)> {
@@ -169,7 +182,8 @@ pub(crate) async fn scheduler_gate(
         return Ok((SchedulerGate::Skip("scheduler_lock_active"), Some(records)));
     };
 
-    let decision = schedule_decision(config, task, &records, now_secs);
+    let activity = load_session_activity(sessions_db_path).await;
+    let decision = schedule_decision(config, task, &records, activity, now_secs);
     if let Some(reason) = scheduler_skip_reason(&decision, task) {
         return Ok((SchedulerGate::Skip(reason), Some(records)));
     }
@@ -180,10 +194,12 @@ pub(crate) async fn scheduler_gate(
 pub(crate) async fn task_run_gate(
     config: &AutomationConfig,
     dashboard_root: &Path,
+    sessions_db_path: &Path,
     task: AgentTaskKind,
     trigger: AutomationTrigger,
 ) -> Result<(SchedulerGate, Option<Vec<AutomationRunLedgerRecord>>)> {
-    let (gate, records) = scheduler_gate(config, dashboard_root, task, trigger).await?;
+    let (gate, records) =
+        scheduler_gate(config, dashboard_root, sessions_db_path, task, trigger).await?;
     let gate = match gate {
         SchedulerGate::Skip(reason) => SchedulerGate::Skip(reason),
         SchedulerGate::Proceed(lock) => match task_skip_reason(config, task) {
@@ -695,6 +711,7 @@ mod tests {
         let config = AutomationConfig::default();
         let mut run = AgentTaskRunContext::new(
             dashboard_root.to_path_buf(),
+            dashboard_root.join("sessions.db"),
             Some(run_id.to_string()),
             "test",
             trigger,
@@ -825,6 +842,7 @@ mod tests {
         let config = scheduler_enabled_config();
         let mut run = AgentTaskRunContext::new(
             dashboard_root.to_path_buf(),
+            dashboard_root.join("sessions.db"),
             Some(run_id.to_string()),
             "test",
             AutomationTrigger::Scheduler,

src/automation/memory_curator.rs

@@ -53,6 +53,7 @@ pub async fn run_memory_curator_with_backend(
 ) -> Result<MemoryCuratorAutomationRun> {
     let mut run = AgentTaskRunContext::new(
         cg.store_layout().dashboard_root.clone(),
+        cg.store_layout().sessions_db_path.clone(),
         options.run_id.clone(),
         "memory_curator",
         options.trigger,

src/automation/runner.rs

@@ -175,6 +175,7 @@ pub async fn run_session_reflector_with_backend(
 ) -> Result<SessionReflectorAutomationRun> {
     let mut run = AgentTaskRunContext::new(
         cg.store_layout().dashboard_root.clone(),
+        cg.store_layout().sessions_db_path.clone(),
         options.run_id.clone(),
         "session_reflector",
         options.trigger,
@@ -446,6 +447,7 @@ pub async fn run_skill_writer_with_backend(
 ) -> Result<SkillWriterAutomationRun> {
     let mut run = AgentTaskRunContext::new(
         cg.store_layout().dashboard_root.clone(),
+        cg.store_layout().sessions_db_path.clone(),
         options.run_id.clone(),
         "skill_writer",
         options.trigger,

src/automation/scheduler.rs

@@ -9,6 +9,7 @@ use super::config::{
 };
 use super::run_ledger::{AutomationRunLedgerRecord, AutomationRunStatus, AutomationTrigger};
 use crate::errors::{Result, TraceDecayError};
+use crate::global_db::GlobalDb;
 
 const DEFAULT_FAILURE_COOLDOWN_SECS: u64 = 300;
 const DEFAULT_STALE_LOCK_SECS: u64 = 6 * 60 * 60;
@@ -27,6 +28,42 @@ pub enum AutomationSchedule {
     Interval { every_secs: u64 },
 }
 
+/// Most recent LCM session ingest activity for the project, in unix seconds.
+///
+/// `None` means the session store does not exist yet or holds no timestamped
+/// messages; gates that need an activity signal treat that as "no activity
+/// observed" rather than an error.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub struct SessionActivity {
+    pub last_activity_secs: Option<i64>,
+}
+
+impl SessionActivity {
+    pub fn none() -> Self {
+        Self::default()
+    }
+
+    pub fn at(last_activity_secs: i64) -> Self {
+        Self {
+            last_activity_secs: Some(last_activity_secs),
+        }
+    }
+}
+
+/// Reads the session-activity signal from the LCM sessions database.
+///
+/// This is a single indexed `ORDER BY timestamp DESC LIMIT 1` lookup against
+/// the read-only store, so it is cheap and race-safe to call from every
+/// scheduler tick; concurrent ingest writers only ever move the value forward.
+pub async fn load_session_activity(sessions_db_path: &Path) -> SessionActivity {
+    let Some(db) = GlobalDb::open_read_only_at(sessions_db_path).await else {
+        return SessionActivity::none();
+    };
+    SessionActivity {
+        last_activity_secs: db.latest_session_activity_secs().await,
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct AutomationScheduleDecision {
     skip_reason: Option<&'static str>,
@@ -170,6 +207,7 @@ pub fn schedule_decision(
     config: &AutomationConfig,
     task: AgentTaskKind,
     records: &[AutomationRunLedgerRecord],
+    activity: SessionActivity,
     now_secs: i64,
 ) -> AutomationScheduleDecision {
     if !config.enabled {
@@ -200,10 +238,12 @@ pub fn schedule_decision(
         return AutomationScheduleDecision::skipped("scheduler_schedule_manual");
     };
 
+    // `min_idle_secs` is a true idle window: the project must have been quiet
+    // (no LCM session ingest activity) for at least this long. An unknown
+    // activity signal (no session store yet) counts as idle.
     if let Some(min_idle_secs) = task_config.min_idle_secs {
-        if let Some(record) = latest_non_skipped_record(records, task, None) {
-            let completed_at = record.completed_at.parse::<i64>().ok().unwrap_or(0);
-            if elapsed_secs(completed_at, now_secs) < min_idle_secs {
+        if let Some(last_activity) = activity.last_activity_secs {
+            if elapsed_secs(last_activity, now_secs) < min_idle_secs {
                 return AutomationScheduleDecision::skipped("scheduler_idle_window_active");
             }
         }
@@ -235,9 +275,34 @@ pub fn schedule_decision(
         }
     }
 
+    // Session-evidence tasks only re-run when new session activity landed
+    // after their last successful run started; a run without fresh evidence
+    // would re-review the same transcript slices. Skips do not consume the
+    // interval clock, so the task fires on the first tick after new activity.
+    if task_consumes_session_evidence(task) {
+        if let Some(record) = latest_successful_record(records, task) {
+            let started_at = record.started_at.parse::<i64>().ok().unwrap_or(0);
+            let has_new_activity = activity
+                .last_activity_secs
+                .is_some_and(|last_activity| last_activity > started_at);
+            if !has_new_activity {
+                return AutomationScheduleDecision::skipped("no_new_session_activity");
+            }
+        }
+    }
+
     AutomationScheduleDecision::due()
 }
 
+/// Tasks whose evidence comes from the LCM session store; they are gated on
+/// new session activity since their last successful run.
+fn task_consumes_session_evidence(task: AgentTaskKind) -> bool {
+    match task {
+        AgentTaskKind::SessionReflector | AgentTaskKind::SkillWriter => true,
+        AgentTaskKind::MemoryCurator => false,
+    }
+}
+
 pub fn stale_lock_secs(config: &AutomationConfig, task: AgentTaskKind) -> Option<u64> {
     task_config(config, task)
         .stale_lock_secs
@@ -303,6 +368,16 @@ fn task_config(config: &AutomationConfig, task: AgentTaskKind) -> &AutomationTas
     }
 }
 
+fn latest_successful_record(
+    records: &[AutomationRunLedgerRecord],
+    task: AgentTaskKind,
+) -> Option<&AutomationRunLedgerRecord> {
+    records
+        .iter()
+        .filter(|record| record.task == task && record.status == AutomationRunStatus::Succeeded)
+        .max_by_key(|record| record.completed_at.parse::<i64>().ok().unwrap_or(0))
+}
+
 fn latest_non_skipped_record(
     records: &[AutomationRunLedgerRecord],
     task: AgentTaskKind,

src/dashboard/automation_scheduler_api.rs

@@ -11,8 +11,8 @@ use crate::automation::backend::{task_key, AgentTaskKind};
 use crate::automation::config::{effective_config, load_project_config, AutomationConfig};
 use crate::automation::run_ledger::{load_run_records, AutomationRunLedgerRecord};
 use crate::automation::scheduler::{
-    load_scheduler_control, save_scheduler_control, schedule_decision, scheduler_control_path,
-    AutomationSchedulerControl,
+    load_scheduler_control, load_session_activity, save_scheduler_control, schedule_decision,
+    scheduler_control_path, AutomationSchedulerControl, SessionActivity,
 };
 use crate::automation::staged_notice::{count_pending_automation_output, AutomationPendingCounts};
 use crate::tracedecay::current_timestamp;
@@ -69,6 +69,8 @@ async fn scheduler_status_payload(state: &DashboardState) -> ApiResult {
         Err(_) => AutomationPendingCounts::default(),
     };
     let now = current_timestamp();
+    let activity =
+        load_session_activity(&state.store_root.join(crate::storage::SESSIONS_DB_FILENAME)).await;
     Ok(Json(json!({
         "status": scheduler_status_label(&effective, control.paused),
         "paused": control.paused,
@@ -77,16 +79,17 @@ async fn scheduler_status_payload(state: &DashboardState) -> ApiResult {
         "enabled": effective.enabled,
         "scheduler_tick_secs": effective.scheduler_tick_secs,
         "now": now,
+        "last_session_activity": activity.last_activity_secs,
         "project_config_path": crate::automation::config::project_config_path(&state.dashboard_root)
             .display()
             .to_string(),
         "control_path": scheduler_control_path(&state.dashboard_root)
             .display()
             .to_string(),
         "tasks": [
-            task_status(&effective, control.paused, &records, now, AgentTaskKind::MemoryCurator),
-            task_status(&effective, control.paused, &records, now, AgentTaskKind::SessionReflector),
-            task_status(&effective, control.paused, &records, now, AgentTaskKind::SkillWriter),
+            task_status(&effective, control.paused, &records, activity, now, AgentTaskKind::MemoryCurator),
+            task_status(&effective, control.paused, &records, activity, now, AgentTaskKind::SessionReflector),
+            task_status(&effective, control.paused, &records, activity, now, AgentTaskKind::SkillWriter),
         ],
     })))
 }
@@ -95,13 +98,14 @@ fn task_status(
     config: &AutomationConfig,
     paused: bool,
     records: &[AutomationRunLedgerRecord],
+    activity: SessionActivity,
     now: i64,
     task: AgentTaskKind,
 ) -> Value {
     let decision = if paused {
         crate::automation::scheduler::AutomationScheduleDecision::skipped("scheduler_paused")
     } else {
-        schedule_decision(config, task, records, now)
+        schedule_decision(config, task, records, activity, now)
     };
     let latest_scheduler = records
         .iter()

src/global_db.rs

@@ -2297,6 +2297,32 @@ impl GlobalDb {
         Ok(out)
     }
 
+    /// Timestamp of the most recent ingested session message, in unix
+    /// seconds, or `None` when no timestamped messages exist. Providers store
+    /// either seconds or milliseconds; millisecond values are normalized.
+    /// Backed by `idx_session_messages_timestamp`, so this is a cheap index
+    /// seek suitable for every automation scheduler tick.
+    pub async fn latest_session_activity_secs(&self) -> Option<i64> {
+        let mut rows = self
+            .conn
+            .query(
+                "SELECT timestamp FROM session_messages
+                 WHERE timestamp IS NOT NULL
+                 ORDER BY timestamp DESC
+                 LIMIT 1",
+                (),
+            )
+            .await
+            .ok()?;
+        let row = rows.next().await.ok()??;
+        let timestamp = row.get::<i64>(0).ok()?;
+        Some(if timestamp >= 1_000_000_000_000 {
+            timestamp / 1000
+        } else {
+            timestamp
+        })
+    }
+
     /// Inserts or replaces a provider message. Returns `false` on any DB error.
     pub async fn upsert_session_message(&self, message: &SessionMessageRecord) -> bool {
         if self.conn.execute("BEGIN IMMEDIATE", ()).await.is_err() {