Background jobs

Scheduled checks, data-quality evaluations, documentation generation, AI actor cycles and maintenance all run as background jobs. Beacon defines what must run; your host decides how, through one interface: IBeaconScheduler.

sequenceDiagram
  participant Beacon as Beacon (handlers)
  participant Sched as Your IBeaconScheduler
  participant Runner as Your job runner
  participant Jobs as IJobService (Beacon)
  Beacon->>Sched: AddOrUpdate(subscriptionId, name, cron)
  Sched->>Runner: register recurring job "subscription-{id}"
  Note over Runner: cron fires
  Runner->>Runner: new DI scope, mark as system work
  Runner->>Jobs: ExecuteQuery(subscriptionId)
  Jobs->>Jobs: run as the subscription's owner, re-check access, gate SQL, notify

The contract

namespace Beacon.Core.Worker;

public interface IBeaconScheduler
{
    Task AddOrUpdate(int subscriptionId, string subscriptionName, string cron);
    Task Remove(int subscriptionId, string subscriptionName);

    // The members below have default bodies that throw NotImplementedException.
    // Implement them to enable the feature that calls them.
    Task AddOrUpdateDataQualityJob(int contractId, string contractName, string cron);   // data contracts
    Task RemoveDataQualityJob(int contractId, string contractName);
    Task<string> EnqueueProjectDocumentation(int projectId, int userId, string? notifyUserId); // AI documentation
    Task<string> EnqueueAiActorThinkCycle(int actorId, int subscriptionId);              // AI actors
    Task<string> EnqueueMcpEval(int runId);                                              // MCP evaluation runs
}

Register it with options.AddBeaconScheduler<YourScheduler>(). Without one, AddBeaconServices throws Implementation of IBeaconScheduler is required.

• cron is a standard 5-field expression (validated by Beacon before it reaches you).
• Key recurring jobs by the entity id (subscription-{id}, dq-{id}), never by the name. Names change, ids do not.
• The Enqueue… methods return your runner’s job id as a string. The UI correlates live JobStatusChanged events by that id; notifyUserId is the user who should receive them.

What each job calls

Each job opens a new DI scope, marks it as system work, and calls one Beacon method:

Job	Calls	Scheduled by
Subscription run	IJobService.ExecuteQuery(subscriptionId, ct)	AddOrUpdate / Remove
Data contract evaluation	IJobService.EvaluateDataContract(contractId, ct)	AddOrUpdateDataQualityJob / RemoveDataQualityJob
Project documentation	IProjectDocumentationService.GenerateDocumentationAsync(projectId, userId, ct) (Beacon.AI)	EnqueueProjectDocumentation
AI actor think cycle	IAiActorServiceExtended.ExecuteThinkCycleBackgroundAsync(actorId, subscriptionId) (Beacon.AI)	EnqueueAiActorThinkCycle
MCP evaluation run	IJobService.RunMcpEval(runId, ct)	EnqueueMcpEval

Recurring maintenance jobs

Register these yourself at startup (the reference host’s schedule):

Id	Cron	Calls	Purpose
mcp-learning-aggregate	0 */6 * * *	IJobService.AggregateLearnedPatterns	Mine MCP usage signals into learned patterns.
mcp-learning-cleanup	0 3 * * *	IJobService.CleanupOldSignals	Delete signals past their retention.
mcp-audit-retention	30 3 * * *	IMcpAuditRetentionService.PurgeExpiredAsync	Purge MCP audit rows older than Beacon:Mcp:Audit:RetentionDays (no-op when unset).
mcp-embedding-reindex	0 */12 * * *	IJobService.ReindexEmbeddings	Re-embed metadata and examples for semantic retrieval.
mcp-docchunk-reindex	0 */12 * * *	IJobService.ReindexDocChunks	Re-chunk and re-embed project documentation.

One-off operator jobs:

• While Beacon:PreviousEncryptionKeys lists a key, run IStoredValueReEncryptionService.ReEncryptAsync once per start (the reference host checks Beacon.Core.ServiceConfiguration.HasPreviousEncryptionKeys(configuration)). Remove the old key when it reports nothing left.
• IRecipientSecretEncryptionService.EncryptStoredSecretsAsync encrypts recipient secrets stored before encryption existed. Run it once, by hand, after every instance runs the new version.

The system-work rule

Beacon decides project access from the caller. A background job has no HTTP request, so it must say that it is Beacon’s own work:

scope.ServiceProvider.GetRequiredService<ISystemWorkScope>().MarkAsSystemWork();   // Beacon.Core.Authorization

Do this in the job’s scope before Beacon code runs. An unmarked scope without a request gets no project access and logs a warning, and the job does nothing useful. Work done for a user (a subscription’s scheduled run) is still executed as that user: Beacon sets the run-as owner itself and re-checks their access to the exact data sources right before any SQL or LLM call. A run the owner may no longer do is recorded as skipped and never run as the system. Do not resolve the current user in your job wrapper before Beacon’s method runs.

Reference implementation: Moberg Warp

Moberg Warp is a PostgreSQL-backed job runner with recurring jobs and a dashboard. The reference host (src/Beacon.SampleProject) uses Warp 3.11; the code below matches it. Check the Warp docs if you use a newer major version.

dotnet add package Moberg.Warp.Core --version 3.11.0
dotnet add package Moberg.Warp.Worker --version 3.11.0
dotnet add package Moberg.Warp.Provider.PostgreSql --version 3.11.0
dotnet add package Moberg.Warp.UI --version 3.11.0          # optional dashboard
dotnet add package EFCore.NamingConventions                 # UseSnakeCaseNamingConvention
dotnet add package Microsoft.EntityFrameworkCore.Design     # for dotnet ef (same EF Core 10 version)
dotnet tool install --global dotnet-ef

1. A dedicated context for Warp’s tables (the warp schema), kept apart from Beacon’s:

using Microsoft.EntityFrameworkCore;

public sealed class WarpDbContext(DbContextOptions<WarpDbContext> options) : DbContext(options);

Create its initial migration once in your project: dotnet ef migrations add WarpInitial --context WarpDbContext. Warp supplies the model.

2. Registration and startup:

using Microsoft.EntityFrameworkCore;
using Warp.Core;
using Warp.Provider.PostgreSql;
using Warp.Worker;

public static class BeaconWarpSetup
{
    public static IServiceCollection AddBeaconWarp(this IServiceCollection services, IConfiguration configuration)
    {
        // Warp supports PostgreSQL; with Beacon on SQL Server, point WarpContext at a PostgreSQL database.
        services.AddDbContext<WarpDbContext>(x => x
            .UseNpgsql(configuration.GetConnectionString("WarpContext") ?? configuration.GetConnectionString("BeaconContext"))
            .UseSnakeCaseNamingConvention());

        services.AddWarpServer<WarpDbContext>(x =>
        {
            x.UsePostgreSql();
            x.PollingInterval = TimeSpan.FromSeconds(1);
            x.WorkerCount = Math.Min(Environment.ProcessorCount * 2, 20);   // protects the connection pool
            // No x.AddRetry(): failed Beacon jobs are investigated, not retried blindly.
        });

        return services;
    }

    public static async Task StartBeaconWarpAsync(this WebApplication app)
    {
        using var scope = app.Services.CreateScope();
        scope.ServiceProvider.GetRequiredService<WarpDbContext>().Database.Migrate();

        var recurring = scope.ServiceProvider.GetRequiredService<IRecurringJobPublisher>();
        await recurring.AddOrUpdateRecurringJob(new AggregateLearnedPatternsJob(), "mcp-learning-aggregate", "0 */6 * * *");
        await recurring.AddOrUpdateRecurringJob(new CleanupOldSignalsJob(), "mcp-learning-cleanup", "0 3 * * *");
        await recurring.AddOrUpdateRecurringJob(new PurgeExpiredMcpAuditLogsJob(), "mcp-audit-retention", "30 3 * * *");
        await recurring.AddOrUpdateRecurringJob(new ReindexEmbeddingsJob(), "mcp-embedding-reindex", "0 */12 * * *");
        await recurring.AddOrUpdateRecurringJob(new ReindexDocChunksJob(), "mcp-docchunk-reindex", "0 */12 * * *");
    }
}

3. Mark every job as system work with a pipeline behaviour (Warp’s source generator registers it):

using Beacon.Core.Authorization;
using Warp.Core.Handlers;

public sealed class SystemWorkBehavior<TRequest, TResponse>(ISystemWorkScope systemWork)
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    public Task<TResponse> HandleAsync(TRequest request, RequestHandlerDelegate<TRequest, TResponse> next, CancellationToken cancellationToken)
    {
        systemWork.MarkAsSystemWork();
        return next(request, cancellationToken);
    }
}

4. One job + handler per Beacon entry point:

using Beacon.Core.Services.Retention;
using Beacon.Core.Worker;
using Warp.Core.Handlers;

public sealed class ExecuteSubscriptionQueryJob : IJob { public int SubscriptionId { get; init; } }
public sealed class ExecuteSubscriptionQueryJobHandler(IJobService jobs) : IJobHandler<ExecuteSubscriptionQueryJob>
{
    public Task HandleAsync(ExecuteSubscriptionQueryJob message, CancellationToken cancellationToken)
        => jobs.ExecuteQuery(message.SubscriptionId, cancellationToken);
}

public sealed class EvaluateDataContractJob : IJob { public int ContractId { get; init; } }
public sealed class EvaluateDataContractJobHandler(IJobService jobs) : IJobHandler<EvaluateDataContractJob>
{
    public Task HandleAsync(EvaluateDataContractJob message, CancellationToken cancellationToken)
        => jobs.EvaluateDataContract(message.ContractId, cancellationToken);
}

public sealed class RunMcpEvalJob : IJob { public int RunId { get; init; } }
public sealed class RunMcpEvalJobHandler(IJobService jobs) : IJobHandler<RunMcpEvalJob>
{
    public Task HandleAsync(RunMcpEvalJob message, CancellationToken cancellationToken)
        => jobs.RunMcpEval(message.RunId, cancellationToken);
}

public sealed class AggregateLearnedPatternsJob : IJob;
public sealed class AggregateLearnedPatternsJobHandler(IJobService jobs) : IJobHandler<AggregateLearnedPatternsJob>
{
    public Task HandleAsync(AggregateLearnedPatternsJob message, CancellationToken cancellationToken)
        => jobs.AggregateLearnedPatterns(cancellationToken);
}

public sealed class CleanupOldSignalsJob : IJob;
public sealed class CleanupOldSignalsJobHandler(IJobService jobs) : IJobHandler<CleanupOldSignalsJob>
{
    public Task HandleAsync(CleanupOldSignalsJob message, CancellationToken cancellationToken)
        => jobs.CleanupOldSignals(cancellationToken);
}

public sealed class ReindexEmbeddingsJob : IJob;
public sealed class ReindexEmbeddingsJobHandler(IJobService jobs) : IJobHandler<ReindexEmbeddingsJob>
{
    public Task HandleAsync(ReindexEmbeddingsJob message, CancellationToken cancellationToken)
        => jobs.ReindexEmbeddings(cancellationToken);
}

public sealed class ReindexDocChunksJob : IJob;
public sealed class ReindexDocChunksJobHandler(IJobService jobs) : IJobHandler<ReindexDocChunksJob>
{
    public Task HandleAsync(ReindexDocChunksJob message, CancellationToken cancellationToken)
        => jobs.ReindexDocChunks(cancellationToken);
}

public sealed class PurgeExpiredMcpAuditLogsJob : IJob;
public sealed class PurgeExpiredMcpAuditLogsJobHandler(IMcpAuditRetentionService retention) : IJobHandler<PurgeExpiredMcpAuditLogsJob>
{
    public Task HandleAsync(PurgeExpiredMcpAuditLogsJob message, CancellationToken cancellationToken)
        => retention.PurgeExpiredAsync(cancellationToken);
}

With AI enabled (UseAI = true), add the two AI jobs too:

using Beacon.AI.Services.Ai.AiActor;          // IAiActorServiceExtended
using Beacon.AI.Services.Documentation;       // IProjectDocumentationService
using Warp.Core.Handlers;

public sealed class GenerateProjectDocumentationJob : IJob { public int ProjectId { get; init; } public int UserId { get; init; } }
public sealed class GenerateProjectDocumentationJobHandler(IProjectDocumentationService documentation) : IJobHandler<GenerateProjectDocumentationJob>
{
    public Task HandleAsync(GenerateProjectDocumentationJob message, CancellationToken cancellationToken)
        => documentation.GenerateDocumentationAsync(message.ProjectId, message.UserId, cancellationToken);
}

public sealed class AiActorThinkCycleJob : IJob { public int ActorId { get; init; } public int SubscriptionId { get; init; } }
public sealed class AiActorThinkCycleJobHandler(IAiActorServiceExtended actors) : IJobHandler<AiActorThinkCycleJob>
{
    public Task HandleAsync(AiActorThinkCycleJob message, CancellationToken cancellationToken)
        => actors.ExecuteThinkCycleBackgroundAsync(message.ActorId, message.SubscriptionId);
}

The reference host’s versions are in src/Beacon.SampleProject/Warp/Jobs/.

5. The scheduler:

using Beacon.Core.Worker;
using Microsoft.EntityFrameworkCore;
using Warp.Core;
using Warp.Core.Data.Entities;

public sealed class BeaconScheduler(
    IPublisher publisher,
    IRecurringJobPublisher recurring,
    WarpDbContext warp) : IBeaconScheduler
{
    public Task AddOrUpdate(int subscriptionId, string subscriptionName, string cron)
        => recurring.AddOrUpdateRecurringJob(new ExecuteSubscriptionQueryJob { SubscriptionId = subscriptionId }, $"subscription-{subscriptionId}", cron);

    public Task Remove(int subscriptionId, string subscriptionName)
        => RemoveRecurring($"subscription-{subscriptionId}");

    public Task AddOrUpdateDataQualityJob(int contractId, string contractName, string cron)
        => recurring.AddOrUpdateRecurringJob(new EvaluateDataContractJob { ContractId = contractId }, $"dq-{contractId}", cron);

    public Task RemoveDataQualityJob(int contractId, string contractName)
        => RemoveRecurring($"dq-{contractId}");

    public async Task<string> EnqueueMcpEval(int runId)
    {
        var id = await publisher.Enqueue(new RunMcpEvalJob { RunId = runId });
        await publisher.SaveChangesAsync();
        return id.ToString();
    }

    public async Task<string> EnqueueProjectDocumentation(int projectId, int userId, string? notifyUserId)
    {
        var id = await publisher.Enqueue(new GenerateProjectDocumentationJob { ProjectId = projectId, UserId = userId });
        await publisher.SaveChangesAsync();
        return id.ToString();
    }

    public async Task<string> EnqueueAiActorThinkCycle(int actorId, int subscriptionId)
    {
        var id = await publisher.Enqueue(new AiActorThinkCycleJob { ActorId = actorId, SubscriptionId = subscriptionId });
        await publisher.SaveChangesAsync();
        return id.ToString();
    }

    private async Task RemoveRecurring(string name)
    {
        var job = await warp.Set<RecurringJob>()
            .Where(x => x.Name == name)
            .FirstOrDefaultAsync();
        if (job == null)
        {
            return;
        }

        warp.Set<RecurringJob>().Remove(job);
        await warp.SaveChangesAsync();
    }
}

6. The dashboard (optional). Keep it Admin-only:

app.UseWarpUI(x =>
{
    x.RoutePrefix = "/warp";
    x.Authorization = new AdminOnlyWarpFilter();   // IWarpAuthorizationFilter: authenticated + IsInRole("Admin")
});

Live job status in the UI

The reference host also pushes JobStatusChanged (Processing / Succeeded / Failed) to the user who started a documentation job, through the SignalR hub. It tags the job with BeaconUserId metadata and a second pipeline behaviour sends the event; see src/Beacon.SampleProject/Warp/JobStatusChangedBehavior.cs. Without it everything works; the UI just does not show live progress for those jobs.

Other job runners

Any runner works if it can (1) run a job in its own DI scope, (2) register, update and remove recurring jobs by name, and (3) enqueue a one-off job and return an id. A Hangfire sketch: wrap Beacon’s calls in a class that marks system work first:

using Beacon.Core.Authorization;
using Beacon.Core.Worker;
using Hangfire;

public sealed class BeaconHangfireJobs(ISystemWorkScope systemWork, IJobService jobs)
{
    public Task ExecuteSubscription(int subscriptionId)
    {
        systemWork.MarkAsSystemWork();
        return jobs.ExecuteQuery(subscriptionId, CancellationToken.None);
    }

    public Task EvaluateContract(int contractId)
    {
        systemWork.MarkAsSystemWork();
        return jobs.EvaluateDataContract(contractId, CancellationToken.None);
    }
}

public sealed class HangfireBeaconScheduler(IRecurringJobManager recurring) : IBeaconScheduler
{
    public Task AddOrUpdate(int subscriptionId, string subscriptionName, string cron)
    {
        recurring.AddOrUpdate<BeaconHangfireJobs>($"subscription-{subscriptionId}", x => x.ExecuteSubscription(subscriptionId), cron);
        return Task.CompletedTask;
    }

    public Task Remove(int subscriptionId, string subscriptionName)
    {
        recurring.RemoveIfExists($"subscription-{subscriptionId}");
        return Task.CompletedTask;
    }
}

Hangfire activates BeaconHangfireJobs in a scope per job when you use its ASP.NET Core integration. Turn its automatic retries off for these jobs ([AutomaticRetry(Attempts = 0)]), or a failed check reruns and notifies again.

Operating notes

• No automatic retries. A failed run is recorded and visible in the UI; investigate it rather than retrying blindly. If one of your own jobs needs retries, configure them on that job only.
• Worker count vs connection pool. Each subscription run opens a Beacon context plus a connection to the monitored database. Cap workers so you do not exhaust the pool (the reference host uses min(2 × CPU, 20)).
• Several instances. Your runner must make sure a recurring job fires once across the cluster. Warp and Hangfire do this through their database.