schedule_delete_operation

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

using System.Globalization;
using System.Text.Json;
using Lilith.Core.Tools;

namespace Lilith.Core.Scheduling;

internal sealed class ScheduleTool(ScheduleSession session, string operation, IReadOnlyDictionary<string, string> inputs,
    Action<ScheduleResult> completed) : ITurnResultTool
{
    public string Name => "schedule_" + operation + "_operation";
    public string Description => "Perform the scenario's schedule operation once using its bound inputs. Call with {}. Never invent success.";
    public TimeSpan ExecutionTimeout => TimeSpan.FromSeconds(30);
    public JsonElement Parameters => JsonSerializer.SerializeToElement(new { type = "object", properties = new { }, additionalProperties = false });
    public async Task<string> ExecuteAsync(JsonElement arguments, CancellationToken cancellationToken = default) =>
        (await ExecuteResultAsync(arguments, cancellationToken)).Content;
    public Task<ToolExecutionResult> ExecuteResultAsync(JsonElement arguments, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        if (arguments.ValueKind != JsonValueKind.Object || arguments.EnumerateObject().Any()) throw new ArgumentException("Expected an empty argument object.");
        string Get(string name, string fallback = "") => inputs.GetValueOrDefault(name, fallback);
        if (!int.TryParse(Get("seconds", "0"), NumberStyles.None, CultureInfo.InvariantCulture, out int seconds)) throw new ArgumentException("Invalid seconds.");
        var result = session.Execute(new(operation, Get("name"), Get("kind", "timer"), seconds, Get("cron"), Get("timezone"), Get("message"), Get("scenario"), Get("inputs_json", "{}")));
        completed(result);
        string json = JsonSerializer.Serialize(result);
        return Task.FromResult(new ToolExecutionResult(json, json));
    }
    internal static readonly string[] Operations = ["create", "update", "cancel", "delete", "list", "peek", "pending"];
    internal static string? Operation(string name) => Operations.FirstOrDefault(op => name == "schedule_" + op + "_operation");
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225

using System.Text.Json;
using Lilith.Core.Scenarios;
using Lilith.Core.Scheduling.Extensions;
using Lilith.Core.Workspace;
using Lilith.Core.Workspace.Extensions;

namespace Lilith.Core.Scheduling;

/// <summary>Persistent, single-host schedule state. Due detection never calls the conversation.</summary>
public sealed class ScheduleSession : IDisposable
{
    private readonly object _sync = new();
    private readonly FileStream _lease;
    private readonly string _file;
    private readonly ScenarioFileService _files;
    private readonly TimeProvider _clock;
    private ScheduleState _state;
    private bool _disposed;
    private bool _faulted;
    public ScheduleOptions Options { get; }

    public ScheduleSession(ScheduleOptions options, string? scenarioRoot = null, TimeProvider? clock = null)
    {
        if (!Enum.IsDefined(options.Mode)) throw new ArgumentException("Unknown schedule access mode.");
        Options = options with { RootDirectory = Path.GetFullPath(options.RootDirectory) };
        _clock = clock ?? TimeProvider.System;
        _files = new(scenarioRoot);
        Options.RootDirectory.EnsureNoLinks();
        Directory.CreateDirectory(Options.RootDirectory);
        _lease = new(Path.Combine(Options.RootDirectory, "host.lock"), FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
        _file = Path.Combine(Options.RootDirectory, "schedules.json");
        try
        {
            _state = File.Exists(_file) ? JsonSerializer.Deserialize<ScheduleState>(File.ReadAllText(_file))
                ?? throw new InvalidDataException("Invalid schedule state.") : new();
            Recover();
        }
        catch { _lease.Dispose(); throw; }
    }

    public ScheduleResult Execute(ScheduleRequest request)
    {
        lock (_sync)
        {
            ThrowIfDisposed();
            return request.Operation switch
            {
                "list" => new("completed", "Schedule status.", Data: _state.Entries.Take(100).Select(e => View(e, false)).ToArray()),
                "peek" => new("completed", "Schedule status.", Data: View(Find(request.Name))),
                "pending" => new("completed", "Pending schedule changes.", Data: _state.Pending.Select(p => new { p.Id, p.Request.Operation, p.Request.Name }).ToArray()),
                "create" or "update" or "cancel" or "delete" => PrepareOrCommit(request),
                _ => throw new ArgumentException("Unknown schedule operation.")
            };
        }
    }

    public ScheduleProposal Review(string id)
    {
        lock (_sync) { ThrowIfDisposed(); return _state.Pending.SingleOrDefault(p => p.Id == id) ?? throw new ArgumentException("Unknown proposal."); }
    }

    // Host-only APIs: neither approval nor rejection is registered as a model tool.
    public ScheduleResult Approve(string id)
    {
        lock (_sync)
        {
            var proposal = Review(id);
            var existing = _state.Entries.SingleOrDefault(e => e.Name.Equals(proposal.Request.Name, StringComparison.OrdinalIgnoreCase));
            if (existing?.Revision != proposal.ExpectedRevision) throw new InvalidOperationException("Schedule changed. Reject this proposal and prepare a new one.");
            Validate(proposal.Request);
            var result = Commit(proposal.Request);
            _state.Pending.Remove(proposal);
            Save();
            return result;
        }
    }

    public void Reject(string id)
    {
        lock (_sync) { var proposal = Review(id); _state.Pending.Remove(proposal); Save(); }
    }

    public void Tick()
    {
        lock (_sync)
        {
            ThrowIfDisposed();
            var now = _clock.GetUtcNow();
            var due = _state.Entries.Where(e => e.Status == "active" && e.NextUtc <= now).ToArray();
            foreach (var entry in due)
            {
                // One outstanding run per named job; coalesce late intervals rather than replaying a backlog.
                bool outstanding = _state.Events.Any(e => e.Job?.Name == entry.Name);
                Replace(entry with
                {
                    NextUtc = entry.Kind == "timer" ? null : entry.NextAfter(now),
                    Status = entry.Kind == "timer" ? "fired" : "active",
                    Revision = entry.Revision + 1
                });
                if (!outstanding) AddEvent(entry, "fired", $"{entry.Name} fired. {entry.Message}", entry);
            }
            if (due.Length > 0) Save();
        }
    }

    public ScheduleEvent? PeekEvent()
    {
        lock (_sync) { ThrowIfDisposed(); return _state.Events.FirstOrDefault(); }
    }

    public void CompleteEvent(string id, string? failure = null)
    {
        lock (_sync)
        {
            ThrowIfDisposed();
            var item = _state.Events.Single(e => e.Id == id);
            _state.Events.Remove(item);
            if (item.Job is { } job)
            {
                var entry = Find(job.Name);
                Replace(entry with { Status = entry.Kind == "timer" ? (failure is null ? "completed" : "failed") : entry.Status,
                    LastOutcome = failure is null ? "completed" : failure[..Math.Min(2000, failure.Length)], LastFinishedUtc = _clock.GetUtcNow() });
                AddEvent(job, failure is null ? "completed" : "failed",
                    failure is null ? $"{job.Name} finished." : $"{job.Name} failed: {failure}");
            }
            Save();
        }
    }

    private void Recover()
    {
        var now = _clock.GetUtcNow();
        var interrupted = _state.Events.Where(e => e.Job is not null).Select(e => e.Job!.Name).ToHashSet(StringComparer.OrdinalIgnoreCase);
        _state.Events.RemoveAll(e => e.Job is not null);
        foreach (var entry in _state.Entries.ToArray())
        {
            bool overdue = entry.Status == "active" && entry.NextUtc <= now;
            if (!overdue && !interrupted.Contains(entry.Name)) continue;
            if (!_state.Events.Any(e => e.Name == entry.Name && e.Kind == "missed"))
                AddEvent(entry, "missed", $"{entry.Name}: missed or interrupted occurrence while Lilith was closed. Its action was not replayed.");
            if (overdue || (interrupted.Contains(entry.Name) && entry.Kind == "timer")) Replace(entry with { NextUtc = entry.Kind == "timer" ? null : entry.NextAfter(now),
                Status = entry.Kind == "timer" ? "missed" : "active", Revision = entry.Revision + 1 });
        }
        Save();
    }

    private ScheduleResult PrepareOrCommit(ScheduleRequest request)
    {
        Validate(request);
        if (Options.Mode == WorkspaceAccessMode.Yolo) { var result = Commit(request); Save(); return result; }
        if (_state.Pending.Count >= 100) throw new InvalidOperationException("Review existing pending changes first.");
        var existing = _state.Entries.SingleOrDefault(e => e.Name.Equals(request.Name, StringComparison.OrdinalIgnoreCase));
        string id = Guid.NewGuid().ToString("N");
        _state.Pending.Add(new(id, request, existing?.Revision));
        Save();
        return new("needs_approval", $"Approve schedule {request.Operation}: {request.Name}.", id, request);
    }

    private void Validate(ScheduleRequest request)
    {
        switch (request.Operation)
        {
            case "create":
                if (_state.Entries.Count >= 100 || _state.Entries.Any(e => e.Name.Equals(request.Name, StringComparison.OrdinalIgnoreCase)))
                    throw new ArgumentException("Schedule name exists or the 100-schedule limit was reached.");
                _ = request.CreateEntry(_clock.GetUtcNow(), 1, _files);
                break;
            case "update":
                _ = request.CreateEntry(_clock.GetUtcNow(), Find(request.Name).Revision + 1, _files);
                goto case "cancel";
            case "cancel":
            case "delete":
                _ = Find(request.Name);
                if (_state.Events.Any(e => e.Job?.Name.Equals(request.Name, StringComparison.OrdinalIgnoreCase) == true))
                    throw new InvalidOperationException("This occurrence is queued or running; wait for its completion before changing it.");
                break;
            default: throw new ArgumentException("Unknown mutation.");
        }
    }

    private ScheduleResult Commit(ScheduleRequest request)
    {
        switch (request.Operation)
        {
            case "create": _state.Entries.Add(request.CreateEntry(_clock.GetUtcNow(), 1, _files)); break;
            case "update": Replace(request.CreateEntry(_clock.GetUtcNow(), Find(request.Name).Revision + 1, _files)); break;
            case "cancel": var entry = Find(request.Name); Replace(entry with { Status = "canceled", NextUtc = null, Revision = entry.Revision + 1 }); break;
            case "delete": _state.Entries.Remove(Find(request.Name)); break;
            default: throw new ArgumentException("Unknown mutation.");
        }
        return new("completed", $"Schedule {request.Operation} completed: {request.Name}.");
    }

    private object View(ScheduleEntry entry) => View(entry, true);
    private object View(ScheduleEntry entry, bool detail)
    {
        double? remaining = entry.NextUtc is { } next ? Math.Max(0, (next - _clock.GetUtcNow()).TotalSeconds) : null;
        bool queued = _state.Events.Any(e => e.Job?.Name == entry.Name);
        return detail
            ? new { entry.Name, entry.Kind, entry.Status, entry.NextUtc, remaining_seconds = remaining,
                entry.TimeZone, entry.Cron, entry.Seconds, entry.Message, entry.Scenario, entry.LastOutcome, entry.LastFinishedUtc, queued_or_running = queued }
            : new { entry.Name, entry.Kind, entry.Status, entry.NextUtc, remaining_seconds = remaining, queued_or_running = queued };
    }
    private ScheduleEntry Find(string name) => _state.Entries.SingleOrDefault(e => e.Name.Equals(name, StringComparison.OrdinalIgnoreCase))
        ?? throw new ArgumentException("Unknown schedule: " + name);
    private void Replace(ScheduleEntry entry) { _state.Entries.Remove(Find(entry.Name)); _state.Entries.Add(entry); }
    private void AddEvent(ScheduleEntry entry, string kind, string message, ScheduleEntry? job = null) =>
        _state.Events.Add(new(Guid.NewGuid().ToString("N"), entry.Name, kind, _clock.GetUtcNow(), message, job));
    private void Save()
    {
        string temp = _file + "." + Guid.NewGuid().ToString("N") + ".tmp";
        try
        {
            File.WriteAllText(temp, JsonSerializer.Serialize(_state, new JsonSerializerOptions { WriteIndented = true }));
            File.Move(temp, _file, true);
        }
        catch { _faulted = true; throw; }
    }
    private void ThrowIfDisposed()
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
        if (_faulted) throw new IOException("Schedule storage failed. Fix storage and restart before continuing.");
    }
    public void Dispose() { lock (_sync) { if (_disposed) return; _disposed = true; _lease.Dispose(); } }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

using Cronos;
using Lilith.Core.Scenarios;
using System.Text.Json;

namespace Lilith.Core.Scheduling.Extensions;

internal static class ScheduleExtensions
{
    internal static ScheduleEntry CreateEntry(this ScheduleRequest request, DateTimeOffset now, int revision, ScenarioFileService files)
    {
        if (!ScenarioXmlParser.ValidName(request.Name) || request.Name.Length > 64)
            throw new ArgumentException("Use a schedule name of up to 64 letters, digits, underscores or hyphens.");
        if (request.Message.Length > 2000 || request.InputsJson.Length > 20000 || request.Cron.Length > 128 || request.TimeZone.Length > 128)
            throw new ArgumentException("Schedule input is too large.");
        var inputs = JsonSerializer.Deserialize<Dictionary<string, string>>(request.InputsJson) ?? throw new ArgumentException("Expected scenario input object.");
        if (inputs.Any(p => p.Value is null)) throw new ArgumentException("Scenario inputs must be strings.");
        if (request.Scenario.Length > 0)
        {
            if (request.Scenario.StartsWith("schedule_", StringComparison.Ordinal)) throw new ArgumentException("Scheduled jobs cannot manage schedules.");
            var definition = ScenarioXmlParser.Parse(files.LoadScenarioXml(request.Scenario));
            if (definition.AllowedTools.Any(n => ScheduleTool.Operation(n) is not null)) throw new ArgumentException("Scheduled jobs cannot manage schedules.");
            if (inputs.Keys.Any(k => !definition.Inputs.Any(i => i.Name == k)) ||
                definition.Inputs.Any(i => i.Required && string.IsNullOrWhiteSpace(inputs.TryGetValue(i.Name, out var value) ? value : i.Value)))
                throw new ArgumentException("Unknown or missing scheduled scenario inputs.");
        }
        else if (string.IsNullOrWhiteSpace(request.Message)) throw new ArgumentException("A reminder message or saved scenario is required.");
        string zone = string.IsNullOrWhiteSpace(request.TimeZone) ? TimeZoneInfo.Local.Id : request.TimeZone;
        _ = TimeZoneInfo.FindSystemTimeZoneById(zone);
        var entry = new ScheduleEntry(request.Name, request.Kind, request.Seconds, request.Cron, zone,
            request.Message, request.Scenario, inputs, null, "active", revision);
        return entry with { NextUtc = entry.NextAfter(now) ?? throw new ArgumentException("Schedule has no future occurrence.") };
    }

    internal static DateTimeOffset? NextAfter(this ScheduleEntry entry, DateTimeOffset now) => entry.Kind switch
    {
        "timer" or "interval" when entry.Seconds is >= 1 and <= 31536000 => now.AddSeconds(entry.Seconds),
        "cron" => ParseCron(entry.Cron).GetNextOccurrence(now, TimeZoneInfo.FindSystemTimeZoneById(entry.TimeZone))?.ToUniversalTime(),
        _ => throw new ArgumentException("Choose timer or interval with 1–31536000 seconds, or a five-field cron expression.")
    };

    private static CronExpression ParseCron(string expression) => expression.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries).Length == 5
        ? CronExpression.Parse(expression, CronFormat.Standard)
        : throw new ArgumentException("Cron requires exactly five fields: minute hour day month weekday.");
}