feat(telegram): scheduler delivery, hosted service, /jobs and /cancel
Spec: /jobs and /cancel explain when scheduling is off; Scheduler options validation.
This commit is contained in:
@@ -2,12 +2,14 @@ using System.Globalization;
|
||||
using System.Text;
|
||||
using Telegram.Bot.Types;
|
||||
using Volcanic.AgenticCore.Abstractions.Providers;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
using Volcanic.AgenticCore.Telegram.Rendering;
|
||||
using Volcanic.AgenticCore.Telegram.Scheduling;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Chats;
|
||||
|
||||
/// <summary>The bot's own slash commands. Anything else starting with <c>/</c> goes to the session (extension commands).</summary>
|
||||
public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zones = null)
|
||||
public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zones = null, JobScheduler? scheduler = null)
|
||||
{
|
||||
public static readonly IReadOnlyList<BotCommand> Menu =
|
||||
[
|
||||
@@ -19,6 +21,7 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
|
||||
new() { Command = "compact", Description = "Summarise the conversation to free context" },
|
||||
new() { Command = "thinking", Description = "Show or set the thinking level" },
|
||||
new() { Command = "timezone", Description = "Show or set this chat's time zone" },
|
||||
.. SchedulerSetup.Menu,
|
||||
new() { Command = "help", Description = "What this bot does" },
|
||||
];
|
||||
|
||||
@@ -102,6 +105,14 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
|
||||
await ReplyAsync(chat, await TimeZoneAsync(chat, command.Args, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
|
||||
return true;
|
||||
|
||||
case "jobs":
|
||||
await ReplyAsync(chat, Jobs(chat), cancellationToken).ConfigureAwait(false);
|
||||
return true;
|
||||
|
||||
case "cancel":
|
||||
await ReplyAsync(chat, await CancelJobAsync(chat, command.Args, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false);
|
||||
return true;
|
||||
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
@@ -125,7 +136,7 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
|
||||
/compact – summarise the conversation to free context
|
||||
/thinking [level] – show or set the thinking level
|
||||
/timezone [IANA name|reset] – this chat's time zone
|
||||
""" + GoogleCalendarSetup.HelpLines(chat.Session.Commands);
|
||||
""" + SchedulerSetup.HelpLines + GoogleCalendarSetup.HelpLines(chat.Session.Commands);
|
||||
|
||||
private string Status(ChatContext chat)
|
||||
{
|
||||
@@ -137,6 +148,8 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
|
||||
sb.AppendLine(CultureInfo.InvariantCulture, $"**Messages:** {session.Messages.Count}");
|
||||
if (session.GetContextUsage() is { } usage)
|
||||
sb.AppendLine(CultureInfo.InvariantCulture, $"**Context:** {(usage.Tokens is { } t ? t.ToString("N0", CultureInfo.InvariantCulture) : "?")} / {usage.ContextWindow.ToString("N0", CultureInfo.InvariantCulture)} tokens{(usage.Percent is { } p ? $" ({p:0}%)" : "")}");
|
||||
if (scheduler is { IsStarted: true })
|
||||
sb.AppendLine(CultureInfo.InvariantCulture, $"**Jobs:** {scheduler.List(TelegramJobDelivery.TargetOf(chat.ChatId)).Count}");
|
||||
var zone = zones?.Describe(chat.ChatId) ?? ChatTimeZoneService.Describe(session.TimeZone, DateTimeOffset.UtcNow);
|
||||
sb.AppendLine(CultureInfo.InvariantCulture, $"**Time zone:** {zone}");
|
||||
sb.AppendLine(CultureInfo.InvariantCulture, $"**Workspace:** `{session.Cwd}`");
|
||||
@@ -181,6 +194,30 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
|
||||
: error;
|
||||
}
|
||||
|
||||
private string Jobs(ChatContext chat)
|
||||
{
|
||||
if (scheduler is not { IsStarted: true }) return SchedulerSetup.DisabledReply;
|
||||
var target = TelegramJobDelivery.TargetOf(chat.ChatId);
|
||||
var jobs = scheduler.List(target);
|
||||
if (jobs.Count == 0) return "No scheduled jobs.";
|
||||
var zone = scheduler.TimeZoneFor(target);
|
||||
var now = scheduler.Now;
|
||||
return $"**Scheduled jobs** ({JobFormat.ZoneName(zone)}, now {JobFormat.DayTime(now, zone)})\n" +
|
||||
string.Join('\n', jobs.Select(j => JobFormat.Line(j, zone, now)));
|
||||
}
|
||||
|
||||
private async Task<string> CancelJobAsync(ChatContext chat, string arg, CancellationToken cancellationToken)
|
||||
{
|
||||
if (scheduler is not { IsStarted: true }) return SchedulerSetup.DisabledReply;
|
||||
var id = arg.Trim().ToLowerInvariant();
|
||||
if (id.Length == 0) return "Usage: /cancel <id> (see /jobs).";
|
||||
var target = TelegramJobDelivery.TargetOf(chat.ChatId);
|
||||
var job = scheduler.List(target).FirstOrDefault(v => v.Job.Id == id)?.Job;
|
||||
return job is not null && await scheduler.CancelAsync(target, id, cancellationToken).ConfigureAwait(false)
|
||||
? $"Cancelled {job.Id} ({job.Name})."
|
||||
: $"No job {id} in this chat.";
|
||||
}
|
||||
|
||||
private static async Task<string> ThinkingAsync(ChatContext chat, string arg, CancellationToken cancellationToken)
|
||||
{
|
||||
var session = chat.Session;
|
||||
|
||||
@@ -5,6 +5,7 @@ using Volcanic.AgenticCore.Abstractions.Extensions.Events;
|
||||
using Volcanic.AgenticCore.Abstractions.Messages;
|
||||
using Volcanic.AgenticCore.Abstractions.Ui;
|
||||
using Volcanic.AgenticCore.Extensions.GoogleCalendar;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
using Volcanic.AgenticCore.Features.Agent;
|
||||
using Volcanic.AgenticCore.Features.Extensions;
|
||||
using Volcanic.AgenticCore.Hosting;
|
||||
@@ -72,7 +73,8 @@ public sealed class ChatRegistry(
|
||||
TimeProvider? time = null,
|
||||
GoogleCalendarOptions? calendar = null,
|
||||
ChatSettingsStore? settings = null,
|
||||
ChatTimeZoneService? zones = null) : IAsyncDisposable
|
||||
ChatTimeZoneService? zones = null,
|
||||
JobScheduler? scheduler = null) : IAsyncDisposable
|
||||
{
|
||||
private readonly ConcurrentDictionary<long, Lazy<Task<ChatContext>>> _chats = new();
|
||||
private readonly ILogger _logger = loggerFactory.CreateLogger<ChatRegistry>();
|
||||
@@ -114,6 +116,8 @@ public sealed class ChatRegistry(
|
||||
var extras = new List<ExtensionRegistration>();
|
||||
if (calendar is not null) extras.Add(GoogleCalendarSetup.Registration(calendar));
|
||||
if (zones is not null) extras.Add(ExtensionRegistration.FromInstance(new ChatSettingsExtension(chatId, zones)));
|
||||
if (scheduler is not null)
|
||||
extras.Add(ExtensionRegistration.FromInstance(new SchedulerExtension(scheduler, chatId.ToString(System.Globalization.CultureInfo.InvariantCulture))));
|
||||
AgentSession session;
|
||||
try
|
||||
{
|
||||
|
||||
@@ -11,6 +11,7 @@ using Microsoft.Extensions.Options;
|
||||
using Telegram.Bot;
|
||||
using Volcanic.AgenticCore.Extensions.ClaudeCode;
|
||||
using Volcanic.AgenticCore.Extensions.Mcp;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
using Volcanic.AgenticCore.Features.Processes;
|
||||
using Volcanic.AgenticCore.Hosting;
|
||||
using Volcanic.AgenticCore.Telegram;
|
||||
@@ -18,6 +19,7 @@ using Volcanic.AgenticCore.Telegram.Chats;
|
||||
using Volcanic.AgenticCore.Telegram.Inbound;
|
||||
using Volcanic.AgenticCore.Telegram.Polling;
|
||||
using Volcanic.AgenticCore.Telegram.Rendering;
|
||||
using Volcanic.AgenticCore.Telegram.Scheduling;
|
||||
using Volcanic.AgenticCore.Telegram.Speech;
|
||||
|
||||
var builder = Host.CreateApplicationBuilder(args);
|
||||
@@ -32,7 +34,8 @@ var processEnvironment = new ProcessEnvironmentOptions();
|
||||
TelegramProcessEnvironment.Configure(processEnvironment, options, processSection, Environment.GetEnvironmentVariable, calendar);
|
||||
var mcp = builder.Configuration.GetSection(McpOptions.SectionName).Get<McpOptions>() ?? new McpOptions();
|
||||
mcp.ApplyEnvironment(Environment.GetEnvironmentVariable);
|
||||
var errors = options.Validate().Concat(processEnvironment.Validate()).Concat(mcp.Validate()).ToList();
|
||||
var scheduling = SchedulerSetup.Options(builder.Configuration.GetSection(SchedulerOptions.SectionName), VolcanicPaths.DefaultAgentDir);
|
||||
var errors = options.Validate().Concat(processEnvironment.Validate()).Concat(mcp.Validate()).Concat(scheduling.Validate()).ToList();
|
||||
if (errors.Count > 0)
|
||||
{
|
||||
foreach (var error in errors) Console.Error.WriteLine(error);
|
||||
@@ -41,6 +44,7 @@ if (errors.Count > 0)
|
||||
|
||||
builder.Services.AddSingleton(options);
|
||||
builder.Services.AddSingleton(mcp);
|
||||
builder.Services.AddSingleton(scheduling);
|
||||
builder.Services.AddSingleton(options.Speech);
|
||||
if (options.Speech.IsEnabled) // no key, no ISpeechToText: voice notes get the setup notice
|
||||
// InboundMedia (a singleton) keeps this typed client for the process lifetime, so the factory never rotates its
|
||||
@@ -80,13 +84,14 @@ builder.Services.AddSingleton(sp => new ChatTimeZoneService(
|
||||
sp.GetRequiredService<ChatSettingsStore>(),
|
||||
chatId => sp.GetRequiredService<ChatRegistry>().Find(chatId)?.Session,
|
||||
sp.GetRequiredService<TimeProvider>(),
|
||||
sp.GetService<Volcanic.AgenticCore.Extensions.Scheduler.JobScheduler>()));
|
||||
sp.GetService<JobScheduler>()));
|
||||
builder.Services.AddSingleton<ChatCommands>();
|
||||
builder.Services.AddSingleton<InboundMedia>();
|
||||
builder.Services.AddSingleton<InboundComposer>();
|
||||
builder.Services.AddSingleton<IUpdateRouter, UpdateRouter>();
|
||||
builder.Services.AddSingleton<PollingService>();
|
||||
builder.Services.AddHostedService(sp => sp.GetRequiredService<PollingService>());
|
||||
if (scheduling.Enabled) SchedulerSetup.Register(builder.Services); // the hosted service starts after PollingService
|
||||
builder.Services.AddHostedService<BotSetupService>();
|
||||
|
||||
using var host = builder.Build();
|
||||
@@ -101,6 +106,7 @@ if (ClaudeInstallation.Locate(Environment.GetEnvironmentVariable) is { } claude)
|
||||
logger.LogInformation("Using Claude Code at {Path}, model {Model}", claude, options.Model);
|
||||
else
|
||||
logger.LogWarning("Claude Code (claude) was not found on PATH; requests will fail until it is installed and logged in. Send /claude-code-doctor to the bot for details.");
|
||||
logger.LogInformation("Scheduler: {State}", scheduling.Enabled ? scheduling.StorePath : "off");
|
||||
logger.LogInformation("Default time zone: {Zone}", TimeZoneIds.Name(options.ConfiguredTimeZone() ?? TimeZoneInfo.Local));
|
||||
logger.LogInformation("Allowed users: {Users}. Workspaces under {Root}", string.Join(", ", options.AllowedUserIds), options.WorkspaceRoot);
|
||||
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Scheduling;
|
||||
|
||||
/// <summary>Starts the scheduler with the Telegram delivery (registered after <c>PollingService</c>) and stops it on shutdown.</summary>
|
||||
public sealed class SchedulerHostedService(JobScheduler scheduler, TelegramJobDelivery delivery, ILogger<SchedulerHostedService> logger) : IHostedService
|
||||
{
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await scheduler.StartAsync(delivery, cancellationToken).ConfigureAwait(false);
|
||||
logger.LogInformation("Scheduler started; jobs in {Path}", scheduler.StorePath);
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken) => scheduler.StopAsync(cancellationToken);
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using Telegram.Bot.Types;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Scheduling;
|
||||
|
||||
/// <summary>Everything the Telegram host does for the scheduler, kept out of the shared host files.</summary>
|
||||
public static class SchedulerSetup
|
||||
{
|
||||
public const string DisabledReply = "Scheduling is turned off on this bot (Scheduler:Enabled).";
|
||||
|
||||
public const string HelpLines = "\n/jobs – this chat's scheduled reminders, prompts and watchers\n/cancel <id> – cancel one of them";
|
||||
|
||||
public static readonly IReadOnlyList<BotCommand> Menu =
|
||||
[
|
||||
new() { Command = "jobs", Description = "Scheduled reminders, prompts and watchers" },
|
||||
new() { Command = "cancel", Description = "Cancel a scheduled job: /cancel <id>" },
|
||||
];
|
||||
|
||||
/// <summary>The <c>Scheduler</c> section; an empty <c>StorePath</c> becomes <c><agentDir>/scheduler/jobs.json</c>.</summary>
|
||||
public static SchedulerOptions Options(IConfiguration section, string agentDir)
|
||||
{
|
||||
var options = section.Get<SchedulerOptions>() ?? new SchedulerOptions();
|
||||
if (string.IsNullOrWhiteSpace(options.StorePath)) options.StorePath = SchedulerOptions.DefaultStorePath(agentDir);
|
||||
return options;
|
||||
}
|
||||
|
||||
/// <summary>Runner, scheduler, delivery and the hosted service (register after <c>PollingService</c>). Only when enabled.</summary>
|
||||
public static void Register(IServiceCollection services)
|
||||
{
|
||||
services.TryAddSingleton(TimeProvider.System);
|
||||
services.AddSingleton<IScriptRunner, ScriptRunner>();
|
||||
services.AddSingleton<JobScheduler>();
|
||||
services.AddSingleton<TelegramJobDelivery>();
|
||||
services.AddHostedService<SchedulerHostedService>();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
using System.Globalization;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
|
||||
using Volcanic.AgenticCore.Extensions.Scheduler;
|
||||
using Volcanic.AgenticCore.Telegram.Chats;
|
||||
using Volcanic.AgenticCore.Telegram.Inbound;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Scheduling;
|
||||
|
||||
/// <summary>
|
||||
/// The scheduler's view of Telegram. Targets are chat ids (invariant culture). Direct texts go through the chat's
|
||||
/// <c>IChatChannel</c> (renderer queue, splitting, HTML with fallback, retries); prompts go straight to the chat's session
|
||||
/// as follow-ups, never through <c>UpdateRouter</c> or the <c>ChatInbox</c>, so they are never debounced into a user's
|
||||
/// message and never answer a pending question. <see cref="WorkspaceFor"/> and <see cref="TimeZoneFor"/> never create or
|
||||
/// look up a session: a silent watcher tick leaves the chat untouched.
|
||||
/// </summary>
|
||||
public sealed class TelegramJobDelivery(ChatRegistry chats, ChatSettingsStore settings, TelegramOptions options, TimeProvider time, ILogger<TelegramJobDelivery> logger) : IJobDelivery
|
||||
{
|
||||
public static long ChatIdOf(string target) => long.Parse(target, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture);
|
||||
|
||||
public static string TargetOf(long chatId) => chatId.ToString(CultureInfo.InvariantCulture);
|
||||
|
||||
public async Task SendTextAsync(string target, string markdown, CancellationToken cancellationToken)
|
||||
{
|
||||
var chat = await chats.GetAsync(ChatIdOf(target), cancellationToken).ConfigureAwait(false);
|
||||
await chat.Channel.SendTextAsync(markdown, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public async Task RunPromptAsync(string target, string prompt, CancellationToken cancellationToken)
|
||||
{
|
||||
var chatId = ChatIdOf(target);
|
||||
var chat = await chats.GetAsync(chatId, cancellationToken).ConfigureAwait(false);
|
||||
var text = MessageContext.TimestampHeader(time.GetUtcNow(), TimeZoneFor(target)) + "\n" + prompt;
|
||||
try
|
||||
{
|
||||
// Idle: starts a run and returns when it settles. Busy: queued as a follow-up (never steering) and returns at once.
|
||||
await chat.Session.PromptAsync(text, null, StreamingBehavior.FollowUp, cancellationToken).ConfigureAwait(false);
|
||||
// Queued: the job stays in flight until the run that consumed it settles. Harmless when it already ran.
|
||||
await chat.Session.WaitForIdleAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Scheduled prompt in chat {ChatId} failed", chatId);
|
||||
chat.Report($"The scheduled job failed: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public string WorkspaceFor(string target) => chats.WorkspaceFor(ChatIdOf(target));
|
||||
|
||||
/// <summary>Private chat id = user id: the chat is served while its user is allow-listed.</summary>
|
||||
public bool IsDeliverable(string target) =>
|
||||
long.TryParse(target, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var chatId) && options.IsAllowed(chatId);
|
||||
|
||||
public TimeZoneInfo TimeZoneFor(string target) => settings.TimeZoneFor(ChatIdOf(target));
|
||||
}
|
||||
Reference in New Issue
Block a user