diff --git a/backend/src/CodeSpace.Api/CodeSpace.Api.csproj b/backend/src/CodeSpace.Api/CodeSpace.Api.csproj index 49da0c929..9eabab1bf 100644 --- a/backend/src/CodeSpace.Api/CodeSpace.Api.csproj +++ b/backend/src/CodeSpace.Api/CodeSpace.Api.csproj @@ -78,6 +78,8 @@ + + diff --git a/backend/src/CodeSpace.Api/Controllers/SealedPayloadsController.cs b/backend/src/CodeSpace.Api/Controllers/SealedPayloadsController.cs new file mode 100644 index 000000000..e9aeed8ae --- /dev/null +++ b/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; + +/// +/// 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 X-Team-Id. +/// +[ApiController] +[Route("api/admin/sealed-payloads")] +public class SealedPayloadsController : ControllerBase +{ + private readonly IMediator _mediator; + + public SealedPayloadsController(IMediator mediator) { _mediator = mediator; } + + /// 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; has the order. + [HttpPost("unseal")] + public async Task Unseal(CancellationToken cancellationToken) + { + var result = await _mediator.Send(new UnsealNonTerminalPayloadsCommand(), cancellationToken).ConfigureAwait(false); + return Ok(result); + } +} diff --git a/backend/src/CodeSpace.Api/Extensions/Hangfire/HangfireRegistrar.Worker.cs b/backend/src/CodeSpace.Api/Extensions/Hangfire/HangfireRegistrar.Worker.cs index cf2d1a47c..08dda7898 100644 --- a/backend/src/CodeSpace.Api/Extensions/Hangfire/HangfireRegistrar.Worker.cs +++ b/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(); } + /// + /// 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 (). Both pools list them — a server without a marker reads to the + /// fleet gate as a previous-version worker. + /// + internal static readonly string[] ControlQueues = [HangfireConstants.DefaultQueue, .. FleetCapabilityMarkers.Advertised]; + + internal static readonly string[] AgentQueues = [HangfireConstants.AgentQueue, .. FleetCapabilityMarkers.Advertised]; + /// 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). private const int ControlWorkerCount = 4; @@ -84,37 +97,20 @@ public override void ApplyHangfire(IApplicationBuilder app, IConfiguration confi /// /// Resolve every from DI and register each with Hangfire's - /// scheduler. The registered call is IJobSafeRunner.Run(jobId, jobType) — Hangfire - /// serialises that call; on each tick a fresh lifetime scope resolves the concrete job + /// scheduler (). The registered call is IJobSafeRunner.Run(jobId, jobType) — + /// Hangfire serialises that call; on each tick a fresh lifetime scope resolves the concrete job /// type from DI + invokes with JobId in the log context. /// private static void ScanHangfireRecurringJobs(IApplicationBuilder app) { - var backgroundJobClient = app.ApplicationServices.GetRequiredService(); - - 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(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); } } diff --git a/backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobSchedule.cs b/backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobSchedule.cs new file mode 100644 index 000000000..ca9420629 --- /dev/null +++ b/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; + +/// +/// The recurring-job set this build schedules: every in CodeSpace.Core, registered +/// with Hangfire as IJobSafeRunner.Run(jobId, jobType). Asserted once when a worker boots and again by +/// ; both go through here so the two can never schedule different sets. +/// +internal static class RecurringJobSchedule +{ + /// Add or update every schedulable job. Returns the jobs it left unscheduled and why; the caller decides whether that is worth a log line. + internal static IReadOnlyList Assert(IServiceProvider services) + { + var backgroundJobClient = services.GetRequiredService(); + var skipped = new List(); + + 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(job.JobId, r => r.Run(job.JobId, type), job.CronExpression, job.TimeZone); + } + + return skipped; + } + + private static IEnumerable JobTypes() => + typeof(IRecurringJob).Assembly.GetTypes().Where(type => type.IsClass && !type.IsAbstract && typeof(IRecurringJob).IsAssignableFrom(type)); + + /// 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. + 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); diff --git a/backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobScheduleKeeper.cs b/backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobScheduleKeeper.cs new file mode 100644 index 000000000..b988ba528 --- /dev/null +++ b/backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobScheduleKeeper.cs @@ -0,0 +1,57 @@ +namespace CodeSpace.Api.Extensions.Hangfire; + +/// +/// Re-asserts this build's recurring-job set every on a processing pod, healing an entry a +/// previous-version scheduler disabled. +/// +/// 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. +/// AddOrUpdate on an entry that carries a scheduling error reschedules it and resets its retries; a healthy +/// entry comes out unchanged. +/// +/// Deliberately not an IRecurringJob: 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. +/// +/// 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. +/// +public sealed class RecurringJobScheduleKeeper : BackgroundService +{ + internal static readonly TimeSpan Interval = TimeSpan.FromMinutes(5); + + private readonly IServiceScopeFactory _scopes; + private readonly TimeProvider _clock; + private readonly ILogger _logger; + + public RecurringJobScheduleKeeper(IServiceScopeFactory scopes, TimeProvider clock, ILogger 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(); + } + + /// One pass, in its own scope. A failed pass is logged and left to the next one: a keeper that threw out of would stop the host. + 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); + } + } +} diff --git a/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/UnsealNonTerminalPayloadsCommandHandler.cs b/backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/UnsealNonTerminalPayloadsCommandHandler.cs new file mode 100644 index 000000000..29cce420e --- /dev/null +++ b/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; + +/// Rule 16 — thin handler. The restore lives in . +public sealed class UnsealNonTerminalPayloadsCommandHandler : IRequestHandler +{ + private readonly ISealedPayloadUnsealService _unseal; + + public UnsealNonTerminalPayloadsCommandHandler(ISealedPayloadUnsealService unseal) { _unseal = unseal; } + + public async Task Handle(UnsealNonTerminalPayloadsCommand request, CancellationToken cancellationToken) => + await _unseal.UnsealNonTerminalAsync(cancellationToken).ConfigureAwait(false); +} diff --git a/backend/src/CodeSpace.Core/Services/Agents/AgentAcceptanceContract.cs b/backend/src/CodeSpace.Core/Services/Agents/AgentAcceptanceContract.cs index a7ac6759d..a015e9afe 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/AgentAcceptanceContract.cs +++ b/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); + /// + /// 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 task_jsonb 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 WorkUnit). + /// + /// 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. + /// + public static IReadOnlyList 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)); + } + /// The 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. public const string FailClosedExitReason = "acceptance-failed"; diff --git a/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs b/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs index 92a7004ef..2b43f1040 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/AgentRunExecutor.cs @@ -4360,8 +4360,8 @@ internal static SecretRedactor WithModelBrokerRunToken(SecretRedactor redactor, /// The credential parts embedded IN a base URL — userinfo and secret-named query values (). private static IEnumerable UrlEmbeddedSecrets(string? baseUrl) => CredentialNeedles.FromUrl(baseUrl); - /// 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 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. - private Task PersistResolvedModelAsync(AgentRunOwnerToken owner, AgentTask taskWithModel, CancellationToken cancellationToken) => PersistRuntimeIdentityAsync(owner, null, AgentTaskSecrets.Seal(taskWithModel, _sealer), cancellationToken); + /// 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 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 (): unsealed, a re-attach would rebuild this launch's redactor without those secrets and fail the handle's fingerprint. + private Task PersistResolvedModelAsync(AgentRunOwnerToken owner, AgentTask taskWithModel, CancellationToken cancellationToken) => PersistRuntimeIdentityAsync(owner, null, AgentTaskSecrets.Reseal(taskWithModel, _sealer), cancellationToken); /// The masked task and its sealed copy move together; a harness-only write (null task) leaves both untouched. private async Task PersistRuntimeIdentityAsync(AgentRunOwnerToken owner, string? harness, Messages.Workflows.SealedPayload? task, CancellationToken cancellationToken) diff --git a/backend/src/CodeSpace.Core/Services/Agents/AgentRunService.cs b/backend/src/CodeSpace.Core/Services/Agents/AgentRunService.cs index 9e2323f9d..f7dc9024b 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/AgentRunService.cs +++ b/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); } diff --git a/backend/src/CodeSpace.Core/Services/Agents/AgentTaskSecrets.cs b/backend/src/CodeSpace.Core/Services/Agents/AgentTaskSecrets.cs index fb5443995..a7bd13016 100644 --- a/backend/src/CodeSpace.Core/Services/Agents/AgentTaskSecrets.cs +++ b/backend/src/CodeSpace.Core/Services/Agents/AgentTaskSecrets.cs @@ -33,19 +33,11 @@ public static AgentTask Embed(AgentTask task, PersistenceSecretRedactor? scope) return secrets.Length == 0 ? task : task with { WorkflowSecrets = secrets }; } - /// The task as the store persists it: the masked task_jsonb and, when the task names secrets it embeds, the sealed exact copy. Both are the column-safe text (), so the copy a run executes is the one its public row describes. - public static SealedPayload Seal(AgentTask task, WorkflowSensitivePayloadSealer? sealer) - { - var exact = PersistedText.SanitizeJson(JsonSerializer.Serialize(task, AgentJson.Options))!; - - if (task.WorkflowSecrets is not { Count: > 0 } secrets) return new SealedPayload(exact, null); - - var seal = sealer ?? throw new InvalidOperationException("An agent task embeds workflow secrets but no payload sealer is available to persist it masked."); + /// The task as the store persists it: the masked task_jsonb and, when the task names secrets it embeds, the sealed exact copy. Both are the column-safe text (), so the copy a run executes is the one its public row describes. While a previous-version worker is live the sealer writes the exact task unsealed instead () — the row that worker launches from. + public static SealedPayload Seal(AgentTask task, WorkflowSensitivePayloadSealer? sealer) => Persist(task, sealer, throughFleetGate: true); - using var document = JsonDocument.Parse(exact); - - return seal.Seal(document.RootElement, new PersistenceSecretRedactor(secrets)); - } + /// A task ed from its row, written back onto that row by the current worker running it: in the shape it was opened from — sealed again when it names secrets (only an opened sealed copy does), whatever the fleet gate says, else exact. A previous-version worker never launches a Running row, and the launch's redactor fingerprint names exactly these secrets, which a re-attach rebuilds from the row. + public static SealedPayload Reseal(AgentTask task, WorkflowSensitivePayloadSealer? sealer) => Persist(task, sealer, throughFleetGate: false); /// The exact task a run executes — opened from its sealed copy, naming the secrets it embeds, when the row was sealed; else its task_jsonb, which is then itself exact. Null for an empty envelope. public static AgentTask? Open(AgentRun run, WorkflowSensitivePayloadSealer? sealer) @@ -57,4 +49,18 @@ public static SealedPayload Seal(AgentTask task, WorkflowSensitivePayloadSealer? return JsonSerializer.Deserialize(opened.Json, AgentJson.Options) is { } task ? task with { WorkflowSecrets = opened.Secrets } : null; } + + private static SealedPayload Persist(AgentTask task, WorkflowSensitivePayloadSealer? sealer, bool throughFleetGate) + { + var exact = PersistedText.SanitizeJson(JsonSerializer.Serialize(task, AgentJson.Options))!; + + if (task.WorkflowSecrets is not { Count: > 0 } secrets) return new SealedPayload(exact, null); + + var seal = sealer ?? throw new InvalidOperationException("An agent task embeds workflow secrets but no payload sealer is available to persist it masked."); + + using var document = JsonDocument.Parse(exact); + var redactor = new PersistenceSecretRedactor(secrets); + + return throughFleetGate ? seal.Seal(document.RootElement, redactor) : seal.Reseal(document.RootElement, redactor); + } } diff --git a/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilities.cs b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilities.cs new file mode 100644 index 000000000..c7ce07861 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilities.cs @@ -0,0 +1,106 @@ +using System.Collections.Concurrent; +using CodeSpace.Core.DependencyInjection; +using CodeSpace.Messages.Fleet; +using Microsoft.Extensions.Logging; + +namespace CodeSpace.Core.Services.Fleet; + +/// +/// over a snapshot of the fleet registry that is re-read at most once per +/// : the gate sits on write paths, the registry is a database. The answer can lag a +/// fleet change by one lifetime and never leads it — a previous-version worker that just left still holds the new +/// shape back for up to that long, which only costs writing the shape it read. +/// +/// A fleet that cannot be read answers false: the gate cannot prove no previous-version worker is in it, and the +/// shape every worker reads is the one a write can always fall back to. A host whose registry observes no fleet at all +/// (no job storage — a test host) answers true without reading. +/// +public sealed class FleetCapabilities : IFleetCapabilities, ISingletonDependency +{ + public static readonly TimeSpan SnapshotLifetime = TimeSpan.FromSeconds(30); + + private readonly IWorkerFleetRegistry _registry; + private readonly TimeProvider _clock; + private readonly ILogger _logger; + private readonly Lock _refresh = new(); + private readonly ConcurrentDictionary _lastAnswers = new(StringComparer.Ordinal); + private FleetSnapshot? _snapshot; + + public FleetCapabilities(IWorkerFleetRegistry registry, TimeProvider clock, ILogger logger) + { + _registry = registry; + _clock = clock; + _logger = logger; + } + + public bool AllWorkersSupport(string capability) + { + if (!_registry.ObservesFleet) return true; + + var snapshot = CurrentSnapshot(); + var supported = snapshot.Servers is { } servers && FleetCapabilityRules.EveryLiveServerAdvertises(servers, capability, snapshot.ReadAtUtc); + + ReportChange(capability, supported, snapshot); + + return supported; + } + + private FleetSnapshot CurrentSnapshot() + { + if (Fresh(Volatile.Read(ref _snapshot)) is { } fresh) return fresh; + + lock (_refresh) + { + if (Fresh(_snapshot) is { } raced) return raced; + + var read = ReadSnapshot(); + Volatile.Write(ref _snapshot, read); + + return read; + } + } + + private FleetSnapshot? Fresh(FleetSnapshot? snapshot) => snapshot is not null && _clock.GetUtcNow().UtcDateTime - snapshot.ReadAtUtc < SnapshotLifetime ? snapshot : null; + + private FleetSnapshot ReadSnapshot() + { + var readAt = _clock.GetUtcNow().UtcDateTime; + + try + { + return new FleetSnapshot(readAt, _registry.ReadServers()); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "The worker fleet could not be read; new data shapes are held back until it can be"); + return new FleetSnapshot(readAt, null); + } + } + + /// Logged on a change of answer only — a rolling deploy flips each capability twice, not once per write. + private void ReportChange(string capability, bool supported, FleetSnapshot snapshot) + { + var previous = _lastAnswers.TryGetValue(capability, out var last) ? last : (bool?)null; + + if (previous == supported) return; + + _lastAnswers[capability] = supported; + + if (!supported) + _logger.LogWarning("Not every live worker advertises {Capability}, or none is live; writing the shape a previous-version worker reads until they all do. Holding it back: {Servers}", capability, HoldingBack(capability, snapshot)); + else if (previous == false) + _logger.LogInformation("Every live worker now advertises {Capability}", capability); + } + + private static IReadOnlyList HoldingBack(string capability, FleetSnapshot snapshot) + { + if (snapshot.Servers is not { } servers) return [""]; + + var lacking = FleetCapabilityRules.LiveServersLacking(servers, capability, snapshot.ReadAtUtc); + + return lacking.Count > 0 ? lacking : [""]; + } + + /// The registry as read at ; null servers when the read failed. + private sealed record FleetSnapshot(DateTime ReadAtUtc, IReadOnlyList? Servers); +} diff --git a/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityMarkers.cs b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityMarkers.cs new file mode 100644 index 000000000..ea61c34cd --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityMarkers.cs @@ -0,0 +1,21 @@ +namespace CodeSpace.Core.Services.Fleet; + +/// +/// What a worker of THIS build can read that a previous build cannot, advertised by every job server it starts as an +/// extra queue name that nothing is ever enqueued to. A previous-version worker does not list a marker it never knew, +/// so its absence from any live server is what holds a new shape back (). +/// +/// A marker is compared across builds, so it is never edited: a new shape gets a new marker, and a build that +/// still reads an old shape keeps advertising the old one. +/// +public static class FleetCapabilityMarkers +{ + /// + /// Reads a sealed payload (0247): an agent_run task or a workflow_run_request payload whose public + /// column is masked and whose exact copy is in the encrypted sealed column. A worker without it executes the mask. + /// + public const string SealedPayloadV1 = "cap-sealed-payload-v1"; + + /// Every marker this build's job servers advertise. + public static readonly IReadOnlyList Advertised = [SealedPayloadV1]; +} diff --git a/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityRules.cs b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityRules.cs new file mode 100644 index 000000000..c09c2bd5e --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityRules.cs @@ -0,0 +1,28 @@ +using CodeSpace.Messages.Fleet; + +namespace CodeSpace.Core.Services.Fleet; + +/// +/// The fleet gate's decision, pure: a capability holds when there is a LIVE job server and every live one advertises +/// its marker. A server whose heartbeat is older than is gone — Hangfire's own +/// watchdog removes its row only after its server timeout — and a server with no recorded heartbeat is assumed alive. +/// A fleet with no live server holds every capability back: the API pod asks too (a replay or rerun fork), and the next +/// worker to join may be of either build — the gap a worker rollback opens between stopping this build's workers and +/// starting the previous build's. +/// +public static class FleetCapabilityRules +{ + /// Four Hangfire heartbeats (30s apart): a busy server is never mistaken for a gone one, and a killed previous-version worker stops holding a capability back within minutes. + public static readonly TimeSpan LiveHeartbeatWindow = TimeSpan.FromMinutes(2); + + public static bool EveryLiveServerAdvertises(IReadOnlyCollection servers, string capability, DateTime nowUtc) => + servers.Any(server => IsLive(server, nowUtc)) && LiveServersLacking(servers, capability, nowUtc).Count == 0; + + /// The ids of the live servers that do not advertise — the ones holding it back. + public static IReadOnlyList LiveServersLacking(IReadOnlyCollection servers, string capability, DateTime nowUtc) => + servers.Where(server => IsLive(server, nowUtc) && !Advertises(server, capability)).Select(server => server.ServerId).ToList(); + + private static bool IsLive(WorkerServerHeartbeat server, DateTime nowUtc) => server.LastHeartbeatUtc is not { } heartbeat || nowUtc - heartbeat <= LiveHeartbeatWindow; + + private static bool Advertises(WorkerServerHeartbeat server, string capability) => server.Queues.Contains(capability, StringComparer.Ordinal); +} diff --git a/backend/src/CodeSpace.Core/Services/Fleet/HangfireWorkerFleetRegistry.cs b/backend/src/CodeSpace.Core/Services/Fleet/HangfireWorkerFleetRegistry.cs new file mode 100644 index 000000000..ae452e1a6 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/HangfireWorkerFleetRegistry.cs @@ -0,0 +1,29 @@ +using CodeSpace.Core.DependencyInjection; +using CodeSpace.Messages.Fleet; +using Hangfire; +using Hangfire.Storage.Monitoring; + +namespace CodeSpace.Core.Services.Fleet; + +/// +/// The fleet as Hangfire's server table records it. Every build — this one and every previous one — announces each of +/// its job servers there with the queues it listens on and refreshes a heartbeat, so it is the one registry a +/// previous-version worker writes without knowing this reader exists. +/// +/// The storage is optional: a host that registered no Hangfire storage (a test container with no job server) +/// observes no fleet. Every deployed role registers storage — the public pod too, to enqueue. +/// +public sealed class HangfireWorkerFleetRegistry : IWorkerFleetRegistry, ISingletonDependency +{ + private readonly JobStorage? _storage; + + public HangfireWorkerFleetRegistry(JobStorage? storage = null) => _storage = storage; + + public bool ObservesFleet => _storage is not null; + + public IReadOnlyList ReadServers() => _storage is null ? [] : _storage.GetMonitoringApi().Servers().Select(ToHeartbeat).ToList(); + + /// Hangfire stores the heartbeat as a UTC timestamp without a zone, so it comes back unspecified; it is UTC. + private static WorkerServerHeartbeat ToHeartbeat(ServerDto server) => + new(server.Name, server.Queues?.ToList() ?? [], server.Heartbeat is { } heartbeat ? DateTime.SpecifyKind(heartbeat, DateTimeKind.Utc) : null); +} diff --git a/backend/src/CodeSpace.Core/Services/Fleet/IFleetCapabilities.cs b/backend/src/CodeSpace.Core/Services/Fleet/IFleetCapabilities.cs new file mode 100644 index 000000000..643bd1eda --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/IFleetCapabilities.cs @@ -0,0 +1,11 @@ +namespace CodeSpace.Core.Services.Fleet; + +/// +/// Whether a new data shape may be written yet: true only when a worker of the deployment is live and every live one +/// advertises the capability to read it (). While a previous-version worker is live +/// — a rolling deploy, or a worker rollback — or no worker is, a writer keeps writing the shape every worker reads. +/// +public interface IFleetCapabilities +{ + bool AllWorkersSupport(string capability); +} diff --git a/backend/src/CodeSpace.Core/Services/Fleet/IWorkerFleetRegistry.cs b/backend/src/CodeSpace.Core/Services/Fleet/IWorkerFleetRegistry.cs new file mode 100644 index 000000000..359961daa --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Fleet/IWorkerFleetRegistry.cs @@ -0,0 +1,12 @@ +using CodeSpace.Messages.Fleet; + +namespace CodeSpace.Core.Services.Fleet; + +/// The registry every job server of the deployment heartbeats into, read as it stands — live and stale servers alike; liveness is ' call. +public interface IWorkerFleetRegistry +{ + /// False on a host with no registry at all (a test host with no job storage): it runs no job and shares its rows with no worker, so it has no fleet to hold a shape back for. + bool ObservesFleet { get; } + + IReadOnlyList ReadServers(); +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/ISealedPayloadUnsealService.cs b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/ISealedPayloadUnsealService.cs new file mode 100644 index 000000000..f61f064b0 --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/ISealedPayloadUnsealService.cs @@ -0,0 +1,9 @@ +using CodeSpace.Messages.Commands.Workflows; + +namespace CodeSpace.Core.Services.Workflows.Runtime; + +/// The pre-rollback restore of sealed payloads (): every non-terminal agent task and run payload that holds ciphertext goes back to the exact, unsealed row a previous-version worker executes. +public interface ISealedPayloadUnsealService +{ + Task UnsealNonTerminalAsync(CancellationToken cancellationToken); +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/SealedPayloadUnsealService.cs b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/SealedPayloadUnsealService.cs new file mode 100644 index 000000000..cf9d7ad7b --- /dev/null +++ b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/SealedPayloadUnsealService.cs @@ -0,0 +1,168 @@ +using System.Security.Cryptography; +using System.Text.Json; +using CodeSpace.Core.DependencyInjection; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Services.Agents; +using CodeSpace.Core.Services.Completion; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Commands.Workflows; +using CodeSpace.Messages.Enums; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace CodeSpace.Core.Services.Workflows.Runtime; + +/// +/// Restores each sealed row a previous-version worker could still execute to the shape the previous build wrote: the +/// exact payload in the public column and no ciphertext. That is every non-terminal run payload (a pending child is +/// started, a suspended one resumed, from its payload) and every Queued agent task. A Running agent task stays sealed: +/// Running is reached only from Queued, so no worker of either build launches it again, and a current worker that +/// re-attaches it — a rollback aborted, a rolling worker deploy — rebuilds the launch's redactor from the sealed copy. +/// Each row is a compare-and-swap on the ciphertext it opened and on the row still being in that state, so a worker +/// that rewrote or claimed the row meanwhile wins and the row is counted as raced. +/// +/// An agent task that carries a workflow-agent acceptance contract was staked from its MASKED task, and the +/// completion composer re-derives the unit's contract hash from task_jsonb of the unit's first attempt. Restoring +/// that first attempt's exact goal would leave its receipt attesting a hash the ledger never staked, so the unit is +/// re-staked from the restored task in the same transaction — exactly what the previous build staked for that row. A +/// retry's restore needs nothing: the composer keys the unit on the first attempt, which this pass does not touch. +/// +public sealed class SealedPayloadUnsealService : ISealedPayloadUnsealService, IScopedDependency +{ + private static readonly WorkflowRunStatus[] LiveRunStatuses = [WorkflowRunStatus.Pending, WorkflowRunStatus.Enqueued, WorkflowRunStatus.Running, WorkflowRunStatus.Suspended]; + + private readonly CodeSpaceDbContext _db; + private readonly WorkflowSensitivePayloadSealer _sealer; + private readonly ICompletionContractStore _contracts; + private readonly ILogger _logger; + + public SealedPayloadUnsealService(CodeSpaceDbContext db, WorkflowSensitivePayloadSealer sealer, ICompletionContractStore contracts, ILogger logger) + { + _db = db; + _sealer = sealer; + _contracts = contracts; + _logger = logger; + } + + public async Task UnsealNonTerminalAsync(CancellationToken cancellationToken) + { + var agentTasks = await UnsealAgentTasksAsync(cancellationToken).ConfigureAwait(false); + var runPayloads = await UnsealRunPayloadsAsync(cancellationToken).ConfigureAwait(false); + + _logger.LogWarning("Unsealed non-terminal payloads for a worker rollback: {AgentTasks} agent tasks, {RunPayloads} run payloads restored; {Raced} raced, {Unopenable} unopenable", agentTasks.Restored, runPayloads.Restored, agentTasks.Raced + runPayloads.Raced, agentTasks.Unopenable + runPayloads.Unopenable); + + return new UnsealNonTerminalPayloadsResponse { AgentTasks = agentTasks.Restored, RunPayloads = runPayloads.Restored, Raced = agentTasks.Raced + runPayloads.Raced, Unopenable = agentTasks.Unopenable + runPayloads.Unopenable }; + } + + // ── Agent tasks ─────────────────────────────────────────────────────────── + + private async Task UnsealAgentTasksAsync(CancellationToken cancellationToken) + { + var rows = await _db.AgentRun.AsNoTracking() + .Where(r => r.SealedTaskCiphertext != null && r.Status == AgentRunStatus.Queued) + .Select(r => new SealedAgentTask(r.Id, r.TeamId, r.WorkflowRunId, r.NodeId, r.IterationKey, r.CreatedDate, r.TaskJson, r.SealedTaskCiphertext!)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var tally = new UnsealTally(); + + foreach (var row in rows) tally.Count(await UnsealAgentTaskAsync(row, cancellationToken).ConfigureAwait(false)); + + return tally; + } + + private async Task UnsealAgentTaskAsync(SealedAgentTask row, CancellationToken cancellationToken) + { + if (TryOpen(row.TaskJson, row.Ciphertext, "agent run", row.Id) is not { } exact) return UnsealOutcome.Unopenable; + + await using var transaction = await _db.Database.BeginTransactionAsync(cancellationToken).ConfigureAwait(false); + + var restored = await _db.AgentRun + .Where(r => r.Id == row.Id && r.SealedTaskCiphertext == row.Ciphertext && r.Status == AgentRunStatus.Queued) + .ExecuteUpdateAsync(s => s.SetProperty(r => r.TaskJson, exact).SetProperty(r => r.SealedTaskCiphertext, (string?)null), cancellationToken).ConfigureAwait(false); + + if (restored != 1) return UnsealOutcome.Raced; + + await RestakeFirstAttemptAsync(row, exact, cancellationToken).ConfigureAwait(false); + await transaction.CommitAsync(cancellationToken).ConfigureAwait(false); + + return UnsealOutcome.Restored; + } + + private async Task RestakeFirstAttemptAsync(SealedAgentTask row, string exactTaskJson, CancellationToken cancellationToken) + { + if (row.WorkflowRunId is not { } workflowRunId) return; + + var task = JsonSerializer.Deserialize(exactTaskJson, AgentJson.Options)!; + var requirements = AgentAcceptanceContract.WorkflowAgentRequirements(task, row.NodeId, row.IterationKey, workflowRunId); + + if (requirements.Count == 0 || await HasEarlierAttemptAsync(row, workflowRunId, cancellationToken).ConfigureAwait(false)) return; + + await _contracts.UpsertRequirementsAsync(workflowRunId, row.TeamId, requirements, cancellationToken).ConfigureAwait(false); + } + + private Task HasEarlierAttemptAsync(SealedAgentTask row, Guid workflowRunId, CancellationToken cancellationToken) => + _db.AgentRun.AnyAsync(r => r.WorkflowRunId == workflowRunId && r.TeamId == row.TeamId && r.NodeId == row.NodeId && r.IterationKey == row.IterationKey && r.CreatedDate < row.CreatedDate, cancellationToken); + + // ── Run payloads ────────────────────────────────────────────────────────── + + private async Task UnsealRunPayloadsAsync(CancellationToken cancellationToken) + { + var rows = await _db.WorkflowRunRequest.AsNoTracking() + .Where(q => q.SealedPayloadCiphertext != null && _db.WorkflowRun.Any(r => r.RunRequestId == q.Id && LiveRunStatuses.Contains(r.Status))) + .Select(q => new SealedRunPayload(q.Id, q.NormalizedPayloadJson, q.SealedPayloadCiphertext!)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var tally = new UnsealTally(); + + foreach (var row in rows) tally.Count(await UnsealRunPayloadAsync(row, cancellationToken).ConfigureAwait(false)); + + return tally; + } + + private async Task UnsealRunPayloadAsync(SealedRunPayload row, CancellationToken cancellationToken) + { + if (TryOpen(row.PayloadJson, row.Ciphertext, "run request", row.Id) is not { } exact) return UnsealOutcome.Unopenable; + + var restored = await _db.WorkflowRunRequest + .Where(q => q.Id == row.Id && q.SealedPayloadCiphertext == row.Ciphertext && _db.WorkflowRun.Any(r => r.RunRequestId == q.Id && LiveRunStatuses.Contains(r.Status))) + .ExecuteUpdateAsync(s => s.SetProperty(q => q.NormalizedPayloadJson, exact).SetProperty(q => q.SealedPayloadCiphertext, (string?)null), cancellationToken).ConfigureAwait(false); + + return restored == 1 ? UnsealOutcome.Restored : UnsealOutcome.Raced; + } + + // ── Shared ──────────────────────────────────────────────────────────────── + + /// The exact payload, or null when this pod's Data Protection keys cannot open the ciphertext — that row stays sealed and is named in the log, never guessed at. + private string? TryOpen(string publicJson, string ciphertext, string kind, Guid id) + { + try + { + return _sealer.Open(publicJson, ciphertext).Json; + } + catch (Exception ex) when (ex is CryptographicException or InvalidOperationException or JsonException) + { + _logger.LogError(ex, "The sealed payload of {Kind} {Id} could not be opened; it stays sealed", kind, id); + return null; + } + } + + private sealed record SealedAgentTask(Guid Id, Guid TeamId, Guid? WorkflowRunId, string? NodeId, string IterationKey, DateTimeOffset CreatedDate, string TaskJson, string Ciphertext); + + private sealed record SealedRunPayload(Guid Id, string PayloadJson, string Ciphertext); + + private enum UnsealOutcome { Restored, Raced, Unopenable } + + private sealed class UnsealTally + { + public int Restored { get; private set; } + public int Raced { get; private set; } + public int Unopenable { get; private set; } + + public void Count(UnsealOutcome outcome) + { + if (outcome == UnsealOutcome.Restored) Restored++; + else if (outcome == UnsealOutcome.Raced) Raced++; + else Unopenable++; + } + } +} diff --git a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/WorkflowSensitivePayloadSealer.cs b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/WorkflowSensitivePayloadSealer.cs index a4fdc876d..bcec3821b 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/Runtime/WorkflowSensitivePayloadSealer.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/Runtime/WorkflowSensitivePayloadSealer.cs @@ -1,5 +1,6 @@ using System.Text.Json; using CodeSpace.Core.DependencyInjection; +using CodeSpace.Core.Services.Fleet; using CodeSpace.Messages.Workflows; namespace CodeSpace.Core.Services.Workflows.Runtime; @@ -13,6 +14,15 @@ namespace CodeSpace.Core.Services.Workflows.Runtime; /// A payload that carries none of the given secrets is returned verbatim with no ciphertext, so a row written /// for it is byte-identical to one written before sealing existed — and reads such a row (and every /// pre-sealing row) straight from its public column. +/// +/// A worker of the previous build reads only the public column and would execute the mask. So while any live +/// worker does not advertise — a rolling deploy, a worker +/// rollback — a secret-bearing payload for a new row (, and a fork's copy, ) is +/// written verbatim and unsealed too: exactly the row the previous build wrote and the only one it executes correctly. +/// That row keeps the plaintext at rest and hands the receiving run no secret to keep masking, as the previous build +/// did; sealing resumes once every live worker reads it. Rows sealed before a previous-version worker joins are opened +/// back by UnsealNonTerminalPayloadsCommand, the pre-rollback step. A row a current worker is already running +/// keeps its shape whatever the gate says (): no previous-version worker launches it. /// public sealed class WorkflowSensitivePayloadSealer : ISingletonDependency { @@ -20,20 +30,30 @@ public sealed class WorkflowSensitivePayloadSealer : ISingletonDependency internal const string EnvelopeKind = "sealed.payload.v1"; private readonly IWorkflowSensitivePayloadProtector _protector; + private readonly IFleetCapabilities _fleet; + + public WorkflowSensitivePayloadSealer(IWorkflowSensitivePayloadProtector protector, IFleetCapabilities fleet) + { + _protector = protector; + _fleet = fleet; + } + + /// Mask the embeds for its public column; encrypt the exact payload together with exactly those values when anything was masked — once every live worker can read that. Until then, and whenever nothing is embedded, the payload is written verbatim with no ciphertext. For a new row: any worker may be the one that executes it. + public SealedPayload Seal(JsonElement payload, PersistenceSecretRedactor secrets) => Seal(payload, secrets, throughFleetGate: true); - public WorkflowSensitivePayloadSealer(IWorkflowSensitivePayloadProtector protector) => _protector = protector; + /// for the row a current worker is already running, rewritten by that worker (an agent run's resolved model): sealed whenever something is embedded, whatever the fleet gate says. A previous-version worker never launches such a row, and the run's launch built its redactor from the secrets the row names, which a re-attach rebuilds from the same row. + public SealedPayload Reseal(JsonElement payload, PersistenceSecretRedactor secrets) => Seal(payload, secrets, throughFleetGate: false); - /// Mask the embeds for its public column; encrypt the exact payload together with exactly those values when anything was masked. - public SealedPayload Seal(JsonElement payload, PersistenceSecretRedactor secrets) + /// A stored payload copied onto a new row — a replay or rerun fork of a run: its exact copy and secrets opened from and , then sealed again through the fleet gate exactly as a fresh payload is. A copy of an unsealed payload is carried over byte-identical. + public SealedPayload SealCopy(string publicJson, string? ciphertext) { - var embedded = secrets.FoundIn(payload); + var opened = Open(publicJson, ciphertext); - if (embedded.Count == 0) return new SealedPayload(payload.GetRawText(), null); + if (opened.Secrets.Count == 0) return new SealedPayload(opened.Json, null); - var masked = secrets.Redact(payload).Value.GetRawText(); - var envelope = JsonSerializer.Serialize(new SealedEnvelope(EnvelopeKind, payload, embedded)); + using var document = JsonDocument.Parse(opened.Json); - return new SealedPayload(masked, _protector.Protect(envelope)); + return Seal(document.RootElement, new PersistenceSecretRedactor(opened.Secrets)); } /// The exact payload and the secret values it embeds — read from when the row was sealed, else is itself exact and embeds none. @@ -49,5 +69,17 @@ public OpenedPayload Open(string publicJson, string? ciphertext) return new OpenedPayload(envelope.Payload.GetRawText(), envelope.Secrets); } + private SealedPayload Seal(JsonElement payload, PersistenceSecretRedactor secrets, bool throughFleetGate) + { + var embedded = secrets.FoundIn(payload); + + if (embedded.Count == 0 || (throughFleetGate && !_fleet.AllWorkersSupport(FleetCapabilityMarkers.SealedPayloadV1))) return new SealedPayload(payload.GetRawText(), null); + + var masked = secrets.Redact(payload).Value.GetRawText(); + var envelope = JsonSerializer.Serialize(new SealedEnvelope(EnvelopeKind, payload, embedded)); + + return new SealedPayload(masked, _protector.Protect(envelope)); + } + private sealed record SealedEnvelope(string Kind, JsonElement Payload, IReadOnlyList Secrets); } diff --git a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs index c47fc341f..18e3744ac 100644 --- a/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs +++ b/backend/src/CodeSpace.Core/Services/Workflows/WorkflowService.cs @@ -44,6 +44,7 @@ public sealed class WorkflowService : IWorkflowService, IScopedDependency private readonly IWorkflowRunDispatcher _runDispatcher; private readonly Engine.IWorkflowResumeService _resumeService; private readonly IPostCommitActions _postCommit; + private readonly WorkflowSensitivePayloadSealer _sealer; private readonly IAgentRunService _agentRunService; private readonly IRunningAgentCancellation _runningAgents; private readonly Rerun.IRerunCellSeeder _cellSeeder; @@ -74,6 +75,7 @@ public WorkflowService(CodeSpaceDbContext db, WorkflowDefinitionServices definit _runDispatcher = launch.Dispatcher; _resumeService = control.Resume; _postCommit = launch.PostCommit; + _sealer = launch.Sealer; _agentRunService = control.Agents; _runningAgents = control.RunningAgents; _cellSeeder = control.Cells; @@ -606,12 +608,16 @@ private async Task StageForkedRunAsync(WorkflowRun original, List StageForkedRunAsync(WorkflowRun original, List +/// The pre-rollback step for workers: every queued agent run and every workflow run that is not yet terminal and still +/// holds a sealed payload (0247) gets its exact payload back in the public column and the ciphertext cleared — the row +/// the previous build wrote and the only one a previous-version worker can execute; left sealed, that worker runs the +/// mask. +/// +/// Order for a worker rollback: stop every current worker and wait until none is live — its servers gone from the +/// Hangfire server list, or two minutes without a heartbeat — and 30 seconds more, one fleet-gate snapshot; send this +/// from a current API pod; then start the previous-version workers. From the end of that wait, the fleet gate holds +/// sealing back (first no worker is live, then a previous-version one is), so current pods write every new hand-off +/// unsealed — an agent task, a child's payload, a replay or rerun fork of a sealed run — and this only has to catch the +/// rows sealed before. Terminal rows stay sealed: nothing executes them again, and a fork of one is written unsealed by +/// the current API that stages it. A Running agent run stays sealed too: no worker launches it again, and the sealed +/// copy is what a current worker's re-attach rebuilds its redactor from. A previous-version API pod knows no sealed +/// column: a replay or rerun it stages of a sealed run copies the masked column alone and runs with the mask, whichever +/// worker executes it. +/// +/// Instance-level and global-admin only: it spans every team and puts secret values back into columns team members +/// can read, exactly as the previous build stored them. NOT transactional (): +/// each row is restored by its own compare-and-swap, so a row a worker rewrote meanwhile is left as found and counted, +/// and one unopenable row never undoes the others. Idempotent — a second send finds nothing left to restore. +/// +public sealed record UnsealNonTerminalPayloadsCommand : ICommand, IRequireGlobalAdmin, INonTransactionalCommand; + +/// What the pass restored, and what it had to leave sealed: rows changed under it (send again once the workers are stopped), rows whose ciphertext this pod's keys cannot open. +public sealed record UnsealNonTerminalPayloadsResponse +{ + public required int AgentTasks { get; init; } + public required int RunPayloads { get; init; } + public required int Raced { get; init; } + public required int Unopenable { get; init; } +} diff --git a/backend/src/CodeSpace.Messages/Fleet/WorkerServerHeartbeat.cs b/backend/src/CodeSpace.Messages/Fleet/WorkerServerHeartbeat.cs new file mode 100644 index 000000000..143b16b2c --- /dev/null +++ b/backend/src/CodeSpace.Messages/Fleet/WorkerServerHeartbeat.cs @@ -0,0 +1,8 @@ +namespace CodeSpace.Messages.Fleet; + +/// +/// One job server of the worker fleet as the fleet registry last recorded it: its id, the queues it listens on — where +/// a capability marker rides, never enqueued to (FleetCapabilityMarkers) — and its last heartbeat in UTC. A +/// null heartbeat means the registry recorded none; it is treated as alive. +/// +public sealed record WorkerServerHeartbeat(string ServerId, IReadOnlyList Queues, DateTime? LastHeartbeatUtc); diff --git a/backend/tests/CodeSpace.E2ETests/Infrastructure/DeferredDrainFleetRegistry.cs b/backend/tests/CodeSpace.E2ETests/Infrastructure/DeferredDrainFleetRegistry.cs new file mode 100644 index 000000000..713acefc0 --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Infrastructure/DeferredDrainFleetRegistry.cs @@ -0,0 +1,21 @@ +using CodeSpace.Core.Constants; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Messages.Fleet; + +namespace CodeSpace.E2ETests.Infrastructure; + +/// +/// The worker fleet of a host whose jobs the TEST drains (): that drain is its one +/// worker, it runs this build, and it is live for as long as the host is. The host itself takes the Api role and starts +/// no job server, so Hangfire's server table is empty, and the production fleet gate reads an empty fleet as "the next +/// worker may be a previous-version one" — right for an API pod between worker generations, wrong for a host that runs +/// every job it enqueues. Only the registry is replaced; the gate itself () is the +/// production one. +/// +public sealed class DeferredDrainFleetRegistry : IWorkerFleetRegistry +{ + public bool ObservesFleet => true; + + /// One server with no recorded heartbeat — the rules take it as alive — listening where the drain runs jobs, with every marker this build advertises. + public IReadOnlyList ReadServers() => [new("deferred-job-drain", [HangfireConstants.DefaultQueue, HangfireConstants.AgentQueue, .. FleetCapabilityMarkers.Advertised], null)]; +} diff --git a/backend/tests/CodeSpace.E2ETests/Infrastructure/RecurringJobWorkerHostFactory.cs b/backend/tests/CodeSpace.E2ETests/Infrastructure/RecurringJobWorkerHostFactory.cs index 3e5b39390..ef1eb8d9e 100644 --- a/backend/tests/CodeSpace.E2ETests/Infrastructure/RecurringJobWorkerHostFactory.cs +++ b/backend/tests/CodeSpace.E2ETests/Infrastructure/RecurringJobWorkerHostFactory.cs @@ -67,6 +67,9 @@ public RecurringJobWorkerHostFactory(ILogEventSink logSink) /// The per-run GUID database. public string DatabaseName { get; } + /// The started host, so a test can stop it the way a worker pod is stopped — its job servers leave the server table — while the database stays for the test to read. + public IHost? StartedHost { get; private set; } + /// The per-run database's connection string, so the test can open its OWN Hangfire storage handle onto the rows this host's job servers drain, rather than reading through the process-wide JobStorage.Current. public string ConnectionString => _testConnectionString; @@ -151,6 +154,6 @@ protected override IHost CreateHost(IHostBuilder builder) { builder.UseSerilog(new LoggerConfiguration().MinimumLevel.Warning().MinimumLevel.Override(typeof(LocalProcessRunner).FullName!, LogEventLevel.Information).WriteTo.Sink(_logSink).CreateLogger(), dispose: true); - return base.CreateHost(builder); + return StartedHost = base.CreateHost(builder); } } diff --git a/backend/tests/CodeSpace.E2ETests/Infrastructure/TaskLaunchApiFactory.cs b/backend/tests/CodeSpace.E2ETests/Infrastructure/TaskLaunchApiFactory.cs index 0a440aab1..0b6bd09c4 100644 --- a/backend/tests/CodeSpace.E2ETests/Infrastructure/TaskLaunchApiFactory.cs +++ b/backend/tests/CodeSpace.E2ETests/Infrastructure/TaskLaunchApiFactory.cs @@ -111,6 +111,8 @@ protected override IHost CreateHost(IHostBuilder builder) { // Singleton so deferred jobs survive past the request scope — the test drains them after the POST. b.RegisterType().As().AsSelf().SingleInstance(); + // The drain is this host's worker, and the fleet gate must see it as one (a current one). + b.RegisterType().As().SingleInstance(); ConfigureContractTestServices(b); }); diff --git a/backend/tests/CodeSpace.E2ETests/Jobs/WorkerFleetCapabilityE2ETests.cs b/backend/tests/CodeSpace.E2ETests/Jobs/WorkerFleetCapabilityE2ETests.cs new file mode 100644 index 000000000..3f437f618 --- /dev/null +++ b/backend/tests/CodeSpace.E2ETests/Jobs/WorkerFleetCapabilityE2ETests.cs @@ -0,0 +1,167 @@ +using CodeSpace.Api.Extensions.Hangfire; +using CodeSpace.Core.Constants; +using CodeSpace.Core.Jobs.RecurringJobs; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.E2ETests.Infrastructure; +using Hangfire; +using Hangfire.PostgreSql; +using Hangfire.PostgreSql.Factories; +using Hangfire.Server; +using Hangfire.Storage; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging.Abstractions; +using Serilog.Core; +using Serilog.Events; +using Shouldly; + +namespace CodeSpace.E2ETests.Jobs; + +/// +/// What a rolling worker deploy depends on, observed on the real worker host: every job server it starts advertises the +/// fleet capability markers in Hangfire's server table, its fleet gate reads that table and says yes for an all-current +/// fleet and no the moment a previous-version worker's servers sit beside it or once no server is left, and its +/// schedule keeper reschedules the recurring entries a previous-version scheduler disabled. +/// +/// Tier: 🟢 High-fidelity (Rule 12) — the production WorkerHangfireRegistrar starts two real +/// BackgroundJobServers against a throw-away Postgres database; the gate is the host's own +/// HangfireWorkerFleetRegistry over the host's own JobStorage; the previous-version servers are +/// announced through Hangfire's own AnnounceServer with the queues the previous build's registrar gave them; +/// the disabled entries are written with Hangfire's own write transaction, field for field what its scheduler leaves. +/// The keeper's pass is invoked directly rather than waiting out its five-minute interval. +/// +[Trait("Category", "E2E")] +[Trait("Surface", "Worker")] +public sealed class WorkerFleetCapabilityE2ETests +{ + [Fact] + public async Task The_worker_host_advertises_its_capabilities_and_its_gate_sees_a_previous_version_worker_join() + { + await using var factory = new RecurringJobWorkerHostFactory(new DiscardingSink()); + await factory.InitializeAsync(); + _ = factory.Services; // builds and starts the host, which starts both job servers + + var storage = OpenStorageOnTheFixtureDatabase(factory); + var servers = await WaitForServersAsync(storage, expected: 2); + + foreach (var server in servers) + FleetCapabilityMarkers.Advertised.Except(server.Queues).ShouldBeEmpty($"server {server.Name} must advertise every capability marker, or its peers take it for a previous-version worker and never seal"); + + FreshGate(factory).AllWorkersSupport(FleetCapabilityMarkers.SealedPayloadV1).ShouldBeTrue("an all-current fleet reads sealed payloads"); + + AnnouncePreviousVersionWorker(storage); + + FreshGate(factory).AllWorkersSupport(FleetCapabilityMarkers.SealedPayloadV1).ShouldBeFalse("a live previous-version worker reads only the public column, so sealing must wait for it to go"); + } + + [Fact] + public async Task Once_the_worker_hosts_servers_stop_the_gate_holds_sealing_back_until_a_worker_is_live_again() + { + // A worker rollback's gap: every current worker stopped, the previous ones not started. The API pod still forks + // runs through the gate, and the next worker to join may be a previous-version one. + await using var factory = new RecurringJobWorkerHostFactory(new DiscardingSink()); + await factory.InitializeAsync(); + _ = factory.Services; + + var storage = OpenStorageOnTheFixtureDatabase(factory); + await WaitForServersAsync(storage, expected: 2); + GateOver(storage).AllWorkersSupport(FleetCapabilityMarkers.SealedPayloadV1).ShouldBeTrue("precondition: the host's own servers are live and current"); + + await factory.StartedHost.ShouldNotBeNull().StopAsync(TimeSpan.FromSeconds(30)); // a worker pod's graceful stop + await WaitForServersGoneAsync(storage); + + GateOver(storage).AllWorkersSupport(FleetCapabilityMarkers.SealedPayloadV1).ShouldBeFalse("with no live worker the gate cannot vouch for the next one to join, so a fork staged now must be the shape every worker reads"); + } + + [Fact] + public async Task The_worker_hosts_keeper_reschedules_entries_a_previous_version_scheduler_disabled() + { + await using var factory = new RecurringJobWorkerHostFactory(new DiscardingSink()); + await factory.InitializeAsync(); + var keeper = factory.Services.GetServices().OfType().ShouldHaveSingleItem("the Worker role must run exactly one schedule keeper"); + + var storage = OpenStorageOnTheFixtureDatabase(factory); + var disabled = new[] { nameof(DepartedMemberCredentialRetirementRecurringJob), nameof(DisconnectedCredentialBindingRepairRecurringJob), nameof(WebhookClaimPurgeRecurringJob) }; + DisableAsAPreviousVersionSchedulerDoes(storage, disabled); + + keeper.Reassert(); + + foreach (var entry in Entries(storage).Where(entry => disabled.Contains(entry.Id))) + entry.NextExecution.ShouldNotBeNull($"{entry.Id} stays off forever unless something reschedules it — no current worker boots again after a rolling deploy finishes. Check RecurringJobScheduleKeeper.Reassert and that AddOrUpdate still reschedules an entry carrying an Error."); + } + + private static IFleetCapabilities FreshGate(RecurringJobWorkerHostFactory factory) => + new FleetCapabilities(factory.Services.GetRequiredService(), TimeProvider.System, NullLogger.Instance); + + private static IFleetCapabilities GateOver(JobStorage storage) => new FleetCapabilities(new HangfireWorkerFleetRegistry(storage), TimeProvider.System, NullLogger.Instance); + + private static JobStorage OpenStorageOnTheFixtureDatabase(RecurringJobWorkerHostFactory factory) + { + var options = HangfireRegistrarBase.BuildStorageOptions(); + return new PostgreSqlStorage(new NpgsqlConnectionFactory(factory.ConnectionString, options), options); + } + + /// Hangfire announces a server shortly after the host starts; the predicate starts false, so an empty read can never pass. + private static async Task> WaitForServersAsync(JobStorage storage, int expected) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); + IReadOnlyList servers = []; + + while (servers.Count < expected && DateTime.UtcNow < deadline) + { + await Task.Delay(TimeSpan.FromMilliseconds(250)); + servers = storage.GetMonitoringApi().Servers().ToList(); + } + + servers.Count.ShouldBe(expected, $"the worker host must start its control and agent servers within 30s; check WorkerHangfireRegistrar.RegisterHangfire and `select id, data from hangfire.server` on the test database"); + + return servers; + } + + /// A gracefully stopped server removes its own row; the predicate reads first, so a table that never held the servers cannot pass for one they left. + private static async Task WaitForServersGoneAsync(JobStorage storage) + { + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(30); + var servers = storage.GetMonitoringApi().Servers().Count; + + while (servers > 0 && DateTime.UtcNow < deadline) + { + await Task.Delay(TimeSpan.FromMilliseconds(250)); + servers = storage.GetMonitoringApi().Servers().Count; + } + + servers.ShouldBe(0, "a stopped BackgroundJobServer must remove its row from hangfire.server within 30s; check `select id, lastheartbeat from hangfire.server` on the test database"); + } + + private static void AnnouncePreviousVersionWorker(JobStorage storage) + { + using var connection = storage.GetConnection(); + connection.AnnounceServer($"codespace-control-previous:{Environment.ProcessId}:{Guid.NewGuid():N}", new ServerContext { Queues = [HangfireConstants.DefaultQueue], WorkerCount = 4 }); + connection.AnnounceServer($"codespace-agents-previous:{Environment.ProcessId}:{Guid.NewGuid():N}", new ServerContext { Queues = [HangfireConstants.AgentQueue], WorkerCount = 4 }); + } + + private static void DisableAsAPreviousVersionSchedulerDoes(JobStorage storage, IEnumerable recurringJobIds) + { + using var connection = storage.GetConnection(); + using var transaction = connection.CreateWriteTransaction(); + + foreach (var id in recurringJobIds) + { + transaction.SetRangeInHash($"recurring-job:{id}", new Dictionary { ["NextExecution"] = string.Empty, ["Error"] = "Hangfire.Common.JobLoadException: Could not load the job.", ["RetryAttempt"] = "5" }); + transaction.AddToSet("recurring-jobs", id, -1.0); + } + + transaction.Commit(); + } + + private static IReadOnlyList Entries(JobStorage storage) + { + using var connection = storage.GetConnection(); + return connection.GetRecurringJobs(); + } + + private sealed class DiscardingSink : ILogEventSink + { + public void Emit(LogEvent logEvent) { } + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Fleet/HangfireFleetHarness.cs b/backend/tests/CodeSpace.IntegrationTests/Fleet/HangfireFleetHarness.cs new file mode 100644 index 000000000..9eab17f78 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Fleet/HangfireFleetHarness.cs @@ -0,0 +1,91 @@ +using Autofac; +using CodeSpace.Api.Extensions.Hangfire; +using CodeSpace.Core.Constants; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Core.Services.Workflows.Runtime; +using CodeSpace.IntegrationTests.Infrastructure; +using Hangfire; +using Hangfire.PostgreSql; +using Hangfire.PostgreSql.Factories; +using Hangfire.Server; +using Npgsql; + +namespace CodeSpace.IntegrationTests.Fleet; + +/// +/// A real Hangfire server table, on the fixture's Postgres in a schema of its own, that a test fills the way job servers +/// fill it in a deployment: registers this build's two pools with the exact queue +/// lists WorkerHangfireRegistrar gives them, registers the two +/// pools the previous build's registrar started (its own queue only — it never knew a capability marker). Both go +/// through Hangfire's own AnnounceServer, the call a starting BackgroundJobServer makes. +/// +/// hands out a fixture scope whose fleet gate, registry and sealer read THIS table +/// through the production classes — fresh instances, so no snapshot is shared with another test or with the +/// fixture's root container (which has no job storage and therefore observes no fleet: its gate holds nothing back). +/// +internal sealed class HangfireFleetHarness : IDisposable +{ + private readonly string _connectionString; + + public HangfireFleetHarness(string connectionString) + { + _connectionString = connectionString; + Schema = $"hangfire_fleet_{Guid.NewGuid():N}"; + + var options = HangfireRegistrarBase.BuildStorageOptions(); + options.SchemaName = Schema; + + Storage = new PostgreSqlStorage(new NpgsqlConnectionFactory(connectionString, options), options); + } + + public string Schema { get; } + + public JobStorage Storage { get; } + + /// This build's worker: its control and agent pools, each listing the capability markers. + public IReadOnlyList AnnounceCurrentWorker() => [Announce("codespace-control-current", WorkerHangfireRegistrar.ControlQueues), Announce("codespace-agents-current", WorkerHangfireRegistrar.AgentQueues)]; + + /// A previous-version worker: the same two pools on the queues its registrar listed — default and agents, nothing else. + public IReadOnlyList AnnouncePreviousVersionWorker() => [Announce("codespace-control-previous", [HangfireConstants.DefaultQueue]), Announce("codespace-agents-previous", [HangfireConstants.AgentQueue])]; + + public string Announce(string serverName, string[] queues) + { + var serverId = $"{serverName}:{Environment.ProcessId}:{Guid.NewGuid():N}"; + + using var connection = Storage.GetConnection(); + connection.AnnounceServer(serverId, new ServerContext { Queues = queues, WorkerCount = 1 }); + + return serverId; + } + + /// Back-date a server's last heartbeat — a worker that was killed and whose row Hangfire's watchdog has not removed yet. + public void AgeHeartbeats(IEnumerable serverIds, TimeSpan age) + { + using var connection = new NpgsqlConnection(_connectionString); + connection.Open(); + + using var command = new NpgsqlCommand($"UPDATE \"{Schema}\".server SET lastheartbeat = lastheartbeat - @age WHERE id = ANY(@ids)", connection); + command.Parameters.AddWithValue("age", age); + command.Parameters.AddWithValue("ids", serverIds.ToArray()); + command.ExecuteNonQuery(); + } + + /// A fixture scope whose job storage is this table, with its own fleet gate, registry and sealer so their snapshot is this test's alone. + public ILifetimeScope BeginScope(PostgresFixture fixture, Action? configure = null) => fixture.BeginScope(builder => + { + builder.RegisterInstance(Storage).As(); + builder.RegisterType().AsSelf().As().SingleInstance(); + builder.RegisterType().AsSelf().As().SingleInstance(); + builder.RegisterType().AsSelf().SingleInstance(); + configure?.Invoke(builder); + }); + + public void Dispose() + { + using var connection = new NpgsqlConnection(_connectionString); + connection.Open(); + + using var command = new NpgsqlCommand($"DROP SCHEMA IF EXISTS \"{Schema}\" CASCADE", connection); + command.ExecuteNonQuery(); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Fleet/RollingWorkerPayloadSealingFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Fleet/RollingWorkerPayloadSealingFlowTests.cs new file mode 100644 index 000000000..b80a8aeac --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Fleet/RollingWorkerPayloadSealingFlowTests.cs @@ -0,0 +1,404 @@ +using System.Text.Json; +using Autofac; +using CodeSpace.Core.Authorization; +using CodeSpace.Core.Persistence.Db; +using CodeSpace.Core.Persistence.Entities; +using CodeSpace.Core.Services.Agents; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Core.Services.Variables; +using CodeSpace.Core.Services.Workflows; +using CodeSpace.Core.Services.Workflows.Engine; +using CodeSpace.Core.Services.Workflows.Runtime; +using CodeSpace.IntegrationTests.Infrastructure; +using CodeSpace.IntegrationTests.Infrastructure.Jobs; +using CodeSpace.IntegrationTests.Workflows.Infrastructure; +using CodeSpace.Messages.Agents; +using CodeSpace.Messages.Commands.Workflows; +using CodeSpace.Messages.Constants; +using CodeSpace.Messages.Contracts; +using CodeSpace.Messages.Dtos.Workflows; +using CodeSpace.Messages.Enums; +using MediatR; +using Microsoft.EntityFrameworkCore; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Fleet; + +/// +/// Sealed hand-offs (0247) during a rolling worker deploy and before a worker rollback, against real Postgres and a real +/// Hangfire server table: while a previous-version worker is live, the engine stages a flow.subworkflow child's +/// payload and an agent.run task in the exact shape the previous build wrote — and the previous build's read of +/// those rows (deserialize the public column, nothing else; quoted from a9cf96d35^ at each assertion) gets the +/// real secret, not the mask. Once every live worker advertises the capability, the same staging seals. A replay or +/// rerun fork of a sealed child is a new row too, written in the shape every worker that may run it reads. The admin +/// unseal restores the sealed rows a previous-version worker could still execute to that same shape and leaves terminal +/// rows and launched agent runs sealed. +/// +/// Real engine, real fleet registry over Hangfire's own AnnounceServer rows, real sealer and Data +/// Protection; the previous build's read path is an inline mirror of its two lines (the scripted two-build run in the +/// PR exercises the previous build's binary itself). +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public sealed class RollingWorkerPayloadSealingFlowTests +{ + private readonly PostgresFixture _fixture; + + public RollingWorkerPayloadSealingFlowTests(PostgresFixture fixture) { _fixture = fixture; } + + [Fact] + public async Task While_a_previous_version_worker_is_live_the_handoffs_it_may_execute_are_written_exact() + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + fleet.AnnounceCurrentWorker(); + fleet.AnnouncePreviousVersionWorker(); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + + var (childRunId, agentRunId) = await StageHandoffsAsync(fleet, team); + + var request = await LoadRunRequestAsync(childRunId); + request.SealedPayloadCiphertext.ShouldBeNull("a previous-version worker never reads the sealed column"); + PreviousBuildTrigger(request).GetProperty("token").GetString().ShouldBe(team.Secret, "the previous build builds the child's trigger scope from normalized_payload_json alone"); + + var agentRun = await LoadAgentRunAsync(agentRunId); + agentRun.SealedTaskCiphertext.ShouldBeNull(); + PreviousBuildTask(agentRun).Goal.ShouldContain(team.Secret, customMessage: "the previous build's executor launches whatever task_jsonb deserializes to"); + + await RunEngineAsync(fleet, childRunId); + (await LoadRunAsync(childRunId)).Status.ShouldBe(WorkflowRunStatus.Success, "a current worker executes the unsealed row as well — it is every pre-0247 row's shape"); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Once_every_live_worker_reads_sealed_payloads_the_handoffs_are_sealed(bool previousVersionWorkerStoppedHeartbeating) + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + fleet.AnnounceCurrentWorker(); + if (previousVersionWorkerStoppedHeartbeating) fleet.AgeHeartbeats(fleet.AnnouncePreviousVersionWorker(), FleetCapabilityRules.LiveHeartbeatWindow + TimeSpan.FromMinutes(1)); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + + var (childRunId, agentRunId) = await StageHandoffsAsync(fleet, team); + + var request = await LoadRunRequestAsync(childRunId); + request.SealedPayloadCiphertext.ShouldNotBeNull(previousVersionWorkerStoppedHeartbeating ? "a killed worker's row lingers until Hangfire's watchdog removes it, but it holds nothing back once its heartbeat is stale" : "an all-current fleet seals"); + request.NormalizedPayloadJson.ShouldNotContain(team.Secret); + + var agentRun = await LoadAgentRunAsync(agentRunId); + agentRun.SealedTaskCiphertext.ShouldNotBeNull(); + agentRun.TaskJson.ShouldNotContain(team.Secret); + } + + public enum Fork { Replay, RerunFromNode } + + /// The fleet the API pod sees when it stages the fork: all current, a previous-version worker live beside them (a rolling deploy, a worker rollback), or no live worker at all (every current worker stopped for a rollback, the previous ones not started yet). + public enum FleetAtFork { AllCurrent, PreviousVersionWorkerLive, NoWorkerLive } + + [Theory] + [InlineData(Fork.Replay, FleetAtFork.AllCurrent)] + [InlineData(Fork.Replay, FleetAtFork.PreviousVersionWorkerLive)] + [InlineData(Fork.Replay, FleetAtFork.NoWorkerLive)] + [InlineData(Fork.RerunFromNode, FleetAtFork.AllCurrent)] + [InlineData(Fork.RerunFromNode, FleetAtFork.PreviousVersionWorkerLive)] + [InlineData(Fork.RerunFromNode, FleetAtFork.NoWorkerLive)] + public async Task A_fork_of_a_sealed_child_is_written_in_the_shape_every_worker_that_may_run_it_reads(Fork fork, FleetAtFork fleetAtFork) + { + // A finished child stays sealed through the pre-rollback unseal, and the API that forks it stays current while + // the workers roll back — so the fork is a new row a previous-version worker may execute. + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + var currentWorker = fleet.AnnounceCurrentWorker(); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + var (childRunId, _) = await StageHandoffsAsync(fleet, team); + await RunEngineAsync(fleet, childRunId); + (await LoadRunRequestAsync(childRunId)).SealedPayloadCiphertext.ShouldNotBeNull("precondition: the finished child was sealed under an all-current fleet"); + if (fleetAtFork == FleetAtFork.PreviousVersionWorkerLive) fleet.AnnouncePreviousVersionWorker(); + if (fleetAtFork == FleetAtFork.NoWorkerLive) fleet.AgeHeartbeats(currentWorker, FleetCapabilityRules.LiveHeartbeatWindow + TimeSpan.FromMinutes(1)); + + var forkRunId = await ForkAsync(fleet, team, childRunId, fork); + + var request = await LoadRunRequestAsync(forkRunId); + if (fleetAtFork == FleetAtFork.AllCurrent) + { + request.SealedPayloadCiphertext.ShouldNotBeNull("an all-current fleet keeps the fork sealed"); + request.NormalizedPayloadJson.ShouldNotContain(team.Secret); + } + else + { + request.SealedPayloadCiphertext.ShouldBeNull("the worker that runs the fork may be a previous-version one, which never reads the sealed column"); + PreviousBuildTrigger(request).GetProperty("token").GetString().ShouldBe(team.Secret, "the previous build builds the fork's trigger scope from normalized_payload_json alone"); + } + + await RunEngineAsync(fleet, forkRunId); + (await LoadRunAsync(forkRunId)).Status.ShouldBe(WorkflowRunStatus.Success, "a current worker executes the fork in either shape"); + } + + [Fact] + public void The_fleet_registry_reads_the_server_table_with_utc_heartbeats() + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + var serverId = fleet.Announce("codespace-control-probe", ["default", FleetCapabilityMarkers.SealedPayloadV1]); + + var registry = new HangfireWorkerFleetRegistry(fleet.Storage); + registry.ObservesFleet.ShouldBeTrue("every deployed role registers job storage, so an empty server table there is a fleet with no live worker, not a host with none"); + var server = registry.ReadServers().ShouldHaveSingleItem(); + + server.ServerId.ShouldBe(serverId); + server.Queues.ShouldBe(["default", FleetCapabilityMarkers.SealedPayloadV1]); + (DateTime.UtcNow - server.LastHeartbeatUtc!.Value).Duration().ShouldBeLessThan(TimeSpan.FromMinutes(1), + "a heartbeat read in the wrong zone would age every server by the host's UTC offset and hide a live previous-version worker"); + } + + [Fact] + public async Task Unsealing_restores_every_non_terminal_sealed_row_and_leaves_terminal_ones_sealed() + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + fleet.AnnounceCurrentWorker(); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + var (liveChildRunId, liveAgentRunId) = await StageHandoffsAsync(fleet, team); + var (finishedChildRunId, finishedAgentRunId) = await StageHandoffsAsync(fleet, team); + await RunEngineAsync(fleet, finishedChildRunId); + await SetAgentRunStatusAsync(finishedAgentRunId, AgentRunStatus.Cancelled); + + var response = await UnsealAsAdminAsync(); + + response.AgentTasks.ShouldBeGreaterThanOrEqualTo(1); + response.RunPayloads.ShouldBeGreaterThanOrEqualTo(1); + + var liveRequest = await LoadRunRequestAsync(liveChildRunId); + liveRequest.SealedPayloadCiphertext.ShouldBeNull(); + PreviousBuildTrigger(liveRequest).GetProperty("token").GetString().ShouldBe(team.Secret, "the pending child must start on a previous-version worker with the real value"); + + var liveAgent = await LoadAgentRunAsync(liveAgentRunId); + liveAgent.SealedTaskCiphertext.ShouldBeNull(); + PreviousBuildTask(liveAgent).Goal.ShouldContain(team.Secret, customMessage: "the queued agent must launch on a previous-version worker with the real goal"); + + (await LoadRunRequestAsync(finishedChildRunId)).SealedPayloadCiphertext.ShouldNotBeNull("a finished child executes nothing again, so its secret stays sealed"); + (await LoadAgentRunAsync(finishedAgentRunId)).SealedTaskCiphertext.ShouldNotBeNull("a cancelled agent run executes nothing again"); + + await UnsealAsAdminAsync(); + + (await LoadAgentRunAsync(liveAgentRunId)).TaskJson.ShouldBe(liveAgent.TaskJson, "a second send finds nothing left to restore"); + (await LoadRunRequestAsync(liveChildRunId)).NormalizedPayloadJson.ShouldBe(liveRequest.NormalizedPayloadJson); + } + + [Fact] + public async Task Unsealing_leaves_an_agent_run_that_already_launched_sealed() + { + // A previous-version worker never launches a Running row — Running is reached only from Queued — it only + // re-attaches one, and its re-attach cannot rebuild the launch's redactor in either shape. A current worker can, + // from the sealed copy alone: unsealed, the row would drop the secrets folded into the handle's fingerprint, and + // a re-attach after an aborted rollback would land the run marker-only, losing its tail. + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + fleet.AnnounceCurrentWorker(); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + var (_, agentRunId) = await StageHandoffsAsync(fleet, team); + await SetAgentRunStatusAsync(agentRunId, AgentRunStatus.Running); + var launched = await LoadAgentRunAsync(agentRunId); + + await UnsealAsAdminAsync(); + + var after = await LoadAgentRunAsync(agentRunId); + after.SealedTaskCiphertext.ShouldBe(launched.SealedTaskCiphertext, "a launched run keeps the shape its launch was built from"); + after.TaskJson.ShouldBe(launched.TaskJson); + } + + [Fact] + public async Task Unsealing_a_contract_bearing_first_attempt_re_stakes_it_from_the_restored_task() + { + // The composer re-derives a workflow agent's contract hash from task_jsonb and admission compares it with the + // staked one. Staked from the MASKED goal, a restored exact goal would attest a contract the ledger never staked. + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + fleet.AnnounceCurrentWorker(); + var team = await SeedTeamWithSecretAsync(); + using var manual = ResolveJobClient().ManualExecution(); + var workflowId = await CreateWorkflowAsync(team, AgentDefinition(acceptance: """{ "kind": "TestsPass", "command": ["sh", "-c", "exit 0"] }""")); + var runId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, workflowId, team.TeamId); + await RunEngineAsync(fleet, runId); + var agentRun = await LoadAgentRunForRunAsync(runId); + + (await StakedAcceptanceHashAsync(runId)).ShouldBe(AgentAcceptanceContract.Hash(PreviousBuildTask(agentRun)), "precondition: the sealed attempt staked the hash of the masked task_jsonb"); + + await UnsealAsAdminAsync(); + + var restored = await LoadAgentRunAsync(agentRun.Id); + (await StakedAcceptanceHashAsync(runId)).ShouldBe(AgentAcceptanceContract.Hash(PreviousBuildTask(restored)), "the unit is re-staked from the restored task — the hash the composer will derive from it"); + } + + [Fact] + public async Task Only_a_global_admin_can_unseal() + { + var team = await SeedTeamWithSecretAsync(); + using var scope = _fixture.BeginScopeAs(team.OwnerId, team.TeamId); + + await Should.ThrowAsync(() => scope.Resolve().Send(new UnsealNonTerminalPayloadsCommand())); + } + + // ── The previous build's read path (a9cf96d35^) ─────────────────────────────────────────────────────────── + + /// AgentRunExecutor.ExecuteAsync at a9cf96d35^: JsonSerializer.Deserialize<AgentTask>(run.TaskJson, AgentJson.Options) — the only copy of the task it knows. + private static AgentTask PreviousBuildTask(AgentRun run) => JsonSerializer.Deserialize(run.TaskJson, AgentJson.Options)!; + + /// WorkflowEngine.BuildScopeFreshAndPersistSnapshotAsync at a9cf96d35^: ParsePayloadObject(run.RunRequest?.NormalizedPayloadJson) — the child's {{trigger.*}}. + private static JsonElement PreviousBuildTrigger(WorkflowRunRequest request) => JsonDocument.Parse(request.NormalizedPayloadJson).RootElement; + + // ── Staging ────────────────────────────────────────────────────────────────────────────────────────────── + + /// A parent whose flow.subworkflow hands the team secret to a child, and a run whose agent.run goal embeds it — each parked at its suspend, so the child is staged (not run) and the agent queued (not launched). + private async Task<(Guid ChildRunId, Guid AgentRunId)> StageHandoffsAsync(HangfireFleetHarness fleet, Team team) + { + var childWorkflowId = await CreateWorkflowAsync(team, ChildDefinition()); + var parentWorkflowId = await CreateWorkflowAsync(team, ParentDefinition(childWorkflowId)); + var agentWorkflowId = await CreateWorkflowAsync(team, AgentDefinition()); + + var parentRunId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, parentWorkflowId, team.TeamId); + var agentWorkflowRunId = await WorkflowsTestSeed.SeedManualRunAsync(_fixture, agentWorkflowId, team.TeamId); + + await RunEngineAsync(fleet, parentRunId); + await RunEngineAsync(fleet, agentWorkflowRunId); + + return (await LoadChildRunIdAsync(parentRunId), (await LoadAgentRunForRunAsync(agentWorkflowRunId)).Id); + } + + private async Task RunEngineAsync(HangfireFleetHarness fleet, Guid runId) + { + using var scope = fleet.BeginScope(_fixture); + await scope.Resolve().ExecuteRunAsync(runId, CancellationToken.None); + } + + private async Task ForkAsync(HangfireFleetHarness fleet, Team team, Guid runId, Fork fork) + { + using var scope = fleet.BeginScope(_fixture); + var workflows = scope.Resolve(); + + return fork == Fork.Replay + ? await workflows.ReplayRunAsync(runId, team.TeamId, team.OwnerId, CancellationToken.None) + : await workflows.RerunFromNodeAsync(runId, "emit", team.TeamId, team.OwnerId, CancellationToken.None); + } + + private async Task SetAgentRunStatusAsync(Guid agentRunId, AgentRunStatus status) + { + using var scope = _fixture.BeginScope(); + await scope.Resolve().AgentRun.Where(run => run.Id == agentRunId).ExecuteUpdateAsync(s => s.SetProperty(run => run.Status, status)); + } + + private async Task UnsealAsAdminAsync() + { + using var scope = _fixture.BeginScopeAs(Guid.NewGuid(), null, Roles.Admin); + return await scope.Resolve().Send(new UnsealNonTerminalPayloadsCommand()); + } + + // ── Definitions + seeding ──────────────────────────────────────────────────────────────────────────────── + + private sealed record Team(Guid TeamId, Guid OwnerId, string Secret); + + private async Task SeedTeamWithSecretAsync() + { + var (teamId, ownerId) = await WorkflowsTestSeed.SeedTeamAsync(_fixture); + var team = new Team(teamId, ownerId, $"CANARY-rolling-{Guid.NewGuid():N}"); + + using var scope = _fixture.BeginScope(); + await scope.Resolve().SetAsync(VariableScope.Team, teamId, teamId, "DEPLOY_TOKEN", VariableValueType.Secret, JsonSerializer.SerializeToElement(team.Secret), null, ownerId, CancellationToken.None); + + return team; + } + + private async Task CreateWorkflowAsync(Team team, WorkflowDefinition definition) + { + using var scope = _fixture.BeginScopeAs(team.OwnerId, team.TeamId, Roles.Admin); + + return await scope.Resolve().Send(new CreateWorkflowCommand { Name = "rolling-" + Guid.NewGuid().ToString("N")[..8], Definition = definition, Activations = [], Enabled = true }); + } + + private static WorkflowDefinition ParentDefinition(Guid childWorkflowId) => new() + { + SchemaVersion = 1, + Nodes = + [ + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "sub", TypeKey = "flow.subworkflow", Config = WorkflowsTestSeed.Json($$"""{ "workflowId": "{{childWorkflowId}}" }"""), Inputs = WorkflowsTestSeed.Json("""{ "inputs": { "token": "{{team.DEPLOY_TOKEN}}" } }""") }, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.Json("""{ "status": "ok" }""") }, + ], + Edges = [new() { From = "start", To = "sub" }, new() { From = "sub", To = "end" }], + }; + + private static WorkflowDefinition ChildDefinition() => new() + { + SchemaVersion = 1, + Nodes = + [ + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "emit", TypeKey = JsonEmitNode.Key, Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.Json("""{ "t": "{{trigger.token}}" }""") }, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.Json("""{ "status": "ok" }""") }, + ], + Edges = [new() { From = "start", To = "emit" }, new() { From = "emit", To = "end" }], + }; + + private static WorkflowDefinition AgentDefinition(string? acceptance = null) => new() + { + SchemaVersion = 1, + Nodes = + [ + new() { Id = "start", TypeKey = "trigger.manual", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.EmptyJson() }, + new() { Id = "agent", TypeKey = "agent.run", Inputs = WorkflowsTestSeed.EmptyJson(), + Config = WorkflowsTestSeed.Json("""{ "goal": "Deploy with bearer {{team.DEPLOY_TOKEN}}", "harness": "codex-cli", "runnerKind": "local", "timeoutSeconds": 300""" + (acceptance is null ? "" : $", \"acceptance\": {acceptance}") + " }") }, + new() { Id = "end", TypeKey = "builtin.terminal", Config = WorkflowsTestSeed.EmptyJson(), Inputs = WorkflowsTestSeed.Json("""{ "summary": "{{nodes.agent.outputs.summary}}" }""") }, + ], + Edges = [new() { From = "start", To = "agent" }, new() { From = "agent", To = "end" }], + }; + + // ── Reading ────────────────────────────────────────────────────────────────────────────────────────────── + + private InMemoryBackgroundJobClient ResolveJobClient() + { + using var scope = _fixture.BeginScope(); + return scope.Resolve(); + } + + private async Task LoadRunAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().WorkflowRun.AsNoTracking().SingleAsync(run => run.Id == runId); + } + + private async Task LoadRunRequestAsync(Guid runId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + var requestId = await db.WorkflowRun.AsNoTracking().Where(run => run.Id == runId).Select(run => run.RunRequestId).SingleAsync(); + + return await db.WorkflowRunRequest.AsNoTracking().SingleAsync(request => request.Id == requestId); + } + + private async Task LoadAgentRunAsync(Guid agentRunId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().AgentRun.AsNoTracking().SingleAsync(run => run.Id == agentRunId); + } + + private async Task LoadAgentRunForRunAsync(Guid workflowRunId) + { + using var scope = _fixture.BeginScope(); + return await scope.Resolve().AgentRun.AsNoTracking().SingleAsync(run => run.WorkflowRunId == workflowRunId); + } + + private async Task LoadChildRunIdAsync(Guid parentRunId) + { + using var scope = _fixture.BeginScope(); + var token = await scope.Resolve().WorkflowRunWait.AsNoTracking().Where(wait => wait.RunId == parentRunId && wait.WaitKind == WorkflowWaitKinds.Subworkflow).Select(wait => wait.Token).SingleAsync(); + + return Guid.Parse(token); + } + + private async Task StakedAcceptanceHashAsync(Guid workflowRunId) + { + using var scope = _fixture.BeginScope(); + var envelope = await scope.Resolve().CompletionRequirement.AsNoTracking().Where(r => r.WorkflowRunId == workflowRunId && r.Kind == ContractKinds.Acceptance).Select(r => r.EnvelopeJson).SingleAsync(); + + return JsonSerializer.Deserialize(envelope, AgentJson.Options)!.SpecHash; + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Hangfire/RecurringJobScheduleKeeperFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Hangfire/RecurringJobScheduleKeeperFlowTests.cs new file mode 100644 index 000000000..7644cba38 --- /dev/null +++ b/backend/tests/CodeSpace.IntegrationTests/Hangfire/RecurringJobScheduleKeeperFlowTests.cs @@ -0,0 +1,144 @@ +using Autofac; +using Autofac.Extensions.DependencyInjection; +using CodeSpace.Api.Extensions.Hangfire; +using CodeSpace.Core.Jobs.RecurringJobs; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Core.Services.Jobs; +using CodeSpace.IntegrationTests.Fleet; +using CodeSpace.IntegrationTests.Infrastructure; +using Hangfire; +using Hangfire.Storage; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Npgsql; +using Shouldly; + +namespace CodeSpace.IntegrationTests.Hangfire; + +/// +/// The recurring entries a previous-version worker's scheduler disables during a rolling deploy — it cannot load a job +/// type it never had, retries, then leaves the entry with no next execution — come back on the keeper's next pass, and +/// the worker's advertised queues carry the fleet capability markers. Real Hangfire storage on Postgres (a schema of +/// its own), the production CodeSpaceBackgroundJobClient, the production RecurringJobSchedule over the +/// fixture container's real recurring jobs; the disabled state is written with Hangfire's own write transaction, +/// field for field what RecurringJobScheduler leaves (observed against the previous build's scheduler: Error, +/// RetryAttempt 5, an empty NextExecution and score -1). +/// +[Collection(PostgresCollection.Name)] +[Trait("Category", "Integration")] +public sealed class RecurringJobScheduleKeeperFlowTests +{ + private const string LoadFailure = "Hangfire.Common.JobLoadException: Could not load the job. See inner exception for the details. ---> Newtonsoft.Json.JsonSerializationException: Error converting value \"CodeSpace.Core.Jobs.RecurringJobs.DepartedMemberCredentialRetirementRecurringJob, CodeSpace.Core\" to type 'System.Type'."; + + private static readonly string[] AddedSinceThePreviousBuild = [nameof(DepartedMemberCredentialRetirementRecurringJob), nameof(DisconnectedCredentialBindingRepairRecurringJob), nameof(WebhookClaimPurgeRecurringJob), nameof(PackCloneUrlBackfillRecurringJob)]; + + private readonly PostgresFixture _fixture; + + public RecurringJobScheduleKeeperFlowTests(PostgresFixture fixture) { _fixture = fixture; } + + [Fact] + public void An_entry_a_previous_version_scheduler_disabled_is_rescheduled_by_the_next_pass() + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + using var scope = BeginScopeOnRealStorage(fleet); + RecurringJobSchedule.Assert(new AutofacServiceProvider(scope)); + + DisableAsAPreviousVersionSchedulerDoes(fleet.Storage, AddedSinceThePreviousBuild); + EntriesNamed(fleet.Storage, AddedSinceThePreviousBuild).ShouldAllBe(entry => entry.NextExecution == null, "precondition: the entries are disabled"); + + Keeper(scope).Reassert(); + + foreach (var entry in EntriesNamed(fleet.Storage, AddedSinceThePreviousBuild)) + { + entry.NextExecution.ShouldNotBeNull($"{entry.Id} must be scheduled again — a disabled entry never fires, and these are the sweeps that clean up after previous-version pods"); + entry.Error.ShouldBeNullOrEmpty(); + entry.RetryAttempt.ShouldBe(0); + ScheduleScore(fleet, entry.Id).ShouldBeGreaterThan(0, $"{entry.Id} must be back in the scheduler's due set, not parked at -1"); + } + } + + [Fact] + public void A_pass_over_a_healthy_schedule_changes_no_entry() + { + using var fleet = new HangfireFleetHarness(_fixture.ConnectionString); + using var scope = BeginScopeOnRealStorage(fleet); + RecurringJobSchedule.Assert(new AutofacServiceProvider(scope)); + var before = AllEntries(fleet.Storage).ToDictionary(entry => entry.Id, entry => (entry.NextExecution, entry.Cron, entry.Queue)); + + Keeper(scope).Reassert(); + + AllEntries(fleet.Storage).ToDictionary(entry => entry.Id, entry => (entry.NextExecution, entry.Cron, entry.Queue)).ShouldBe(before); + } + + [Fact] + public void The_pass_interval_is_pinned() => + // Each pass takes one Hangfire lock per recurring entry on every worker; five minutes keeps that negligible while + // a disabled sweep — the departed-member token sweep runs every five minutes itself — misses at most one tick. + RecurringJobScheduleKeeper.Interval.ShouldBe(TimeSpan.FromMinutes(5)); + + [Fact] + public void Both_worker_pools_advertise_every_capability_marker_after_their_own_queue() + { + // A pool without the marker reads to every peer as a previous-version worker, and sealing would never resume. + WorkerHangfireRegistrar.ControlQueues.ShouldBe([Core.Constants.HangfireConstants.DefaultQueue, .. FleetCapabilityMarkers.Advertised]); + WorkerHangfireRegistrar.AgentQueues.ShouldBe([Core.Constants.HangfireConstants.AgentQueue, .. FleetCapabilityMarkers.Advertised]); + } + + /// The keeper as a worker host builds it, with its scopes opened under the test's storage override — the container's own scope factory would open them from the fixture root, whose job client is the in-memory one. + private static RecurringJobScheduleKeeper Keeper(ILifetimeScope scope) => new(new ChildScopeFactory(scope), TimeProvider.System, NullLogger.Instance); + + private ILifetimeScope BeginScopeOnRealStorage(HangfireFleetHarness fleet) => + fleet.BeginScope(_fixture, builder => builder.RegisterInstance(new CodeSpaceBackgroundJobClient(() => new BackgroundJobClient(fleet.Storage), () => new RecurringJobManager(fleet.Storage))).As()); + + /// The writes RecurringJobScheduler makes when it gives up on an entry it cannot load: Disable(error) through UpdateRecurringJob. + private static void DisableAsAPreviousVersionSchedulerDoes(JobStorage storage, IEnumerable recurringJobIds) + { + using var connection = storage.GetConnection(); + using var transaction = connection.CreateWriteTransaction(); + + foreach (var id in recurringJobIds) + { + transaction.SetRangeInHash($"recurring-job:{id}", new Dictionary { ["NextExecution"] = string.Empty, ["Error"] = LoadFailure, ["RetryAttempt"] = "5" }); + transaction.AddToSet("recurring-jobs", id, -1.0); + } + + transaction.Commit(); + } + + private static IReadOnlyList AllEntries(JobStorage storage) + { + using var connection = storage.GetConnection(); + return connection.GetRecurringJobs(); + } + + private static IReadOnlyList EntriesNamed(JobStorage storage, IReadOnlyCollection ids) + { + var entries = AllEntries(storage).Where(entry => ids.Contains(entry.Id)).ToList(); + entries.Count.ShouldBe(ids.Count, "every job added since the previous build must be in this build's schedule"); + + return entries; + } + + private double ScheduleScore(HangfireFleetHarness fleet, string recurringJobId) + { + using var connection = new NpgsqlConnection(_fixture.ConnectionString); + connection.Open(); + + using var command = new NpgsqlCommand($"SELECT score FROM \"{fleet.Schema}\".set WHERE key = 'recurring-jobs' AND value = @id", connection); + command.Parameters.AddWithValue("id", recurringJobId); + + return Convert.ToDouble(command.ExecuteScalar()); + } + + private sealed class ChildScopeFactory(ILifetimeScope parent) : IServiceScopeFactory + { + public IServiceScope CreateScope() => new ChildScope(parent.BeginLifetimeScope()); + } + + private sealed class ChildScope(ILifetimeScope scope) : IServiceScope + { + public IServiceProvider ServiceProvider { get; } = new AutofacServiceProvider(scope); + + public void Dispose() => scope.Dispose(); + } +} diff --git a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs index 0755354c6..05c528590 100644 --- a/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs +++ b/backend/tests/CodeSpace.IntegrationTests/Workflows/AgentRunReattachFlowTests.cs @@ -12,6 +12,8 @@ using CodeSpace.Core.Services.Agents.Sandbox.Runners; using CodeSpace.Core.Services.Agents.Workspace; using CodeSpace.Core.Services.Credentials; +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Core.Services.Workflows.Runtime; using CodeSpace.IntegrationTests.Infrastructure; using CodeSpace.IntegrationTests.Infrastructure.Jobs; using CodeSpace.Messages.Agents; @@ -1064,6 +1066,75 @@ public async Task Revived_original_observer_cannot_append_or_kill_successors_pro } } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task A_sealed_auto_run_keeps_its_sealed_task_and_a_current_worker_re_attaches_its_tail_whatever_the_fleet_gate_said_at_launch(bool everyLiveWorkerReadsSealedPayloadsAtLaunch) + { + if (OperatingSystem.IsWindows()) return; + + // An auto run (no pinned model) re-persists its task with the resolved model at launch. Written through the fleet + // gate while a previous-version worker was live, that write unsealed the Running row: a re-attach then rebuilt + // the launch's redactor without the task's workflow secrets, failed the handle's fingerprint and landed the run + // marker-only, its tail lost. No previous-version worker ever launches a Running row, so nothing gained. + var secret = $"CANARY-reattach-{Guid.NewGuid():N}"; + var teamId = await SeedTeamAsync(); + var credentialId = await SeedModelCredentialAsync(teamId, "scripted-provider", "sk-reattach-gate-5e81d0"); + await SeedDefaultModelAsync(credentialId, "credential-default-model"); + var runId = await CreateSealedAutoRunAsync(teamId, credentialId, secret); + var gate = Path.Combine(NewSpoolDir(), "release"); + var harness = new ProjectingHarness("scripted-provider", "SCRIPTED_MODEL_KEY", script: $"echo started; while [ ! -f '{gate}' ]; do sleep 0.1; done; echo recovered"); + using var originalScope = _fixture.BeginScope(); + var reached = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var resume = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var instrumented = new InstrumentedAgentRunService(originalScope.Resolve()) { BeforeOwnedAppendAsync = async _ => { reached.TrySetResult(); await resume.Task; } }; + using var bounded = new CancellationTokenSource(TimeSpan.FromSeconds(60)); + var original = BuildExecutor(originalScope, harness, runs: instrumented, sealer: GatedSealer(originalScope, everyLiveWorkerReadsSealedPayloadsAtLaunch)).ExecuteAsync(runId, bounded.Token); + Task? successor = null; + try + { + await reached.Task.WaitAsync(bounded.Token); + using var reclaim = _fixture.BeginScope(); + var runs = reclaim.Resolve(); + var launched = await runs.GetAsync(runId, bounded.Token); + var handle = JsonSerializer.Deserialize(launched.RunnerHandleJson!, AgentJson.Options)!; + _pidsToKill.Add(handle.ProcessId); + _spoolDirs.Add(handle.SpoolDirectory); + + launched.SealedTaskCiphertext.ShouldNotBeNull("the launch re-persisted the resolved model onto a row that was sealed, so it stays sealed"); + launched.TaskJson.ShouldNotContain(secret); + JsonSerializer.Deserialize(launched.TaskJson, AgentJson.Options)!.Model.ShouldBe("credential-default-model", "precondition: the auto run re-persisted its resolved model"); + + await LapseLeaseAsync(runId); + _reservations[runId] = (await runs.ReserveReattachAsync(runId, bounded.Token))!; + successor = ReattachAsync(runId, harness, sealer: GatedSealer(reclaim, everyLiveWorkerReadsSealedPayloads: true)); + AgentRun current; + do + { + current = await runs.GetAsync(runId, bounded.Token); + if (current.OwnerId is null || current.OwnerId == launched.OwnerId) await Task.Delay(20, bounded.Token); + } while (current.OwnerId is null || current.OwnerId == launched.OwnerId); + resume.SetResult(); + await Should.ThrowAsync(() => original); + await File.WriteAllTextAsync(gate, "release", bounded.Token); + await successor.WaitAsync(bounded.Token); + + var completed = await runs.GetAsync(runId, bounded.Token); + completed.Status.ShouldBe(AgentRunStatus.Succeeded, "a re-attach that cannot rebuild the launch's redactor (the handle's fingerprint differs) does not read the tail; a later sweep lands the run marker-only"); + completed.ResultJson.ShouldNotBeNull().ShouldNotContain("reattach-marker-only"); + var events = await reclaim.Resolve().AgentRunEvent.AsNoTracking().Where(e => e.AgentRunId == runId).ToListAsync(bounded.Token); + events.Count(e => e.Text == "recovered").ShouldBe(1, "the line the agent wrote while no worker observed it is captured exactly once"); + } + finally + { + resume.TrySetResult(); + await File.WriteAllTextAsync(gate, "release"); + bounded.Cancel(); + try { await original; } catch { } + if (successor != null) try { await successor.WaitAsync(TimeSpan.FromSeconds(5)); } catch { } + } + } + // ── helpers ───────────────────────────────────────────────────────────────────────────────────────── private async Task LaunchAliveSupervisorAsync(Guid runId) @@ -1087,13 +1158,13 @@ await scope.Resolve().Database .ExecuteSqlInterpolatedAsync($"UPDATE agent_run SET lease_expires_at = {DateTimeOffset.UtcNow.AddMinutes(-1)} WHERE id = {runId}"); } - private async Task ReattachAsync(Guid runId, IAgentHarness harness, IAgentRunLogCaptureBridge? logCapture = null, CodeSpace.Core.Services.Agents.Credentials.IModelCredentialBroker? credentialBroker = null) + private async Task ReattachAsync(Guid runId, IAgentHarness harness, IAgentRunLogCaptureBridge? logCapture = null, CodeSpace.Core.Services.Agents.Credentials.IModelCredentialBroker? credentialBroker = null, WorkflowSensitivePayloadSealer? sealer = null) { using var scope = _fixture.BeginScope(); - await BuildExecutor(scope, harness, logCapture, credentialBroker: credentialBroker).ReattachAsync(_reservations[runId], CancellationToken.None); + await BuildExecutor(scope, harness, logCapture, credentialBroker: credentialBroker, sealer: sealer).ReattachAsync(_reservations[runId], CancellationToken.None); } - private static AgentRunExecutor BuildExecutor(ILifetimeScope scope, IAgentHarness harness, IAgentRunLogCaptureBridge? logCapture = null, IAgentRunService? runs = null, CodeSpace.Core.Services.Agents.Credentials.IModelCredentialBroker? credentialBroker = null) + private static AgentRunExecutor BuildExecutor(ILifetimeScope scope, IAgentHarness harness, IAgentRunLogCaptureBridge? logCapture = null, IAgentRunService? runs = null, CodeSpace.Core.Services.Agents.Credentials.IModelCredentialBroker? credentialBroker = null, WorkflowSensitivePayloadSealer? sealer = null) { return new AgentRunExecutor( runs ?? scope.Resolve(), @@ -1114,7 +1185,16 @@ private static AgentRunExecutor BuildExecutor(ILifetimeScope scope, IAgentHarnes NullLogger.Instance, logCapture, credentialBroker: credentialBroker, - logs: scope.Resolve()); + logs: scope.Resolve(), + sealer: sealer); + } + + /// The production sealer over the fixture's Data Protection keys, with the fleet gate's answer fixed — the one a worker would read off a fleet with (false) or without (true) a previous-version worker in it. + private static WorkflowSensitivePayloadSealer GatedSealer(ILifetimeScope scope, bool everyLiveWorkerReadsSealedPayloads) => new(scope.Resolve(), new FixedFleetGate(everyLiveWorkerReadsSealedPayloads)); + + private sealed class FixedFleetGate(bool supported) : IFleetCapabilities + { + public bool AllWorkersSupport(string capability) => supported; } private sealed class RecordingLogCaptureBridge : IAgentRunLogCaptureBridge @@ -1169,6 +1249,23 @@ private async Task CreateRunWithCredentialAsync(Guid teamId, Guid modelCre return run.Id; } + /// An auto run (no pinned model) whose goal embeds a workflow secret, staged the way a workflow's agent.run stages it: named, so the store seals it. + private async Task CreateSealedAutoRunAsync(Guid teamId, Guid modelCredentialId, string secret) + { + using var scope = await WorkflowsTestSeed.BeginSeedOperatorScopeAsync(_fixture, teamId); + var task = new AgentTask { Goal = $"Deploy with {secret}", Harness = "scripted-projector", Model = null, ModelCredentialId = modelCredentialId, TimeoutSeconds = 1800, WorkflowSecrets = [secret] }; + + return (await scope.Resolve().CreateAsync(task, teamId, null, null, iterationKey: "", cancellationToken: CancellationToken.None)).Id; + } + + private async Task SeedDefaultModelAsync(Guid modelCredentialId, string modelId) + { + using var scope = _fixture.BeginScope(); + var db = scope.Resolve(); + db.ModelCredentialModel.Add(new ModelCredentialModel { Id = Guid.NewGuid(), ModelCredentialId = modelCredentialId, ModelId = modelId, IsDefault = true, Enabled = true }); + await db.SaveChangesAsync(); + } + private async Task SeedModelCredentialAsync(Guid teamId, string provider, string plaintextKey) { using var scope = _fixture.BeginScope(); @@ -1242,20 +1339,23 @@ private sealed class ProjectingHarness : IAgentHarness, IModelCredentialProjecto private readonly string _provider; private readonly string _envVar; private readonly int _maxLineChars; + private readonly string _script; /// A longer line becomes no event, as a production CLI parser drops a line it cannot parse. - public ProjectingHarness(string provider, string envVar, int maxLineChars = int.MaxValue) + /// What the launched agent runs — only a test that launches through this harness needs more than true. + public ProjectingHarness(string provider, string envVar, int maxLineChars = int.MaxValue, string script = "true") { _provider = provider; _envVar = envVar; _maxLineChars = maxLineChars; + _script = script; } public string Kind => "scripted-projector"; public string Version => "test"; public IReadOnlyList Models { get; } = new[] { "test-model" }; - public SandboxSpec BuildInvocation(AgentTask task) => new() { Command = "/bin/sh", Args = new[] { "-c", "true" }, Environment = task.Environment, TimeoutSeconds = task.TimeoutSeconds }; + public SandboxSpec BuildInvocation(AgentTask task) => new() { Command = "/bin/sh", Args = new[] { "-c", _script }, Environment = task.Environment, TimeoutSeconds = task.TimeoutSeconds }; public IReadOnlyList ParseEvents(string rawLine) { diff --git a/backend/tests/CodeSpace.UnitTests/Agents/AgentTaskSecretsTests.cs b/backend/tests/CodeSpace.UnitTests/Agents/AgentTaskSecretsTests.cs index 493bd111e..fb83f0af4 100644 --- a/backend/tests/CodeSpace.UnitTests/Agents/AgentTaskSecretsTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Agents/AgentTaskSecretsTests.cs @@ -3,6 +3,7 @@ using CodeSpace.Core.Services.Agents; using CodeSpace.Core.Services.Workflows.Runtime; using CodeSpace.Messages.Agents; +using CodeSpace.UnitTests.Fleet; using Microsoft.AspNetCore.DataProtection; using Shouldly; @@ -18,7 +19,7 @@ public sealed class AgentTaskSecretsTests { private const string Secret = "CANARY-goal-7d2e91b4c0"; - private static readonly WorkflowSensitivePayloadSealer Sealer = new(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider())); + private static readonly WorkflowSensitivePayloadSealer Sealer = new(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), FixedFleetCapabilities.AllCurrent); private static AgentTask Task(string goal) => new() { Goal = goal, Harness = "codex-cli" }; @@ -88,4 +89,45 @@ public void A_sealed_row_cannot_be_opened_without_a_sealer_rather_than_running_t Should.Throw(() => AgentTaskSecrets.Open(run, sealer: null)); } + + [Fact] + public void While_a_previous_version_worker_is_live_a_secret_bearing_task_persists_as_the_previous_build_wrote_it() + { + var task = Task($"Call the deploy API with bearer token {Secret}") with { WorkflowSecrets = [Secret] }; + var sealer = new WorkflowSensitivePayloadSealer(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), FixedFleetCapabilities.WithPreviousVersionWorker); + + var written = AgentTaskSecrets.Seal(task, sealer); + + // The previous build's AgentRunService wrote exactly this expression into task_jsonb and its executor + // deserialized task_jsonb to launch; WorkflowSecrets is [JsonIgnore], so the list never reaches the column. + written.PublicJson.ShouldBe(CodeSpace.Core.Persistence.PersistedText.SanitizeJson(JsonSerializer.Serialize(task, AgentJson.Options))); + written.Ciphertext.ShouldBeNull(); + JsonSerializer.Deserialize(written.PublicJson, AgentJson.Options)!.Goal.ShouldBe(task.Goal, "the previous build launches the exact goal, never the mask"); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void A_task_a_current_worker_rewrites_onto_its_own_row_keeps_the_shape_it_was_opened_from(bool openedFromSealedRow) + { + // The executor re-persists the opened task with its resolved model while the run is Running. A previous-version + // worker never launches a Running row, so the fleet gate has nothing to protect there; the row's shape is what + // the launch's redactor fingerprint was built from, and a re-attach rebuilds it from that shape. + var protector = new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()); + var stagedWhileAllCurrent = new WorkflowSensitivePayloadSealer(protector, FixedFleetCapabilities.AllCurrent); + var previousVersionWorkerLive = new WorkflowSensitivePayloadSealer(protector, FixedFleetCapabilities.WithPreviousVersionWorker); + var staged = AgentTaskSecrets.Seal(Task($"Deploy with {Secret}") with { WorkflowSecrets = openedFromSealedRow ? [Secret] : null }, stagedWhileAllCurrent); + var row = new AgentRun { Id = Guid.NewGuid(), TaskJson = staged.PublicJson, SealedTaskCiphertext = staged.Ciphertext, Harness = "codex-cli" }; + var opened = AgentTaskSecrets.Open(row, previousVersionWorkerLive)!; + + var written = AgentTaskSecrets.Reseal(opened with { Model = "resolved-model" }, previousVersionWorkerLive); + + (written.Ciphertext is not null).ShouldBe(openedFromSealedRow); + written.PublicJson.Contains(Secret).ShouldBe(!openedFromSealedRow); + var reopened = AgentTaskSecrets.Open(new AgentRun { Id = row.Id, TaskJson = written.PublicJson, SealedTaskCiphertext = written.Ciphertext, Harness = "codex-cli" }, previousVersionWorkerLive)!; + reopened.Model.ShouldBe("resolved-model"); + reopened.Goal.ShouldBe($"Deploy with {Secret}"); + reopened.WorkflowSecrets.ShouldBe(opened.WorkflowSecrets, "a re-attach rebuilds the launch's redactor from exactly these"); + } } + diff --git a/backend/tests/CodeSpace.UnitTests/Agents/SupervisorBranchlessStopGradeTests.cs b/backend/tests/CodeSpace.UnitTests/Agents/SupervisorBranchlessStopGradeTests.cs index ebaccd67a..997976c64 100644 --- a/backend/tests/CodeSpace.UnitTests/Agents/SupervisorBranchlessStopGradeTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Agents/SupervisorBranchlessStopGradeTests.cs @@ -251,7 +251,7 @@ public async Task A_branchless_gate_masks_the_units_sealed_secrets_and_the_scope // artifact — so the request must name both, and nothing else of the scope. // MUTATIONS this pins: drop WorkflowSecrets from the branchless lane's request; read the unit's secrets without // opening its sealed task; or read the spec's without the ambient scope → one of the two canaries goes missing → red. - var sealer = new CodeSpace.Core.Services.Workflows.Runtime.WorkflowSensitivePayloadSealer(new CodeSpace.Core.Services.Workflows.Runtime.DataProtectionWorkflowSensitivePayloadProtector(new Microsoft.AspNetCore.DataProtection.EphemeralDataProtectionProvider())); + var sealer = new CodeSpace.Core.Services.Workflows.Runtime.WorkflowSensitivePayloadSealer(new CodeSpace.Core.Services.Workflows.Runtime.DataProtectionWorkflowSensitivePayloadProtector(new Microsoft.AspNetCore.DataProtection.EphemeralDataProtectionProvider()), CodeSpace.UnitTests.Fleet.FixedFleetCapabilities.AllCurrent); var db = Infrastructure.EmptyTestDb.New(); var task = new AgentTask { Goal = $"write the report, deploy key {UnitSecret}", Harness = "codex-cli", WorkflowSecrets = [UnitSecret] }; var sealedTask = AgentTaskSecrets.Seal(task, sealer); diff --git a/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs b/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs index dfa0b3a72..88ff62660 100644 --- a/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Architecture/ScopedTransactionInventoryTests.cs @@ -81,6 +81,7 @@ public class ScopedTransactionInventoryTests ["Workflows/Budget/BudgetLedger.PhysicalLlm.cs"] = (2, "both physical-POST sites: the receipt authorizing a POST, and the receipt recording what that POST actually cost, must be committed independently of a caller that may roll back."), ["Workflows/Engine/WorkflowEngine.cs"] = (5, "three terminal writes that must land whatever their caller decides (two under CancellationToken.None), a redaction fallback whose fresh-scope retry depends on this transaction's discard, and a step's park, whose share lock on the run row is what a Continue's bump waits on and so must end at the park's own commit."), ["Workflows/ModelCalls/WorkflowRunModelCallBodyMaterializer.cs"] = (2, "the same private CreateDb() context; nothing ambient can reach it."), + ["Workflows/Runtime/SealedPayloadUnsealService.cs"] = (1, "the pre-rollback unseal is a non-transactional sweep: each restored agent task commits together with its unit's re-stake and on its own, so a raced or unopenable row never undoes another; joined, one transaction over the batch would."), }; [Fact] diff --git a/backend/tests/CodeSpace.UnitTests/Fleet/FixedFleetCapabilities.cs b/backend/tests/CodeSpace.UnitTests/Fleet/FixedFleetCapabilities.cs new file mode 100644 index 000000000..7b31cf716 --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Fleet/FixedFleetCapabilities.cs @@ -0,0 +1,19 @@ +using CodeSpace.Core.Services.Fleet; + +namespace CodeSpace.UnitTests.Fleet; + +/// A fleet whose answer the test fixes: true is a fleet of current workers, false one with a previous-version worker still live. Counts the asks so a test can prove the gate is consulted only when something would be sealed. +public sealed class FixedFleetCapabilities(bool supported) : IFleetCapabilities +{ + public static FixedFleetCapabilities AllCurrent => new(true); + + public static FixedFleetCapabilities WithPreviousVersionWorker => new(false); + + public int Asks { get; private set; } + + public bool AllWorkersSupport(string capability) + { + Asks++; + return supported; + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilitiesTests.cs b/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilitiesTests.cs new file mode 100644 index 000000000..c6330021f --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilitiesTests.cs @@ -0,0 +1,107 @@ +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Messages.Fleet; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Time.Testing; +using Shouldly; + +namespace CodeSpace.UnitTests.Fleet; + +/// +/// The gate every sealing write asks: read the fleet at most once per snapshot lifetime (a seal is a hot path, the +/// registry is a database), see a fleet change once the snapshot ages out, and answer with the shape every worker can +/// read when the fleet cannot be read at all. +/// +[Trait("Category", "Unit")] +public sealed class FleetCapabilitiesTests +{ + private const string Marker = FleetCapabilityMarkers.SealedPayloadV1; + + [Fact] + public void The_fleet_is_read_once_per_snapshot_lifetime() + { + var clock = new FakeTimeProvider(new DateTimeOffset(2026, 10, 13, 12, 0, 0, TimeSpan.Zero)); + var registry = new ScriptedRegistry(clock, () => [Live("new", Marker)]); + var fleet = new FleetCapabilities(registry, clock, NullLogger.Instance); + + fleet.AllWorkersSupport(Marker).ShouldBeTrue(); + clock.Advance(FleetCapabilities.SnapshotLifetime - TimeSpan.FromSeconds(1)); + fleet.AllWorkersSupport(Marker).ShouldBeTrue(); + + registry.Reads.ShouldBe(1, "a second ask inside the snapshot lifetime must not hit the registry again"); + + clock.Advance(TimeSpan.FromSeconds(1)); + fleet.AllWorkersSupport(Marker).ShouldBeTrue(); + + registry.Reads.ShouldBe(2, "an aged-out snapshot is re-read"); + } + + [Fact] + public void A_previous_version_worker_holds_the_capability_back_until_it_is_gone_and_the_snapshot_ages_out() + { + var clock = new FakeTimeProvider(new DateTimeOffset(2026, 10, 13, 12, 0, 0, TimeSpan.Zero)); + var oldWorkerAlive = true; + var registry = new ScriptedRegistry(clock, () => oldWorkerAlive ? [Live("new", Marker), Live("old", "default")] : [Live("new", Marker)]); + var fleet = new FleetCapabilities(registry, clock, NullLogger.Instance); + + fleet.AllWorkersSupport(Marker).ShouldBeFalse("a live previous-version worker reads only the legacy shape"); + + oldWorkerAlive = false; + fleet.AllWorkersSupport(Marker).ShouldBeFalse("the cached snapshot still holds the old worker — the answer may lag by one lifetime, never lead"); + + clock.Advance(FleetCapabilities.SnapshotLifetime); + fleet.AllWorkersSupport(Marker).ShouldBeTrue(); + } + + [Fact] + public void An_unreadable_fleet_answers_with_the_shape_every_worker_reads() + { + var clock = new FakeTimeProvider(new DateTimeOffset(2026, 10, 13, 12, 0, 0, TimeSpan.Zero)); + var registry = new ScriptedRegistry(clock, () => throw new InvalidOperationException("storage down")); + var fleet = new FleetCapabilities(registry, clock, NullLogger.Instance); + + Should.NotThrow(() => fleet.AllWorkersSupport(Marker)).ShouldBeFalse( + "a gate that cannot see the fleet cannot prove no previous-version worker is in it; the caller must still be able to write"); + } + + [Fact] + public void A_fleet_with_no_live_worker_holds_every_capability_back() + { + // The API pod asks too (a replay or rerun fork). With every current worker stopped for a rollback, the next worker + // to join may be a previous-version one, and the fork it picks up must be the shape it reads. + var clock = new FakeTimeProvider(new DateTimeOffset(2026, 10, 13, 12, 0, 0, TimeSpan.Zero)); + var fleet = new FleetCapabilities(new ScriptedRegistry(clock, () => []), clock, NullLogger.Instance); + + fleet.AllWorkersSupport(Marker).ShouldBeFalse(); + } + + [Fact] + public void A_host_with_no_job_storage_observes_no_fleet_and_holds_nothing_back() + { + // Every deployed role registers job storage; a host without it runs no job and shares its rows with no worker. + var clock = new FakeTimeProvider(new DateTimeOffset(2026, 10, 13, 12, 0, 0, TimeSpan.Zero)); + var registry = new ScriptedRegistry(clock, () => [Live("old", "default")]) { ObservesFleet = false }; + var fleet = new FleetCapabilities(registry, clock, NullLogger.Instance); + + fleet.AllWorkersSupport(Marker).ShouldBeTrue(); + registry.Reads.ShouldBe(0); + } + + [Fact] + public void The_hangfire_registry_observes_a_fleet_only_over_job_storage() => new HangfireWorkerFleetRegistry().ObservesFleet.ShouldBeFalse(); + + private static Func Live(string id, params string[] queues) => now => new WorkerServerHeartbeat(id, queues, now.UtcDateTime); + + /// A registry whose servers heartbeat "now" by the test clock, so only presence decides liveness here; staleness is FleetCapabilityRulesTests' job. + private sealed class ScriptedRegistry(TimeProvider clock, Func>> servers) : IWorkerFleetRegistry + { + public int Reads { get; private set; } + + public bool ObservesFleet { get; init; } = true; + + public IReadOnlyList ReadServers() + { + Reads++; + return servers().Select(server => server(clock.GetUtcNow())).ToList(); + } + } +} diff --git a/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilityRulesTests.cs b/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilityRulesTests.cs new file mode 100644 index 000000000..4361448f6 --- /dev/null +++ b/backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilityRulesTests.cs @@ -0,0 +1,80 @@ +using CodeSpace.Core.Services.Fleet; +using CodeSpace.Messages.Fleet; +using Shouldly; + +namespace CodeSpace.UnitTests.Fleet; + +/// +/// The fleet gate's one decision: a capability is usable only when there is a live job server and EVERY live one +/// advertises it. A server that stopped heartbeating is gone (Hangfire's own watchdog may not have removed its row yet), +/// a server whose heartbeat was never recorded is assumed alive, and a fleet with no live server holds every capability +/// back: the next worker to join may be of either build — the gap a worker rollback opens between stopping this +/// build's workers and starting the previous build's, while the API still stages forks. +/// +[Trait("Category", "Unit")] +public sealed class FleetCapabilityRulesTests +{ + private const string Marker = FleetCapabilityMarkers.SealedPayloadV1; + private static readonly DateTime Now = new(2026, 10, 13, 12, 0, 0, DateTimeKind.Utc); + + [Theory] + [InlineData("all-new", true)] + [InlineData("mixed", false)] + [InlineData("old-only", false)] + [InlineData("old-stale-new-live", true)] + [InlineData("old-live-new-stale", false)] + [InlineData("old-at-window-edge", false)] + [InlineData("old-just-past-window", true)] + [InlineData("old-never-heartbeat", false)] + [InlineData("only-stale", false)] + [InlineData("empty", false)] + [InlineData("newer-marker-only", false)] + [InlineData("marker-case-differs", false)] + public void A_capability_holds_only_when_every_live_server_advertises_it(string fleet, bool expected) => + FleetCapabilityRules.EveryLiveServerAdvertises(Fleet(fleet), Marker, Now).ShouldBe(expected); + + [Fact] + public void The_live_servers_lacking_a_capability_are_named() + { + var servers = new[] + { + Server("old-live", Now.AddSeconds(-10), "default"), + Server("old-stale", Now.AddMinutes(-10), "default"), + Server("new-live", Now.AddSeconds(-5), "default", Marker), + }; + + FleetCapabilityRules.LiveServersLacking(servers, Marker, Now).ShouldBe(["old-live"]); + } + + [Fact] + public void The_marker_and_the_liveness_window_are_pinned() + { + // The marker is compared across builds: a renamed marker makes every new worker look like a previous-version one + // to its peers, and sealing never resumes. A new payload shape is a NEW marker, never an edited one. + FleetCapabilityMarkers.SealedPayloadV1.ShouldBe("cap-sealed-payload-v1"); + FleetCapabilityMarkers.Advertised.ShouldContain(FleetCapabilityMarkers.SealedPayloadV1); + + // Four Hangfire heartbeats (30s each): long enough that a busy server is never mistaken for a gone one, short + // enough that a killed previous-version worker stops holding sealing back within minutes. + FleetCapabilityRules.LiveHeartbeatWindow.ShouldBe(TimeSpan.FromMinutes(2)); + } + + private static IReadOnlyList Fleet(string name) => name switch + { + "all-new" => [Server("new-a", Now.AddSeconds(-5), "default", Marker), Server("new-b", Now.AddSeconds(-20), "agents", Marker)], + "mixed" => [Server("new", Now.AddSeconds(-5), "default", Marker), Server("old", Now.AddSeconds(-5), "default")], + "old-only" => [Server("old", Now.AddSeconds(-5), "default")], + "old-stale-new-live" => [Server("new", Now.AddSeconds(-5), "default", Marker), Server("old", Now.AddMinutes(-3), "default")], + "old-live-new-stale" => [Server("new", Now.AddMinutes(-3), "default", Marker), Server("old", Now.AddSeconds(-5), "default")], + "old-at-window-edge" => [Server("new", Now, "default", Marker), Server("old", Now - FleetCapabilityRules.LiveHeartbeatWindow, "default")], + "old-just-past-window" => [Server("new", Now, "default", Marker), Server("old", Now - FleetCapabilityRules.LiveHeartbeatWindow - TimeSpan.FromSeconds(1), "default")], + "old-never-heartbeat" => [Server("new", Now, "default", Marker), Server("old", null, "default")], + "only-stale" => [Server("old", Now.AddMinutes(-30), "default")], + "empty" => [], + "newer-marker-only" => [Server("future", Now, "default", "cap-sealed-payload-v2")], + "marker-case-differs" => [Server("odd", Now, "default", Marker.ToUpperInvariant())], + _ => throw new ArgumentOutOfRangeException(nameof(name), name, null), + }; + + private static WorkerServerHeartbeat Server(string id, DateTime? heartbeat, params string[] queues) => new(id, queues, heartbeat); +} diff --git a/backend/tests/CodeSpace.UnitTests/Workflows/WorkflowSensitivePayloadSealerTests.cs b/backend/tests/CodeSpace.UnitTests/Workflows/WorkflowSensitivePayloadSealerTests.cs index da7997616..b843c78e0 100644 --- a/backend/tests/CodeSpace.UnitTests/Workflows/WorkflowSensitivePayloadSealerTests.cs +++ b/backend/tests/CodeSpace.UnitTests/Workflows/WorkflowSensitivePayloadSealerTests.cs @@ -1,5 +1,6 @@ using System.Text.Json; using CodeSpace.Core.Services.Workflows.Runtime; +using CodeSpace.UnitTests.Fleet; using Microsoft.AspNetCore.DataProtection; using Shouldly; @@ -8,14 +9,16 @@ namespace CodeSpace.UnitTests.Workflows; /// /// The sealer is the seam that keeps a resolved workflow secret out of a payload's PUBLIC column while the run it is /// handed to still executes with the exact value: the public copy is masked, the exact copy and the values it embeds -/// are encrypted, and a payload carrying no secret is left byte-identical with no ciphertext at all. +/// are encrypted, and a payload carrying no secret is left byte-identical with no ciphertext at all. While a +/// previous-version worker is live — it reads only the public column — a secret-bearing payload is written verbatim and +/// unsealed, the one shape that worker can execute. /// [Trait("Category", "Unit")] public sealed class WorkflowSensitivePayloadSealerTests { private const string Secret = "CANARY-sealer-3f1c9a7e5b2d"; - private static readonly WorkflowSensitivePayloadSealer Sealer = new(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider())); + private static readonly WorkflowSensitivePayloadSealer Sealer = new(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), FixedFleetCapabilities.AllCurrent); private static JsonElement Payload(object value) => JsonSerializer.SerializeToElement(value); @@ -62,9 +65,92 @@ public void An_unsealed_row_opens_from_its_public_column_with_no_secrets() public void A_ciphertext_of_another_kind_under_the_same_purpose_is_refused() { var protector = new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()); - var sealer = new WorkflowSensitivePayloadSealer(protector); + var sealer = new WorkflowSensitivePayloadSealer(protector, FixedFleetCapabilities.AllCurrent); var nodeOutputs = protector.Protect("""{"outputs":{"x":"y"}}"""); Should.Throw(() => sealer.Open("{}", nodeOutputs)); } + + [Fact] + public void While_a_previous_version_worker_is_live_a_secret_bearing_payload_is_written_verbatim_and_unsealed() + { + var fleet = FixedFleetCapabilities.WithPreviousVersionWorker; + var sealer = new WorkflowSensitivePayloadSealer(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), fleet); + var payload = Payload(new { token = Secret, header = $"Bearer {Secret}" }); + + var written = sealer.Seal(payload, new PersistenceSecretRedactor([Secret])); + + written.PublicJson.ShouldBe(payload.GetRawText(), "the previous build executes the public column, so it must hold the exact payload — byte for byte what that build wrote itself"); + written.Ciphertext.ShouldBeNull("a sealed copy beside the exact one could go stale the moment a previous-version worker rewrites the public column"); + fleet.Asks.ShouldBe(1); + } + + [Fact] + public void A_payload_with_nothing_to_seal_never_consults_the_fleet() + { + var fleet = FixedFleetCapabilities.WithPreviousVersionWorker; + var sealer = new WorkflowSensitivePayloadSealer(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), fleet); + + sealer.Seal(Payload(new { token = "public" }), new PersistenceSecretRedactor([Secret])).Ciphertext.ShouldBeNull(); + + fleet.Asks.ShouldBe(0, "the fleet read is for payloads that would be sealed — every other write stays off the registry"); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public void A_copy_of_a_sealed_payload_for_a_new_row_is_sealed_again_through_the_fleet_gate(bool everyLiveWorkerReadsSealedPayloads) + { + // A replay or rerun fork is a new row that any live worker may execute — a previous-version one included. + var protector = new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()); + var payload = Payload(new { token = Secret, other = "kept" }); + var original = new WorkflowSensitivePayloadSealer(protector, FixedFleetCapabilities.AllCurrent).Seal(payload, new PersistenceSecretRedactor([Secret, "never-present-secret"])); + var sealer = new WorkflowSensitivePayloadSealer(protector, new FixedFleetCapabilities(everyLiveWorkerReadsSealedPayloads)); + + var copy = sealer.SealCopy(original.PublicJson, original.Ciphertext); + + if (!everyLiveWorkerReadsSealedPayloads) + { + copy.PublicJson.ShouldBe(payload.GetRawText(), "the previous build builds a fork's trigger scope from the public column alone"); + copy.Ciphertext.ShouldBeNull(); + return; + } + + copy.PublicJson.ShouldBe(original.PublicJson); + copy.Ciphertext.ShouldNotBeNull(); + var opened = sealer.Open(copy.PublicJson, copy.Ciphertext); + opened.Json.ShouldBe(payload.GetRawText()); + opened.Secrets.ShouldBe([Secret]); + } + + [Fact] + public void A_copy_of_an_unsealed_payload_is_carried_over_byte_identical_without_consulting_the_fleet() + { + var fleet = FixedFleetCapabilities.WithPreviousVersionWorker; + var sealer = new WorkflowSensitivePayloadSealer(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), fleet); + const string legacy = """{ "token" : "written-before-sealing" }"""; + + var copy = sealer.SealCopy(legacy, null); + + copy.PublicJson.ShouldBe(legacy, "a fork of an unsealed run carries the original's column exactly as before"); + copy.Ciphertext.ShouldBeNull(); + fleet.Asks.ShouldBe(0); + } + + [Fact] + public void Resealing_a_payload_for_the_row_a_current_worker_already_runs_ignores_the_fleet_gate() + { + // The row is Running on a current worker: no previous-version worker ever launches it, and writing it unsealed + // would drop the secrets its launch folded into the handle's redactor fingerprint. + var fleet = FixedFleetCapabilities.WithPreviousVersionWorker; + var sealer = new WorkflowSensitivePayloadSealer(new DataProtectionWorkflowSensitivePayloadProtector(new EphemeralDataProtectionProvider()), fleet); + var payload = Payload(new { goal = $"Deploy with {Secret}" }); + + var written = sealer.Reseal(payload, new PersistenceSecretRedactor([Secret])); + + written.PublicJson.ShouldNotContain(Secret); + written.Ciphertext.ShouldNotBeNull(); + sealer.Open(written.PublicJson, written.Ciphertext).Secrets.ShouldBe([Secret]); + fleet.Asks.ShouldBe(0); + } }