opencrabs 0.3.37

The autonomous, self-improving AI agent. Single Rust binary. Every channel. Install with: cargo install opencrabs
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
//! Cron Scheduler
//!
//! Background task that checks the `cron_jobs` table every 60 seconds,
//! executes due jobs in a dedicated "Cron" session, and delivers results
//! to the configured channel. Cron jobs are fully isolated from the TUI —
//! they never share or mutate the user's active session.

use crate::channels::ChannelFactory;
use crate::config::Config;
use crate::db::CronJobRepository;
use crate::db::CronJobRunRepository;
use crate::db::models::{CronJob, CronJobRun};
use crate::services::{ServiceContext, SessionService};
use chrono::Utc;
use cron::Schedule;
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::Mutex;
use uuid::Uuid;

/// Name used for the shared cron session.
const CRON_SESSION_NAME: &str = "Cron";

/// Whether `job_profile` is the active process profile (so the cheap, already
/// wired factory agent can run it) rather than a foreign profile that needs its
/// own config + brain materialized. `None` = legacy pre-stamping row, treated as
/// the active profile. The base profile is stored as the literal "default".
fn is_active_profile(job_profile: Option<&str>, active: Option<&str>) -> bool {
    match job_profile {
        None => true,
        Some(stamped) => stamped == active.unwrap_or("default"),
    }
}

/// Reserved cron-job name for a one-shot background `/rebuild`. The scheduler
/// special-cases this name: instead of running an agent prompt it builds from
/// source and exec-restarts into the new binary, then the job removes itself.
/// The originating session id is carried in `prompt` so the restart resumes
/// the user's session.
pub const REBUILD_JOB_NAME: &str = "__opencrabs_rebuild__";

/// Schedule a one-shot background rebuild for `session_id`. Returns once the
/// job is queued — the build runs out-of-band on the scheduler's next tick
/// (within ~60s), so the calling session is never blocked. `deliver_to` (if
/// set) receives a status message; the reload resumes `session_id`.
pub async fn schedule_background_rebuild(
    pool: crate::db::Pool,
    session_id: Uuid,
    deliver_to: Option<String>,
) -> anyhow::Result<()> {
    let repo = CronJobRepository::new(pool);
    // Remove any stale rebuild job first so we never stack two builds.
    if let Ok(existing) = repo.list_all().await {
        for j in existing.iter().filter(|j| j.name == REBUILD_JOB_NAME) {
            let _ = repo.delete(&j.id.to_string()).await;
        }
    }
    let now = Utc::now();
    let job = CronJob {
        id: Uuid::new_v4(),
        name: REBUILD_JOB_NAME.to_string(),
        // Every minute → the next tick (within 60s) picks it up; the job
        // deletes itself on pickup so it runs exactly once.
        cron_expr: "* * * * *".to_string(),
        timezone: "UTC".to_string(),
        prompt: session_id.to_string(),
        provider: None,
        model: None,
        thinking: "off".to_string(),
        auto_approve: true,
        deliver_to,
        deliver_api_key: None,
        enabled: true,
        last_run_at: None,
        next_run_at: None,
        created_at: now,
        updated_at: now,
        // Stamp the current profile so the guard in `tick()` lets it run here.
        // current_profile_name() honors the task-local profile scope.
        profile_name: Some(crate::config::profile::current_profile_name()),
    };
    repo.insert(&job).await?;
    tracing::info!("Background rebuild queued for session {session_id}");
    Ok(())
}

/// Execute the reserved background-rebuild job: delete it first (one-shot, no
/// retry on the 60s tick), build from source, then exec-restart into the
/// freshly-built binary (replaces the whole process). On failure it reports
/// to `deliver_to` and returns. The originating session id is in `job.prompt`.
async fn run_rebuild_job(job: &CronJob, ctx: &ServiceContext) -> anyhow::Result<()> {
    use crate::brain::SelfUpdater;

    // Delete up front so a long/failed build can't re-trigger next tick.
    let repo = CronJobRepository::new(ctx.pool());
    if let Err(e) = repo.delete(&job.id.to_string()).await {
        tracing::error!("rebuild job: failed to delete self: {e}");
    }

    let session_id = Uuid::parse_str(job.prompt.trim()).unwrap_or_else(|_| Uuid::nil());
    tracing::info!("Background rebuild starting (will resume session {session_id})");

    let updater =
        SelfUpdater::auto_detect().map_err(|e| anyhow::anyhow!("rebuild: auto_detect: {e}"))?;

    match updater
        .build_streaming(|line| tracing::debug!("rebuild: {line}"))
        .await
    {
        Ok(built_path) => {
            tracing::info!(
                "Background rebuild succeeded: {} — reloading",
                built_path.display()
            );
            deliver_rebuild_status(
                job,
                "✅ Rebuilt from source — reloading into the new binary now.",
            )
            .await;
            // exec() replaces the entire process (this scheduler task too).
            if let Err(e) = SelfUpdater::restart_into(&built_path, session_id) {
                tracing::error!("Background rebuild restart failed: {e}");
                return Err(anyhow::anyhow!("rebuild restart failed: {e}"));
            }
            Ok(()) // unreachable on success
        }
        Err(out) => {
            tracing::error!("Background rebuild failed: {out}");
            deliver_rebuild_status(job, &format!("⚠️ Background rebuild failed:\n{out}")).await;
            Ok(())
        }
    }
}

/// Deliver a rebuild status line to the job's configured channels (if any).
async fn deliver_rebuild_status(job: &CronJob, msg: &str) {
    if let Some(ref deliver_to) = job.deliver_to {
        for target in deliver_to
            .split(',')
            .map(str::trim)
            .filter(|s| !s.is_empty())
        {
            deliver_result(target, &job.name, msg, job.deliver_api_key.as_deref()).await;
        }
    }
}

/// Background cron scheduler that polls the database and executes due jobs.
pub struct CronScheduler {
    repo: CronJobRepository,
    run_repo: CronJobRunRepository,
    factory: Arc<ChannelFactory>,
    service_context: ServiceContext,
    /// Dedicated session for all cron jobs — isolated from TUI sessions.
    cron_session_id: Option<Uuid>,
    /// Kept for API compatibility but no longer used for session resolution.
    #[allow(dead_code)]
    shared_session_id: Arc<Mutex<Option<Uuid>>>,
}

impl CronScheduler {
    pub fn new(
        repo: CronJobRepository,
        run_repo: CronJobRunRepository,
        factory: Arc<ChannelFactory>,
        service_context: ServiceContext,
        shared_session_id: Arc<Mutex<Option<Uuid>>>,
    ) -> Self {
        Self {
            repo,
            run_repo,
            factory,
            service_context,
            cron_session_id: None,
            shared_session_id,
        }
    }

    /// Spawn the scheduler as a background tokio task.
    /// Polls every 60 seconds for due jobs.
    pub fn spawn(self) -> tokio::task::JoinHandle<()> {
        tokio::spawn(self.run())
    }

    /// Run the polling loop in the CURRENT task (no internal spawn). The
    /// multi-profile daemon drives this directly inside a
    /// `with_profile_home_async(profile, ...)` scope so the scheduler's own
    /// setup (cron session, config reads) resolves to that profile's home.
    /// `spawn()` is the thin wrapper for callers that just want it backgrounded.
    pub async fn run(mut self) {
        // Find or create the dedicated cron session
        match self.resolve_or_create_cron_session().await {
            Ok(id) => {
                self.cron_session_id = Some(id);
                tracing::info!(
                    "Cron scheduler started — polling every 60s, cron session: {}",
                    id
                );
            }
            Err(e) => {
                tracing::error!("Cron scheduler failed to create session: {e}");
            }
        }
        loop {
            if let Err(e) = self.tick().await {
                tracing::error!("Cron scheduler tick error: {e}");
            }
            tokio::time::sleep(std::time::Duration::from_secs(60)).await;
        }
    }

    /// Find an existing "Cron" session or create one.
    async fn resolve_or_create_cron_session(&self) -> anyhow::Result<Uuid> {
        use crate::db::repository::SessionListOptions;
        let session_svc = SessionService::new(self.service_context.clone());
        // Look for an existing session named "Cron"
        let sessions = session_svc
            .list_sessions(SessionListOptions {
                include_archived: false,
                limit: None,
                offset: 0,
            })
            .await?;
        if let Some(existing) = sessions
            .iter()
            .find(|s| s.title.as_deref().is_some_and(|n| n == CRON_SESSION_NAME))
        {
            return Ok(existing.id);
        }
        // Create a new dedicated cron session
        let config = Config::load()?;
        let provider = config.cron.default_provider.clone();
        let model = config.cron.default_model.clone();
        let session = session_svc
            .create_session_with_provider(Some(CRON_SESSION_NAME.to_string()), provider, model)
            .await?;
        Ok(session.id)
    }

    /// One scheduler tick: check all enabled jobs and execute any that are due.
    async fn tick(&self) -> anyhow::Result<()> {
        let jobs = self.repo.list_enabled().await?;
        let now = Utc::now();

        for job in &jobs {
            if self.is_due(job, now) {
                tracing::info!("Cron job '{}' ({}) is due — executing", job.name, job.id);

                // Calculate next run time before executing (so we don't re-trigger)
                let next_run = self.next_run_after(job, now);
                let next_run_str = next_run.map(|dt| dt.to_rfc3339());
                self.repo
                    .update_last_run(&job.id.to_string(), next_run_str.as_deref())
                    .await?;

                // Execute in background so we don't block other jobs
                let Some(cron_sid) = self.cron_session_id else {
                    tracing::error!(
                        "Cron job '{}' — no cron session available, skipping",
                        job.name
                    );
                    continue;
                };
                let job = job.clone();
                let factory = self.factory.clone();
                let ctx = self.service_context.clone();
                let run_repo = self.run_repo.clone();
                tokio::spawn(async move {
                    // For foreign-profile jobs, wrap the ENTIRE execution in a
                    // task-local profile home scope. This means every tool call
                    // the agent makes (memory writes, config reads, file ops,
                    // brain reads) resolves to the job's profile home, not the
                    // process profile. The scope lives until the task ends, so
                    // it persists across every .await inside the agent loop.
                    //
                    // This spawned task does NOT inherit the scheduler's own
                    // task-local home (tokio::spawn drops it), so it defaults to
                    // the process global. We therefore scope whenever the job's
                    // profile differs from the process global, which is exactly
                    // the multi-profile daemon case: a per-profile scheduler's
                    // jobs are stamped with a non-global profile and get scoped
                    // here.
                    let profile = job.profile_name.as_deref();
                    let active = crate::config::profile::active_profile().unwrap_or("default");
                    let needs_scope = profile.is_some() && profile != Some(active);

                    let result = if needs_scope {
                        crate::config::profile::with_profile_home_async(profile, async {
                            tracing::info!(
                                "Cron job '{}' — task-local profile home set to {:?}",
                                job.name,
                                crate::config::opencrabs_home()
                            );
                            execute_job(&job, &factory, &ctx, cron_sid, &run_repo).await
                        })
                        .await
                    } else {
                        execute_job(&job, &factory, &ctx, cron_sid, &run_repo).await
                    };

                    if let Err(e) = result {
                        tracing::error!("Cron job '{}' failed: {e}", job.name);
                    }
                });
            }
        }

        Ok(())
    }

    /// Check if a job is due to run.
    fn is_due(&self, job: &CronJob, now: chrono::DateTime<Utc>) -> bool {
        match &job.next_run_at {
            // If next_run_at is set and is in the past (or now), it's due
            Some(next) => *next <= now,
            // If next_run_at is None (first run), calculate from cron and check
            None => {
                // For first-time jobs, check if the current minute matches
                let cron_str = format!("0 {}", job.cron_expr);
                if let Ok(schedule) = Schedule::from_str(&cron_str) {
                    // If any upcoming time is within the next 60s, it's due
                    if let Some(next) = schedule.upcoming(Utc).next() {
                        let diff = next - now;
                        diff.num_seconds() <= 60
                    } else {
                        false
                    }
                } else {
                    tracing::warn!(
                        "Invalid cron expression for job '{}': {}",
                        job.name,
                        job.cron_expr
                    );
                    false
                }
            }
        }
    }

    /// Calculate the next run time after a given point.
    fn next_run_after(
        &self,
        job: &CronJob,
        after: chrono::DateTime<Utc>,
    ) -> Option<chrono::DateTime<Utc>> {
        let cron_str = format!("0 {}", job.cron_expr);
        Schedule::from_str(&cron_str)
            .ok()
            .and_then(|s| s.after(&after).next())
    }
}

/// Resolve the `(Config, AgentService)` a job should run with.
///
/// Jobs created in the active profile (and legacy unstamped jobs) use the
/// already-wired factory agent. A job stamped with a DIFFERENT profile
/// (shared-DB case, #182) gets its own config + brain + provider built from
/// that profile's home. The shared DB pool is reused since that's exactly why
/// the foreign job is visible to this scheduler at all.
async fn resolve_job_agent(
    job: &CronJob,
    factory: &ChannelFactory,
    ctx: &ServiceContext,
) -> anyhow::Result<(Config, Arc<crate::brain::agent::AgentService>)> {
    // Use the task-local profile (set by the per-job home scope above) rather
    // than the process global, so a job running under its own profile scope
    // is recognized as "local" and reuses the in-scope factory/config instead
    // of needlessly re-materializing. Falls back to the global when unscoped.
    let current = crate::config::profile::current_profile_name();
    if is_active_profile(job.profile_name.as_deref(), Some(&current)) {
        return Ok((Config::load()?, factory.create_agent_service().await));
    }

    let profile = job.profile_name.as_deref();
    tracing::info!(
        "Cron job '{}' belongs to profile {:?} (current profile {:?}); \
         running under its own profile context",
        job.name,
        profile,
        current
    );

    // Materialize config + brain from the foreign profile's home.
    // with_profile_home sets a sync scope just for these two loads.
    let (config, brain, home) = crate::config::profile::with_profile_home(profile, || {
        let config = Config::load()?;
        let home = crate::config::opencrabs_home();
        let brain = crate::brain::prompt_builder::BrainLoader::new(home.clone())
            .build_core_brain(None, None);
        anyhow::Ok((config, brain, home))
    })?;

    // Provider is built from the foreign profile's keys.
    let provider = crate::brain::provider::create_provider(&config).await?;
    let mut builder = crate::brain::agent::AgentService::new(provider, ctx.clone(), &config)
        .await
        .with_system_brain(brain)
        .with_working_directory(home.clone())
        .with_brain_path(home);
    if let Some(registry) = factory.tool_registry() {
        builder = builder.with_tool_registry(registry);
    }
    Ok((config, Arc::new(builder)))
}

/// Execute a single cron job in the shared cron session.
/// Isolated from TUI — never touches the user's active session.
/// Results are always stored in the DB; channel delivery is optional.
async fn execute_job(
    job: &CronJob,
    factory: &ChannelFactory,
    ctx: &ServiceContext,
    cron_session_id: Uuid,
    run_repo: &CronJobRunRepository,
) -> anyhow::Result<()> {
    // Reserved one-shot background rebuild — build + exec-restart, never an
    // agent prompt.
    if job.name == REBUILD_JOB_NAME {
        return run_rebuild_job(job, ctx).await;
    }

    // Resolve the config + agent for this job's profile. A job created in a
    // non-active profile (shared-DB case, #182) runs under its own profile's
    // config + brain, not the process profile's.
    let (config, agent) = resolve_job_agent(job, factory, ctx).await?;
    let effective_provider = job
        .provider
        .clone()
        .or_else(|| config.cron.default_provider.clone());
    let effective_model = job
        .model
        .clone()
        .or_else(|| config.cron.default_model.clone());

    // Pre-validate the {provider, model} pair before spawning the agent.
    // A reversed cron config (e.g. model="zhipu", provider="glm-5.1") or a
    // typo would otherwise reach the tool loop and produce confusing RSI
    // entries like "dialagram/zhipu" where a provider name leaked into the
    // model slot. Catch it early, log loudly, skip the job.
    if let Some(ref provider_name) = effective_provider
        && let Some(ref model) = effective_model
    {
        match crate::brain::provider::create_provider_by_name(&config, provider_name).await {
            Ok(provider) => {
                let supported = provider.supported_models();
                if !supported.is_empty() && !supported.iter().any(|m| m == model) {
                    tracing::error!(
                        "Cron job '{}' — model '{}' is NOT supported by provider '{}' \
                         (supported: {}). SKIPPING job — fix cron config. \
                         Either set a valid model or remove the model override to use \
                         the provider's default ('{}').",
                        job.name,
                        model,
                        provider_name,
                        supported.join(", "),
                        provider.default_model(),
                    );
                    // Record the failure so RSI surfaces it
                    let run = CronJobRun::new_running(
                        job.id,
                        job.name.clone(),
                        effective_provider.clone(),
                        effective_model.clone(),
                    );
                    let run_id = run.id.to_string();
                    if let Err(e) = run_repo.insert(&run).await {
                        tracing::error!("Failed to insert cron run record: {e}");
                    }
                    let err_msg = format!(
                        "model '{}' not supported by provider '{}' — cron config invalid",
                        model, provider_name
                    );
                    if let Err(db_err) = run_repo.complete_error(&run_id, &err_msg).await {
                        tracing::error!("Failed to save cron run error to DB: {db_err}");
                    }
                    return Ok(());
                }
            }
            Err(e) => {
                tracing::warn!(
                    "Cron job '{}' — cannot pre-validate model (provider '{}' creation \
                     failed: {e}) — proceeding with default validation",
                    job.name,
                    provider_name
                );
            }
        }
    }

    // Create a run record in the DB (status = "running")
    let run = CronJobRun::new_running(
        job.id,
        job.name.clone(),
        effective_provider.clone(),
        effective_model.clone(),
    );
    let run_id = run.id.to_string();
    if let Err(e) = run_repo.insert(&run).await {
        tracing::error!("Failed to insert cron run record: {e}");
    }

    let session_id = cron_session_id;
    tracing::info!(
        "Cron job '{}' — using cron session {}",
        job.name,
        session_id
    );

    // Swap to cron-specific provider if configured
    if let Some(ref provider_name) = effective_provider {
        match crate::brain::provider::create_provider_by_name(&config, provider_name).await {
            Ok(provider) => {
                tracing::info!(
                    "Cron job '{}' — using provider '{}'",
                    job.name,
                    provider_name
                );
                agent.swap_provider_for_session(
                    cron_session_id,
                    provider.clone(),
                    provider.default_model().to_string(),
                );
            }
            Err(e) => {
                tracing::warn!(
                    "Cron job '{}' — failed to create provider '{}': {e}, using system default",
                    job.name,
                    provider_name
                );
            }
        }
    }

    // Execute with auto-approved tools (no interactive user)
    let result = agent
        .send_message_with_tools_and_callback(
            session_id,
            job.prompt.clone(),
            effective_model,
            None, // no cancel token
            Some(Arc::new(|_| {
                // Auto-approve all tools for cron jobs
                Box::pin(async { Ok((true, false)) })
            })),
            None, // no progress callback
            None, // no follow_up_question surface for cron
            "cron",
            None,
        )
        .await;

    match result {
        Ok(response) => {
            let clean = crate::utils::sanitize::strip_llm_artifacts(&response.content);

            tracing::info!(
                "Cron job '{}' completed — {} tokens, ${:.6}",
                job.name,
                response.usage.input_tokens + response.usage.output_tokens,
                response.cost
            );

            // Save result to DB
            if let Err(e) = run_repo
                .complete_success(
                    &run_id,
                    &clean,
                    response.usage.input_tokens as i64,
                    response.usage.output_tokens as i64,
                    response.cost,
                )
                .await
            {
                tracing::error!("Failed to save cron run result to DB: {e}");
            }

            // Optionally deliver to configured channels too
            if let Some(ref deliver_to) = job.deliver_to {
                for target in deliver_to
                    .split(',')
                    .map(str::trim)
                    .filter(|s| !s.is_empty())
                {
                    deliver_result(target, &job.name, &clean, job.deliver_api_key.as_deref()).await;
                }
            }
        }
        Err(e) => {
            tracing::error!("Cron job '{}' agent error: {e}", job.name);

            // Save error to DB
            let error_msg = format!("{e}");
            if let Err(db_err) = run_repo.complete_error(&run_id, &error_msg).await {
                tracing::error!("Failed to save cron run error to DB: {db_err}");
            }

            // Optionally deliver error to configured channels too
            if let Some(ref deliver_to) = job.deliver_to {
                let msg = format!("Cron job '{}' failed: {e}", job.name);
                for target in deliver_to
                    .split(',')
                    .map(str::trim)
                    .filter(|s| !s.is_empty())
                {
                    deliver_result(target, &job.name, &msg, job.deliver_api_key.as_deref()).await;
                }
            }
        }
    }

    Ok(())
}

/// Deliver a cron job result to the specified channel.
/// Format: "telegram:chat_id", "discord:channel_id", "slack:channel_id",
/// or an HTTP(S) URL for generic webhook delivery.
async fn deliver_result(deliver_to: &str, job_name: &str, content: &str, api_key: Option<&str>) {
    // HTTP(S) URL — generic webhook delivery
    if deliver_to.starts_with("http://") || deliver_to.starts_with("https://") {
        deliver_http(deliver_to, job_name, content, api_key).await;
        return;
    }

    let parts: Vec<&str> = deliver_to.splitn(2, ':').collect();
    if parts.len() != 2 {
        tracing::warn!(
            "Invalid deliver_to format '{}' for job '{}' — expected 'channel:id' or HTTP URL",
            deliver_to,
            job_name
        );
        return;
    }

    let (channel, target_id) = (parts[0], parts[1]);

    // Truncate content for delivery (channels have message limits)
    let max_len = 4000;
    let msg = if content.len() > max_len {
        format!(
            "{}...\n\n(truncated — full output in session)",
            &content[..max_len]
        )
    } else {
        content.to_string()
    };

    let delivery_msg = format!("⏰ **Cron: {job_name}**\n\n{msg}");

    match channel {
        "telegram" => {
            #[cfg(feature = "telegram")]
            {
                tracing::info!("Delivering cron result to Telegram chat {target_id}");
                deliver_telegram(target_id, &delivery_msg).await;
            }
            #[cfg(not(feature = "telegram"))]
            {
                tracing::warn!("Telegram feature not enabled — cannot deliver cron result");
            }
        }
        "discord" => {
            tracing::info!("Delivering cron result to Discord channel {target_id}");
            tracing::warn!("Discord cron delivery not yet wired — result logged only");
        }
        "slack" => {
            tracing::info!("Delivering cron result to Slack channel {target_id}");
            tracing::warn!("Slack cron delivery not yet wired — result logged only");
        }
        other => {
            tracing::warn!("Unknown delivery channel '{other}' for job '{job_name}'");
        }
    }
}

/// Deliver cron result via HTTP POST to a generic webhook URL.
async fn deliver_http(url: &str, job_name: &str, content: &str, api_key: Option<&str>) {
    let client = reqwest::Client::new();
    let body = serde_json::json!({
        "job_name": job_name,
        "content": content,
        "timestamp": chrono::Utc::now().to_rfc3339(),
    });

    let mut request = client.post(url).json(&body);

    // Attach Bearer token if the job has one configured.
    if let Some(key) = api_key {
        request = request.header("Authorization", format!("Bearer {key}"));
    }

    match request.send().await {
        Ok(resp) if resp.status().is_success() => {
            tracing::info!("Cron result for '{job_name}' delivered to {url}");
        }
        Ok(resp) => {
            tracing::warn!(
                "HTTP delivery to {url} failed ({}): {:?}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            );
        }
        Err(e) => {
            tracing::error!("HTTP delivery to {url} error: {e}");
        }
    }
}

/// Deliver via Telegram Bot API (direct HTTP POST).
#[cfg(feature = "telegram")]
async fn deliver_telegram(chat_id: &str, message: &str) {
    // We need the bot token — read from config
    let brain_path = crate::brain::BrainLoader::resolve_path();
    let keys_path = brain_path.join("keys.toml");
    let token = if let Ok(content) = std::fs::read_to_string(&keys_path) {
        content.parse::<toml::Table>().ok().and_then(|t| {
            t.get("channels")?
                .as_table()?
                .get("telegram")?
                .as_table()?
                .get("token")?
                .as_str()
                .map(String::from)
        })
    } else {
        None
    };

    let Some(token) = token else {
        tracing::warn!("No Telegram bot token found in keys.toml — cannot deliver cron result");
        return;
    };

    let url = format!("https://api.telegram.org/bot{}/sendMessage", token);

    let client = reqwest::Client::new();
    let body = serde_json::json!({
        "chat_id": chat_id,
        "text": message,
        "parse_mode": "Markdown"
    });

    match client.post(&url).json(&body).send().await {
        Ok(resp) if resp.status().is_success() => {
            tracing::info!("Cron result delivered to Telegram chat {chat_id}");
        }
        Ok(resp) => {
            tracing::warn!(
                "Telegram delivery failed ({}): {:?}",
                resp.status(),
                resp.text().await.unwrap_or_default()
            );
        }
        Err(e) => {
            tracing::error!("Telegram delivery HTTP error: {e}");
        }
    }
}