admin/VolcanicAgenticCore
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scheduler): store jobs per chat and import VolcanicSecretary job files

Browse Source 
Each chat's jobs live in <workspace>/.volcanic/scheduler/jobs.json. On start the
scheduler loads every chat's file, moves the retired shared
<agentDir>/scheduler/jobs.json into them and renames it .migrated-<ts>.
A VolcanicSecretary jobs.json (bare array) in a chat's folder is converted
in place for that chat; the original is kept as .secretary-<ts>.

Also adds the VolcanicSecretary data migration plan.
This commit is contained in:
EugeneTes
2026-10-02 07:29:48 +00:00
parent 28c6f90d91
commit 28bbfbc759
19 changed files with 609 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobDelivery.cs
View File

@@ -12,6 +12,9 @@ public interface IJobDelivery
/// <summary>Working directory for the target's scripts. A path computation: must not create or wake a session.</summary>
string WorkspaceFor(string target);
/// <summary>Targets whose workspace may hold a job file (<see cref="JobStore.PathIn"/>), read once on start. Must not create or wake a session.</summary>
IEnumerable<string> KnownTargets();
/// <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);

127
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobScheduler.cs
View File

@@ -1,10 +1,12 @@
using System.Globalization;
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
/// 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>,
@@ -22,7 +24,7 @@ public sealed class JobScheduler : IAsyncDisposable
private readonly ISecretRedactor _redactor;
private readonly TimeProvider _time;
private readonly ILogger _logger;
private readonly JobStore _store;
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);
@@ -38,18 +40,17 @@ public sealed class JobScheduler : IAsyncDisposable
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>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; } }
@@ -60,11 +61,11 @@ public sealed class JobScheduler : IAsyncDisposable
/// <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 the store, fire overdue one-offs (late), compute cron occurrences strictly after now (no catch-up), arm the timer.</summary>
/// <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 _store.LoadAsync(cancellationToken).ConfigureAwait(false);
var jobs = await LoadAllAsync(delivery, cancellationToken).ConfigureAwait(false);
var now = _time.GetUtcNow();
lock (_lock)
{
@@ -97,6 +98,90 @@ public sealed class JobScheduler : IAsyncDisposable
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)
{
@@ -158,7 +243,7 @@ public sealed class JobScheduler : IAsyncDisposable
next = schedule.At ?? Schedules.NextCron(schedule.Cron!, now, zone);
_jobs[created.Id] = new Entry(created) { Next = next };
}
await PersistAsync(cancellationToken).ConfigureAwait(false);
await PersistAsync(target, cancellationToken).ConfigureAwait(false);
Arm();
return new JobView(created, next);
}
@@ -180,7 +265,7 @@ public sealed class JobScheduler : IAsyncDisposable
_inFlight.TryGetValue(id, out running);
}
if (running is not null) await TryCancelAsync(running.Cancel).ConfigureAwait(false);
await PersistAsync(cancellationToken).ConfigureAwait(false);
await PersistAsync(target, cancellationToken).ConfigureAwait(false);
Arm();
return removed;
}
@@ -356,7 +441,7 @@ public sealed class JobScheduler : IAsyncDisposable
result = await RunWatcherAsync(job, delivery, zone, scriptToken).ConfigureAwait(false);
break;
}
if (result is not null) await RecordAsync(job.Id, result).ConfigureAwait(false);
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>
@@ -423,17 +508,17 @@ public sealed class JobScheduler : IAsyncDisposable
}
/// <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)
private async Task RecordAsync(ScheduledJob job, 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);
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(CancellationToken.None).ConfigureAwait(false);
await PersistAsync(job.Target, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
@@ -442,15 +527,15 @@ public sealed class JobScheduler : IAsyncDisposable
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)
/// <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).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)];
await _store.SaveAsync(snapshot, cancellationToken).ConfigureAwait(false);
lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).Where(j => j.Target == target).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)];
await StoreFor(_delivery!, target).SaveAsync(snapshot, cancellationToken).ConfigureAwait(false);
}
finally
{
@@ -475,11 +560,13 @@ public sealed class JobScheduler : IAsyncDisposable
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()
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 (_jobs.ContainsKey(id) || _inFlight.ContainsKey(id));
while (taken(id));
return id;
}

142
extensions/Volcanic.AgenticCore.Extensions.Scheduler/JobStore.cs
View File

@@ -11,8 +11,14 @@ namespace Volcanic.AgenticCore.Extensions.Scheduler;
/// (temp file + rename) through one semaphore. A corrupt file is set aside and loading starts empty, so a bad file never
/// keeps the host down; invalid jobs (including entries that don't deserialise) are dropped with a warning. A job of the
/// retired <c>reminder</c> kind is loaded as a prompt asking the agent to pass the reminder on.
/// <para>
/// A chat's own file (<see cref="PathIn"/>, with <paramref name="target"/>) owns its jobs: every entry gets that target,
/// whatever it stores. A VolcanicSecretary <c>jobs.json</c> there (a bare array without targets) is converted, saved in
/// this format and the original kept as <c>jobs.json.secretary-&lt;timestamp&gt;</c>. Without a target (the retired
/// shared file) entries keep their own targets and a VolcanicSecretary file is set aside.
/// </para>
/// </summary>
public sealed class JobStore(string path, TimeProvider time, ILogger logger)
public sealed class JobStore(string path, TimeProvider time, ILogger logger, string? target = null)
{
public const int Version = 1;
private const string ReminderPrefix = "Remind me: ";
@@ -20,14 +26,30 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger)
public string Path => path;
/// <summary>A chat's job file: <c>&lt;workspace&gt;/.volcanic/scheduler/jobs.json</c>.</summary>
public static string PathIn(string workspace) => System.IO.Path.Combine(workspace, ".volcanic", "scheduler", "jobs.json");
public async Task<IReadOnlyList<ScheduledJob>> LoadAsync(CancellationToken cancellationToken)
{
if (!File.Exists(path)) return [];
var json = await File.ReadAllTextAsync(path, cancellationToken).ConfigureAwait(false);
JsonElement root;
try
{
using var document = JsonDocument.Parse(json);
root = document.RootElement.Clone();
}
catch (JsonException)
{
root = default;
}
if (root.ValueKind == JsonValueKind.Array)
return await LoadSecretaryAsync(root, cancellationToken).ConfigureAwait(false);
StoredFile? file;
try
{
file = VolcanicJson.Deserialize<StoredFile>(json);
file = root.ValueKind == JsonValueKind.Object ? VolcanicJson.Deserialize<StoredFile>(root) : null;
}
catch (JsonException)
{
@@ -35,17 +57,36 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger)
}
if (file is not { Version: Version, Jobs: not null })
{
var aside = $"{path}.corrupt-{time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}";
File.Move(path, aside, overwrite: true);
logger.LogError("The job store {Path} could not be read; it was moved to {Aside} and the scheduler starts empty", path, aside);
SetAside("could not be read");
return [];
}
return ReadAll([.. file.Jobs.Select(Owned)]);
}
private async Task<IReadOnlyList<ScheduledJob>> LoadSecretaryAsync(JsonElement entries, CancellationToken cancellationToken)
{
if (string.IsNullOrWhiteSpace(target))
{
SetAside("is a VolcanicSecretary job file, which belongs in a chat's workspace (.volcanic/scheduler/jobs.json)");
return [];
}
var source = entries.EnumerateArray().ToList();
var jobs = ReadAll([.. source.Select(FromSecretary)]);
var backup = $"{path}.secretary-{Stamp()}";
File.Copy(path, backup, overwrite: true);
await SaveAsync(jobs, cancellationToken).ConfigureAwait(false);
logger.LogInformation("Converted the VolcanicSecretary job file {Path} for target {Target}: kept {Kept} of {Total} jobs; the original is {Backup}",
path, target, jobs.Count, source.Count, backup);
return jobs;
}
private List<ScheduledJob> ReadAll(IReadOnlyList<JsonElement> entries)
{
var jobs = new List<ScheduledJob>();
var ids = new HashSet<string>(StringComparer.Ordinal);
for (var i = 0; i < file.Jobs.Count; i++)
for (var i = 0; i < entries.Count; i++)
{
var (job, problem) = Read(file.Jobs[i]);
var (job, problem) = Read(entries[i]);
problem ??= ids.Add(job!.Id) ? null : "duplicate id";
if (problem is not null)
{
@@ -57,6 +98,15 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger)
return jobs;
}
private void SetAside(string why)
{
var aside = $"{path}.corrupt-{Stamp()}";
File.Move(path, aside, overwrite: true);
logger.LogError("The job store {Path} {Why}; it was moved to {Aside} and the scheduler starts empty", path, why, aside);
}
private string Stamp() => time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture);
public async Task SaveAsync(IReadOnlyList<ScheduledJob> jobs, CancellationToken cancellationToken)
{
await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
@@ -96,6 +146,15 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger)
return (job, JobValidation.CheckStored(job));
}
/// <summary>In a chat's file the entry belongs to that chat, whatever target it stores.</summary>
private JsonElement Owned(JsonElement entry)
{
if (string.IsNullOrWhiteSpace(target) || entry.ValueKind != JsonValueKind.Object) return entry;
var node = JsonNode.Parse(entry.GetRawText())!.AsObject();
node["target"] = target;
return JsonSerializer.SerializeToElement(node);
}
/// <summary>A stored <c>reminder</c> (fixed text sent as is) becomes a prompt; other entries pass through unchanged.</summary>
private static JsonElement FromReminder(JsonElement entry)
{
@@ -104,12 +163,75 @@ public sealed class JobStore(string path, TimeProvider time, ILogger logger)
return entry;
var node = JsonNode.Parse(entry.GetRawText())!.AsObject();
node["kind"] = "prompt";
if (entry.TryGetProperty("text", out var text) && text.ValueKind == JsonValueKind.String && text.GetString() is { } body && !string.IsNullOrWhiteSpace(body)
&& body.Length + ReminderPrefix.Length <= JobValidation.MaxTextLength)
node["text"] = ReminderPrefix + body;
if (entry.TryGetProperty("text", out var text) && Remind(text) is { } reminder) node["text"] = reminder;
return JsonSerializer.SerializeToElement(node);
}
/// <summary>The text of a reminder with <see cref="ReminderPrefix"/>, or null when it's empty or the prefix doesn't fit.</summary>
private static string? Remind(JsonElement text) =>
text.ValueKind == JsonValueKind.String && text.GetString() is { } body && !string.IsNullOrWhiteSpace(body)
&& body.Length + ReminderPrefix.Length <= JobValidation.MaxTextLength
? ReminderPrefix + body
: null;
/// <summary>
/// One VolcanicSecretary job as an entry of this format. <c>source.kind</c> <c>text</c> becomes a prompt (a
/// <c>deliver: direct</c> one, which sent its text as is, a reminder), <c>script</c> a watcher; a 6-field cron with
/// seconds <c>0</c> loses that field. Anything else is left for <see cref="Read"/> to reject.
/// </summary>
private JsonElement FromSecretary(JsonElement entry)
{
if (entry.ValueKind != JsonValueKind.Object) return entry;
var node = new JsonObject { ["target"] = target };
foreach (var name in (string[])["id", "name", "createdAt", "lastRunAt"])
if (entry.TryGetProperty(name, out var value)) node[name] = JsonNode.Parse(value.GetRawText());
if (entry.TryGetProperty("schedule", out var schedule))
{
var copy = JsonNode.Parse(schedule.GetRawText());
if (copy is JsonObject scheduleNode && scheduleNode["cron"] is JsonValue cron && cron.TryGetValue<string>(out var expression))
scheduleNode["cron"] = WithoutZeroSeconds(expression);
node["schedule"] = copy;
}
if (entry.TryGetProperty("source", out var source) && source.ValueKind == JsonValueKind.Object
&& source.TryGetProperty("kind", out var kind) && kind.ValueKind == JsonValueKind.String)
{
switch (kind.GetString())
{
case "text":
node["kind"] = "prompt";
if (source.TryGetProperty("text", out var text))
{
var direct = !entry.TryGetProperty("deliver", out var deliver) || deliver.ValueKind != JsonValueKind.String || deliver.GetString() != "agent";
node["text"] = (direct ? Remind(text) : null) ?? JsonNode.Parse(text.GetRawText());
}
break;
case "script":
node["kind"] = "watcher";
if (source.TryGetProperty("command", out var command)) node["command"] = JsonNode.Parse(command.GetRawText());
if (source.TryGetProperty("timeoutSec", out var timeout)) node["timeoutSeconds"] = JsonNode.Parse(timeout.GetRawText());
break;
default:
node["kind"] = kind.GetString();
break;
}
}
if (entry.TryGetProperty("lastResult", out var result) && JsonNode.Parse(result.GetRawText()) is JsonObject resultNode)
{
resultNode["timedOut"] ??= false;
node["lastResult"] = resultNode;
}
return JsonSerializer.SerializeToElement(node);
}
private static string WithoutZeroSeconds(string cron)
{
var fields = cron.Split((char[]?)null, StringSplitOptions.RemoveEmptyEntries);
return fields.Length == 6 && fields[0] == "0" ? string.Join(' ', fields[1..]) : cron;
}
private sealed record JobFile(int Version, List<ScheduledJob> Jobs);
private sealed record StoredFile(int Version, List<JsonElement>? Jobs);

10
extensions/Volcanic.AgenticCore.Extensions.Scheduler/SchedulerOptions.cs
View File

@@ -7,8 +7,12 @@ public sealed class SchedulerOptions
public bool Enabled { get; set; } = true;
/// <summary><c>jobs.json</c>. Empty: the host fills <see cref="DefaultStorePath"/> under its agent directory.</summary>
public string? StorePath { get; set; }
/// <summary>
/// The retired shared <c>jobs.json</c>. Jobs are kept per chat (<see cref="JobStore.PathIn"/>); on start the jobs
/// found here move into their chats' files and this file is renamed <c>jobs.json.migrated-&lt;timestamp&gt;</c>.
/// Empty: the host fills <see cref="DefaultLegacyStorePath"/> under its agent directory.
/// </summary>
public string? LegacyStorePath { get; set; }
/// <summary>Watcher timeout when the job sets none.</summary>
public TimeSpan DefaultScriptTimeout { get; set; } = TimeSpan.FromSeconds(300);
@@ -19,7 +23,7 @@ public sealed class SchedulerOptions
public List<string> Shell { get; set; } = [];
/// <summary><c>&lt;agentDir&gt;/scheduler/jobs.json</c>.</summary>
public static string DefaultStorePath(string agentDir) => Path.Combine(agentDir, "scheduler", "jobs.json");
public static string DefaultLegacyStorePath(string agentDir) => Path.Combine(agentDir, "scheduler", "jobs.json");
/// <summary>Configuration problems that must stop the host, empty when valid.</summary>
public IReadOnlyList<string> Validate()