From bfa5a5296519d2b7693e0a3bf674bb44ccd34bd5 Mon Sep 17 00:00:00 2001 From: Highbyte Date: Wed, 16 Sep 2026 13:24:21 +0200 Subject: [PATCH 1/2] Add reviewed CLI batches with shared web execution --- docs/reference/actions.md | 64 ++++++ docs/reference/agent-skills.md | 5 + docs/reference/storage.md | 1 + skills/wrighty/SKILL.md | 4 +- .../wrighty/references/board-and-workers.md | 34 ++++ src/Highbyte.Wrighty.Cli/CliApplication.cs | 1 + src/Highbyte.Wrighty.Cli/CliBatchCommands.cs | 106 ++++++++++ .../Output/BatchOutput.cs | 31 +++ .../Actions/WorkflowBatch.cs | 49 +++++ .../Actions/WorkflowBatchExecutor.cs | 85 ++++++++ .../Actions/WorkflowBatchService.cs | 48 +++++ .../Actions/WorkflowBatchStore.cs | 182 +++++++++++++++++ .../Caching/CachePaths.cs | 2 + .../Storage/StorageLocations.cs | 10 + .../BoardBatchExecution.cs | 34 ++++ src/Highbyte.Wrighty.Web/BoardBatchStore.cs | 7 +- .../Pages/Index.cshtml.cs | 106 ++-------- .../Pages/Shared/_BoardBatchResult.cshtml | 4 +- .../Actions/WorkflowBatchExecutorTests.cs | 77 +++++++ .../Actions/WorkflowBatchStoreTests.cs | 192 ++++++++++++++++++ .../Cli/BatchCommandTests.cs | 84 ++++++++ .../Web/BatchExecutionParityTests.cs | 86 ++++++++ .../Web/WrightyWebServerTests.cs | 2 +- .../Workers/WorkflowBatchExecutionTests.cs | 70 +++++++ 24 files changed, 1183 insertions(+), 101 deletions(-) create mode 100644 src/Highbyte.Wrighty.Cli/CliBatchCommands.cs create mode 100644 src/Highbyte.Wrighty.Cli/Output/BatchOutput.cs create mode 100644 src/Highbyte.Wrighty.Core/Actions/WorkflowBatch.cs create mode 100644 src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs create mode 100644 src/Highbyte.Wrighty.Core/Actions/WorkflowBatchService.cs create mode 100644 src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs create mode 100644 src/Highbyte.Wrighty.Web/BoardBatchExecution.cs create mode 100644 tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchExecutorTests.cs create mode 100644 tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchStoreTests.cs create mode 100644 tests/Highbyte.Wrighty.UnitTests/Cli/BatchCommandTests.cs create mode 100644 tests/Highbyte.Wrighty.UnitTests/Web/BatchExecutionParityTests.cs create mode 100644 tests/Highbyte.Wrighty.UnitTests/Workers/WorkflowBatchExecutionTests.cs diff --git a/docs/reference/actions.md b/docs/reference/actions.md index 3215ddda..dc815828 100644 --- a/docs/reference/actions.md +++ b/docs/reference/actions.md @@ -104,3 +104,67 @@ with `wrighty actions` before deciding what is currently available. See [worker lifecycle](worker.md), [claims and ownership](claims.md), and [operator actions by surface](operator-actions.md) for the underlying procedures. + +## Batch workflow actions + +```shell +wrighty batch preview queue --status "Todo" --json +wrighty batch preview queue --status "Todo" --field area=api --json +wrighty batch preview send-back --id local:12 --id local:19 --json +wrighty batch preview resume --status "In Progress" --json +wrighty batch show --json +wrighty batch execute --yes --json +``` + +Select explicit IDs (duplicates collapse to canonical IDs), or one configured workflow status +with optional exact-match `--field` filters. Filters use the same AND semantics as `list`. +These selections cannot be mixed. Only active, eligible Local Markdown items enter the frozen set. +Previews sort canonical IDs ordinally and freeze the first 100 eligible items; JSON reports +`selectedCount`, `eligibleCount`, the exact `candidates`, and `limited` so truncation is explicit. +Each candidate includes its title, reviewed state fingerprint, before state, and authorization +consequence. Previewing persists display data and hashes but does not mutate or claim an item. + +A preview expires five minutes after creation. Execute requires `--yes` in both human and JSON +modes, authorizing only this preview's operation and candidates. It starts no worker. Execution +rechecks configuration before starting and between candidates, then executes each item sequentially +through the individual action service. New matches cannot join the set; changed content, claim, +session, or eligibility produces a skipped item. The reviewed-state fingerprint is stricter than +the web Board's fresh eligibility check, so an otherwise harmless content edit also requires a +new CLI preview. The CLI and Board use the same Core batch loop for sequencing, conflict classification, +cancellation, and partial results after a systemic failure. Limits, lifetime, action eligibility, +and backend execution are also shared. Each interface retains its own preview storage and +revalidation inputs. + +All three commands return `schemaVersion: 1` and `result` containing `preview`, `state` (`preview` +or `completed`), `items`, `stopCode`, and `hasIssues`. Item outcomes are `applied`, `skipped`, +`failed`, or `unprocessed`. Applied items include their single-action before/after result. +A systemic failure stops remaining work; no successful mutation is rolled back. An ambiguous +failure has `mutationMayHaveApplied: true` and must be inspected before any retry. Execution exits +0 when all items applied, 6 for partial results, or 130 for cancellation; these outcomes retain +stdout JSON. Validation failures use normal stderr errors. `show` exits 0 for a readable record. + +Preview and result journals are scoped by the absolute configuration path under +`/workflow-batches-v1//`. Separate invocations using that +configuration and cache can share previews. An exclusive configuration-scoped file lock serializes +execution and inspection; contention returns `STORE_BUSY`. Completed records are returned on +repeat execution, even after preview expiry, without reapplying any item. Explicit `--yes` remains +required. A preview contains no claim credentials and is not execution authority. + +Before each mutation, Wrighty writes an in-flight marker and flushes it to disk. If execution +stops or the host restarts, the next `show`/`execute` marks an unfinished run interrupted: the +in-flight item is failed with an uncertain outcome, and remaining items are unprocessed. Wrighty +does not resume an interrupted batch automatically. Cancellation between items preserves definite +outcomes; cancellation during a mutation is conservatively uncertain. If journal persistence or +stdout fails, inspect the existing batch before attempting another mutation. + +The Board keeps previews/results in its web process and maps the shared executor's outcomes into +its warning panel. An accepted web batch continues if the browser disconnects. Unexpected backend +failures are retained as partial results and identify any item that may have been mutated, so a +repeated submission returns the result instead of replaying the batch. Stopping the web process +still loses its in-memory record; the CLI journal's restart recovery is specific to CLI batches. + +Records untouched for 24 hours are removed when creating another preview; each configuration holds +at most 512 records. Cache deletion loses preview/result evidence and makes old IDs unavailable; +it does not undo item mutations. Never restore, copy, or edit journals to retry work. For an expired +or missing preview, inspect current items and obtain a newly reviewed selection. Follow up with +`workers --item --json` when pickup assessment is needed. diff --git a/docs/reference/agent-skills.md b/docs/reference/agent-skills.md index e57f02b7..168e43c4 100644 --- a/docs/reference/agent-skills.md +++ b/docs/reference/agent-skills.md @@ -191,3 +191,8 @@ through `actions --exec --yes --expected-version --js requested effect is authorized. CLI and web share eligibility and locked revalidation. Execution returns the resulting state and refreshed worker pickup evidence; it does not launch a worker. See [action execution](actions.md). + +The skill also supports reviewed Local Markdown batches through `batch preview`, `batch show`, +and `batch execute --yes`. It reports the exact frozen subset and partial results, including +uncertain outcomes after interruption. CLI and web batch operations share the Core execution loop; +the CLI persists its own previews/results for cross-process use. See [batch workflow actions](actions.md#batch-workflow-actions). diff --git a/docs/reference/storage.md b/docs/reference/storage.md index 39aaa934..d3c8c4c3 100644 --- a/docs/reference/storage.md +++ b/docs/reference/storage.md @@ -68,6 +68,7 @@ runtime-specific filename placeholders rather than additional base directories. | | `sessions-v1.json` | Legacy runtime state | GitHub | No | Pre-migration session records may be lost | [Legacy files](#legacy-files) | | | `provider-availability-v1.json` | Legacy operational cache | Both | No | Pre-migration provider state may be lost | [Legacy files](#legacy-files) | | | `*.tmp` | Atomic-write temporary files | Both | No | Normally safe only when no Wrighty process is running | [Installation state and cache](#installation-state-and-cache) | +| `/workflow-batches-v1/` | `/.json`, `.lock`, `*.tmp` | Machine-local execution journal | Local Markdown | No | Frozen previews and interruption/replay evidence are lost; mutations remain applied | [Batch workflow actions](actions.md#batch-workflow-actions) | | `/worker-instances-v1/` | `/.json` | Machine-local runtime state | Both | No | Worker liveness and configuration-drift observations disappear until processes register again | [Installation state and cache](#installation-state-and-cache) | | | `/.stop.json` | Temporary machine-local control request | Both | No | A pending cooperative drain/interrupt request is lost | [Installation state and cache](#installation-state-and-cache) | | `/worker-interruptions-v1/` | `-.json` | Temporary interruption-recovery breadcrumb | Both | No | An incomplete interrupted-run finalizer is harder to diagnose; item and claim state remain authoritative | [Installation state and cache](#installation-state-and-cache) | diff --git a/skills/wrighty/SKILL.md b/skills/wrighty/SKILL.md index ba47e2c5..7b5c4c32 100644 --- a/skills/wrighty/SKILL.md +++ b/skills/wrighty/SKILL.md @@ -5,13 +5,13 @@ description: Safely operate Wrighty through the `wrighty` CLI. Use only when the # Wrighty - + Operate Wrighty state only through the `wrighty` command. Never mutate tracked state by editing local Markdown, invoking `gh`, calling GitHub APIs/MCP, writing claim comments, or changing Project fields directly. -For board overview, triage, individual Queue/Send back/Resume actions, worker discovery, and pickup prospects, read +For board overview, triage, individual and batch Queue/Send back/Resume actions, worker discovery, and pickup prospects, read [references/board-and-workers.md](references/board-and-workers.md). Use `list --json` for workflow columns, `actions --json` for the shared action catalogue, and `workers [--item ] --json` for scoped worker evidence. Discovery does not authorize mutations or worker launches. diff --git a/skills/wrighty/references/board-and-workers.md b/skills/wrighty/references/board-and-workers.md index 4c6e89c5..67dda368 100644 --- a/skills/wrighty/references/board-and-workers.md +++ b/skills/wrighty/references/board-and-workers.md @@ -65,6 +65,40 @@ but worker assessment failed: inspect again without replaying the mutation. If a without a definitive outcome, re-read the item before any retry. Other catalogue entries remain manual-only; use their documented focused procedure only within the user's authorization. +## Batch workflow actions + +For a requested Local Markdown batch, freeze the selection before authorizing execution: + +```shell +wrighty batch preview queue --status "Todo" --json +wrighty batch preview send-back --id local:12 --id local:19 --json +wrighty batch show --json +wrighty batch execute --yes --json +``` + +Use configured statuses, or explicit IDs from the user's selection. A status selection can include +`--field name=value` filters with the same exact-match AND semantics as `list`. Use `resume` for +retained sessions. Review `result.preview.candidates`, each item's `consequence`, `selectedCount`, +`eligibleCount`, `limited`, and `expiresAt`. At most 100 eligible items are frozen for five minutes. +If limited, explain which exact subset will run; never silently process subsequent batches. + +Pass `--yes` only when the user's authorization covers the frozen items and their consequences. +If that effect is already authorized, proceed without another confirmation. Execution starts no +worker and never claims newly eligible items outside the frozen selection. Each candidate's +reviewed state is revalidated under the same mutation lock as individual actions. + +Read stdout JSON even on exit 6 (partial result) or 130 (cancellation). Report `result.items` as +applied, skipped, failed, or unprocessed; explain `code` and `mutationMayHaveApplied`. A systemic +failure stops the rest without rolling back applied items. Repeating execute returns the stored +result; it does not retry skipped, failed, or unprocessed items. After a process interruption, +`batch show` recovers the journal and flags an in-flight item as uncertain. Inspect that item's +current state before proposing a new preview. Do not recreate and replay the whole batch blindly. + +A missing/expired preview requires a fresh review. Previews/results belong to the same local +configuration and cache directory across processes and restarts; do not copy or edit the files, +or treat a preview ID as authority to execute. Use `workers --item --json` for pickup evidence +after a successful action when the user's request needs it. + ## Workers and pickup prospects Run `wrighty workers --json` for worker discovery alone. It reads the configuration-scoped local diff --git a/src/Highbyte.Wrighty.Cli/CliApplication.cs b/src/Highbyte.Wrighty.Cli/CliApplication.cs index 1faedbad..03c9aab8 100644 --- a/src/Highbyte.Wrighty.Cli/CliApplication.cs +++ b/src/Highbyte.Wrighty.Cli/CliApplication.cs @@ -214,6 +214,7 @@ private RootCommand BuildRootCommand() root.Subcommands.Add(BuildWorkersCommand()); root.Subcommands.Add(BuildGetCommand()); root.Subcommands.Add(BuildActionsCommand()); + root.Subcommands.Add(BuildBatchCommand()); root.Subcommands.Add(BuildContextCommand()); root.Subcommands.Add(BuildApproveCommand()); root.Subcommands.Add(BuildApprovalWorkflowCommand()); diff --git a/src/Highbyte.Wrighty.Cli/CliBatchCommands.cs b/src/Highbyte.Wrighty.Cli/CliBatchCommands.cs new file mode 100644 index 00000000..85ca1ee4 --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/CliBatchCommands.cs @@ -0,0 +1,106 @@ +using System.CommandLine; +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.Cli; + +public sealed partial class CliApplication +{ + private Command BuildBatchCommand() + { + var command = new Command("batch", "Preview, execute, or inspect a frozen workflow batch"); + command.Subcommands.Add(BuildBatchPreviewCommand()); + command.Subcommands.Add(BuildBatchRecordCommand(execute: false)); + command.Subcommands.Add(BuildBatchRecordCommand(execute: true)); + return command; + } + + private WorkflowBatchStore BatchStore => new(storageLocations.WorkflowBatchesRoot); + + private Command BuildBatchPreviewCommand() + { + var action = new Argument("action-name"); + var ids = new Option("--id") { Description = "Explicit item ID; repeat to select several items." }; + var status = new Option("--status") { Description = "Select active items in this workflow status." }; + var fields = FieldOption("Filter the status selection by name=value; repeat for AND semantics."); + var json = JsonOption(); + var command = new Command("preview", "Freeze up to 100 eligible items for five minutes; no items are changed"); + command.Arguments.Add(action); + command.Options.Add(ids); + command.Options.Add(status); + command.Options.Add(fields); + command.Options.Add(json); + command.SetAction((parsed, token) => ExecuteAsync(parsed.GetValue(json), async config => + { + WorkflowActionService.EnsureSupported(parsed.GetValue(action)!); + var selection = await SelectBatchItemsAsync(config, parsed.GetValue(ids) ?? [], + parsed.GetValue(status), parsed.GetValue(fields) ?? [], token); + var preview = await new WorkflowBatchService(tracker, BatchStore).PreviewAsync( + config, parsed.GetValue(action)!, selection, token); + await writer.WriteBatchAsync(new(preview, "preview", []), parsed.GetValue(json)); + }, token)); + return command; + } + + private async Task> SelectBatchItemsAsync(TrackerConfig config, + string[] ids, string? status, string[] fields, CancellationToken cancellationToken) + { + if (tracker.Backend(config) is not IWorkflowActionBackend) + throw new TrackerException("NOT_SUPPORTED", "Batch workflow actions require Local Markdown.", 3); + if ((ids.Length == 0 && string.IsNullOrWhiteSpace(status)) || + (ids.Length > 0 && (status is not null || fields.Length > 0))) + throw new TrackerException("ARGUMENT_INVALID", "Select --id values or --status with optional --field filters.", 2); + if (ids.Length > 0) return ids.Select(id => tracker.ResolveId(config, id)).ToArray(); + var items = await tracker.ListAsync(config, new ListWorkItemsRequest(status, null, ArchiveScope.Active, + ParseFields(fields, allowDeletion: false).ToDictionary(pair => pair.Key, pair => pair.Value!, StringComparer.Ordinal)), + cancellationToken); + return items.Select(item => item.Id).ToArray(); + } + + private Command BuildBatchRecordCommand(bool execute) + { + var id = new Argument("preview-id"); + var yes = new Option("--yes") { Description = "Authorize exactly the persisted preview's items and action." }; + var json = JsonOption(); + var command = new Command(execute ? "execute" : "show", execute + ? "Execute a reviewed preview once, or return its recorded result" + : "Inspect a preview or its recorded result without executing items"); + command.Arguments.Add(id); + command.Options.Add(json); + if (execute) command.Options.Add(yes); + command.SetAction((parsed, token) => RunBatchRecordAsync( + new(parsed.GetValue(id)!, execute, parsed.GetValue(yes), parsed.GetValue(json)), token)); + return command; + } + + private sealed record BatchRecordRequest(string Id, bool Execute, bool Yes, bool Json); + + private async Task RunBatchRecordAsync(BatchRecordRequest request, CancellationToken token) + { + var resultExit = 0; + var exit = await ExecuteAsync(request.Json, async config => + { + var store = BatchStore; + var record = await store.ReadAsync(config, request.Id, token); + if (request.Execute) + { + if (!request.Yes) + throw new TrackerException("BATCH_CONFIRMATION_REQUIRED", + "Review batch show output, then pass --yes to authorize exactly that preview.", 2); + record = await new WorkflowBatchService(tracker, store).ExecuteAsync(config, + request.Id, ct => configLoader.LoadAsync(workingDirectory, ct), token); + resultExit = BatchExitCode(record); + } + await writer.WriteBatchAsync(record, request.Json); + }, token); + return exit == 0 ? resultExit : exit; + } + + private static int BatchExitCode(WorkflowBatchRecord record) + { + if (record.StopCode == "BATCH_CANCELLED") return 130; + return record.HasIssues ? 6 : 0; + } +} diff --git a/src/Highbyte.Wrighty.Cli/Output/BatchOutput.cs b/src/Highbyte.Wrighty.Cli/Output/BatchOutput.cs new file mode 100644 index 00000000..cdda8646 --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/Output/BatchOutput.cs @@ -0,0 +1,31 @@ +using Highbyte.Wrighty.Actions; + +namespace Highbyte.Wrighty.Cli.Output; + +public sealed partial class OutputWriter +{ + public async Task WriteBatchAsync(WorkflowBatchRecord record, bool json) + { + if (json) + { + await WriteJsonAsync(new { schemaVersion = 1, result = record }); + return; + } + var preview = record.Preview; + await output.WriteLineAsync($"Batch {preview.Id}: {preview.Action}; {record.State}."); + await output.WriteLineAsync($"Selected {preview.SelectedCount}; eligible {preview.EligibleCount}; frozen {preview.Candidates.Count}. " + + $"Preview expires {preview.ExpiresAt:O}. No worker is started."); + foreach (var candidate in preview.Candidates) + { + await output.WriteLineAsync($" {candidate.Id}: {candidate.Title}"); + await output.WriteLineAsync($" {candidate.Consequence}"); + } + if (record.State == "preview") + await output.WriteLineAsync($"After review: wrighty batch execute {preview.Id} --yes"); + foreach (var item in record.Items) + await output.WriteLineAsync($" {item.Id}: {item.Outcome}{(item.Code is null ? "" : $" ({item.Code})")}" + + (item.MutationMayHaveApplied ? "; mutation may have applied — inspect before retrying." : ".")); + if (record.StopCode is not null) + await output.WriteLineAsync($"Stopped: {record.StopCode}. Remaining items were not processed."); + } +} diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatch.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatch.cs new file mode 100644 index 00000000..545e3283 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatch.cs @@ -0,0 +1,49 @@ +using System.Security.Cryptography; +using System.Text.Json; +using Highbyte.Wrighty.Configuration; + +namespace Highbyte.Wrighty.Actions; + +public sealed record WorkflowBatchCandidate( + string Id, string Title, string StateVersion, WorkflowActionState Before, string Consequence); + +public sealed record WorkflowBatchPreview( + string Id, string Action, string ConfigurationVersion, DateTimeOffset CreatedAt, + IReadOnlyList Candidates, int SelectedCount, int EligibleCount) +{ + public DateTimeOffset ExpiresAt => CreatedAt + WorkflowBatchPolicy.PreviewLifetime; + public bool Limited => EligibleCount > Candidates.Count; + public bool StartsWorker { get; } = false; +} + +public sealed record WorkflowBatchItemResult( + string Id, string Outcome, string? Code = null, WorkflowActionResult? Applied = null, + bool MutationMayHaveApplied = false); + +public sealed record WorkflowBatchRecord( + WorkflowBatchPreview Preview, string State, IReadOnlyList Items, + string? ActiveItemId = null, string? StopCode = null) +{ + public bool HasIssues => StopCode is not null || Items.Any(item => item.Outcome != "applied"); +} + +public static class WorkflowBatchPolicy +{ + public const int MaximumCandidates = 100; + public static readonly TimeSpan PreviewLifetime = TimeSpan.FromMinutes(5); + + public static string ConfigurationVersion(TrackerConfig config) => + Convert.ToHexString(SHA256.HashData(JsonSerializer.SerializeToUtf8Bytes(new + { + config, config.SourcePath, config.SourceRevision + }))); + + public static bool IsItemConflict(string code) => code is + "BATCH_ITEM_INELIGIBLE" or "WORK_ITEM_NOT_FOUND" or "WORK_ITEM_ARCHIVED" or "ITEM_ARCHIVED" or + "CLAIM_NOT_OWNER" or "CLAIM_REQUIRED" or "CLAIM_HELD" or "CLAIM_HELD_BY_LOCAL_CLAIMANT" or + "CLAIM_STALE" or "CLAIM_TOKEN_REQUIRED" or "UPDATE_CONFLICT" or "WEB_CLAIM_GENERATION_STALE" or + "WORKER_ITEM_INELIGIBLE" or "ACTION_STATE_CHANGED" or "WORKFLOW_STATE_INVALID" or + "WORKER_RECOVERY_PENDING" or "WORKER_ITEM_NOT_PAUSED" or "RESUME_SESSION_CHANGED" or + "SESSION_LAUNCH_NOT_ALLOWED" or "RESUME_ADDRESS_UNAVAILABLE" or "RESUME_ADDRESS_NOT_LOCAL" or + "RESUME_WORKTREE_ABSENT" or "STATUS_MOVE_NOT_ALLOWED"; +} diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs new file mode 100644 index 00000000..9f5de126 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs @@ -0,0 +1,85 @@ +using Highbyte.Wrighty.Errors; + +namespace Highbyte.Wrighty.Actions; + +/// Storage hooks run outside mutation error handling: a failed journal write must stop execution. +public interface IWorkflowBatchProgress +{ + void Starting(string id); + void Completed(WorkflowBatchItemResult result); +} + +public sealed record WorkflowBatchExecution(IReadOnlyList Items, string? StopCode); + +/// +/// The single batch loop for every surface. Callers supply their reviewed selection and per-item +/// revalidation; this owns sequencing, conflict handling, cancellation, and partial outcomes. +/// +public static class WorkflowBatchExecutor +{ + public static async Task ExecuteAsync( + IReadOnlyList itemIds, + Func> execute, + CancellationToken cancellationToken, + IWorkflowBatchProgress? progress = null, + Action? onFailure = null) + { + // Do not let changes to the caller's selection add work after execution has started. + var frozen = itemIds.ToArray(); + if (frozen.Length > WorkflowBatchPolicy.MaximumCandidates || + frozen.Any(string.IsNullOrWhiteSpace) || frozen.Distinct(StringComparer.Ordinal).Count() != frozen.Length) + throw new TrackerException("BATCH_SELECTION_INVALID", "A batch requires at most 100 distinct item IDs.", 2); + List items = []; + string? stopCode = null; + foreach (var id in frozen) + { + if (cancellationToken.IsCancellationRequested) + { + stopCode = "BATCH_CANCELLED"; + break; + } + progress?.Starting(id); + var item = await ApplyAsync(id, execute, cancellationToken, onFailure); + items.Add(item); + progress?.Completed(item); + if (item.Outcome != "failed") continue; + stopCode = item.Code; + break; + } + return Complete(frozen, items, stopCode); + } + + internal static WorkflowBatchExecution Complete(IReadOnlyList ids, + IReadOnlyList items, string? stopCode) + { + var processed = items.Select(item => item.Id).ToHashSet(StringComparer.Ordinal); + return new([.. items, .. ids.Where(id => !processed.Contains(id)) + .Select(id => new WorkflowBatchItemResult(id, "unprocessed", stopCode))], stopCode); + } + + private static async Task ApplyAsync(string id, + Func> execute, + CancellationToken cancellationToken, Action? onFailure) + { + try + { + return new(id, "applied", Applied: await execute(id, cancellationToken)); + } + catch (TrackerException exception) when (WorkflowBatchPolicy.IsItemConflict(exception.Code)) + { + return new(id, "skipped", exception.Code); + } + catch (Exception exception) + { + onFailure?.Invoke(exception); + var code = exception switch + { + TrackerException trackerError => trackerError.Code, + OperationCanceledException => "BATCH_CANCELLED", + _ => "BATCH_EXECUTION_FAILED" + }; + // A failure can follow a backend write. Only known pre-mutation refusals are definite. + return new(id, "failed", code, MutationMayHaveApplied: code != "BATCH_CONFIG_CHANGED"); + } + } +} diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchService.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchService.cs new file mode 100644 index 00000000..e2378f28 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchService.cs @@ -0,0 +1,48 @@ +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.Actions; + +public sealed class WorkflowBatchService(TrackerService tracker, WorkflowBatchStore store) +{ + public async Task PreviewAsync(TrackerConfig config, string action, + IReadOnlyList selection, CancellationToken cancellationToken) + { + WorkflowActionService.EnsureSupported(action); + if (tracker.Backend(config) is not IWorkflowActionBackend) + throw new TrackerException("NOT_SUPPORTED", "This backend does not support batch workflow execution.", 3); + var ids = selection.Distinct().OrderBy(id => id.Value, StringComparer.Ordinal).ToArray(); + List candidates = []; + var eligible = 0; + foreach (var id in ids) + { + try + { + var state = await tracker.GetOperationalAsync(config, id, cancellationToken); + var descriptor = WorkflowActionService.Select(config, state, action); + if (descriptor.Availability != "available") continue; + eligible++; + if (candidates.Count < WorkflowBatchPolicy.MaximumCandidates) + candidates.Add(new(id.Value, state.Item.Title, WorkflowActionService.Version(config, state), + WorkflowActionService.Describe(state), descriptor.Description)); + } + catch (TrackerException exception) when (WorkflowBatchPolicy.IsItemConflict(exception.Code)) + { + // A deleted or newly unavailable item is outside the eligible frozen set. + } + } + return await store.CreateAsync(config, action, candidates, ids.Length, eligible, cancellationToken); + } + + public Task ExecuteAsync(TrackerConfig config, string previewId, + Func> reload, CancellationToken cancellationToken) => + store.ExecuteAsync(config, previewId, async (preview, candidate, token) => + { + var currentConfig = await reload(token); + if (WorkflowBatchPolicy.ConfigurationVersion(currentConfig) != preview.ConfigurationVersion) + throw new TrackerException("BATCH_CONFIG_CHANGED", "Configuration changed; review a new batch.", 6); + return await new WorkflowActionService(tracker).ExecuteAsync(currentConfig, + new WorkItemId(candidate.Id), preview.Action, candidate.StateVersion, token); + }, cancellationToken); +} diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs new file mode 100644 index 00000000..bd6d5abb --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs @@ -0,0 +1,182 @@ +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.LocalMarkdown; + +namespace Highbyte.Wrighty.Actions; + +/// +/// Configuration-scoped, cross-process preview journal. Write-ahead item markers prevent +/// replay after interruption; a cache record never substitutes for fresh execution consent. +/// +public sealed class WorkflowBatchStore(string root, TimeProvider? timeProvider = null) +{ + private readonly TimeProvider clock = timeProvider ?? TimeProvider.System; + private const int MaximumEntries = 512; + private static readonly TimeSpan Retention = TimeSpan.FromHours(24); + + public async Task CreateAsync(TrackerConfig config, string action, + IReadOnlyList candidates, int selectedCount, int eligibleCount, + CancellationToken cancellationToken) + { + WorkflowActionService.EnsureSupported(action); + if (candidates.Count > WorkflowBatchPolicy.MaximumCandidates) + throw Error("BATCH_TOO_LARGE", "A batch may contain at most 100 items."); + var scope = Scope(config); + await using var gate = await LocalStoreLock.AcquireAsync(scope, cancellationToken); + Purge(scope); + if (Directory.EnumerateFiles(scope, "*.json").Take(MaximumEntries).Count() >= MaximumEntries) + throw Error("BATCH_STORE_FULL", "The batch cache is full. Wait for its 24-hour retention period."); + var preview = new WorkflowBatchPreview(RandomNumberGenerator.GetHexString(32).ToLowerInvariant(), + action, WorkflowBatchPolicy.ConfigurationVersion(config), clock.GetUtcNow(), + candidates.ToArray(), selectedCount, eligibleCount); + Save(Path.Combine(scope, preview.Id + ".json"), new(preview, "preview", [])); + return preview; + } + + public async Task ReadAsync(TrackerConfig config, string id, + CancellationToken cancellationToken) + { + var path = RecordPath(config, id); + await using var gate = await LocalStoreLock.AcquireAsync(Scope(config), cancellationToken); + return Recover(path, Load(path)); + } + + public async Task ExecuteAsync(TrackerConfig config, string id, + Func> execute, + CancellationToken cancellationToken) + { + var path = RecordPath(config, id); + await using var gate = await LocalStoreLock.AcquireAsync(Scope(config), cancellationToken); + var record = Recover(path, Load(path)); + if (record.State == "completed") return record; + if (clock.GetUtcNow() >= record.Preview.ExpiresAt) + throw Error("BATCH_EXPIRED", "The preview expired. Review a new batch."); + if (record.Preview.ConfigurationVersion != WorkflowBatchPolicy.ConfigurationVersion(config)) + throw Error("BATCH_CONFIG_CHANGED", "Configuration changed. Review a new batch."); + record = record with { State = "running" }; + Save(path, record); + var journal = new Journal(path, record); + var candidates = record.Preview.Candidates.ToDictionary(candidate => candidate.Id, StringComparer.Ordinal); + var execution = await WorkflowBatchExecutor.ExecuteAsync( + record.Preview.Candidates.Select(candidate => candidate.Id).ToArray(), + async (itemId, token) => await execute(record.Preview, candidates[itemId], token), + cancellationToken, journal); + record = record with { State = "completed", Items = execution.Items, StopCode = execution.StopCode }; + Save(path, record); + return record; + } + + private sealed class Journal(string path, WorkflowBatchRecord record) : IWorkflowBatchProgress + { + public void Starting(string id) + { + // Persist before the mutation so recovery cannot replay an ambiguous item. + record = record with { ActiveItemId = id }; + Save(path, record); + } + + public void Completed(WorkflowBatchItemResult item) + { + record = record with { Items = [.. record.Items, item], ActiveItemId = null, + StopCode = item.Outcome == "failed" ? item.Code : null }; + Save(path, record); + } + } + + private WorkflowBatchRecord Recover(string path, WorkflowBatchRecord record) + { + if (record.State != "running") return record; + if (record.ActiveItemId is { } id) + record = record with { Items = [.. record.Items, + new(id, "failed", "BATCH_INTERRUPTED", MutationMayHaveApplied: true)] }; + record = Complete(record with { ActiveItemId = null, StopCode = "BATCH_INTERRUPTED" }); + Save(path, record); + return record; + } + + private static WorkflowBatchRecord Complete(WorkflowBatchRecord record) + { + var execution = WorkflowBatchExecutor.Complete( + record.Preview.Candidates.Select(candidate => candidate.Id).ToArray(), record.Items, record.StopCode); + return record with { State = "completed", ActiveItemId = null, Items = execution.Items }; + } + + private void Purge(string scope) + { + var cutoff = clock.GetUtcNow() - Retention; + foreach (var path in Directory.EnumerateFiles(scope, "*.json")) + if (File.GetLastWriteTimeUtc(path) < cutoff.UtcDateTime) File.Delete(path); + } + + private static TrackerException Error(string code, string message) => new(code, message, 6); + + private string Scope(TrackerConfig config) + { + if (string.IsNullOrWhiteSpace(config.SourcePath)) + throw Error("BATCH_SCOPE_REQUIRED", "Batch operations require a saved repository configuration."); + var source = Path.GetFullPath(config.SourcePath); + if (OperatingSystem.IsWindows()) source = source.ToUpperInvariant(); + return Path.Combine(root, Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(source)))); + } + + private string RecordPath(TrackerConfig config, string id) + { + if (id.Length != 32 || !id.All(char.IsAsciiHexDigit)) + throw Error("BATCH_UNKNOWN", "The batch preview ID is invalid or unavailable."); + return Path.Combine(Scope(config), id.ToLowerInvariant() + ".json"); + } + + private static WorkflowBatchRecord Load(string path) + { + if (!File.Exists(path)) throw Error("BATCH_UNKNOWN", "This batch is unavailable. Review a new batch."); + try + { + if (new FileInfo(path).Length > 2 * 1024 * 1024) throw new JsonException(); + var record = JsonSerializer.Deserialize(File.ReadAllText(path)); + if (record?.Preview?.Candidates is null || record.Items is null || + record.State is not ("preview" or "running" or "completed") || + record.Preview.Id != Path.GetFileNameWithoutExtension(path) || + record.Preview.Candidates.Count > WorkflowBatchPolicy.MaximumCandidates || + !ValidCandidates(record) || + record.Preview.Candidates.Select(item => item.Id).Distinct().Count() != record.Preview.Candidates.Count) + throw new JsonException(); + WorkflowActionService.EnsureSupported(record.Preview.Action); + return record; + } + catch (JsonException) + { + throw Error("BATCH_RECORD_INVALID", "The saved batch is invalid. Inspect item state before creating another preview."); + } + } + + private static bool ValidCandidates(WorkflowBatchRecord record) + { + var preview = record.Preview; + if (preview.CreatedAt == default || preview.EligibleCount < preview.Candidates.Count || + preview.SelectedCount < preview.EligibleCount || string.IsNullOrWhiteSpace(preview.ConfigurationVersion)) + return false; + if (preview.Candidates.Any(item => item is null || string.IsNullOrWhiteSpace(item.Id) || + item.StateVersion is not { Length: 64 } || !item.StateVersion.All(char.IsAsciiHexDigit) || + item.Before is null || item.Consequence is null || item.Title is null)) + return false; + var ids = preview.Candidates.Select(item => item.Id).ToHashSet(StringComparer.Ordinal); + if (record.Items.Any(item => item is null || !ids.Contains(item.Id)) || + record.Items.Select(item => item.Id).Distinct().Count() != record.Items.Count) + return false; + return record.State != "preview" || (record.Items.Count == 0 && record.ActiveItemId is null); + } + + private static void Save(string path, WorkflowBatchRecord record) + { + var temporary = path + ".tmp"; + using (var stream = new FileStream(temporary, FileMode.Create, FileAccess.Write, FileShare.None)) + { + JsonSerializer.Serialize(stream, record); + stream.Flush(flushToDisk: true); + } + File.Move(temporary, path, overwrite: true); + } +} diff --git a/src/Highbyte.Wrighty.Core/Caching/CachePaths.cs b/src/Highbyte.Wrighty.Core/Caching/CachePaths.cs index 57d59419..e48972d7 100644 --- a/src/Highbyte.Wrighty.Core/Caching/CachePaths.cs +++ b/src/Highbyte.Wrighty.Core/Caching/CachePaths.cs @@ -32,6 +32,8 @@ public CachePaths(string? overrideRoot = null) public string ProviderCapacityLockPath => Path.Combine(Root, "provider-capacity-v1.lock"); + public string WorkflowBatchesRoot => Path.Combine(Root, "workflow-batches-v1"); + public string WorkerInstancesRoot => Path.Combine(Root, "worker-instances-v1"); diff --git a/src/Highbyte.Wrighty.Core/Storage/StorageLocations.cs b/src/Highbyte.Wrighty.Core/Storage/StorageLocations.cs index ed51b019..9a8c6a7d 100644 --- a/src/Highbyte.Wrighty.Core/Storage/StorageLocations.cs +++ b/src/Highbyte.Wrighty.Core/Storage/StorageLocations.cs @@ -64,6 +64,8 @@ public sealed record StorageLocationDescriptor( /// public sealed class StorageLocationCatalog(CachePaths cachePaths) { + public string WorkflowBatchesRoot => cachePaths.WorkflowBatchesRoot; + private const string GitHubBackend = "github"; private const string LocalMarkdownBackend = "local-markdown"; private const string LocalMarkdownPathSource = "localMarkdown.path"; @@ -172,6 +174,14 @@ public IReadOnlyList Describe( new("all", cachePaths.RootSource), sensitive: false, "Cross-process lock protecting provider-capacity updates."), + Directory( + "cache.workflow-batches", + "Workflow batch previews and results", + Full(cachePaths.WorkflowBatchesRoot), + StorageLifecycle.RuntimeState, + new("local-markdown", cachePaths.RootSource), + sensitive: true, + "Configuration-scoped previews and execution journals; deleting them loses replay and interruption evidence."), Directory( "cache.worker-instances", "Worker-instance registry", diff --git a/src/Highbyte.Wrighty.Web/BoardBatchExecution.cs b/src/Highbyte.Wrighty.Web/BoardBatchExecution.cs new file mode 100644 index 00000000..7f100ff5 --- /dev/null +++ b/src/Highbyte.Wrighty.Web/BoardBatchExecution.cs @@ -0,0 +1,34 @@ +using Highbyte.Wrighty.Actions; + +namespace Highbyte.Wrighty.Web; + +/// Adapts Board display data and its accepted-request lifetime to the shared Core loop. +public static class BoardBatchExecution +{ + public static async Task ExecuteAsync(BoardBatchIntent intent, + Func> execute, + Action? onFailure = null) + { + var candidates = intent.Candidates.ToDictionary(candidate => candidate.Id, StringComparer.Ordinal); + // A browser disconnect does not cancel an already accepted bounded operation. + var result = await WorkflowBatchExecutor.ExecuteAsync( + intent.Candidates.Select(candidate => candidate.Id).ToArray(), + (id, token) => execute(candidates[id], token), CancellationToken.None, onFailure: onFailure); + return new(intent.Id, intent.Action, DateTimeOffset.UtcNow, + result.Items.Select(item => new BoardBatchItemResult(item.Id, candidates[item.Id].DisplayId, + item.Outcome == "applied", item.Outcome == "skipped", Reason(item), + item.Outcome == "unprocessed", item.MutationMayHaveApplied)).ToArray(), + result.StopCode is null ? null : $"The batch stopped because Wrighty could not continue safely ({result.StopCode})."); + } + + private static string? Reason(WorkflowBatchItemResult item) => item.Outcome switch + { + "applied" => null, + "unprocessed" => "Not processed because the batch stopped.", + "skipped" when item.Code == "BATCH_ITEM_INELIGIBLE" => "No longer eligible for this action.", + "skipped" => "The item changed before Wrighty could apply the action.", + _ when item.MutationMayHaveApplied => + $"The action may have applied ({item.Code}). Inspect this item before retrying.", + _ => $"Wrighty could not complete this item ({item.Code})." + }; +} diff --git a/src/Highbyte.Wrighty.Web/BoardBatchStore.cs b/src/Highbyte.Wrighty.Web/BoardBatchStore.cs index 67c2f8b1..d21dbd09 100644 --- a/src/Highbyte.Wrighty.Web/BoardBatchStore.cs +++ b/src/Highbyte.Wrighty.Web/BoardBatchStore.cs @@ -31,7 +31,8 @@ public sealed record BoardBatchItemResult( bool Succeeded, bool Skipped, string? Reason = null, - bool Aborted = false); + bool Aborted = false, + bool MutationMayHaveApplied = false); public sealed record BoardBatchResult( string IntentId, @@ -63,9 +64,9 @@ public sealed class BoardBatchStore( TimeSpan? intentLifetime = null, int maximumEntries = 512) { - public const int MaximumCandidates = 100; + public const int MaximumCandidates = Actions.WorkflowBatchPolicy.MaximumCandidates; private readonly TimeProvider clock = timeProvider ?? TimeProvider.System; - private readonly TimeSpan lifetime = intentLifetime ?? TimeSpan.FromMinutes(5); + private readonly TimeSpan lifetime = intentLifetime ?? Actions.WorkflowBatchPolicy.PreviewLifetime; private readonly ConcurrentDictionary entries = new(StringComparer.Ordinal); private readonly object latestLock = new(); private BoardBatchResult? latestResult; diff --git a/src/Highbyte.Wrighty.Web/Pages/Index.cshtml.cs b/src/Highbyte.Wrighty.Web/Pages/Index.cshtml.cs index b95a518a..7657979d 100644 --- a/src/Highbyte.Wrighty.Web/Pages/Index.cshtml.cs +++ b/src/Highbyte.Wrighty.Web/Pages/Index.cshtml.cs @@ -3633,92 +3633,19 @@ public IActionResult OnPostDismissBoardBatch(string? intentId) return new NoContentResult(); } - private async Task ExecuteBoardBatchAsync(BoardBatchIntent intent) - { - List results = []; - string? abortReason = null; - foreach (var candidate in intent.Candidates) - { - try - { - if (!string.Equals( - intent.ConfigurationRevision, - state.ActiveConfigurationRevision ?? string.Empty, - StringComparison.Ordinal)) - { - throw new TrackerException( - "BOARD_BATCH_CONFIG_CHANGED", - "Wrighty's configuration changed while the batch was running.", - 6); - } - if (!await BoardBatchCandidateStillEligibleAsync( - candidate.Id, - intent.Action, - CancellationToken.None)) - { - results.Add(new BoardBatchItemResult( - candidate.Id, - candidate.DisplayId, - Succeeded: false, - Skipped: true, - "No longer eligible for this action.")); - continue; - } - - await ExecuteBoardBatchItemAsync( - candidate.Id, - intent.Action, - CancellationToken.None); - results.Add(new BoardBatchItemResult( - candidate.Id, - candidate.DisplayId, - Succeeded: true, - Skipped: false)); - } - catch (TrackerException exception) when (IsBoardBatchItemConflict(exception)) - { - results.Add(new BoardBatchItemResult( - candidate.Id, - candidate.DisplayId, - Succeeded: false, - Skipped: true, - "The item changed before Wrighty could apply the action.")); - } - catch (TrackerException exception) - { - WebDiagnostics.RetainFailure(HttpContext, exception.Code, exception); - results.Add(new BoardBatchItemResult( - candidate.Id, - candidate.DisplayId, - Succeeded: false, - Skipped: false, - $"Wrighty could not complete this item ({exception.Code}).")); - abortReason = $"The batch stopped because Wrighty could not continue safely ({exception.Code})."; - break; - } - } - - if (abortReason is not null) - { - var completedIds = results.Select(result => result.Id).ToHashSet(StringComparer.Ordinal); - results.AddRange(intent.Candidates - .Where(candidate => !completedIds.Contains(candidate.Id)) - .Select(candidate => new BoardBatchItemResult( - candidate.Id, - candidate.DisplayId, - Succeeded: false, - Skipped: false, - "Not processed because the batch stopped.", - Aborted: true))); - } - - return new BoardBatchResult( - intent.Id, - intent.Action, - DateTimeOffset.UtcNow, - results, - abortReason); - } + private Task ExecuteBoardBatchAsync(BoardBatchIntent intent) => + BoardBatchExecution.ExecuteAsync(intent, async (candidate, token) => + { + if (!string.Equals(intent.ConfigurationRevision, + state.ActiveConfigurationRevision ?? string.Empty, StringComparison.Ordinal)) + throw new TrackerException("BATCH_CONFIG_CHANGED", + "Wrighty's configuration changed while the batch was running.", 6); + if (!await BoardBatchCandidateStillEligibleAsync(candidate.Id, intent.Action, token)) + throw new TrackerException("BATCH_ITEM_INELIGIBLE", "No longer eligible for this action.", 6); + await ExecuteBoardBatchItemAsync(candidate.Id, intent.Action, token); + return null; + }, exception => WebDiagnostics.RetainFailure(HttpContext, + exception is TrackerException trackerError ? trackerError.Code : "BATCH_EXECUTION_FAILED", exception)); private async Task BoardBatchCandidateStillEligibleAsync( string id, @@ -3754,13 +3681,6 @@ private Task ExecuteBoardBatchItemAsync( _ => ResumeSessionAsync(id, cancellationToken) }; - private static bool IsBoardBatchItemConflict(TrackerException exception) => - exception.Code is "WORK_ITEM_NOT_FOUND" or "WORK_ITEM_ARCHIVED" or - "CLAIM_NOT_OWNER" or "WORKER_ITEM_INELIGIBLE" or "ACTION_STATE_CHANGED" or - "WORKFLOW_STATE_INVALID" or "WORKER_RECOVERY_PENDING" or "ITEM_ARCHIVED" or - "RESUME_ADDRESS_NOT_LOCAL" || - Status(exception) == StatusCodes.Status409Conflict; - /// /// Moves a retained session's dispatch state to needs-attention. The item stays where it is /// and keeps its recorded session; only the marker changes. diff --git a/src/Highbyte.Wrighty.Web/Pages/Shared/_BoardBatchResult.cshtml b/src/Highbyte.Wrighty.Web/Pages/Shared/_BoardBatchResult.cshtml index a328bca1..aa5804d9 100644 --- a/src/Highbyte.Wrighty.Web/Pages/Shared/_BoardBatchResult.cshtml +++ b/src/Highbyte.Wrighty.Web/Pages/Shared/_BoardBatchResult.cshtml @@ -19,7 +19,7 @@ -

@Model.NotProcessedCount item@(Model.NotProcessedCount == 1 ? "" : "s") could not be processed. +

@Model.NotProcessedCount item@(Model.NotProcessedCount == 1 ? "" : "s") require@(Model.NotProcessedCount == 1 ? "s" : "") review. @Model.SucceededCount completed successfully; @Model.SkippedCount skipped, @Model.FailedCount failed@(aborted).

@if (Model.AbortReason is not null) @@ -29,7 +29,7 @@ @if (Model.Items.Any(item => !item.Succeeded)) {
- Review items not processed + Review affected items
    @foreach (var item in Model.Items.Where(item => !item.Succeeded).Take(100)) { diff --git a/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchExecutorTests.cs b/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchExecutorTests.cs new file mode 100644 index 00000000..5e2c622c --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchExecutorTests.cs @@ -0,0 +1,77 @@ +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.Errors; + +namespace Highbyte.Wrighty.UnitTests.Actions; + +public sealed class WorkflowBatchExecutorTests +{ + [Fact] + public async Task Cancellation_before_first_item_leaves_everything_unprocessed() + { + using var cancellation = new CancellationTokenSource(); + await cancellation.CancelAsync(); + var result = await WorkflowBatchExecutor.ExecuteAsync(["1", "2"], + (_, _) => throw new InvalidOperationException("must not execute"), cancellation.Token); + Assert.Equal("BATCH_CANCELLED", result.StopCode); + Assert.All(result.Items, item => Assert.Equal("unprocessed", item.Outcome)); + } + + [Fact] + public async Task Sequence_is_frozen_before_callbacks_and_progress_surrounds_each_mutation() + { + List selection = ["1", "2"]; + List events = []; + var result = await WorkflowBatchExecutor.ExecuteAsync(selection, (id, _) => + { + selection.Clear(); + selection.Add("unapproved"); + events.Add("apply " + id); + return Task.FromResult(null); + }, default, new Progress(events)); + Assert.Equal(["1", "2"], result.Items.Select(item => item.Id)); + Assert.Equal(["before 1", "apply 1", "after 1", "before 2", "apply 2", "after 2"], events); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task Persistence_failure_stops_before_any_more_mutations(bool beforeMutation) + { + var calls = 0; + await Assert.ThrowsAsync(() => WorkflowBatchExecutor.ExecuteAsync(["1", "2"], (_, _) => + { + calls++; + return Task.FromResult(null); + }, default, new FailedProgress(beforeMutation))); + Assert.Equal(beforeMutation ? 0 : 1, calls); + } + + [Theory] + [InlineData("duplicate")] + [InlineData("limit")] + [InlineData("empty-id")] + public async Task Invalid_selection_is_rejected_before_any_work(string scenario) + { + string[] ids = scenario switch + { + "duplicate" => ["1", "1"], + "limit" => Enumerable.Range(1, 101).Select(i => i.ToString()).ToArray(), + _ => [" "] + }; + var error = await Assert.ThrowsAsync(() => WorkflowBatchExecutor.ExecuteAsync(ids, + (_, _) => throw new InvalidOperationException("must not execute"), default)); + Assert.Equal("BATCH_SELECTION_INVALID", error.Code); + } + + private sealed class Progress(List events) : IWorkflowBatchProgress + { + public void Starting(string id) => events.Add("before " + id); + public void Completed(WorkflowBatchItemResult result) => events.Add("after " + result.Id); + } + + private sealed class FailedProgress(bool before) : IWorkflowBatchProgress + { + public void Starting(string id) { if (before) throw new IOException("Journal unavailable"); } + public void Completed(WorkflowBatchItemResult result) { if (!before) throw new IOException("Journal unavailable"); } + } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchStoreTests.cs b/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchStoreTests.cs new file mode 100644 index 00000000..b16140e6 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Actions/WorkflowBatchStoreTests.cs @@ -0,0 +1,192 @@ +using System.Text.Json; +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; + +namespace Highbyte.Wrighty.UnitTests.Actions; + +public sealed class WorkflowBatchStoreTests : IDisposable +{ + private readonly string root = Path.Combine(Path.GetTempPath(), "wrighty-batch-" + Guid.NewGuid().ToString("N")); + private readonly BatchClock clock = new(); + private TrackerConfig Config => new() { SourcePath = Path.Combine(root, ".wrighty.json"), Backend = "local-markdown" }; + private WorkflowBatchStore Store => new(root, clock); + private Task Preview(int count = 3) => Store.CreateAsync(Config, "queue", + Enumerable.Range(1, count).Select(i => new WorkflowBatchCandidate($"local:{i}", $"Item {i}", new string('A', 64), + new("Todo", "ready", false, null), "Authorizes automatic execution")).ToArray(), count, count, default); + private static Task Applied(WorkflowBatchPreview preview, WorkflowBatchCandidate candidate) => + Task.FromResult(new WorkflowActionResult(candidate.Id, preview.Action, "applied", DateTimeOffset.UtcNow, + candidate.StateVersion, candidate.Before, candidate.Before with { Status = "Worker queue", AutomaticExecutionAllowed = true })); + + [Fact] + public async Task Persisted_preview_survives_new_store_and_replay_does_not_execute_again() + { + var preview = await Preview(); + var calls = 0; + var first = await Store.ExecuteAsync(Config, preview.Id, (p, c, ct) => { calls++; return Applied(p,c); }, default); + Assert.Equal(3, calls); + Assert.All(first.Items, item => Assert.Equal("applied", item.Outcome)); + clock.Now += TimeSpan.FromMinutes(6); + var replay = await Store.ExecuteAsync(Config, preview.Id, (_, _, _) => throw new InvalidOperationException("replayed"), default); + Assert.Equal(JsonSerializer.Serialize(first), JsonSerializer.Serialize(replay)); + Assert.DoesNotContain("claimToken", await File.ReadAllTextAsync(RecordPath())); + } + + [Fact] + public async Task Concurrent_executors_apply_the_frozen_set_only_once() + { + var preview = await Preview(); + var calls = 0; + async Task Execute(WorkflowBatchPreview p, WorkflowBatchCandidate c, CancellationToken ct) + { + Interlocked.Increment(ref calls); + await Task.Delay(30, ct); + return await Applied(p, c); + } + var results = await Task.WhenAll(Store.ExecuteAsync(Config, preview.Id, Execute, default), + Store.ExecuteAsync(Config, preview.Id, Execute, default)); + Assert.Equal(3, calls); + Assert.All(results, result => Assert.Equal(3, result.Items.Count)); + } + + [Theory] + [InlineData("expired", "BATCH_EXPIRED")] + [InlineData("config", "BATCH_CONFIG_CHANGED")] + [InlineData("scope", "BATCH_UNKNOWN")] + [InlineData("id", "BATCH_UNKNOWN")] + public async Task Invalid_execution_refuses_before_any_item(string scenario, string code) + { + var preview = await Preview(); + var config = Config; + var id = preview.Id; + if (scenario == "expired") clock.Now += TimeSpan.FromMinutes(5); + if (scenario == "config") config = config with { DefaultPickFrom = "Other queue" }; + if (scenario == "scope") config = config with { SourcePath = Path.Combine(root, "other.json") }; + if (scenario == "id") id = "../../not-a-preview"; + var error = await Assert.ThrowsAsync(() => Store.ExecuteAsync(config, id, + (_, _, _) => throw new InvalidOperationException("must not execute"), default)); + Assert.Equal(code, error.Code); + } + + [Theory] + [InlineData("ACTION_STATE_CHANGED", "skipped", 3, false)] + [InlineData("CLAIM_HELD", "skipped", 3, false)] + [InlineData("BATCH_CONFIG_CHANGED", "failed", 2, false)] + [InlineData("STORE_BROKEN", "failed", 2, true)] + public async Task Conflict_continues_but_systemic_failure_stops_with_partial_results( + string code, string outcome, int callsExpected, bool ambiguous) + { + var preview = await Preview(); + var calls = 0; + var result = await Store.ExecuteAsync(Config, preview.Id, (p, c, ct) => + { + if (++calls == 2) throw new TrackerException(code, "private diagnostics"); + return Applied(p,c); + }, default); + Assert.Equal(callsExpected, calls); + Assert.Equal("applied", result.Items[0].Outcome); + Assert.Equal(outcome, result.Items[1].Outcome); + Assert.Equal(ambiguous, result.Items[1].MutationMayHaveApplied); + Assert.Equal(callsExpected == 2 ? "unprocessed" : "applied", result.Items[2].Outcome); + Assert.DoesNotContain("private diagnostics", JsonSerializer.Serialize(result)); + } + + [Fact] + public async Task Cancellation_between_items_preserves_applied_work_and_leaves_rest_unprocessed() + { + var preview = await Preview(); + using var cancel = new CancellationTokenSource(); + var result = await Store.ExecuteAsync(Config, preview.Id, (p, c, ct) => + { cancel.Cancel(); return Applied(p,c); }, cancel.Token); + Assert.Equal("BATCH_CANCELLED", result.StopCode); + Assert.Equal(["applied", "unprocessed", "unprocessed"], result.Items.Select(item => item.Outcome)); + } + + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task Interrupted_run_is_never_restarted_and_marks_only_inflight_item_uncertain(bool inFlight) + { + var preview = await Preview(); + var applied = new WorkflowBatchItemResult("local:1", "applied", Applied: await Applied(preview, preview.Candidates[0])); + var interrupted = new WorkflowBatchRecord(preview, "running", [applied], inFlight ? "local:2" : null); + await File.WriteAllTextAsync(RecordPath(), JsonSerializer.Serialize(interrupted)); + var result = await Store.ExecuteAsync(Config, preview.Id, (_, _, _) => throw new InvalidOperationException("must not resume"), default); + Assert.Equal("BATCH_INTERRUPTED", result.StopCode); + Assert.Equal("applied", result.Items[0].Outcome); + Assert.Equal(inFlight ? "failed" : "unprocessed", result.Items[1].Outcome); + Assert.Equal(inFlight, result.Items[1].MutationMayHaveApplied); + Assert.Equal("unprocessed", result.Items[2].Outcome); + } + + [Fact] + public async Task Cancellation_during_mutation_is_reported_as_uncertain() + { + var preview = await Preview(); + var result = await Store.ExecuteAsync(Config, preview.Id, (_, _, _) => throw new OperationCanceledException(), default); + Assert.Equal("BATCH_CANCELLED", result.StopCode); + Assert.True(result.Items[0].MutationMayHaveApplied); + Assert.All(result.Items.Skip(1), item => Assert.Equal("unprocessed", item.Outcome)); + } + + [Theory] + [InlineData("null-version")] + [InlineData("duplicate")] + [InlineData("invalid-json")] + public async Task Corrupt_records_fail_closed(string scenario) + { + var preview = await Preview(); + if (scenario == "invalid-json") await File.WriteAllTextAsync(RecordPath(), "{"); + else + { + var candidates = preview.Candidates.ToArray(); + candidates[1] = scenario == "duplicate" ? candidates[0] : candidates[1] with { StateVersion = null! }; + await File.WriteAllTextAsync(RecordPath(), JsonSerializer.Serialize(new WorkflowBatchRecord(preview with { Candidates = candidates }, "preview", []))); + } + var error = await Assert.ThrowsAsync(() => Store.ReadAsync(Config, preview.Id, default)); + Assert.Equal("BATCH_RECORD_INVALID", error.Code); + } + + [Fact] + public async Task Preview_limit_scope_requirement_and_retention_are_enforced() + { + var error = await Assert.ThrowsAsync(() => Preview(101)); + Assert.Equal("BATCH_TOO_LARGE", error.Code); + error = await Assert.ThrowsAsync(() => Store.CreateAsync(Config with { SourcePath = null }, "queue", [], 0, 0, default)); + Assert.Equal("BATCH_SCOPE_REQUIRED", error.Code); + await Preview(); + var old = RecordPath(); + File.SetLastWriteTimeUtc(old, clock.Now.AddDays(-2).UtcDateTime); + await Preview(); + Assert.False(File.Exists(old)); + } + + [Fact] + public async Task Failed_result_persistence_retains_uncertain_marker_and_prevents_replay() + { + var preview = await Preview(); + var calls = 0; + var path = RecordPath(); + var error = await Record.ExceptionAsync(() => Store.ExecuteAsync(Config, preview.Id, (p, c, ct) => + { + calls++; + Directory.CreateDirectory(path + ".tmp"); // Block the post-mutation journal replacement. + return Applied(p, c); + }, default)); + Assert.True(error is IOException or UnauthorizedAccessException); + Directory.Delete(path + ".tmp"); + var recovered = await Store.ExecuteAsync(Config, preview.Id, + (_, _, _) => throw new InvalidOperationException("must not replay"), default); + Assert.Equal(1, calls); + Assert.True(recovered.Items[0].MutationMayHaveApplied); + Assert.All(recovered.Items.Skip(1), item => Assert.Equal("unprocessed", item.Outcome)); + } + + private string RecordPath() => Directory.GetFiles(root, "*.json", SearchOption.AllDirectories).Single(); + public void Dispose() { if (Directory.Exists(root)) Directory.Delete(root, true); } + private sealed class BatchClock : TimeProvider + { + public DateTimeOffset Now { get; set; } = DateTimeOffset.UtcNow; + public override DateTimeOffset GetUtcNow() => Now; + } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Cli/BatchCommandTests.cs b/tests/Highbyte.Wrighty.UnitTests/Cli/BatchCommandTests.cs new file mode 100644 index 00000000..5eb41f70 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Cli/BatchCommandTests.cs @@ -0,0 +1,84 @@ +using System.Text.Json; +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.AgentContext; +using Highbyte.Wrighty.Backends; +using Highbyte.Wrighty.Cli; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.UnitTests.Cli; + +public sealed partial class CliApplicationTests +{ + [Fact] + public async Task Batch_cli_freezes_selection_skips_new_claims_and_replays_results() + { + var f = await WorkflowCliFixture(); + var backend = f.Tracker.Backend(f.Config); + var second = await backend.CreateAsync(f.Config, new CreateWorkItemOperation(new("Second", "Body", "Ideas", "P1"), false), default); + var output = new StringWriter(); + var error = new StringWriter(); + CliApplication App() => Application(new RecordingBackend(), new StringReader(""), output, error, + config: f.Config, trackerOverride: f.Tracker, storageLocationCatalog: new(new Highbyte.Wrighty.Caching.CachePaths(Path.GetDirectoryName(f.Config.SourcePath)))); + Assert.Equal(0, await App().InvokeAsync(["batch", "preview", "queue", "--status", "Ideas", "--json"])); + using var preview = JsonDocument.Parse(output.ToString()); + var frozen = preview.RootElement.GetProperty("result").GetProperty("preview"); + Assert.Equal(2, frozen.GetProperty("candidates").GetArrayLength()); + Assert.False(frozen.GetProperty("startsWorker").GetBoolean()); + var id = frozen.GetProperty("id").GetString()!; + var later = await backend.CreateAsync(f.Config, new CreateWorkItemOperation(new("Later", "Body", "Ideas", "P1"), false), default); + var claim = await backend.TryClaimAsync(f.Config, second.Id, AgentExecutionContext.Human, default); + output.GetStringBuilder().Clear(); + Assert.Equal(2, await App().InvokeAsync(["batch", "execute", id, "--json"])); + Assert.Equal("Ideas", (await f.Tracker.GetAsync(f.Config, f.Id, default)).Status); + Assert.Equal(6, await App().InvokeAsync(["batch", "execute", id, "--yes", "--json"])); + using var result = JsonDocument.Parse(output.ToString()); + var items = result.RootElement.GetProperty("result").GetProperty("items"); + Assert.Equal("applied", items[0].GetProperty("outcome").GetString()); + Assert.Equal("skipped", items[1].GetProperty("outcome").GetString()); + Assert.Equal("Ideas", (await f.Tracker.GetAsync(f.Config, later.Id, default)).Status); + Assert.Equal(claim.ClaimantId, (await backend.GetClaimOwnershipAsync(f.Config, second.Id, default)).ClaimantId); + var firstResult = output.ToString(); + output.GetStringBuilder().Clear(); + Assert.Equal(6, await App().InvokeAsync(["batch", "execute", id, "--yes", "--json"])); + Assert.Equal(firstResult, output.ToString()); + output.GetStringBuilder().Clear(); + Assert.Equal(0, await App().InvokeAsync(["batch", "show", id])); + Assert.Contains("skipped", output.ToString()); + } + + [Theory] + [InlineData("none")] + [InlineData("mixed")] + [InlineData("fields-with-ids")] + [InlineData("unsupported")] + public async Task Batch_cli_rejects_implicit_mixed_and_unsupported_selections(string scenario) + { + var f = await WorkflowCliFixture(); + List args = ["batch", "preview", scenario == "unsupported" ? "launch" : "queue", "--json"]; + if (scenario == "mixed") args.AddRange(["--id", "1", "--status", "Ideas"]); + if (scenario == "fields-with-ids") args.AddRange(["--id", "1", "--field", "x=y"]); + var error = new StringWriter(); + Assert.Equal(2, await Application(new RecordingBackend(), new StringReader(""), new StringWriter(), error, + config: f.Config, trackerOverride: f.Tracker).InvokeAsync(args.ToArray())); + Assert.Equal("Ideas", (await f.Tracker.GetAsync(f.Config, f.Id, default)).Status); + } + + [Fact] + public async Task Batch_cli_explicit_ids_deduplicate_and_support_send_back() + { + var f = await WorkflowCliFixture(); + await new WorkflowActionService(f.Tracker).ExecuteAsync(f.Config, f.Id, "queue", null, default); + var output = new StringWriter(); + var app = Application(new RecordingBackend(), new StringReader(""), output, + config: f.Config, trackerOverride: f.Tracker, storageLocationCatalog: new(new Highbyte.Wrighty.Caching.CachePaths(Path.GetDirectoryName(f.Config.SourcePath)))); + Assert.Equal(0, await app.InvokeAsync(["batch", "preview", "send-back", "--id", "1", "--id", "local:1", "--json"])); + using var json = JsonDocument.Parse(output.ToString()); + var preview = json.RootElement.GetProperty("result").GetProperty("preview"); + Assert.Equal(1, preview.GetProperty("selectedCount").GetInt32()); + output.GetStringBuilder().Clear(); + Assert.Equal(0, await app.InvokeAsync(["batch", "execute", preview.GetProperty("id").GetString()!, "--yes"])); + Assert.Equal("Ideas", (await f.Tracker.GetAsync(f.Config, f.Id, default)).Status); + Assert.False((await f.Tracker.GetAsync(f.Config, f.Id, default)).AutomaticExecutionAllowed); + } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Web/BatchExecutionParityTests.cs b/tests/Highbyte.Wrighty.UnitTests/Web/BatchExecutionParityTests.cs new file mode 100644 index 00000000..bc987eb5 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Web/BatchExecutionParityTests.cs @@ -0,0 +1,86 @@ +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Web; + +namespace Highbyte.Wrighty.UnitTests.Web; + +public sealed class BatchExecutionParityTests : IDisposable +{ + private readonly string root = Path.Combine(Path.GetTempPath(), "wrighty-batch-parity-" + Guid.NewGuid().ToString("N")); + + [Theory] + [InlineData("CLAIM_HELD", "skipped", 3, false)] + [InlineData("ACTION_STATE_CHANGED", "skipped", 3, false)] + [InlineData("BATCH_ITEM_INELIGIBLE", "skipped", 3, false)] + [InlineData("BATCH_CONFIG_CHANGED", "failed", 2, false)] + [InlineData("STORE_BROKEN", "failed", 2, true)] + [InlineData("io", "failed", 2, true)] + [InlineData("cancel", "failed", 2, true)] + public async Task Web_and_persisted_batches_share_partial_outcomes_and_do_not_replay( + string scenario, string secondOutcome, int callCount, bool uncertain) + { + var config = new TrackerConfig { SourcePath = Path.Combine(root, ".wrighty.json") }; + var cli = new WorkflowBatchStore(root); + var candidates = Enumerable.Range(1, 3).Select(i => new WorkflowBatchCandidate($"local:{i}", $"Item {i}", + new string('A', 64), new("Todo", "ready", false, null), "Queue")).ToArray(); + var preview = await cli.CreateAsync(config, "queue", candidates, 3, 3, default); + var web = new BoardBatchStore(); + var intent = web.Create(BoardBatchAction.Queue, "revision", + candidates.Select(c => new BoardBatchCandidate(c.Id, c.Id, c.Title)).ToArray(), 3, 3); + var cliCalls = new List(); + var webCalls = new List(); + var logged = new List(); + var cliResult = await cli.ExecuteAsync(config, preview.Id, (_, c, _) => + { + cliCalls.Add(c.Id); + return Apply(c.Id, scenario); + }, default); + var webResult = await web.ExecuteAsync(intent.Id, "revision", i => BoardBatchExecution.ExecuteAsync(i, async (c, _) => + { + webCalls.Add(c.Id); + return await Apply(c.Id, scenario); + }, logged.Add)); + Assert.Equal(cliCalls, webCalls); + Assert.Equal(callCount, cliCalls.Count); + Assert.Equal(secondOutcome, cliResult.Items[1].Outcome); + Assert.Equal(cliResult.Items.Select(i => i.Outcome), webResult.Items.Select(Outcome)); + Assert.Equal(cliResult.Items.Select(i => i.MutationMayHaveApplied), webResult.Items.Select(i => i.MutationMayHaveApplied)); + Assert.Equal(uncertain, webResult.Items[1].MutationMayHaveApplied); + Assert.Equal(secondOutcome == "failed" ? 1 : 0, logged.Count); + if (uncertain) Assert.Contains("may have applied", webResult.Items[1].Reason); + if (callCount == 2) + { + Assert.Equal("unprocessed", cliResult.Items[2].Outcome); + Assert.Contains(cliResult.StopCode!, webResult.AbortReason); + } + // Both surface stores retain partial results, including ambiguous failures. + var replay = await cli.ExecuteAsync(config, preview.Id, + (_, _, _) => throw new InvalidOperationException("must not replay"), default); + Assert.Equal(cliResult.Items, replay.Items); + var webReplay = await web.ExecuteAsync(intent.Id, "revision", + _ => throw new InvalidOperationException("must not replay")); + Assert.Same(webResult, webReplay); + } + + private static string Outcome(BoardBatchItemResult item) + { + if (item.Succeeded) return "applied"; + if (item.Skipped) return "skipped"; + return item.Aborted ? "unprocessed" : "failed"; + } + + private static Task Apply(string id, string scenario) + { + if (id == "local:2") + { + if (scenario == "io") throw new IOException("private backend details"); + if (scenario == "cancel") throw new OperationCanceledException(); + throw new TrackerException(scenario, "private backend details"); + } + return Task.FromResult(new WorkflowActionResult(id, "queue", "applied", DateTimeOffset.UtcNow, + "version", new("Todo", "ready", false, null), new("Queue", "queued", true, null))); + } + + public void Dispose() { if (Directory.Exists(root)) Directory.Delete(root, true); } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Web/WrightyWebServerTests.cs b/tests/Highbyte.Wrighty.UnitTests/Web/WrightyWebServerTests.cs index 4eb32530..e7a59c64 100644 --- a/tests/Highbyte.Wrighty.UnitTests/Web/WrightyWebServerTests.cs +++ b/tests/Highbyte.Wrighty.UnitTests/Web/WrightyWebServerTests.cs @@ -2221,7 +2221,7 @@ public async Task Board_bulk_queue_freezes_filtered_candidates_revalidates_and_i var result = await (await client.SendAsync(resultRequest)).Content.ReadAsStringAsync(); Assert.Contains("Queue all warning", result); Assert.Contains("role=\"alert\"", result); - Assert.Contains("1 item could not be processed.", result); + Assert.Contains("1 item requires review.", result); Assert.Contains("1 completed successfully; 1 skipped", result); Assert.DoesNotContain("@if", result); Assert.Contains("No longer eligible for this action.", result); diff --git a/tests/Highbyte.Wrighty.UnitTests/Workers/WorkflowBatchExecutionTests.cs b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkflowBatchExecutionTests.cs new file mode 100644 index 00000000..af45780a --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkflowBatchExecutionTests.cs @@ -0,0 +1,70 @@ +using Highbyte.Wrighty.Actions; +using Highbyte.Wrighty.AgentContext; +using Highbyte.Wrighty.Backends; +using Highbyte.Wrighty.Claims; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.UnitTests.Workers; + +public sealed partial class LocalDispatchStateTests +{ + [Fact] + public async Task Batch_resume_uses_shared_session_fencing_and_preserves_recorded_session() + { + var (backend, config, id, oldHandle) = await CreatePausedItemAsync(); + var tracker = new TrackerService(new TrackerBackendRegistry([backend])); + var before = await tracker.GetOperationalAsync(config, id, default); + var store = new WorkflowBatchStore(Path.Combine(Path.GetDirectoryName(config.SourcePath)!, "batches")); + var service = new WorkflowBatchService(tracker, store); + var preview = await service.PreviewAsync(config, "resume", [id], default); + Assert.Single(preview.Candidates); + var result = await service.ExecuteAsync(config, preview.Id, _ => Task.FromResult(config), default); + Assert.Equal("applied", Assert.Single(result.Items).Outcome); + var after = await tracker.GetOperationalAsync(config, id, default); + Assert.Equal(DispatchStates.Queued, after.Item.DispatchState); + Assert.Equal(before.Session!.SessionId, after.Session!.SessionId); + await Assert.ThrowsAsync(() => backend.RenewClaimAsync(config, id, oldHandle, null, null, default)); + } + + [Fact] + public async Task Batch_revalidates_reviewed_content_and_configuration_between_items() + { + var f = await WorkflowFixture(); + var store = new WorkflowBatchStore(Path.Combine(Path.GetDirectoryName(f.Config.SourcePath)!, "batches")); + var service = new WorkflowBatchService(f.Tracker, store); + var preview = await service.PreviewAsync(f.Config, "queue", [f.Id], default); + var claim = await f.Backend.TryClaimAsync(f.Config, f.Id, AgentExecutionContext.Human, default); + var handle = new ClaimHandle(AgentExecutionContext.Human, claim.ClaimToken); + await f.Backend.UpdateAsync(f.Config, f.Id, new(new(Title: OptionalValue.From("Edited since review"), + Body: default, Status: default, Priority: default), false, ClaimHandle: handle), default); + await f.Backend.ReleaseAsync(f.Config, f.Id, handle, false, DispatchStateOnRelease.Preserve, default); + var result = await service.ExecuteAsync(f.Config, preview.Id, _ => Task.FromResult(f.Config), default); + Assert.Equal("ACTION_STATE_CHANGED", Assert.Single(result.Items).Code); + Assert.Equal("Ideas", (await f.Tracker.GetAsync(f.Config, f.Id, default)).Status); + preview = await service.PreviewAsync(f.Config, "queue", [f.Id], default); + result = await service.ExecuteAsync(f.Config, preview.Id, + _ => Task.FromResult(f.Config with { DefaultPickFrom = "Other queue" }), default); + Assert.Equal("BATCH_CONFIG_CHANGED", result.StopCode); + Assert.False(Assert.Single(result.Items).MutationMayHaveApplied); + } + + [Fact] + public async Task Batch_preview_counts_eligible_items_and_caps_the_frozen_set() + { + var f = await WorkflowFixture(); + var ids = new List { f.Id }; + for (var i = 0; i < 101; i++) + ids.Add((await f.Backend.CreateAsync(f.Config, + new CreateWorkItemOperation(new($"Batch {i}", "Body", "Ideas", "P1"), false), default)).Id); + await f.Actions.ExecuteAsync(f.Config, ids[1], "queue", null, default); + var store = new WorkflowBatchStore(Path.Combine(Path.GetDirectoryName(f.Config.SourcePath)!, "batches")); + var preview = await new WorkflowBatchService(f.Tracker, store).PreviewAsync(f.Config, "queue", ids, default); + Assert.Equal(102, preview.SelectedCount); + Assert.Equal(101, preview.EligibleCount); + Assert.Equal(100, preview.Candidates.Count); + Assert.True(preview.Limited); + Assert.DoesNotContain(preview.Candidates, candidate => candidate.Id == ids[1].Value); + Assert.Equal(preview.Candidates.OrderBy(item => item.Id, StringComparer.Ordinal), preview.Candidates); + } +} From 7a941a7903de7afefe1e82d462658fac15f3c51f Mon Sep 17 00:00:00 2001 From: Highbyte Date: Wed, 16 Sep 2026 13:30:54 +0200 Subject: [PATCH 2/2] Address batch execution quality findings --- .../Actions/WorkflowBatchExecutor.cs | 4 ++-- .../Actions/WorkflowBatchStore.cs | 17 +++++++++-------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs index 9f5de126..b5c9fba6 100644 --- a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchExecutor.cs @@ -39,7 +39,7 @@ public static async Task ExecuteAsync( break; } progress?.Starting(id); - var item = await ApplyAsync(id, execute, cancellationToken, onFailure); + var item = await ApplyAsync(id, execute, onFailure, cancellationToken); items.Add(item); progress?.Completed(item); if (item.Outcome != "failed") continue; @@ -59,7 +59,7 @@ internal static WorkflowBatchExecution Complete(IReadOnlyList ids, private static async Task ApplyAsync(string id, Func> execute, - CancellationToken cancellationToken, Action? onFailure) + Action? onFailure, CancellationToken cancellationToken) { try { diff --git a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs index bd6d5abb..ac3af03a 100644 --- a/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs +++ b/src/Highbyte.Wrighty.Core/Actions/WorkflowBatchStore.cs @@ -15,6 +15,7 @@ public sealed class WorkflowBatchStore(string root, TimeProvider? timeProvider = { private readonly TimeProvider clock = timeProvider ?? TimeProvider.System; private const int MaximumEntries = 512; + private const string CompletedState = "completed"; private static readonly TimeSpan Retention = TimeSpan.FromHours(24); public async Task CreateAsync(TrackerConfig config, string action, @@ -51,7 +52,7 @@ public async Task ExecuteAsync(TrackerConfig config, string var path = RecordPath(config, id); await using var gate = await LocalStoreLock.AcquireAsync(Scope(config), cancellationToken); var record = Recover(path, Load(path)); - if (record.State == "completed") return record; + if (record.State == CompletedState) return record; if (clock.GetUtcNow() >= record.Preview.ExpiresAt) throw Error("BATCH_EXPIRED", "The preview expired. Review a new batch."); if (record.Preview.ConfigurationVersion != WorkflowBatchPolicy.ConfigurationVersion(config)) @@ -64,7 +65,7 @@ record = record with { State = "running" }; record.Preview.Candidates.Select(candidate => candidate.Id).ToArray(), async (itemId, token) => await execute(record.Preview, candidates[itemId], token), cancellationToken, journal); - record = record with { State = "completed", Items = execution.Items, StopCode = execution.StopCode }; + record = record with { State = CompletedState, Items = execution.Items, StopCode = execution.StopCode }; Save(path, record); return record; } @@ -78,15 +79,15 @@ record = record with { ActiveItemId = id }; Save(path, record); } - public void Completed(WorkflowBatchItemResult item) + public void Completed(WorkflowBatchItemResult result) { - record = record with { Items = [.. record.Items, item], ActiveItemId = null, - StopCode = item.Outcome == "failed" ? item.Code : null }; + record = record with { Items = [.. record.Items, result], ActiveItemId = null, + StopCode = result.Outcome == "failed" ? result.Code : null }; Save(path, record); } } - private WorkflowBatchRecord Recover(string path, WorkflowBatchRecord record) + private static WorkflowBatchRecord Recover(string path, WorkflowBatchRecord record) { if (record.State != "running") return record; if (record.ActiveItemId is { } id) @@ -101,7 +102,7 @@ private static WorkflowBatchRecord Complete(WorkflowBatchRecord record) { var execution = WorkflowBatchExecutor.Complete( record.Preview.Candidates.Select(candidate => candidate.Id).ToArray(), record.Items, record.StopCode); - return record with { State = "completed", ActiveItemId = null, Items = execution.Items }; + return record with { State = CompletedState, ActiveItemId = null, Items = execution.Items }; } private void Purge(string scope) @@ -137,7 +138,7 @@ private static WorkflowBatchRecord Load(string path) if (new FileInfo(path).Length > 2 * 1024 * 1024) throw new JsonException(); var record = JsonSerializer.Deserialize(File.ReadAllText(path)); if (record?.Preview?.Candidates is null || record.Items is null || - record.State is not ("preview" or "running" or "completed") || + record.State is not ("preview" or "running" or CompletedState) || record.Preview.Id != Path.GetFileNameWithoutExtension(path) || record.Preview.Candidates.Count > WorkflowBatchPolicy.MaximumCandidates || !ValidCandidates(record) ||