From 1345beaf01c77279427d8f1f52143731d0329bba Mon Sep 17 00:00:00 2001 From: "Mars.P" Date: Tue, 13 Oct 2026 13:01:43 +0800 Subject: [PATCH] Seal payloads only once every live worker can read them A worker on the build before 0247 executes an agent task or a sub-workflow child from the public column alone. Once that column holds the masked copy, such a worker launches the agent with "[REDACTED]" for the goal's secret and starts the child with "[REDACTED]" inputs. That is what any rolling worker deploy (k8s RollingUpdate included) or any worker rollback runs into, and nothing on the previous build can tell. Every job server of this build now lists a capability marker, cap-sealed-payload-v1, among its Hangfire queues. Nothing is ever enqueued to it; Hangfire's server table carries it to every peer, and a previous build never lists it. The sealer asks the fleet gate before it seals a payload for a new row. While any server with a heartbeat in the last two minutes lacks the marker, or no server is live at all, the payload is written verbatim and unsealed: the row the previous build wrote and the only one it executes correctly. The gate reads the server table at most once per 30 seconds, and an unreadable table answers no. A host with no job storage (a test host) observes no fleet and holds nothing back. A replay or rerun fork is such a new row, staged on the API pod, but it copied the original's ciphertext verbatim. A fork of a sealed child stayed sealed whatever the fleet held, and a previous-version worker ran it with the mask. The fork now opens the original and seals the copy through the gate. Since the API pod asks too, a fleet with no live worker answers no: in a worker rollback's gap between stopping this build's workers and starting the previous ones, the next worker to join may be either build. A row a current worker is already running keeps its shape. The executor re-persists an auto run's resolved model onto its Running row; written through the gate, that unsealed the row, and a later re-attach rebuilt the launch's redactor without the task's workflow secrets, failed the handle's fingerprint and lost the run's tail. No previous-version worker launches a Running row, so that write now seals regardless of the gate. Rows sealed before a previous-version worker joins still need opening, so UnsealNonTerminalPayloadsCommand (POST api/admin/sealed-payloads/unseal, global admin) restores every non-terminal sealed run payload and every queued sealed agent task. A running agent task stays sealed: no worker launches it again, and a current worker's re-attach needs the sealed copy. Each row is a compare-and-swap on its ciphertext. A restored first attempt of a workflow agent with an acceptance contract is re-staked from the restored task, because the composer re-derives the unit's hash from task_jsonb and a masked stake would no longer match. Send it once no current worker is live plus one 30-second gate snapshot, before starting the previous ones. The previous build's scheduler also cannot load the recurring job types this build added. It retries and then disables those entries, and only a current worker's boot re-enabled them, so after a rolling deploy the departed-member token sweep and three others could stay off. A keeper on the worker role now re-runs the boot registration every five minutes. AddOrUpdate reschedules an entry that carries a scheduling error and leaves a healthy one unchanged. Old builds cannot be made to ignore an unknown job type, so healing is the remedy. --- .../src/CodeSpace.Api/CodeSpace.Api.csproj | 2 + .../Controllers/SealedPayloadsController.cs | 26 ++ .../Hangfire/HangfireRegistrar.Worker.cs | 56 ++- .../Hangfire/RecurringJobSchedule.cs | 53 +++ .../Hangfire/RecurringJobScheduleKeeper.cs | 57 +++ ...UnsealNonTerminalPayloadsCommandHandler.cs | 16 + .../Agents/AgentAcceptanceContract.cs | 19 + .../Services/Agents/AgentRunExecutor.cs | 4 +- .../Services/Agents/AgentRunService.cs | 13 +- .../Services/Agents/AgentTaskSecrets.cs | 30 +- .../Services/Fleet/FleetCapabilities.cs | 106 +++++ .../Services/Fleet/FleetCapabilityMarkers.cs | 21 + .../Services/Fleet/FleetCapabilityRules.cs | 28 ++ .../Fleet/HangfireWorkerFleetRegistry.cs | 29 ++ .../Services/Fleet/IFleetCapabilities.cs | 11 + .../Services/Fleet/IWorkerFleetRegistry.cs | 12 + .../Runtime/ISealedPayloadUnsealService.cs | 9 + .../Runtime/SealedPayloadUnsealService.cs | 168 ++++++++ .../Runtime/WorkflowSensitivePayloadSealer.cs | 48 ++- .../Services/Workflows/WorkflowService.cs | 14 +- .../Workflows/WorkflowServiceDependencies.cs | 5 +- .../UnsealNonTerminalPayloadsCommand.cs | 37 ++ .../Fleet/WorkerServerHeartbeat.cs | 8 + .../DeferredDrainFleetRegistry.cs | 21 + .../RecurringJobWorkerHostFactory.cs | 5 +- .../Infrastructure/TaskLaunchApiFactory.cs | 2 + .../Jobs/WorkerFleetCapabilityE2ETests.cs | 167 ++++++++ .../Fleet/HangfireFleetHarness.cs | 91 ++++ .../RollingWorkerPayloadSealingFlowTests.cs | 404 ++++++++++++++++++ .../RecurringJobScheduleKeeperFlowTests.cs | 144 +++++++ .../Workflows/AgentRunReattachFlowTests.cs | 112 ++++- .../Agents/AgentTaskSecretsTests.cs | 44 +- .../SupervisorBranchlessStopGradeTests.cs | 2 +- .../ScopedTransactionInventoryTests.cs | 1 + .../Fleet/FixedFleetCapabilities.cs | 19 + .../Fleet/FleetCapabilitiesTests.cs | 107 +++++ .../Fleet/FleetCapabilityRulesTests.cs | 80 ++++ .../WorkflowSensitivePayloadSealerTests.cs | 92 +++- 38 files changed, 1983 insertions(+), 80 deletions(-) create mode 100644 backend/src/CodeSpace.Api/Controllers/SealedPayloadsController.cs create mode 100644 backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobSchedule.cs create mode 100644 backend/src/CodeSpace.Api/Extensions/Hangfire/RecurringJobScheduleKeeper.cs create mode 100644 backend/src/CodeSpace.Core/Handlers/CommandHandlers/Workflows/UnsealNonTerminalPayloadsCommandHandler.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilities.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityMarkers.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/FleetCapabilityRules.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/HangfireWorkerFleetRegistry.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/IFleetCapabilities.cs create mode 100644 backend/src/CodeSpace.Core/Services/Fleet/IWorkerFleetRegistry.cs create mode 100644 backend/src/CodeSpace.Core/Services/Workflows/Runtime/ISealedPayloadUnsealService.cs create mode 100644 backend/src/CodeSpace.Core/Services/Workflows/Runtime/SealedPayloadUnsealService.cs create mode 100644 backend/src/CodeSpace.Messages/Commands/Workflows/UnsealNonTerminalPayloadsCommand.cs create mode 100644 backend/src/CodeSpace.Messages/Fleet/WorkerServerHeartbeat.cs create mode 100644 backend/tests/CodeSpace.E2ETests/Infrastructure/DeferredDrainFleetRegistry.cs create mode 100644 backend/tests/CodeSpace.E2ETests/Jobs/WorkerFleetCapabilityE2ETests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Fleet/HangfireFleetHarness.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Fleet/RollingWorkerPayloadSealingFlowTests.cs create mode 100644 backend/tests/CodeSpace.IntegrationTests/Hangfire/RecurringJobScheduleKeeperFlowTests.cs create mode 100644 backend/tests/CodeSpace.UnitTests/Fleet/FixedFleetCapabilities.cs create mode 100644 backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilitiesTests.cs create mode 100644 backend/tests/CodeSpace.UnitTests/Fleet/FleetCapabilityRulesTests.cs 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); + } }