Seal payloads only once every live worker can read them

backend/src/CodeSpace.Api/CodeSpace.Api.csproj

@@ -78,6 +78,8 @@
     <!-- The integration test reaches internal registrar helpers (CodeSpaceHangfireRegistrar.IsProcessingEnabled)
          to pin the deployment-topology gate's polarity without standing up the web host. -->
     <InternalsVisibleTo Include="CodeSpace.IntegrationTests" />
+    <!-- The worker-host E2E drives RecurringJobScheduleKeeper's single pass rather than waiting out its interval. -->
+    <InternalsVisibleTo Include="CodeSpace.E2ETests" />
   </ItemGroup>
 
 </Project>

backend/src/CodeSpace.Api/Controllers/SealedPayloadsController.cs

@@ -0,0 +1,26 @@
+using CodeSpace.Messages.Commands.Workflows;
+using MediatR;
+using Microsoft.AspNetCore.Mvc;
+
+namespace CodeSpace.Api.Controllers;
+
+/// <summary>
+/// Deployment maintenance for sealed payloads (0247). Instance-level and global-admin only, enforced by the command's
+/// mediator marker — it spans every team, so nothing here reads <c>X-Team-Id</c>.
+/// </summary>
+[ApiController]
+[Route("api/admin/sealed-payloads")]
+public class SealedPayloadsController : ControllerBase
+{
+    private readonly IMediator _mediator;
+
+    public SealedPayloadsController(IMediator mediator) { _mediator = mediator; }
+
+    /// <summary>The pre-rollback step for workers: restore every sealed row a previous-version worker could still execute — non-terminal run payloads, queued agent tasks — to the exact, unsealed shape it executes. Send it once no current worker is live (plus 30 seconds) and before starting the previous ones; <see cref="UnsealNonTerminalPayloadsCommand"/> has the order.</summary>
+    [HttpPost("unseal")]
+    public async Task<IActionResult> Unseal(CancellationToken cancellationToken)
+    {
+        var result = await _mediator.Send(new UnsealNonTerminalPayloadsCommand(), cancellationToken).ConfigureAwait(false);
+        return Ok(result);
+    }
+}

backend/src/CodeSpace.Api/Extensions/Hangfire/HangfireRegistrar.Worker.cs

@@ -3,7 +3,7 @@
 using CodeSpace.Core.Jobs;
 using CodeSpace.Core.Services.Agents;
 using CodeSpace.Core.Services.Agents.Sandbox.Runners;
-using CodeSpace.Core.Services.Jobs;
+using CodeSpace.Core.Services.Fleet;
 using Hangfire;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -46,7 +46,7 @@ public override void RegisterHangfire(IServiceCollection services, IConfiguratio
         {
             opt.WorkerCount = ControlWorkerCount;
             opt.ServerName = $"codespace-control-{Environment.MachineName}";
-            opt.Queues = new[] { HangfireConstants.DefaultQueue };
+            opt.Queues = ControlQueues;
         });
 
         // Agent pool — the long IAgentRunExecutor jobs. (WorkerCount preserves the prior agent concurrency; tune
@@ -55,10 +55,23 @@ public override void RegisterHangfire(IServiceCollection services, IConfiguratio
         {
             opt.WorkerCount = Environment.ProcessorCount * 2;
             opt.ServerName = $"codespace-agents-{Environment.MachineName}";
-            opt.Queues = new[] { HangfireConstants.AgentQueue };
+            opt.Queues = AgentQueues;
         });
+
+        // Heals the recurring entries a previous-version worker's scheduler disables during a rolling deploy.
+        services.AddHostedService<RecurringJobScheduleKeeper>();
     }
 
+    /// <summary>
+    /// Each pool's own queue, then the fleet capability markers this build advertises: queue names nothing is ever
+    /// enqueued to, which the server table carries to every peer so it can tell this build's workers from a previous
+    /// build's (<see cref="FleetCapabilityMarkers"/>). Both pools list them — a server without a marker reads to the
+    /// fleet gate as a previous-version worker.
+    /// </summary>
+    internal static readonly string[] ControlQueues = [HangfireConstants.DefaultQueue, .. FleetCapabilityMarkers.Advertised];
+
+    internal static readonly string[] AgentQueues = [HangfireConstants.AgentQueue, .. FleetCapabilityMarkers.Advertised];
+
     /// <summary>Dedicated control-plane workers — a deliberate small FLOOR (control jobs are short + low-volume + self-drain at the 2s poll interval, so they need little). Raise it if control-plane latency is ever observed under a recovery burst (many simultaneous resumes / reconciler re-dispatches / decision-expiry).</summary>
     private const int ControlWorkerCount = 4;
 
@@ -84,37 +97,20 @@ public override void ApplyHangfire(IApplicationBuilder app, IConfiguration confi
 
     /// <summary>
     /// Resolve every <see cref="IRecurringJob"/> from DI and register each with Hangfire's
-    /// scheduler. The registered call is <c>IJobSafeRunner.Run(jobId, jobType)</c> — Hangfire
-    /// serialises that call; on each tick a fresh lifetime scope resolves the concrete job
+    /// scheduler (<see cref="RecurringJobSchedule"/>). The registered call is <c>IJobSafeRunner.Run(jobId, jobType)</c> —
+    /// Hangfire serialises that call; on each tick a fresh lifetime scope resolves the concrete job
     /// type from DI + invokes <see cref="IJob.Execute"/> with <c>JobId</c> in the log context.
     /// </summary>
     private static void ScanHangfireRecurringJobs(IApplicationBuilder app)
     {
-        var backgroundJobClient = app.ApplicationServices.GetRequiredService<ICodeSpaceBackgroundJobClient>();
-
-        var recurringJobTypes = typeof(IRecurringJob).Assembly.GetTypes()
-            .Where(type => type.IsClass && !type.IsAbstract && typeof(IRecurringJob).IsAssignableFrom(type))
-            .ToList();
-
-        foreach (var type in recurringJobTypes)
-        {
-            var job = (IRecurringJob)app.ApplicationServices.GetRequiredService(type);
-
-            // A conditionally-registered job (feature-gated, e.g. a flag-OFF lane's reaper) is skipped ENTIRELY when it
-            // opts out — no recurring entry is created, so flag-OFF is byte-identical (no tick ever fires).
-            if (job is IConditionalRecurringJob { ShouldRegister: false })
-            {
-                Log.Information("Recurring job {Job} skipped — conditional registration is off", job.GetType().FullName);
-                continue;
-            }
-
-            if (string.IsNullOrEmpty(job.CronExpression))
-            {
-                Log.Error("Recurring job cron expression empty, {Job}", job.GetType().FullName);
-                continue;
-            }
+        foreach (var skip in RecurringJobSchedule.Assert(app.ApplicationServices)) LogSkip(skip);
+    }
 
-            backgroundJobClient.AddOrUpdateRecurringJob<IJobSafeRunner>(job.JobId, r => r.Run(job.JobId, type), job.CronExpression, job.TimeZone);
-        }
+    private static void LogSkip(RecurringJobSkip skip)
+    {
+        if (skip.Reason == RecurringJobSkipReason.ConditionallyOff)
+            Log.Information("Recurring job {Job} skipped — conditional registration is off", skip.JobType.FullName);
+        else
+            Log.Error("Recurring job cron expression empty, {Job}", skip.JobType.FullName);
     }
 }

backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobSchedule.cs

@@ -0,0 +1,53 @@
+using CodeSpace.Core.Jobs;
+using CodeSpace.Core.Services.Jobs;
+
+namespace CodeSpace.Api.Extensions.Hangfire;
+
+/// <summary>
+/// The recurring-job set this build schedules: every <see cref="IRecurringJob"/> in <c>CodeSpace.Core</c>, registered
+/// with Hangfire as <c>IJobSafeRunner.Run(jobId, jobType)</c>. Asserted once when a worker boots and again by
+/// <see cref="RecurringJobScheduleKeeper"/>; both go through here so the two can never schedule different sets.
+/// </summary>
+internal static class RecurringJobSchedule
+{
+    /// <summary>Add or update every schedulable job. Returns the jobs it left unscheduled and why; the caller decides whether that is worth a log line.</summary>
+    internal static IReadOnlyList<RecurringJobSkip> Assert(IServiceProvider services)
+    {
+        var backgroundJobClient = services.GetRequiredService<ICodeSpaceBackgroundJobClient>();
+        var skipped = new List<RecurringJobSkip>();
+
+        foreach (var type in JobTypes())
+        {
+            var job = (IRecurringJob)services.GetRequiredService(type);
+
+            if (SkipReasonOf(job) is { } reason)
+            {
+                skipped.Add(new RecurringJobSkip(type, reason));
+                continue;
+            }
+
+            backgroundJobClient.AddOrUpdateRecurringJob<IJobSafeRunner>(job.JobId, r => r.Run(job.JobId, type), job.CronExpression, job.TimeZone);
+        }
+
+        return skipped;
+    }
+
+    private static IEnumerable<Type> JobTypes() =>
+        typeof(IRecurringJob).Assembly.GetTypes().Where(type => type.IsClass && !type.IsAbstract && typeof(IRecurringJob).IsAssignableFrom(type));
+
+    /// <summary>A conditionally-registered job (feature-gated, e.g. a flag-OFF lane's reaper) that opts out gets NO recurring entry, so flag-OFF is byte-identical; a job with no cron cannot be scheduled at all.</summary>
+    private static RecurringJobSkipReason? SkipReasonOf(IRecurringJob job)
+    {
+        if (job is IConditionalRecurringJob { ShouldRegister: false }) return RecurringJobSkipReason.ConditionallyOff;
+
+        return string.IsNullOrEmpty(job.CronExpression) ? RecurringJobSkipReason.NoCronExpression : null;
+    }
+}
+
+internal enum RecurringJobSkipReason
+{
+    ConditionallyOff,
+    NoCronExpression,
+}
+
+internal sealed record RecurringJobSkip(Type JobType, RecurringJobSkipReason Reason);

backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobScheduleKeeper.cs

@@ -0,0 +1,57 @@
+namespace CodeSpace.Api.Extensions.Hangfire;
+
+/// <summary>
+/// Re-asserts this build's recurring-job set every <see cref="Interval"/> on a processing pod, healing an entry a
+/// previous-version scheduler disabled.
+///
+/// <para>Every worker's Hangfire scheduler fires every recurring entry, its own or not. A previous-version worker cannot
+/// load a job type it never had, retries the entry a few times, then disables it — no next execution — and nothing
+/// re-enables it until a current worker boots again. In a rolling deploy the last current worker usually boots before
+/// the last previous-version one stops, so the sweeps this build added could stay off indefinitely.
+/// <c>AddOrUpdate</c> on an entry that carries a scheduling error reschedules it and resets its retries; a healthy
+/// entry comes out unchanged.</para>
+///
+/// <para>Deliberately not an <c>IRecurringJob</c>: a previous-version scheduler would disable it exactly like the
+/// entries it exists to heal. It runs only where the Worker registrar adds it, so never on a public pod, and it only
+/// re-runs the registration that worker boot already performs — no business work.</para>
+///
+/// <para>A previous-version worker can still fetch a background job of a new type off the shared queue and fail it
+/// (no retries); the next tick runs on a current worker. Nothing here can prevent that, and nothing is lost but the one
+/// tick.</para>
+/// </summary>
+public sealed class RecurringJobScheduleKeeper : BackgroundService
+{
+    internal static readonly TimeSpan Interval = TimeSpan.FromMinutes(5);
+
+    private readonly IServiceScopeFactory _scopes;
+    private readonly TimeProvider _clock;
+    private readonly ILogger<RecurringJobScheduleKeeper> _logger;
+
+    public RecurringJobScheduleKeeper(IServiceScopeFactory scopes, TimeProvider clock, ILogger<RecurringJobScheduleKeeper> logger)
+    {
+        _scopes = scopes;
+        _clock = clock;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        using var timer = new PeriodicTimer(Interval, _clock);
+
+        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) Reassert();
+    }
+
+    /// <summary>One pass, in its own scope. A failed pass is logged and left to the next one: a keeper that threw out of <see cref="ExecuteAsync"/> would stop the host.</summary>
+    internal void Reassert()
+    {
+        try
+        {
+            using var scope = _scopes.CreateScope();
+            RecurringJobSchedule.Assert(scope.ServiceProvider);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Re-asserting the recurring-job schedule failed; the next pass in {Interval} retries", Interval);
+        }
+    }
+}

backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/UnsealNonTerminalPayloadsCommandHandler.cs

@@ -0,0 +1,16 @@
+using CodeSpace.Core.Services.Workflows.Runtime;
+using CodeSpace.Messages.Commands.Workflows;
+using MediatR;
+
+namespace CodeSpace.Core.Handlers.CommandHandlers.Workflows;
+
+/// <summary>Rule 16 — thin handler. The restore lives in <see cref="ISealedPayloadUnsealService"/>.</summary>
+public sealed class UnsealNonTerminalPayloadsCommandHandler : IRequestHandler<UnsealNonTerminalPayloadsCommand, UnsealNonTerminalPayloadsResponse>
+{
+    private readonly ISealedPayloadUnsealService _unseal;
+
+    public UnsealNonTerminalPayloadsCommandHandler(ISealedPayloadUnsealService unseal) { _unseal = unseal; }
+
+    public async Task<UnsealNonTerminalPayloadsResponse> Handle(UnsealNonTerminalPayloadsCommand request, CancellationToken cancellationToken) =>
+        await _unseal.UnsealNonTerminalAsync(cancellationToken).ConfigureAwait(false);
+}

backend/src/CodeSpace.Core/Services/Agents/AgentAcceptanceContract.cs

@@ -50,6 +50,25 @@ public static string UnitId(string? nodeId, string iterationKey) =>
     public static string Hash(AgentTask task) =>
         Messages.Contracts.ContractHashing.Hash(new { instruction = task.Goal, acceptance = task.Acceptance, expectsChanges = task.ExpectsChanges }, AgentJson.Options);
 
+    /// <summary>
+    /// P4-U1/U2: the obligations a contract-bearing workflow agent dispatched OUTSIDE the supervisor stakes for its unit
+    /// — acceptance Required plus delivery/output per its change expectation, hashed from THIS task, which must be the
+    /// task as <c>task_jsonb</c> holds it (the completion composer re-derives the hash from that column). Empty for a
+    /// task with no contract or one the supervisor stakes itself (a <c>WorkUnit</c>).
+    ///
+    /// <para>P5-4: the task carries its contract's PROVENANCE (stamped by the projection builder that knows the source
+    /// — the quick tier's spec is the operator's launch floor). Null = unknown/model-authored → the ModelProposal
+    /// default; authority is only ever under-claimed, never inflated. planRef mirrors the composer's workflow-agent
+    /// convention exactly (WorkPlanId = the run, PlanVersion 0) — the same coordinates its attempts and receipts
+    /// already carry. OwesAcceptance is true by construction — only a task with an Acceptance stakes.</para>
+    /// </summary>
+    public static IReadOnlyList<Messages.Contracts.RequirementEnvelope> WorkflowAgentRequirements(AgentTask task, string? nodeId, string iterationKey, Guid workflowRunId)
+    {
+        if (task.Acceptance is null || task.WorkUnit is not null) return [];
+
+        return SupervisorUnitContract.BuildStakedRequirements([(UnitId(nodeId, iterationKey), Hash(task), true, ExpectsChanges(task))], task.AcceptanceAuthority ?? Messages.Contracts.ContractAuthority.ModelProposal, (workflowRunId, 0));
+    }
+
     /// <summary>The <see cref="AgentRunResult.ExitReason"/> a fail-closed acceptance re-grade stamps — the machine-readable marker consumers (the agent.run node's retry verdict) key on to tell a DETERMINISTIC verdict failure from a transient death. Pinned by a unit test (Rule 8) so producer + consumer can't drift.</summary>
     public const string FailClosedExitReason = "acceptance-failed";
 

backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs

@@ -4360,8 +4360,8 @@
     /// <summary>The credential parts embedded IN a base URL — userinfo and secret-named query values (<see cref="CredentialNeedles.FromUrl"/>).</summary>
     private static IEnumerable<string> UrlEmbeddedSecrets(string? baseUrl) => CredentialNeedles.FromUrl(baseUrl);
 
-    /// <summary>Re-persist the run's stored task with its RESOLVED model filled, so the live projection shows what an "auto" run actually dispatches from the moment it starts (mirrors the harness-reconciliation write). The task is the ORIGINAL (no injected secret env) with only <see cref="AgentTask.Model"/> set — but it is the OPENED task, so it is re-sealed exactly as the store first persisted it: written back raw, the workflow secrets it embeds would land in task_jsonb again.</summary>
-    private Task PersistResolvedModelAsync(AgentRunOwnerToken owner, AgentTask taskWithModel, CancellationToken cancellationToken) => PersistRuntimeIdentityAsync(owner, null, AgentTaskSecrets.Seal(taskWithModel, _sealer), cancellationToken);
+    /// <summary>Re-persist the run's stored task with its RESOLVED model filled, so the live projection shows what an "auto" run actually dispatches from the moment it starts (mirrors the harness-reconciliation write). The task is the ORIGINAL (no injected secret env) with only <see cref="AgentTask.Model"/> set — but it is the OPENED task, so it is re-sealed exactly as the store first persisted it: written back raw, the workflow secrets it embeds would land in task_jsonb again. The row is Running, so it keeps that shape whatever the fleet gate says (<see cref="AgentTaskSecrets.Reseal"/>): unsealed, a re-attach would rebuild this launch's redactor without those secrets and fail the handle's fingerprint.</summary>
+    private Task PersistResolvedModelAsync(AgentRunOwnerToken owner, AgentTask taskWithModel, CancellationToken cancellationToken) => PersistRuntimeIdentityAsync(owner, null, AgentTaskSecrets.Reseal(taskWithModel, _sealer), cancellationToken);
 
     /// <summary>The masked task and its sealed copy move together; a harness-only write (null task) leaves both untouched.</summary>
     private async Task PersistRuntimeIdentityAsync(AgentRunOwnerToken owner, string? harness, Messages.Workflows.SealedPayload? task, CancellationToken cancellationToken)
@@ -5347,10 +5347,10 @@
     /// <summary>The same reconstruction from a payload that came from somewhere other than the row — an offloaded one fetched back out of the artifact store.</summary>
     private static AgentEvent ReplayedEvent(AgentEventKind kind, string? text, string? dataJson)
     {
         if (dataJson is not { Length: > 0 } json) return new AgentEvent { Kind = kind, Text = text };
 
         try { using var doc = JsonDocument.Parse(json); return new AgentEvent { Kind = kind, Text = text, Data = doc.RootElement.Clone() }; }
         catch (JsonException) { return new AgentEvent { Kind = kind, Text = text }; }
     }
 
     /// <summary>Ask the row, on a token of its own, whether the run actually reached a terminal state — the only honest answer to "did the landing take?" once an exception has been raised somewhere after the fenced write.</summary>

backend/src/CodeSpace.Core/Services/Agents/AgentRunService.cs

@@ -308,16 +308,9 @@ private async Task StakeWorkflowAgentObligationsAsync(AgentTask task, Guid teamI
 
         try
         {
-            // P5-4: the task carries its contract's PROVENANCE (stamped by the projection builder that knows the
-            // source — the quick tier's spec is the operator's launch floor). Null = unknown/model-authored → the
-            // ModelProposal default; authority is only ever under-claimed, never inflated.
-            // planRef mirrors the composer's workflow-agent convention exactly (WorkPlanId = the run, PlanVersion 0)
-            // — the same coordinates its attempts and receipts already carry.
-            // OwesAcceptance is true by construction — this method only stakes when task.Acceptance is present.
-            var requirements = Supervisor.SupervisorUnitContract.BuildStakedRequirements(new[]
-            {
-                (AgentAcceptanceContract.UnitId(nodeId, iterationKey), AgentAcceptanceContract.Hash(task), true, AgentAcceptanceContract.ExpectsChanges(task)),
-            }, task.AcceptanceAuthority ?? Messages.Contracts.ContractAuthority.ModelProposal, (runId, 0));
+            // The one builder the pre-rollback unseal re-stakes through too (SealedPayloadUnsealService), so the two
+            // can never stake different shapes for the same unit.
+            var requirements = AgentAcceptanceContract.WorkflowAgentRequirements(task, nodeId, iterationKey, runId);
 
             await _contracts.UpsertRequirementsAsync(runId, teamId, requirements, cancellationToken).ConfigureAwait(false);
         }