admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(scheduler): JobScheduler on TimeProvider with watcher, reminder and prompt firing

Browse Source 
Spec: a skipped one-off fires after the next start; the late note's placement and no-exit-code failure wording are pinned down; IsDeliverable is part of the watcher invariant and must not create a session; saves snapshot inside a serialising gate.
This commit is contained in:
EugeneTes
2026-09-27 10:20:22 +02:00
parent 51bc0cedf1
commit 4bf99dc57c
8 changed files with 1145 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
docs/superpowers/specs/2026-09-26-scheduler-design.md
View File

@@ -199,11 +199,11 @@ public interface IJobDelivery
Task SendTextAsync(string target, string markdown, CancellationToken ct);
/// <summary>Run a prompt in the target's session (follow-up when busy); completes when the run that consumed it settles.</summary>
Task RunPromptAsync(string target, string prompt, CancellationToken ct);
/// <summary>Working directory for the target's scripts.</summary>
/// <summary>Working directory for the target's scripts. A path computation: must not create or wake a session.</summary>
string WorkspaceFor(string target);
/// <summary>False when the target may no longer be served (e.g. removed from the allow-list): the job is skipped, not deleted.</summary>
/// <summary>False when the target may no longer be served (e.g. removed from the allow-list): the job is skipped, not deleted. Must not create or wake a session.</summary>
bool IsDeliverable(string target);
/// <summary>The target's current zone (the chat's, else the host default). Read on every computation, never cached.</summary>
/// <summary>The target's current zone (the chat's, else the host default). Read on every computation, never cached. Must not create or wake a session.</summary>
TimeZoneInfo TimeZoneFor(string target);
}
```
@@ -215,15 +215,16 @@ Firing (`FireAsync(job)`, tracked in an in-flight set and never awaited by the t
| Kind | Action | Message format |
|---|---|---|
| reminder | `SendTextAsync` | `⏰ <text>` |
| prompt | `RunPromptAsync` | `[Scheduled job "<name>" (<id>)]\n<text>` |
| reminder | `SendTextAsync` | `⏰ <text>`; fired more than 60 s late: `⏰ <text> (due <d MMM HH:mm>, delivered late)` |
| prompt | `RunPromptAsync` | `[Scheduled job "<name>" (<id>)]\n<text>`; fired more than 60 s late, the note gets its own line after the header: `[Scheduled job "<name>" (<id>)]\nThis was due at <d MMM HH:mm>; the bot was offline.\n<text>` |
| watcher, matched (exit 0, non-empty stdout) | `RunPromptAsync` | `[Scheduled job "<name>" (<id>)] The watcher script printed:\n<stdout>` |
| watcher, silent (exit 0, empty stdout) | nothing: no model call, no session activity, no transcript entry | only `LastRunAt`/`LastResult` are updated |
| watcher, failed | `SendTextAsync` once per failure streak | `⚠️ Job "<name>" (<id>) failed (exit <n>[, timed out]):\n```\n<stderr tail>\n```` |
| watcher, failed | `SendTextAsync` once per failure streak | `⚠️ Job "<name>" (<id>) failed (exit <n>[, timed out]):\n```\n<stderr tail>\n````; with no exit code it reads `failed (timed out)` or `failed (did not start)` |
- **Watcher invariant.** Running a watcher's script touches only `IScriptRunner`, `WorkspaceFor` and `TimeZoneFor`.
- Watcher matches never get a late note (a watcher only runs when its occurrence comes due; there's no catch-up).
- **Watcher invariant.** Running a watcher's script touches only `IScriptRunner`, `IsDeliverable`, `WorkspaceFor` and `TimeZoneFor`.
It never calls `RunPromptAsync` or `SendTextAsync`, and the Telegram delivery's `WorkspaceFor`/`TimeZoneFor` don't
create or look up a chat session. Only the outcome decides what follows: `Output` → exactly one `RunPromptAsync`,
create or look up a chat session (nor does `IsDeliverable`). Only the outcome decides what follows: `Output` → exactly one `RunPromptAsync`,
`Failed` → at most one `SendTextAsync` per failure streak, `Silent` → nothing.
- **Redaction.** A watcher's stdout goes through `redactor.Redact` before it's put into the prompt, and the stderr
tail before it's put into the failure alert (direct to the chat, but still not a place for the bot token). Known
@@ -237,7 +238,9 @@ Firing (`FireAsync(job)`, tracked in an in-flight set and never awaited by the t
throws `ChatDeliveryException`. The job's state still advances: there's no redelivery queue.
- `CancelAsync` removes the job. If that job's watcher script is running it's killed and its outcome thrown away. A
prompt run that's in progress isn't aborted: that's what `/stop` is for.
- `!IsDeliverable(target)` at fire time → skip, log a warning, keep the job (a cron job advances).
- `!IsDeliverable(target)` at fire time → skip, log a warning, keep the job (a cron job advances); a skipped one-off
stays in the store and fires again after the next start.
- Saves are serialised and each one snapshots the jobs inside that gate, so a save never overwrites newer state.
- `TimeZoneChangedAsync(target)` recomputes `NextRunAt` for the target's cron jobs in the new zone (strictly after now,
no catch-up), leaves one-offs untouched, re-arms the timer and returns
`ZoneChange(int CronJobs, int OneOffs)`. A cron occurrence already in flight isn't affected.

20
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs Normal file
View File

@@ -0,0 +1,20 @@
namespace Volcanic.AgenticCore.Extensions.Scheduler;
/// <summary>What the host does for the scheduler. Passed to <see cref="JobScheduler.StartAsync"/>, not the constructor, to break the host's DI cycle.</summary>
public interface IJobDelivery
{
/// <summary>Send Markdown to the target's chat, in order with that chat's rendered output.</summary>
Task SendTextAsync(string target, string markdown, CancellationToken cancellationToken);
/// <summary>Run a prompt in the target's session (follow-up when busy); completes when the run that consumed it settles.</summary>
Task RunPromptAsync(string target, string prompt, CancellationToken cancellationToken);
/// <summary>Working directory for the target's scripts. A path computation: must not create or wake a session.</summary>
string WorkspaceFor(string target);
/// <summary>False when the target may no longer be served (e.g. removed from the allow-list): the job is skipped, not deleted. Must not create or wake a session.</summary>
bool IsDeliverable(string target);
/// <summary>The target's current zone (the chat's, else the host default). Read on every computation, never cached. Must not create or wake a session.</summary>
TimeZoneInfo TimeZoneFor(string target);
}

56
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobFormat.cs Normal file
View File

@@ -0,0 +1,56 @@
using System.Globalization;
using System.Text;
namespace Volcanic.AgenticCore.Extensions.Scheduler;
/// <summary>How jobs and times are shown (the <c>schedule</c> tool, <c>/jobs</c>, zone-change replies). Invariant culture, 24 h.</summary>
public static class JobFormat
{
public static string Icon(JobKind kind) => kind switch { JobKind.Reminder => "⏰", JobKind.Prompt => "💬", _ => "🔍" };
/// <summary><c>27 Sep 09:00</c> in <paramref name="zone"/>.</summary>
public static string DayTime(DateTimeOffset at, TimeZoneInfo zone) =>
TimeZoneInfo.ConvertTime(at, zone).ToString("d MMM HH:mm", CultureInfo.InvariantCulture);
/// <summary><c>14:10</c> when on the same local day as <paramref name="now"/>, else <see cref="DayTime"/>.</summary>
public static string When(DateTimeOffset at, TimeZoneInfo zone, DateTimeOffset now)
{
var local = TimeZoneInfo.ConvertTime(at, zone);
return local.Date == TimeZoneInfo.ConvertTime(now, zone).Date
? local.ToString("HH:mm", CultureInfo.InvariantCulture)
: DayTime(at, zone);
}
/// <summary><c>Sat 2026-09-26 14:03</c> (the model-facing form: weekday and full date).</summary>
public static string Stamp(DateTimeOffset at, TimeZoneInfo zone) =>
TimeZoneInfo.ConvertTime(at, zone).ToString("ddd yyyy-MM-dd HH:mm", CultureInfo.InvariantCulture);
public static string ZoneName(TimeZoneInfo zone) => ZoneNames.Name(zone);
/// <summary><c>UTC+02:00</c> at <paramref name="at"/>.</summary>
public static string Offset(TimeZoneInfo zone, DateTimeOffset at) => ZoneNames.Offset(zone.GetUtcOffset(at));
/// <summary><c>exit 1</c>, <c>exit 1, timed out</c>, <c>timed out</c>, or <c>did not start</c>.</summary>
public static string FailureReason(JobResult result) => result switch
{
{ ExitCode: { } code, TimedOut: true } => $"exit {code.ToString(CultureInfo.InvariantCulture)}, timed out",
{ ExitCode: { } code } => $"exit {code.ToString(CultureInfo.InvariantCulture)}",
{ TimedOut: true } => "timed out",
_ => "did not start",
};
/// <summary><c>`k7m2q` ⏰ Call the dentist · once 27 Sep 09:00</c>; cron jobs show the expression and the next run; a failed last run is appended.</summary>
public static string Line(JobView view, TimeZoneInfo zone, DateTimeOffset now)
{
var job = view.Job;
var sb = new StringBuilder().Append('`').Append(job.Id).Append("` ").Append(Icon(job.Kind)).Append(' ').Append(job.Name);
if (job.Schedule.At is { } at) sb.Append(" · once ").Append(DayTime(at, zone));
else
{
sb.Append(" · `").Append(job.Schedule.Cron).Append('`');
if (view.NextRunAt is { } next) sb.Append(" · next ").Append(When(next, zone, now));
}
if (job.LastResult is { Ok: false } failed) sb.Append(" · last: failed (").Append(FailureReason(failed)).Append(')');
return sb.ToString();
}
}

18
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobMessages.cs Normal file
View File

@@ -0,0 +1,18 @@
namespace Volcanic.AgenticCore.Extensions.Scheduler;
/// <summary>The texts a fire sends (spec §4 table).</summary>
internal static class JobMessages
{
public static string Reminder(ScheduledJob job, DateTimeOffset? lateDue, TimeZoneInfo zone) =>
$"⏰ {job.Text}" + (lateDue is { } due ? $" (due {JobFormat.DayTime(due, zone)}, delivered late)" : "");
public static string Prompt(ScheduledJob job, DateTimeOffset? lateDue, TimeZoneInfo zone) =>
$"{Header(job)}\n" + (lateDue is { } due ? $"This was due at {JobFormat.DayTime(due, zone)}; the bot was offline.\n" : "") + job.Text;
public static string WatcherMatch(ScheduledJob job, string output) => $"{Header(job)} The watcher script printed:\n{output}";
public static string Failure(ScheduledJob job, JobResult result) =>
$"⚠️ Job \"{job.Name}\" ({job.Id}) failed ({JobFormat.FailureReason(result)}):\n```\n{(string.IsNullOrWhiteSpace(result.StderrTail) ? "(no output on stderr)" : result.StderrTail)}\n```";
private static string Header(ScheduledJob job) => $"[Scheduled job \"{job.Name}\" ({job.Id})]";
}

460
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs Normal file
View File

@@ -0,0 +1,460 @@
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Abstractions.Processes;
namespace Volcanic.AgenticCore.Extensions.Scheduler;
/// <summary>
/// The process-wide scheduler: loads <c>jobs.json</c>, arms one <see cref="TimeProvider"/> timer for the soonest job
/// (capped at 1 h, re-armed on every tick, create, cancel, fire and zone change), and fires jobs through the host's
/// <see cref="IJobDelivery"/>. A fire is tracked in flight and never awaited by the tick; an occurrence that comes due
/// while the job is still in flight is skipped. A watcher's script run touches only the runner, <c>IsDeliverable</c>,
/// <c>WorkspaceFor</c> and <c>TimeZoneFor</c>: a silent run makes no delivery call at all.
/// </summary>
public sealed class JobScheduler : IAsyncDisposable
{
public static readonly TimeSpan MaxTimerDelay = TimeSpan.FromHours(1);
public static readonly TimeSpan LateThreshold = TimeSpan.FromSeconds(60);
public static readonly TimeSpan StopGrace = TimeSpan.FromSeconds(10);
private const string IdAlphabet = "abcdefghjkmnpqrstuvwxyz23456789";
private readonly SchedulerOptions _options;
private readonly IScriptRunner _runner;
private readonly ISecretRedactor _redactor;
private readonly TimeProvider _time;
private readonly ILogger _logger;
private readonly JobStore _store;
private readonly SemaphoreSlim _saveGate = new(1, 1);
private readonly Lock _lock = new();
private readonly Dictionary<string, Entry> _jobs = new(StringComparer.Ordinal);
private readonly Dictionary<string, Fire> _inFlight = new(StringComparer.Ordinal);
private readonly CancellationTokenSource _stopping = new();
private IJobDelivery? _delivery;
private ITimer? _timer;
private DateTimeOffset? _wakeAt;
private bool _stopped;
private Task _pendingTick = Task.CompletedTask;
public JobScheduler(SchedulerOptions options, IScriptRunner runner, ISecretRedactor redactor, TimeProvider time, ILogger<JobScheduler> logger)
{
ArgumentNullException.ThrowIfNull(options);
if (string.IsNullOrWhiteSpace(options.StorePath)) throw new ArgumentException("SchedulerOptions.StorePath is required.", nameof(options));
_options = options;
_runner = runner;
_redactor = redactor;
_time = time;
_logger = logger;
_store = new JobStore(options.StorePath, time, logger);
}
public bool IsStarted { get { lock (_lock) return _delivery is not null && !_stopped; } }
public DateTimeOffset Now => _time.GetUtcNow();
public string StorePath => _store.Path;
/// <summary>When the timer fires next (tests).</summary>
internal DateTimeOffset? NextWakeAt { get { lock (_lock) return _wakeAt; } }
/// <summary>Load the store, fire overdue one-offs (late), compute cron occurrences strictly after now (no catch-up), arm the timer.</summary>
public async Task StartAsync(IJobDelivery delivery, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(delivery);
var jobs = await _store.LoadAsync(cancellationToken).ConfigureAwait(false);
var now = _time.GetUtcNow();
lock (_lock)
{
if (_delivery is not null || _stopped) throw new InvalidOperationException("The scheduler was already started.");
_delivery = delivery;
foreach (var job in jobs)
{
var entry = new Entry(job);
if (job.Schedule.At is { } at) entry.Next = at;
else
{
entry.Next = NextCron(job, now);
var missed = Schedules.NextCron(job.Schedule.Cron!, job.LastRunAt ?? job.CreatedAt, delivery.TimeZoneFor(job.Target));
if (missed is { } m && m < now)
_logger.LogInformation("Job {Id} ({Name}): occurrences missed while the bot was down are skipped; next {Next}", job.Id, job.Name, entry.Next);
}
_jobs[job.Id] = entry;
}
_timer = _time.CreateTimer(_ => OnTimer(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
}
Tick(); // overdue one-offs fire now
}
/// <summary>Disarm, kill running scripts, and wait at most <see cref="StopGrace"/> for in-flight fires.</summary>
public async Task StopAsync(CancellationToken cancellationToken)
{
Fire[] fires;
ITimer? timer;
lock (_lock)
{
if (_stopped) return;
_stopped = true;
fires = [.. _inFlight.Values];
timer = _timer;
}
timer?.Dispose();
await _stopping.CancelAsync().ConfigureAwait(false); // every fire's token is linked to this one
try
{
await Task.WhenAll(fires.Select(f => f.Task)).WaitAsync(StopGrace, _time, cancellationToken).ConfigureAwait(false);
}
catch (TimeoutException)
{
_logger.LogWarning("Scheduler stop: {Count} job(s) still running after {Grace}", fires.Count(f => !f.Task.IsCompleted), StopGrace);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { }
}
public ValueTask DisposeAsync() => new(StopAsync(CancellationToken.None));
public IReadOnlyList<JobView> List(string target)
{
lock (_lock)
return _jobs.Values.Where(e => e.Job.Target == target)
.Select(e => new JobView(e.Job, e.Next))
.OrderBy(v => v.NextRunAt ?? DateTimeOffset.MaxValue).ThenBy(v => v.Job.CreatedAt)
.ToList();
}
public TimeZoneInfo TimeZoneFor(string target) => Volatile.Read(ref _delivery)?.TimeZoneFor(target) ?? TimeZoneInfo.Local;
/// <summary>Validate (an offset-less <c>at</c> is read in the target's zone), persist, arm. Throws <see cref="JobValidationException"/>.</summary>
public async Task<JobView> CreateAsync(string target, NewJob job, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(job);
if (!IsStarted) throw new InvalidOperationException("The scheduler isn't running.");
var zone = TimeZoneFor(target);
var now = _time.GetUtcNow();
var schedule = JobValidation.ValidateNew(job, now, zone);
ScheduledJob created;
DateTimeOffset? next;
lock (_lock)
{
if (_jobs.Values.Count(e => e.Job.Target == target) >= _options.MaxJobsPerTarget)
throw new JobValidationException($"This chat already has {_options.MaxJobsPerTarget} scheduled jobs, the maximum. Cancel one first.");
created = new ScheduledJob
{
Id = NewId(), Target = target, Name = job.Name.Trim(), Kind = job.Kind, Schedule = schedule,
Text = job.Kind == JobKind.Watcher ? null : job.Text, Command = job.Kind == JobKind.Watcher ? job.Command : null,
TimeoutSeconds = job.TimeoutSeconds, CreatedAt = now,
};
next = schedule.At ?? Schedules.NextCron(schedule.Cron!, now, zone);
_jobs[created.Id] = new Entry(created) { Next = next };
}
await PersistAsync(cancellationToken).ConfigureAwait(false);
Arm();
return new JobView(created, next);
}
/// <summary>Remove a job of <paramref name="target"/>. A running watcher is killed and its outcome dropped; a prompt run in progress isn't aborted. False: unknown, or another target's.</summary>
public async Task<bool> CancelAsync(string target, string id, CancellationToken cancellationToken)
{
Fire? running;
lock (_lock)
{
if (!_jobs.TryGetValue(id, out var entry) || entry.Job.Target != target) return false;
_jobs.Remove(id);
_inFlight.TryGetValue(id, out running);
}
if (running is not null) await TryCancelAsync(running.Cancel).ConfigureAwait(false);
await PersistAsync(cancellationToken).ConfigureAwait(false);
Arm();
return true;
}
/// <summary>The target's zone changed: cron jobs get their next occurrence in the new zone (strictly after now), one-offs keep their instant.</summary>
public Task<ZoneChange> TimeZoneChangedAsync(string target, CancellationToken cancellationToken)
{
var now = _time.GetUtcNow();
var zone = TimeZoneFor(target);
int cron = 0, oneOffs = 0;
lock (_lock)
{
foreach (var entry in _jobs.Values.Where(e => e.Job.Target == target))
{
if (entry.Job.Schedule.Cron is { } expression)
{
cron++;
entry.Next = Schedules.NextCron(expression, now, zone);
}
else oneOffs++;
}
}
Arm();
return Task.FromResult(new ZoneChange(cron, oneOffs));
}
/// <summary>Wait for the pending tick and every fire in flight (tests).</summary>
internal async Task IdleAsync()
{
while (true)
{
Task tick;
Task[] fires;
lock (_lock)
{
tick = _pendingTick;
fires = _inFlight.Values.Select(f => f.Task).ToArray();
}
if (tick.IsCompleted && fires.All(f => f.IsCompleted)) return;
await Task.WhenAll(fires.Append(tick)).WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
}
}
// ---- Timer and ticks ----
/// <summary>The timer callback: queues a tick behind the previous one (it may run synchronously inside <c>ITimer.Change</c>).</summary>
private void OnTimer()
{
lock (_lock)
{
if (_stopped) return;
_pendingTick = _pendingTick.ContinueWith(_ => Tick(), CancellationToken.None, TaskContinuationOptions.None, TaskScheduler.Default);
}
}
/// <summary>Starts every due job that isn't in flight, advances each due job's next occurrence, re-arms. Never awaits a fire.</summary>
private void Tick()
{
try
{
var now = _time.GetUtcNow();
lock (_lock)
{
if (_stopped || _delivery is null) return;
foreach (var entry in _jobs.Values)
{
if (entry.Next is not { } next || next > now) continue;
if (_inFlight.ContainsKey(entry.Job.Id))
_logger.LogInformation("Job {Id} ({Name}) is still running; skipped the occurrence due {Due}", entry.Job.Id, entry.Job.Name, next);
else StartFire(entry.Job, next, now);
// Cron: the next occurrence strictly after now. One-off: none (it's removed once its fire completes).
entry.Next = null;
if (entry.Job.Schedule.Cron is null) continue;
try
{
entry.Next = NextCron(entry.Job, now);
}
catch (Exception ex)
{
// Left unarmed rather than due forever (a zero-delay timer loop); a restart or zone change recomputes it.
_logger.LogError(ex, "Job {Id} ({Name}): computing the next occurrence failed", entry.Job.Id, entry.Job.Name);
}
}
}
}
finally
{
Arm();
}
}
private void Arm()
{
ITimer timer;
TimeSpan delay;
lock (_lock)
{
if (_stopped || _timer is null) return;
var now = _time.GetUtcNow();
var soonest = _jobs.Values.Select(e => e.Next).Where(n => n is not null).DefaultIfEmpty(null).Min();
var wake = soonest is { } s && s - now < MaxTimerDelay ? s : now + MaxTimerDelay;
if (wake < now) wake = now;
_wakeAt = wake;
timer = _timer;
delay = wake - now;
}
try
{
timer.Change(delay, Timeout.InfiniteTimeSpan); // outside the lock: a zero delay may call back synchronously
}
catch (ObjectDisposedException)
{
// Stopped meanwhile.
}
}
/// <summary>Under the lock. The next cron occurrence strictly after <paramref name="after"/> in the target's current zone.</summary>
private DateTimeOffset? NextCron(ScheduledJob job, DateTimeOffset after) =>
Schedules.NextCron(job.Schedule.Cron!, after, _delivery!.TimeZoneFor(job.Target));
// ---- Firing ----
/// <summary>Called under the lock: registers the fire in flight before it can complete.</summary>
private void StartFire(ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now)
{
var cancel = CancellationTokenSource.CreateLinkedTokenSource(_stopping.Token);
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var task = RunFireAsync(gate.Task, job, dueAt, now, cancel);
_inFlight[job.Id] = new Fire(task, cancel);
gate.SetResult();
}
private async Task RunFireAsync(Task gate, ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now, CancellationTokenSource cancel)
{
await gate.ConfigureAwait(false);
try
{
await FireAsync(job, dueAt, now, cancel.Token).ConfigureAwait(false);
}
catch (Exception ex)
{
_logger.LogError(ex, "Job {Id} ({Name}) failed to fire", job.Id, job.Name);
}
finally
{
lock (_lock) _inFlight.Remove(job.Id);
cancel.Dispose();
}
}
private async Task FireAsync(ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now, CancellationToken scriptToken)
{
var delivery = _delivery!;
if (!delivery.IsDeliverable(job.Target))
{
// Kept: a cron job has already advanced; a one-off fires again after the next start.
_logger.LogWarning("Job {Id} ({Name}): target {Target} can't be served any more; skipped", job.Id, job.Name, job.Target);
return;
}
var zone = delivery.TimeZoneFor(job.Target);
DateTimeOffset? lateDue = now - dueAt > LateThreshold ? dueAt : null;
JobResult? result;
switch (job.Kind)
{
case JobKind.Reminder:
await DeliverAsync(job, ct => delivery.SendTextAsync(job.Target, JobMessages.Reminder(job, lateDue, zone), ct)).ConfigureAwait(false);
result = new JobResult(true, null, false, null, _time.GetUtcNow());
break;
case JobKind.Prompt:
await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.Prompt(job, lateDue, zone), ct)).ConfigureAwait(false);
result = new JobResult(true, null, false, null, _time.GetUtcNow());
break;
default:
result = await RunWatcherAsync(job, delivery, zone, scriptToken).ConfigureAwait(false);
break;
}
if (result is not null) await RecordAsync(job.Id, result).ConfigureAwait(false);
}
/// <summary>Null when the run was cancelled (job cancelled or scheduler stopping): the outcome is dropped.</summary>
private async Task<JobResult?> RunWatcherAsync(ScheduledJob job, IJobDelivery delivery, TimeZoneInfo zone, CancellationToken scriptToken)
{
var timeout = job.TimeoutSeconds is { } seconds ? TimeSpan.FromSeconds(seconds) : _options.DefaultScriptTimeout;
var request = new ScriptRequest(job.Command!, delivery.WorkspaceFor(job.Target), timeout, job.Target, job.Id, ZoneNames.Iana(zone));
ScriptOutcome outcome;
try
{
outcome = await _runner.RunAsync(request, scriptToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (scriptToken.IsCancellationRequested)
{
return null;
}
if (scriptToken.IsCancellationRequested || !Exists(job.Id)) return null;
var at = _time.GetUtcNow();
switch (outcome)
{
case ScriptOutcome.Output output:
// The condition is met: exactly one agent prompt. Redacted: watcher stdout isn't a tool result.
await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.WatcherMatch(job, _redactor.Redact(output.Text)), ct)).ConfigureAwait(false);
return new JobResult(true, 0, false, null, at);
case ScriptOutcome.Failed failed:
var result = new JobResult(false, failed.ExitCode, failed.TimedOut, _redactor.Redact(failed.StderrTail), at);
if (!IsRepeat(LastResult(job.Id), result))
await DeliverAsync(job, ct => delivery.SendTextAsync(job.Target, JobMessages.Failure(job, result), ct)).ConfigureAwait(false);
return result;
default:
return new JobResult(true, 0, false, null, at); // silent: nothing but LastRunAt/LastResult
}
}
/// <summary>The previous run failed the same way (exit code and timeout): the alert isn't sent again.</summary>
private static bool IsRepeat(JobResult? previous, JobResult current) =>
previous is { Ok: false } p && p.ExitCode == current.ExitCode && p.TimedOut == current.TimedOut;
/// <summary>A delivery exception is logged and swallowed; the job's state still advances (no redelivery queue).</summary>
private async Task DeliverAsync(ScheduledJob job, Func<CancellationToken, Task> deliver)
{
try
{
await deliver(_stopping.Token).ConfigureAwait(false); // a prompt run isn't aborted by a cancel, only by shutdown
}
catch (OperationCanceledException) when (_stopping.IsCancellationRequested) { }
catch (Exception ex)
{
_logger.LogError(ex, "Job {Id} ({Name}): delivery to {Target} failed", job.Id, job.Name, job.Target);
}
}
/// <summary>Update and persist the run fields; a one-off is removed now that its fire has completed.</summary>
private async Task RecordAsync(string id, JobResult result)
{
lock (_lock)
{
if (!_jobs.TryGetValue(id, out var entry)) return; // cancelled meanwhile
if (entry.Job.Schedule.At is not null) _jobs.Remove(id);
else entry.Job = entry.Job with { LastRunAt = result.At, LastResult = result };
}
try
{
await PersistAsync(CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
_logger.LogError(ex, "Saving the job store failed");
}
Arm();
}
/// <summary>Saves the current jobs. The snapshot is taken inside the gate, so a later save never writes older state.</summary>
private async Task PersistAsync(CancellationToken cancellationToken)
{
await _saveGate.WaitAsync(cancellationToken).ConfigureAwait(false);
try
{
List<ScheduledJob> snapshot;
lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)];
await _store.SaveAsync(snapshot, cancellationToken).ConfigureAwait(false);
}
finally
{
_saveGate.Release();
}
}
private static async Task TryCancelAsync(CancellationTokenSource cancel)
{
try
{
await cancel.CancelAsync().ConfigureAwait(false);
}
catch (ObjectDisposedException)
{
// The fire completed meanwhile.
}
}
private bool Exists(string id) { lock (_lock) return _jobs.ContainsKey(id); }
private JobResult? LastResult(string id) { lock (_lock) return _jobs.TryGetValue(id, out var e) ? e.Job.LastResult : null; }
/// <summary>Under the lock. Not reused while a cancelled job with that id is still in flight.</summary>
private string NewId()
{
string id;
do id = new string(Random.Shared.GetItems(IdAlphabet.AsSpan(), 5));
while (_jobs.ContainsKey(id) || _inFlight.ContainsKey(id));
return id;
}
private sealed class Entry(ScheduledJob job)
{
public ScheduledJob Job { get; set; } = job;
public DateTimeOffset? Next { get; set; }
}
private sealed record Fire(Task Task, CancellationTokenSource Cancel);
}

40
tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobFormatTests.cs Normal file
View File

@@ -0,0 +1,40 @@
using Volcanic.AgenticCore.Extensions.Scheduler;
using Volcanic.AgenticCore.Extensions.Tests.Scheduler.Support;
namespace Volcanic.AgenticCore.Extensions.Tests.Scheduler;
public sealed class JobFormatTests
{
private static readonly DateTimeOffset Now = SchedulerHarness.Start; // Sat 26 Sep 14:03 Berlin
private static readonly TimeZoneInfo Berlin = SchedulerHarness.Berlin;
private static ScheduledJob Job(string id, string name, JobKind kind, JobSchedule schedule) => new()
{
Id = id, Target = "111", Name = name, Kind = kind, Schedule = schedule, CreatedAt = Now,
Text = kind == JobKind.Watcher ? null : name, Command = kind == JobKind.Watcher ? "check" : null,
};
[Fact]
public void Job_lines_show_kind_schedule_next_and_last_failure()
{
var once = new DateTimeOffset(2026, 9, 27, 7, 0, 0, TimeSpan.Zero);
Assert.Equal("`k7m2q` ⏰ Call the dentist · once 27 Sep 09:00",
JobFormat.Line(new JobView(Job("k7m2q", "Call the dentist", JobKind.Reminder, new JobSchedule(once, null)), once), Berlin, Now));
Assert.Equal("`p3xz9` 💬 Inbox summary · `0 8 * * 1-5` · next 28 Sep 08:00",
JobFormat.Line(new JobView(Job("p3xz9", "Inbox summary", JobKind.Prompt, new JobSchedule(null, "0 8 * * 1-5")), new DateTimeOffset(2026, 9, 28, 6, 0, 0, TimeSpan.Zero)), Berlin, Now));
var failing = Job("w8r4t", "Price check", JobKind.Watcher, new JobSchedule(null, "*/10 * * * *")) with { LastResult = new JobResult(false, 1, false, "x", Now) };
Assert.Equal("`w8r4t` 🔍 Price check · `*/10 * * * *` · next 14:10 · last: failed (exit 1)",
JobFormat.Line(new JobView(failing, new DateTimeOffset(2026, 9, 26, 12, 10, 0, TimeSpan.Zero)), Berlin, Now));
}
[Fact]
public void When_omits_the_date_today()
{
Assert.Equal("14:10", JobFormat.When(new DateTimeOffset(2026, 9, 26, 12, 10, 0, TimeSpan.Zero), Berlin, Now));
Assert.Equal("27 Sep 00:10", JobFormat.When(new DateTimeOffset(2026, 9, 26, 22, 10, 0, TimeSpan.Zero), Berlin, Now));
Assert.Equal("Sat 2026-09-26 14:03", JobFormat.Stamp(Now, Berlin));
Assert.Equal("UTC+02:00", JobFormat.Offset(Berlin, Now));
Assert.Equal("exit 1, timed out", JobFormat.FailureReason(new JobResult(false, 1, true, null, Now)));
Assert.Equal("did not start", JobFormat.FailureReason(new JobResult(false, null, false, "failed to start", Now)));
}
}

384
tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/JobSchedulerTests.cs Normal file
View File

@@ -0,0 +1,384 @@
using Volcanic.AgenticCore.Extensions.Scheduler;
using Volcanic.AgenticCore.Extensions.Tests.GoogleCalendar;
using Volcanic.AgenticCore.Extensions.Tests.Scheduler.Support;
namespace Volcanic.AgenticCore.Extensions.Tests.Scheduler;
public sealed class JobSchedulerTests
{
private static readonly TimeSpan Minute = TimeSpan.FromMinutes(1);
private static readonly DateTimeOffset Start = SchedulerHarness.Start; // 12:03Z = Sat 14:03 Berlin
private static ScheduledJob Stored(string id, JobKind kind, JobSchedule schedule, string? text = "Call the dentist", string target = "111") => new()
{
Id = id, Target = target, Name = text ?? "Watch", Kind = kind, Schedule = schedule, CreatedAt = Start.AddDays(-1),
Text = kind == JobKind.Watcher ? null : text, Command = kind == JobKind.Watcher ? "scripts/check.sh" : null,
};
private static NewJob Watcher(string cron = "* * * * *") => new("Price check", JobKind.Watcher, Cron: cron, Command: "scripts/price.sh");
[Fact]
public async Task An_overdue_one_off_fires_on_start_with_the_late_note_and_is_removed_after_delivery()
{
await using var h = new SchedulerHarness();
await h.Store.SaveAsync([Stored("k7m2q", JobKind.Reminder, new JobSchedule(Start.AddMinutes(-5), null))], CancellationToken.None);
h.Delivery.HoldTexts = new TaskCompletionSource();
await h.StartAsync();
await Eventually.UntilAsync(() => h.Delivery.Texts.Count == 1, "the reminder send");
Assert.Single(await h.Store.LoadAsync(CancellationToken.None)); // not removed while the delivery is in progress
h.Delivery.HoldTexts.SetResult();
await h.Scheduler.IdleAsync();
Assert.Equal(("111", "⏰ Call the dentist (due 26 Sep 13:58, delivered late)"), Assert.Single(h.Delivery.Texts));
Assert.Empty(await h.Store.LoadAsync(CancellationToken.None));
Assert.Empty(h.Scheduler.List("111"));
}
[Fact]
public async Task An_overdue_one_off_within_60_s_has_no_late_note()
{
await using var h = new SchedulerHarness();
await h.Store.SaveAsync([Stored("k7m2q", JobKind.Reminder, new JobSchedule(Start.AddSeconds(-30), null))], CancellationToken.None);
await h.StartAsync();
await h.Scheduler.IdleAsync();
Assert.Equal(("111", "⏰ Call the dentist"), Assert.Single(h.Delivery.Texts));
}
[Fact]
public async Task An_overdue_prompt_gets_the_offline_note()
{
await using var h = new SchedulerHarness();
await h.Store.SaveAsync([Stored("p3xz9", JobKind.Prompt, new JobSchedule(Start.AddHours(-1), null), text: "Summarise inbox/")], CancellationToken.None);
await h.StartAsync();
await h.Scheduler.IdleAsync();
Assert.Equal("[Scheduled job \"Summarise inbox/\" (p3xz9)]\nThis was due at 26 Sep 13:03; the bot was offline.\nSummarise inbox/",
Assert.Single(h.Delivery.Prompts).Prompt);
}
[Fact]
public async Task Cron_does_not_catch_up_after_downtime()
{
await using var h = new SchedulerHarness();
var job = Stored("c1c1c", JobKind.Reminder, new JobSchedule(null, "*/10 * * * *")) with { LastRunAt = Start.AddHours(-3) };
await h.Store.SaveAsync([job], CancellationToken.None);
await h.StartAsync();
await h.Scheduler.IdleAsync();
Assert.Empty(h.Delivery.Texts);
Assert.Equal(new DateTimeOffset(2026, 9, 26, 12, 10, 0, TimeSpan.Zero), Assert.Single(h.Scheduler.List("111")).NextRunAt);
}
[Fact]
public async Task Advancing_the_clock_fires_and_the_1_h_cap_rearms_without_firing()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(new NewJob("Later", JobKind.Reminder, At: "2026-09-26T17:03", Text: "Later")); // 15:03Z, 3 h away
Assert.Equal(Start.AddHours(1), h.Scheduler.NextWakeAt);
await h.AdvanceAsync(TimeSpan.FromHours(1));
Assert.Empty(h.Delivery.Texts);
Assert.Equal(Start.AddHours(2), h.Scheduler.NextWakeAt);
await h.AdvanceAsync(TimeSpan.FromHours(1));
Assert.Empty(h.Delivery.Texts);
Assert.Equal(Start.AddHours(3), h.Scheduler.NextWakeAt);
await h.AdvanceAsync(TimeSpan.FromHours(1));
Assert.Equal(("111", "⏰ Later"), Assert.Single(h.Delivery.Texts));
Assert.Empty(h.Scheduler.List("111"));
}
[Fact]
public async Task A_reminder_is_sent_as_text_and_a_prompt_as_a_prompt()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(new NewJob("Stretch", JobKind.Reminder, At: "2026-09-26T14:04", Text: "Stretch"));
var prompt = await h.CreateAsync(new NewJob("Summary", JobKind.Prompt, At: "2026-09-26T14:04", Text: "Summarise inbox/"));
await h.AdvanceAsync(Minute);
Assert.Equal(("111", "⏰ Stretch"), Assert.Single(h.Delivery.Texts));
Assert.Equal(("111", $"[Scheduled job \"Summary\" ({prompt.Job.Id})]\nSummarise inbox/"), Assert.Single(h.Delivery.Prompts));
}
[Fact]
public async Task A_silent_watcher_makes_no_delivery_calls_and_only_updates_its_run_fields()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
var created = (await h.CreateAsync(Watcher())).Job;
h.Runner.Then(Enumerable.Repeat<ScriptOutcome>(new ScriptOutcome.Silent(), 10).Append(new ScriptOutcome.Output("price 48")).ToArray());
for (var i = 0; i < 10; i++) await h.AdvanceAsync(Minute);
Assert.Equal(10, h.Runner.Requests.Count);
Assert.Empty(h.Delivery.Prompts);
Assert.Empty(h.Delivery.Texts);
Assert.DoesNotContain(h.Delivery.Calls, c => c.StartsWith("RunPrompt", StringComparison.Ordinal) || c.StartsWith("SendText", StringComparison.Ordinal));
var stored = Assert.Single(await h.Store.LoadAsync(CancellationToken.None));
Assert.Equal(created, stored with { LastRunAt = null, LastResult = null }); // nothing but the run fields changed
Assert.Equal(Start.AddMinutes(10), stored.LastRunAt);
Assert.True(stored.LastResult!.Ok);
await h.AdvanceAsync(Minute);
Assert.Equal(("111", $"[Scheduled job \"Price check\" ({created.Id})] The watcher script printed:\nprice 48"), Assert.Single(h.Delivery.Prompts));
Assert.Empty(h.Delivery.Texts);
}
[Fact]
public async Task A_failure_is_reported_once_per_streak()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
var id = (await h.CreateAsync(Watcher())).Job.Id;
h.Runner.Then(
new ScriptOutcome.Failed(1, "boom", false),
new ScriptOutcome.Failed(1, "boom again", false),
new ScriptOutcome.Failed(2, "other", false),
new ScriptOutcome.Failed(null, "slow", true),
new ScriptOutcome.Silent(),
new ScriptOutcome.Failed(2, "back", false));
var counts = new List<int>();
for (var i = 0; i < 6; i++)
{
await h.AdvanceAsync(Minute);
counts.Add(h.Delivery.Texts.Count);
}
Assert.Equal([1, 1, 2, 3, 3, 4], counts);
Assert.Equal($"⚠️ Job \"Price check\" ({id}) failed (exit 1):\n```\nboom\n```", h.Delivery.Texts[0].Text);
Assert.Equal($"⚠️ Job \"Price check\" ({id}) failed (timed out):\n```\nslow\n```", h.Delivery.Texts[2].Text);
Assert.Empty(h.Delivery.Prompts); // failures never go through the model
}
[Fact]
public async Task Overlapping_occurrences_are_skipped_while_a_watcher_runs()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(Watcher());
h.Runner.Hold = new TaskCompletionSource();
h.Time.Advance(Minute);
await Eventually.UntilAsync(() => h.Runner.Requests.Count == 1, "the first run");
h.Time.Advance(Minute);
h.Time.Advance(Minute);
await Task.Delay(100);
Assert.Single(h.Runner.Requests);
h.Runner.Hold.SetResult();
h.Runner.Hold = null;
await h.Scheduler.IdleAsync();
await h.AdvanceAsync(Minute);
Assert.Equal(2, h.Runner.Requests.Count);
}
[Fact]
public async Task Overlapping_occurrences_are_skipped_while_a_prompt_run_is_held_open()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(new NewJob("Every minute", JobKind.Prompt, Cron: "* * * * *", Text: "tick"));
h.Delivery.HoldPrompts = new TaskCompletionSource();
h.Time.Advance(Minute);
await Eventually.UntilAsync(() => h.Delivery.Prompts.Count == 1, "the first prompt");
h.Time.Advance(Minute);
await Task.Delay(100);
Assert.Single(h.Delivery.Prompts);
h.Delivery.HoldPrompts.SetResult();
h.Delivery.HoldPrompts = null;
await h.Scheduler.IdleAsync();
await h.AdvanceAsync(Minute);
Assert.Equal(2, h.Delivery.Prompts.Count);
}
[Fact]
public async Task Targets_are_isolated_in_List_and_CancelAsync()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
var mine = (await h.CreateAsync(new NewJob("Mine", JobKind.Reminder, Cron: "@daily", Text: "a"), "111")).Job;
var theirs = (await h.CreateAsync(new NewJob("Theirs", JobKind.Reminder, Cron: "@daily", Text: "b"), "222")).Job;
Assert.Equal([mine.Id], h.Scheduler.List("111").Select(v => v.Job.Id));
Assert.False(await h.Scheduler.CancelAsync("222", mine.Id, CancellationToken.None));
Assert.True(await h.Scheduler.CancelAsync("111", mine.Id, CancellationToken.None));
Assert.False(await h.Scheduler.CancelAsync("111", mine.Id, CancellationToken.None));
Assert.Equal([theirs.Id], (await h.Store.LoadAsync(CancellationToken.None)).Select(j => j.Id));
}
[Fact]
public async Task Cancel_kills_a_running_watcher_and_drops_its_output()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
var id = (await h.CreateAsync(Watcher())).Job.Id;
h.Runner.Fallback = new ScriptOutcome.Output("late output");
h.Runner.IgnoreCancellation = true; // the outcome still arrives after the cancel; it must be thrown away
h.Runner.Hold = new TaskCompletionSource();
h.Time.Advance(Minute);
await Eventually.UntilAsync(() => h.Runner.Requests.Count == 1, "the run");
Assert.True(await h.Scheduler.CancelAsync("111", id, CancellationToken.None));
Assert.Equal(1, h.Runner.CancellationsSeen);
h.Runner.Hold.SetResult();
await h.Scheduler.IdleAsync();
Assert.Empty(h.Delivery.Prompts);
Assert.Empty(await h.Store.LoadAsync(CancellationToken.None));
}
[Fact]
public async Task An_undeliverable_target_is_skipped_and_its_job_kept()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(new NewJob("Hourly", JobKind.Reminder, Cron: "0 * * * *", Text: "ping"));
await h.CreateAsync(new NewJob("Once", JobKind.Reminder, At: "2026-09-26T14:30", Text: "once"));
h.Delivery.Undeliverable.Add("111");
await h.AdvanceAsync(TimeSpan.FromMinutes(57)); // past 14:30 and 15:00 Berlin
Assert.Empty(h.Delivery.Texts);
var jobs = h.Scheduler.List("111");
Assert.Equal(2, jobs.Count);
Assert.Equal(new DateTimeOffset(2026, 9, 26, 14, 0, 0, TimeSpan.Zero), jobs.Single(j => j.Job.Name == "Hourly").NextRunAt); // advanced
Assert.Equal(2, (await h.Store.LoadAsync(CancellationToken.None)).Count);
}
[Fact]
public async Task StopAsync_kills_scripts_and_returns()
{
var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(Watcher());
h.Runner.Hold = new TaskCompletionSource();
h.Time.Advance(Minute);
await Eventually.UntilAsync(() => h.Runner.Requests.Count == 1, "the run");
await h.Scheduler.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(5));
Assert.Equal(1, h.Runner.CancellationsSeen);
await h.DisposeAsync();
}
[Fact]
public async Task StopAsync_gives_up_after_the_grace_period()
{
var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(Watcher());
h.Runner.IgnoreCancellation = true;
h.Runner.Hold = new TaskCompletionSource();
h.Time.Advance(Minute);
await Eventually.UntilAsync(() => h.Runner.Requests.Count == 1, "the run");
var stop = h.Scheduler.StopAsync(CancellationToken.None);
await Task.Delay(100);
Assert.False(stop.IsCompleted);
h.Time.Advance(TimeSpan.FromSeconds(10));
await stop.WaitAsync(TimeSpan.FromSeconds(5));
h.Runner.Hold.SetResult();
await h.DisposeAsync();
}
[Fact]
public async Task A_zone_change_recomputes_cron_and_leaves_one_offs()
{
var start = new DateTimeOffset(2026, 9, 26, 23, 30, 0, TimeSpan.Zero); // Sun 01:30 Berlin, Sun 08:30 Tokyo
await using var h = new SchedulerHarness(start);
await h.StartAsync();
var cron = (await h.CreateAsync(new NewJob("Morning", JobKind.Reminder, Cron: "0 9 * * *", Text: "Good morning"))).Job;
var once = (await h.CreateAsync(new NewJob("Dentist", JobKind.Reminder, At: "2026-09-27T10:00", Text: "Dentist"))).Job;
var other = (await h.CreateAsync(new NewJob("Other", JobKind.Reminder, Cron: "0 9 * * *", Text: "x"), "222")).Job;
var berlinNine = new DateTimeOffset(2026, 9, 27, 7, 0, 0, TimeSpan.Zero);
Assert.Equal(start.AddHours(1), h.Scheduler.NextWakeAt); // capped
h.Delivery.Zones["111"] = SchedulerHarness.Tokyo;
var change = await h.Scheduler.TimeZoneChangedAsync("111", CancellationToken.None);
Assert.Equal(new ZoneChange(1, 1), change);
var views = h.Scheduler.List("111").ToDictionary(v => v.Job.Id);
Assert.Equal(new DateTimeOffset(2026, 9, 27, 0, 0, 0, TimeSpan.Zero), views[cron.Id].NextRunAt); // 09:00 Tokyo
Assert.Equal(new DateTimeOffset(2026, 9, 27, 8, 0, 0, TimeSpan.Zero), views[once.Id].NextRunAt); // unchanged instant
Assert.Equal(berlinNine, Assert.Single(h.Scheduler.List("222")).NextRunAt);
Assert.Equal(new DateTimeOffset(2026, 9, 27, 0, 0, 0, TimeSpan.Zero), h.Scheduler.NextWakeAt); // re-armed
Assert.Equal(other.Id, h.Scheduler.List("222")[0].Job.Id);
}
[Fact]
public async Task Watcher_output_and_failure_tails_are_redacted()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(Watcher());
h.Runner.Then(new ScriptOutcome.Output("token=hunter2-secret"), new ScriptOutcome.Failed(1, "auth hunter2-secret rejected", false));
await h.AdvanceAsync(Minute);
await h.AdvanceAsync(Minute);
var prompt = Assert.Single(h.Delivery.Prompts).Prompt;
Assert.EndsWith("token=[REDACTED]", prompt);
var alert = Assert.Single(h.Delivery.Texts).Text;
Assert.Contains("auth [REDACTED] rejected", alert);
Assert.DoesNotContain("hunter2", prompt + alert);
Assert.Equal("auth [REDACTED] rejected", Assert.Single(await h.Store.LoadAsync(CancellationToken.None)).LastResult!.StderrTail);
}
[Fact]
public async Task Delivery_failures_are_logged_and_the_job_still_advances()
{
await using var h = new SchedulerHarness();
await h.StartAsync();
await h.CreateAsync(new NewJob("Tick", JobKind.Reminder, Cron: "* * * * *", Text: "tick"));
h.Delivery.FailSends = new InvalidOperationException("Telegram is down");
await h.AdvanceAsync(Minute);
var view = Assert.Single(h.Scheduler.List("111"));
Assert.Equal(Start.AddMinutes(1), view.Job.LastRunAt);
Assert.Equal(Start.AddMinutes(2), view.NextRunAt);
}
[Fact]
public async Task CreateAsync_validates_and_caps_jobs_per_target()
{
await using var h = new SchedulerHarness(configure: o => o.MaxJobsPerTarget = 2);
await h.StartAsync();
await h.CreateAsync(new NewJob("a", JobKind.Reminder, Cron: "@daily", Text: "a"));
await h.CreateAsync(new NewJob("b", JobKind.Reminder, Cron: "@daily", Text: "b"));
var full = await Assert.ThrowsAsync<JobValidationException>(() => h.CreateAsync(new NewJob("c", JobKind.Reminder, Cron: "@daily", Text: "c")));
Assert.Equal("This chat already has 2 scheduled jobs, the maximum. Cancel one first.", full.Message);
await h.CreateAsync(new NewJob("c", JobKind.Reminder, Cron: "@daily", Text: "c"), "222"); // other targets have their own cap
await Assert.ThrowsAsync<JobValidationException>(() => h.CreateAsync(new NewJob("bad", JobKind.Reminder, Cron: "0 0 9 * * *", Text: "x"), "333"));
var ids = h.Scheduler.List("111").Select(v => v.Job.Id).ToList();
Assert.All(ids, id => Assert.Matches("^[abcdefghjkmnpqrstuvwxyz23456789]{5}$", id));
}
[Fact]
public async Task Watcher_runs_with_the_target_workspace_zone_and_timeout()
{
await using var h = new SchedulerHarness();
h.Delivery.Zones["111"] = SchedulerHarness.Tokyo;
await h.StartAsync();
var job = (await h.CreateAsync(Watcher())).Job;
await h.CreateAsync(new NewJob("Quick", JobKind.Watcher, Cron: "* * * * *", Command: "true", TimeoutSeconds: 5), "222");
await h.AdvanceAsync(Minute);
var requests = h.Runner.Requests.OrderBy(r => r.Target).ToList();
Assert.Equal(new ScriptRequest("scripts/price.sh", Path.Combine(h.Delivery.WorkspaceRoot, "111"), TimeSpan.FromSeconds(300), "111", job.Id, "Asia/Tokyo"), requests[0]);
Assert.Equal(TimeSpan.FromSeconds(5), requests[1].Timeout);
Assert.Equal("Europe/Berlin", requests[1].TimeZoneId);
}
}

155
tests/Volcanic.AgenticCore.Extensions.Tests/Scheduler/Support/SchedulerFakes.cs Normal file
View File

@@ -0,0 +1,155 @@
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Time.Testing;
using Volcanic.AgenticCore.Abstractions.Processes;
using Volcanic.AgenticCore.Extensions.Scheduler;
namespace Volcanic.AgenticCore.Extensions.Tests.Scheduler.Support;
/// <summary>Records every delivery call; can hold sends and prompt runs open, fail sends, and switch a target's zone.</summary>
internal sealed class FakeJobDelivery : IJobDelivery
{
private readonly Lock _lock = new();
private readonly List<(string Target, string Text)> _texts = [];
private readonly List<(string Target, string Prompt)> _prompts = [];
private readonly List<string> _calls = [];
public Dictionary<string, TimeZoneInfo> Zones { get; } = new();
public TimeZoneInfo DefaultZone { get; set; } = TimeZoneInfo.FindSystemTimeZoneById("Europe/Berlin");
public HashSet<string> Undeliverable { get; } = [];
public TaskCompletionSource? HoldTexts { get; set; }
public TaskCompletionSource? HoldPrompts { get; set; }
public Exception? FailSends { get; set; }
public string WorkspaceRoot { get; set; } = Path.Combine(Path.GetTempPath(), "fake-workspaces");
public IReadOnlyList<(string Target, string Text)> Texts { get { lock (_lock) return _texts.ToList(); } }
public IReadOnlyList<(string Target, string Prompt)> Prompts { get { lock (_lock) return _prompts.ToList(); } }
/// <summary>"SendText:111", "RunPrompt:111", "WorkspaceFor:111", "TimeZoneFor:111", "IsDeliverable:111".</summary>
public IReadOnlyList<string> Calls { get { lock (_lock) return _calls.ToList(); } }
public async Task SendTextAsync(string target, string markdown, CancellationToken cancellationToken)
{
lock (_lock) { _calls.Add("SendText:" + target); _texts.Add((target, markdown)); }
if (HoldTexts is { } hold) await hold.Task.WaitAsync(cancellationToken);
if (FailSends is { } error) throw error;
}
public async Task RunPromptAsync(string target, string prompt, CancellationToken cancellationToken)
{
lock (_lock) { _calls.Add("RunPrompt:" + target); _prompts.Add((target, prompt)); }
if (HoldPrompts is { } hold) await hold.Task.WaitAsync(cancellationToken);
}
public string WorkspaceFor(string target)
{
lock (_lock) _calls.Add("WorkspaceFor:" + target);
return Path.Combine(WorkspaceRoot, target);
}
public bool IsDeliverable(string target)
{
lock (_lock) _calls.Add("IsDeliverable:" + target);
return !Undeliverable.Contains(target);
}
public TimeZoneInfo TimeZoneFor(string target)
{
lock (_lock) _calls.Add("TimeZoneFor:" + target);
return Zones.GetValueOrDefault(target, DefaultZone);
}
}
/// <summary>Returns queued outcomes (then <see cref="Fallback"/>); can hold runs open, honouring or ignoring cancellation.</summary>
internal sealed class FakeScriptRunner : IScriptRunner
{
private readonly Lock _lock = new();
private readonly Queue<ScriptOutcome> _outcomes = new();
private readonly List<ScriptRequest> _requests = [];
private int _cancellations;
public ScriptOutcome Fallback { get; set; } = new ScriptOutcome.Silent();
public TaskCompletionSource? Hold { get; set; }
public bool IgnoreCancellation { get; set; }
public IReadOnlyList<ScriptRequest> Requests { get { lock (_lock) return _requests.ToList(); } }
public int CancellationsSeen => Volatile.Read(ref _cancellations);
public FakeScriptRunner Then(params ScriptOutcome[] outcomes)
{
lock (_lock) foreach (var o in outcomes) _outcomes.Enqueue(o);
return this;
}
public async Task<ScriptOutcome> RunAsync(ScriptRequest request, CancellationToken cancellationToken)
{
ScriptOutcome outcome;
lock (_lock)
{
_requests.Add(request);
outcome = _outcomes.Count > 0 ? _outcomes.Dequeue() : Fallback;
}
if (Hold is { } hold)
{
if (IgnoreCancellation)
{
using var _ = cancellationToken.Register(() => Interlocked.Increment(ref _cancellations));
await hold.Task;
}
else
{
try { await hold.Task.WaitAsync(cancellationToken); }
catch (OperationCanceledException) { Interlocked.Increment(ref _cancellations); throw; }
}
}
return outcome;
}
}
internal sealed class FakeRedactor(params string[] secrets) : ISecretRedactor
{
public string Redact(string text) => secrets.Aggregate(text, (t, s) => t.Replace(s, "[REDACTED]", StringComparison.Ordinal));
}
/// <summary>A <see cref="JobScheduler"/> on a <see cref="FakeTimeProvider"/> at Sat 2026-09-26 14:03 Berlin, a temp store and the fakes.</summary>
internal sealed class SchedulerHarness : IAsyncDisposable
{
public static readonly DateTimeOffset Start = new(2026, 9, 26, 12, 3, 0, TimeSpan.Zero);
public static readonly TimeZoneInfo Berlin = TimeZoneInfo.FindSystemTimeZoneById("Europe/Berlin");
public static readonly TimeZoneInfo Tokyo = TimeZoneInfo.FindSystemTimeZoneById("Asia/Tokyo");
public SchedulerHarness(DateTimeOffset? start = null, Action<SchedulerOptions>? configure = null)
{
Time = new FakeTimeProvider(start ?? Start);
Options = new SchedulerOptions { StorePath = Path.Combine(Root, "scheduler", "jobs.json") };
configure?.Invoke(Options);
Delivery.WorkspaceRoot = Path.Combine(Root, "workspaces");
Scheduler = new JobScheduler(Options, Runner, Redactor, Time, NullLogger<JobScheduler>.Instance);
}
public string Root { get; } = Directory.CreateTempSubdirectory("scheduler-").FullName;
public FakeTimeProvider Time { get; }
public SchedulerOptions Options { get; }
public FakeJobDelivery Delivery { get; } = new();
public FakeScriptRunner Runner { get; } = new();
public FakeRedactor Redactor { get; } = new("hunter2-secret");
public JobScheduler Scheduler { get; }
public JobStore Store => new(Options.StorePath!, Time, NullLogger.Instance);
public Task StartAsync() => Scheduler.StartAsync(Delivery, CancellationToken.None);
public Task<JobView> CreateAsync(NewJob job, string target = "111") => Scheduler.CreateAsync(target, job, CancellationToken.None);
/// <summary>Advance the clock and wait for the tick and every fire it started (don't use while a fire is held open).</summary>
public async Task AdvanceAsync(TimeSpan by)
{
Time.Advance(by);
await Scheduler.IdleAsync();
}
public async ValueTask DisposeAsync()
{
Runner.Hold?.TrySetResult();
Delivery.HoldTexts?.TrySetResult();
Delivery.HoldPrompts?.TrySetResult();
await Scheduler.StopAsync(CancellationToken.None);
try { Directory.Delete(Root, recursive: true); } catch (IOException) { }
}
}