From e8a7191782699df7cd98d8516757b538b59e0460 Mon Sep 17 00:00:00 2001 From: Proxicon Date: Mon, 19 Oct 2026 11:33:35 +0200 Subject: [PATCH] Fix exact job mutation and run-action confirmation replay for beta.30 --- Directory.Build.props | 2 +- docs/mcp-http/campaign-retest.md | 97 ++++- docs/release-notes/0.1.1-beta.30.md | 49 +++ docs/release-notes/README.md | 1 + .../Client/McpOperatorJobEndpoints.cs | 107 ++++- .../Operations/McpOperatorContracts.cs | 16 +- .../McpOperatorConfirmationService.cs | 3 + .../Services/McpOperatorJobStore.cs | 24 +- .../API/McpOperatorJobEndpointTests.cs | 378 ++++++++++++++++++ .../API/McpOperatorJobStartEndpointTests.cs | 233 ++++++++++- .../McpOperatorConfirmationServiceTests.cs | 21 +- 11 files changed, 893 insertions(+), 38 deletions(-) create mode 100644 docs/release-notes/0.1.1-beta.30.md diff --git a/Directory.Build.props b/Directory.Build.props index b9bb479b..8e0b66e4 100644 --- a/Directory.Build.props +++ b/Directory.Build.props @@ -6,7 +6,7 @@ --> 0.1.1 - beta.29 + beta.30 $(VersionPrefix) $(VersionPrefix)-$(VersionSuffix) $(Version) diff --git a/docs/mcp-http/campaign-retest.md b/docs/mcp-http/campaign-retest.md index 8db3778a..a667dcc2 100644 --- a/docs/mcp-http/campaign-retest.md +++ b/docs/mcp-http/campaign-retest.md @@ -7,7 +7,102 @@ publication integrity and owner functional acceptance are separate stages. [campaign #294](https://github.com/BostonTechnologies/netratel/issues/294) remain open until the owner explicitly disposes the campaign contexts. -## Current authority: beta.27 owner checkpoint and beta.28 continuation +## Current authority: beta.29 owner checkpoint and beta.30 continuation + +The owner tested beta.29 at public source +`4b148ebe824b3de445a12d30e95b8f5c30d53655`. Its API, Web, HTTP MCP and migration +deployment digests matched publication; an independent read-only database check +confirmed the required migrations. Retained key-ring storage was observed; +full key-file and backup continuity was not established. + +The checkpoint contains 33/35 fresh bounded functional assertions. Both failures +are manifestations of [#333](https://github.com/BostonTechnologies/netratel/issues/333): +exact successful definition-update and parameter-add confirmations were rejected +by mutable ETag validation before replay reconciliation. Historical remediation +(12/12), metadata-only projections (2/2), access controls (7/7), and new-fixture +cleanup (3/3) are separate populations. + +Beta.30 authenticates recorded definition outcomes before fresh-write validation. +For newly completed mutations, replay must preserve the original job ID, +resulting version and added item ID even after later edits or soft deletion, +without another write. Current authority, ownership, target/policy binding and +plan expiry remain enforced. Legacy ID-only receipts retain their documented +current ownership/tombstone-version fallback; do not invent missing historical +fields or replay expired historical plans. + +The release source review also identified and repairs +[#334](https://github.com/BostonTechnologies/netratel/issues/334): exact successful +cancellation replay after termination and deletion replay after tombstoning. +This is a separate source-review finding, not a second beta.29 owner-observed +defect. Retest both on fresh owned disposable runs within the valid confirmation +window. Compare the original run ID and action-result state, and prove no repeat +cancellation, deletion or audit mutation. Keep current gateway prerequisites +for cancellation and current authority/ownership/expiry checks for both actions. + +### Beta.30 execution order + +1. The owner deploys the verified beta.30 artifacts and confirms actual running + image digests, source/version and migration state. Preserve the existing + Data Protection key ring and backup. Distribution proof alone does not prove + which components are deployed. +2. The owner recovers the original QA container/volume and supplies independently + enrolled beta.30 Windows/Linux disposable executors. The beta.29 campaign's + original executors still ran beta.25 and its original QA gateway remained + offline. Follow the [isolated fixture guidance](../client/CLIENT.md#disposable-executors-for-owner-acceptance); + an extra executable directory does not isolate a canonical device-key store. +3. The owner replaces/reconnects the selected consumer through supported controls. + Prove its exact selected process, loaded implementation assembly and complete + raw native schema identity using the [bridge identity guide](../integrations/http-mcp-bridge.md#prove-which-bridge-the-consumer-is-running). + The beta.29 checkpoint observed beta.26/beta.27 bridges. Current descriptions, + diagnostic bridges and configuration are separate from selected-process proof. +4. Revalidate exact current target, profile, revision, grant and expiry state. + Preserve the primary and auxiliary scope and original deadlines. Do not + replay consumed reviews. Four positive contexts, `clients/update_metadata` + and `monitoring/permissions` in both original tenants, denied exact admission + despite configured grants. Obtain legitimate review from a distinct eligible + authenticated reviewer if those positive cases are required; neighboring + admitted reads do not establish their admission. +5. Retest #333 on newly reviewed disposable definitions, covering update, + parameter/step operations, deletion, original-result fidelity after later + changes, and no duplicate writes. Retain create/start replay, changed-request, + stale-new-write, current-authority and single-winner controls. Complete the + remaining #305/#306 profile/bootstrap/upsert/classification/tag/revision/race/ + rollback, fresh secret-reader and output-surface, service timing, injection, + monitoring trigger/clear and fresh flow/event/delivery assertions. +6. Complete all eight actual native prompt workflows with real required human + decisions and useful effects. The checkpoint completed 0/8. Prove actual + native progress notifications and in-flight protocol cancellation under #288; + polling output and application-job cancellation do not supply that evidence. + Bounded Linux/Windows browse behavior now has owner regression evidence under + #321, including filtered-entry behavior at the exact eligible limit. Preserve + the two unsupported continuation contexts and deferred autofill. +7. After all other campaign and paired-product traffic ends, run #310's unchanged + exact isolated hour: 240 serial calls at offsets 0, 15 through 3585 seconds, + with observation through 3600 seconds. Repeat tenant A flow list, auth status, + tenant B flow list, system version 60 times. Preserve every failed, missed or + overrun slot without retries or catch-up; separate 120 core samples and 60 + admitted reads per tenant. No qualifying endurance hour ran in beta.29. +8. Reconcile the original receiver incidents and background artifacts without + duplicate delivery, then finish original dependency-ordered cleanup, effective + auxiliary revocation and update-intent restoration. New beta.29 fixtures and + inspection sessions were removed; original cleanup remains open. Preserve + terminal audit and protected state. Infrastructure disposal is owner work. + +The local test agent performs acceptance, evidence collection and explicitly +authorized campaign-fixture operations. Deployment, infrastructure recovery, +restarts/upgrades, installation/enrolment and consumer reconnection are owner +actions. Continue unaffected admitted checks when a prerequisite blocks one +area; do not turn a failed acceptance check into source or release work. + +Preserve all 518 context IDs and history: 438 original, 60 earlier additions and +20 beta.27 additions. Beta.29 invocation coverage is 110/518; dispositions are +104 bounded passes, 2 failed contexts, 4 exact admission denials, 391 untested, +14 prohibited, 2 unsupported and 1 deferred. All prohibited contexts remained +uninvoked. Append beta.30 attempts and record each evidence population separately. +Keep #282/#294/#288/#304/#305/#306/#308/#310 open for their actual remaining +criteria. Delivered repairs #315–#318 and #321/#322 remain closed. + +## Historical beta.27 owner checkpoint and beta.28 continuation The beta.27 owner checkpoint supplies bounded deployed regression evidence for the repairs delivered by [#319](https://github.com/BostonTechnologies/netratel/pull/319). diff --git a/docs/release-notes/0.1.1-beta.30.md b/docs/release-notes/0.1.1-beta.30.md new file mode 100644 index 00000000..14020352 --- /dev/null +++ b/docs/release-notes/0.1.1-beta.30.md @@ -0,0 +1,49 @@ +# 0.1.1-beta.30 — exact job mutation and run-action replay + +The beta.29 owner campaign reproduced successful job updates and parameter +additions whose exact confirmation replay returned `job_etag_mismatch`. +The original writes had succeeded once. [#333](https://github.com/BostonTechnologies/netratel/issues/333) +corrects reconciliation of those completed mutations. + +All ten job-definition mutations now check for an authenticated recorded outcome +before validating the current definition version. An exact successful replay +returns the original job ID, resulting version and added parameter or step ID, +with `replayed: true`, including after later edits or soft deletion. It does not +repeat the mutation or create another mutation audit record. Unused plans stay +unconsumed until fresh-request validation succeeds. + +Current caller, tenant, target, policy, ownership, plan binding and expiry still +apply. Fresh stale writes and changed confirmations remain rejected. Replays +of older receipts that stored only a job ID retain a compatibility fallback: +the current authorized ownership/tombstone version and no historical item ID. +Those older receipts cannot reconstruct fields they never stored. + +The source review also identified +[#334](https://github.com/BostonTechnologies/netratel/issues/334): cancellation +replay was rejected after a run became terminal, and deletion replay was rejected +after the first deletion hid the run. The run-action repair applies the same +authenticated reconciliation before mutable state checks. New receipts preserve +the original run ID and action-result state; replay does not request another +cancellation or repeat the deletion/audit. Current cancellation gateway +requirements remain enforced. This additional finding came from source review, +separately from the beta.29 owner evidence. + +The repair uses the existing confirmation and job-ownership storage, without a +new database migration. Existing endpoint/store tests cover the replay behavior +and denial controls. Required hosted validation, release builds and publication +use the existing generic workflows. + +## Next owner campaign + +Deploy the verified beta.30 server artifacts, prepare independently enrolled +beta.30 Windows/Linux fixtures, recover the original QA fixture, and reconnect +the selected consumer using the supported controls. Verify actual loaded +component identities before release-specific acceptance. The +[owner repeat campaign](../mcp-http/campaign-retest.md) preserves the beta.29 +checkpoint and remaining work. + +Source repair and distribution integrity are separate from fresh owner +acceptance in [#294](https://github.com/BostonTechnologies/netratel/issues/294). +That campaign retains all 518 contexts, the eight actual native workflows, +the exact isolated 240-call hour, and the original cleanup obligations. +Autofill remains deferred. diff --git a/docs/release-notes/README.md b/docs/release-notes/README.md index ee9b5076..c91cd528 100644 --- a/docs/release-notes/README.md +++ b/docs/release-notes/README.md @@ -2,6 +2,7 @@ These notes describe the named release and preserve historical status. [Release engineering](../deployment/RELEASES.md) describes the current publication process; the product version is controlled only by `Directory.Build.props`. +- [0.1.1-beta.30 — exact job mutation and run-action replay](0.1.1-beta.30.md) - [0.1.1-beta.29 candidate — managed automation retry attempts](0.1.1-beta.29.md) - [0.1.1-beta.11 — Client operations and Netclaw investigations](0.1.1-beta.11.md) diff --git a/src/NetRatel/NetRatel.API/Endpoints/Client/McpOperatorJobEndpoints.cs b/src/NetRatel/NetRatel.API/Endpoints/Client/McpOperatorJobEndpoints.cs index 78ec86a3..fceac89f 100644 --- a/src/NetRatel/NetRatel.API/Endpoints/Client/McpOperatorJobEndpoints.cs +++ b/src/NetRatel/NetRatel.API/Endpoints/Client/McpOperatorJobEndpoints.cs @@ -1,3 +1,4 @@ +using System.Globalization; using System.Security.Cryptography; using System.Text; using System.Text.Json; @@ -189,11 +190,28 @@ private static async Task ConfirmMutationAsync(int tenantId, Guid agent if (result.Failure is { } failure) return failure; var context = result.Context!; if (!HasPlan(request.PlanToken, request.IdempotencyKey)) return Failure("confirmation_plan_invalid", context); + var confirmationRequest = new McpOperatorConfirmationRequest(request.PlanToken!, request.IdempotencyKey!, MutationHash(action, request), context.Decision); + // Successful mutations advance the version or hide the definition. Only + // an authenticated existing admission may bypass those mutable checks; + // fresh stale/invalid attempts must not consume their confirmation plan. + var replayRequest = confirmationRequest with { ReplayOnly = true }; + var replay = await confirmations.ConfirmAsync(replayRequest, cancellationToken).ConfigureAwait(false); + if (replay.FailureCode is { } replayCode) return Failure(replayCode, context); + if (replay.IsReplay) return await ReplayMutationAsync(action, request, replay, jobs, context, cancellationToken).ConfigureAwait(false); var validation = await ValidateMutationAsync(action, request, jobs, context, cancellationToken).ConfigureAwait(false); - if (validation.FailureCode is { } code) return Failure(code, context); - var confirmation = await confirmations.ConfirmAsync(new McpOperatorConfirmationRequest(request.PlanToken!, request.IdempotencyKey!, MutationHash(action, request), context.Decision), cancellationToken).ConfigureAwait(false); + if (validation.FailureCode is { } code) + { + // A concurrent confirmation can finish between the replay lookup + // and validation. Reconcile it without admitting a new dispatch. + replay = await confirmations.ConfirmAsync(replayRequest, cancellationToken).ConfigureAwait(false); + if (replay.FailureCode is { } concurrentCode) return Failure(concurrentCode, context); + return replay.IsReplay + ? await ReplayMutationAsync(action, request, replay, jobs, context, cancellationToken).ConfigureAwait(false) + : Failure(code, context); + } + var confirmation = await confirmations.ConfirmAsync(confirmationRequest, cancellationToken).ConfigureAwait(false); if (confirmation.FailureCode is { } confirmationCode) return Failure(confirmationCode, context); - if (confirmation.IsReplay) return await ReplayMutationAsync(action, confirmation, jobs, context, cancellationToken).ConfigureAwait(false); + if (confirmation.IsReplay) return await ReplayMutationAsync(action, request, confirmation, jobs, context, cancellationToken).ConfigureAwait(false); if (!confirmation.IsNewDispatch || confirmation.IdempotencyId is not { } idempotencyId) return Failure("confirmation_plan_invalid", context); try @@ -201,7 +219,8 @@ private static async Task ConfirmMutationAsync(int tenantId, Guid agent var audit = await admission.RecordAcceptedAsync(context.Request, cancellationToken).ConfigureAwait(false); var mutation = await ApplyMutationAsync(action, request, jobs, context.Decision, audit, cancellationToken).ConfigureAwait(false); if (mutation is null) { await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Failed, "job_not_found", CancellationToken.None).ConfigureAwait(false); return Failure("job_not_found", context); } - await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Succeeded, mutation.Value.Job.JobId.ToString(), cancellationToken).ConfigureAwait(false); + var resultReference = string.Create(CultureInfo.InvariantCulture, $"job-mutation:{mutation.Value.Job.JobId}:{mutation.Value.Job.Version}:{mutation.Value.ItemId}"); + await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Succeeded, resultReference, cancellationToken).ConfigureAwait(false); return Results.Ok(new McpOperatorJobMutationResult(Decimal(mutation.Value.Job.JobId), Decimal(mutation.Value.Job.Version), Decimal(mutation.Value.ItemId), action == "delete", false, context.CorrelationId)); } catch (McpOperatorJobConcurrencyException) @@ -408,11 +427,25 @@ private static async Task ConfirmRunActionAsync(string action, int tena if (result.Failure is { } failure) return failure; var context = result.Context!; if (!HasPlan(request.PlanToken, request.IdempotencyKey)) return Failure("confirmation_plan_invalid", context); + var confirmationRequest = new McpOperatorConfirmationRequest(request.PlanToken!, request.IdempotencyKey!, RunActionHash(action, runId, request.Reason), context.Decision); + var replayRequest = confirmationRequest with { ReplayOnly = true }; + var replay = await confirmations.ConfirmAsync(replayRequest, cancellationToken).ConfigureAwait(false); + if (replay.FailureCode is { } replayCode) return Failure(replayCode, context); + if (replay.IsReplay) return await ReplayRunActionAsync(runId, replay, jobs, context, cancellationToken).ConfigureAwait(false); var run = await GetOwnedRunAsync(runId, jobs, context, cancellationToken).ConfigureAwait(false); - if (run is null || run.TargetSetDigest != context.Decision.Request.TargetSetDigest || (action == "cancel" && IsTerminal(run.State)) || (action == "delete" && !IsTerminal(run.State))) return Failure("job_run_action_invalid", context); - var confirmation = await confirmations.ConfirmAsync(new McpOperatorConfirmationRequest(request.PlanToken!, request.IdempotencyKey!, RunActionHash(action, runId, request.Reason), context.Decision), cancellationToken).ConfigureAwait(false); + if (run is null || run.TargetSetDigest != context.Decision.Request.TargetSetDigest || (action == "cancel" && IsTerminal(run.State)) || (action == "delete" && !IsTerminal(run.State))) + { + // Cancellation can finish, or deletion can hide the run, while the + // same confirmation is being retried. Fresh invalid plans stay unused. + replay = await confirmations.ConfirmAsync(replayRequest, cancellationToken).ConfigureAwait(false); + if (replay.FailureCode is { } concurrentCode) return Failure(concurrentCode, context); + return replay.IsReplay + ? await ReplayRunActionAsync(runId, replay, jobs, context, cancellationToken).ConfigureAwait(false) + : Failure("job_run_action_invalid", context); + } + var confirmation = await confirmations.ConfirmAsync(confirmationRequest, cancellationToken).ConfigureAwait(false); if (confirmation.FailureCode is { } code) return Failure(code, context); - if (confirmation.IsReplay) return ReplayRunAction(confirmation, run, context); + if (confirmation.IsReplay) return await ReplayRunActionAsync(runId, confirmation, jobs, context, cancellationToken).ConfigureAwait(false); if (!confirmation.IsNewDispatch || confirmation.IdempotencyId is not { } idempotencyId) return Failure("confirmation_plan_invalid", context); try { @@ -436,7 +469,8 @@ private static async Task ConfirmRunActionAsync(string action, int tena await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Failed, "job_run_action_invalid", CancellationToken.None).ConfigureAwait(false); return Failure("job_run_action_invalid", context); } - await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Succeeded, runId.ToString(), cancellationToken).ConfigureAwait(false); + var resultReference = string.Create(CultureInfo.InvariantCulture, $"job-run-action:{runId}:{updated.State}"); + await confirmations.CompleteAsync(idempotencyId, McpOperatorIdempotencyOutcome.Succeeded, resultReference, cancellationToken).ConfigureAwait(false); return Results.Ok(new McpOperatorJobRunResult(runId.ToString(), updated.State, false, context.CorrelationId)); } catch (McpOperatorAdmissionRejectedException rejection) @@ -523,15 +557,33 @@ private static async Task ValidateMutationAsync(string actio private static (McpOperatorJobLease Job, long? ItemId)? ToMutation(McpOperatorJobLease? job) => job is null ? null : (job, null); - private static async Task ReplayMutationAsync(string action, McpOperatorConfirmationAdmission confirmation, IMcpOperatorJobStore jobs, McpOperatorJobContext context, CancellationToken cancellationToken) + private static async Task ReplayMutationAsync(string action, McpOperatorJobMutationRequest request, McpOperatorConfirmationAdmission confirmation, IMcpOperatorJobStore jobs, McpOperatorJobContext context, CancellationToken cancellationToken) { if (confirmation.Outcome == McpOperatorIdempotencyOutcome.Pending) return Failure("idempotency_pending", context); - if (confirmation.Outcome != McpOperatorIdempotencyOutcome.Succeeded || !long.TryParse(confirmation.ResultReference, out var jobId)) return Failure("idempotency_replay_unavailable", context); - if (action == "delete") return Results.Ok(new McpOperatorJobMutationResult(Decimal(jobId), "0", null, true, true, context.CorrelationId)); - var job = await GetOwnedJobAsync(jobId, jobs, context, cancellationToken).ConfigureAwait(false); - return job is null ? Failure("idempotency_replay_unavailable", context) : Results.Ok(new McpOperatorJobMutationResult(Decimal(job.JobId), Decimal(job.Version), null, false, true, context.CorrelationId)); + if (confirmation.Outcome != McpOperatorIdempotencyOutcome.Succeeded || ReadMutationReference(confirmation.ResultReference) is not { } recorded || + (action != "create" && request.JobId != recorded.JobId)) return Failure("idempotency_replay_unavailable", context); + var job = await jobs.GetOwnedForReplayAsync(recorded.JobId, context.TenantId, context.AgentId, + context.Request.Principal, context.Request.McpResource!, context.Request.McpInstance!, cancellationToken).ConfigureAwait(false); + if (job is null || job.TargetSetDigest != context.Decision.Request.TargetSetDigest) return Failure("idempotency_replay_unavailable", context); + // Older receipts retained only the job id. They still require current + // ownership, and can report only the current owner/tombstone version; + // their original version and created item id cannot be reconstructed. + return Results.Ok(new McpOperatorJobMutationResult(Decimal(recorded.JobId), Decimal(recorded.Version ?? job.Version), + Decimal(recorded.ItemId), action == "delete", true, context.CorrelationId)); + } + + private static MutationReference? ReadMutationReference(string? reference) + { + if (PositiveDecimal(reference, out var legacyJobId)) return new(legacyJobId, null, null); + if (reference?.Split(':') is not ["job-mutation", var jobIdText, var versionText, var itemIdText] || + !PositiveDecimal(jobIdText, out var jobId) || !PositiveDecimal(versionText, out var version)) return null; + if (itemIdText.Length == 0) return new(jobId, version, null); + return PositiveDecimal(itemIdText, out var itemId) ? new(jobId, version, itemId) : null; } + private static bool PositiveDecimal(string? value, out long parsed) => + long.TryParse(value, NumberStyles.None, CultureInfo.InvariantCulture, out parsed) && parsed > 0; + private static async Task CanStartAsync(long jobId, McpOperatorJobRunRequest request, IMcpOperatorJobStore jobs, McpOperatorJobContext context, CancellationToken cancellationToken) { // The signed start action requires current job.manage and script.edit @@ -551,10 +603,28 @@ private static async Task ReplayRunStartAsync(McpOperatorConfirmationAd return run is null ? Failure("idempotency_replay_unavailable", context) : Results.Ok(new McpOperatorJobRunResult(runId.ToString(), run.State, true, context.CorrelationId)); } - private static IResult ReplayRunAction(McpOperatorConfirmationAdmission confirmation, McpOperatorJobRunLease run, McpOperatorJobContext context) => - confirmation.Outcome == McpOperatorIdempotencyOutcome.Succeeded && confirmation.ResultReference == run.RunId.ToString() - ? Results.Ok(new McpOperatorJobRunResult(run.RunId.ToString(), run.State, true, context.CorrelationId)) - : Failure(confirmation.Outcome == McpOperatorIdempotencyOutcome.Pending ? "idempotency_pending" : "idempotency_replay_unavailable", context); + private static async Task ReplayRunActionAsync(ulong runId, McpOperatorConfirmationAdmission confirmation, + IMcpOperatorJobStore jobs, McpOperatorJobContext context, CancellationToken cancellationToken) + { + var run = await jobs.GetRunOwnedForReplayAsync(runId, context.TenantId, context.AgentId, context.Request.Principal, + context.Request.McpResource!, context.Request.McpInstance!, cancellationToken).ConfigureAwait(false); + if (run is null || run.TargetSetDigest != context.Decision.Request.TargetSetDigest) return Failure("idempotency_replay_unavailable", context); + if (confirmation.Outcome == McpOperatorIdempotencyOutcome.Pending) return Failure("idempotency_pending", context); + if (confirmation.Outcome != McpOperatorIdempotencyOutcome.Succeeded || ReadRunActionReference(confirmation.ResultReference) is not { } recorded || + recorded.RunId != runId) return Failure("idempotency_replay_unavailable", context); + // Integer-only legacy receipts can report only the current owned run + // state. New receipts retain the original confirmation response state. + return Results.Ok(new McpOperatorJobRunResult(runId.ToString(CultureInfo.InvariantCulture), recorded.State ?? run.State, true, context.CorrelationId)); + } + + private static RunActionReference? ReadRunActionReference(string? reference) + { + if (PositiveDecimal(reference, out var legacyRunId)) return new((ulong)legacyRunId, null); + if (reference?.Split(':') is not ["job-run-action", var runIdText, var stateText] || + !PositiveDecimal(runIdText, out var runId) || !Enum.TryParse(stateText, out var state) || + !Enum.IsDefined(state) || state.ToString() != stateText) return null; + return new((ulong)runId, stateText); + } private static async Task GetOwnedJobAsync(long jobId, IMcpOperatorJobStore jobs, McpOperatorJobContext context, CancellationToken cancellationToken) => await jobs.GetOwnedAsync(jobId, context.TenantId, context.AgentId, context.Request.Principal, context.Request.McpResource!, context.Request.McpInstance!, cancellationToken).ConfigureAwait(false); @@ -712,6 +782,9 @@ private static IResult Failure(string code, McpOperatorJobContext context) private sealed record McpOperatorJobContext(McpOperatorRouteAccessRequest Request, McpOperatorDecision Decision, string RequiredScope, string Tool, string Operation, int TenantId, Guid AgentId, string CorrelationId); private sealed record McpOperatorJobContextResult(McpOperatorJobContext? Context, IResult? Failure); + private readonly record struct MutationReference(long JobId, long? Version, long? ItemId); + private readonly record struct RunActionReference(ulong RunId, string? State); + private readonly record struct MutationValidation(string? FailureCode) { public static MutationValidation Valid { get; } = new(null); diff --git a/src/NetRatel/NetRatel.Application/Operations/McpOperatorContracts.cs b/src/NetRatel/NetRatel.Application/Operations/McpOperatorContracts.cs index 50bc9a3c..666c0a3f 100644 --- a/src/NetRatel/NetRatel.Application/Operations/McpOperatorContracts.cs +++ b/src/NetRatel/NetRatel.Application/Operations/McpOperatorContracts.cs @@ -484,7 +484,14 @@ public sealed record McpOperatorConfirmationRequest( string PlanToken, string IdempotencyKey, string PayloadHash, - McpOperatorDecision CurrentDecision); + McpOperatorDecision CurrentDecision) +{ + /// + /// Authenticates an existing replay without consuming a fresh plan. A valid, + /// unconsumed plan returns neither a replay nor a new dispatch admission. + /// + public bool ReplayOnly { get; init; } +} /// /// Result of confirmation-plan consumption and idempotency admission. Routes @@ -1205,6 +1212,11 @@ Task> ListOwnedAsync( long jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken); + /// Checks current ownership, including retained deletion tombstones, for an authenticated mutation replay. + Task GetOwnedForReplayAsync( + long jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, + CancellationToken cancellationToken); + /// Ensures every enabled step still points at the reviewed exact script revision. Task IsExecutableAsync( long jobId, McpOperatorDecision decision, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, @@ -1236,6 +1248,8 @@ Task> ListStepsAsync( Task DeleteStepAsync(long jobId, long stepId, long expectedVersion, McpOperatorDecision decision, McpOperatorAcceptedAudit acceptedAudit, DateTimeOffset occurredAtUtc, CancellationToken cancellationToken); Task GetRunOwnedAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken); + /// Checks current run ownership, including retained deletion tombstones, for an authenticated action replay. + Task GetRunOwnedForReplayAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken); Task> ListRunsOwnedAsync(long? jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken); Task RecordStartedRunAsync(long jobId, ulong runId, Guid idempotencyId, McpOperatorDecision decision, McpOperatorAcceptedAudit acceptedAudit, DateTimeOffset occurredAtUtc, CancellationToken cancellationToken); Task RecordCancellationRequestedAsync(ulong runId, McpOperatorDecision decision, McpOperatorAcceptedAudit acceptedAudit, DateTimeOffset occurredAtUtc, CancellationToken cancellationToken); diff --git a/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorConfirmationService.cs b/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorConfirmationService.cs index ad3b07bc..d2aa48fc 100644 --- a/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorConfirmationService.cs +++ b/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorConfirmationService.cs @@ -96,6 +96,9 @@ public async Task ConfirmAsync( if (plan.ConsumedAtUtc is not null) return McpOperatorConfirmationAdmission.Denied("confirmation_plan_stale"); + if (request.ReplayOnly) + return new McpOperatorConfirmationAdmission(false, false, null, null, null, null); + var idempotency = new McpOperatorIdempotencyRecord { Id = Guid.NewGuid(), diff --git a/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorJobStore.cs b/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorJobStore.cs index eb32cbb5..5f0d96b4 100644 --- a/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorJobStore.cs +++ b/src/NetRatel/NetRatel.Infrastructure/Services/McpOperatorJobStore.cs @@ -65,14 +65,24 @@ public async Task> ListOwnedAsync( return rows.Select(entry => ToLease(entry.owner, entry.job)).ToArray(); } - public async Task GetOwnedAsync( + public Task GetOwnedAsync( long jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, + CancellationToken cancellationToken) => + GetOwnedAsync(jobId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: false, cancellationToken); + + public Task GetOwnedForReplayAsync( + long jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, + CancellationToken cancellationToken) => + GetOwnedAsync(jobId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: true, cancellationToken); + + private async Task GetOwnedAsync( + long jobId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, bool includeDeleted, CancellationToken cancellationToken) { if (jobId <= 0 || !IsOwnerInputValid(tenantId, agentId, principal, mcpResource, mcpInstance)) return null; - var row = await FindOwnedAsync(jobId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: false, cancellationToken) + var row = await FindOwnedAsync(jobId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted, cancellationToken) .ConfigureAwait(false); return row is null ? null : ToLease(row.owner, row.job); } @@ -449,10 +459,16 @@ public async Task CreateAsync(McpOperatorJobCreateRequest r return ToLease(row.owner, row.job); } - public async Task GetRunOwnedAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken) + public Task GetRunOwnedAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken) => + GetRunOwnedAsync(runId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: false, cancellationToken); + + public Task GetRunOwnedForReplayAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, CancellationToken cancellationToken) => + GetRunOwnedAsync(runId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: true, cancellationToken); + + private async Task GetRunOwnedAsync(ulong runId, int tenantId, Guid agentId, McpOperatorPrincipal principal, string mcpResource, string mcpInstance, bool includeDeleted, CancellationToken cancellationToken) { if (runId == 0 || runId > long.MaxValue || !IsOwnerInputValid(tenantId, agentId, principal, mcpResource, mcpInstance)) return null; - var row = await FindOwnedRunAsync(runId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted: false, cancellationToken).ConfigureAwait(false); + var row = await FindOwnedRunAsync(runId, tenantId, agentId, principal, mcpResource, mcpInstance, includeDeleted, cancellationToken).ConfigureAwait(false); return row is null ? null : ToRunLease(row.run, row.projection); } diff --git a/src/NetRatel/NetRatel.Tests/API/McpOperatorJobEndpointTests.cs b/src/NetRatel/NetRatel.Tests/API/McpOperatorJobEndpointTests.cs index 47fc3f4c..9563698e 100644 --- a/src/NetRatel/NetRatel.Tests/API/McpOperatorJobEndpointTests.cs +++ b/src/NetRatel/NetRatel.Tests/API/McpOperatorJobEndpointTests.cs @@ -1,8 +1,22 @@ +using System.Net; +using System.Net.Http.Json; +using System.Security.Claims; +using System.Security.Cryptography; +using System.Text; +using System.Text.Encodings.Web; +using System.Text.Json; using AwesomeAssertions; +using Microsoft.AspNetCore.Authentication; using Microsoft.AspNetCore.Authorization; using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Routing; +using Microsoft.AspNetCore.TestHost; +using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; using NetRatel.API.Endpoints.Client; using NetRatel.API.Gateway; using NetRatel.API.Middleware; @@ -11,6 +25,8 @@ using NetRatel.Application.Jobs; using NetRatel.Application.Operations; using NetRatel.Application.Presence; +using NetRatel.Infrastructure.Persistence; +using NetRatel.Infrastructure.Services; using NetRatel.Shared.Operations; using Xunit; @@ -18,6 +34,203 @@ namespace NetRatel.Tests.API; public sealed class McpOperatorJobEndpointTests { + private const string MutationResource = "https://mcp.example.test/mcp"; + private const string MutationSubject = "definition-operator"; + private const string MutationClientId = "definition-client"; + private const string ScriptContent = "printf 'definition-marker'"; + private const long MutationJobId = 901; + private const long MutationScriptId = 801; + private const long MutationParameterId = 601; + private const long MutationStepId = 701; + private static readonly Guid MutationAgentId = Guid.Parse("965d8a10-b2aa-4f11-93c0-a4d2f3a400e0"); + private static string MutationRoot => $"/api/v2/mcp/operator/agents/42/{MutationAgentId:D}/jobs"; + + [Theory] + [InlineData("create")] + [InlineData("update")] + [InlineData("delete")] + [InlineData("param_add")] + [InlineData("param_update")] + [InlineData("param_delete")] + [InlineData("step_add")] + [InlineData("step_update")] + [InlineData("step_reorder")] + [InlineData("step_delete")] + public async Task Completed_definition_mutations_replay_the_original_result_without_persistence(string action) + { + await using var app = await BuildMutationAppAsync(); + using var client = app.GetTestClient(); + var request = await PreviewMutationAsync(client, action, MutationRequest(action)); + var confirmed = await ConfirmMutationAsync(client, action, request); + confirmed.Replayed.Should().BeFalse(); + confirmed.Deleted.Should().Be(action == "delete"); + confirmed.Version.Should().Be(action == "create" ? "1" : "2"); + if (action is "param_add" or "step_add") confirmed.ItemId.Should().NotBeNullOrEmpty(); + var writes = app.Services.GetRequiredService(); + var afterConfirmation = writes.Count; + + var replayed = await ConfirmMutationAsync(client, action, request); + + replayed.Should().Be(confirmed with { Replayed = true }); + writes.Count.Should().Be(afterConfirmation); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + (await db.McpOperatorIdempotencyRecords.CountAsync()).Should().Be(1); + (await db.McpOperatorAcceptedAudits.CountAsync()).Should().Be(1); + (await db.McpOperatorJobAudits.CountAsync()).Should().Be(1); + var jobId = long.Parse(confirmed.JobId); + var owner = await db.McpOperatorJobs.SingleAsync(row => row.JobId == jobId); + owner.Version.Should().Be(long.Parse(confirmed.Version)); + // Later edits/deletion and a changed script must not replace the + // recorded outcome or require a new validation of execution material. + owner.Version++; + owner.DeletedAtUtc = DateTimeOffset.UtcNow; + (await db.Jobs.SingleAsync(row => row.Id == jobId)).Name = "later-definition"; + (await db.McpOperatorScripts.SingleAsync()).Version++; + await db.SaveChangesAsync(); + var store = scope.ServiceProvider.GetRequiredService(); + (await store.GetOwnedAsync(jobId, 42, MutationAgentId, MutationPrincipal(), MutationResource, "prod", default)).Should().BeNull(); + } + var afterLaterChange = writes.Count; + + var replayedAfterChange = await ConfirmMutationAsync(client, action, request); + + replayedAfterChange.Should().Be(confirmed with { Replayed = true }); + writes.Count.Should().Be(afterLaterChange); + } + + [Fact] + public async Task Changed_confirmation_does_not_consume_the_reviewed_parameter_add() + { + await using var app = await BuildMutationAppAsync(); + using var client = app.GetTestClient(); + var request = await PreviewMutationAsync(client, "param_add", MutationRequest("param_add")); + var writes = app.Services.GetRequiredService(); + var beforeAttempt = writes.Count; + + var changed = await client.PostAsJsonAsync(MutationRoot + "/confirm/param_add", + request with { Parameter = request.Parameter! with { DefaultValue = "unreviewed" } }); + + changed.StatusCode.Should().Be(HttpStatusCode.Conflict); + (await changed.Content.ReadAsStringAsync()).Should().Contain("confirmation_plan_stale"); + writes.Count.Should().Be(beforeAttempt); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + (await db.McpOperatorConfirmationPlans.SingleAsync()).ConsumedAtUtc.Should().BeNull(); + (await db.McpOperatorIdempotencyRecords.CountAsync()).Should().Be(0); + (await db.JobParameters.CountAsync()).Should().Be(1); + } + + var confirmed = await ConfirmMutationAsync(client, "param_add", request); + confirmed.ItemId.Should().NotBeNullOrEmpty(); + confirmed.Version.Should().Be("2"); + } + + [Fact] + public async Task Fresh_competing_stale_confirmation_remains_unconsumed_after_the_winning_write() + { + await using var app = await BuildMutationAppAsync(); + using var client = app.GetTestClient(); + var stale = await PreviewMutationAsync(client, "update", MutationRequest("update")); + var winning = await PreviewMutationAsync(client, "update", + MutationRequest("update") with { Job = new McpOperatorJobDraft("winning-definition", "/") }); + await ConfirmMutationAsync(client, "update", winning); + var writes = app.Services.GetRequiredService(); + var beforeAttempt = writes.Count; + + var rejected = await client.PostAsJsonAsync(MutationRoot + "/confirm/update", stale); + + rejected.StatusCode.Should().Be(HttpStatusCode.Conflict); + (await rejected.Content.ReadAsStringAsync()).Should().Contain("job_etag_mismatch"); + writes.Count.Should().Be(beforeAttempt); + await using var scope = app.Services.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + (await db.McpOperatorConfirmationPlans.CountAsync(plan => plan.ConsumedAtUtc == null)).Should().Be(1); + (await db.McpOperatorIdempotencyRecords.CountAsync()).Should().Be(1); + (await db.Jobs.SingleAsync()).Name.Should().Be("winning-definition"); + (await db.McpOperatorJobs.SingleAsync()).Version.Should().Be(2); + (await db.McpOperatorJobAudits.CountAsync()).Should().Be(1); + } + + [Theory] + [InlineData("policy-version", "confirmation_plan_stale")] + [InlineData("policy-disabled", "tenant_not_authorized")] + [InlineData("expired", "confirmation_plan_expired")] + [InlineData("scope", "oauth_scope_missing")] + [InlineData("client", "confirmation_plan_stale")] + [InlineData("resource", "confirmation_plan_stale")] + [InlineData("target-version", "confirmation_plan_stale")] + [InlineData("owner", "idempotency_replay_unavailable")] + [InlineData("owner-tenant", "idempotency_replay_unavailable")] + public async Task Deleted_definition_replay_rechecks_current_authority_binding_and_tombstone_ownership(string change, string failureCode) + { + await using var app = await BuildMutationAppAsync(); + using var client = app.GetTestClient(); + var request = await PreviewMutationAsync(client, "delete", MutationRequest("delete")); + await ConfirmMutationAsync(client, "delete", request); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + var caller = app.Services.GetRequiredService(); + switch (change) + { + case "policy-version": (await db.McpOperatorPolicies.SingleAsync()).Version++; break; + case "policy-disabled": (await db.McpOperatorPolicies.SingleAsync()).DisabledAtUtc = DateTimeOffset.UtcNow; break; + case "expired": (await db.McpOperatorConfirmationPlans.SingleAsync()).ExpiresAtUtc = DateTimeOffset.UtcNow.AddMinutes(-1); break; + case "scope": caller.Scopes = []; break; + case "client": caller.ClientId = "other-client"; break; + case "resource": caller.Resource = "https://other.example.test/mcp"; break; + case "target-version": (await db.McpOperatorTargetProfiles.SingleAsync()).Version++; break; + case "owner": (await db.McpOperatorJobs.SingleAsync()).Subject = "other-owner"; break; + case "owner-tenant": (await db.McpOperatorJobs.SingleAsync()).TenantId = 43; break; + } + await db.SaveChangesAsync(); + } + var writes = app.Services.GetRequiredService(); + var beforeReplay = writes.Count; + + var replayed = await client.PostAsJsonAsync(MutationRoot + "/confirm/delete", request); + + replayed.IsSuccessStatusCode.Should().BeFalse(); + (await replayed.Content.ReadAsStringAsync()).Should().Contain(failureCode); + writes.Count.Should().Be(beforeReplay); + await using var verification = app.Services.CreateAsyncScope(); + var verificationDb = verification.ServiceProvider.GetRequiredService(); + (await verificationDb.McpOperatorJobAudits.CountAsync()).Should().Be(1); + (await verificationDb.McpOperatorIdempotencyRecords.SingleAsync()).Outcome.Should().Be(McpOperatorIdempotencyOutcome.Succeeded); + } + + [Theory] + [InlineData("param_add")] + [InlineData("delete")] + public async Task Legacy_job_id_receipts_replay_with_the_authorized_current_version(string action) + { + await using var app = await BuildMutationAppAsync(); + using var client = app.GetTestClient(); + var request = await PreviewMutationAsync(client, action, MutationRequest(action)); + var confirmed = await ConfirmMutationAsync(client, action, request); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + (await db.McpOperatorIdempotencyRecords.SingleAsync()).ResultReference = confirmed.JobId; + (await db.McpOperatorJobs.SingleAsync()).Version++; + await db.SaveChangesAsync(); + } + var writes = app.Services.GetRequiredService(); + var beforeReplay = writes.Count; + + var replayed = await ConfirmMutationAsync(client, action, request); + + replayed.JobId.Should().Be(confirmed.JobId); + replayed.Version.Should().Be("3"); + replayed.ItemId.Should().BeNull(); + replayed.Deleted.Should().Be(action == "delete"); + replayed.Replayed.Should().BeTrue(); + writes.Count.Should().Be(beforeReplay); + } + [Fact] public void Gateway_terminal_result_supplies_bounded_redacted_job_output() { @@ -64,6 +277,171 @@ public void Output_respects_remaining_line_budget_and_message_bound() "client", 42, "exec-library-script", "Completed", null, DateTimeOffset.Parse("2026-09-05T00:00:00Z"), DateTimeOffset.Parse("2026-09-05T00:00:01Z"), ResultJson: result); + private static McpOperatorJobMutationRequest MutationRequest(string action) => action switch + { + "create" => new(Job: new McpOperatorJobDraft("created-definition", "/")), + "update" => new(Job: new McpOperatorJobDraft("updated-definition", "/"), JobId: MutationJobId, ExpectedVersion: 1), + "delete" => new(JobId: MutationJobId, ExpectedVersion: 1), + "param_add" => new(JobId: MutationJobId, ExpectedVersion: 1, Parameter: new McpOperatorJobParameter("new_parameter", "string", false, DefaultValue: "reviewed")), + "param_update" => new(JobId: MutationJobId, ExpectedVersion: 1, ParameterId: MutationParameterId, Parameter: new McpOperatorJobParameter("region", "string", false, DefaultValue: "updated")), + "param_delete" => new(JobId: MutationJobId, ExpectedVersion: 1, ParameterId: MutationParameterId), + "step_add" => new(JobId: MutationJobId, ExpectedVersion: 1, Step: new McpOperatorJobStep(2, MutationScriptId, 1, MutationHash(ScriptContent))), + "step_update" => new(JobId: MutationJobId, ExpectedVersion: 1, StepId: MutationStepId, Step: new McpOperatorJobStep(1, MutationScriptId, 1, MutationHash(ScriptContent), Enabled: false)), + "step_reorder" => new(JobId: MutationJobId, ExpectedVersion: 1, StepId: MutationStepId, Ordinal: 2), + "step_delete" => new(JobId: MutationJobId, ExpectedVersion: 1, StepId: MutationStepId), + _ => throw new ArgumentOutOfRangeException(nameof(action)) + }; + + private static async Task PreviewMutationAsync(HttpClient client, string action, McpOperatorJobMutationRequest request) + { + using var response = await client.PostAsJsonAsync(MutationRoot + "/preview/" + action, request); + response.StatusCode.Should().Be(HttpStatusCode.OK, await response.Content.ReadAsStringAsync()); + var preview = (await response.Content.ReadFromJsonAsync())!; + return request with { PlanToken = preview.PlanToken, IdempotencyKey = preview.IdempotencyKey }; + } + + private static async Task ConfirmMutationAsync(HttpClient client, string action, McpOperatorJobMutationRequest request) + { + using var response = await client.PostAsJsonAsync(MutationRoot + "/confirm/" + action, request); + response.StatusCode.Should().Be(HttpStatusCode.OK, await response.Content.ReadAsStringAsync()); + return (await response.Content.ReadFromJsonAsync())!; + } + + private static async Task BuildMutationAppAsync() + { + var builder = WebApplication.CreateBuilder(new WebApplicationOptions { EnvironmentName = "Production" }); + builder.WebHost.UseTestServer(); + builder.Services.AddAuthentication("Test").AddScheme("Test", _ => { }); + builder.Services.AddAuthorization(options => options.AddPolicy("M2MOnly", policy => policy.RequireAuthenticatedUser())); + var databaseName = "job-definition-" + Guid.NewGuid(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddDbContext((provider, options) => options.UseInMemoryDatabase(databaseName) + .AddInterceptors(provider.GetRequiredService())); + builder.Services.AddScoped(); + builder.Services.AddScoped(); + builder.Services.AddScoped(); + builder.Services.AddScoped(); + builder.Services.AddScoped(_ => throw new NotSupportedException()); + builder.Services.AddScoped(_ => throw new NotSupportedException()); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + builder.Services.AddSingleton(); + var app = builder.Build(); + app.UseAuthentication(); + // Exercise real route policy, ownership and durable confirmation state + // after the signed delegation middleware's separately tested boundary. + app.Use(async (context, next) => + { + var caller = context.RequestServices.GetRequiredService(); + var action = context.Request.Path.Value!.Split('/')[^1]; + context.Items[McpOperatorDelegationMiddleware.HttpContextItemKey] = new McpOperatorDelegation( + new McpOperatorDelegationIdentity(MutationSubject, caller.ClientId, null, [], ["AutomationOperator"], caller.Scopes), + "mcp-service", "netratel_jobs", action, "definition-request", DateTimeOffset.UtcNow.AddMinutes(1), + caller.Resource, "prod", 42, MutationAgentId, "definition-correlation"); + await next(context); + }); + app.UseAuthorization(); + app.MapMcpOperatorJobEndpoints(); + await app.StartAsync(); + await using var scope = app.Services.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var now = DateTimeOffset.UtcNow; + var policyId = Guid.NewGuid(); + db.Agents.Add(new Agent { Id = MutationAgentId, TenantId = 42, IsEnabled = true, Status = AgentStatus.Active, CreatedAtUtc = now }); + db.McpOperatorTargetProfiles.Add(new McpOperatorTargetProfileRecord + { + AgentId = MutationAgentId, TenantId = 42, Classification = McpOperatorTargetClassification.ManagedStandard, + TagsJson = "[]", Version = 1, UpdatedAtUtc = now, UpdatedBy = "independent-admin" + }); + db.McpOperatorPolicies.Add(new McpOperatorPolicyRecord + { + Id = policyId, Name = "owned definition mutations", Environment = McpOperatorEnvironment.Production, + Effect = McpOperatorPolicyEffect.Allow, Priority = 10, + PrincipalSelectorKind = McpOperatorPrincipalSelectorKind.OAuthSubject, PrincipalSelectorValue = MutationSubject, + TargetSelectorKind = McpOperatorTargetSelectorKind.ExactAgent, TenantId = 42, AgentId = MutationAgentId, + TargetClassification = McpOperatorTargetClassification.ManagedStandard, OperationFamily = McpOperatorOperationFamily.AutomationWrite, + ConstraintsJson = JsonSerializer.Serialize(new McpOperatorConstraints(MaxJobTargetCount: 1, MaxFanOut: 1), McpOperatorJsonContext.Default.McpOperatorConstraints), + CreatedAtUtc = now, CreatedBy = "independent-admin", Version = 1 + }); + await db.SaveChangesAsync(); + var route = new McpOperatorRouteAccessRequest(McpOperatorEnvironment.Production, MutationPrincipal(), "mcp-service", MutationResource, + "prod", "netratel_jobs", "update", 42, MutationAgentId, new HashSet(["netratel.mcp.write"], StringComparer.Ordinal), + "definition-correlation", "definition-request", true, true); + var admitted = await scope.ServiceProvider.GetRequiredService().EvaluateAsync(route, default); + admitted.Decision.IsAllowed.Should().BeTrue(); + db.Jobs.Add(new JobDefinition { Id = MutationJobId, Name = "owned-definition", TenantId = 42, AgentId = MutationAgentId, CreatedAtUtc = now, UpdatedAtUtc = now }); + db.McpOperatorJobs.Add(new McpOperatorJobRecord + { + Id = Guid.NewGuid(), JobId = MutationJobId, TenantId = 42, AgentId = MutationAgentId, Subject = MutationSubject, ClientId = MutationClientId, + McpResource = MutationResource, McpInstance = "prod", PolicyId = policyId, PolicyVersion = 1, + TargetSetDigest = admitted.Decision.TargetSetDigest!, CreatedAtUtc = now, UpdatedAtUtc = now, Version = 1 + }); + db.JobParameters.Add(new JobParameterDefinition { Id = MutationParameterId, JobId = MutationJobId, Name = "region", Type = "string" }); + db.Scripts.Add(new ScriptDefinition { Id = MutationScriptId, Name = "owned-script", ScriptType = "bash", Content = ScriptContent }); + db.McpOperatorScripts.Add(new McpOperatorScriptRecord + { + Id = Guid.NewGuid(), ScriptId = MutationScriptId, TenantId = 42, Subject = MutationSubject, ClientId = MutationClientId, + McpResource = MutationResource, McpInstance = "prod", PolicyId = policyId, PolicyVersion = 1, + Name = "owned-script", ShellType = "sh", ContentHash = MutationHash(ScriptContent), ManifestHash = MutationHash(string.Empty), + TimeoutSeconds = 15, WorkingDirectory = "/tmp", Version = 1 + }); + for (var ordinal = 1; ordinal <= 2; ordinal++) db.JobSteps.Add(new JobStepDefinition + { + Id = MutationStepId + ordinal - 1, JobId = MutationJobId, Ordinal = ordinal, Type = (int)JobStepKind.LibraryScript, + ScriptId = MutationScriptId, Enabled = true, + PayloadJson = JsonSerializer.Serialize(new { ScriptVersion = 1, ScriptContentHash = MutationHash(ScriptContent) }) + }); + await db.SaveChangesAsync(); + return app; + } + + private static McpOperatorPrincipal MutationPrincipal() => new(MutationSubject, MutationClientId, null, + new HashSet(), new HashSet(["AutomationOperator"], StringComparer.Ordinal), + new HashSet(["netratel.mcp.write"], StringComparer.Ordinal)); + + private static string MutationHash(string value) => Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(value))); + + private sealed class MutationCaller + { + public string ClientId { get; set; } = MutationClientId; + public string Resource { get; set; } = MutationResource; + public string[] Scopes { get; set; } = ["netratel.mcp.write"]; + } + + private sealed class MutationWriteCounter : SaveChangesInterceptor + { + public int Count { get; private set; } + public override ValueTask> SavingChangesAsync(DbContextEventData eventData, + InterceptionResult result, CancellationToken cancellationToken = default) + { + Count++; + return ValueTask.FromResult(result); + } + } + + private sealed class MutationAuthentication(IOptionsMonitor options, ILoggerFactory logger, UrlEncoder encoder) + : AuthenticationHandler(options, logger, encoder) + { + protected override Task HandleAuthenticateAsync() => Task.FromResult(AuthenticateResult.Success( + new AuthenticationTicket(new ClaimsPrincipal(new ClaimsIdentity([new Claim("sub", "mcp-service")], Scheme.Name)), Scheme.Name))); + } + + private sealed class MutationPresence : IClientPresenceRouter + { + public Task StartSessionAsync(StartGatewayPresenceSession message, CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task RecordHeartbeatAsync(RecordGatewayHeartbeat message, CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task EndSessionAsync(EndGatewayPresenceSession message, CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task ProbeAsync(CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task GetSnapshotAsync(ClientKey client, CancellationToken cancellationToken) => throw new NotSupportedException(); + } + + private sealed class MutationDispatcher : IAgentCommandAuthorityDispatcher + { + public bool IsAvailable(ClientKey client) => throw new NotSupportedException(); + public Task DispatchAsync(ClientKey client, string commandId, string correlationId, string taskType, string payloadJson, int environment, CancellationToken cancellationToken) => throw new NotSupportedException(); + } + [Fact] public void Production_job_routes_are_versioned_owned_and_m2m_only() { diff --git a/src/NetRatel/NetRatel.Tests/API/McpOperatorJobStartEndpointTests.cs b/src/NetRatel/NetRatel.Tests/API/McpOperatorJobStartEndpointTests.cs index 7b5a1e76..bc146263 100644 --- a/src/NetRatel/NetRatel.Tests/API/McpOperatorJobStartEndpointTests.cs +++ b/src/NetRatel/NetRatel.Tests/API/McpOperatorJobStartEndpointTests.cs @@ -11,6 +11,7 @@ using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.TestHost; using Microsoft.EntityFrameworkCore; +using Microsoft.EntityFrameworkCore.Diagnostics; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; @@ -37,8 +38,165 @@ public sealed class McpOperatorJobStartEndpointTests private const string Content = "printf 'owned-job-marker'"; private const long JobId = 901; private const long ScriptId = 801; + private const long RunId = 1001; private static readonly Guid AgentId = Guid.Parse("965d8a10-b2aa-4f11-93c0-a4d2f3a400e0"); private static string Root => $"/api/v2/mcp/operator/agents/42/{AgentId:D}/jobs/{JobId}/runs"; + private static string ActionRoot(string action) => $"/api/v2/mcp/operator/agents/42/{AgentId:D}/jobs/runs/{RunId}/{action}"; + + [Theory] + [InlineData("cancel")] + [InlineData("delete")] + public async Task Completed_run_actions_replay_the_original_state_without_repeating_dispatch_or_persistence(string action) + { + await using var app = await BuildAsync(action); + using var client = app.GetTestClient(); + var request = await PreviewRunActionAsync(client, action); + var confirmed = await ConfirmRunActionAsync(client, action, request); + confirmed.Replayed.Should().BeFalse(); + confirmed.State.Should().Be(action == "cancel" ? nameof(JobRunState.Running) : nameof(JobRunState.Succeeded)); + var dispatch = app.Services.GetRequiredService(); + dispatch.Cancels.Should().Be(action == "cancel" ? 1 : 0); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + if (action == "cancel") (await db.JobRuns.SingleAsync()).Status = (int)JobRunState.Cancelled; + (await db.McpOperatorJobs.SingleAsync()).DeletedAtUtc = DateTimeOffset.UtcNow; + await db.SaveChangesAsync(); + if (action == "delete") + { + var store = scope.ServiceProvider.GetRequiredService(); + (await store.GetRunOwnedAsync(RunId, 42, AgentId, StartRoute().Principal, Resource, "prod", default)).Should().BeNull(); + } + } + var writes = app.Services.GetRequiredService(); + var beforeReplay = writes.Count; + + var replayed = await ConfirmRunActionAsync(client, action, request); + + replayed.Should().Be(confirmed with { Replayed = true }); + writes.Count.Should().Be(beforeReplay); + dispatch.Cancels.Should().Be(action == "cancel" ? 1 : 0); + await using var verification = app.Services.CreateAsyncScope(); + var verificationDb = verification.ServiceProvider.GetRequiredService(); + (await verificationDb.McpOperatorAcceptedAudits.CountAsync()).Should().Be(1); + (await verificationDb.McpOperatorJobRunAudits.CountAsync()).Should().Be(1); + (await verificationDb.McpOperatorJobRuns.SingleAsync()).Version.Should().Be(2); + } + + [Theory] + [InlineData("cancel")] + [InlineData("delete")] + public async Task Fresh_run_action_with_invalid_current_state_does_not_consume_the_plan(string action) + { + await using var app = await BuildAsync(action); + using var client = app.GetTestClient(); + var request = await PreviewRunActionAsync(client, action); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + if (action == "cancel") (await db.JobRuns.SingleAsync()).Status = (int)JobRunState.Succeeded; + else (await db.McpOperatorJobRuns.SingleAsync()).DeletedAtUtc = DateTimeOffset.UtcNow; + await db.SaveChangesAsync(); + } + var writes = app.Services.GetRequiredService(); + var beforeAttempt = writes.Count; + + using var rejected = await client.PostAsJsonAsync(ActionRoot(action) + "/confirm", request); + + rejected.StatusCode.Should().Be(HttpStatusCode.BadRequest); + (await rejected.Content.ReadAsStringAsync()).Should().Contain("job_run_action_invalid"); + writes.Count.Should().Be(beforeAttempt); + app.Services.GetRequiredService().Cancels.Should().Be(0); + await using var verification = app.Services.CreateAsyncScope(); + var verificationDb = verification.ServiceProvider.GetRequiredService(); + (await verificationDb.McpOperatorConfirmationPlans.SingleAsync()).ConsumedAtUtc.Should().BeNull(); + (await verificationDb.McpOperatorIdempotencyRecords.CountAsync()).Should().Be(0); + (await verificationDb.McpOperatorAcceptedAudits.CountAsync()).Should().Be(0); + } + + [Theory] + [InlineData("cancel", "pending", "idempotency_pending")] + [InlineData("delete", "pending", "idempotency_pending")] + [InlineData("cancel", "gateway", "capability_unavailable")] + [InlineData("cancel", "expired", "confirmation_plan_expired")] + [InlineData("delete", "owner", "idempotency_replay_unavailable")] + [InlineData("delete", "target", "idempotency_replay_unavailable")] + [InlineData("delete", "policy", "tenant_not_authorized")] + [InlineData("delete", "reason", "confirmation_plan_stale")] + public async Task Run_action_replay_preserves_pending_and_current_admission_denials(string action, string change, string failureCode) + { + await using var app = await BuildAsync(action); + using var client = app.GetTestClient(); + var request = await PreviewRunActionAsync(client, action); + if (change != "pending") await ConfirmRunActionAsync(client, action, request); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + if (change == "pending") + { + var plan = await db.McpOperatorConfirmationPlans.SingleAsync(); + var decision = await scope.ServiceProvider.GetRequiredService() + .EvaluateAsync(StartRoute() with { Operation = action }, default); + var admission = await scope.ServiceProvider.GetRequiredService() + .ConfirmAsync(new(request.PlanToken!, request.IdempotencyKey!, plan.PayloadHash, decision.Decision), default); + admission.IsNewDispatch.Should().BeTrue(); + } + if (action == "cancel") (await db.JobRuns.SingleAsync()).Status = (int)JobRunState.Cancelled; + else (await db.McpOperatorJobRuns.SingleAsync()).DeletedAtUtc = DateTimeOffset.UtcNow; + switch (change) + { + case "gateway": app.Services.GetRequiredService().Available = false; break; + case "expired": (await db.McpOperatorConfirmationPlans.SingleAsync()).ExpiresAtUtc = DateTimeOffset.UtcNow.AddMinutes(-1); break; + case "owner": (await db.McpOperatorJobs.SingleAsync()).Subject = "another-owner"; break; + case "target": (await db.McpOperatorJobRuns.SingleAsync()).TargetSetDigest = Hash("different-target"); break; + case "policy": (await db.McpOperatorPolicies.SingleAsync()).DisabledAtUtc = DateTimeOffset.UtcNow; break; + case "reason": request = request with { Reason = "unreviewed reason" }; break; + } + await db.SaveChangesAsync(); + } + var writes = app.Services.GetRequiredService(); + var beforeReplay = writes.Count; + var dispatch = app.Services.GetRequiredService(); + var cancelsBeforeReplay = dispatch.Cancels; + + using var replayed = await client.PostAsJsonAsync(ActionRoot(action) + "/confirm", request); + + replayed.IsSuccessStatusCode.Should().BeFalse(); + (await replayed.Content.ReadAsStringAsync()).Should().Contain(failureCode); + writes.Count.Should().Be(beforeReplay); + dispatch.Cancels.Should().Be(cancelsBeforeReplay); + await using var verification = app.Services.CreateAsyncScope(); + var verificationDb = verification.ServiceProvider.GetRequiredService(); + (await verificationDb.McpOperatorIdempotencyRecords.SingleAsync()).Outcome.Should().Be( + change == "pending" ? McpOperatorIdempotencyOutcome.Pending : McpOperatorIdempotencyOutcome.Succeeded); + (await verificationDb.McpOperatorJobRunAudits.CountAsync()).Should().Be(change == "pending" ? 0 : 1); + } + + [Theory] + [InlineData("cancel")] + [InlineData("delete")] + public async Task Legacy_run_id_receipts_replay_the_current_owned_state(string action) + { + await using var app = await BuildAsync(action); + using var client = app.GetTestClient(); + var request = await PreviewRunActionAsync(client, action); + var confirmed = await ConfirmRunActionAsync(client, action, request); + await using (var scope = app.Services.CreateAsyncScope()) + { + var db = scope.ServiceProvider.GetRequiredService(); + (await db.McpOperatorIdempotencyRecords.SingleAsync()).ResultReference = confirmed.JobRunId; + if (action == "cancel") (await db.JobRuns.SingleAsync()).Status = (int)JobRunState.Cancelled; + await db.SaveChangesAsync(); + } + var writes = app.Services.GetRequiredService(); + var beforeReplay = writes.Count; + + var replayed = await ConfirmRunActionAsync(client, action, request); + + replayed.Should().Be(confirmed with { State = action == "cancel" ? nameof(JobRunState.Cancelled) : confirmed.State, Replayed = true }); + writes.Count.Should().Be(beforeReplay); + app.Services.GetRequiredService().Cancels.Should().Be(action == "cancel" ? 1 : 0); + } [Fact] public async Task Execute_only_start_previews_confirms_and_replays_one_owned_pinned_job() @@ -150,14 +308,32 @@ public async Task Start_assertion_cannot_be_reused_to_read_arbitrary_scripts() decision.Decision.FailureCode.Should().Be("oauth_scope_missing"); } - private static async Task BuildAsync() + private static async Task PreviewRunActionAsync(HttpClient client, string action) + { + var request = new McpOperatorJobRunActionRequest("reviewed action"); + using var response = await client.PostAsJsonAsync(ActionRoot(action) + "/preview", request); + response.StatusCode.Should().Be(HttpStatusCode.OK, await response.Content.ReadAsStringAsync()); + var preview = (await response.Content.ReadFromJsonAsync())!; + return request with { PlanToken = preview.PlanToken, IdempotencyKey = preview.IdempotencyKey }; + } + + private static async Task ConfirmRunActionAsync(HttpClient client, string action, McpOperatorJobRunActionRequest request) + { + using var response = await client.PostAsJsonAsync(ActionRoot(action) + "/confirm", request); + response.StatusCode.Should().Be(HttpStatusCode.OK, await response.Content.ReadAsStringAsync()); + return (await response.Content.ReadFromJsonAsync())!; + } + + private static async Task BuildAsync(string operation = "start") { var builder = WebApplication.CreateBuilder(new WebApplicationOptions { EnvironmentName = "Production" }); builder.WebHost.UseTestServer(); builder.Services.AddAuthentication("Test").AddScheme("Test", _ => { }); builder.Services.AddAuthorization(options => options.AddPolicy("M2MOnly", policy => policy.RequireAuthenticatedUser())); var databaseName = "job-start-" + Guid.NewGuid(); - builder.Services.AddDbContext(options => options.UseInMemoryDatabase(databaseName)); + builder.Services.AddSingleton(); + builder.Services.AddDbContext((provider, options) => options.UseInMemoryDatabase(databaseName) + .AddInterceptors(provider.GetRequiredService())); builder.Services.AddScoped(); builder.Services.AddScoped(); builder.Services.AddScoped(); @@ -177,7 +353,7 @@ private static async Task BuildAsync() { context.Items[McpOperatorDelegationMiddleware.HttpContextItemKey] = new McpOperatorDelegation( new McpOperatorDelegationIdentity(Subject, ClientId, null, [], ["AutomationOperator"], ["netratel.mcp.execute"]), - "mcp-service", "netratel_job_runs", "start", "start-request", DateTimeOffset.UtcNow.AddMinutes(1), + "mcp-service", "netratel_job_runs", operation, "start-request", DateTimeOffset.UtcNow.AddMinutes(1), Resource, "prod", 42, AgentId, "start-correlation"); await next(context); }); @@ -201,20 +377,22 @@ private static async Task BuildAsync() PrincipalSelectorKind = McpOperatorPrincipalSelectorKind.OAuthSubject, PrincipalSelectorValue = Subject, TargetSelectorKind = McpOperatorTargetSelectorKind.ExactAgent, TenantId = 42, AgentId = AgentId, TargetClassification = McpOperatorTargetClassification.ManagedStandard, - OperationFamily = McpOperatorOperationFamily.JobExecution, Operation = "netratel_job_runs/start", + OperationFamily = McpOperatorOperationFamily.JobExecution, Operation = "netratel_job_runs/" + operation, ConstraintsJson = JsonSerializer.Serialize(new McpOperatorConstraints(MaxJobTargetCount: 1, MaxFanOut: 1), McpOperatorJsonContext.Default.McpOperatorConstraints), CreatedAtUtc = now, CreatedBy = "independent-admin", Version = 1 }); await db.SaveChangesAsync(); - var admitted = await scope.ServiceProvider.GetRequiredService().EvaluateAsync(StartRoute(), default); + var route = StartRoute() with { Operation = operation }; + var admitted = await scope.ServiceProvider.GetRequiredService().EvaluateAsync(route, default); admitted.Decision.IsAllowed.Should().BeTrue(); db.Jobs.Add(new JobDefinition { Id = JobId, Name = "owned-job", TenantId = 42, AgentId = AgentId, CreatedAtUtc = now, UpdatedAtUtc = now }); - db.McpOperatorJobs.Add(new McpOperatorJobRecord + var owner = new McpOperatorJobRecord { Id = Guid.NewGuid(), JobId = JobId, TenantId = 42, AgentId = AgentId, Subject = Subject, ClientId = ClientId, McpResource = Resource, McpInstance = "prod", PolicyId = policyId, PolicyVersion = 1, TargetSetDigest = admitted.Decision.TargetSetDigest!, CreatedAtUtc = now, UpdatedAtUtc = now, Version = 2 - }); + }; + db.McpOperatorJobs.Add(owner); db.Scripts.Add(new ScriptDefinition { Id = ScriptId, Name = "owned-script", ScriptType = "bash", Content = Content }); db.McpOperatorScripts.Add(new McpOperatorScriptRecord { @@ -228,6 +406,19 @@ private static async Task BuildAsync() Id = 701, JobId = JobId, Ordinal = 1, Type = (int)JobStepKind.LibraryScript, ScriptId = ScriptId, Enabled = true, PayloadJson = JsonSerializer.Serialize(new { ScriptVersion = 1, ScriptContentHash = Hash(Content) }) }); + if (operation != "start") + { + db.JobRuns.Add(new JobRunRecord + { + Id = RunId, JobId = JobId, TenantId = 42, AgentId = AgentId, StartedBy = "reviewed-operator", + Status = (int)(operation == "cancel" ? JobRunState.Running : JobRunState.Succeeded), CreatedAtUtc = now, StartedAtUtc = now + }); + db.McpOperatorJobRuns.Add(new McpOperatorJobRunRecord + { + Id = Guid.NewGuid(), JobRunId = RunId, JobRecordId = owner.Id, JobId = JobId, TenantId = 42, AgentId = AgentId, + AcceptedAuditId = Guid.NewGuid(), TargetSetDigest = owner.TargetSetDigest, CreatedAtUtc = now, UpdatedAtUtc = now, Version = 1 + }); + } await db.SaveChangesAsync(); return app; } @@ -240,7 +431,23 @@ private static async Task BuildAsync() private static IReadOnlySet Set(params string[] values) => values.ToHashSet(StringComparer.Ordinal); private static string Hash(string value) => Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(value))); - private sealed class DispatchCounter { public int Starts { get; set; } } + private sealed class DispatchCounter + { + public int Starts { get; set; } + public int Cancels { get; set; } + public bool Available { get; set; } = true; + } + + private sealed class RunWriteCounter : SaveChangesInterceptor + { + public int Count { get; private set; } + public override ValueTask> SavingChangesAsync(DbContextEventData eventData, + InterceptionResult result, CancellationToken cancellationToken = default) + { + Count++; + return ValueTask.FromResult(result); + } + } private sealed class RecordingAuthority(OrchestratorDbContext db, DispatchCounter counter) : IAkkaJobAuthorityService { @@ -261,12 +468,16 @@ public async Task StartReviewedMcpAsync(McpOperatorJobExecutionSnaps 1, now, now, now, null, request.InputsJson, null, AgentId); } public Task RecordLifecycleAsync(ClientKey client, JobLifecycleUpdateEnvelope lifecycle, CancellationToken cancellationToken) => throw new NotSupportedException(); - public Task CancelAsync(ulong jobRunId, string reason, CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task CancelAsync(ulong jobRunId, string reason, CancellationToken cancellationToken) + { + counter.Cancels++; + return Task.FromResult(true); + } } - private sealed class Dispatcher : IAgentCommandAuthorityDispatcher + private sealed class Dispatcher(DispatchCounter counter) : IAgentCommandAuthorityDispatcher { - public bool IsAvailable(ClientKey client) => true; + public bool IsAvailable(ClientKey client) => counter.Available; public Task DispatchAsync(ClientKey client, string commandId, string correlationId, string taskType, string payloadJson, int environment, CancellationToken cancellationToken) => throw new NotSupportedException(); } diff --git a/src/NetRatel/NetRatel.Tests/Infrastructure/McpOperatorConfirmationServiceTests.cs b/src/NetRatel/NetRatel.Tests/Infrastructure/McpOperatorConfirmationServiceTests.cs index e6f5b8cc..45352b44 100644 --- a/src/NetRatel/NetRatel.Tests/Infrastructure/McpOperatorConfirmationServiceTests.cs +++ b/src/NetRatel/NetRatel.Tests/Infrastructure/McpOperatorConfirmationServiceTests.cs @@ -21,10 +21,23 @@ public async Task ConfirmAsync_ConsumesTheBoundPlanOnceAndReplaysTheCompletedRes var decision = await AllowedDecisionAsync(db); var service = new McpOperatorConfirmationService(db); var plan = await service.CreatePlanAsync(new(decision, PayloadHash), Ct); - - var admitted = await service.ConfirmAsync(new(plan.PlanToken, plan.IdempotencyKey, PayloadHash, decision), Ct); + var request = new McpOperatorConfirmationRequest(plan.PlanToken, plan.IdempotencyKey, PayloadHash, decision); + + var freshLookup = await service.ConfirmAsync(request with { ReplayOnly = true }, Ct); + freshLookup.IsNewDispatch.Should().BeFalse(); + freshLookup.IsReplay.Should().BeFalse(); + freshLookup.FailureCode.Should().BeNull(); + (await db.McpOperatorConfirmationPlans.SingleAsync(Ct)).ConsumedAtUtc.Should().BeNull(); + (await db.McpOperatorIdempotencyRecords.CountAsync(Ct)).Should().Be(0); + + var admitted = await service.ConfirmAsync(request, Ct); + var pendingLookup = await service.ConfirmAsync(request with { ReplayOnly = true }, Ct); + pendingLookup.IsNewDispatch.Should().BeFalse(); + pendingLookup.IsReplay.Should().BeTrue(); + pendingLookup.Outcome.Should().Be(McpOperatorIdempotencyOutcome.Pending); await service.CompleteAsync(admitted.IdempotencyId!.Value, McpOperatorIdempotencyOutcome.Succeeded, "run-42", Ct); - var replayed = await service.ConfirmAsync(new(plan.PlanToken, plan.IdempotencyKey, PayloadHash, decision), Ct); + var replayed = await service.ConfirmAsync(request with { ReplayOnly = true }, Ct); + (await service.ConfirmAsync(request, Ct)).Should().Be(replayed); admitted.IsNewDispatch.Should().BeTrue(); admitted.IsReplay.Should().BeFalse(); @@ -50,9 +63,11 @@ public async Task ConfirmAsync_RejectsAPlanWhenThePolicyVersionChangesAfterPrevi var current = await new McpOperatorAuthorization(db).EvaluateAsync(decision.Request, Ct); var result = await service.ConfirmAsync(new(plan.PlanToken, plan.IdempotencyKey, PayloadHash, current), Ct); + var replayLookup = await service.ConfirmAsync(new(plan.PlanToken, plan.IdempotencyKey, PayloadHash, current) { ReplayOnly = true }, Ct); result.IsNewDispatch.Should().BeFalse(); result.FailureCode.Should().Be("confirmation_plan_stale"); + replayLookup.Should().Be(result); } [Fact]