VolcanicAgenticCore/extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs

using System.Globalization;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Abstractions.Processes;
using Volcanic.AgenticCore.Abstractions.Skills;

namespace Volcanic.AgenticCore.Extensions.Scheduler;

/// <summary>
/// The process-wide scheduler: loads every known target's <c>jobs.json</c> (<see cref="JobStore.PathIn"/> in its
/// workspace; saves rewrite only the changed target's file), 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.
/// <para>
/// Skill jobs (<c>.volcanic/skills/&lt;skill&gt;/jobs/&lt;job&gt;.md</c>, id <c>&lt;skill&gt;/&lt;job&gt;</c>) run on the same
/// timer, but their source of truth is the file: <see cref="SyncSkillJobsAsync"/> re-reads a target's files (on start,
/// every <see cref="SchedulerOptions.SkillJobSyncInterval"/>, around agent runs and before they are listed) and adds,
/// re-plans or removes them. They are never written to <c>jobs.json</c>, can't be cancelled, and keep their run fields
/// in <c>skill-jobs-state.json</c>.
/// </para>
/// </summary>
public sealed partial 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);
    public const string WorkspaceVariable = SkillHookFiles.WorkspaceVariable;
    public const string SkillDirVariable = SkillHookFiles.SkillDirVariable;
    private const string IdAlphabet = "abcdefghjkmnpqrstuvwxyz23456789";

    private readonly SchedulerOptions _options;
    private readonly IScriptRunner _runner;
    private readonly ISecretRedactor _redactor;
    private readonly IScriptEnvironment? _scriptEnvironment;
    private readonly TimeProvider _time;
    private readonly ILogger _logger;
    private readonly Dictionary<string, JobStore> _stores = new(StringComparer.Ordinal);
    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 ITimer? _syncTimer;
    private Task _pendingSync = Task.CompletedTask;
    private readonly SemaphoreSlim _syncGate = new(1, 1);
    private readonly Dictionary<string, List<SkillJobView>> _skillViews = new(StringComparer.Ordinal);
    private readonly Dictionary<string, Dictionary<string, SkillJobState>> _skillStates = new(StringComparer.Ordinal);
    private DateTimeOffset? _wakeAt;
    private bool _stopped;
    private Task _pendingTick = Task.CompletedTask;
    private int _firesStarted;

    public JobScheduler(SchedulerOptions options, IScriptRunner runner, ISecretRedactor redactor, TimeProvider time, ILogger<JobScheduler> logger,
        IScriptEnvironment? scriptEnvironment = null)
    {
        ArgumentNullException.ThrowIfNull(options);
        _options = options;
        _runner = runner;
        _redactor = redactor;
        _scriptEnvironment = scriptEnvironment;
        _time = time;
        _logger = logger;
    }

    public bool IsStarted { get { lock (_lock) return _delivery is not null && !_stopped; } }
    public DateTimeOffset Now => _time.GetUtcNow();
    /// <summary>How many jobs are loaded, across all targets.</summary>
    public int Count { get { lock (_lock) return _jobs.Count; } }

    /// <summary>When the timer fires next (tests).</summary>
    internal DateTimeOffset? NextWakeAt { get { lock (_lock) return _wakeAt; } }

    /// <summary>How many fires have been started (tests: a skipped occurrence doesn't count).</summary>
    internal int FiresStarted { get { lock (_lock) return _firesStarted; } }

    /// <summary>The latest queued tick, without the fires it started (tests; unlike <see cref="IdleAsync"/> it doesn't wait for held fires).</summary>
    internal Task PendingTick { get { lock (_lock) return _pendingTick; } }

    /// <summary>Load every target's file (moving the retired shared file's jobs in first), 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 LoadAllAsync(delivery, 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
                {
                    try
                    {
                        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);
                    }
                    catch (Exception ex)
                    {
                        // One bad entry never keeps the host down: it stays listed but unarmed until a zone change recomputes it.
                        entry.Next = null;
                        _logger.LogError(ex, "Job {Id} ({Name}): computing the next occurrence on start failed; the job isn't armed", job.Id, job.Name);
                    }
                }
                _jobs[job.Id] = entry;
            }
            _timer = _time.CreateTimer(_ => OnTimer(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
        }
        await SyncAllSkillJobsAsync().ConfigureAwait(false);
        lock (_lock)
        {
            if (!_stopped)
                _syncTimer = _time.CreateTimer(_ => OnSyncTimer(), null, _options.SkillJobSyncInterval, _options.SkillJobSyncInterval);
        }
        Tick(); // overdue one-offs fire now
    }

    // ---- Loading ----

    /// <summary>
    /// Every known target's jobs. The retired shared file's jobs are merged into their targets' files first (an id the
    /// target already has is skipped, so an interrupted move is safe to repeat), then the shared file is renamed. An id
    /// taken by another target gets a new one. Changed files are saved.
    /// </summary>
    private async Task<List<ScheduledJob>> LoadAllAsync(IJobDelivery delivery, CancellationToken cancellationToken)
    {
        var byTarget = new Dictionary<string, List<ScheduledJob>>(StringComparer.Ordinal);
        var changed = new HashSet<string>(StringComparer.Ordinal);
        foreach (var target in delivery.KnownTargets().Distinct(StringComparer.Ordinal))
            await TargetJobsAsync(target).ConfigureAwait(false);

        var legacy = _options.LegacyStorePath;
        if (!string.IsNullOrWhiteSpace(legacy) && File.Exists(legacy))
        {
            var moved = 0;
            foreach (var job in await new JobStore(legacy, _time, _logger).LoadAsync(cancellationToken).ConfigureAwait(false))
            {
                var list = await TargetJobsAsync(job.Target).ConfigureAwait(false);
                if (list is null || list.Any(j => j.Id == job.Id)) continue;
                list.Add(job);
                changed.Add(job.Target);
                moved++;
            }
            await SaveAsync().ConfigureAwait(false);
            if (File.Exists(legacy)) // not set aside as unreadable
            {
                var renamed = $"{legacy}.migrated-{_time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}";
                File.Move(legacy, renamed, overwrite: true);
                _logger.LogInformation("Moved {Count} job(s) from the shared job file {Path} into the chats' files; it is now {Renamed}", moved, legacy, renamed);
            }
        }

        var ids = new HashSet<string>(StringComparer.Ordinal);
        foreach (var (target, list) in byTarget)
            for (var i = 0; i < list.Count; i++)
            {
                if (ids.Add(list[i].Id)) continue;
                var id = NewId(ids.Contains);
                _logger.LogWarning("Job {Id} ({Name}) of target {Target}: the id is taken by another target; it is now {NewId}", list[i].Id, list[i].Name, target, id);
                list[i] = list[i] with { Id = id };
                ids.Add(id);
                changed.Add(target);
            }
        await SaveAsync().ConfigureAwait(false);
        return [.. byTarget.Values.SelectMany(j => j)];

        // Null when the target has no usable workspace (the host can't map it): its jobs are skipped.
        async Task<List<ScheduledJob>?> TargetJobsAsync(string target)
        {
            if (byTarget.TryGetValue(target, out var list)) return list;
            JobStore store;
            try
            {
                store = StoreFor(delivery, target);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Target {Target} has no workspace for its job file; its jobs are skipped", target);
                return null;
            }
            return byTarget[target] = [.. await store.LoadAsync(cancellationToken).ConfigureAwait(false)];
        }

        async Task SaveAsync()
        {
            foreach (var target in changed)
                await StoreFor(delivery, target).SaveAsync(byTarget[target], cancellationToken).ConfigureAwait(false);
            changed.Clear();
        }
    }

    private JobStore StoreFor(IJobDelivery delivery, string target)
    {
        lock (_lock)
        {
            if (!_stores.TryGetValue(target, out var store))
                _stores[target] = store = new JobStore(JobStore.PathIn(delivery.WorkspaceFor(target)), _time, _logger, target);
            return store;
        }
    }

    /// <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();
        ITimer? syncTimer;
        lock (_lock) syncTimer = _syncTimer;
        syncTimer?.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 && e.Job.Skill is null)
                .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 && e.Job.Skill is null) >= _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(target, 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) =>
        await CancelJobAsync(target, id, cancellationToken).ConfigureAwait(false) is not null;

    /// <summary><see cref="CancelAsync"/> returning the removed job (null: unknown, another target's, or a skill job); looked up and removed in one step.</summary>
    public async Task<ScheduledJob?> CancelJobAsync(string target, string id, CancellationToken cancellationToken)
    {
        Fire? running;
        ScheduledJob removed;
        lock (_lock)
        {
            if (!_jobs.TryGetValue(id, out var entry) || entry.Job.Target != target || entry.Job.Skill is not null) return null;
            removed = entry.Job;
            _jobs.Remove(id);
            _inFlight.TryGetValue(id, out running);
        }
        if (running is not null) await TryCancelAsync(running.Cancel).ConfigureAwait(false);
        await PersistAsync(target, cancellationToken).ConfigureAwait(false);
        Arm();
        return removed;
    }

    /// <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, sync;
            Task[] fires;
            lock (_lock)
            {
                tick = _pendingTick;
                sync = _pendingSync;
                fires = _inFlight.Values.Select(f => f.Task).ToArray();
            }
            if (tick.IsCompleted && sync.IsCompleted && fires.All(f => f.IsCompleted)) return;
            await Task.WhenAll(fires.Append(tick).Append(sync)).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(KeyOf(entry.Job)))
                        _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();
        }
    }

    /// <summary>
    /// Computes the wake time and applies it to the timer under the lock, so two concurrent arms can't apply an older
    /// wake after a newer one. A synchronous callback from <c>Change</c> (zero delay) is safe: <see cref="Lock"/> is
    /// re-entrant and <see cref="OnTimer"/> only queues a tick.
    /// </summary>
    private void Arm()
    {
        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;
            try
            {
                _timer.Change(wake - now, Timeout.InfiniteTimeSpan);
            }
            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[KeyOf(job)] = new Fire(task, cancel);
        _firesStarted++;
        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(KeyOf(job));
            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.Prompt:
                result = await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.Prompt(job, lateDue, zone), ct)).ConfigureAwait(false)
                    ? new JobResult(true, null, false, null, _time.GetUtcNow())
                    : null;
                break;
            default:
                result = await RunWatcherAsync(job, delivery, zone, scriptToken).ConfigureAwait(false);
                break;
        }
        if (result is not null) await RecordAsync(job, result).ConfigureAwait(false);
    }

    /// <summary>Null when the run or its delivery 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;
        IReadOnlyDictionary<string, string>? extra;
        ScriptOutcome outcome;
        try
        {
            extra = await ScriptEnvironmentAsync(job, timeout, scriptToken).ConfigureAwait(false);
            var workspace = delivery.WorkspaceFor(job.Target);
            var request = job.Skill is { } skill
                ? new ScriptRequest(job.Command!, skill.Directory, timeout, job.Target, job.Id, ZoneNames.Iana(zone), SkillEnvironment(extra, workspace, skill))
                : new ScriptRequest(job.Command!, workspace, timeout, job.Target, job.Id, ZoneNames.Iana(zone), extra);
            outcome = await _runner.RunAsync(request, scriptToken).ConfigureAwait(false);
        }
        catch (OperationCanceledException) when (scriptToken.IsCancellationRequested)
        {
            return null;
        }
        string Redact(string text) => RedactRun(text, extra);
        if (scriptToken.IsCancellationRequested || !Exists(KeyOf(job))) 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.
                return await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.WatcherMatch(job, Redact(output.Text)), ct)).ConfigureAwait(false)
                    ? new JobResult(true, 0, false, null, at)
                    : null;
            case ScriptOutcome.Failed failed:
                var result = new JobResult(false, failed.ExitCode, failed.TimedOut, Redact(failed.StderrTail), at);
                if (!IsRepeat(LastResult(KeyOf(job)), result)
                    && !await DeliverAsync(job, ct => delivery.SendTextAsync(job.Target, JobMessages.Failure(job, result), ct)).ConfigureAwait(false))
                    return null; // the alert was cut off by shutdown: not recorded, so the next failure alerts again
                return result;
            default:
                return new JobResult(true, 0, false, null, at); // silent: nothing but LastRunAt/LastResult
        }
    }

    /// <summary>A skill watcher's variables: the host's, plus <c>VOLCANIC_WORKSPACE</c> and <c>VOLCANIC_SKILL_DIR</c> (paths, not redacted).</summary>
    private static Dictionary<string, string> SkillEnvironment(IReadOnlyDictionary<string, string>? extra, string workspace, SkillJobSource skill)
    {
        var variables = extra is null ? new Dictionary<string, string>(StringComparer.Ordinal) : new Dictionary<string, string>(extra, StringComparer.Ordinal);
        variables[WorkspaceVariable] = Path.GetFullPath(workspace);
        variables[SkillDirVariable] = Path.GetFullPath(skill.Directory);
        return variables;
    }

    /// <summary>The host's extra variables for this run; null when there are none or the host failed (logged).</summary>
    private async Task<IReadOnlyDictionary<string, string>?> ScriptEnvironmentAsync(ScheduledJob job, TimeSpan timeout, CancellationToken scriptToken)
    {
        if (_scriptEnvironment is null) return null;
        try
        {
            var extra = await _scriptEnvironment.ForRunAsync(job.Target, timeout, scriptToken).ConfigureAwait(false);
            return extra.Count == 0 ? null : extra;
        }
        catch (Exception ex) when (ex is not OperationCanceledException || !scriptToken.IsCancellationRequested)
        {
            _logger.LogWarning(ex, "Job {Id} ({Name}): the script environment failed; the watcher runs without it", job.Id, job.Name);
            return null;
        }
    }

    /// <summary>The host's secrets, then this run's extra values (8 characters or more, like the policy's secret values).</summary>
    private string RedactRun(string text, IReadOnlyDictionary<string, string>? extra)
    {
        text = _redactor.Redact(text);
        if (extra is null) return text;
        foreach (var value in extra.Values.Where(v => v.Length >= 8).OrderByDescending(v => v.Length))
            text = text.Replace(value, "[REDACTED]", StringComparison.Ordinal);
        return text;
    }

    /// <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). False only
    /// when shutdown interrupted the delivery: the fire then isn't recorded, so a one-off stays stored and fires again
    /// after the next start (at-least-once).
    /// </summary>
    private async Task<bool> 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
            return true;
        }
        catch (OperationCanceledException) when (_stopping.IsCancellationRequested)
        {
            _logger.LogInformation("Job {Id} ({Name}): delivery interrupted by shutdown; kept for the next start", job.Id, job.Name);
            return false;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Job {Id} ({Name}): delivery to {Target} failed", job.Id, job.Name, job.Target);
            return true;
        }
    }

    /// <summary>Update and persist the run fields; a one-off is removed now that its fire has completed.</summary>
    private async Task RecordAsync(ScheduledJob job, JobResult result)
    {
        if (job.Skill is not null)
        {
            await RecordSkillJobAsync(job, result).ConfigureAwait(false);
            return;
        }
        lock (_lock)
        {
            if (!_jobs.TryGetValue(job.Id, out var entry)) return; // cancelled meanwhile
            if (entry.Job.Schedule.At is not null) _jobs.Remove(job.Id);
            else entry.Job = entry.Job with { LastRunAt = result.At, LastResult = result };
        }
        try
        {
            await PersistAsync(job.Target, 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 target's current jobs to its file. The snapshot is taken inside the gate, so a later save never writes older state.</summary>
    private async Task PersistAsync(string target, CancellationToken cancellationToken)
    {
        await _saveGate.WaitAsync(cancellationToken).ConfigureAwait(false);
        try
        {
            List<ScheduledJob> snapshot;
            lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).Where(j => j.Target == target && j.Skill is null).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)];
            await StoreFor(_delivery!, target).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() => NewId(id => _jobs.ContainsKey(id) || _inFlight.ContainsKey(id));

    private static string NewId(Func<string, bool> taken)
    {
        string id;
        do id = new string(Random.Shared.GetItems(IdAlphabet.AsSpan(), 5));
        while (taken(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);
}