From 443df6991f33ade20c7cd498bccec6d3950350d2 Mon Sep 17 00:00:00 2001 From: Owain Lewis Date: Thu, 30 Jul 2026 15:42:33 +0100 Subject: [PATCH] feat(jobs): require schedule activation review --- ARCHITECTURE.md | 45 +- assistant/skills/push/SKILL.md | 10 +- docs/core-system/design.md | 2 +- docs/index.md | 4 +- docs/jobs.md | 53 +- docs/jobs/design.md | 11 +- docs/reference/cli.md | 35 +- docs/security.md | 29 +- docs/services.md | 16 +- src/approval.rs | 35 +- src/assistant.rs | 4 +- src/audit.rs | 121 +- src/cli_json.rs | 63 +- src/doctor.rs | 1 + src/gateway/mod.rs | 129 ++- src/gateway/tests.rs | 227 ++++ src/gateway/worker.rs | 68 +- src/history.rs | 130 ++- src/jobs.rs | 1996 +++++++++++++++++++++++++++++++- src/main.rs | 52 +- src/prompt.rs | 5 +- src/util.rs | 24 + tests/init_cli.rs | 4 +- tests/json_cli.rs | 80 ++ 24 files changed, 3030 insertions(+), 114 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 7349cac..ce4e9c8 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -596,6 +596,7 @@ timeout. | `messages` | canonical inbound and outbound content, generation state, delivery state, chunk checkpoint | | `approval_questions` | retained durable question and answer state | | `job_runs` | immutable job claims, bounded results, evaluation, scheduling, and delivery | +| `job_schedule_reviews` and `job_schedule_events` | exact-revision schedule proposals, decisions, activation, and history | | `gateway_control_actions` | idempotent control actions such as `/stop` targets | | `channel_cursors` | monotonic per-channel polling checkpoints | | `backend_sessions` | current backend session for each channel and thread | @@ -609,9 +610,12 @@ Important constraints: - scheduled occurrence identity is unique; - control actions are idempotent by inbound message. -The approval tables and inbound answer path remain durable, but current -production job creation writes runbooks directly and does not create approval -questions. +Job files are authored directly. Enabled schedule activation uses the durable +question path and a separate activation ledger. A question is bound to one +allowlisted channel identity and the validated content revision, file identity, +effective backend, timeout, work directory, enabled triggers, and delivery +target. Answer selection and the schedule decision are recoverable across a +crash, and the scheduler still revalidates the exact revision before planning. ### Slack Inbox @@ -627,6 +631,12 @@ Events include message metadata, routing, backend starts and failures, answer outcomes, delivery results, and completion. Content is omitted by default. `audit_log_content = true` opts into message and reply text. +Schedule lifecycle events first enter `job_schedule_events` as a SQLite +outbox. The gateway syncs each JSONL append before acknowledging it in SQLite +and retries pending rows on startup and scheduler or conversation activity. +Replay is at least once, so a crash between append and acknowledgement may +duplicate the stable schedule `event_id`. + --- ## 8. Job System @@ -683,20 +693,33 @@ The gateway ticks the scheduler once per second: ```text 1. RECOVER inspect stale runs and delivery claims 2. CATALOG reload and validate installed jobs -3. PLAN calculate next cron occurrence in its IANA timezone -4. ENQUEUE record one due occurrence in push.db -5. CLAIM take work up to jobs_max_workers -6. EXECUTE run a fresh unattended backend session -7. EVALUATE optionally run the restricted evaluator -8. STORE commit result, error, and evaluation state -9. CLAIM SEND take due delivery work across gateway processes -10. DELIVER send stored chunks and checkpoint progress +3. REVIEW reconcile the exact revision and require durable activation +4. PLAN calculate next cron occurrence in its IANA timezone +5. ENQUEUE record one due occurrence in push.db +6. CLAIM take work up to jobs_max_workers +7. EXECUTE run a fresh unattended backend session +8. EVALUATE optionally run the restricted evaluator +9. STORE commit result, error, and evaluation state +10. CLAIM SEND take due delivery work across gateway processes +11. DELIVER send stored chunks and checkpoint progress ``` Push does not catch up occurrences missed while offline. A clock jump queues at most one occurrence, daylight-saving gaps are skipped, and repeated local times run once at their first instant. +Direct Markdown authoring, validation, inspection, disabled triggers, and +manual runs do not require schedule activation. A new or changed enabled +revision is proposed but omitted from planning until the exact owner-bound +review is approved. Any later validation failure, content change, path or +symlink replacement, or change to effective execution or delivery settings +invalidates it. A version-11 database migration activates only valid enabled +schedules whose exact revisions are captured by the first config-aware open +after upgrade, so upgrades preserve existing intended recurrence without +creating a later grandfathering window. When that capture has no valid primary +destination, the migration records an empty baseline and closes without +activating schedules. + ### Execution and Delivery Semantics - A failed or timed-out job is not rerun because the agent may already have diff --git a/assistant/skills/push/SKILL.md b/assistant/skills/push/SKILL.md index 796cb64..b9b37b7 100644 --- a/assistant/skills/push/SKILL.md +++ b/assistant/skills/push/SKILL.md @@ -4,7 +4,7 @@ description: Operate a Push personal assistant, inspect its health and jobs, and license: MIT compatibility: Requires the Push CLI and an initialized assistant repository. metadata: - push-managed-version: "1" + push-managed-version: "2" --- # Push @@ -43,6 +43,7 @@ are: - `push job show ` - `push job run ` - `push job runs []` +- `push job reviews []` All commands accept `--config `. Do not assume machine-readable output unless `push help` documents it in the installed version. Never expose tokens, @@ -57,9 +58,12 @@ message content, or sensitive runtime state in diagnostics or replies. environment for credentials. 4. Run `push job validate` after every job change. Do not claim success if validation fails. -5. Use `push job show ` to inspect the installed result and +5. Saving a new or changed enabled schedule does not activate it. Tell the user + that Push will present the exact revision for separate owner review. +6. Use `push job show ` to inspect the installed result, + `push job reviews ` to inspect schedule activation state, and `push job runs ` to inspect execution and delivery history. -6. Run `push job run ` only when the user asked for the job to execute or +7. Run `push job run ` only when the user asked for the job to execute or when execution is a clearly authorized part of the task. ## Reply normally diff --git a/docs/core-system/design.md b/docs/core-system/design.md index dc6a6c7..7e49451 100644 --- a/docs/core-system/design.md +++ b/docs/core-system/design.md @@ -90,7 +90,7 @@ Jobs: /resolved/path/to/assistant/jobs Begin with context/README.md when user context is relevant. Do not modify SOUL.md or evals unless the user asks. -When the user asks to create or change a job, write the complete runbook directly under Jobs and run `push job validate` before saying it succeeded. +When the user asks to create or change a job, write the complete runbook directly under Jobs and run `push job validate` before saying it was saved. A new or changed enabled schedule remains inactive until Push presents its exact revision for owner review. ``` Claude Code and Pi receive the composed text as appended system instructions. diff --git a/docs/index.md b/docs/index.md index b489172..8fc3ea8 100644 --- a/docs/index.md +++ b/docs/index.md @@ -46,8 +46,8 @@ sends the result back when the work is done.
Telegram Push is online
You · 18:12

Every weekday at 8am, run my morning brief and send me the three things that need my attention.

-
Push → CodexDraft ready for approval
-
Push · 18:14

I drafted your morning brief for weekdays at 8am. Approve it to start the schedule.

+
Push → CodexSchedule ready for review
+
Push · 18:14

I saved your morning brief for weekdays at 8am. Approve this exact revision to start the schedule.

Delivered in chatConversation saved
diff --git a/docs/jobs.md b/docs/jobs.md index dfb6d6a..80e5bd9 100644 --- a/docs/jobs.md +++ b/docs/jobs.md @@ -58,7 +58,8 @@ push job show repo-review ``` Validation reports every valid and invalid file. An invalid job is disabled -individually and does not stop messaging or other valid jobs. +individually and does not stop messaging or other valid jobs. Validation does +not activate an enabled schedule. ## Evaluate completed work @@ -140,6 +141,30 @@ Scheduling starts only when the primary destination is enabled and allowlisted. A missing or invalid destination disables new scheduled starts without affecting conversations or manual jobs. +Saving `enabled = true` creates a schedule activation proposal. The Markdown +file remains available for validation, inspection, and `push job run`, but the +scheduler does not plan the enabled trigger until the proposal is approved. +Push assigns each proposal to one allowlisted conversation when it presents a +durable question showing the exact job name, content revision, enabled cron +schedules, effective backend, timeout, work directory, and primary delivery +target. That persisted identity is the review owner; concurrent conversations +cannot adopt or answer its question. Reply with the question UUID followed by +the number for Approve or Reject. A number alone is rejected so a delayed reply +cannot select a replacement question. The question expires after 24 hours. + +Approval is bound to the exact channel, sender, chat, thread or topic, validated +file revision, file identity, effective execution settings, and delivery +target. A file edit, invalid file, symlink or path replacement, schedule change, +backend change, timeout change, work-directory change, or delivery-target +change invalidates the prior activation before it can run. A later valid +revision receives a new review. Disabled triggers and jobs without triggers do +not require activation review. + +Use `push job reviews []` to inspect current and historical schedule +review state. A manually edited schedule is still detected and kept inactive; +the next completed request from an allowlisted conversation can receive its +review question. + Push runs at most `jobs_max_workers` scheduled jobs concurrently. It does not catch up cron occurrences missed while offline. Daylight-saving gaps are skipped; repeated local times run once at their first instant. Cron expressions @@ -189,20 +214,30 @@ delivery attempts, destination, bounded results, and error details. When a user asks for a job, the assistant writes the complete runbook directly to `/jobs/.md` and runs `push job validate`. -There is no separate draft or approval step. The selected agent's filesystem -permissions control whether it can change the assistant repository. +There is no separate draft-file or installation approval step. The selected +agent's filesystem permissions control whether it can change the assistant +repository. A new or changed enabled schedule remains inactive until its +separate activation review succeeds. For an assistant repository created before this change, replace any `AGENTS.md` instruction that says to propose jobs through approval with the direct-write rule above. The gateway's runtime instruction overrides that old rule, but updating the repository keeps its checked-in guidance accurate. -Pending job approvals from older Push versions are cancelled during database -migration. Replying to one explains that the job must be requested again. +Pending draft-install approvals from older Push versions are cancelled during +database migration. Replying to one explains that the job must be requested +again. On upgrade to schedule activation review, each valid enabled schedule +whose exact revision exists at the first config-aware Push command is captured +as the migration baseline. Those revisions are recorded as approved and +activated once when a valid primary destination exists. If that first command +has no valid primary destination, Push records an empty baseline and closes the +migration without grandfathering any schedules. Disabled and invalid jobs are +not grandfathered. !!! warning - Jobs have no interactive approval path. Push runs Codex jobs with full - access and no prompts and Claude jobs in `bypassPermissions` mode. Treat - every enabled job as code execution by the Push service user, review - changes to the assistant repository, and allow only trusted senders. + Schedule activation review is not an agent permission prompt. After + activation, Push runs Codex jobs with full access and no prompts and Claude + jobs in `bypassPermissions` mode. Treat every activated job as code + execution by the Push service user, review changes to the assistant + repository, and allow only trusted senders. diff --git a/docs/jobs/design.md b/docs/jobs/design.md index 0edb942..58f1532 100644 --- a/docs/jobs/design.md +++ b/docs/jobs/design.md @@ -256,10 +256,13 @@ execution: - `push job run ` - `push job runs []` -Push is the only writer to the run ledger. The CLI owns the manual run it -claims, and the gateway owns scheduled runs. Agents may write requested job -files directly under `/jobs` when their filesystem permissions -allow it, then validate the catalog with `push job validate`. +Push is the only writer to the run and schedule-activation ledgers. The CLI +owns the manual run it claims, and the gateway owns scheduled runs. Agents may +write requested job files directly under `/jobs` when their +filesystem permissions allow it, then validate the catalog with +`push job validate`. Direct authoring does not activate a new or changed enabled +schedule. The gateway binds owner review to the exact validated revision and +effective execution and delivery settings before planning it. ## Alternatives and tradeoffs diff --git a/docs/reference/cli.md b/docs/reference/cli.md index dc90625..709ba64 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -21,6 +21,7 @@ root. | `push job show ` | Print the parsed installed job | | `push job run ` | Claim and run one job in the CLI process | | `push job runs []` | Print run and delivery history, optionally for one job | +| `push job reviews []` | Print schedule activation review state and exact revision metadata | Examples: @@ -36,6 +37,7 @@ push reload push job validate push job run repo-review push job runs repo-review +push job reviews repo-review ``` Unknown commands and missing values fail with the accepted command forms. The @@ -43,6 +45,13 @@ CLI does not currently provide shell completion or separate help pages for subcommands. A `--help` flag anywhere in the argument list prints the global help shown by `push --help`. +`push job validate`, `list`, and `show` never activate a schedule. Use +`push job reviews []` to inspect proposed, approved, rejected, +invalidated, and activated revisions, including their schedules, effective +backend, timeout, work directory, and delivery target. Activation decisions +are made by replying to the durable review question from the exact persisted +allowlisted channel identity that received it. + `push reload` and its `push restart` alias target the service definitions documented by Push: `com.owainlewis.push` under launchd on macOS and the `push.service` user unit under systemd on Linux. The service definition controls its config path, @@ -72,7 +81,7 @@ available for: - `help` and `version` - `doctor`, `status`, and `paths` -- `job validate`, `job list`, `job show`, and `job runs` +- `job validate`, `job list`, `job show`, `job runs`, and `job reviews` Commands that start or mutate runtime state reject `--json`. This includes the gateway, `init`, `reload`, `restart`, and `job run`. In particular, Push does @@ -256,6 +265,30 @@ The run projection queries only `job_runs` from the shared SQLite database. It does not include co-located channel cursors, backend session IDs, conversation messages, stored job output, evaluation text, or error text. +`job reviews` data: + +| Field | Type | Values | +| --- | --- | --- | +| `job_name` | string or null | Requested job filter, or null for all jobs | +| `reviews` | array of review objects | Up to 100 newest stored schedule review revisions | +| `reviews[].review_id` | string | Exact activation fingerprint | +| `reviews[].job_name` | string | Job slug | +| `reviews[].status` | string enum | `proposed`, `approved`, `rejected`, `invalidated`, or `activated` | +| `reviews[].content_hash` | string | Authored Markdown SHA-256 | +| `reviews[].schedules` | array of trigger objects | Enabled triggers bound to the review | +| `reviews[].schedules[].id` | string | Trigger slug | +| `reviews[].schedules[].kind` | string constant | `cron` | +| `reviews[].schedules[].schedule` | string | Five-field cron expression | +| `reviews[].schedules[].timezone` | string | IANA timezone name | +| `reviews[].schedules[].enabled` | boolean | Always `true` for a reviewed trigger | +| `reviews[].backend` | string enum | Effective `claude`, `codex`, or `pi` backend | +| `reviews[].timeout_ms` | integer | Effective timeout in milliseconds | +| `reviews[].workdir` | string | Resolved backend working directory | +| `reviews[].delivery.channel` | string | Bound delivery channel | +| `reviews[].delivery.target` | string | Bound delivery target | +| `reviews[].reviewed_by` | string or null | Bound actor for a decided revision | +| `reviews[].reason` | string or null | Invalidation or migration reason | + Fields may be added compatibly within version 1. Existing fields, meanings, category names, and types will not change without a schema-version change. diff --git a/docs/security.md b/docs/security.md index d56b288..724e900 100644 --- a/docs/security.md +++ b/docs/security.md @@ -46,12 +46,14 @@ runs Claude jobs in `bypassPermissions` mode. Pi already has no native filesystem sandbox or interactive permission prompt. Evaluators remain read-only with tools disabled. -This makes job bodies equivalent to unattended code execution as the Push -service user. The assistant repository is the default work directory. An -explicit work directory must exist. Push rejects overlap with runtime state and -a loaded config stored outside the assistant repository. Keep allowed senders -and job definitions trusted, and run the service with only the OS permissions -its jobs require. +This makes activated job bodies equivalent to unattended code execution as the +Push service user. New and changed enabled schedules stay inactive until an +allowlisted owner approves the exact revision and effective execution settings. +The assistant repository is the default work directory. An explicit work +directory must exist. Push rejects overlap with runtime state and a loaded +config stored outside the assistant repository. Keep allowed senders and job +definitions trusted, and run the service with only the OS permissions its jobs +require. Do not place secrets in a job body. Make them available through the backend or service environment using the narrowest policy that works. @@ -96,9 +98,13 @@ This reduces exposure, but it does not make an allowed message harmless. Bounded `ask_user` questions are stored before delivery, survive restart, expire, and can be consumed once. Mismatched, duplicate, ambiguous, cancelled, -and expired answers do not reach an agent. Job creation does not use this -mechanism. The selected agent's filesystem permissions control access to jobs -in the assistant repository. +and expired answers do not reach an agent. Direct job-file creation does not +use this mechanism. Enabled schedule activation does: the review binds the +exact allowlisted channel identity to the content hash, file identity, +validated schedules, effective backend, timeout, work directory, and delivery +target. Revalidation and scheduler claims fail closed when those values no +longer match. Schedule reviews require the question UUID with the answer +number; uncorrelated numbers cannot approve a replacement question. ## Audit log @@ -107,6 +113,11 @@ include metadata such as row ID, channel, thread, backend, decision, target, error, and character counts. Message and reply content are omitted unless `audit_log_content = true`. +Schedule lifecycle events use a durable SQLite outbox. Push syncs each JSONL +event before marking it delivered and retries pending events after a write +failure or restart. A crash after append but before acknowledgement can produce +a duplicate with the same `event_id`; consumers should deduplicate that ID. + The redacted log is still sensitive because it can contain handles, thread IDs, file paths, and backend errors. Protect and rotate it like a service log. diff --git a/docs/services.md b/docs/services.md index 43bc37e..090b6ce 100644 --- a/docs/services.md +++ b/docs/services.md @@ -234,11 +234,23 @@ backup. As a last resort, move the unusable database aside and restart with the retained JSON to recover its older cursors and sessions, understanding that conversation, job, and delivery records not present in JSON will be absent. +New and changed enabled schedules are detected on each scheduler tick but stay +inactive until their exact validated revision is approved from the bound +allowlisted conversation. Review questions and accepted activations are stored +in `database_path`, so restart does not lose them. Use +`push job reviews` to inspect +proposed, rejected, invalidated, approved, and activated revisions. Editing or +replacing an activated job invalidates its schedule before the changed revision +can run. Schedule audit events also remain pending in the database until their +JSONL append is synced, then replay after an audit write failure or restart. + ## Agent-created jobs When asked, the agent writes jobs directly under `/jobs` and -runs `push job validate`. There is no approval step. The agent's configuration -decides whether it may write to the assistant repository. +runs `push job validate`. There is no draft installation step. The agent's +configuration decides whether it may write to the assistant repository. Saving +an enabled schedule and activating unattended recurrence are separate actions; +the latter requires durable owner review. ## Restart Behavior diff --git a/src/approval.rs b/src/approval.rs index e6c5c80..b71ecb4 100644 --- a/src/approval.rs +++ b/src/approval.rs @@ -1,17 +1,12 @@ //! Channel-neutral, durable user questions and normalized answers. -#[cfg(test)] use anyhow::{bail, Result}; use serde::{Deserialize, Serialize}; use uuid::Uuid; -#[cfg(test)] pub const MAX_CHOICES: usize = 9; -#[cfg(test)] -const MAX_PROMPT_CHARS: usize = 2_000; -#[cfg(test)] +const MAX_PROMPT_CHARS: usize = 64 * 1024; const MAX_LABEL_CHARS: usize = 256; -#[cfg(test)] const MAX_VALUE_CHARS: usize = 256; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -20,7 +15,6 @@ pub struct Choice { pub value: String, } -#[cfg(test)] #[derive(Debug, Clone, PartialEq, Eq)] pub struct Question { pub id: String, @@ -34,7 +28,6 @@ pub struct Question { pub expires_at_ms: i64, } -#[cfg(test)] impl Question { pub fn new( origin: AnswerOrigin, @@ -91,15 +84,28 @@ impl Question { Ok(()) } + #[cfg_attr(not(test), allow(dead_code))] pub fn render_text(&self) -> String { + self.render_text_with_instruction(&format!( + "Reply with a number, or `{} `. Expires automatically.", + self.id + )) + } + + pub fn render_correlated_text(&self) -> String { + self.render_text_with_instruction(&format!( + "Reply with `{} `. A number alone is not accepted. Expires automatically.", + self.id + )) + } + + fn render_text_with_instruction(&self, instruction: &str) -> String { let mut text = format!("{}\n", self.prompt.trim()); for (index, choice) in self.choices.iter().enumerate() { text.push_str(&format!("\n{}. {}", index + 1, choice.label.trim())); } - text.push_str(&format!( - "\n\nReply with a number, or `{} `. Expires automatically.", - self.id - )); + text.push_str("\n\n"); + text.push_str(instruction); text } } @@ -158,7 +164,6 @@ pub enum AnswerOutcome { Ambiguous, } -#[cfg(test)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DeliveryStatus { Delivered, @@ -189,7 +194,6 @@ impl QuestionState { } } -#[cfg(test)] impl DeliveryStatus { pub fn as_str(self) -> &'static str { match self { @@ -264,6 +268,9 @@ mod tests { ) .is_err()); let mut question = Question::new(origin, "me", "Continue?", choices, 1).unwrap(); + let correlated = question.render_correlated_text(); + assert!(correlated.contains(&format!("Reply with `{} `", question.id))); + assert!(correlated.contains("A number alone is not accepted")); question.id = "not-a-uuid".to_string(); assert!(question.validate().is_err()); } diff --git a/src/assistant.rs b/src/assistant.rs index 3d37f32..e72f92e 100644 --- a/src/assistant.rs +++ b/src/assistant.rs @@ -31,7 +31,7 @@ const AGENTS: &str = r#"# Assistant repository instructions - Treat `SOUL.md` as user-owned identity. Do not edit it unless the user asks. - Use `context/` for durable user context and working notes. - Treat `evals/` as user-owned evaluation criteria. Do not edit them during evaluation. -- Store job runbooks in `jobs/`. Create or update them directly when the user asks, then run `push job validate`. +- Store job runbooks in `jobs/`. Create or update them directly when the user asks, then run `push job validate`. Say when an enabled schedule is saved but still awaiting Push's separate owner review. - Keep secrets, sessions, databases, logs, and other runtime state outside this repository. "#; @@ -61,7 +61,7 @@ Good examples include preferences, active projects, people, recurring processes, "#; const PUSH_SKILL: &str = include_str!("../assistant/skills/push/SKILL.md"); -const PUSH_SKILL_VERSION: u32 = 1; +const PUSH_SKILL_VERSION: u32 = 2; const PUSH_SKILL_LINK: &str = "../../skills/push"; const PUSH_SKILL_MANIFEST: &str = ".push-managed.json"; const PUSH_SKILL_PROVIDERS: [&str; 2] = [".agents", ".claude"]; diff --git a/src/audit.rs b/src/audit.rs index 122c921..bdd08e5 100644 --- a/src/audit.rs +++ b/src/audit.rs @@ -1,5 +1,6 @@ //! Local JSONL audit log for production debugging. +use std::io::{Read, Seek, SeekFrom, Write}; use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -23,6 +24,8 @@ pub struct AuditEvent { pub ts_ms: u64, pub event: String, #[serde(skip_serializing_if = "Option::is_none")] + pub event_id: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub row_id: Option, #[serde(skip_serializing_if = "Option::is_none")] pub channel: Option, @@ -52,6 +55,12 @@ pub struct AuditEvent { pub reply: Option, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub job_name: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub revision: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub actor: Option, } #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -88,8 +97,10 @@ impl AuditLog { std::fs::create_dir_all(parent) .with_context(|| format!("create audit log directory {}", parent.display()))?; } + let mut encoded = serde_json::to_vec(&event).context("encode audit event")?; + encoded.push(b'\n'); let mut options = std::fs::OpenOptions::new(); - options.create(true).append(true); + options.create(true).read(true).write(true).append(true); #[cfg(unix)] { use std::os::unix::fs::OpenOptionsExt; @@ -100,16 +111,32 @@ impl AuditLog { .with_context(|| format!("open audit log {}", self.path.display()))?; crate::util::restrict_permissions(path, false) .with_context(|| format!("restrict audit log permissions {}", self.path.display()))?; - serde_json::to_writer(&mut file, &event).context("write audit event")?; - use std::io::Write; - writeln!(file).context("finish audit event")?; + repair_incomplete_tail(&mut file).context("repair incomplete audit event")?; + file.write_all(&encoded).context("write audit event")?; + file.sync_data().context("sync audit event")?; Ok(()) } + pub(crate) fn flush_schedule_reviews(&self, ledger: &mut crate::jobs::Ledger) -> Result { + let mut written = 0usize; + loop { + let events = ledger.pending_schedule_audit_events(100)?; + if events.is_empty() { + return Ok(written); + } + for event in events { + self.record(self.schedule_review(&event))?; + ledger.mark_schedule_audit_logged(event.id, crate::util::now_ms())?; + written += 1; + } + } + } + pub fn inbound(&self, msg: &RawMessage) -> AuditEvent { AuditEvent { ts_ms: now_ms(), event: "message_inbound".to_string(), + event_id: None, row_id: Some(msg.row_id), channel: Some(msg.channel.to_string()), thread: None, @@ -125,6 +152,9 @@ impl AuditLog { message: Some(content(&msg.text, self.include_content)), reply: None, error: None, + job_name: None, + revision: None, + actor: None, } } @@ -236,6 +266,29 @@ impl AuditLog { event.reason = Some(reason.into()); event } + + pub fn schedule_review(&self, event: &crate::jobs::ScheduleReviewEvent) -> AuditEvent { + let mut audit = self.base( + match event.event.as_str() { + "proposed" => "schedule_review_proposed", + "approved" => "schedule_review_approved", + "rejected" => "schedule_review_rejected", + "invalidated" => "schedule_review_invalidated", + "activated" => "schedule_review_activated", + _ => "schedule_review_unknown", + }, + None, + None, + None, + ); + audit.ts_ms = u64::try_from(event.created_at_ms).unwrap_or_default(); + audit.event_id = Some(format!("job_schedule_event:{}", event.audit_event_id)); + audit.job_name = Some(event.job_name.clone()); + audit.revision = Some(event.content_hash.clone()); + audit.reason = event.reason.clone(); + audit.actor = event.actor.clone(); + audit + } fn base( &self, event: &'static str, @@ -246,6 +299,7 @@ impl AuditLog { AuditEvent { ts_ms: now_ms(), event: event.to_string(), + event_id: None, row_id, channel: Some(self.channel.clone()), thread: thread.map(str::to_string), @@ -261,6 +315,9 @@ impl AuditLog { message: None, reply: None, error: None, + job_name: None, + revision: None, + actor: None, } } } @@ -277,6 +334,35 @@ impl AuditEvent { } } +fn repair_incomplete_tail(file: &mut std::fs::File) -> std::io::Result<()> { + let len = file.metadata()?.len(); + if len == 0 { + return Ok(()); + } + file.seek(SeekFrom::End(-1))?; + let mut last = [0u8; 1]; + file.read_exact(&mut last)?; + if last[0] == b'\n' { + return Ok(()); + } + + let mut end = len; + let mut buffer = [0u8; 8 * 1024]; + while end > 0 { + let start = end.saturating_sub(buffer.len() as u64); + let count = usize::try_from(end - start).unwrap_or(buffer.len()); + file.seek(SeekFrom::Start(start))?; + file.read_exact(&mut buffer[..count])?; + if let Some(index) = buffer[..count].iter().rposition(|byte| *byte == b'\n') { + file.set_len(start + index as u64 + 1)?; + return Ok(()); + } + end = start; + } + file.set_len(0)?; + Ok(()) +} + fn content(text: &str, include: bool) -> AuditContent { AuditContent { chars: text.chars().count(), @@ -375,6 +461,33 @@ mod tests { assert_eq!(failed.error.as_deref(), Some("send failed")); } + #[test] + fn schedule_review_events_keep_job_and_revision_context() { + let audit = AuditLog::new("audit.jsonl".to_string(), false, "scheduler"); + let event = audit.schedule_review(&crate::jobs::ScheduleReviewEvent { + id: 17, + audit_event_id: "9ec38fe9-a6c8-4e74-b2cf-d6b8a943188a".to_string(), + event: "invalidated".to_string(), + job_name: "daily-review".to_string(), + content_hash: "abc123".to_string(), + review_id: "review-id".to_string(), + actor: Some("scheduler".to_string()), + reason: Some("job revision changed".to_string()), + created_at_ms: 1234, + }); + + assert_eq!(event.event, "schedule_review_invalidated"); + assert_eq!( + event.event_id.as_deref(), + Some("job_schedule_event:9ec38fe9-a6c8-4e74-b2cf-d6b8a943188a") + ); + assert_eq!(event.ts_ms, 1234); + assert_eq!(event.job_name.as_deref(), Some("daily-review")); + assert_eq!(event.revision.as_deref(), Some("abc123")); + assert_eq!(event.actor.as_deref(), Some("scheduler")); + assert_eq!(event.reason.as_deref(), Some("job revision changed")); + } + #[test] fn writes_jsonl_events() { let path = temp_path("audit-jsonl"); diff --git a/src/cli_json.rs b/src/cli_json.rs index 8db0189..c8ed921 100644 --- a/src/cli_json.rs +++ b/src/cli_json.rs @@ -271,6 +271,58 @@ fn run_job_command(config_path: &str, command: JobCommand) -> Result<(), CliErro }), ) } + JobCommand::Reviews(name) => { + if let Some(name) = name.as_deref() { + jobs::validate_job_name(name).map_err(CliError::invalid_input)?; + } + let ledger = jobs::Ledger::open(&cfg.paths.database).map_err(|error| { + CliError::configuration("the schedule review ledger could not be opened", error) + })?; + let rows = ledger.schedule_reviews(name.as_deref()).map_err(|error| { + CliError::configuration("the schedule review ledger could not be read", error) + })?; + let reviews = rows + .into_iter() + .map(|review| { + let schedules = review + .schedules + .into_iter() + .map(|trigger| { + json!({ + "id": trigger.id, + "kind": trigger.kind, + "schedule": trigger.schedule, + "timezone": trigger.timezone, + "enabled": trigger.enabled, + }) + }) + .collect::>(); + json!({ + "review_id": review.review_id, + "job_name": review.job_name, + "status": review.status, + "content_hash": review.content_hash, + "schedules": schedules, + "backend": review.backend, + "timeout_ms": review.timeout_ms, + "workdir": review.workdir, + "delivery": { + "channel": review.delivery_channel, + "target": review.delivery_target, + }, + "reviewed_by": review.reviewed_by, + "reason": review.reason, + }) + }) + .collect::>(); + write_success( + "job.reviews", + json!({ + "job_name": name, + "reviews": reviews, + }), + ) + } JobCommand::Run(_) => unreachable!("job run JSON mode is rejected before config loading"), } } @@ -282,12 +334,19 @@ fn load_config(path: &str) -> Result { anyhow!("configuration not found at {path}"), )); } - config::Config::load(path).map_err(|error| { + let cfg = config::Config::load(path).map_err(|error| { CliError::configuration( format!("configuration at {path} could not be loaded; run `push doctor` for details"), error, ) - }) + })?; + jobs::Ledger::capture_legacy_schedule_baseline(&cfg).map_err(|error| { + CliError::configuration( + "the existing schedule migration baseline could not be captured", + error, + ) + })?; + Ok(cfg) } fn catalog_value(catalog: &jobs::Catalog) -> Value { diff --git a/src/doctor.rs b/src/doctor.rs index 5010789..cc92381 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -41,6 +41,7 @@ pub fn doctor(config_path: &str) -> Result<()> { bail!("doctor found 1 failed check"); } }; + jobs::Ledger::capture_legacy_schedule_baseline(&cfg)?; let report = run_checks(&cfg); print!("{report}"); if report.is_ok() { diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 47fe528..33a3bcc 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -17,9 +17,7 @@ use tokio::task::{JoinHandle, JoinSet}; use tracing::{error, info, warn}; use crate::agent::Runner; -#[cfg(test)] -use crate::approval::AnswerOrigin; -use crate::approval::AnswerOutcome; +use crate::approval::{AnswerOrigin, AnswerOutcome}; use crate::audit::{AuditEvent, AuditLog}; use crate::channel::{Channel, InboundVoice, RawMessage}; use crate::config::{AgentBackend, ChannelKind, Config, PrimaryDeliveryConfig}; @@ -44,6 +42,7 @@ struct Job { text: String, reply_with_voice: bool, voice_attachment: Option, + approval_origin: AnswerOrigin, } /// Shared, cheaply cloneable context handed to each worker task. @@ -60,6 +59,7 @@ struct Ctx { assistant_dir: String, audit: Arc, voice: Option, + schedule_destination: Option, #[cfg(test)] setup_failure_replies: Arc>>, #[cfg(test)] @@ -115,6 +115,7 @@ struct WorkerState { impl GatewayGroup { pub fn new(cfg: Config) -> Result { + jobs::Ledger::capture_legacy_schedule_baseline(&cfg)?; let enabled = cfg.enabled_channel_kinds()?; let store = Arc::new(Mutex::new(Store::open(&cfg.paths)?)); let history = Arc::new(Mutex::new( @@ -138,11 +139,32 @@ impl GatewayGroup { audit_lock.clone(), )?); } - Ok(Self { + let mut group = Self { gateways, primary_delivery: cfg.primary_delivery.clone(), cfg, - }) + }; + let destination = group.primary_destination().ok(); + let mut ledger = jobs::Ledger::open(&group.cfg.paths.database)?; + if let Some(destination) = destination { + let catalog = jobs::Catalog::load(&group.cfg)?; + ledger.recover_answered_schedule_reviews(&group.cfg, now_ms())?; + ledger.reconcile_schedule_reviews( + &catalog, + &destination.channel, + &destination.target, + now_ms(), + )?; + for gateway in &mut group.gateways { + gateway.ctx.schedule_destination = Some(destination.clone()); + } + } else { + ledger.settle_legacy_schedule_migration_without_destination()?; + } + if let Some(gateway) = group.gateways.first() { + audit_schedule_events(&gateway.ctx, &mut ledger); + } + Ok(group) } pub fn primary_destination(&self) -> Result { @@ -259,7 +281,8 @@ async fn run_scheduler( } _ = ticker.tick() => { let contexts = contexts.clone(); - if let Err(error) = scheduler.tick(now_ms(), move |channel, target, text, start_chunk, progress| { + let audit_contexts = contexts.clone(); + let result = scheduler.tick(now_ms(), move |channel, target, text, start_chunk, progress| { let contexts = contexts.clone(); async move { let Some(ctx) = contexts.get(&channel) else { @@ -281,7 +304,17 @@ async fn run_scheduler( }; scheduled_reply_to(ctx, &target, &text, start_chunk, progress).await } - }).await { + }).await; + if let Some(ctx) = audit_contexts.values().next() { + scheduler.take_schedule_events(); + match jobs::Ledger::open(&ctx.cfg.paths.database) { + Ok(mut ledger) => audit_schedule_events(ctx, &mut ledger), + Err(error) => { + error!("open schedule audit outbox: {error:#}"); + } + } + } + if let Err(error) = result { error!("job scheduler tick failed: {error:#}"); } } @@ -378,6 +411,7 @@ impl Gateway { reply_marker: crate::channel::REPLY_MARKER.to_string(), assistant_dir: cfg.assistant_dir.clone(), audit, + schedule_destination: None, #[cfg(not(test))] voice: Voice::from_config(&cfg), #[cfg(test)] @@ -584,6 +618,80 @@ impl Gateway { Ok(AnswerOutcome::NotAnAnswer) => {} Ok(outcome @ (AnswerOutcome::Selected(_) | AnswerOutcome::Duplicate(_))) => { self.audit_approval(m.row_id, &thread, &outcome); + let correlation_id = match &outcome { + AnswerOutcome::Selected(answer) => &answer.correlation_id, + AnswerOutcome::Duplicate(id) => id, + _ => unreachable!(), + }; + let reviewer = format!( + "channel={} thread={} sender={} chat={}", + approval_origin.channel, + approval_origin.thread_key, + approval_origin.sender_key, + approval_origin.chat_key + ); + let decision = + jobs::Ledger::open(&self.cfg.paths.database).and_then(|mut ledger| { + let decision = ledger.resolve_schedule_answer( + &self.cfg, + correlation_id, + &reviewer, + now_ms(), + )?; + audit_schedule_events(&self.ctx, &mut ledger); + Ok(decision) + }); + let decision = match decision { + Ok(decision) => decision, + Err(error) => { + error!("[{thread}] schedule review decision failed: {error:#}"); + self.audit(self.ctx.audit.failed( + "schedule_review_decision_failed", + m.row_id, + &thread, + None, + error.to_string(), + )); + return; + } + }; + let confirmation = match decision { + jobs::ScheduleDecision::Approved { + job_name, + content_hash, + event: _, + } => { + Some(format!( + "Approved schedule activation for `{job_name}` revision `{content_hash}`. It will activate on the next scheduler tick." + )) + } + jobs::ScheduleDecision::Rejected { + job_name, + content_hash, + event: _, + } => { + Some(format!( + "Rejected schedule activation for `{job_name}` revision `{content_hash}`. The Markdown file remains available for manual use." + )) + } + jobs::ScheduleDecision::Invalidated { + job_name, + content_hash, + reason, + event: _, + } => { + Some(format!( + "Could not activate `{job_name}` revision `{content_hash}` because it changed or became invalid: {reason}" + )) + } + jobs::ScheduleDecision::AlreadyHandled + | jobs::ScheduleDecision::NotScheduleReview => None, + }; + if let Some(confirmation) = confirmation { + if !reply_to(&self.ctx, &target, &confirmation).await { + warn!("[{thread}] schedule review confirmation delivery failed"); + } + } self.complete_row(m.row_id, "approval_answer"); continue; } @@ -691,6 +799,7 @@ impl Gateway { text: message_text, reply_with_voice, voice_attachment: m.voice.clone(), + approval_origin, }; if job.text.trim().eq_ignore_ascii_case("/stop") { if !self.stop(job).await { @@ -1008,6 +1117,12 @@ fn audit(ctx: &Ctx, event: AuditEvent) { } } +fn audit_schedule_events(ctx: &Ctx, ledger: &mut jobs::Ledger) { + if let Err(error) = ctx.audit.flush_schedule_reviews(ledger) { + error!("schedule audit outbox error: {error:#}"); + } +} + async fn reply_to(ctx: &Ctx, target: &str, text: &str) -> bool { let chunks = ctx.channel.outbound_chunks(text, &ctx.reply_marker); if chunks.is_empty() { diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index ef773f6..21218bf 100644 --- a/src/gateway/tests.rs +++ b/src/gateway/tests.rs @@ -224,6 +224,7 @@ fn setup_failure_ctx( false, "imessage", )), + schedule_destination: None, voice: None, setup_failure_replies: Arc::new(Mutex::new(Vec::new())), sent_replies: Arc::new(Mutex::new(Vec::new())), @@ -243,6 +244,12 @@ fn setup_failure_job(row_id: i64) -> Job { text: "hello".to_string(), reply_with_voice: false, voice_attachment: None, + approval_origin: AnswerOrigin { + channel: "imessage".to_string(), + thread_key: "imessage:self:me".to_string(), + sender_key: "me".to_string(), + chat_key: "me".to_string(), + }, } } @@ -1267,6 +1274,77 @@ async fn pending_outbound_is_delivered_after_restart_without_backend_rerun() { let _ = std::fs::remove_dir_all(assistant_dir); } +#[tokio::test(flavor = "current_thread")] +async fn recovered_outbound_still_presents_authored_schedule_review() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("schedule-history-recovery-sessions"); + let assistant_dir = temp_path("schedule-history-recovery-assistant"); + let jobs_dir = assistant_dir.join("jobs"); + let workdir = assistant_dir.join("work"); + std::fs::create_dir_all(&jobs_dir).unwrap(); + std::fs::create_dir_all(&workdir).unwrap(); + let mut config = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + config.jobs_dir = jobs_dir.to_string_lossy().to_string(); + std::fs::write( + jobs_dir.join("recovered-schedule.md"), + format!( + "+++\nversion = 1\ntimeout = \"5s\"\nworkdir = {:?}\nbackend = \"codex\"\n\n[[triggers]]\nid = \"morning\"\nkind = \"cron\"\nschedule = \"0 8 * * *\"\ntimezone = \"Europe/London\"\nenabled = true\n+++\n\nPrepare a note.\n", + workdir.to_string_lossy() + ), + ) + .unwrap(); + let mut history = History::open(&config.paths.database).unwrap(); + let inbound_id = history + .record_inbound( + "imessage", + "imessage:dm:+15551234567", + "imessage:1", + "create a schedule", + ) + .unwrap(); + history + .record_outbound( + inbound_id, + OutboundOrigin::Backend, + Some("codex"), + "stored reply", + ) + .unwrap(); + drop(history); + + let calls = Arc::new(Mutex::new(Vec::new())); + let mut gateway = Gateway::new(config).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + gateway.ctx.schedule_destination = Some(PrimaryDestination { + channel: "imessage".to_string(), + target: "+15551234567".to_string(), + }); + gateway + .tick_fake(vec![message( + 1, + "+15551234567", + "+15551234567", + false, + "create a schedule", + )]) + .await; + gateway.queues.clear(); + gateway.drain_workers().await; + + assert!(calls.lock().unwrap().is_empty()); + let replies = gateway.ctx.sent_replies.lock().unwrap(); + assert!(replies + .iter() + .any(|(_, text)| text.contains("stored reply"))); + assert!(replies.iter().any(|(_, text)| { + text.contains("Review schedule activation") && text.contains("recovered-schedule") + })); +} + #[tokio::test(flavor = "current_thread")] async fn session_state_save_failure_keeps_reply_for_restart_without_backend_rerun() { let state_path = temp_state_path(); @@ -1781,6 +1859,12 @@ async fn closed_worker_queue_is_recovered_without_another_message() { text: "recover older".to_string(), reply_with_voice: false, voice_attachment: None, + approval_origin: AnswerOrigin { + channel: "imessage".to_string(), + thread_key: thread.to_string(), + sender_key: "me@icloud.com".to_string(), + chat_key: "me@icloud.com".to_string(), + }, }; let (jobs, rx) = mpsc::channel(QUEUE_DEPTH); @@ -2075,6 +2159,12 @@ async fn stop_targets_the_current_row_ahead_of_retained_failures() { text: text.to_string(), reply_with_voice: false, voice_attachment: None, + approval_origin: AnswerOrigin { + channel: "imessage".to_string(), + thread_key: thread.to_string(), + sender_key: "me@icloud.com".to_string(), + chat_key: "me@icloud.com".to_string(), + }, }; let inbound_ids = ["failed", "active", "/stop"] .into_iter() @@ -2579,6 +2669,66 @@ async fn missing_primary_disables_new_schedules_without_stopping_gateway() { let _ = std::fs::remove_dir_all(workdir); } +#[test] +fn missing_primary_closes_upgrade_migration_before_later_schedule_creation() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("missing-primary-migration-sessions"); + let assistant_dir = temp_path("missing-primary-migration-assistant"); + let jobs_dir = temp_path("missing-primary-migration-jobs"); + let workdir = temp_path("missing-primary-migration-work"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + std::fs::create_dir_all(&jobs_dir).unwrap(); + std::fs::create_dir_all(&workdir).unwrap(); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.jobs_dir = jobs_dir.to_string_lossy().to_string(); + let history = crate::history::History::open(&cfg.paths.database).unwrap(); + history.execute_batch_for_test( + "DROP TABLE job_schedule_review_questions; + DROP TABLE job_schedule_events; + DROP TABLE job_schedule_reviews; + DROP TABLE job_schedule_legacy_baseline; + DROP TABLE job_schedule_meta; + PRAGMA user_version = 11;", + ); + drop(history); + + let missing = GatewayGroup::new(cfg.clone()).unwrap(); + drop(missing); + std::fs::write( + jobs_dir.join("later.md"), + format!( + "+++\nversion = 1\ntimeout = \"5s\"\nworkdir = {:?}\nbackend = \"codex\"\n\n[[triggers]]\nid = \"minute\"\nkind = \"cron\"\nschedule = \"* * * * *\"\ntimezone = \"UTC\"\nenabled = true\n+++\n\nRun.\n", + workdir.to_string_lossy() + ), + ) + .unwrap(); + cfg.primary_delivery = Some(PrimaryDeliveryConfig { + channel: "imessage".to_string(), + target: "+15551234567".to_string(), + }); + + let valid = GatewayGroup::new(cfg.clone()).unwrap(); + drop(valid); + let reviews = crate::jobs::Ledger::open(&cfg.paths.database) + .unwrap() + .schedule_reviews(Some("later")) + .unwrap(); + assert_eq!(reviews.len(), 1); + assert_eq!(reviews[0].status, "proposed"); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.db")); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); + let _ = std::fs::remove_dir_all(jobs_dir); + let _ = std::fs::remove_dir_all(workdir); +} + #[tokio::test] async fn one_channel_failure_does_not_stop_another_and_shutdown_reaches_survivor() { let (shutdown_tx, shutdown_rx) = watch::channel(false); @@ -2709,6 +2859,83 @@ async fn route_agent_writes_job_directly_without_approval() { assert!(!replies.iter().any(|(_, text)| text.contains("Approve"))); } +#[tokio::test] +async fn direct_authored_schedule_is_saved_then_reviewed_in_its_owner_channel() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("direct-schedule-sessions"); + let assistant_dir = temp_path("direct-schedule-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.jobs_dir = assistant_dir.join("jobs").to_string_lossy().to_string(); + std::fs::create_dir_all(&cfg.jobs_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut gateway = Gateway::new(cfg.clone()).unwrap(); + gateway.ctx.schedule_destination = Some(PrimaryDestination { + channel: "imessage".to_string(), + target: "+15551234567".to_string(), + }); + let job_path = Path::new(&cfg.jobs_dir).join("agent-schedule.md"); + let hook = Arc::new(move || { + std::fs::write( + &job_path, + "+++\nversion = 1\ntimeout = \"5s\"\nbackend = \"codex\"\n\n[[triggers]]\nid = \"morning\"\nkind = \"cron\"\nschedule = \"0 8 * * *\"\ntimezone = \"Europe/London\"\nenabled = true\n+++\n\nPrepare a note.\n", + ) + .unwrap(); + }); + gateway.ctx.runners = Arc::new(fake_runners_with_hook(calls, Some(hook))); + + run_messages( + &mut gateway, + vec![message( + 1, + "+15551234567", + "+15551234567", + false, + "Create a morning schedule", + )], + ) + .await; + + let replies = gateway.ctx.sent_replies.lock().unwrap().clone(); + let review = replies + .iter() + .map(|(_, text)| text) + .find(|text| text.contains("Review schedule activation")) + .expect("schedule review should be delivered after direct authoring"); + assert!(review.contains("Job: agent-schedule")); + assert!(review.contains("0 8 * * *")); + let question_id = review + .split(|character: char| character.is_whitespace() || character == '`') + .find(|part| Uuid::parse_str(part).is_ok()) + .unwrap() + .to_string(); + drop(replies); + + run_messages( + &mut gateway, + vec![message( + 2, + "+15551234567", + "+15551234567", + false, + &format!("{question_id} 1"), + )], + ) + .await; + + assert!(gateway + .ctx + .sent_replies + .lock() + .unwrap() + .iter() + .any(|(_, text)| text.contains("Approved schedule activation"))); +} + #[tokio::test] async fn retired_job_approval_reply_explains_direct_creation() { let state_path = temp_state_path(); diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 05e7403..458184f 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -14,7 +14,7 @@ use crate::history::{DeliveryStatus, OutboundMessage, OutboundOrigin}; use crate::prompt::{ComposedPrompt, Composer}; use crate::voice::MAX_AUDIO_BYTES; -use super::{audit, complete_row, Ctx, Job, WorkerState}; +use super::{audit, audit_schedule_events, complete_row, Ctx, Job, WorkerState}; pub(super) const SESSION_SETUP_FAILURE: &str = "Push could not prepare this conversation. Check the local logs, then resend."; @@ -71,6 +71,7 @@ where "recovered_outbound", "recover outbound", ); + review_changed_schedules(ctx, &job).await; return; } @@ -489,6 +490,71 @@ where .await; } } + review_changed_schedules(ctx, &job).await; +} + +async fn review_changed_schedules(ctx: &Ctx, job: &Job) { + let Some(destination) = &ctx.schedule_destination else { + return; + }; + let result = (|| { + let catalog = crate::jobs::Catalog::load(&ctx.cfg)?; + let mut ledger = crate::jobs::Ledger::open(&ctx.cfg.paths.database)?; + let (_, events) = ledger.reconcile_schedule_reviews( + &catalog, + &destination.channel, + &destination.target, + crate::util::now_ms(), + )?; + let questions = ledger.schedule_review_questions( + &job.approval_origin, + &job.target, + crate::util::now_ms(), + )?; + Ok::<_, anyhow::Error>((ledger, events, questions)) + })(); + let (mut ledger, _events, questions) = match result { + Ok(result) => result, + Err(error) => { + error!( + "[{}] reconcile schedule activation reviews: {error:#}", + job.thread + ); + audit( + ctx, + ctx.audit.failed( + "schedule_review_failed", + job.row_id, + &job.thread, + None, + error.to_string(), + ), + ); + return; + } + }; + audit_schedule_events(ctx, &mut ledger); + for question in questions { + let delivered = + super::reply_to(ctx, &question.target, &question.render_correlated_text()).await; + if let Err(error) = ledger.mark_schedule_question_delivery( + &question.id, + if delivered { + crate::approval::DeliveryStatus::Delivered + } else { + crate::approval::DeliveryStatus::Failed + }, + crate::util::now_ms(), + ) { + error!( + "[{}] persist schedule review delivery: {error:#}", + job.thread + ); + } + if !delivered { + warn!("[{}] schedule review delivery failed", job.thread); + } + } } async fn interrupt(cancel: &mut watch::Receiver, row_id: i64) { diff --git a/src/history.rs b/src/history.rs index bae0342..3acb0c5 100644 --- a/src/history.rs +++ b/src/history.rs @@ -6,13 +6,11 @@ use std::time::Duration; use anyhow::{bail, Context, Result}; use rusqlite::{params, Connection, OptionalExtension, Transaction}; -#[cfg(test)] -use crate::approval::QuestionState; use crate::approval::{parse_answer, AnswerOrigin, AnswerOutcome, NormalizedAnswer}; #[cfg(test)] -use crate::approval::{DeliveryStatus as ApprovalDeliveryStatus, Question}; +use crate::approval::{DeliveryStatus as ApprovalDeliveryStatus, Question, QuestionState}; -const SCHEMA_VERSION: i64 = 11; +const SCHEMA_VERSION: i64 = 12; const RETIRED_JOB_APPROVAL_ERROR: &str = "job approval was removed; request direct job creation"; const MAX_HISTORY_READ_BYTES: usize = 8 * 1024; const READ_TRUNCATED: &str = "\n[truncated by push while reading history]"; @@ -332,7 +330,6 @@ impl History { Ok(messages) } - /// Test-only seeding for the inbound answer-resolution flow. #[cfg(test)] pub fn create_question(&mut self, question: &Question, now_ms: i64) -> Result<()> { question.validate()?; @@ -401,6 +398,10 @@ impl History { WHERE channel = ?1 AND thread_key = ?2 AND sender_key = ?3 AND chat_key = ?4 AND status = 'pending' + AND NOT EXISTS ( + SELECT 1 FROM job_schedule_review_questions + WHERE question_id = approval_questions.id + ) ORDER BY created_at_ms, id", )?; let ids = statement @@ -853,6 +854,91 @@ fn migrate(conn: &Connection) -> Result<()> { PRAGMA user_version = 11;", )?; } + if version <= 11 { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS job_schedule_reviews ( + id TEXT PRIMARY KEY, + job_name TEXT NOT NULL, + content_hash TEXT NOT NULL, + snapshot_hash TEXT NOT NULL, + file_identity TEXT NOT NULL, + path TEXT NOT NULL, + schedules_json TEXT NOT NULL, + backend TEXT NOT NULL, + timeout_ms INTEGER NOT NULL, + workdir TEXT NOT NULL, + delivery_channel TEXT NOT NULL, + delivery_target TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ( + 'proposed', 'approved', 'rejected', 'invalidated', 'activated' + )), + proposed_at_ms INTEGER NOT NULL, + decided_at_ms INTEGER, + activated_at_ms INTEGER, + invalidated_at_ms INTEGER, + reviewed_by TEXT, + reason TEXT + ); + CREATE INDEX IF NOT EXISTS job_schedule_reviews_current_idx + ON job_schedule_reviews(job_name, status, proposed_at_ms); + CREATE TABLE IF NOT EXISTS job_schedule_review_questions ( + question_id TEXT PRIMARY KEY REFERENCES approval_questions(id), + review_id TEXT NOT NULL REFERENCES job_schedule_reviews(id), + created_at_ms INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS job_schedule_review_questions_review_idx + ON job_schedule_review_questions(review_id, created_at_ms); + CREATE TABLE IF NOT EXISTS job_schedule_events ( + id INTEGER PRIMARY KEY, + audit_event_id TEXT NOT NULL UNIQUE, + review_id TEXT NOT NULL REFERENCES job_schedule_reviews(id), + job_name TEXT NOT NULL, + content_hash TEXT NOT NULL, + event TEXT NOT NULL CHECK(event IN ( + 'proposed', 'approved', 'rejected', 'invalidated', 'activated' + )), + actor TEXT, + reason TEXT, + created_at_ms INTEGER NOT NULL, + audit_logged_at_ms INTEGER + ); + CREATE INDEX IF NOT EXISTS job_schedule_events_job_idx + ON job_schedule_events(job_name, created_at_ms, id); + CREATE TABLE IF NOT EXISTS job_schedule_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS job_schedule_legacy_baseline ( + review_id TEXT PRIMARY KEY, + job_name TEXT NOT NULL, + content_hash TEXT NOT NULL, + snapshot_hash TEXT NOT NULL, + file_identity TEXT NOT NULL, + path TEXT NOT NULL, + schedules_json TEXT NOT NULL, + backend TEXT NOT NULL, + timeout_ms INTEGER NOT NULL, + workdir TEXT NOT NULL, + delivery_channel TEXT NOT NULL, + delivery_target TEXT NOT NULL + );", + )?; + conn.execute( + "INSERT OR IGNORE INTO job_schedule_meta(key, value) + VALUES ('legacy_schedule_migration', ?1)", + [if version == 0 { "complete" } else { "pending" }], + )?; + conn.execute( + "INSERT OR IGNORE INTO job_schedule_meta(key, value) + VALUES ('legacy_schedule_baseline', ?1)", + [if version == 0 { + "not_required" + } else { + "unclaimed" + }], + )?; + conn.execute_batch("PRAGMA user_version = 12;")?; + } conn.execute_batch("COMMIT;")?; Ok(()) } @@ -1022,6 +1108,40 @@ mod tests { let _ = std::fs::remove_file(path); } + #[test] + fn v11_migration_marks_installed_schedules_for_one_time_preservation() { + let path = temp_path("schedule-activation-v11-migration"); + let history = History::open(path.to_str().unwrap()).unwrap(); + history.execute_batch_for_test( + "DROP TABLE job_schedule_review_questions; + DROP TABLE job_schedule_events; + DROP TABLE job_schedule_reviews; + DROP TABLE job_schedule_legacy_baseline; + DROP TABLE job_schedule_meta; + PRAGMA user_version = 11;", + ); + drop(history); + + let reopened = History::open(path.to_str().unwrap()).unwrap(); + + let migration: String = reopened + .conn + .query_row( + "SELECT value FROM job_schedule_meta + WHERE key = 'legacy_schedule_migration'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(migration, "pending"); + let version: i64 = reopened + .conn + .query_row("PRAGMA user_version", [], |row| row.get(0)) + .unwrap(); + assert_eq!(version, SCHEMA_VERSION); + let _ = std::fs::remove_file(path); + } + #[test] fn outbound_delivery_chunk_progress_survives_reopen() { let path = temp_path("outbound-delivery-progress"); diff --git a/src/jobs.rs b/src/jobs.rs index c9cde17..3000dc7 100644 --- a/src/jobs.rs +++ b/src/jobs.rs @@ -9,18 +9,19 @@ use std::time::{Duration, Instant}; use anyhow::{bail, Context, Result}; use chrono::{Datelike, LocalResult, TimeZone, Timelike, Utc}; -use rusqlite::{params, Connection, TransactionBehavior}; -use serde::Deserialize; +use rusqlite::{params, Connection, OptionalExtension, Transaction, TransactionBehavior}; +use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use tokio::sync::{mpsc, oneshot}; use tokio::task::JoinSet; use uuid::Uuid; use crate::agent::{Request, RunError, Runner}; +use crate::approval::{AnswerOrigin, Choice, DeliveryStatus as ApprovalDeliveryStatus, Question}; use crate::config::{AgentBackend, Config}; use crate::history::History; use crate::prompt::Composer; -use crate::util::{expand_home, now_ms, restrict_permissions, same_file}; +use crate::util::{expand_home, file_identity, now_ms, restrict_permissions, same_file}; const MAX_STORED_RESULT_BYTES: usize = 64 * 1024; const MAX_EVAL_BYTES: usize = 64 * 1024; @@ -54,7 +55,7 @@ pub struct Eval { pub body: String, } -#[derive(Debug, Deserialize, Clone)] +#[derive(Debug, Deserialize, Serialize, Clone)] #[serde(deny_unknown_fields)] pub struct Trigger { pub id: String, @@ -72,7 +73,9 @@ pub struct Job { pub timeout: Duration, pub workdir: PathBuf, pub backend: AgentBackend, + pub content_hash: String, pub snapshot_hash: String, + pub file_identity: String, pub evals: Vec, pub triggers: Vec, } @@ -183,7 +186,9 @@ fn load_file(cfg: &Config, name: &str, path: &Path) -> Result { let mut bytes = Vec::new(); file.read_to_end(&mut bytes) .with_context(|| format!("read job {}", path.display()))?; - validate_contents(cfg, name, path, &bytes) + let mut job = validate_contents(cfg, name, path, &bytes)?; + job.file_identity = file_identity(&opened); + Ok(job) } pub(crate) fn validate_contents( @@ -224,6 +229,7 @@ pub(crate) fn validate_contents( .context("canonicalize default job workdir from assistant_root")?, }; cfg.validate_job_workdir(&workdir)?; + let content_hash = hash_bytes(bytes); let mut snapshot = Sha256::new(); snapshot.update(bytes); for eval in &evals { @@ -240,7 +246,9 @@ pub(crate) fn validate_contents( timeout, workdir, backend, + content_hash, snapshot_hash, + file_identity: String::new(), evals, triggers: metadata.triggers, }) @@ -270,6 +278,10 @@ fn load_evals(cfg: &Config, names: &[String]) -> Result> { Ok(evals) } +fn hash_bytes(bytes: &[u8]) -> String { + format!("{:x}", Sha256::digest(bytes)) +} + fn load_eval(cfg: &Config, name: &str) -> Result { let root = std::fs::canonicalize(&cfg.assistant_root) .with_context(|| format!("resolve assistant root {}", cfg.assistant_root))?; @@ -633,6 +645,59 @@ pub struct Ledger { conn: Connection, } +const SCHEDULE_REVIEW_TTL_MS: i64 = 24 * 60 * 60 * 1_000; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ScheduleReviewEvent { + pub id: i64, + pub audit_event_id: String, + pub event: String, + pub job_name: String, + pub content_hash: String, + pub review_id: String, + pub actor: Option, + pub reason: Option, + pub created_at_ms: i64, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ScheduleDecision { + Approved { + job_name: String, + content_hash: String, + event: ScheduleReviewEvent, + }, + Rejected { + job_name: String, + content_hash: String, + event: ScheduleReviewEvent, + }, + Invalidated { + job_name: String, + content_hash: String, + reason: String, + event: Option, + }, + AlreadyHandled, + NotScheduleReview, +} + +#[derive(Debug, Clone)] +struct ScheduleReview { + id: String, + job_name: String, + content_hash: String, + snapshot_hash: String, + file_identity: String, + path: String, + schedules_json: String, + backend: String, + timeout_ms: i64, + workdir: String, + delivery_channel: String, + delivery_target: String, +} + #[derive(Debug)] pub struct RunRow { pub id: String, @@ -655,12 +720,30 @@ pub struct RunRow { pub delivery_target: Option, } +#[derive(Debug)] +pub struct ScheduleReviewRow { + pub review_id: String, + pub job_name: String, + pub content_hash: String, + pub status: String, + pub schedules: Vec, + pub backend: String, + pub timeout_ms: i64, + pub workdir: String, + pub delivery_channel: String, + pub delivery_target: String, + pub reviewed_by: Option, + pub reason: Option, +} + #[derive(Debug, Clone)] pub struct QueuedRun { pub id: String, pub job_name: String, pub snapshot_hash: String, pub trigger_id: String, + pub delivery_channel: Option, + pub delivery_target: Option, } #[derive(Debug)] @@ -748,9 +831,601 @@ impl Ledger { drop(History::open(database_path)?); let conn = Connection::open(database_path)?; conn.busy_timeout(Duration::from_secs(5))?; + conn.execute_batch("PRAGMA foreign_keys = ON;")?; Ok(Self { conn }) } + pub fn capture_legacy_schedule_baseline(cfg: &Config) -> Result<()> { + let mut ledger = Self::open(&cfg.paths.database)?; + let migration: String = ledger.conn.query_row( + "SELECT value FROM job_schedule_meta + WHERE key = 'legacy_schedule_migration'", + [], + |row| row.get(0), + )?; + let baseline: String = ledger.conn.query_row( + "SELECT value FROM job_schedule_meta + WHERE key = 'legacy_schedule_baseline'", + [], + |row| row.get(0), + )?; + if migration != "pending" || baseline != "unclaimed" { + return Ok(()); + } + + let destination = schedule_migration_destination(cfg).ok(); + let catalog = Catalog::load(cfg)?; + let candidates = destination + .as_ref() + .map(|(channel, target)| { + catalog + .jobs + .values() + .filter(|job| job.triggers.iter().any(|trigger| trigger.enabled)) + .map(|job| schedule_review(job, channel, target)) + .collect::>>() + }) + .transpose()? + .unwrap_or_default(); + let tx = ledger + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let still_unclaimed = tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM job_schedule_meta + WHERE key = 'legacy_schedule_baseline' AND value = 'unclaimed' + )", + [], + |row| row.get::<_, bool>(0), + )?; + if !still_unclaimed { + tx.rollback()?; + return Ok(()); + } + for review in candidates { + tx.execute( + "INSERT OR IGNORE INTO job_schedule_legacy_baseline ( + review_id, job_name, content_hash, snapshot_hash, file_identity, + path, schedules_json, backend, timeout_ms, workdir, + delivery_channel, delivery_target + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)", + params![ + review.id, + review.job_name, + review.content_hash, + review.snapshot_hash, + review.file_identity, + review.path, + review.schedules_json, + review.backend, + review.timeout_ms, + review.workdir, + review.delivery_channel, + review.delivery_target, + ], + )?; + } + tx.execute( + "UPDATE job_schedule_meta SET value = 'captured' + WHERE key = 'legacy_schedule_baseline' AND value = 'unclaimed'", + [], + )?; + tx.commit()?; + Ok(()) + } + + pub fn reconcile_schedule_reviews( + &mut self, + catalog: &Catalog, + delivery_channel: &str, + delivery_target: &str, + now: i64, + ) -> Result<(HashSet, Vec)> { + let candidates = catalog + .jobs + .values() + .filter(|job| job.triggers.iter().any(|trigger| trigger.enabled)) + .map(|job| { + schedule_review(job, delivery_channel, delivery_target) + .map(|review| (review.id.clone(), review)) + }) + .collect::>>()?; + let mut events = Vec::new(); + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let migration: String = tx.query_row( + "SELECT value FROM job_schedule_meta WHERE key = 'legacy_schedule_migration'", + [], + |row| row.get(0), + )?; + if migration == "pending" { + for review in candidates.values() { + let preserved = tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM job_schedule_legacy_baseline + WHERE review_id = ?1 + )", + [&review.id], + |row| row.get::<_, bool>(0), + )?; + if preserved && insert_schedule_review(&tx, review, "activated", now)? { + push_schedule_event( + &tx, + &mut events, + review, + "approved", + Some("migration"), + Some("preserved existing installed schedule"), + now, + )?; + push_schedule_event( + &tx, + &mut events, + review, + "activated", + Some("migration"), + Some("preserved existing installed schedule"), + now, + )?; + } + } + tx.execute( + "UPDATE job_schedule_meta SET value = 'complete' + WHERE key = 'legacy_schedule_migration'", + [], + )?; + tx.execute("DELETE FROM job_schedule_legacy_baseline", [])?; + } + + let live = tx + .prepare( + "SELECT id, job_name, content_hash, snapshot_hash, file_identity, path, + schedules_json, backend, timeout_ms, workdir, + delivery_channel, delivery_target + FROM job_schedule_reviews + WHERE status IN ('proposed', 'approved', 'activated') + ORDER BY proposed_at_ms, id", + )? + .query_map([], map_schedule_review)? + .collect::>>()?; + for review in live { + if !candidates.contains_key(&review.id) { + let changed = tx.execute( + "UPDATE job_schedule_reviews + SET status = 'invalidated', invalidated_at_ms = ?2, + reason = 'job revision or effective schedule changed' + WHERE id = ?1 AND status IN ('proposed', 'approved', 'activated')", + params![review.id, now], + )?; + if changed == 1 { + push_schedule_event( + &tx, + &mut events, + &review, + "invalidated", + Some("scheduler"), + Some("job revision or effective schedule changed"), + now, + )?; + cancel_review_questions(&tx, &review.id, now)?; + } + } + } + + for review in candidates.values() { + if insert_schedule_review(&tx, review, "proposed", now)? { + push_schedule_event( + &tx, + &mut events, + review, + "proposed", + Some("scheduler"), + None, + now, + )?; + } else { + let reproposed = tx.execute( + "UPDATE job_schedule_reviews + SET status = 'proposed', proposed_at_ms = ?2, + decided_at_ms = NULL, activated_at_ms = NULL, + invalidated_at_ms = NULL, reviewed_by = NULL, reason = NULL + WHERE id = ?1 AND status = 'invalidated'", + params![review.id, now], + )?; + if reproposed == 1 { + cancel_review_questions(&tx, &review.id, now)?; + push_schedule_event( + &tx, + &mut events, + review, + "proposed", + Some("scheduler"), + Some("valid schedule revision returned after invalidation"), + now, + )?; + } + } + let changed = tx.execute( + "UPDATE job_schedule_reviews + SET status = 'activated', activated_at_ms = ?2 + WHERE id = ?1 AND status = 'approved'", + params![review.id, now], + )?; + if changed == 1 { + push_schedule_event( + &tx, + &mut events, + review, + "activated", + Some("scheduler"), + None, + now, + )?; + } + } + + let active = tx + .prepare( + "SELECT id FROM job_schedule_reviews + WHERE status = 'activated' ORDER BY id", + )? + .query_map([], |row| row.get::<_, String>(0))? + .collect::>>()?; + tx.commit()?; + Ok((active, events)) + } + + pub fn settle_legacy_schedule_migration_without_destination(&mut self) -> Result<()> { + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute( + "UPDATE job_schedule_meta SET value = 'complete' + WHERE key = 'legacy_schedule_migration' AND value = 'pending'", + [], + )?; + tx.commit()?; + Ok(()) + } + + pub fn schedule_review_questions( + &mut self, + origin: &AnswerOrigin, + target: &str, + now: i64, + ) -> Result> { + self.conn.execute( + "UPDATE approval_questions SET status = 'expired', updated_at_ms = ?1 + WHERE status = 'pending' AND expires_at_ms <= ?1", + [now], + )?; + let proposed = self + .conn + .prepare( + "SELECT id, job_name, content_hash, snapshot_hash, file_identity, path, + schedules_json, backend, timeout_ms, workdir, + delivery_channel, delivery_target + FROM job_schedule_reviews + WHERE status = 'proposed' + ORDER BY proposed_at_ms, job_name, id", + )? + .query_map([], map_schedule_review)? + .collect::>>()?; + let mut questions = Vec::new(); + for review in proposed { + let existing = + self.conn + .query_row( + "SELECT q.id, q.channel, q.thread_key, q.sender_key, q.chat_key, + q.target, q.prompt, q.choices_json, q.expires_at_ms, + q.delivery_status + FROM job_schedule_review_questions rq + JOIN approval_questions q ON q.id = rq.question_id + WHERE rq.review_id = ?1 + AND q.status IN ('pending', 'answered') + AND q.expires_at_ms > ?2 + ORDER BY rq.created_at_ms DESC LIMIT 1", + params![review.id, now], + |row| { + Ok(( + Question { + id: row.get(0)?, + channel: row.get(1)?, + thread_key: row.get(2)?, + sender_key: row.get(3)?, + chat_key: row.get(4)?, + target: row.get(5)?, + prompt: row.get(6)?, + choices: serde_json::from_str(&row.get::<_, String>(7)?) + .map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 7, + rusqlite::types::Type::Text, + Box::new(error), + ) + })?, + expires_at_ms: row.get(8)?, + }, + row.get::<_, String>(9)?, + )) + }, + ) + .optional()?; + if let Some((question, delivery_status)) = existing { + if question.channel == origin.channel + && question.thread_key == origin.thread_key + && question.sender_key == origin.sender_key + && question.chat_key == origin.chat_key + && delivery_status != "delivered" + { + questions.push(question); + } + continue; + } + + let question = Question::new( + origin.clone(), + target, + schedule_review_prompt(&review)?, + vec![ + Choice { + label: "Approve this exact schedule revision".to_string(), + value: "approve".to_string(), + }, + Choice { + label: "Reject this schedule revision".to_string(), + value: "reject".to_string(), + }, + ], + now.saturating_add(SCHEDULE_REVIEW_TTL_MS), + )?; + let choices = serde_json::to_string(&question.choices)?; + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + if !insert_schedule_question(&tx, &review, &question, &choices, now)? { + tx.rollback()?; + continue; + } + tx.commit()?; + questions.push(question); + } + Ok(questions) + } + + pub fn mark_schedule_question_delivery( + &mut self, + id: &str, + status: ApprovalDeliveryStatus, + now: i64, + ) -> Result<()> { + let changed = self.conn.execute( + "UPDATE approval_questions + SET delivery_status = ?2, updated_at_ms = ?3 + WHERE id = ?1 + AND EXISTS ( + SELECT 1 FROM job_schedule_review_questions + WHERE question_id = ?1 + )", + params![id, status.as_str(), now], + )?; + if changed != 1 { + bail!("schedule review question {id:?} does not exist"); + } + Ok(()) + } + + pub fn resolve_schedule_answer( + &mut self, + cfg: &Config, + question_id: &str, + reviewer: &str, + now: i64, + ) -> Result { + let row = self + .conn + .query_row( + "SELECT r.id, r.job_name, r.content_hash, r.snapshot_hash, + r.file_identity, r.path, r.schedules_json, r.backend, + r.timeout_ms, r.workdir, r.delivery_channel, + r.delivery_target, r.status, q.choices_json, q.answer_index, + q.status + FROM job_schedule_review_questions rq + JOIN job_schedule_reviews r ON r.id = rq.review_id + JOIN approval_questions q ON q.id = rq.question_id + WHERE rq.question_id = ?1", + [question_id], + |row| { + Ok(( + ScheduleReview { + id: row.get(0)?, + job_name: row.get(1)?, + content_hash: row.get(2)?, + snapshot_hash: row.get(3)?, + file_identity: row.get(4)?, + path: row.get(5)?, + schedules_json: row.get(6)?, + backend: row.get(7)?, + timeout_ms: row.get(8)?, + workdir: row.get(9)?, + delivery_channel: row.get(10)?, + delivery_target: row.get(11)?, + }, + row.get::<_, String>(12)?, + row.get::<_, String>(13)?, + row.get::<_, Option>(14)?, + row.get::<_, String>(15)?, + )) + }, + ) + .optional()?; + let Some((review, status, choices, answer_index, question_status)) = row else { + return Ok(ScheduleDecision::NotScheduleReview); + }; + if status != "proposed" { + return Ok(ScheduleDecision::AlreadyHandled); + } + if question_status != "answered" { + return Ok(ScheduleDecision::AlreadyHandled); + } + let choices: Vec = serde_json::from_str(&choices)?; + let answer = answer_index + .and_then(|value| value.checked_sub(1)) + .and_then(|index| choices.get(index as usize)) + .context("stored schedule review answer is invalid")?; + if answer.value == "reject" { + let Some(event) = + self.finish_schedule_review(&review, "rejected", reviewer, None, question_id, now)? + else { + return Ok(ScheduleDecision::AlreadyHandled); + }; + return Ok(ScheduleDecision::Rejected { + job_name: review.job_name, + content_hash: review.content_hash, + event, + }); + } + if answer.value != "approve" { + bail!("unsupported schedule review answer {:?}", answer.value); + } + + let current = Catalog::load_named(cfg, &review.job_name) + .and_then(|job| { + let candidate = + schedule_review(&job, &review.delivery_channel, &review.delivery_target)?; + if candidate.id != review.id + || job.content_hash != review.content_hash + || job.snapshot_hash != review.snapshot_hash + || job.file_identity != review.file_identity + || job.path.to_string_lossy() != review.path + { + bail!("job changed after its schedule was presented"); + } + Ok(()) + }) + .context("revalidate exact schedule revision"); + if let Err(error) = current { + let reason = format!("{error:#}"); + let event = self.finish_schedule_review( + &review, + "invalidated", + reviewer, + Some(&reason), + question_id, + now, + )?; + let Some(event) = event else { + return Ok(ScheduleDecision::AlreadyHandled); + }; + return Ok(ScheduleDecision::Invalidated { + job_name: review.job_name, + content_hash: review.content_hash, + reason, + event: Some(event), + }); + } + let Some(event) = + self.finish_schedule_review(&review, "approved", reviewer, None, question_id, now)? + else { + return Ok(ScheduleDecision::AlreadyHandled); + }; + Ok(ScheduleDecision::Approved { + job_name: review.job_name, + content_hash: review.content_hash, + event, + }) + } + + pub fn recover_answered_schedule_reviews( + &mut self, + cfg: &Config, + now: i64, + ) -> Result> { + let answered = self + .conn + .prepare( + "SELECT q.id, q.channel, q.thread_key, q.sender_key, q.chat_key + FROM job_schedule_review_questions rq + JOIN approval_questions q ON q.id = rq.question_id + JOIN job_schedule_reviews r ON r.id = rq.review_id + WHERE q.status = 'answered' AND r.status = 'proposed' + ORDER BY q.answered_at_ms, q.id", + )? + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + row.get::<_, String>(3)?, + row.get::<_, String>(4)?, + )) + })? + .collect::>>()?; + let mut events = Vec::new(); + for (id, channel, thread, sender, chat) in answered { + let reviewer = format!("channel={channel} thread={thread} sender={sender} chat={chat}"); + match self.resolve_schedule_answer(cfg, &id, &reviewer, now)? { + ScheduleDecision::Approved { event, .. } + | ScheduleDecision::Rejected { event, .. } => events.push(event), + ScheduleDecision::Invalidated { + event: Some(event), .. + } => events.push(event), + ScheduleDecision::Invalidated { event: None, .. } + | ScheduleDecision::AlreadyHandled + | ScheduleDecision::NotScheduleReview => {} + } + } + Ok(events) + } + + fn finish_schedule_review( + &mut self, + review: &ScheduleReview, + status: &str, + reviewer: &str, + reason: Option<&str>, + question_id: &str, + now: i64, + ) -> Result> { + if !matches!(status, "approved" | "rejected" | "invalidated") { + bail!("invalid schedule review decision {status:?}"); + } + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed = tx.execute( + "UPDATE job_schedule_reviews + SET status = ?2, decided_at_ms = ?3, reviewed_by = ?4, reason = ?5, + invalidated_at_ms = CASE WHEN ?2 = 'invalidated' THEN ?3 + ELSE invalidated_at_ms END + WHERE id = ?1 AND status = 'proposed'", + params![review.id, status, now, reviewer, reason], + )?; + if changed != 1 { + tx.commit()?; + return Ok(None); + } + tx.execute( + "UPDATE approval_questions + SET status = 'consumed', consumed_at_ms = ?2, updated_at_ms = ?2 + WHERE id = ?1 AND status = 'answered'", + params![question_id, now], + )?; + let mut events = Vec::new(); + push_schedule_event( + &tx, + &mut events, + review, + status, + Some(reviewer), + reason, + now, + )?; + tx.commit()?; + Ok(Some(events.remove(0))) + } + pub fn start_manual(&mut self, cfg: &Config, job: &Job) -> Result { let now = now_ms(); let Some(lock) = JobLock::try_acquire(&cfg.paths.jobs_run, &job.name)? else { @@ -900,6 +1575,85 @@ impl Ledger { Ok(rows) } + pub fn schedule_reviews(&self, name: Option<&str>) -> Result> { + let mut statement = self.conn.prepare( + "SELECT id, job_name, content_hash, status, schedules_json, backend, + timeout_ms, workdir, delivery_channel, delivery_target, + reviewed_by, reason + FROM job_schedule_reviews + WHERE (?1 IS NULL OR job_name = ?1) + ORDER BY proposed_at_ms DESC, id DESC LIMIT 100", + )?; + let rows = statement + .query_map([name], |row| { + let schedules_json = row.get::<_, String>(4)?; + let schedules = serde_json::from_str(&schedules_json).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 4, + rusqlite::types::Type::Text, + Box::new(error), + ) + })?; + Ok(ScheduleReviewRow { + review_id: row.get(0)?, + job_name: row.get(1)?, + content_hash: row.get(2)?, + status: row.get(3)?, + schedules, + backend: row.get(5)?, + timeout_ms: row.get(6)?, + workdir: row.get(7)?, + delivery_channel: row.get(8)?, + delivery_target: row.get(9)?, + reviewed_by: row.get(10)?, + reason: row.get(11)?, + }) + })? + .collect::>>() + .map_err(anyhow::Error::from)?; + Ok(rows) + } + + pub(crate) fn pending_schedule_audit_events( + &self, + limit: usize, + ) -> Result> { + let mut statement = self.conn.prepare( + "SELECT id, audit_event_id, event, job_name, content_hash, review_id, + actor, reason, created_at_ms + FROM job_schedule_events + WHERE audit_logged_at_ms IS NULL + ORDER BY id + LIMIT ?1", + )?; + let rows = statement + .query_map([limit as i64], |row| { + Ok(ScheduleReviewEvent { + id: row.get(0)?, + audit_event_id: row.get(1)?, + event: row.get(2)?, + job_name: row.get(3)?, + content_hash: row.get(4)?, + review_id: row.get(5)?, + actor: row.get(6)?, + reason: row.get(7)?, + created_at_ms: row.get(8)?, + }) + })? + .collect::>>()?; + Ok(rows) + } + + pub(crate) fn mark_schedule_audit_logged(&mut self, id: i64, now: i64) -> Result<()> { + self.conn.execute( + "UPDATE job_schedule_events + SET audit_logged_at_ms = ?2 + WHERE id = ?1 AND audit_logged_at_ms IS NULL", + params![id, now], + )?; + Ok(()) + } + pub fn enqueue_scheduled( &mut self, job: &Job, @@ -960,7 +1714,8 @@ impl Ledger { pub fn queued_runs(&self, limit: usize) -> Result> { let mut statement = self.conn.prepare( - "SELECT id, job_name, snapshot_hash, trigger_id + "SELECT id, job_name, snapshot_hash, trigger_id, + delivery_channel, delivery_target FROM job_runs WHERE state = 'queued' ORDER BY scheduled_at_ms, queued_at_ms LIMIT ?1", )?; @@ -971,6 +1726,8 @@ impl Ledger { job_name: row.get(1)?, snapshot_hash: row.get(2)?, trigger_id: row.get(3)?, + delivery_channel: row.get(4)?, + delivery_target: row.get(5)?, }) })? .collect::>>()?; @@ -1013,9 +1770,40 @@ impl Ledger { return Ok(None); } }; + let activation_id = queued + .delivery_channel + .as_deref() + .zip(queued.delivery_target.as_deref()) + .and_then(|(channel, target)| schedule_review(&job, channel, target).ok()) + .map(|review| review.id); let tx = self .conn .transaction_with_behavior(TransactionBehavior::Immediate)?; + let activated = activation_id + .as_deref() + .map(|id| { + tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM job_schedule_reviews + WHERE id = ?1 AND status = 'activated' + )", + [id], + |row| row.get::<_, bool>(0), + ) + }) + .transpose()? + .unwrap_or(false); + if !activated { + tx.execute( + "UPDATE job_runs SET state = 'cancelled', finished_at_ms = ?2, + error = 'schedule activation changed before execution', + delivery_state = 'pending' + WHERE id = ?1 AND state = 'queued'", + params![queued.id, now], + )?; + tx.commit()?; + return Ok(None); + } tx.execute( "UPDATE job_runs SET state = CASE WHEN result IS NOT NULL AND evaluation_state = 'running' @@ -1283,37 +2071,305 @@ fn duration_ms(duration: Duration) -> i64 { duration.as_millis().min(i64::MAX as u128) as i64 } -fn delivery_backoff_ms(attempts: i64) -> i64 { - match attempts { - 0 => 0, - 1 => 30_000, - 2 => 2 * 60_000, - 3 => 10 * 60_000, - _ => 30 * 60_000, - } +fn schedule_review( + job: &Job, + delivery_channel: &str, + delivery_target: &str, +) -> Result { + let schedules = job + .triggers + .iter() + .filter(|trigger| trigger.enabled) + .cloned() + .collect::>(); + if schedules.is_empty() { + bail!("schedule review requires at least one enabled trigger"); + } + let schedules_json = serde_json::to_string(&schedules)?; + let path = job.path.to_string_lossy().to_string(); + let workdir = job.workdir.to_string_lossy().to_string(); + let timeout_ms = duration_ms(job.timeout); + let mut fingerprint = Sha256::new(); + for value in [ + job.name.as_str(), + job.content_hash.as_str(), + job.snapshot_hash.as_str(), + job.file_identity.as_str(), + path.as_str(), + schedules_json.as_str(), + job.backend.as_str(), + &timeout_ms.to_string(), + workdir.as_str(), + delivery_channel, + delivery_target, + ] { + fingerprint.update(value.as_bytes()); + fingerprint.update(b"\0"); + } + Ok(ScheduleReview { + id: format!("{:x}", fingerprint.finalize()), + job_name: job.name.clone(), + content_hash: job.content_hash.clone(), + snapshot_hash: job.snapshot_hash.clone(), + file_identity: job.file_identity.clone(), + path, + schedules_json, + backend: job.backend.as_str().to_string(), + timeout_ms, + workdir, + delivery_channel: delivery_channel.to_string(), + delivery_target: delivery_target.to_string(), + }) } -#[derive(Clone)] -struct NextOccurrence { - schedule: String, - timezone: String, - snapshot_hash: String, - at_ms: Option, +fn schedule_migration_destination(cfg: &Config) -> Result<(String, String)> { + let configured = cfg + .primary_delivery + .as_ref() + .context("primary delivery is not configured")?; + let kind = crate::config::ChannelKind::parse(&configured.channel) + .context("invalid primary delivery channel")?; + if !cfg.enabled_channel_kinds()?.contains(&kind) { + bail!("primary delivery channel is not enabled"); + } + let channel = crate::channel::Channel::new_for(cfg, kind)?; + let target = channel.primary_target(&configured.target)?; + Ok((kind.as_str().to_string(), target)) } -pub struct Scheduler { - cfg: Config, - delivery_channel: String, - delivery_target: String, - next: HashMap<(String, String), NextOccurrence>, - workers: JoinSet>, - delivery_workers: JoinSet>, - delivery_owner: String, - validation_errors: HashMap, - validation_initialized: bool, - scheduling_enabled: bool, - ledger: Option, -} +fn insert_schedule_review( + tx: &Transaction<'_>, + review: &ScheduleReview, + status: &str, + now: i64, +) -> Result { + Ok(tx.execute( + "INSERT INTO job_schedule_reviews ( + id, job_name, content_hash, snapshot_hash, file_identity, path, + schedules_json, backend, timeout_ms, workdir, delivery_channel, + delivery_target, status, proposed_at_ms, decided_at_ms, activated_at_ms, + reviewed_by, reason + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, + ?13, ?14, + CASE WHEN ?13 = 'activated' THEN ?14 ELSE NULL END, + CASE WHEN ?13 = 'activated' THEN ?14 ELSE NULL END, + CASE WHEN ?13 = 'activated' THEN 'migration' ELSE NULL END, + CASE WHEN ?13 = 'activated' + THEN 'preserved existing installed schedule' ELSE NULL END) + ON CONFLICT(id) DO NOTHING", + params![ + review.id, + review.job_name, + review.content_hash, + review.snapshot_hash, + review.file_identity, + review.path, + review.schedules_json, + review.backend, + review.timeout_ms, + review.workdir, + review.delivery_channel, + review.delivery_target, + status, + now, + ], + )? == 1) +} + +fn insert_schedule_question( + tx: &Transaction<'_>, + review: &ScheduleReview, + question: &Question, + choices_json: &str, + now: i64, +) -> Result { + let proposed = tx.query_row( + "SELECT EXISTS( + SELECT 1 FROM job_schedule_reviews + WHERE id = ?1 AND status = 'proposed' + )", + [&review.id], + |row| row.get::<_, bool>(0), + )?; + if !proposed { + return Ok(false); + } + let claimed = tx.query_row( + "SELECT EXISTS( + SELECT 1 + FROM job_schedule_review_questions rq + JOIN approval_questions q ON q.id = rq.question_id + WHERE rq.review_id = ?1 + AND q.status IN ('pending', 'answered') + AND q.expires_at_ms > ?2 + )", + params![review.id, now], + |row| row.get::<_, bool>(0), + )?; + if claimed { + return Ok(false); + } + tx.execute( + "INSERT INTO approval_questions ( + id, channel, thread_key, sender_key, chat_key, target, + prompt, choices_json, expires_at_ms, status, delivery_status, + created_at_ms, updated_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, + 'pending', 'pending', ?10, ?10)", + params![ + question.id, + question.channel, + question.thread_key, + question.sender_key, + question.chat_key, + question.target, + question.prompt, + choices_json, + question.expires_at_ms, + now, + ], + )?; + tx.execute( + "INSERT INTO job_schedule_review_questions ( + question_id, review_id, created_at_ms + ) VALUES (?1, ?2, ?3)", + params![question.id, review.id, now], + )?; + Ok(true) +} + +fn push_schedule_event( + tx: &Transaction<'_>, + events: &mut Vec, + review: &ScheduleReview, + event: &str, + actor: Option<&str>, + reason: Option<&str>, + now: i64, +) -> Result<()> { + tx.execute( + "INSERT INTO job_schedule_events ( + audit_event_id, review_id, job_name, content_hash, event, actor, + reason, created_at_ms + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", + params![ + Uuid::new_v4().to_string(), + review.id, + review.job_name, + review.content_hash, + event, + actor, + reason, + now + ], + )?; + let audit_event_id = tx.query_row( + "SELECT audit_event_id FROM job_schedule_events WHERE id = last_insert_rowid()", + [], + |row| row.get(0), + )?; + events.push(ScheduleReviewEvent { + id: tx.last_insert_rowid(), + audit_event_id, + event: event.to_string(), + job_name: review.job_name.clone(), + content_hash: review.content_hash.clone(), + review_id: review.id.clone(), + actor: actor.map(str::to_string), + reason: reason.map(str::to_string), + created_at_ms: now, + }); + Ok(()) +} + +fn cancel_review_questions(tx: &Transaction<'_>, review_id: &str, now: i64) -> Result<()> { + tx.execute( + "UPDATE approval_questions + SET status = 'cancelled', updated_at_ms = ?2 + WHERE id IN ( + SELECT question_id FROM job_schedule_review_questions WHERE review_id = ?1 + ) AND status IN ('pending', 'answered')", + params![review_id, now], + )?; + Ok(()) +} + +fn map_schedule_review(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(ScheduleReview { + id: row.get(0)?, + job_name: row.get(1)?, + content_hash: row.get(2)?, + snapshot_hash: row.get(3)?, + file_identity: row.get(4)?, + path: row.get(5)?, + schedules_json: row.get(6)?, + backend: row.get(7)?, + timeout_ms: row.get(8)?, + workdir: row.get(9)?, + delivery_channel: row.get(10)?, + delivery_target: row.get(11)?, + }) +} + +fn schedule_review_prompt(review: &ScheduleReview) -> Result { + let schedules: Vec = serde_json::from_str(&review.schedules_json)?; + let schedules = schedules + .iter() + .map(|trigger| { + format!( + "- {}: {:?} in {}", + trigger.id, trigger.schedule, trigger.timezone + ) + }) + .collect::>() + .join("\n"); + Ok(format!( + "Review schedule activation\n\nJob: {}\nContent revision: {}\nSchedules:\n{}\nBackend: {}\nTimeout: {}\nWork directory: {}\nDelivery target: {}:{}\n\nThe Markdown file is already saved and remains available for validation and manual inspection. Approval activates only this exact validated revision and effective schedule configuration.", + review.job_name, + review.content_hash, + schedules, + review.backend, + humantime::format_duration(Duration::from_millis( + review.timeout_ms.try_into().unwrap_or(u64::MAX) + )), + review.workdir, + review.delivery_channel, + review.delivery_target, + )) +} + +fn delivery_backoff_ms(attempts: i64) -> i64 { + match attempts { + 0 => 0, + 1 => 30_000, + 2 => 2 * 60_000, + 3 => 10 * 60_000, + _ => 30 * 60_000, + } +} + +#[derive(Clone)] +struct NextOccurrence { + schedule: String, + timezone: String, + snapshot_hash: String, + at_ms: Option, +} + +pub struct Scheduler { + cfg: Config, + delivery_channel: String, + delivery_target: String, + next: HashMap<(String, String), NextOccurrence>, + workers: JoinSet>, + delivery_workers: JoinSet>, + delivery_owner: String, + validation_errors: HashMap, + validation_initialized: bool, + scheduling_enabled: bool, + ledger: Option, + schedule_events: Vec, +} impl Scheduler { pub fn new(cfg: Config, delivery_channel: String, delivery_target: String) -> Self { @@ -1329,6 +2385,7 @@ impl Scheduler { validation_initialized: false, scheduling_enabled: true, ledger: None, + schedule_events: Vec::new(), } } @@ -1345,6 +2402,7 @@ impl Scheduler { validation_initialized: false, scheduling_enabled: false, ledger: None, + schedule_events: Vec::new(), } } @@ -1375,11 +2433,32 @@ impl Scheduler { None => Ledger::open(&self.cfg.paths.database)?, }; ledger.recover_stale_runs(&self.cfg, now)?; + self.schedule_events + .extend(ledger.recover_answered_schedule_reviews(&self.cfg, now)?); let catalog = Catalog::load(&self.cfg)?; self.report_catalog_errors(&catalog); + let active_reviews = if self.scheduling_enabled { + let (active, events) = ledger.reconcile_schedule_reviews( + &catalog, + &self.delivery_channel, + &self.delivery_target, + now, + )?; + self.schedule_events.extend(events); + active + } else { + HashSet::new() + }; let mut seen = HashSet::new(); for job in catalog.jobs.values().filter(|_| self.scheduling_enabled) { + let review = schedule_review(job, &self.delivery_channel, &self.delivery_target).ok(); + if review + .as_ref() + .is_none_or(|review| !active_reviews.contains(&review.id)) + { + continue; + } for trigger in job.triggers.iter().filter(|trigger| trigger.enabled) { let key = (job.name.clone(), trigger.id.clone()); seen.insert(key.clone()); @@ -1484,6 +2563,10 @@ impl Scheduler { Ok(()) } + pub fn take_schedule_events(&mut self) -> Vec { + std::mem::take(&mut self.schedule_events) + } + pub async fn shutdown(&mut self) { self.shutdown_with_grace(SCHEDULER_SHUTDOWN_GRACE).await; } @@ -2091,6 +3174,7 @@ pub fn format_job(job: &Job) -> String { #[cfg(test)] mod tests { use super::*; + use crate::audit::AuditLog; use crate::test_support::{sh_arg, temp_dir, temp_path, FakeCli}; use std::io::Write; use std::sync::{Arc, Mutex}; @@ -2138,6 +3222,15 @@ mod tests { cfg.jobs_dir = jobs_dir.to_string_lossy().to_string(); cfg.paths.database = database.to_path_buf(); cfg.paths.jobs_run = run_dir.to_path_buf(); + cfg.channel = "telegram".to_string(); + cfg.self_handles.clear(); + cfg.allow_from.clear(); + cfg.telegram_bot_token = Some("test-token".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + cfg.primary_delivery = Some(crate::config::PrimaryDeliveryConfig { + channel: "telegram".to_string(), + target: "7".to_string(), + }); cfg } @@ -2173,6 +3266,86 @@ mod tests { ) } + fn review_origin() -> AnswerOrigin { + AnswerOrigin { + channel: "telegram".to_string(), + thread_key: "telegram:dm:7".to_string(), + sender_key: "7".to_string(), + chat_key: "7".to_string(), + } + } + + fn preserve_existing_schedules(cfg: &Config) { + let ledger = Ledger::open(&cfg.paths.database).unwrap(); + ledger + .conn + .execute_batch( + "UPDATE job_schedule_meta SET value = 'pending' + WHERE key = 'legacy_schedule_migration'; + UPDATE job_schedule_meta SET value = 'unclaimed' + WHERE key = 'legacy_schedule_baseline'; + DELETE FROM job_schedule_legacy_baseline;", + ) + .unwrap(); + drop(ledger); + Ledger::capture_legacy_schedule_baseline(cfg).unwrap(); + } + + fn activate_existing_schedules(cfg: &Config) { + preserve_existing_schedules(cfg); + let catalog = Catalog::load(cfg).unwrap(); + let enabled = catalog + .jobs + .values() + .filter(|job| job.triggers.iter().any(|trigger| trigger.enabled)) + .count(); + let (active, _) = Ledger::open(&cfg.paths.database) + .unwrap() + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + assert_eq!(active.len(), enabled); + } + + fn propose_schedule(cfg: &Config, ledger: &mut Ledger, now: i64) -> (Catalog, Question) { + let catalog = Catalog::load(cfg).unwrap(); + let (active, events) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", now) + .unwrap(); + assert!(active.is_empty()); + assert!(events.iter().any(|event| event.event == "proposed")); + let questions = ledger + .schedule_review_questions(&review_origin(), "7", now) + .unwrap(); + assert_eq!(questions.len(), 1); + (catalog, questions.into_iter().next().unwrap()) + } + + fn answer_schedule( + cfg: &Config, + ledger: &mut Ledger, + question: &Question, + answer: usize, + now: i64, + ) -> ScheduleDecision { + let mut history = History::open(&cfg.paths.database).unwrap(); + let outcome = history + .answer_question(&review_origin(), &format!("{} {answer}", question.id), now) + .unwrap(); + assert!(matches!( + outcome, + crate::approval::AnswerOutcome::Selected(_) + )); + drop(history); + ledger + .resolve_schedule_answer( + cfg, + &question.id, + "channel=telegram thread=telegram:dm:7 sender=7 chat=7", + now, + ) + .unwrap() + } + async fn delivery_ok( _channel: String, _target: String, @@ -2895,6 +4068,7 @@ mod tests { write_job(&jobs_dir, "delivery", &scheduled_job(&workdir, false)); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = slow.bin(); + activate_existing_schedules(&cfg); let catalog = Catalog::load(&cfg).unwrap(); let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); let execution_id = ledger @@ -3070,6 +4244,7 @@ mod tests { .errors .iter() .all(|error| error.message.contains("no possible calendar date"))); + preserve_existing_schedules(&cfg); let mut scheduler = Scheduler::new(cfg, "telegram".into(), "7".into()); tokio::time::timeout(Duration::from_millis(100), scheduler.tick(0, delivery_ok)) @@ -3097,6 +4272,753 @@ mod tests { validate_triggers(&[trigger]).unwrap(); } + #[tokio::test] + async fn new_enabled_schedule_stays_unplanned_until_exact_review() { + let jobs_dir = temp_dir("schedule-review-pending-jobs"); + let workdir = temp_dir("schedule-review-pending-work"); + let database = temp_path("schedule-review-pending-db"); + let run_dir = temp_dir("schedule-review-pending-run"); + write_job(&jobs_dir, "pending", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut scheduler = Scheduler::new(cfg.clone(), "telegram".into(), "7".into()); + + scheduler.tick(0, delivery_ok).await.unwrap(); + scheduler.tick(60_000, delivery_ok).await.unwrap(); + + assert!(scheduler.next.is_empty()); + assert!(Ledger::open(&cfg.paths.database) + .unwrap() + .runs(Some("pending")) + .unwrap() + .is_empty()); + let events = scheduler.take_schedule_events(); + assert_eq!( + events + .iter() + .map(|event| event.event.as_str()) + .collect::>(), + vec!["proposed"] + ); + } + + #[test] + fn disabled_schedule_remains_manually_claimable_without_review() { + let jobs_dir = temp_dir("schedule-review-disabled-jobs"); + let workdir = temp_dir("schedule-review-disabled-work"); + let database = temp_path("schedule-review-disabled-db"); + let run_dir = temp_dir("schedule-review-disabled-run"); + write_job(&jobs_dir, "disabled", &scheduled_job(&workdir, false)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let catalog = Catalog::load(&cfg).unwrap(); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + + let (active, events) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + assert!(active.is_empty()); + assert!(events.is_empty()); + assert!(matches!( + ledger + .start_manual(&cfg, &catalog.jobs["disabled"]) + .unwrap(), + StartOutcome::Claimed { .. } + )); + } + + #[test] + fn schedule_question_binds_channel_owner_and_rejects_duplicate_approval() { + let jobs_dir = temp_dir("schedule-review-owner-jobs"); + let workdir = temp_dir("schedule-review-owner-work"); + let database = temp_path("schedule-review-owner-db"); + let run_dir = temp_dir("schedule-review-owner-run"); + write_job(&jobs_dir, "owner-bound", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + assert!(question.prompt.contains("Job: owner-bound")); + assert!(question.prompt.contains("Backend: codex")); + assert!(question.prompt.contains("Delivery target: telegram:7")); + assert!(question.prompt.contains("every-minute")); + let reviews = ledger.schedule_reviews(Some("owner-bound")).unwrap(); + assert_eq!(reviews.len(), 1); + assert_eq!(reviews[0].status, "proposed"); + assert_eq!(reviews[0].delivery_target, "7"); + let mut competing = Ledger::open(&cfg.paths.database).unwrap(); + let competing_origin = AnswerOrigin { + channel: "imessage".to_string(), + thread_key: "imessage:self:me".to_string(), + sender_key: "me".to_string(), + chat_key: "me".to_string(), + }; + assert!(competing + .schedule_review_questions(&competing_origin, "me", 1_050) + .unwrap() + .is_empty()); + + let mut history = History::open(&cfg.paths.database).unwrap(); + let wrong_channel = AnswerOrigin { + channel: "imessage".to_string(), + thread_key: "imessage:self:7".to_string(), + sender_key: "7".to_string(), + chat_key: "7".to_string(), + }; + assert!(matches!( + history + .answer_question(&wrong_channel, &format!("{} 1", question.id), 1_100) + .unwrap(), + crate::approval::AnswerOutcome::Mismatched(_) + )); + let wrong_sender = AnswerOrigin { + sender_key: "8".to_string(), + ..review_origin() + }; + assert!(matches!( + history + .answer_question(&wrong_sender, &format!("{} 1", question.id), 1_200) + .unwrap(), + crate::approval::AnswerOutcome::Mismatched(_) + )); + assert!(matches!( + history + .answer_question(&review_origin(), &format!("{} 1", question.id), 1_300) + .unwrap(), + crate::approval::AnswerOutcome::Selected(_) + )); + drop(history); + assert!(matches!( + ledger + .resolve_schedule_answer(&cfg, &question.id, "owner", 1_300) + .unwrap(), + ScheduleDecision::Approved { .. } + )); + + let mut history = History::open(&cfg.paths.database).unwrap(); + assert!(matches!( + history + .answer_question(&review_origin(), &format!("{} 1", question.id), 1_400) + .unwrap(), + crate::approval::AnswerOutcome::Duplicate(_) + )); + assert_eq!( + ledger + .resolve_schedule_answer(&cfg, &question.id, "owner", 1_400) + .unwrap(), + ScheduleDecision::AlreadyHandled + ); + } + + #[test] + fn concurrent_channels_cannot_adopt_the_same_schedule_review() { + let jobs_dir = temp_dir("schedule-review-owner-race-jobs"); + let workdir = temp_dir("schedule-review-owner-race-work"); + let database = temp_path("schedule-review-owner-race-db"); + let run_dir = temp_dir("schedule-review-owner-race-run"); + write_job(&jobs_dir, "owner-race", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let catalog = Catalog::load(&cfg).unwrap(); + Ledger::open(&database) + .unwrap() + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + + let origins = [ + (review_origin(), "7".to_string()), + ( + AnswerOrigin { + channel: "imessage".to_string(), + thread_key: "imessage:self:me".to_string(), + sender_key: "me".to_string(), + chat_key: "me".to_string(), + }, + "me".to_string(), + ), + ]; + let barrier = std::sync::Arc::new(std::sync::Barrier::new(origins.len())); + let handles = origins + .into_iter() + .map(|(origin, target)| { + let database = database.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + let mut ledger = Ledger::open(database).unwrap(); + barrier.wait(); + let questions = ledger + .schedule_review_questions(&origin, &target, 1_100) + .unwrap(); + (origin, questions) + }) + }) + .collect::>(); + let results = handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect::>(); + + assert_eq!( + results + .iter() + .map(|(_, questions)| questions.len()) + .sum::(), + 1 + ); + let (owner, question) = results + .iter() + .find_map(|(origin, questions)| questions.first().map(|question| (origin, question))) + .unwrap(); + let competing = results + .iter() + .find(|(origin, _)| origin != owner) + .map(|(origin, _)| origin) + .unwrap(); + assert_eq!(question.channel, owner.channel); + assert_eq!(question.thread_key, owner.thread_key); + assert_eq!(question.sender_key, owner.sender_key); + assert_eq!(question.chat_key, owner.chat_key); + + let mut history = History::open(&database).unwrap(); + assert!(matches!( + history + .answer_question(competing, &format!("{} 1", question.id), 1_200) + .unwrap(), + crate::approval::AnswerOutcome::Mismatched(_) + )); + assert!(matches!( + history + .answer_question(owner, &format!("{} 1", question.id), 1_300) + .unwrap(), + crate::approval::AnswerOutcome::Selected(_) + )); + } + + #[test] + fn expired_schedule_answer_cannot_select_reissued_question_without_correlation() { + let jobs_dir = temp_dir("schedule-review-expired-answer-jobs"); + let workdir = temp_dir("schedule-review-expired-answer-work"); + let database = temp_path("schedule-review-expired-answer-db"); + let run_dir = temp_dir("schedule-review-expired-answer-run"); + write_job(&jobs_dir, "expired-answer", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, first) = propose_schedule(&cfg, &mut ledger, 1_000); + let reissued_at = first.expires_at_ms; + let second = ledger + .schedule_review_questions(&review_origin(), "7", reissued_at) + .unwrap() + .remove(0); + assert_ne!(first.id, second.id); + + let mut history = History::open(&cfg.paths.database).unwrap(); + assert!(matches!( + history + .answer_question(&review_origin(), "1", reissued_at + 1) + .unwrap(), + crate::approval::AnswerOutcome::Expired(id) if id == first.id + )); + assert_eq!( + history.question_state(&second.id, reissued_at + 1).unwrap(), + Some(crate::approval::QuestionState::Pending) + ); + assert!(matches!( + history + .answer_question( + &review_origin(), + &format!("{} 1", second.id), + reissued_at + 2, + ) + .unwrap(), + crate::approval::AnswerOutcome::Selected(_) + )); + } + + #[test] + fn invalidated_review_cannot_gain_a_question_from_a_stale_selection() { + let jobs_dir = temp_dir("schedule-review-question-race-jobs"); + let workdir = temp_dir("schedule-review-question-race-work"); + let database = temp_path("schedule-review-question-race-db"); + let run_dir = temp_dir("schedule-review-question-race-run"); + let original = scheduled_job(&workdir, true); + write_job(&jobs_dir, "question-race", &original); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let catalog = Catalog::load(&cfg).unwrap(); + ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + let review = schedule_review(&catalog.jobs["question-race"], "telegram", "7").unwrap(); + + write_job( + &jobs_dir, + "question-race", + &original.replace("Run once.", "Run changed."), + ); + let changed = Catalog::load(&cfg).unwrap(); + ledger + .reconcile_schedule_reviews(&changed, "telegram", "7", 1_100) + .unwrap(); + let question = Question::new( + review_origin(), + "7", + schedule_review_prompt(&review).unwrap(), + vec![ + Choice { + label: "Approve".to_string(), + value: "approve".to_string(), + }, + Choice { + label: "Reject".to_string(), + value: "reject".to_string(), + }, + ], + 2_000, + ) + .unwrap(); + let choices = serde_json::to_string(&question.choices).unwrap(); + let tx = ledger + .conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .unwrap(); + assert!(!insert_schedule_question(&tx, &review, &question, &choices, 1_200).unwrap()); + tx.commit().unwrap(); + let question_count: i64 = ledger + .conn + .query_row( + "SELECT count(*) FROM job_schedule_review_questions + WHERE review_id = ?1", + [&review.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(question_count, 0); + + write_job(&jobs_dir, "question-race", &original); + let restored = Catalog::load(&cfg).unwrap(); + ledger + .reconcile_schedule_reviews(&restored, "telegram", "7", 1_300) + .unwrap(); + assert!(ledger + .recover_answered_schedule_reviews(&cfg, 1_400) + .unwrap() + .is_empty()); + assert_eq!( + ledger + .schedule_reviews(Some("question-race")) + .unwrap() + .remove(0) + .status, + "proposed" + ); + } + + #[test] + fn revision_race_invalidates_schedule_approval() { + let jobs_dir = temp_dir("schedule-review-race-jobs"); + let workdir = temp_dir("schedule-review-race-work"); + let database = temp_path("schedule-review-race-db"); + let run_dir = temp_dir("schedule-review-race-run"); + write_job(&jobs_dir, "raced", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + write_job( + &jobs_dir, + "raced", + &scheduled_job(&workdir, true).replace("Run once.", "Run changed."), + ); + + let decision = answer_schedule(&cfg, &mut ledger, &question, 1, 1_100); + + assert!(matches!(decision, ScheduleDecision::Invalidated { .. })); + let events = ledger + .conn + .query_row( + "SELECT group_concat(event, ',') FROM job_schedule_events + WHERE job_name = 'raced' ORDER BY id", + [], + |row| row.get::<_, String>(0), + ) + .unwrap(); + assert_eq!(events, "proposed,invalidated"); + } + + #[cfg(unix)] + #[test] + fn same_content_in_place_rewrite_invalidates_schedule_approval() { + let jobs_dir = temp_dir("schedule-review-in-place-answer-jobs"); + let workdir = temp_dir("schedule-review-in-place-answer-work"); + let database = temp_path("schedule-review-in-place-answer-db"); + let run_dir = temp_dir("schedule-review-in-place-answer-run"); + let contents = scheduled_job(&workdir, true); + write_job(&jobs_dir, "in-place-answer", &contents); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (catalog, question) = propose_schedule(&cfg, &mut ledger, 1_000); + let original_identity = catalog.jobs["in-place-answer"].file_identity.clone(); + std::thread::sleep(Duration::from_millis(10)); + write_job(&jobs_dir, "in-place-answer", &contents); + let rewritten = Catalog::load(&cfg).unwrap(); + assert_ne!( + rewritten.jobs["in-place-answer"].file_identity, + original_identity + ); + + assert!(matches!( + answer_schedule(&cfg, &mut ledger, &question, 1, 1_100), + ScheduleDecision::Invalidated { .. } + )); + } + + #[test] + fn schedule_audit_outbox_replays_after_write_failure_and_restart() { + let jobs_dir = temp_dir("schedule-review-audit-outbox-jobs"); + let workdir = temp_dir("schedule-review-audit-outbox-work"); + let database = temp_path("schedule-review-audit-outbox-db"); + let run_dir = temp_dir("schedule-review-audit-outbox-run"); + let audit_path = temp_path("schedule-review-audit-outbox-log"); + write_job(&jobs_dir, "audit-outbox", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let catalog = Catalog::load(&cfg).unwrap(); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + std::fs::create_dir(&audit_path).unwrap(); + let audit = AuditLog::new(&audit_path, false, "scheduler"); + + assert!(audit.flush_schedule_reviews(&mut ledger).is_err()); + drop(ledger); + let mut reopened = Ledger::open(&cfg.paths.database).unwrap(); + assert_eq!( + reopened.pending_schedule_audit_events(100).unwrap().len(), + 1 + ); + + std::fs::remove_dir(&audit_path).unwrap(); + std::fs::write(&audit_path, "{\"partial\":").unwrap(); + assert_eq!(audit.flush_schedule_reviews(&mut reopened).unwrap(), 1); + assert!(reopened + .pending_schedule_audit_events(100) + .unwrap() + .is_empty()); + let contents = std::fs::read_to_string(&audit_path).unwrap(); + let lines = contents.lines().collect::>(); + assert_eq!(lines.len(), 1); + let event: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(event["event"], "schedule_review_proposed"); + let event_id = event["event_id"].as_str().unwrap(); + let uuid = event_id.strip_prefix("job_schedule_event:").unwrap(); + Uuid::parse_str(uuid).unwrap(); + } + + #[cfg(unix)] + #[test] + fn symlink_and_regular_path_replacement_fail_closed() { + use std::os::unix::fs::symlink; + + for replacement in ["regular", "symlink"] { + let jobs_dir = temp_dir(&format!("schedule-review-{replacement}-jobs")); + let workdir = temp_dir(&format!("schedule-review-{replacement}-work")); + let database = temp_path(&format!("schedule-review-{replacement}-db")); + let run_dir = temp_dir(&format!("schedule-review-{replacement}-run")); + let contents = scheduled_job(&workdir, true); + write_job(&jobs_dir, "replaced", &contents); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + let path = jobs_dir.join("replaced.md"); + let replacement_path = jobs_dir.join("replacement"); + std::fs::write(&replacement_path, &contents).unwrap(); + if replacement == "regular" { + std::fs::rename(&replacement_path, &path).unwrap(); + } else { + std::fs::remove_file(&path).unwrap(); + symlink(&replacement_path, &path).unwrap(); + } + + assert!(matches!( + answer_schedule(&cfg, &mut ledger, &question, 1, 1_100), + ScheduleDecision::Invalidated { .. } + )); + } + } + + #[test] + fn rejection_is_terminal_and_records_history() { + let jobs_dir = temp_dir("schedule-review-reject-jobs"); + let workdir = temp_dir("schedule-review-reject-work"); + let database = temp_path("schedule-review-reject-db"); + let run_dir = temp_dir("schedule-review-reject-run"); + write_job(&jobs_dir, "rejected", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + + assert!(matches!( + answer_schedule(&cfg, &mut ledger, &question, 2, 1_100), + ScheduleDecision::Rejected { .. } + )); + let status: String = ledger + .conn + .query_row( + "SELECT status FROM job_schedule_reviews WHERE job_name = 'rejected'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(status, "rejected"); + } + + #[tokio::test] + async fn accepted_review_survives_restart_and_wakes_scheduler() { + let jobs_dir = temp_dir("schedule-review-restart-jobs"); + let workdir = temp_dir("schedule-review-restart-work"); + let database = temp_path("schedule-review-restart-db"); + let run_dir = temp_dir("schedule-review-restart-run"); + write_job(&jobs_dir, "restart-review", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + assert!(matches!( + answer_schedule(&cfg, &mut ledger, &question, 1, 1_100), + ScheduleDecision::Approved { .. } + )); + drop(ledger); + + let mut scheduler = Scheduler::new(cfg.clone(), "telegram".into(), "7".into()); + scheduler.tick(0, delivery_ok).await.unwrap(); + + assert!(scheduler + .next + .contains_key(&("restart-review".to_string(), "every-minute".to_string()))); + assert!(scheduler + .take_schedule_events() + .iter() + .any(|event| event.event == "activated")); + drop(scheduler); + + let mut restarted = Scheduler::new(cfg, "telegram".into(), "7".into()); + restarted.tick(0, delivery_ok).await.unwrap(); + assert!(restarted + .next + .contains_key(&("restart-review".to_string(), "every-minute".to_string()))); + } + + #[test] + fn migration_preserves_only_valid_existing_enabled_schedules() { + let jobs_dir = temp_dir("schedule-review-migration-jobs"); + let workdir = temp_dir("schedule-review-migration-work"); + let database = temp_path("schedule-review-migration-db"); + let run_dir = temp_dir("schedule-review-migration-run"); + write_job(&jobs_dir, "existing", &scheduled_job(&workdir, true)); + write_job(&jobs_dir, "disabled", &scheduled_job(&workdir, false)); + write_job(&jobs_dir, "invalid", "not a runbook"); + let cfg = cfg(&jobs_dir, &database, &run_dir); + preserve_existing_schedules(&cfg); + write_job( + &jobs_dir, + "added-after-migration", + &scheduled_job(&workdir, true), + ); + let catalog = Catalog::load(&cfg).unwrap(); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + + let (active, events) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + + assert_eq!(active.len(), 1); + for expected in [ + ("existing", "approved"), + ("existing", "activated"), + ("added-after-migration", "proposed"), + ] { + assert!(events + .iter() + .any(|event| { (event.job_name.as_str(), event.event.as_str()) == expected })); + } + assert_eq!( + ledger + .schedule_reviews(Some("added-after-migration")) + .unwrap() + .remove(0) + .status, + "proposed" + ); + let migration: String = ledger + .conn + .query_row( + "SELECT value FROM job_schedule_meta + WHERE key = 'legacy_schedule_migration'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(migration, "complete"); + } + + #[test] + fn invalid_edit_removes_an_activated_schedule_before_execution() { + let jobs_dir = temp_dir("schedule-review-invalid-edit-jobs"); + let workdir = temp_dir("schedule-review-invalid-edit-work"); + let database = temp_path("schedule-review-invalid-edit-db"); + let run_dir = temp_dir("schedule-review-invalid-edit-run"); + write_job(&jobs_dir, "invalidated", &scheduled_job(&workdir, true)); + let cfg = cfg(&jobs_dir, &database, &run_dir); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (_, question) = propose_schedule(&cfg, &mut ledger, 1_000); + assert!(matches!( + answer_schedule(&cfg, &mut ledger, &question, 1, 1_100), + ScheduleDecision::Approved { .. } + )); + let catalog = Catalog::load(&cfg).unwrap(); + let (active, _) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_200) + .unwrap(); + assert_eq!(active.len(), 1); + write_job(&jobs_dir, "invalidated", "not a runbook"); + + let invalid_catalog = Catalog::load(&cfg).unwrap(); + let (active, events) = ledger + .reconcile_schedule_reviews(&invalid_catalog, "telegram", "7", 1_300) + .unwrap(); + + assert!(active.is_empty()); + assert!(events.iter().any(|event| event.event == "invalidated")); + } + + #[test] + fn queued_run_is_cancelled_after_effective_backend_invalidates_activation() { + let jobs_dir = temp_dir("schedule-review-queued-backend-jobs"); + let workdir = temp_dir("schedule-review-queued-backend-work"); + let database = temp_path("schedule-review-queued-backend-db"); + let run_dir = temp_dir("schedule-review-queued-backend-run"); + let runbook = scheduled_job(&workdir, true).replace("backend = \"codex\"\n", ""); + write_job(&jobs_dir, "queued-backend", &runbook); + let cfg = cfg(&jobs_dir, &database, &run_dir); + preserve_existing_schedules(&cfg); + let catalog = Catalog::load(&cfg).unwrap(); + let job = &catalog.jobs["queued-backend"]; + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (active, _) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + assert_eq!(active.len(), 1); + ledger + .enqueue_scheduled(job, &job.triggers[0], 60_000, 1_100, "telegram", "7") + .unwrap(); + let queued = ledger.queued_runs(1).unwrap().remove(0); + + let mut changed_cfg = cfg.clone(); + changed_cfg.jobs_agent = Some("claude".to_string()); + let reviews = ledger.schedule_reviews(Some("queued-backend")).unwrap(); + assert_eq!(reviews.len(), 1); + assert_eq!(reviews[0].status, "activated"); + + assert!(ledger + .claim_scheduled(&changed_cfg, &queued, 1_300) + .unwrap() + .is_none()); + let row = ledger.runs(Some("queued-backend")).unwrap().remove(0); + assert_eq!(row.state, "cancelled"); + assert_eq!( + row.error.as_deref(), + Some("schedule activation changed before execution") + ); + } + + #[cfg(unix)] + #[test] + fn queued_run_is_cancelled_after_same_content_path_replacement() { + let jobs_dir = temp_dir("schedule-review-queued-path-jobs"); + let workdir = temp_dir("schedule-review-queued-path-work"); + let database = temp_path("schedule-review-queued-path-db"); + let run_dir = temp_dir("schedule-review-queued-path-run"); + let runbook = scheduled_job(&workdir, true); + write_job(&jobs_dir, "queued-path", &runbook); + let cfg = cfg(&jobs_dir, &database, &run_dir); + preserve_existing_schedules(&cfg); + let catalog = Catalog::load(&cfg).unwrap(); + let job = &catalog.jobs["queued-path"]; + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (active, _) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + assert_eq!(active.len(), 1); + ledger + .enqueue_scheduled(job, &job.triggers[0], 60_000, 1_100, "telegram", "7") + .unwrap(); + let queued = ledger.queued_runs(1).unwrap().remove(0); + + let replacement = jobs_dir.join("replacement.md"); + std::fs::write(&replacement, &runbook).unwrap(); + std::fs::rename(&replacement, jobs_dir.join("queued-path.md")).unwrap(); + let reviews = ledger.schedule_reviews(Some("queued-path")).unwrap(); + assert_eq!(reviews.len(), 1); + assert_eq!(reviews[0].status, "activated"); + + assert!(ledger + .claim_scheduled(&cfg, &queued, 1_300) + .unwrap() + .is_none()); + let row = ledger.runs(Some("queued-path")).unwrap().remove(0); + assert_eq!(row.state, "cancelled"); + assert_eq!( + row.error.as_deref(), + Some("schedule activation changed before execution") + ); + } + + #[cfg(unix)] + #[test] + fn queued_run_is_cancelled_after_same_content_in_place_rewrite() { + let jobs_dir = temp_dir("schedule-review-queued-in-place-jobs"); + let workdir = temp_dir("schedule-review-queued-in-place-work"); + let database = temp_path("schedule-review-queued-in-place-db"); + let run_dir = temp_dir("schedule-review-queued-in-place-run"); + let runbook = scheduled_job(&workdir, true); + write_job(&jobs_dir, "queued-in-place", &runbook); + let cfg = cfg(&jobs_dir, &database, &run_dir); + preserve_existing_schedules(&cfg); + let catalog = Catalog::load(&cfg).unwrap(); + let job = &catalog.jobs["queued-in-place"]; + let original_identity = job.file_identity.clone(); + let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); + let (active, _) = ledger + .reconcile_schedule_reviews(&catalog, "telegram", "7", 1_000) + .unwrap(); + assert_eq!(active.len(), 1); + ledger + .enqueue_scheduled(job, &job.triggers[0], 60_000, 1_100, "telegram", "7") + .unwrap(); + let queued = ledger.queued_runs(1).unwrap().remove(0); + + std::thread::sleep(Duration::from_millis(10)); + write_job(&jobs_dir, "queued-in-place", &runbook); + let rewritten = Catalog::load(&cfg).unwrap(); + assert_ne!( + rewritten.jobs["queued-in-place"].file_identity, + original_identity + ); + assert_eq!( + ledger + .schedule_reviews(Some("queued-in-place")) + .unwrap() + .remove(0) + .status, + "activated" + ); + + assert!(ledger + .claim_scheduled(&cfg, &queued, 1_300) + .unwrap() + .is_none()); + let row = ledger.runs(Some("queued-in-place")).unwrap().remove(0); + assert_eq!(row.state, "cancelled"); + assert_eq!( + row.error.as_deref(), + Some("schedule activation changed before execution") + ); + } + #[tokio::test] async fn scheduler_skips_missed_ticks_and_retries_stored_output_without_rerunning() { let jobs_dir = temp_dir("scheduler-jobs"); @@ -3122,6 +5044,7 @@ printf '%s\n' '{{"type":"thread.started","thread_id":"scheduled"}}' write_job(&jobs_dir, "scheduled", &scheduled_job(&workdir, true)); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = cli.bin(); + preserve_existing_schedules(&cfg); let mut scheduler = Scheduler::new(cfg.clone(), "telegram".into(), "7".into()); let start = Utc .with_ymd_and_hms(2026, 1, 1, 0, 0, 0) @@ -3208,6 +5131,7 @@ printf '%s\n' '{{"type":"thread.started","thread_id":"restart"}}' write_job(&jobs_dir, "restart", &scheduled_job(&workdir, true)); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = cli.bin(); + activate_existing_schedules(&cfg); let job = Catalog::load_named(&cfg, "restart").unwrap(); let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); let first_id = ledger @@ -3310,6 +5234,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"delivery-only"}' ); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = slow.bin(); + activate_existing_schedules(&cfg); let job = Catalog::load_named(&cfg, "timeout").unwrap(); Ledger::open(&cfg.paths.database) .unwrap() @@ -3342,6 +5267,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"delivery-only"}' write_job(&jobs_dir, "failure", &scheduled_job(&workdir, true)); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = failed.bin(); + activate_existing_schedules(&cfg); let job = Catalog::load_named(&cfg, "failure").unwrap(); Ledger::open(&cfg.paths.database) .unwrap() @@ -3389,6 +5315,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"limited"}' let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = cli.bin(); cfg.jobs_max_workers = 1; + activate_existing_schedules(&cfg); let catalog = Catalog::load(&cfg).unwrap(); let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); for job in catalog.jobs.values() { @@ -3443,6 +5370,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"limited"}' let run_dir = temp_dir("scheduled-stale-run"); write_job(&jobs_dir, "stale", &scheduled_job(&workdir, true)); let cfg = cfg(&jobs_dir, &database, &run_dir); + activate_existing_schedules(&cfg); let job = Catalog::load_named(&cfg, "stale").unwrap(); let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); let id = ledger @@ -3484,6 +5412,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"limited"}' "backend = \"codex\"\nevals = [\"quality\"]\n", ), ); + activate_existing_schedules(&cfg); let job = Catalog::load_named(&cfg, "recover-eval").unwrap(); let mut ledger = Ledger::open(&cfg.paths.database).unwrap(); let run_id = ledger @@ -3539,6 +5468,7 @@ printf '%s\n' '{"type":"thread.started","thread_id":"after-crash"}' write_job(&jobs_dir, "cli-live", &scheduled_job(&workdir, true)); let mut cfg = cfg(&jobs_dir, &database, &run_dir); cfg.agent_commands.codex = cli.bin(); + preserve_existing_schedules(&cfg); let start = 1_800_000_000_000i64; let mut before_restart = Scheduler::new(cfg.clone(), "telegram".into(), "7".into()); before_restart.tick(start, delivery_ok).await.unwrap(); diff --git a/src/main.rs b/src/main.rs index 4a50460..2b0a975 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,6 +49,7 @@ Commands: job show Show an installed job job run Run an installed job job runs [name] Show job run history + job reviews [name] Show schedule activation review history Options: --config Use a configuration file (default: $PUSH_HOME/config.toml) @@ -168,7 +169,10 @@ fn load_run_config(path: &str) -> Result { bail!(message); } let expanded_path = util::expand_home(path); - config::Config::load(path).with_context(|| format!("load config {expanded_path}")) + let cfg = config::Config::load(path).with_context(|| format!("load config {expanded_path}"))?; + jobs::Ledger::capture_legacy_schedule_baseline(&cfg) + .context("capture existing schedule migration baseline")?; + Ok(cfg) } fn missing_config_message(path: &str) -> Option { @@ -230,6 +234,7 @@ pub(crate) enum JobCommand { Show(String), Run(String), Runs(Option), + Reviews(Option), } impl Command { @@ -250,6 +255,7 @@ impl Command { | JobCommand::List | JobCommand::Show(_) | JobCommand::Runs(_) + | JobCommand::Reviews(_) ) ) } @@ -332,8 +338,12 @@ impl Args { ["job", "run", name] => Command::Job(JobCommand::Run((*name).to_string())), ["job", "runs"] => Command::Job(JobCommand::Runs(None)), ["job", "runs", name] => Command::Job(JobCommand::Runs(Some((*name).to_string()))), + ["job", "reviews"] => Command::Job(JobCommand::Reviews(None)), + ["job", "reviews", name] => { + Command::Job(JobCommand::Reviews(Some((*name).to_string()))) + } _ => bail!( - "unknown command; expected help, version, init [path], doctor, status, paths, reload, restart, job validate, job list, job show , job run , job runs [], --config , or --json" + "unknown command; expected help, version, init [path], doctor, status, paths, reload, restart, job validate, job list, job show , job run , job runs [], job reviews [], --config , or --json" ), }; Ok(Self { @@ -439,6 +449,38 @@ async fn run_job_command(config_path: &str, command: JobCommand) -> Result<()> { } Ok(()) } + JobCommand::Reviews(name) => { + if let Some(name) = name.as_deref() { + jobs::validate_job_name(name)?; + } + let ledger = jobs::Ledger::open(&cfg.paths.database)?; + for review in ledger.schedule_reviews(name.as_deref())? { + let schedules = review + .schedules + .iter() + .map(|trigger| { + format!("{}:{:?}:{}", trigger.id, trigger.schedule, trigger.timezone) + }) + .collect::>() + .join(","); + println!( + "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}:{}\t{}\t{}", + review.review_id, + review.job_name, + review.status, + review.content_hash, + schedules, + review.backend, + review.timeout_ms, + review.workdir, + review.delivery_channel, + review.delivery_target, + review.reviewed_by.unwrap_or_else(|| "-".to_string()), + review.reason.unwrap_or_else(|| "-".to_string()), + ); + } + Ok(()) + } } } @@ -623,6 +665,12 @@ mod tests { .command, Command::Job(JobCommand::Runs(Some("daily".to_string()))) ); + assert_eq!( + Args::parse(vec!["job".into(), "reviews".into(), "daily".into()]) + .unwrap() + .command, + Command::Job(JobCommand::Reviews(Some("daily".to_string()))) + ); } #[test] diff --git a/src/prompt.rs b/src/prompt.rs index ae907da..214736a 100644 --- a/src/prompt.rs +++ b/src/prompt.rs @@ -19,7 +19,7 @@ const BASE_POLICY: &str = "\ - Follow the user-owned system identity unless it conflicts with this Push-owned policy. - Begin with README.md under the `context` value in Resolved workspace paths when user context is relevant. - Do not modify SOUL.md under `assistant_root` or files under the `evals` value unless the user asks. -- When asked to create or change a job, write the complete runbook under the `jobs` value and run `push job validate` before saying it succeeded. This supersedes older draft or approval instructions."; +- When asked to create or change a job, write the complete runbook under the `jobs` value and run `push job validate` before saying it succeeded. A new or changed enabled schedule remains inactive until Push presents its exact revision for owner review. This supersedes older draft or installation-approval instructions."; /// One composed backend turn. Instructions go through the backend's native /// system-prompt mechanism; content goes through its ordinary prompt input. @@ -356,6 +356,9 @@ mod tests { assert!(prompt .instructions .contains("complete runbook under the `jobs` value")); + assert!(prompt + .instructions + .contains("enabled schedule remains inactive")); let _ = std::fs::remove_dir_all(root); let _ = std::fs::remove_dir_all(work_dir); } diff --git a/src/util.rs b/src/util.rs index 828379d..e7c676b 100644 --- a/src/util.rs +++ b/src/util.rs @@ -41,6 +41,30 @@ pub(crate) fn same_file(expected: &std::fs::Metadata, opened: &std::fs::Metadata && opened.is_file() } +#[cfg(unix)] +pub(crate) fn file_identity(metadata: &std::fs::Metadata) -> String { + use std::os::unix::fs::MetadataExt; + format!( + "unix:{}:{}:{}:{}:{}", + metadata.dev(), + metadata.ino(), + metadata.len(), + metadata.ctime(), + metadata.ctime_nsec() + ) +} + +#[cfg(not(unix))] +pub(crate) fn file_identity(metadata: &std::fs::Metadata) -> String { + let modified = metadata + .modified() + .ok() + .and_then(|value| value.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|value| value.as_nanos()) + .unwrap_or_default(); + format!("portable:{}:{modified}", metadata.len()) +} + /// Restricts a Push-owned path to owner-only access (0o700 directories, /// 0o600 files). A no-op on non-Unix platforms. #[cfg(unix)] diff --git a/tests/init_cli.rs b/tests/init_cli.rs index 004c752..b096b5d 100644 --- a/tests/init_cli.rs +++ b/tests/init_cli.rs @@ -93,7 +93,9 @@ fn init_without_path_creates_assistant_in_current_directory() { let canonical_skill = assistant.join("skills/push/SKILL.md"); let skill = std::fs::read_to_string(&canonical_skill).unwrap(); assert!(skill.contains("name: push")); - assert!(skill.contains("push-managed-version: \"1\"")); + assert!(skill.contains("push-managed-version: \"2\"")); + assert!(skill.contains("push job reviews []")); + assert!(skill.contains("separate owner review")); assert!(!skill.contains(&root.to_string_lossy().to_string())); for provider in [".agents", ".claude"] { let exposure = assistant.join(provider).join("skills/push"); diff --git a/tests/json_cli.rs b/tests/json_cli.rs index a8fa2d3..ea70615 100644 --- a/tests/json_cli.rs +++ b/tests/json_cli.rs @@ -219,6 +219,12 @@ fn scoped_commands_emit_one_json_document_without_unrelated_output() { assert_keys(&runs["data"], &["job_name", "runs"]); assert!(runs["data"]["job_name"].is_null()); assert!(runs["data"]["runs"].is_array()); + + let reviews = json_stdout(&fixture.command().args(["job", "reviews"]).output().unwrap()); + assert_success_envelope(&reviews, "job.reviews"); + assert_keys(&reviews["data"], &["job_name", "reviews"]); + assert!(reviews["data"]["job_name"].is_null()); + assert!(reviews["data"]["reviews"].is_array()); } #[test] @@ -618,6 +624,80 @@ fn job_runs_json_omits_stored_content_fields() { } } +#[test] +fn job_reviews_json_exposes_exact_activation_metadata() { + let fixture = Fixture::new("schedule-review"); + let initial = fixture.command().args(["job", "reviews"]).output().unwrap(); + json_stdout(&initial); + let database = fixture.home.join(".push/push.db"); + let connection = rusqlite::Connection::open(database).unwrap(); + connection + .execute( + "INSERT INTO job_schedule_reviews ( + id, job_name, content_hash, snapshot_hash, file_identity, path, + schedules_json, backend, timeout_ms, workdir, delivery_channel, + delivery_target, status, proposed_at_ms, decided_at_ms, + activated_at_ms, reviewed_by, reason + ) VALUES ( + ?1, 'daily', ?2, ?3, ?4, ?5, ?6, 'codex', 300000, ?7, + 'telegram', '123', 'activated', 1000, 1100, 1200, ?8, ?9 + )", + params![ + "review-fingerprint", + "content-sha256", + "snapshot-sha256", + "unix:1:2", + fixture.assistant.join("jobs/daily.md").to_string_lossy(), + r#"[{"id":"daily","kind":"cron","schedule":"0 9 * * *","timezone":"Europe/London","enabled":true}]"#, + fixture.assistant.to_string_lossy(), + "channel=telegram thread=dm:123 sender=123 chat=123", + "approved exact revision", + ], + ) + .unwrap(); + + let output = fixture + .command() + .args(["job", "reviews", "daily"]) + .output() + .unwrap(); + let payload = json_stdout(&output); + assert_success_envelope(&payload, "job.reviews"); + assert_keys(&payload["data"], &["job_name", "reviews"]); + assert_eq!(payload["data"]["job_name"], "daily"); + let review = &payload["data"]["reviews"][0]; + assert_keys( + review, + &[ + "backend", + "content_hash", + "delivery", + "job_name", + "reason", + "review_id", + "reviewed_by", + "schedules", + "status", + "timeout_ms", + "workdir", + ], + ); + assert_eq!(review["review_id"], "review-fingerprint"); + assert_eq!(review["job_name"], "daily"); + assert_eq!(review["status"], "activated"); + assert_eq!(review["content_hash"], "content-sha256"); + assert_eq!(review["backend"], "codex"); + assert_eq!(review["timeout_ms"], 300000); + assert_eq!(review["delivery"]["channel"], "telegram"); + assert_eq!(review["delivery"]["target"], "123"); + assert_keys( + &review["schedules"][0], + &["enabled", "id", "kind", "schedule", "timezone"], + ); + assert_eq!(review["schedules"][0]["id"], "daily"); + assert_eq!(review["schedules"][0]["enabled"], true); +} + #[cfg(unix)] fn make_executable(path: &Path) { use std::os::unix::fs::PermissionsExt;