diff --git a/docs/reference/README.md b/docs/reference/README.md index 53d651f5..9cd71d6e 100644 --- a/docs/reference/README.md +++ b/docs/reference/README.md @@ -34,3 +34,5 @@ first-run guide. These pages are the authority for options, guarantees, and edge Related: [workflow guide](../workflows.md), [item metadata](../item-metadata/README.md), and [design documents](../design/). + +- [Worker discovery](workers.md): scoped run inspection and item pickup assessment. diff --git a/docs/reference/agent-skills.md b/docs/reference/agent-skills.md index f36bea71..b39e98ff 100644 --- a/docs/reference/agent-skills.md +++ b/docs/reference/agent-skills.md @@ -179,3 +179,9 @@ only for unresolved decisions that materially affect the result. Explicit tracke incomplete, but the skill does not present them as ready or enable automatic processing until the same assessment passes. Fresh worker sessions independently assess the approved context they receive; the skill does not stamp items with a reusable “verified” marker. + +The bundled skill supports board overview using configured workflow order, operational triage, +shared action discovery, and scoped worker/pickup assessment. `list --json` keeps its `result` array +and adds `listing` metadata: status order, archive/filter/limit scope, returned count, and whether +the result may be truncated. See [worker discovery](workers.md). Update an installed skill through +the existing `wrighty skill update` command to receive these workflows. diff --git a/docs/reference/worker.md b/docs/reference/worker.md index e30f6578..7f6cf251 100644 --- a/docs/reference/worker.md +++ b/docs/reference/worker.md @@ -1436,3 +1436,6 @@ target agent is told why it has no history instead of silently assuming there wa These CLI surfaces are version-sensitive. Validate vendor upgrades in a throwaway repository before unattended use. + +Use [`wrighty workers`](workers.md) for focused local run discovery or `workers --item --json` +for advisory pickup prospects. Inspection alone does not start a worker. diff --git a/docs/reference/workers.md b/docs/reference/workers.md new file mode 100644 index 00000000..080869ed --- /dev/null +++ b/docs/reference/workers.md @@ -0,0 +1,70 @@ +# Worker discovery + +`wrighty workers` lists registered worker runs in the current local configuration scope. Plain +listing reads configuration and registry files and probes only registered PIDs and their start +identities. It does not contact the tracker, inspect retained worktrees, start an agent, or clean +up expired registry records. + +```sh +wrighty workers --json +wrighty workers --item local:42 --json +``` + +`--item` resolves the canonical item ID and adds advisory pickup assessment using tracker state +and the same selection/admission primitives as the worker. It can read local runtime, context, +provider-cache, and workspace evidence, but does not claim work, authorize the queue, prepare a +workspace, or run a paid provider probe. Execution still revalidates and atomically claims work. + +## JSON contract + +The versioned envelope has `schemaVersion: 1` and a `result` with: + +| Field | Meaning | +| --- | --- | +| `observedAt` | Start of this observation; the snapshot is not a reservation. | +| `scope`, `configurationPathHash` | Registered workers in this installation/configuration. | +| `coverage`, `detail` | `complete`, `incomplete`, or `unavailable` registry coverage and any explanation. | +| `configurationRevision` | Current repository configuration revision, when readable. | +| `itemId` | Canonical assessed item, when requested. | +| `localWorkers` | Individual run entries; an empty array does not prove global absence. | + +Each entry reuses the `instance` and `detail` projection from `status --json` and adds named +`liveness` (`Running`, `Stale`, `Unknown`), `origin` (`cli-process`, `web-hosted`, `unknown`), +`reportedState`, `intake`, `remainingItemAllowance`, `idleExpiresAt`, `configurationDrift`, and +optional `pickup`. The nested instance retains its existing enum representation; `status --json` +is unchanged apart from the additive scheduling/progress fields. + +`instance` includes run ID, PID/start identity, CLI/web-host origin, last heartbeat/state/item/agent, +startup configuration revision, and cooperative control capabilities. Several web-hosted runs may +share a PID; address a run by its run ID, never by PID alone. + +New registrations include `scheduling`: continuous/bounded/targeted mode, canonical target and +intent, effective source/active statuses, explicit/default agent, filters, workspace mode/repository, +item limit, idle/item timeout, profile, and dry-run mode. These values are captured at startup; +`invocationSummary` remains display text. No claim token is included. `progress` reports completed +item accounting and the current idle-period start from the worker loop. Missing legacy fields mean +unknown, not unrestricted eligibility. Remaining allowance excludes an active item; no configured +limit means unlimited, while missing progress means unknown. + +A `pickup` includes outcome (`could-pick-up`, `cannot-pick-up`, `unknown`), a stable reason code, +explanation, resolved agent when known, and flags for already processing or eligibility after the +current item. It considers intake/lifetime, target/limits, filters, workflow and execution policy, +claims, recorded sessions/dispatch, runtime/agent enablement, approved context, cached provider +state, workspace evidence, and configuration drift. Inconclusive checks yield unknown. + +A due retry remains subject to retained-session rules; discovery never bypasses its timer. +Already-processing reports the worker's last-reported activity and does not authorize competing +work. A possible pickup is not a promise about queue order or timing. Workspace and provider state +can change immediately afterwards. Reassess after queueing or resuming; this command does not +predict a proposed mutation's consequences. + +## Liveness and coverage + +Records use cached heartbeat metadata plus targeted OS process/start-identity checks. A missing +process, reused PID, or expired heartbeat is stale; denied identity inspection is unknown. Older +workers, failed registration, other users/configurations, and remote machines can be outside the +scope. An unreadable record makes coverage incomplete. Discovery does not broaden permissions or +scan process command lines/environments. Cooperative stop still requires fresh verified identity. + +`workers` is read-only. For launching see [worker.md](worker.md); for item action discovery see +[actions.md](actions.md). OS service installation and startup management are operator-managed. diff --git a/skills/wrighty/SKILL.md b/skills/wrighty/SKILL.md index 571db018..8f5f68f8 100644 --- a/skills/wrighty/SKILL.md +++ b/skills/wrighty/SKILL.md @@ -1,16 +1,21 @@ --- name: wrighty -description: Safely operate Wrighty through the `wrighty` CLI. Use only when the user explicitly asks to use Wrighty, the Wrighty CLI, or a Wrighty work-item ID, including requests to list, inspect, create, pick, claim, edit, move, finish, archive, or release tracked work. Do not trigger for generic tasks, GitHub issues, planning, backlogs, or project management that do not explicitly identify Wrighty. +description: Safely operate Wrighty through the `wrighty` CLI. Use only when the user explicitly asks to use Wrighty, the Wrighty CLI, or a Wrighty work-item ID, including board overview, worker discovery, pickup assessment, and requests to list, inspect, create, pick, claim, edit, move, finish, archive, or release tracked work. Do not trigger for generic tasks, GitHub issues, planning, backlogs, or project management that do not explicitly identify Wrighty. --- # 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, available actions, worker discovery, and item 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. + ## Workflow 1. Verify `wrighty` is callable. diff --git a/skills/wrighty/references/board-and-workers.md b/skills/wrighty/references/board-and-workers.md new file mode 100644 index 00000000..160e5abc --- /dev/null +++ b/skills/wrighty/references/board-and-workers.md @@ -0,0 +1,83 @@ +# Board and worker overview + +Use these workflows when the user asks to see their Wrighty board, triage blocked work, discover +workers, or assess whether a worker could pick up a named item. These are observations, not +permission to change items or launch processes. + +## Board by workflow status + +Run `wrighty list --json`. Its existing `result` array contains canonical IDs, workflow status, +priority, claims, retained sessions, and operational state. The additive `listing` block supplies: + +- `statusOrder` and `statusOrderSource`: use configured column order when supplied. If the source + is `unknown`, group by the returned status names and say configured ordering is unavailable. + Do not invent a standard Todo/In Progress/Done workflow. +- `archiveScope`, `statusFilter`, `fields`, and `limit`: describe the scope actually requested. +- `returnedCount`, `countScope`, and `completeness`: counts describe returned items. A + `possibly-truncated` listing cannot establish whole-board totals. + +For full-board counts, omit `--limit`, count the full returned array by workflow status, then +present a bounded selection with the total and omitted count. For a deliberately limited view, +use `--limit ` and label counts as that subset. Honor `--status`, `--archived`, and +`--include-archived` when requested. Never mix archived history into the default active board. +Show empty configured columns as zero only when the requested scope is complete and unfiltered. + +Keep workflow columns separate from operational state. For example, an active-work column may +contain an item that is paused, awaiting clarification, or retry-scheduled. Include canonical IDs +in concise rows so follow-up requests identify the intended items unambiguously. + +## Triage and available actions + +Use `wrighty status --json` for operational groups, and `wrighty get --json` to inspect a +specific blocker. Read `lastRun` and dispatch details before explaining what happened. +Retry-scheduled and handoff-queued work is deferred; do not classify it as awaiting clarification. + +Use `wrighty actions --json` for available actions; add `--all` when the user asks why an +alternative is unavailable. Use action names, reasons, recommendation, and execution metadata +from the response. A recommendation is advice, not execution authority. The current catalogue is +manual-only: `actions --exec` refuses execution. Queue/Send back/Resume descriptors currently +refer to the matching web Board operations; do not substitute a generic move or direct Markdown +edit to imitate an unavailable typed operation. + +## Workers and pickup prospects + +Run `wrighty workers --json` for worker discovery alone. It reads the configuration-scoped local +registry and checks the recorded PIDs and process-start identities; it does not contact the +tracker or inspect retained worktrees. Each run has a `runId`. Several web-hosted runs may share +one PID, so do not collapse them into one worker. + +The result includes observation time, scope/coverage, `localWorkers`, named `liveness`, `origin`, +`reportedState`, heartbeat, structured startup `instance.scheduling`, loop `instance.progress`, intake, remaining +allowance, and configuration drift. Report old or unreadable scheduling as unknown. Do not parse +`invocationSummary`, use current agent as the worker's only supported agent, or reconstruct its +startup settings from current repository configuration. + +For a named item, run `wrighty workers --item --json`. This additionally reads the tracker and +shared selection/admission evidence. For each run, read `pickup.outcome`, `code`, `message`, +`alreadyProcessing`, and `afterCurrentItem`: + +- `alreadyProcessing`: identify that run/session; do not launch competing work. +- `could-pick-up`: observed selection allows pickup, possibly after its current item. This is + neither a reservation nor a timing guarantee; another item may be selected first. +- `cannot-pick-up`: explain the reason, such as closed intake, exhausted allowance, a different + targeted item, filters, claim ownership, workflow state, or provider deferral. +- `unknown`: explain the missing evidence. Do not equate it with absence or launch a replacement + automatically. Older registrations, sandbox denial, configuration drift, and inaccessible + workspace evidence can produce this result. + +An empty complete scope means no registrations were observed in this local configuration. +Unregistered workers, other configurations/accounts, and remote workers may still exist. +Incomplete/unavailable coverage cannot establish even that scoped absence. A stale registration +is not evidence of open intake. Capability fields do not waive fresh identity checks for control. + +Assessment describes the item's current state, not its state after a proposed Queue/Resume action. +After an authorized mutation, re-read the item and assessment before describing pickup prospects. +Queueing and launching are separate permissions. With no suitable observed worker, explain the +reason and offer an appropriate bounded or continuous launch only if the user's intent calls for +it. A next-item `worker --once` does not target a named item; `worker --item ` does. + +Waiting must be bounded or explicitly hosted by the agent platform. Recheck only when asked to +wait or when confirming an authorized operation, and stop on meaningful progress, failure, or a +needed user decision. A conversation is not a guarantee of unattended monitoring. Worker-spawned +implementation sessions must not recursively launch workers. Never run a paid provider probe +merely to strengthen an assessment. diff --git a/src/Highbyte.Wrighty.Cli/CliApplication.cs b/src/Highbyte.Wrighty.Cli/CliApplication.cs index 1f34db24..1faedbad 100644 --- a/src/Highbyte.Wrighty.Cli/CliApplication.cs +++ b/src/Highbyte.Wrighty.Cli/CliApplication.cs @@ -211,6 +211,7 @@ private RootCommand BuildRootCommand() root.Subcommands.Add(BuildInitCommand()); root.Subcommands.Add(BuildListCommand()); root.Subcommands.Add(BuildStatusCommand()); + root.Subcommands.Add(BuildWorkersCommand()); root.Subcommands.Add(BuildGetCommand()); root.Subcommands.Add(BuildActionsCommand()); root.Subcommands.Add(BuildContextCommand()); @@ -2753,24 +2754,25 @@ private Command BuildListCommand() 2); } - var items = await tracker.ListOperationalAsync( - config, - new ListWorkItemsRequest( + var scope = ArchiveScope.Active; + if (parseResult.GetValue(archived)) + scope = ArchiveScope.Archived; + else if (parseResult.GetValue(includeArchived)) + scope = ArchiveScope.All; + var request = new ListWorkItemsRequest( parseResult.GetValue(status), parseResult.GetValue(limit), - parseResult.GetValue(archived) - ? ArchiveScope.Archived - : parseResult.GetValue(includeArchived) - ? ArchiveScope.All - : ArchiveScope.Active, + scope, ParseFields(parseResult.GetValue(fields), allowDeletion: false) - .ToDictionary(pair => pair.Key, pair => pair.Value!, StringComparer.Ordinal)), - cancellationToken); + .ToDictionary(pair => pair.Key, pair => pair.Value!, StringComparer.Ordinal)); + var items = await tracker.ListOperationalAsync(config, request, cancellationToken); + var listing = parseResult.GetValue(json) + ? await DescribeListingAsync(config, request, items.Count, cancellationToken) : null; await writer.WriteOperationalItemsAsync( items, parseResult.GetValue(compact), parseResult.GetValue(json), - id => tracker.FormatShort(config, id)); + id => tracker.FormatShort(config, id), listing); }, cancellationToken)); return command; diff --git a/src/Highbyte.Wrighty.Cli/CliListingContext.cs b/src/Highbyte.Wrighty.Cli/CliListingContext.cs new file mode 100644 index 00000000..fba3e663 --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/CliListingContext.cs @@ -0,0 +1,26 @@ +using Highbyte.Wrighty.Cli.Output; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.Cli; + +public sealed partial class CliApplication +{ + private async Task DescribeListingAsync(TrackerConfig config, + ListWorkItemsRequest request, int count, CancellationToken cancellationToken) + { + IReadOnlyList statuses; + try + { + statuses = await tracker.Backend(config).WorkflowStatusesAsync(config, cancellationToken); + } + catch (Exception exception) when (exception is TrackerException or IOException or UnauthorizedAccessException) + { + statuses = []; + } + return new(statuses, statuses.Count > 0 ? "configured" : "unknown", + request.ArchiveScope.ToString().ToLowerInvariant(), request.Status, request.Fields, + request.Limit, count, request.Limit is { } limit && count >= limit ? "possibly-truncated" : "complete"); + } +} diff --git a/src/Highbyte.Wrighty.Cli/CliWorkerDiscoveryCommands.cs b/src/Highbyte.Wrighty.Cli/CliWorkerDiscoveryCommands.cs new file mode 100644 index 00000000..a46a1429 --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/CliWorkerDiscoveryCommands.cs @@ -0,0 +1,45 @@ +using System.CommandLine; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Workers; + +namespace Highbyte.Wrighty.Cli; + +public sealed partial class CliApplication +{ + private Command BuildWorkersCommand() + { + var json = JsonOption(); + var item = new Option("--item") { Description = "Assess whether each observed worker could pick up this item." }; + var command = new Command("workers", "Inspect registered local workers without starting or controlling them"); + command.Options.Add(json); + command.Options.Add(item); + command.SetAction((parsed, cancellationToken) => ExecuteAsync(parsed.GetValue(json), + config => InspectWorkersAsync(config, parsed.GetValue(item), parsed.GetValue(json), cancellationToken), + cancellationToken)); + return command; + } + + private async Task InspectWorkersAsync(TrackerConfig config, string? item, bool json, + CancellationToken cancellationToken) + { + var configurationPath = config.SourcePath ?? Path.Combine(workingDirectory, TrackerConfigLoader.FileName); + var snapshot = await workerInstances.InspectAsync(configurationPath, cancellationToken); + var revision = await StatusConfigurationRevisionAsync(config, cancellationToken); + var state = item is null ? null : await tracker.GetOperationalAsync(config, + tracker.ResolveId(config, item), cancellationToken); + var workers = new List(); + foreach (var status in snapshot.Workers) + { + WorkerPickupAssessment? pickup = null; + if (state is not null) + pickup = workerService is null + ? new(status.Instance.RunId, state.Item.Id.Value, "unknown", "ASSESSMENT_UNAVAILABLE", + "Worker selection assessment is unavailable here.") + : await workerService.AssessPickupAsync(config, status, state, revision, + snapshot.ObservedAt, cancellationToken); + workers.Add(WorkerDiscoveryEntry.From(status, revision, pickup)); + } + await writer.WriteWorkersAsync(new(snapshot.ObservedAt, snapshot.ConfigurationPathHash, + snapshot.Coverage, snapshot.Detail, revision, state?.Item.Id.Value, workers), json); + } +} diff --git a/src/Highbyte.Wrighty.Cli/Output/OutputWriter.cs b/src/Highbyte.Wrighty.Cli/Output/OutputWriter.cs index a148385e..e675b966 100644 --- a/src/Highbyte.Wrighty.Cli/Output/OutputWriter.cs +++ b/src/Highbyte.Wrighty.Cli/Output/OutputWriter.cs @@ -82,7 +82,8 @@ public async Task WriteOperationalItemsAsync( IEnumerable items, bool compact, bool json, - Func formatShort) + Func formatShort, + WorkItemListingContext? listing = null) { var materialized = items.ToArray(); if (json) @@ -90,6 +91,7 @@ public async Task WriteOperationalItemsAsync( await WriteJsonAsync(new { schemaVersion = 1, + listing, result = materialized .Select(item => OperationalDto(item, formatShort)) .ToArray() diff --git a/src/Highbyte.Wrighty.Cli/Output/WorkItemListingContext.cs b/src/Highbyte.Wrighty.Cli/Output/WorkItemListingContext.cs new file mode 100644 index 00000000..295d8022 --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/Output/WorkItemListingContext.cs @@ -0,0 +1,14 @@ +namespace Highbyte.Wrighty.Cli.Output; + +public sealed record WorkItemListingContext( + IReadOnlyList StatusOrder, + string StatusOrderSource, + string ArchiveScope, + string? StatusFilter, + IReadOnlyDictionary? Fields, + int? Limit, + int ReturnedCount, + string Completeness) +{ + public string CountScope { get; } = "returned-items"; +} diff --git a/src/Highbyte.Wrighty.Cli/Output/WorkerDiscoveryOutput.cs b/src/Highbyte.Wrighty.Cli/Output/WorkerDiscoveryOutput.cs new file mode 100644 index 00000000..952afbbd --- /dev/null +++ b/src/Highbyte.Wrighty.Cli/Output/WorkerDiscoveryOutput.cs @@ -0,0 +1,45 @@ +using Highbyte.Wrighty.Workers; + +namespace Highbyte.Wrighty.Cli.Output; + +public sealed partial class OutputWriter +{ + public async Task WriteWorkersAsync(WorkerDiscovery discovery, bool json) + { + if (json) + { + await WriteJsonAsync(new { schemaVersion = 1, result = discovery }); + return; + } + await output.WriteLineAsync($"Registered local workers ({discovery.LocalWorkers.Count}); coverage: {discovery.Coverage}"); + await output.WriteLineAsync("Scope: this installation and configuration. Unregistered or remote workers may exist."); + if (discovery.Detail is not null) + await output.WriteLineAsync(discovery.Detail); + foreach (var worker in discovery.LocalWorkers) + await WriteWorkerDiscoveryEntryAsync(worker); + } + + private static string AllowanceLabel(WorkerDiscoveryEntry worker) + { + if (worker.RemainingItemAllowance is { } remaining) + return remaining.ToString(System.Globalization.CultureInfo.InvariantCulture); + return worker.Instance.Scheduling is { ItemLimit: null } scheduling && scheduling.IsUsable() + ? "unlimited" : "unknown"; + } + + private async Task WriteWorkerDiscoveryEntryAsync(WorkerDiscoveryEntry worker) + { + var instance = worker.Instance; + await output.WriteLineAsync($" {instance.RunId} pid {instance.ProcessId} {instance.HostKind} {worker.Liveness} intake: {worker.Intake}"); + await output.WriteLineAsync($" Last reported: {instance.State}; item: {instance.CurrentItemId ?? "none"}; agent: {instance.CurrentAgent ?? "none"}; heartbeat: {instance.LastHeartbeatAt:O}"); + if (instance.Scheduling is { } scheduling) + await output.WriteLineAsync($" Selection: {scheduling.Mode}; source: {scheduling.FromStatus}; target: {scheduling.TargetItemId ?? "any eligible item"}; remaining allowance: {AllowanceLabel(worker)}"); + if (worker.ConfigurationDrift == true) + await output.WriteLineAsync(" Configuration differs from the startup snapshot."); + if (worker.Detail is not null) + await output.WriteLineAsync($" {worker.Detail}"); + if (worker.Pickup is { } pickup) + await output.WriteLineAsync($" {pickup.ItemId}: {pickup.Outcome} ({pickup.Code}) — {pickup.Message}" + + (pickup.AfterCurrentItem ? " Pickup would follow the current item." : string.Empty)); + } +} diff --git a/src/Highbyte.Wrighty.Core/Backends/ITrackerBackend.cs b/src/Highbyte.Wrighty.Core/Backends/ITrackerBackend.cs index 9f13ea29..c71f5d2d 100644 --- a/src/Highbyte.Wrighty.Core/Backends/ITrackerBackend.cs +++ b/src/Highbyte.Wrighty.Core/Backends/ITrackerBackend.cs @@ -21,6 +21,9 @@ public interface IWorkItemContentReader public interface ITrackerBackend : IWorkItemContentReader { + Task> WorkflowStatusesAsync(TrackerConfig config, CancellationToken cancellationToken) => + Task.FromResult>(config.LocalMarkdown?.Statuses ?? []); + string Name { get; } IWorkItemAddressResolver AddressResolver { get; } diff --git a/src/Highbyte.Wrighty.Core/GitHub/GitHubTrackerBackend.cs b/src/Highbyte.Wrighty.Core/GitHub/GitHubTrackerBackend.cs index 826a54bb..ba8f4804 100644 --- a/src/Highbyte.Wrighty.Core/GitHub/GitHubTrackerBackend.cs +++ b/src/Highbyte.Wrighty.Core/GitHub/GitHubTrackerBackend.cs @@ -20,6 +20,9 @@ public sealed class GitHubTrackerBackend( private readonly Func retryDelay = delay ?? Task.Delay; + public Task> WorkflowStatusesAsync(TrackerConfig config, CancellationToken cancellationToken) => + projects.WorkflowStatusesAsync(config, cancellationToken); + public string Name => "github"; public IWorkItemAddressResolver AddressResolver => resolver; diff --git a/src/Highbyte.Wrighty.Core/Projects/GitHubProjectClient.cs b/src/Highbyte.Wrighty.Core/Projects/GitHubProjectClient.cs index 61eb3882..3ea99e9c 100644 --- a/src/Highbyte.Wrighty.Core/Projects/GitHubProjectClient.cs +++ b/src/Highbyte.Wrighty.Core/Projects/GitHubProjectClient.cs @@ -388,6 +388,10 @@ private static string TitleCaseProfile(string profile) => new("P3", "Low priority", "GRAY") ]; + public async Task> WorkflowStatusesAsync( + TrackerConfig config, CancellationToken cancellationToken) => + (await GetMetadataAsync(config, cancellationToken)).StatusOptions.Keys.ToArray(); + public async Task InitializeAsync( TrackerConfig config, bool checkOnly, diff --git a/src/Highbyte.Wrighty.Core/Projects/IProjectClient.cs b/src/Highbyte.Wrighty.Core/Projects/IProjectClient.cs index ac41a9d6..e079dc94 100644 --- a/src/Highbyte.Wrighty.Core/Projects/IProjectClient.cs +++ b/src/Highbyte.Wrighty.Core/Projects/IProjectClient.cs @@ -6,6 +6,9 @@ namespace Highbyte.Wrighty.Projects; public interface IProjectClient { + Task> WorkflowStatusesAsync(TrackerConfig config, CancellationToken cancellationToken) => + Task.FromResult>([]); + /// /// Whether this run created the Project, rather than adopting one that already existed. /// Wrighty owns the schema of a Project it created and provisions the priority scale there; diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerDiscovery.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerDiscovery.cs new file mode 100644 index 00000000..730f2bf7 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerDiscovery.cs @@ -0,0 +1,73 @@ +using System.Text.Json.Serialization; + +namespace Highbyte.Wrighty.Workers; + +public sealed record WorkerDiscovery( + DateTimeOffset ObservedAt, + string ConfigurationPathHash, + string Coverage, + string? Detail, + string? ConfigurationRevision, + string? ItemId, + IReadOnlyList LocalWorkers) +{ + public string Scope { get; } = "registered-workers-in-local-configuration"; +} + +public sealed record WorkerDiscoveryEntry( + WorkerInstance Instance, + [property: JsonConverter(typeof(JsonStringEnumConverter))] + WorkerInstanceLiveness Liveness, + string? Detail, + string Intake, + int? RemainingItemAllowance, + DateTimeOffset? IdleExpiresAt, + bool? ConfigurationDrift, + WorkerPickupAssessment? Pickup) +{ + public string Origin => Instance.HostKind switch + { + WorkerHostKind.CliProcess => "cli-process", + WorkerHostKind.WebHosted => "web-hosted", + _ => "unknown" + }; + public string ReportedState => Instance.State.ToString(); + + public static WorkerDiscoveryEntry From(WorkerInstanceStatus worker, string? revision, + WorkerPickupAssessment? pickup = null) + { + var instance = worker.Instance; + int? remaining = null; + if (instance.Scheduling is { ItemLimit: { } limit } selection && selection.IsUsable() && + instance.Progress is { Processed: >= 0 } progress) + remaining = (int)Math.Max(0L, (long)limit - progress.Processed - (instance.CurrentItemId is null ? 0 : 1)); + return new(instance, worker.Liveness, worker.Detail, IntakeState(worker, remaining), remaining, + IdleExpiry(instance), + revision is null || string.IsNullOrEmpty(instance.ConfigurationRevision) + ? null : revision != instance.ConfigurationRevision, pickup); + } + + private static DateTimeOffset? IdleExpiry(WorkerInstance instance) + { + if (instance.CurrentItemId is not null || instance.Scheduling?.IdleTimeout is not { } timeout || + instance.Progress is not { } idle) + return null; + try { return idle.IdleSince + timeout; } + catch (ArgumentOutOfRangeException) { return null; } + } + + private static string IntakeState(WorkerInstanceStatus worker, int? remaining) + { + if (worker.Liveness != WorkerInstanceLiveness.Running) + return "unknown"; + if (worker.Instance.State == WorkerInstanceState.Draining) + return "draining"; + if (worker.Instance.State is WorkerInstanceState.Stopping or WorkerInstanceState.StoppingNow + or WorkerInstanceState.Finalizing) + return "stopping"; + if (worker.Instance.Scheduling is not { } scheduling || !scheduling.IsUsable() || + worker.Instance.Progress is not { Processed: >= 0 } || !Enum.IsDefined(worker.Instance.State)) + return "unknown"; + return scheduling.TargetItemId is not null || scheduling.DryRun || remaining == 0 ? "closed" : "open"; + } +} diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerInstances.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerInstances.cs index 47111059..d02d1243 100644 --- a/src/Highbyte.Wrighty.Core/Workers/WorkerInstances.cs +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerInstances.cs @@ -53,7 +53,9 @@ public sealed record WorkerInstance( string? CurrentAgent = null, int? ControlProtocolVersion = null, IReadOnlyList? SupportedStopModes = null, - string? CurrentItemTitle = null); + string? CurrentItemTitle = null, + WorkerScheduling? Scheduling = null, + WorkerRunProgress? Progress = null); public sealed record WorkerInstanceStatus( WorkerInstance Instance, @@ -65,7 +67,8 @@ public sealed record WorkerProcessObservation(bool Exists, string? StartIdentity public sealed record WorkerRegistrationMetadata( WorkerHostKind HostKind, int ControlProtocolVersion = 1, - IReadOnlyList? SupportedStopModes = null) + IReadOnlyList? SupportedStopModes = null, + WorkerScheduling? Scheduling = null) { public IReadOnlyList EffectiveSupportedStopModes => SupportedStopModes ?? [WorkerStopMode.Drain, WorkerStopMode.Interrupt]; @@ -103,6 +106,9 @@ Task UpdateAsync( CancellationToken cancellationToken) => UpdateAsync(currentItemId, currentAgent, state, cancellationToken); + Task UpdateProgressAsync(WorkerRunProgress progress, CancellationToken cancellationToken) => + Task.CompletedTask; + Task ReadStopRequestAsync(CancellationToken cancellationToken) => Task.FromResult(null); @@ -114,16 +120,15 @@ Task UpdateStateAsync( public interface IWorkerInstanceRegistry { + Task InspectAsync(string configurationPath, CancellationToken cancellationToken) => + Task.FromResult(WorkerRegistrySnapshot.Unavailable(configurationPath)); + Task RegisterAsync( string configurationPath, string configurationRevision, string invocationSummary, CancellationToken cancellationToken); - Task> ListAsync( - string configurationPath, - CancellationToken cancellationToken); - Task RegisterAsync( string configurationPath, string configurationRevision, @@ -136,6 +141,10 @@ Task RegisterAsync( invocationSummary, cancellationToken); + Task> ListAsync( + string configurationPath, + CancellationToken cancellationToken); + Task RequestStopAsync( string configurationPath, WorkerStopTarget target, @@ -230,7 +239,9 @@ public async Task RegisterAsync( metadata.HostKind, CurrentAgent: null, metadata.ControlProtocolVersion, - metadata.EffectiveSupportedStopModes); + metadata.EffectiveSupportedStopModes, + Scheduling: metadata.Scheduling, + Progress: metadata.Scheduling is null ? null : new(0, timestamp)); var registration = new Registration( RecordPath(pathHash, runId), StopRequestPath(pathHash, runId), @@ -241,9 +252,27 @@ public async Task RegisterAsync( return registration; } - public async Task> ListAsync( - string configurationPath, - CancellationToken cancellationToken) + public Task> ListAsync( + string configurationPath, CancellationToken cancellationToken) => + ListCoreAsync(configurationPath, cleanup: true, cancellationToken); + + public async Task InspectAsync( + string configurationPath, CancellationToken cancellationToken) + { + var observedAt = now(); + var workers = await ListCoreAsync(configurationPath, cleanup: false, cancellationToken); + var unreadable = workers.Any(value => value.Instance.ProcessId == 0); + var unavailable = workers.Any(value => value.Instance.RunId == "registry-unavailable" && value.Instance.ProcessId == 0); + var coverage = unavailable ? "unavailable" : "complete"; + if (unreadable && !unavailable) + coverage = "incomplete"; + return new(observedAt, ConfigurationPathHash(configurationPath), + coverage, workers, + unreadable ? "Some registry records could not be read; coverage is incomplete." : null); + } + + private async Task> ListCoreAsync( + string configurationPath, bool cleanup, CancellationToken cancellationToken) { var directory = Path.Combine( paths.WorkerInstancesRoot, @@ -251,12 +280,16 @@ public async Task> ListAsync( string[] records; try { - if (!Directory.Exists(directory)) - return []; records = Directory.GetFiles(directory, "*.json") .Where(path => !path.EndsWith(".stop.json", StringComparison.Ordinal)) .ToArray(); - CleanupExpiredStopRequests(directory); + if (cleanup) + CleanupExpiredStopRequests(directory); + } + catch (DirectoryNotFoundException) + { + return MissingDirectoryIsReadable(directory, paths.WorkerInstancesRoot, paths.Root) + ? [] : [UnreadableStatus(configurationPath, "Worker registry directory could not be read.")]; } catch (Exception exception) when ( exception is IOException or UnauthorizedAccessException) @@ -270,50 +303,73 @@ public async Task> ListAsync( foreach (var path in records) { cancellationToken.ThrowIfCancellationRequested(); - WorkerInstance? instance; + var status = await ReadStatusAsync(path, configurationPath, cancellationToken); + statuses.Add(status); + if (cleanup) + CleanupExpiredRecord(path, status.Instance); + } + + return statuses + .OrderBy(value => LivenessOrder(value.Liveness)) + .ThenByDescending(value => value.Instance.StartedAt) + .ThenBy(value => value.Instance.RunId, StringComparer.Ordinal) + .ToArray(); + } + + private static bool MissingDirectoryIsReadable(params string[] paths) + { + foreach (var path in paths) + { try { - await using var stream = File.OpenRead(path); - instance = await JsonSerializer.DeserializeAsync( - stream, - JsonOptions, - cancellationToken); + if (!File.GetAttributes(path).HasFlag(FileAttributes.Directory)) + return false; } - catch (Exception exception) when ( - exception is IOException or JsonException or UnauthorizedAccessException) + catch (Exception exception) when (exception is FileNotFoundException or DirectoryNotFoundException) { - statuses.Add(UnreadableStatus( - configurationPath, - "Worker record could not be read.", - Path.GetFileNameWithoutExtension(path))); - continue; + // A genuinely absent directory is an observable empty scope. } - if (instance is null) - continue; - - statuses.Add(Status(instance)); - if (now() - instance.LastHeartbeatAt > TimeSpan.FromHours(24)) + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) { - try - { - File.Delete(path); - File.Delete(StopRequestPath( - instance.ConfigurationPathHash, - instance.RunId)); - } - catch (Exception exception) when ( - exception is IOException or UnauthorizedAccessException) - { - // Expired records are best-effort cleanup; listing remains authoritative. - } + return false; } } + return true; + } - return statuses - .OrderBy(value => LivenessOrder(value.Liveness)) - .ThenByDescending(value => value.Instance.StartedAt) - .ThenBy(value => value.Instance.RunId, StringComparer.Ordinal) - .ToArray(); + private async Task ReadStatusAsync( + string path, string configurationPath, CancellationToken cancellationToken) + { + WorkerInstance? instance; + try + { + await using var stream = File.OpenRead(path); + instance = await JsonSerializer.DeserializeAsync(stream, JsonOptions, cancellationToken); + } + catch (Exception exception) when (exception is IOException or JsonException or UnauthorizedAccessException) + { + return UnreadableStatus(configurationPath, "Worker record could not be read.", Path.GetFileNameWithoutExtension(path)); + } + if (instance is null || instance.ProcessId <= 0 || + instance.RunId != Path.GetFileNameWithoutExtension(path) || + instance.ConfigurationPathHash != ConfigurationPathHash(configurationPath)) + return UnreadableStatus(configurationPath, "Worker record identity is invalid.", Path.GetFileNameWithoutExtension(path)); + return Status(instance); + } + + private void CleanupExpiredRecord(string path, WorkerInstance instance) + { + if (instance.ProcessId <= 0 || now() - instance.LastHeartbeatAt <= TimeSpan.FromHours(24)) + return; + try + { + File.Delete(path); + File.Delete(StopRequestPath(instance.ConfigurationPathHash, instance.RunId)); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + // Cleanup belongs to the legacy listing path; read-only inspection never calls this. + } } public async Task RequestStopAsync( @@ -452,7 +508,16 @@ public static string ConfigurationPathHash(string configurationPath) private WorkerInstanceStatus Status(WorkerInstance instance) { - var observation = observe(instance.ProcessId); + WorkerProcessObservation observation; + try + { + observation = observe(instance.ProcessId); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException + or System.ComponentModel.Win32Exception or System.Security.SecurityException) + { + observation = new(true, null); + } if (!observation.Exists) { return new WorkerInstanceStatus( @@ -460,19 +525,19 @@ private WorkerInstanceStatus Status(WorkerInstance instance) WorkerInstanceLiveness.Stale, "The recorded process no longer exists."); } - if (now() - instance.LastHeartbeatAt > staleAfter) + if (observation.StartIdentity is null || instance.ProcessStartIdentity is null) { return new WorkerInstanceStatus( instance, - WorkerInstanceLiveness.Stale, - $"No heartbeat since {instance.LastHeartbeatAt:O}."); + WorkerInstanceLiveness.Unknown, + "The operating system could not verify process-start identity."); } - if (observation.StartIdentity is null || instance.ProcessStartIdentity is null) + if (now() - instance.LastHeartbeatAt > staleAfter) { return new WorkerInstanceStatus( instance, - WorkerInstanceLiveness.Unknown, - "The operating system could not verify process-start identity."); + WorkerInstanceLiveness.Stale, + $"No heartbeat since {instance.LastHeartbeatAt:O}."); } if (!string.Equals( observation.StartIdentity, @@ -566,7 +631,8 @@ private static WorkerProcessObservation ObserveProcess(int processId) return new WorkerProcessObservation(false, null); } catch (Exception exception) when ( - exception is InvalidOperationException or System.ComponentModel.Win32Exception) + exception is InvalidOperationException or System.ComponentModel.Win32Exception + or UnauthorizedAccessException or System.Security.SecurityException) { return new WorkerProcessObservation(true, null); } @@ -576,7 +642,8 @@ private static WorkerProcessObservation ObserveProcess(int processId) { try { return process.StartTime.ToUniversalTime().Ticks.ToString(); } catch (Exception exception) when ( - exception is InvalidOperationException or System.ComponentModel.Win32Exception) + exception is InvalidOperationException or System.ComponentModel.Win32Exception + or UnauthorizedAccessException or System.Security.SecurityException) { return null; } @@ -642,6 +709,7 @@ public async Task UpdateAsync( } current = current with { + Progress = currentItemId is null && current.CurrentItemId is not null ? null : current.Progress, CurrentItemId = currentItemId, CurrentItemTitle = nextItemTitle, CurrentAgent = currentAgent, @@ -656,6 +724,22 @@ public async Task UpdateAsync( } } + public async Task UpdateProgressAsync(WorkerRunProgress progress, CancellationToken cancellationToken) + { + await gate.WaitAsync(cancellationToken); + try + { + if (disposed) + return; + current = current with { Progress = progress, LastHeartbeatAt = clock() }; + await WriteWithoutGateAsync(cancellationToken); + } + finally + { + gate.Release(); + } + } + public async Task ReadStopRequestAsync( CancellationToken cancellationToken) { diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerPickupAssessment.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerPickupAssessment.cs new file mode 100644 index 00000000..dcac9974 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerPickupAssessment.cs @@ -0,0 +1,88 @@ +using static Highbyte.Wrighty.Workers.WorkerPickupOutcomes; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.Workers; + +public static class WorkerPickupOutcomes +{ + public const string CouldPickUp = "could-pick-up"; + public const string CannotPickUp = "cannot-pick-up"; + public const string Unknown = "unknown"; +} + +public sealed record WorkerPickupAssessment( + string RunId, + string ItemId, + string Outcome, + string Code, + string Message, + bool AlreadyProcessing = false, + bool AfterCurrentItem = false, + string? Agent = null); + +/// Registry evidence only. Item eligibility is evaluated by WorkerService afterwards. +public static class WorkerPickupPolicy +{ + public static WorkerPickupAssessment? AssessRegistration(WorkerInstanceStatus status, + WorkItemId itemId, string? configurationRevision, DateTimeOffset observedAt) + { + var worker = status.Instance; + WorkerPickupAssessment Result(string outcome, string code, string message) => + new(worker.RunId, itemId.Value, outcome, code, message); + if (status.Liveness == WorkerInstanceLiveness.Unknown) + return Result(Unknown, "WORKER_NOT_VERIFIED", "Worker liveness could not be verified."); + if (status.Liveness != WorkerInstanceLiveness.Running) + return Result(CannotPickUp, "WORKER_STALE", "This registration is stale; it does not establish active intake."); + if (worker.CurrentItemId == itemId.Value && worker.State is + WorkerInstanceState.PreparingItem or WorkerInstanceState.RunningItem or + WorkerInstanceState.Draining or WorkerInstanceState.StoppingNow) + return Result(CouldPickUp, "ALREADY_PROCESSING", "This worker reports that it is processing this item.") + with { AlreadyProcessing = true, Agent = worker.CurrentAgent }; + if (worker.State is WorkerInstanceState.Draining or WorkerInstanceState.Stopping or + WorkerInstanceState.StoppingNow or WorkerInstanceState.Finalizing) + return Result(CannotPickUp, "INTAKE_CLOSED", "The worker is draining, stopping, or finalizing."); + if (!Enum.IsDefined(worker.State)) + return Result(Unknown, "INTAKE_UNKNOWN", "The worker's intake state is not recognized."); + if (worker.Scheduling is not { } scheduling) + return Result(Unknown, "SCHEDULING_UNKNOWN", "This registration has no structured startup selection."); + if (!scheduling.IsUsable()) + return Result(Unknown, "SCHEDULING_UNKNOWN", "The startup selection is incomplete or unsupported."); + return AssessSelection(worker, itemId, configurationRevision, observedAt); + } + + private static WorkerPickupAssessment? AssessSelection(WorkerInstance worker, + WorkItemId itemId, string? configurationRevision, DateTimeOffset observedAt) + { + var scheduling = worker.Scheduling!; + WorkerPickupAssessment Result(string outcome, string code, string message) => + new(worker.RunId, itemId.Value, outcome, code, message); + if (scheduling.DryRun) + return Result(CannotPickUp, "DRY_RUN", "This run only previews work."); + if (scheduling.TargetItemId is { } target) + return target == itemId.Value + ? Result(Unknown, "TARGETED_STARTING", "This run targets the item but has not reported processing it yet.") + : Result(CannotPickUp, "TARGET_MISMATCH", "This run targets a different item."); + if (scheduling.Mode is not ("continuous" or "bounded")) + return Result(Unknown, "SCHEDULING_UNKNOWN", "The worker selection mode is not recognized."); + return AssessLifetime(worker, itemId, configurationRevision, observedAt); + } + + private static WorkerPickupAssessment? AssessLifetime(WorkerInstance worker, + WorkItemId itemId, string? configurationRevision, DateTimeOffset observedAt) + { + var scheduling = worker.Scheduling!; + WorkerPickupAssessment Result(string outcome, string code, string message) => + new(worker.RunId, itemId.Value, outcome, code, message); + if (worker.Progress is not { } progress || progress.Processed < 0) + return Result(Unknown, "PROGRESS_UNKNOWN", "Remaining allowance and idle lifetime could not be established."); + var reserved = worker.CurrentItemId is null ? 0 : 1; + if (scheduling.ItemLimit is { } limit && progress.Processed >= limit - reserved) + return Result(CannotPickUp, "ITEM_LIMIT_REACHED", "The run has no allowance for another item."); + if (worker.CurrentItemId is null && scheduling.IdleTimeout is { } timeout && + observedAt - progress.IdleSince >= timeout) + return Result(CannotPickUp, "IDLE_EXPIRED", "The reported idle lifetime has expired."); + if (configurationRevision is null || worker.ConfigurationRevision != configurationRevision) + return Result(Unknown, "CONFIGURATION_DRIFT", "The worker's startup configuration cannot be matched to the current configuration."); + return null; + } +} diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerRunHost.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerRunHost.cs index d297c3c8..b58d7f50 100644 --- a/src/Highbyte.Wrighty.Core/Workers/WorkerRunHost.cs +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerRunHost.cs @@ -75,6 +75,11 @@ internal static class WorkerInstanceEventProjection /// public sealed class WorkerRunControl : IDisposable { + internal Func? ReportProgress { get; set; } + + internal Task ReportProgressAsync(WorkerRunProgress progress) => + ReportProgress?.Invoke(progress) ?? Task.CompletedTask; + private static readonly ConcurrentDictionary Controls = []; private readonly CancellationTokenSource intake = new(); private readonly CancellationTokenSource interruption = new(); @@ -237,11 +242,23 @@ public async Task RunAsync( identity.ConfigurationRevision, identity.InvocationSummary, identity.HostKind, + WorkerScheduling.From(config, options, identity, selection), callbacks.Warn, hostCancellationToken); await using var registrationScope = registration; control.RunId = registration.RunId; var warningState = new RegistryWarningState(); + control.ReportProgress = async progress => + { + try + { + await registration.UpdateProgressAsync(progress, CancellationToken.None); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + await WarnOnceAsync(warningState, callbacks.Warn, "Local worker progress could not be updated."); + } + }; void ControlStateChanged() => _ = ReflectControlStateAsync( registration, control, @@ -306,6 +323,7 @@ await TryUpdateAsync( // Expected when the worker run ends before another control poll. } control.StateChanged -= ControlStateChanged; + control.ReportProgress = null; } } @@ -314,6 +332,7 @@ private async Task RegisterAsync( string configurationRevision, string invocationSummary, WorkerHostKind hostKind, + WorkerScheduling scheduling, Func? warn, CancellationToken cancellationToken) { @@ -323,7 +342,7 @@ private async Task RegisterAsync( configurationPath, configurationRevision, invocationSummary, - new WorkerRegistrationMetadata(hostKind), + new WorkerRegistrationMetadata(hostKind, Scheduling: scheduling), cancellationToken); } catch (Exception exception) when ( diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerScheduling.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerScheduling.cs new file mode 100644 index 00000000..95657452 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerScheduling.cs @@ -0,0 +1,70 @@ +using Highbyte.Wrighty.Configuration; + +namespace Highbyte.Wrighty.Workers; + +/// Effective startup selection, without claim credentials or parsed display commands. +public sealed record WorkerScheduling( + string Mode, + string? TargetItemId, + WorkerItemIntent Intent, + string FromStatus, + string ToStatus, + string? AgentOverride, + string? DefaultAgent, + WorkspaceMode WorkspaceMode, + string RepositoryPath, + IReadOnlyDictionary Filters, + int? ItemLimit, + TimeSpan? IdleTimeout, + TimeSpan ItemTimeout, + string? Profile, + bool DryRun) +{ + public static WorkerScheduling From(TrackerConfig config, WorkerOptions options, + WorkerRunIdentity identity, WorkerRunSelection selection) => new( + SelectionMode(options, selection), + selection.ItemId?.Value, selection.Intent, + options.FromStatus ?? config.DefaultPickFrom, options.ToStatus ?? config.DefaultPickTo, + options.Agent, config.EffectiveWorker.DefaultAgent, options.WorkspaceMode, + Path.GetFullPath(identity.RepositoryPath), new Dictionary(options.Filters), + selection.ItemId is not null || options.Once ? 1 : options.MaxItems, + options.IdleTimeout, options.ItemTimeout, options.Profile, options.DryRun); + + public bool IsUsable() => + !string.IsNullOrWhiteSpace(FromStatus) && !string.IsNullOrWhiteSpace(ToStatus) && + !string.IsNullOrWhiteSpace(RepositoryPath) && Path.IsPathFullyQualified(RepositoryPath) && + Filters is not null && Enum.IsDefined(WorkspaceMode) && Enum.IsDefined(Intent) && + ItemTimeout > TimeSpan.Zero && (IdleTimeout is null || IdleTimeout > TimeSpan.Zero) && + (ItemLimit is null || ItemLimit > 0) && Mode switch + { + "continuous" => TargetItemId is null && ItemLimit is null, + "bounded" => TargetItemId is null && ItemLimit is not null, + "targeted" => TargetItemId is not null && ItemLimit == 1, + _ => false + }; + + private static string SelectionMode(WorkerOptions options, WorkerRunSelection selection) + { + if (selection.ItemId is not null) + return "targeted"; + return options.Once || options.MaxItems.HasValue ? "bounded" : "continuous"; + } + + public WorkerOptions Options() => new(AgentOverride, ItemLimit == 1, ItemLimit, WorkspaceMode, + Filters, IdleTimeout, ItemTimeout, FencedAction.Kill, null, "agent", DryRun, true, + FromStatus, ToStatus, Profile: Profile); +} + +public sealed record WorkerRunProgress(int Processed, DateTimeOffset IdleSince); + +public sealed record WorkerRegistrySnapshot( + DateTimeOffset ObservedAt, + string ConfigurationPathHash, + string Coverage, + IReadOnlyList Workers, + string? Detail = null) +{ + public static WorkerRegistrySnapshot Unavailable(string configurationPath) => new( + DateTimeOffset.UtcNow, JsonWorkerInstanceRegistry.ConfigurationPathHash(configurationPath), + "unavailable", [], "This registry does not support read-only inspection."); +} diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerService.Discovery.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerService.Discovery.cs new file mode 100644 index 00000000..a582cb36 --- /dev/null +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerService.Discovery.cs @@ -0,0 +1,106 @@ +using static Highbyte.Wrighty.Workers.WorkerPickupOutcomes; +using Highbyte.Wrighty.Claims; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Errors; +using Highbyte.Wrighty.Models; + +namespace Highbyte.Wrighty.Workers; + +public sealed partial class WorkerService +{ + /// Advisory selection only. No claim, queue repair, workspace creation, or vendor probe. + public async Task AssessPickupAsync( + TrackerConfig config, WorkerInstanceStatus worker, WorkItemOperationalState state, + string? configurationRevision, DateTimeOffset observedAt, CancellationToken cancellationToken) + { + if (WorkerPickupPolicy.AssessRegistration(worker, state.Item.Id, configurationRevision, observedAt) is { } registration) + return registration; + try + { + return await AssessSelectionAsync(config, worker, state, cancellationToken); + } + catch (TrackerException exception) + { + return Pickup(worker, state, Unknown, exception.Code, exception.Message); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException + or System.ComponentModel.Win32Exception) + { + return Pickup(worker, state, Unknown, "ASSESSMENT_UNAVAILABLE", "Required local or tracker evidence could not be read."); + } + } + + private async Task AssessSelectionAsync(TrackerConfig config, + WorkerInstanceStatus worker, WorkItemOperationalState state, CancellationToken cancellationToken) + { + var scheduling = worker.Instance.Scheduling!; + var options = scheduling.Options(); + if (state.Item.Archived) + return Pickup(worker, state, CannotPickUp, "ITEM_ARCHIVED", "Archived work is outside worker intake."); + if (!WorkerPolicyGate.MatchesFilters(state.Item, options.Filters)) + return Pickup(worker, state, CannotPickUp, "FILTER_MISMATCH", "The item does not match the worker's startup filters."); + if (state.Claim.State == ClaimOwnershipState.HeldByOther) + return Pickup(worker, state, CannotPickUp, "CLAIM_HELD", "Another installation holds this item's claim."); + if (!Directory.Exists(scheduling.RepositoryPath)) + return Pickup(worker, state, Unknown, "WORKSPACE_UNAVAILABLE", "The worker's repository is unavailable to this observer."); + + string agent; + if (string.Equals(state.Item.Status, scheduling.ToStatus, StringComparison.OrdinalIgnoreCase)) + { + // Reuse the continuous loop's retained-session selection, including due local dispatch, + // directed handoffs, claims, recorded workspace, and per-agent enablement. + var queued = await QueuedCandidatesAsync(config, options, scheduling.RepositoryPath, null, cancellationToken); + var candidate = queued.FirstOrDefault(value => value.Detail.Id == state.Item.Id); + if (candidate is null) + return Pickup(worker, state, CannotPickUp, "NOT_QUEUED_FOR_THIS_WORKER", + "The continuous worker's retained-session rules do not admit this item now; inspect its actions and dispatch timing."); + agent = candidate.AgentName; + } + else + { + var fresh = await AssessFreshPickupAsync(config, worker, state, options, cancellationToken); + if (fresh.Outcome != CouldPickUp) + return fresh; + agent = fresh.Agent!; + } + var workspace = state.Session?.WorkspacePath ?? scheduling.RepositoryPath; + if ((state.Session is not null || scheduling.WorkspaceMode == WorkspaceMode.Current) && + workspaceLocks.Inspect(workspace) is { State: not "available" } workspaceState) + return Pickup(worker, state, Unknown, "WORKSPACE_LOCK_UNVERIFIED", workspaceState.Detail ?? "Workspace availability is unknown."); + if (config.Testing?.FindCapacityProbe(agent) is not null) + return Pickup(worker, state, Unknown, "SIMULATED_CAPACITY", "Provider simulation is configured; discovery does not create or consume its state."); + var capacity = await providerCapacity.GetAsync(agent, cancellationToken); + if (capacity is { State: not ProviderCapacityState.Available }) + return Pickup(worker, state, CannotPickUp, "PROVIDER_DEFERRED", + "Cached provider state delays pickup; no paid capacity probe was run.") with { Agent = agent }; + return Pickup(worker, state, CouldPickUp, "ELIGIBLE", + "Observed selection rules allow pickup. This is not a reservation or a timing guarantee; launch checks run again at pickup.") + with { Agent = agent, AfterCurrentItem = worker.Instance.CurrentItemId is not null }; + } + + private async Task AssessFreshPickupAsync(TrackerConfig config, + WorkerInstanceStatus worker, WorkItemOperationalState state, WorkerOptions options, + CancellationToken cancellationToken) + { + var scheduling = worker.Instance.Scheduling!; + if (!string.Equals(state.Item.Status, scheduling.FromStatus, StringComparison.OrdinalIgnoreCase)) + return Pickup(worker, state, CannotPickUp, "STATUS_MISMATCH", "The item is outside this worker's source and retained-session statuses."); + if (state.Claim.State != ClaimOwnershipState.Unclaimed) + return Pickup(worker, state, CannotPickUp, "CLAIM_HELD", "The item is already claimed."); + var candidate = ProjectWorkerQueueAuthorization(config, state.Item, scheduling.FromStatus); + var diagnostics = new WorkerCandidateDiagnostics(scheduling.FromStatus); + var evaluation = EvaluateCandidate(candidate, options, scheduling.DefaultAgent, + await LoadUserSettingsAsync(cancellationToken), diagnostics); + if (!evaluation.Eligible) + return Pickup(worker, state, CannotPickUp, "WORKER_POLICY_REFUSED", diagnostics.Describe(options.Filters.Count > 0)); + EnsurePreflightWorkspaceReady(options, scheduling.RepositoryPath, evaluation.Agent!, new HashSet()); + var verdict = await LaunchPreflight.EvaluateAsync(new LaunchPreflightRequest( + config, options, candidate, evaluation.Agent!, LaunchKind.Fresh, LaunchStage.PreClaim), cancellationToken); + if (!verdict.Admitted) + return Pickup(worker, state, CannotPickUp, verdict.Code ?? "CONTEXT_BLOCKED", verdict.Message ?? "Pre-claim admission refused this item."); + return Pickup(worker, state, CouldPickUp, "ELIGIBLE", "Fresh selection policy admits this item.") with { Agent = evaluation.Agent }; + } + + private static WorkerPickupAssessment Pickup(WorkerInstanceStatus worker, WorkItemOperationalState state, + string outcome, string code, string message) => new(worker.Instance.RunId, state.Item.Id.Value, outcome, code, message); +} diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkerService.cs b/src/Highbyte.Wrighty.Core/Workers/WorkerService.cs index 235e2ebd..72c8d0ab 100644 --- a/src/Highbyte.Wrighty.Core/Workers/WorkerService.cs +++ b/src/Highbyte.Wrighty.Core/Workers/WorkerService.cs @@ -10,7 +10,7 @@ namespace Highbyte.Wrighty.Workers; -public sealed class WorkerService( +public sealed partial class WorkerService( TrackerService tracker, IAgentProcessRunner processes, IWorkspaceManager workspaces, @@ -413,6 +413,7 @@ await emit(new WorkerEvent( } var state = new WorkerLoopState(now()); + await control.ReportProgressAsync(new(state.Processed, state.IdleStarted)); while (!control.IntakeClosed && (!options.MaxItems.HasValue || state.Processed < options.MaxItems.Value)) { @@ -463,6 +464,7 @@ private async Task RunIterationAsync( if (queued is not null) { state.Record(queued, now()); + await control.ReportProgressAsync(new(state.Processed, state.IdleStarted)); return options.Once; } @@ -474,6 +476,7 @@ private async Task RunIterationAsync( var disposition = await RunFreshCandidateAsync( config, options, repositoryPath, diagnostics, emit, cancellationToken); state.Record(disposition, now()); + await control.ReportProgressAsync(new(state.Processed, state.IdleStarted)); return options.Once; } catch (TrackerException exception) when (exception.Code == "NO_ITEM_AVAILABLE") diff --git a/src/Highbyte.Wrighty.Core/Workers/WorkspaceExecutionLock.cs b/src/Highbyte.Wrighty.Core/Workers/WorkspaceExecutionLock.cs index 18afdc4e..3dc9ca05 100644 --- a/src/Highbyte.Wrighty.Core/Workers/WorkspaceExecutionLock.cs +++ b/src/Highbyte.Wrighty.Core/Workers/WorkspaceExecutionLock.cs @@ -4,8 +4,12 @@ namespace Highbyte.Wrighty.Workers; +public sealed record WorkspaceLockObservation(string State, string? Detail = null); + public interface IWorkspaceExecutionLock { + WorkspaceLockObservation Inspect(string workspacePath) => new("unknown", "Workspace lock inspection is unavailable."); + ValueTask AcquireAsync( string workspacePath, CancellationToken cancellationToken); @@ -20,6 +24,28 @@ public sealed class FileWorkspaceExecutionLock(string? lockRoot = null) : IWorks private readonly string root = lockRoot ?? DefaultRoot; + public WorkspaceLockObservation Inspect(string workspacePath) + { + try + { + var canonicalPath = CanonicalPath(workspacePath); + var key = OperatingSystem.IsWindows() ? canonicalPath.ToUpperInvariant() : canonicalPath; + var digest = Convert.ToHexStringLower(SHA256.HashData(Encoding.UTF8.GetBytes(key))); + // Open an existing lock for reading only. Never create a directory, lock file, or lease. + using var stream = new FileStream(Path.Combine(root, $"{digest}.lock"), FileMode.Open, + FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); + return new("available"); + } + catch (Exception exception) when (exception is FileNotFoundException or DirectoryNotFoundException) + { + return new("available"); + } + catch (Exception exception) when (exception is IOException or UnauthorizedAccessException) + { + return new("unknown", "The workspace lock is busy or could not be inspected."); + } + } + public ValueTask AcquireAsync( string workspacePath, CancellationToken cancellationToken) @@ -130,6 +156,8 @@ internal sealed class NoOpWorkspaceExecutionLock : IWorkspaceExecutionLock { public static NoOpWorkspaceExecutionLock Instance { get; } = new(); + public WorkspaceLockObservation Inspect(string workspacePath) => new("available"); + public ValueTask AcquireAsync( string workspacePath, CancellationToken cancellationToken) => diff --git a/tests/Highbyte.Wrighty.UnitTests/Cli/CliApplicationTests.cs b/tests/Highbyte.Wrighty.UnitTests/Cli/CliApplicationTests.cs index 4b71fbdb..9e8129f9 100644 --- a/tests/Highbyte.Wrighty.UnitTests/Cli/CliApplicationTests.cs +++ b/tests/Highbyte.Wrighty.UnitTests/Cli/CliApplicationTests.cs @@ -4556,7 +4556,8 @@ public Task LoadAsync( private sealed class RecordingBackend( bool automaticExecutionAllowed = false, - bool? contextApprovalFieldApproved = null) + bool? contextApprovalFieldApproved = null, + bool failReads = false) : IWorkItemBackend, IExistingWorkItemAdoptionBackend { public CreateWorkItemRequest? Request { get; private set; } @@ -4589,7 +4590,11 @@ public Task AdoptAsync( public Task GetAsync( TrackerConfig config, WorkItemId id, - CancellationToken cancellationToken) => Task.FromResult(new WorkItemDetail( + CancellationToken cancellationToken) + { + if (failReads) + throw new InvalidOperationException("Tracker must not be read."); + return Task.FromResult(new WorkItemDetail( id, "Example", "Body", @@ -4599,6 +4604,7 @@ public Task AdoptAsync( AutomaticExecutionAllowed: automaticExecutionAllowed, AgentPolicy: automaticExecutionAllowed ? "claude" : null, ContextApprovalFieldApproved: contextApprovalFieldApproved)); + } public Task CreateAsync( TrackerConfig config, diff --git a/tests/Highbyte.Wrighty.UnitTests/Cli/WorkerDiscoveryCommandTests.cs b/tests/Highbyte.Wrighty.UnitTests/Cli/WorkerDiscoveryCommandTests.cs new file mode 100644 index 00000000..080a1963 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Cli/WorkerDiscoveryCommandTests.cs @@ -0,0 +1,133 @@ +using System.Text.Json; +using Highbyte.Wrighty.Workers; + +namespace Highbyte.Wrighty.UnitTests.Cli; + +public sealed partial class CliApplicationTests +{ + [Fact] + public async Task Workers_plain_listing_uses_read_only_registry_without_tracker_access() + { + var output = new StringWriter(); + var registry = new DiscoveryRegistry([]); + Assert.Equal(0, await Application(new RecordingBackend(failReads: true), new StringReader(""), output, + config: Config with { SourcePath = "/configuration/board.json", SourceRevision = "revision" }, + workerInstanceRegistry: registry).InvokeAsync(["workers", "--json"])); + using var json = JsonDocument.Parse(output.ToString()); + var result = json.RootElement.GetProperty("result"); + Assert.Equal(1, json.RootElement.GetProperty("schemaVersion").GetInt32()); + Assert.Equal("complete", result.GetProperty("coverage").GetString()); + Assert.Equal(0, result.GetProperty("localWorkers").GetArrayLength()); + Assert.Equal("/configuration/board.json", registry.ObservedPath); + } + + [Fact] + public async Task Workers_with_legacy_registry_reports_unavailable_not_empty_coverage() + { + var output = new StringWriter(); + Assert.Equal(0, await Application(new RecordingBackend(failReads: true), new StringReader(""), output) + .InvokeAsync(["workers", "--json"])); + using var json = JsonDocument.Parse(output.ToString()); + Assert.Equal("unavailable", json.RootElement.GetProperty("result").GetProperty("coverage").GetString()); + } + + [Theory] + [InlineData("continuous", "could-pick-up", "ELIGIBLE")] + [InlineData("busy-continuous", "could-pick-up", "ELIGIBLE")] + [InlineData("busy-once", "cannot-pick-up", "ITEM_LIMIT_REACHED")] + [InlineData("filtered", "cannot-pick-up", "FILTER_MISMATCH")] + [InlineData("drift", "unknown", "CONFIGURATION_DRIFT")] + [InlineData("denied-runtime", "unknown", "ASSESSMENT_UNAVAILABLE")] + public async Task Workers_item_assessment_uses_selection_and_launch_evidence(string scenario, string outcome, string code) + { + var config = Config with { SourceRevision = "revision", SourcePath = "/configuration/board.json" }; + var observed = DateTimeOffset.UtcNow; + var scheduling = new WorkerScheduling("continuous", null, WorkerItemIntent.Auto, + config.DefaultPickFrom, config.DefaultPickTo, null, "claude", WorkspaceMode.Current, + Directory.GetCurrentDirectory(), new Dictionary(), null, null, TimeSpan.FromHours(1), null, false); + var busy = scenario.StartsWith("busy", StringComparison.Ordinal); + if (scenario == "busy-once") scheduling = scheduling with { Mode = "bounded", ItemLimit = 1 }; + if (scenario == "filtered") scheduling = scheduling with { Filters = new Dictionary { ["priority"] = "P99" } }; + var instance = new WorkerInstance("run", 123, "start", observed, observed, "scope", + scenario == "drift" ? "old-revision" : "revision", "1", "do not parse", busy ? "github:owner/repo#5" : null, + busy ? WorkerInstanceState.RunningItem : WorkerInstanceState.Idle, + Scheduling: scheduling, Progress: new(0, observed)); + var registry = new DiscoveryRegistry([new(instance, WorkerInstanceLiveness.Running, null)]); + var output = new StringWriter(); + var error = new StringWriter(); + var backend = new RecordingBackend(automaticExecutionAllowed: true); + var app = Application(backend, new StringReader(""), output, error, + workerCandidate: true, config: config, workerInstanceRegistry: registry, + runtimeCatalog: scenario == "denied-runtime" ? new DeniedRuntimeCatalog() : new FixedRuntimeCatalog("claude")); + Assert.Equal(0, await app.InvokeAsync(["workers", "--item", "42", "--json"])); + using var json = JsonDocument.Parse(output.ToString()); + var result = json.RootElement.GetProperty("result"); + Assert.Equal("github:owner/repo#42", result.GetProperty("itemId").GetString()); + var entry = Assert.Single(result.GetProperty("localWorkers").EnumerateArray()); + Assert.Equal("Running", entry.GetProperty("liveness").GetString()); + Assert.Equal("cli-process", entry.GetProperty("origin").GetString()); + var pickup = entry.GetProperty("pickup"); + Assert.Equal(outcome, pickup.GetProperty("outcome").GetString()); + Assert.Equal(code, pickup.GetProperty("code").GetString()); + Assert.Null(backend.Patch); + Assert.Null(backend.Operation); + } + + [Theory] + [InlineData(false, "complete")] + [InlineData(true, "possibly-truncated")] + public async Task List_preserves_item_array_and_labels_count_scope(bool limited, string completeness) + { + var output = new StringWriter(); + var args = limited ? new[] { "list", "--limit", "1", "--json" } : ["list", "--json"]; + Assert.Equal(0, await Application(new RecordingBackend(), new StringReader(""), output, workerCandidate: true).InvokeAsync(args)); + using var json = JsonDocument.Parse(output.ToString()); + Assert.Equal(JsonValueKind.Array, json.RootElement.GetProperty("result").ValueKind); + var listing = json.RootElement.GetProperty("listing"); + Assert.Equal("returned-items", listing.GetProperty("countScope").GetString()); + Assert.Equal(completeness, listing.GetProperty("completeness").GetString()); + Assert.Equal("unknown", listing.GetProperty("statusOrderSource").GetString()); + } + + [Fact] + public async Task Human_worker_output_preserves_run_identity_scope_and_uncertainty() + { + var time = DateTimeOffset.UtcNow; + var scheduling = new WorkerScheduling("continuous", null, WorkerItemIntent.Auto, + Config.DefaultPickFrom, Config.DefaultPickTo, null, "claude", WorkspaceMode.Current, + Directory.GetCurrentDirectory(), new Dictionary(), null, null, TimeSpan.FromHours(1), null, false); + var known = new WorkerInstance("hosted-one", 123, "start", time, time, "scope", "old", "1", "display", null, + WorkerInstanceState.Idle, WorkerHostKind.WebHosted, Scheduling: scheduling, Progress: new(0, time)); + var unknown = known with { RunId = "hosted-two", Scheduling = null, Progress = null }; + var registry = new DiscoveryRegistry([ + new(known, WorkerInstanceLiveness.Running, null), + new(unknown, WorkerInstanceLiveness.Unknown, "Process identity could not be verified.")]); + var output = new StringWriter(); + Assert.Equal(0, await Application(new RecordingBackend(), new StringReader(""), output, + config: Config with { SourceRevision = "revision" }, workerCandidate: true, workerInstanceRegistry: registry) + .InvokeAsync(["workers", "--item", "42"])); + var text = output.ToString(); + Assert.Contains("hosted-one", text); + Assert.Contains("hosted-two", text); + Assert.Contains("Unregistered or remote workers may exist", text); + Assert.Contains("unlimited", text); + Assert.Contains("Configuration differs", text); + Assert.Contains("CONFIGURATION_DRIFT", text); + Assert.Contains("WORKER_NOT_VERIFIED", text); + Assert.Contains("Process identity could not be verified", text); + } + + private sealed class DiscoveryRegistry(IReadOnlyList workers) : IWorkerInstanceRegistry + { + public string? ObservedPath { get; private set; } + public Task InspectAsync(string configurationPath, CancellationToken cancellationToken) + { + ObservedPath = configurationPath; + return Task.FromResult(new WorkerRegistrySnapshot(DateTimeOffset.UtcNow, "scope", "complete", workers)); + } + public Task> ListAsync(string configurationPath, CancellationToken cancellationToken) => + throw new InvalidOperationException("Discovery must not use a listing that may clean up records."); + public Task RegisterAsync(string configurationPath, string configurationRevision, string invocationSummary, CancellationToken cancellationToken) => + throw new InvalidOperationException("Discovery must not register a worker."); + } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Workers/LocalWorkerStateTests.cs b/tests/Highbyte.Wrighty.UnitTests/Workers/LocalWorkerStateTests.cs index 1dd5008e..5e8e216d 100644 --- a/tests/Highbyte.Wrighty.UnitTests/Workers/LocalWorkerStateTests.cs +++ b/tests/Highbyte.Wrighty.UnitTests/Workers/LocalWorkerStateTests.cs @@ -14,7 +14,7 @@ namespace Highbyte.Wrighty.UnitTests.Workers; -public sealed class LocalDispatchStateTests : IDisposable +public sealed partial class LocalDispatchStateTests : IDisposable { private readonly string directory = Path.Combine(Path.GetTempPath(), $"wrighty-worker-{Guid.NewGuid():N}"); private readonly FakeClock clock = new(DateTimeOffset.Parse("2026-07-17T10:00:00Z")); diff --git a/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerDiscoveryTests.cs b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerDiscoveryTests.cs new file mode 100644 index 00000000..ee6ecda2 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerDiscoveryTests.cs @@ -0,0 +1,226 @@ +using System.Text.Json; +using System.Globalization; +using Highbyte.Wrighty.Caching; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.Models; +using Highbyte.Wrighty.Workers; + +namespace Highbyte.Wrighty.UnitTests.Workers; + +public sealed class WorkerDiscoveryTests : IDisposable +{ + private readonly string root = Path.Combine(Path.GetTempPath(), $"wrighty-discovery-{Guid.NewGuid():N}"); + private static readonly DateTimeOffset Now = DateTimeOffset.Parse("2026-09-14T10:00:00Z", CultureInfo.InvariantCulture); + private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web); + private string ConfigPath => Path.Combine(root, "config.json"); + private string RecordDirectory => Path.Combine(new CachePaths(root).WorkerInstancesRoot, + JsonWorkerInstanceRegistry.ConfigurationPathHash(ConfigPath)); + + [Fact] + public async Task Inspection_retains_expired_records_and_stop_requests() + { + var registry = Registry(); + await WriteRecord(Instance() with { LastHeartbeatAt = Now.AddDays(-2) }); + var stop = Path.Combine(RecordDirectory, "old.stop.json"); + await File.WriteAllTextAsync(stop, "{}"); + File.SetLastWriteTimeUtc(stop, Now.AddDays(-2).UtcDateTime); + var before = Directory.GetFiles(RecordDirectory).ToDictionary(path => path, File.ReadAllText); + var snapshot = await registry.InspectAsync(ConfigPath, default); + Assert.Equal("complete", snapshot.Coverage); + Assert.Equal(WorkerInstanceLiveness.Stale, Assert.Single(snapshot.Workers).Liveness); + Assert.Equal(before, Directory.GetFiles(RecordDirectory).ToDictionary(path => path, File.ReadAllText)); + } + + [Fact] + public async Task Empty_scope_is_distinct_from_unreadable_records_and_legacy_registry() + { + Assert.Equal("complete", (await Registry().InspectAsync(ConfigPath, default)).Coverage); + Assert.False(Directory.Exists(root)); + Directory.CreateDirectory(RecordDirectory); + await File.WriteAllTextAsync(Path.Combine(RecordDirectory, "broken.json"), "{"); + var partial = await Registry().InspectAsync(ConfigPath, default); + Assert.Equal("incomplete", partial.Coverage); + Assert.Equal(WorkerInstanceLiveness.Unknown, Assert.Single(partial.Workers).Liveness); + IWorkerInstanceRegistry legacy = NoOpWorkerInstanceRegistry.Instance; + Assert.Equal("unavailable", (await legacy.InspectAsync(ConfigPath, default)).Coverage); + } + + [Fact] + public async Task Inaccessible_registry_path_is_not_a_confidently_empty_scope() + { + Directory.CreateDirectory(root); + await File.WriteAllTextAsync(new CachePaths(root).WorkerInstancesRoot, "not a directory"); + var snapshot = await Registry().InspectAsync(ConfigPath, default); + Assert.Equal("unavailable", snapshot.Coverage); + Assert.Equal(WorkerInstanceLiveness.Unknown, Assert.Single(snapshot.Workers).Liveness); + } + + [Theory] + [InlineData("null")] + [InlineData("wrong-scope")] + [InlineData("wrong-run")] + public async Task Invalid_record_identity_is_not_treated_as_empty_or_live(string variant) + { + var instance = Instance(); + if (variant == "wrong-scope") instance = instance with { ConfigurationPathHash = "elsewhere" }; + if (variant == "wrong-run") instance = instance with { RunId = "elsewhere" }; + Directory.CreateDirectory(RecordDirectory); + await File.WriteAllTextAsync(Path.Combine(RecordDirectory, "run.json"), + variant == "null" ? "null" : JsonSerializer.Serialize(instance, JsonOptions)); + Assert.Equal("incomplete", (await Registry().InspectAsync(ConfigPath, default)).Coverage); + } + + [Theory] + [InlineData("denied", WorkerInstanceLiveness.Unknown)] + [InlineData("missing", WorkerInstanceLiveness.Stale)] + [InlineData("reused", WorkerInstanceLiveness.Stale)] + public async Task Targeted_process_inspection_degrades_honestly(string scenario, WorkerInstanceLiveness expected) + { + await WriteRecord(Instance()); + var inspected = new List(); + var registry = new JsonWorkerInstanceRegistry(new CachePaths(root), () => Now, pid => + { + inspected.Add(pid); + return scenario switch + { + "denied" => throw new UnauthorizedAccessException(), + "missing" => new(false, null), + _ => new(true, "reused") + }; + }); + Assert.Equal(expected, Assert.Single((await registry.InspectAsync(ConfigPath, default)).Workers).Liveness); + Assert.Equal([123], inspected); + } + + [Fact] + public async Task Web_hosted_runs_sharing_a_pid_keep_distinct_run_ids_and_effective_scheduling() + { + var registry = new JsonWorkerInstanceRegistry(new CachePaths(root), heartbeatInterval: TimeSpan.FromMinutes(5)); + var metadata = new WorkerRegistrationMetadata(WorkerHostKind.WebHosted, Scheduling: Scheduling()); + await using var one = await registry.RegisterAsync(ConfigPath, "revision", "display only", metadata, default); + await using var two = await registry.RegisterAsync(ConfigPath, "revision", "display only", metadata, default); + await one.UpdateAsync("local:1", "Title", "codex", WorkerInstanceState.RunningItem, default); + await one.UpdateProgressAsync(new(2, Now), default); + var runs = (await registry.InspectAsync(ConfigPath, default)).Workers; + Assert.Equal(2, runs.Count); + Assert.Single(runs.Select(run => run.Instance.ProcessId).Distinct()); + Assert.Equal(2, runs.Select(run => run.Instance.RunId).Distinct().Count()); + var record = runs.Single(run => run.Instance.RunId == one.RunId).Instance; + Assert.Equal(2, record.Progress!.Processed); + Assert.Equal("Ideas", record.Scheduling!.FromStatus); + Assert.Equal("claude", record.Scheduling.DefaultAgent); + Assert.Equal("codex", record.CurrentAgent); + await one.UpdateAsync(null, null, null, WorkerInstanceState.Idle, default); + record = (await registry.InspectAsync(ConfigPath, default)).Workers.Single(run => run.Instance.RunId == one.RunId).Instance; + Assert.Null(record.Progress); // no transient allowance between completion and loop accounting + } + + [Theory] + [InlineData("once-busy", "cannot-pick-up", "ITEM_LIMIT_REACHED")] + [InlineData("bounded-exhausted", "cannot-pick-up", "ITEM_LIMIT_REACHED")] + [InlineData("targeted-other", "cannot-pick-up", "TARGET_MISMATCH")] + [InlineData("targeted-same", "unknown", "TARGETED_STARTING")] + [InlineData("drain", "cannot-pick-up", "INTAKE_CLOSED")] + [InlineData("unknown", "unknown", "WORKER_NOT_VERIFIED")] + [InlineData("legacy", "unknown", "SCHEDULING_UNKNOWN")] + [InlineData("missing-progress", "unknown", "PROGRESS_UNKNOWN")] + [InlineData("drift", "unknown", "CONFIGURATION_DRIFT")] + [InlineData("expired", "cannot-pick-up", "IDLE_EXPIRED")] + [InlineData("same-item", "could-pick-up", "ALREADY_PROCESSING")] + public void Assessment_distinguishes_run_selection_and_lifetime(string scenario, string outcome, string code) + { + var worker = Instance(); + worker = scenario switch + { + "once-busy" => worker with { CurrentItemId = "local:other", State = WorkerInstanceState.RunningItem, Scheduling = Scheduling() with { Mode = "bounded", ItemLimit = 1 } }, + "bounded-exhausted" => worker with { Progress = new(3, Now), Scheduling = Scheduling() with { Mode = "bounded", ItemLimit = 3 } }, + "targeted-other" => worker with { Scheduling = Scheduling() with { Mode = "targeted", TargetItemId = "local:other", ItemLimit = 1 } }, + "targeted-same" => worker with { Scheduling = Scheduling() with { Mode = "targeted", TargetItemId = "local:1", ItemLimit = 1 } }, + "drain" => worker with { State = WorkerInstanceState.Draining }, + "legacy" => worker with { Scheduling = null }, + "missing-progress" => worker with { Progress = null }, + "drift" => worker with { ConfigurationRevision = "changed" }, + "expired" => worker with { Progress = new(0, Now.AddHours(-1)), Scheduling = Scheduling() with { IdleTimeout = TimeSpan.FromMinutes(1) } }, + "same-item" => worker with { CurrentItemId = "local:1", State = WorkerInstanceState.RunningItem }, + _ => worker + }; + var liveness = scenario == "unknown" ? WorkerInstanceLiveness.Unknown : WorkerInstanceLiveness.Running; + var result = WorkerPickupPolicy.AssessRegistration(new(worker, liveness, null), new("local:1"), "revision", Now); + Assert.NotNull(result); + Assert.Equal(outcome, result.Outcome); + Assert.Equal(code, result.Code); + Assert.Equal(scenario == "same-item", result.AlreadyProcessing); + } + + [Fact] + public void Continuous_busy_worker_still_requires_item_assessment_and_reports_remaining_allowance() + { + var worker = Instance() with { CurrentItemId = "local:other", State = WorkerInstanceState.RunningItem, + Scheduling = Scheduling() with { Mode = "bounded", ItemLimit = 5 }, Progress = new(2, Now) }; + var status = new WorkerInstanceStatus(worker, WorkerInstanceLiveness.Running, null); + Assert.Null(WorkerPickupPolicy.AssessRegistration(status, new("local:1"), "revision", Now)); + var row = WorkerDiscoveryEntry.From(status, "revision"); + Assert.Equal(2, row.RemainingItemAllowance); + Assert.Equal("open", row.Intake); + } + + [Fact] + public void Workspace_lock_inspection_never_creates_a_file() + { + var locks = new FileWorkspaceExecutionLock(Path.Combine(root, "locks")); + Assert.Equal("available", locks.Inspect(root).State); + Assert.False(Directory.Exists(root)); + } + + [Fact] + public async Task Existing_busy_workspace_is_not_advertised_as_available() + { + var locks = new FileWorkspaceExecutionLock(Path.Combine(root, "locks")); + await using var lease = await locks.AcquireAsync(root, default); + Assert.Equal("unknown", locks.Inspect(root).State); + } + + [Theory] + [InlineData(WorkerInstanceState.Draining, "draining")] + [InlineData(WorkerInstanceState.StoppingNow, "stopping")] + [InlineData(WorkerInstanceState.Finalizing, "stopping")] + public void Discovery_projects_closed_intake(WorkerInstanceState state, string expected) + { + var entry = WorkerDiscoveryEntry.From(new(Instance() with { State = state }, WorkerInstanceLiveness.Running, null), "revision"); + Assert.Equal(expected, entry.Intake); + } + + [Fact] + public void Startup_capture_copies_filters_and_retains_targeted_limits() + { + var filters = new Dictionary { ["priority"] = "P1" }; + var options = new WorkerOptions(null, false, 5, WorkspaceMode.Worktree, filters, TimeSpan.FromMinutes(5), + TimeSpan.FromHours(1), FencedAction.Kill, null, "agent", false, true, Profile: "deep"); + var captured = WorkerScheduling.From(new TrackerConfig(), options, + new(root, ConfigPath, "revision", "display", WorkerHostKind.CliProcess), new(new WorkItemId("local:1"))); + filters["priority"] = "P2"; + Assert.Equal("P1", captured.Filters["priority"]); + Assert.Equal("targeted", captured.Mode); + Assert.Equal("local:1", captured.TargetItemId); + Assert.Equal(1, captured.ItemLimit); + Assert.Equal("deep", captured.Profile); + Assert.True(captured.IsUsable()); + } + + private JsonWorkerInstanceRegistry Registry() => new(new CachePaths(root), () => Now, _ => new(true, "start")); + private WorkerInstance Instance() => new("run", 123, "start", Now, Now, + JsonWorkerInstanceRegistry.ConfigurationPathHash(ConfigPath), "revision", "1", "not parsed", null, + WorkerInstanceState.Idle, Scheduling: Scheduling(), Progress: new(0, Now)); + private WorkerScheduling Scheduling() => new("continuous", null, WorkerItemIntent.Auto, "Ideas", "Doing", null, + "claude", WorkspaceMode.Current, root, new Dictionary(), null, null, TimeSpan.FromHours(1), null, false); + private async Task WriteRecord(WorkerInstance instance) + { + Directory.CreateDirectory(RecordDirectory); + await File.WriteAllTextAsync(Path.Combine(RecordDirectory, instance.RunId + ".json"), + JsonSerializer.Serialize(instance, JsonOptions)); + } + public void Dispose() + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } +} diff --git a/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerPickupIntegrationTests.cs b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerPickupIntegrationTests.cs new file mode 100644 index 00000000..b1c74384 --- /dev/null +++ b/tests/Highbyte.Wrighty.UnitTests/Workers/WorkerPickupIntegrationTests.cs @@ -0,0 +1,134 @@ +using Highbyte.Wrighty.AgentContext; +using Highbyte.Wrighty.Backends; +using Highbyte.Wrighty.Caching; +using Highbyte.Wrighty.Claims; +using Highbyte.Wrighty.Configuration; +using Highbyte.Wrighty.LocalMarkdown; +using Highbyte.Wrighty.Models; +using Highbyte.Wrighty.Processes; +using Highbyte.Wrighty.Workers; + +namespace Highbyte.Wrighty.UnitTests.Workers; + +public sealed partial class LocalDispatchStateTests +{ + [Theory] + [InlineData("claude")] + [InlineData("codex")] + [InlineData("copilot")] + [InlineData("opencode")] + public async Task Pickup_assessment_resolves_item_agent_without_launching_or_mutating(string agent) + { + var config = WorkerConfig(); + var backend = new LocalMarkdownTrackerBackend(new FakeIdentity(), clock); + await backend.InitializeAsync(config, false, default); + var created = await backend.CreateAsync(config, new CreateWorkItemOperation( + new("Assess", "Requirements", "Todo", "P1", AutomaticExecutionAllowed: true, AgentPolicy: agent), false), default); + var tracker = new TrackerService(new TrackerBackendRegistry([backend])); + var registry = BuiltInAgentRegistry.Create(new PathExecutableResolver()); + var worker = new WorkerService(tracker, new FailIfRunRunner(), new DiscoveryWorkspaces(), + registry.ExecutionAdapters, clock: () => clock.UtcNow); + var before = TrackerContents(); + var state = await tracker.GetOperationalAsync(config, created.Id, default); + var status = PickupRun(config); + var result = await worker.AssessPickupAsync(config, status, state, "revision", clock.UtcNow, default); + Assert.Equal("could-pick-up", result.Outcome); + Assert.Equal(agent, result.Agent); + Assert.Equal(before, TrackerContents()); + Assert.Equal(ClaimOwnershipState.Unclaimed, (await backend.GetClaimOwnershipAsync(config, created.Id, default)).State); + } + + [Theory] + [InlineData("paused", "cannot-pick-up")] + [InlineData("queued", "could-pick-up")] + [InlineData("claimed", "cannot-pick-up")] + [InlineData("provider", "cannot-pick-up")] + [InlineData("missing-workspace", "unknown")] + public async Task Pickup_assessment_reuses_retained_session_rules_and_cached_capacity(string scenario, string outcome) + { + var (backend, config, id, _) = await CreatePausedItemAsync(); + if (scenario != "paused") await backend.QueuePausedAsync(config, id, default); + if (scenario == "claimed") + await backend.TryClaimAsync(config, id, new AgentExecutionContext("codex", null, AgentContextSource.ExplicitOption, + ClaimantKind: ClaimantKind.Agent, ClaimantId: "agent:another"), default); + var provider = new JsonProviderCapacityStore(new CachePaths(Path.Combine(directory, "capacity"))); + if (scenario == "provider") + await provider.RecordUnavailableAsync("codex", "Quota window", clock.UtcNow.AddHours(1), + AgentFailureConfidence.Authoritative, clock.UtcNow, default); + var tracker = new TrackerService(new TrackerBackendRegistry([backend])); + var worker = new WorkerService(tracker, new FailIfRunRunner(), new DiscoveryWorkspaces(), + [new CodexAgentAdapter()], clock: () => clock.UtcNow, providerCapacityStore: provider); + var status = PickupRun(config); + if (scenario == "missing-workspace") status = status with { Instance = status.Instance with + { Scheduling = status.Instance.Scheduling! with { RepositoryPath = Path.Combine(directory, "missing") } } }; + var before = TrackerContents(); + var result = await worker.AssessPickupAsync(config, status, await tracker.GetOperationalAsync(config, id, default), + "revision", clock.UtcNow, default); + Assert.Equal(outcome, result.Outcome); + if (scenario == "provider") Assert.Equal("PROVIDER_DEFERRED", result.Code); + Assert.Equal(before, TrackerContents()); + } + + [Fact] + public async Task Worker_host_records_effective_startup_and_loop_allowance() + { + var config = WorkerConfig(); + var backend = new LocalMarkdownTrackerBackend(new FakeIdentity(), clock); + await backend.InitializeAsync(config, false, default); + await backend.CreateAsync(config, new CreateWorkItemOperation( + new("Bounded run", "Body", "Todo", "P1", AutomaticExecutionAllowed: true, AgentPolicy: "claude"), false), default); + var worker = new WorkerService(new TrackerService(new TrackerBackendRegistry([backend])), + new CapturingRejectedRunner(), new TrackingWorktree(directory), [new ClaudeAgentAdapter()], clock: () => clock.UtcNow); + var registry = new ProgressRegistry(); + var host = new WorkerRunHost(worker, registry); + using var control = new WorkerRunControl(); + var options = Options() with { Once = false, MaxItems = 1, Agent = null, IdleTimeout = TimeSpan.FromSeconds(1) }; + await host.RunAsync(config, options, new(directory, config.SourcePath!, "revision", "unparsed", WorkerHostKind.WebHosted), + new(null), control, new(_ => Task.CompletedTask, null), default); + Assert.NotNull(registry.Metadata?.Scheduling); + Assert.Equal("bounded", registry.Metadata.Scheduling.Mode); + Assert.Equal(1, registry.Metadata.Scheduling.ItemLimit); + Assert.Null(registry.Metadata.Scheduling.AgentOverride); + Assert.Equal([0, 1], registry.Progress.Select(value => value.Processed)); + } + + private WorkerInstanceStatus PickupRun(TrackerConfig config) => new(new("run", 123, "start", clock.UtcNow, + clock.UtcNow, "scope", "revision", "1", "display only", null, WorkerInstanceState.Idle, + Scheduling: new("continuous", null, WorkerItemIntent.Auto, config.DefaultPickFrom, config.DefaultPickTo, + null, "claude", WorkspaceMode.Current, directory, new Dictionary(), null, null, + TimeSpan.FromHours(1), null, false), Progress: new(0, clock.UtcNow)), WorkerInstanceLiveness.Running, null); + private Dictionary TrackerContents() => Directory.GetFiles(directory, "*.md", SearchOption.AllDirectories) + .Concat(Directory.GetFiles(directory, "state.json", SearchOption.AllDirectories)) + .ToDictionary(path => path, File.ReadAllText); + + private sealed class DiscoveryWorkspaces : IWorkspaceManager + { + public Task PrepareAsync(WorkspaceRequest request, CancellationToken cancellationToken) => + throw new InvalidOperationException("Discovery must not prepare a workspace."); + public Task CleanupAsync(Workspace workspace, CancellationToken cancellationToken) => + throw new InvalidOperationException("Discovery must not clean up a workspace."); + } + + private sealed class ProgressRegistry : IWorkerInstanceRegistry, IWorkerInstanceRegistration + { + public WorkerRegistrationMetadata? Metadata { get; private set; } + public List Progress { get; } = []; + public string RunId => "run"; + public Task RegisterAsync(string configurationPath, string configurationRevision, string invocationSummary, CancellationToken cancellationToken) => + throw new InvalidOperationException("Expected structured registration."); + public Task RegisterAsync(string configurationPath, string configurationRevision, string invocationSummary, + WorkerRegistrationMetadata metadata, CancellationToken cancellationToken) + { + Metadata = metadata; + return Task.FromResult(this); + } + public Task> ListAsync(string configurationPath, CancellationToken cancellationToken) => throw new NotSupportedException(); + public Task UpdateAsync(string? currentItemId, WorkerInstanceState state, CancellationToken cancellationToken) => Task.CompletedTask; + public Task UpdateProgressAsync(WorkerRunProgress progress, CancellationToken cancellationToken) + { + Progress.Add(progress); + return Task.CompletedTask; + } + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +}