V7 / 2a442462cf3f
ScheduleSession.cs
Lilith/Core/Scheduling/ScheduleSession.cs
Unchanged from the preceding versionCS
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(); } }
}