diff --git a/config.toml.example b/config.toml.example index e5cf7df3d..c3d1df7b2 100644 --- a/config.toml.example +++ b/config.toml.example @@ -334,12 +334,12 @@ error_hold_ms = 2500 # [[cron.jobs]] # enabled = true # optional, default: true # schedule = "0 9 * * 1-5" # weekdays at 9:00 AM -# channel = "123456789" # target channel/thread ID +# channel = "123456789012345678" # target channel/thread ID # message = "summarize yesterday's merged PRs" # prompt for the agent # platform = "discord" # "discord" or "slack" # sender_name = "DailyOps" # attribution (default: "openab-cron") # timezone = "America/New_York" # IANA timezone (default: "UTC") -# thread_id = "" # optional: post to existing thread +# thread_id = "234567890123456789" # optional: post to existing thread (omit for a new thread) # [[cron.jobs]] # schedule = "0 0 * * 0" diff --git a/crates/openab-core/src/cron.rs b/crates/openab-core/src/cron.rs index 40e9a67ca..ea6bec58a 100644 --- a/crates/openab-core/src/cron.rs +++ b/crates/openab-core/src/cron.rs @@ -327,8 +327,23 @@ pub fn load_usercron_file(path: &Path, configured_platforms: &[&str]) -> Vec = HashSet::new(); + parsed.jobs.into_iter().enumerate().filter(move |(i, job)| { + if let Some(id) = non_empty_opt(job.id.as_deref()) { + if !seen_ids.insert(id.to_owned()) { + warn!(index = i, id, "usercron: duplicate id (compared after trimming), skipping"); + return false; + } + } if let Err(e) = parse_cron_expr(&job.schedule) { warn!(index = i, schedule = %job.schedule, error = %e, "usercron: invalid cron expression, skipping"); return false; @@ -411,12 +426,26 @@ fn parse_job_list( Some(ParsedJob { schedule, tz, - config: job.clone(), + config: normalize_job_config(job), usercron_path: usercron_path.map(Path::to_path_buf), }) }).collect() } +/// Canonicalize optional identifiers so blank values behave like omitted ones. +/// +/// `thread_id = ""` would otherwise deserialize to `Some("")`, skip thread +/// creation, and fail the send. Both fields also have surrounding whitespace +/// trimmed: a padded `thread_id` never matches a real platform thread ID, and +/// trimming `id` keeps writeback lookups consistent with `update_usercron_job`, +/// which also compares trimmed values. +fn normalize_job_config(job: &CronJobConfig) -> CronJobConfig { + let mut config = job.clone(); + config.thread_id = non_empty_opt(job.thread_id.as_deref()).map(str::to_owned); + config.id = non_empty_opt(job.id.as_deref()).map(str::to_owned); + config +} + /// Run the internal cron scheduler. Evaluates cron expressions once per minute. /// `usercron_path` enables hot-reload of an external cronjob.toml file. pub async fn run_scheduler( @@ -898,7 +927,14 @@ fn update_usercron_job( let mut found = false; for table in jobs.iter_mut() { - if table.get("id").and_then(|item| item.as_str()) != Some(id) { + // Trim the file side too: callers pass a trimmed id, so a padded + // `id = " x "` would otherwise never match and writeback would fail. + if table + .get("id") + .and_then(|item| item.as_str()) + .map(str::trim) + != Some(id.trim()) + { continue; } if let Some(enabled) = enabled { @@ -1511,6 +1547,218 @@ message = "a" assert!(err.to_string().contains("not found")); } + /// Parse one job through `parse_job_list` and return the stored config. + fn parsed_config(job: CronJobConfig, usercron_path: Option<&Path>) -> CronJobConfig { + let parsed = parse_job_list(&[job], "test", usercron_path); + assert_eq!(parsed.len(), 1); + parsed[0].config.clone() + } + + #[test] + fn parse_job_list_normalizes_thread_id_on_baseline_and_usercron() { + let usercron = Path::new("/tmp/cronjob.toml"); + let cases: [(Option<&str>, Option<&str>); 5] = [ + (None, None), + (Some(""), None), + (Some(" "), None), + (Some(" 1556488188002959532 "), Some("1556488188002959532")), + (Some("1556488188002959532"), Some("1556488188002959532")), + ]; + for path in [None, Some(usercron)] { + for (input, expected) in cases { + let mut job = test_cron_job(); + job.thread_id = input.map(str::to_owned); + let config = parsed_config(job, path); + assert_eq!( + config.thread_id.as_deref(), + expected, + "thread_id {input:?} (usercron_path {path:?})" + ); + // Blank thread_id must take the create-thread path, not send to "". + assert_eq!(should_create_cron_thread(&config), expected.is_none()); + } + } + } + + #[test] + fn parse_job_list_trims_and_drops_blank_id() { + let cases: [(Option<&str>, Option<&str>); 4] = [ + (None, None), + (Some(""), None), + (Some(" "), None), + (Some(" daily-news "), Some("daily-news")), + ]; + for (input, expected) in cases { + let mut job = test_cron_job(); + job.id = input.map(str::to_owned); + assert_eq!( + parsed_config(job, None).id.as_deref(), + expected, + "id {input:?}" + ); + } + } + + #[test] + fn update_usercron_job_matches_padded_id_in_file() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cronjob.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = " padded " +schedule = "* * * * *" +channel = "123" +message = "a" +"#, + ) + .unwrap(); + update_usercron_job(&path, "padded", None, Some("thread-789")).unwrap(); + let doc = std::fs::read_to_string(&path) + .unwrap() + .parse::() + .unwrap(); + let job = doc["jobs"].as_array_of_tables().unwrap().get(0).unwrap(); + assert_eq!(job["thread_id"].as_str(), Some("thread-789")); + } + + #[test] + fn load_usercron_file_skips_ids_that_collide_after_trimming() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cronjob.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = "x" +schedule = "* * * * *" +channel = "123" +message = "first" + +[[jobs]] +id = " x " +schedule = "* * * * *" +channel = "123" +message = "second" +"#, + ) + .unwrap(); + // Both ids trim to "x" and would share one writeback target, so only the + // first survives; the padded duplicate is skipped. + let jobs = load_usercron_file(&path, &["discord"]); + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].message, "first"); + } + + #[test] + fn load_usercron_file_skips_exact_duplicate_ids() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("cronjob.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = "x" +schedule = "* * * * *" +channel = "123" +message = "first" + +[[jobs]] +id = "x" +schedule = "* * * * *" +channel = "123" +message = "second" +"#, + ) + .unwrap(); + // Exact duplicates also share one writeback target (behavior change from + // main, where both used to load and run): only the first survives. + let jobs = load_usercron_file(&path, &["discord"]); + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].message, "first"); + } + + #[test] + fn load_usercron_file_first_entry_owns_id_even_when_invalid() { + // The first file-order entry with a given trimmed id owns that id, even + // if it is itself skipped. A later valid same-id entry must not load, + // because its writeback would land on the first raw entry. + let dir = tempfile::tempdir().unwrap(); + let configured = &["discord"]; + + // (a) first entry has an invalid cron expression + let path = dir.path().join("invalid_cron.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = "x" +schedule = "not a cron" +channel = "123" +message = "first" + +[[jobs]] +id = " x " +schedule = "* * * * *" +channel = "123" +message = "second" +"#, + ) + .unwrap(); + assert!(load_usercron_file(&path, configured).is_empty()); + + // (b) first entry has an invalid timezone + let path = dir.path().join("invalid_tz.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = "x" +schedule = "* * * * *" +timezone = "Not/AZone" +channel = "123" +message = "first" + +[[jobs]] +id = " x " +schedule = "* * * * *" +channel = "123" +message = "second" +"#, + ) + .unwrap(); + assert!(load_usercron_file(&path, configured).is_empty()); + + // (c) first entry is disabled + let path = dir.path().join("disabled.toml"); + std::fs::write( + &path, + r#" +[[jobs]] +id = "x" +schedule = "* * * * *" +enabled = false +channel = "123" +message = "first" + +[[jobs]] +id = " x " +schedule = "* * * * *" +channel = "123" +message = "second" +"#, + ) + .unwrap(); + // `enabled = false` entries are kept by the loader (parse_job_list drops + // them later), but they still claim the id, so the later entry is + // skipped and only the disabled first entry remains. + let jobs = load_usercron_file(&path, configured); + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].message, "first"); + assert!(!jobs[0].enabled); + } + #[tokio::test] async fn disable_on_success_requires_exit_zero_and_marker() { let mut job = test_cron_job(); diff --git a/docs/config-reference.md b/docs/config-reference.md index 838ceacb6..0d3362076 100644 --- a/docs/config-reference.md +++ b/docs/config-reference.md @@ -785,16 +785,16 @@ usercron_path = "cronjob.toml" # relative to $HOME/.openab/, or ab [[cron.jobs]] enabled = true # optional, default: true schedule = "0 9 * * 1-5" # cron expression (5-field POSIX) -channel = "123456789" # target channel/thread ID +channel = "123456789012345678" # target channel/thread ID message = "summarize yesterday's merged PRs" # message sent to agent platform = "discord" # optional, default: "discord" sender_name = "DailyOps" # optional, default: "openab-cron" timezone = "America/New_York" # optional, default: "UTC" -thread_id = "" # optional, post to existing thread +# thread_id = "234567890123456789" # optional, post to existing thread (omit for a new thread per run) [[cron.jobs]] schedule = "0 0 * * 0" -channel = "123456789" +channel = "123456789012345678" message = "generate weekly status report" platform = "discord" timezone = "UTC" @@ -818,17 +818,17 @@ timezone = "UTC" | `platform` | string | `"discord"` | Target platform (`"discord"` or `"slack"`). | | `sender_name` | string | `"openab-cron"` | Sender attribution shown in the prompt context. | | `timezone` | string | `"UTC"` | IANA timezone for schedule evaluation (e.g. `"America/New_York"`, `"Europe/Berlin"`). | -| `thread_id` | string | `""` | Optional thread ID to post into an existing thread. | +| `thread_id` | string | — (unset) | Optional thread ID to post into an existing thread. Omit the field for a new thread per run, unless a usercron `id` pins the job to its first thread. See [Thread Behavior](cronjob.md#thread-behavior). | The external `cronjob.toml` uses `[[jobs]]` (same fields). See [Usercron docs](cronjob.md#usercron--hot-reload-with-cronjobtoml) for details. ### Usercron-only `[[jobs]]` fields -These fields are valid only in the external usercron file, for example `$HOME/.openab/cronjob.toml`. They are rejected in baseline `[[cron.jobs]]` because OpenAB only writes state back to the user-managed cron file. +These fields only take effect in the external usercron file, for example `$HOME/.openab/cronjob.toml`. `disable_on_success` is rejected in baseline `[[cron.jobs]]`; the others, including `id`, are parsed but ignored there, because OpenAB only writes state back to the user-managed cron file. | Key | Type | Default | Description | |-----|------|---------|-------------| -| `id` | string | *required with `disable_on_success`* | Stable job ID used when the scheduler writes `enabled = false` or `thread_id` back to `cronjob.toml`. | +| `id` | string | *required with `disable_on_success`* | Unique job ID that enables scheduler writeback of `enabled`/`thread_id` to `cronjob.toml`. See [Thread Behavior](cronjob.md#thread-behavior). | | `disable_on_success` | string | — | Command to run before sending the scheduled prompt. | | `disable_on_success_match` | string | *required with `disable_on_success`* | Marker that must appear in stdout or stderr, in addition to exit code `0`, before the job is considered complete. | | `disable_on_success_timeout_secs` | integer | `60` | Timeout for the completion check command. | diff --git a/docs/cronjob.md b/docs/cronjob.md index 653d52e76..0a44c10d7 100644 --- a/docs/cronjob.md +++ b/docs/cronjob.md @@ -37,7 +37,7 @@ message = "summarize yesterday's merged PRs" # required: prompt for the agent platform = "discord" # optional, default: "discord" sender_name = "DailyOps" # optional, default: "openab-cron" timezone = "America/New_York" # optional, default: "UTC" -thread_id = "" # optional: post to existing thread +# thread_id = "234567890123456789" # optional: post to existing thread (omit to create a new thread) ``` | Field | Required | Default | Description | @@ -49,7 +49,34 @@ thread_id = "" # optional: post to existing thread | `platform` | | `"discord"` | `"discord"`, `"slack"`, `"telegram"`, `"googlechat"`, or `"lineworks"` (non-default platforms require their feature) | | `sender_name` | | `"openab-cron"` | Attribution shown in prompt context | | `timezone` | | `"UTC"` | IANA timezone (e.g. `"America/New_York"`, `"Europe/Berlin"`) | -| `thread_id` | | — | Post into an existing thread instead of the channel | +| `thread_id` | | — | Post into an existing thread instead of creating a new one. Omit the field to create a new thread per run, unless a usercron `id` pins the job to its first thread. See [Thread Behavior](#thread-behavior) | +| `id` | | — | Usercron only (ignored in baseline `[[cron.jobs]]`); a unique job identifier that enables scheduler writeback and is required for `disable_on_success`. See [Thread Behavior](#thread-behavior) | + +### Thread Behavior + +Where a job posts depends on `thread_id` and, for usercron jobs, `id`. This applies to platforms with thread support; Google Chat and LINE WORKS differ (see [Platform Prerequisites](#platform-prerequisites)). + +| `thread_id` | `id` | Behavior | +|---|---|---| +| set | any | Every run posts into that thread. No new thread and no `thread_id` writeback. | +| omitted or blank | omitted | Every run creates a new thread. Since sessions are keyed by thread, each run also starts a fresh agent session. | +| omitted or blank | set (usercron) | The first run creates a thread and writes its ID back as `thread_id`; later runs post into that thread. | + +A blank value (`""` or whitespace) is treated the same as omitting the field, for both `thread_id` and `id`. Surrounding whitespace is trimmed from both fields, so `thread_id = " 123 "` is stored and matched as `"123"`. + +`id` does not select a thread — it only tells the scheduler which entry to update. If you remove a written-back `thread_id` but keep `id`, the next run creates a new thread and pins the job to it again. + +> [!CAUTION] +> **Breaking Change** — Duplicate usercron ids now skip. On earlier versions, two `[[jobs]]` entries with the same `id` both loaded and ran; as of this version only the first (in file order) runs and the rest are skipped at load. A job with an accidental copy-paste duplicate id stops firing after upgrading. +> Before upgrading, check `cronjob.toml` for ids that are equal after trimming (including values that differ only by surrounding whitespace) and make them unique. + +Writeback edge cases for the `id` row: + +- A blank `id` counts as unset. Ids are compared after trimming, so any duplicates — including identical values and ids that differ only by surrounding whitespace — share one writeback target. The scheduler keeps the first entry in file order and skips the rest at load (`usercron: duplicate id (compared after trimming), skipping`). The first entry owns the id even when it is itself invalid or disabled, so a later entry with the same id still does not run (see the Breaking Change note). +- The pinned thread takes effect once the scheduler reloads the file (next tick, normally within 60s). +- Writeback re-resolves the first entry by trimmed `id` at write time, not against the snapshot the job was loaded from. `usercron_write_lock` serializes OpenAB's own writes but not external edits. Editing `cronjob.toml` while a job is in flight — inserting a same-id entry above it, reordering entries, or reusing an id for another job — can redirect the writeback to a different entry: `thread_id` pins the wrong entry (the job that ran keeps creating a new thread each run), or `enabled = false` disables an unrelated job. Avoid reusing or moving ids while jobs are running. +- If OpenAB cannot write `cronjob.toml` (see [Choosing the right scope](#choosing-the-right-scope)), nothing is pinned and every run creates a new thread. The logs show `failed to persist usercron thread_id`. This also means `disable_on_success` cannot write `enabled = false` back, so the goal command keeps running on every match — avoid both patterns when the file is unwritable. +- **Known limitation** — On gateway platforms (Telegram), if topic creation fails or times out (5s), the job posts to the chat itself and the **chat ID** is written back as `thread_id`. Because Telegram chat IDs and topic IDs are both numeric, a written-back value cannot be distinguished from a correct topic ID. Later runs then target that ID as a topic. Remove the written-back `thread_id` line to retry topic creation. This is a documented workaround, not the final behavior; the root-cause fix (gating the writeback on a thread actually being created) is a follow-up. ## Cron Expression Format @@ -366,7 +393,7 @@ Prompts and `disable_on_success` live in the same file, so write access cannot b `usercron_path` may be absolute, so protect the file and directory it actually points to. Agent CLI path rules alone are not enough, because an agent with a shell tool can bypass them. - OpenAB rewrites the file through a temporary file in the same directory, so scheduler writeback fails whenever OpenAB itself cannot write the file's directory. In that case, do not use `disable_on_success`: an achieved goal cannot be written back as `enabled = false`, so the command keeps running and `✅ Goal achieved` is posted on every schedule match. Also avoid jobs that set `id` but leave `thread_id` unset: the new thread's ID cannot be saved, so a new thread is created on every run. Instead, set a real `thread_id`, or omit `id` if you want a new thread per run. + OpenAB rewrites the file through a temporary file in the same directory, so scheduler writeback fails whenever OpenAB itself cannot write the file's directory. In that case, do not use `disable_on_success`, and avoid jobs that set `id` but leave `thread_id` unset; see the [Thread Behavior](#thread-behavior) writeback edge cases for the resulting symptoms. For an audit trail, record changes somewhere the agent cannot write: push commits to a protected remote, or persist file-monitor events (e.g. from `inotifywait`) to an external or append-only store. A local git repository is not enough, because an agent with a shell tool can rewrite its history. This is detective, not preventive: changes are hot-reloaded and take effect within about a minute. @@ -375,7 +402,7 @@ For an audit trail, record changes somewhere the agent cannot write: push commit - **Minute-aligned**: The scheduler aligns to minute boundaries (`:00`), so `0 9 * * *` fires at exactly 09:00:00, not at whatever second the process started. - **Overlap protection**: If a previous execution of the same job is still running, the next tick is skipped. - **Isolation**: Cron failures are logged but never block interactive chat traffic. -- **Usercron persistence**: For usercron jobs, the scheduler may write `thread_id` and `enabled = false` back to `cronjob.toml`. +- **Usercron persistence**: For usercron jobs with an `id`, the scheduler may write `thread_id` and `enabled = false` back to `cronjob.toml` (see [Thread Behavior](#thread-behavior)). - **Graceful shutdown**: In-flight cron tasks are waited on (up to 30 seconds) during shutdown. ## Sender Identity @@ -438,4 +465,7 @@ See [Kubernetes CronJob Reference Architecture](cronjob_k8s_refarch.md) for the | Channel not found | Bot not in channel | Invite the bot to the target channel | | Usercron not reloading | File not saved / wrong path | Check logs for `usercron file changed, reloading` | | Usercron parse error | Invalid TOML syntax | Check logs for `failed to parse usercron file` | +| Usercron job stopped running (e.g. after upgrade) | Another entry earlier in the file has the same `id` (compared after trimming), so this one is skipped | Check logs for `usercron: duplicate id (compared after trimming), skipping`; make the ids unique | | Goal job does not auto-disable | Command did not exit `0` or output did not include `disable_on_success_match` | Run the command manually and confirm both conditions | +| Usercron job with `id` creates a new thread on every run | `thread_id` writeback failed (e.g. agent or OpenAB cannot write the file) | Check logs for `failed to persist usercron thread_id`; see [Thread Behavior](#thread-behavior) | +| Telegram job with `id`: written-back `thread_id` equals the chat ID | Topic creation failed or timed out on the first run, so the chat ID was persisted | Check logs for `create_topic failed` / `create_topic timeout`; remove the `thread_id` line |