feat: validate single-client broadcast operations
Build web service image / build (push) Successful in 1m9s

This commit is contained in:
2026-09-19 14:33:26 +08:00
parent e321d38fa3
commit c7c0ab273f
62 changed files with 2861 additions and 12651 deletions
+123 -8
View File
@@ -133,19 +133,134 @@ public sealed class AgentService(IAgentBackend backend, ServiceSecurity security
public Task<Page<MessageInfo>> MessagesAsync(string? session, int limit, int offset, bool includeContent, CancellationToken ct) =>
MessagesAsync(null, session, limit, offset, includeContent, ct);
public OperationRecord SubmitOperation(OperationSubmitRequest request, CancellationToken ct)
public async Task<OperationRecord> SubmitOperationAsync(OperationSubmitRequest request, CancellationToken ct)
{
if (!string.Equals(request.Kind, "send-text", StringComparison.Ordinal))
throw new ServiceException("UnsupportedOperation", 400, "Only send-text is available in this phase.");
var kind = RequireText(request.Kind, "kind", 80);
var accountId = RequireText(request.AccountId, "accountId", 200);
var targetId = RequireText(request.TargetId, "targetId", 512);
var text = PrepareText(request.Text);
var idempotencyKey = RequireText(request.IdempotencyKey, "idempotencyKey", 128);
if (!request.Confirmed) throw new ServiceException("ConfirmationRequired", 409, "Explicit send confirmation is required.");
var capability = RequireCapability("send-text");
var canonical = System.Text.Json.JsonSerializer.Serialize(new { request.Kind, accountId, targetId, text });
return operations.Submit(Identity, accountId, capability, idempotencyKey, canonical,
cancellation => backend.SendTextAsync(accountId, targetId, text, cancellation));
if (string.Equals(kind, "send-text", StringComparison.Ordinal))
{
var targetId = RequireText(request.TargetId, "targetId", 512);
var capability = RequireCapability("send-text");
var canonical = System.Text.Json.JsonSerializer.Serialize(new { kind, accountId, targetId, text });
return operations.Submit(Identity, accountId, capability, idempotencyKey, canonical,
cancellation => backend.SendTextAsync(accountId, targetId, text, cancellation));
}
if (!string.Equals(kind, "broadcast-text", StringComparison.Ordinal))
throw new ServiceException("UnsupportedOperation", 400, "Only send-text and broadcast-text are available in this phase.");
if (!string.IsNullOrWhiteSpace(request.TargetId))
throw new ServiceException("InvalidRequest", 400, "broadcast-text uses the targets list, not targetId.");
var capabilityForBroadcast = RequireCapability("broadcast-text");
var targets = await FreezeBroadcastTargetsAsync(accountId, request.Targets, ct).ConfigureAwait(false);
var initialDetails = System.Text.Json.JsonSerializer.Serialize(new
{
targetIds = targets,
stopOnError = request.StopOnError
}, ServiceHost.Json);
var canonicalBroadcast = System.Text.Json.JsonSerializer.Serialize(new
{
kind,
accountId,
targetIds = targets,
text,
stopOnError = request.StopOnError
});
return operations.SubmitResult(Identity, accountId, capabilityForBroadcast, idempotencyKey, canonicalBroadcast,
initialDetails,
cancellation => ExecuteBroadcastAsync(accountId, targets, text, request.StopOnError, cancellation));
}
private async Task<IReadOnlyList<string>> FreezeBroadcastTargetsAsync(string accountId, IReadOnlyList<string>? requested, CancellationToken ct)
{
if (requested is null || requested.Count is < 1 or > 20)
throw new ServiceException("InvalidRequest", 400, "broadcast-text requires 1..20 targets.");
var targets = requested
.Select(target => RequireText(target, "target", 512))
.Distinct(StringComparer.Ordinal)
.ToArray();
if (targets.Length == 0) throw new ServiceException("InvalidRequest", 400, "broadcast-text requires at least one unique target.");
// Freeze the current UI target list before enqueueing. The account binding and identity are
// still revalidated by every side-effecting send, so this preflight cannot bypass account safety.
var visible = await backend.SessionsAsync(ct).ConfigureAwait(false);
var visibleIds = visible.Select(session => session.AutomationId).ToHashSet(StringComparer.Ordinal);
var missing = targets.Where(target => !visibleIds.Contains(target)).ToArray();
if (missing.Length > 0)
throw new ServiceException("TargetNotFound", 409, "Every broadcast target must be a visible, uniquely bound session.");
return targets;
}
private async Task<OperationExecutionResult> ExecuteBroadcastAsync(string accountId, IReadOnlyList<string> targets,
string text, bool stopOnError, CancellationToken ct)
{
var items = new List<BroadcastItemResult>(targets.Count);
var stopped = false;
string? stopReason = null;
string? errorCode = null;
var unconfirmed = false;
foreach (var target in targets)
{
try
{
await backend.SendTextAsync(accountId, target, text, ct).ConfigureAwait(false);
items.Add(new BroadcastItemResult(target, "Succeeded", null));
}
catch (OperationCanceledException)
{
items.Add(new BroadcastItemResult(target, "Unconfirmed", "Cancelled"));
stopped = true;
stopReason = "Cancelled";
errorCode = "Cancelled";
unconfirmed = true;
break;
}
catch (ServiceException exception)
{
var itemState = exception.Code == "ResultUnconfirmed" ? "Unconfirmed" : "Failed";
items.Add(new BroadcastItemResult(target, itemState, exception.Code));
errorCode ??= exception.Code == "ResultUnconfirmed" ? "ResultUnconfirmed" : "BroadcastItemFailed";
if (itemState == "Unconfirmed" || stopOnError)
{
stopped = true;
stopReason = exception.Code;
unconfirmed = itemState == "Unconfirmed";
break;
}
}
catch (WxAgentException exception)
{
var code = exception.Code.ToString();
var itemState = exception.Code == WxAgentErrorCode.ResultUnconfirmed ? "Unconfirmed" : "Failed";
items.Add(new BroadcastItemResult(target, itemState, code));
errorCode ??= itemState == "Unconfirmed" ? "ResultUnconfirmed" : "BroadcastItemFailed";
if (itemState == "Unconfirmed" || stopOnError)
{
stopped = true;
stopReason = code;
unconfirmed = itemState == "Unconfirmed";
break;
}
}
catch (Exception)
{
items.Add(new BroadcastItemResult(target, "Unconfirmed", "ExecutionFailed"));
stopped = true;
stopReason = "ExecutionFailed";
errorCode = "ExecutionFailed";
unconfirmed = true;
break;
}
}
var details = System.Text.Json.JsonSerializer.Serialize(new BroadcastResult(targets, items, stopped, stopReason), ServiceHost.Json);
return new OperationExecutionResult(details, errorCode, unconfirmed);
}
private static string RequireText(string? value, string name, int maxLength)
+6 -2
View File
@@ -57,9 +57,13 @@ public sealed class AgentTools(AgentService service)
[McpServerTool(Name = "operations_list", ReadOnly = true), Description("List bounded operations owned by the current identity; accountId is a caller-side view filter, not an authorization boundary.")]
public Task<CallToolResult> Operations(string? accountId = null, int limit = 50, int offset = 0) => Result(() => Task.FromResult<object>(service.Operations(accountId, limit, offset)));
[McpServerTool(Name = "operation_submit"), Description("Submit an explicitly confirmed operation. send-text remains disabled until Windows validation is complete.")]
[McpServerTool(Name = "operation_submit"), Description("Submit one explicitly confirmed send-text operation. The targetId must be a frozen visible session automation ID.")]
public Task<CallToolResult> SubmitOperation(string kind, string accountId, string targetId, string? text, string idempotencyKey, bool confirmed = false, CancellationToken cancellationToken = default) =>
Result(() => Task.FromResult<object>(service.SubmitOperation(new OperationSubmitRequest(kind, accountId, targetId, text, idempotencyKey, confirmed), cancellationToken)));
Result(async () => await service.SubmitOperationAsync(new OperationSubmitRequest(kind, accountId, targetId, text, idempotencyKey, confirmed), cancellationToken));
[McpServerTool(Name = "broadcast_text"), Description("Submit an explicitly confirmed broadcast-text operation for a frozen list of 1..20 visible session automation IDs. Sends sequentially, stops on the first known failure by default, and returns per-target results; replaying the same idempotency key never replays terminal writes.")]
public Task<CallToolResult> BroadcastText(string accountId, string[] targets, string text, string idempotencyKey, bool confirmed = false, bool stopOnError = true, CancellationToken cancellationToken = default) =>
Result(async () => await service.SubmitOperationAsync(new OperationSubmitRequest("broadcast-text", accountId, null, text, idempotencyKey, confirmed, targets, stopOnError), cancellationToken));
[McpServerTool(Name = "operation_get", ReadOnly = true), Description("Read one operation owned by the current identity; terminal writes are never replayed.")]
public Task<CallToolResult> Operation(string operationId) => Result(() => Task.FromResult<object>(service.Operation(operationId)));
+23 -5
View File
@@ -6,11 +6,13 @@ using Microsoft.Extensions.Hosting;
namespace WxAgent.Service;
public sealed record OperationExecutionResult(string? Details = null, string? ErrorCode = null, bool Unconfirmed = false);
public sealed class OperationQueue(OperationStore store, ServiceSecurity security) : BackgroundService
{
private const int QueueCapacity = 100;
private sealed record Work(OperationRecord Record, ServiceIdentity Identity, string Permission,
Func<CancellationToken, Task> Action, CancellationTokenSource Cancel);
Func<CancellationToken, Task<OperationExecutionResult>> Action, CancellationTokenSource Cancel);
private readonly Channel<Work> queue = Channel.CreateBounded<Work>(new BoundedChannelOptions(QueueCapacity)
{ SingleReader = true, FullMode = BoundedChannelFullMode.Wait });
private readonly ConcurrentDictionary<string, CancellationTokenSource> cancellations = new();
@@ -18,7 +20,17 @@ public sealed class OperationQueue(OperationStore store, ServiceSecurity securit
private bool stopping;
public OperationRecord Submit(ServiceIdentity identity, string accountId, AgentCapability capability,
string? idempotencyKey, string canonicalParameters, Func<CancellationToken, Task> action)
string? idempotencyKey, string canonicalParameters, Func<CancellationToken, Task> action) =>
SubmitResult(identity, accountId, capability, idempotencyKey, canonicalParameters, null,
async cancellationToken =>
{
await action(cancellationToken).ConfigureAwait(false);
return new OperationExecutionResult();
});
public OperationRecord SubmitResult(ServiceIdentity identity, string accountId, AgentCapability capability,
string? idempotencyKey, string canonicalParameters, string? initialDetails,
Func<CancellationToken, Task<OperationExecutionResult>> action)
{
security.RequireCurrent(identity, capability.Permission, accountId);
if (!capability.Enabled) throw new ServiceException("CapabilityDisabled", 409, capability.DisabledReason ?? "Capability unavailable.");
@@ -30,7 +42,7 @@ public sealed class OperationQueue(OperationStore store, ServiceSecurity securit
{
if (stopping) throw new ServiceException("Unavailable", 503, "Agent is stopping.");
var (record, created) = store.Enqueue(identity.PrincipalId, accountId, capability.Operation,
idempotencyKey, digest, capability.HasSideEffects, TimeSpan.FromSeconds(capability.TimeoutSeconds), QueueCapacity);
idempotencyKey, digest, capability.HasSideEffects, TimeSpan.FromSeconds(capability.TimeoutSeconds), QueueCapacity, initialDetails);
if (!created) return record;
var cancel = new CancellationTokenSource();
cancellations[record.Id] = cancel;
@@ -92,10 +104,16 @@ public sealed class OperationQueue(OperationStore store, ServiceSecurity securit
started = true;
}
// Never release this slot with WaitAsync: the actual action must have stopped first.
await work.Action(budget.Token);
var result = await work.Action(budget.Token);
if (result.ErrorCode is not null)
{
store.Transition(work.Record.Id, result.Unconfirmed ? "Unconfirmed" : "Failed",
"execution", result.ErrorCode, result.Details);
continue;
}
budget.Token.ThrowIfCancellationRequested();
security.RequireCurrent(work.Identity, work.Permission, work.Record.AccountId);
store.Transition(work.Record.Id, "Succeeded", "complete");
store.Transition(work.Record.Id, "Succeeded", "complete", details: result.Details);
}
catch (Exception e)
{
+12 -7
View File
@@ -4,7 +4,7 @@ namespace WxAgent.Service;
public sealed record OperationRecord(string Id, string PrincipalId, string AccountId, string Capability,
string State, string Stage, string CorrelationId, DateTimeOffset CreatedAt, DateTimeOffset ExpiresAt,
string? ErrorCode, bool HasSideEffects);
string? ErrorCode, bool HasSideEffects, string? Details = null);
public sealed record OperationSummary(string Id, string AccountId, string Capability,
string State, string Stage, string CorrelationId, DateTimeOffset CreatedAt, DateTimeOffset ExpiresAt,
@@ -37,10 +37,12 @@ public sealed class OperationStore : IDisposable
stage='restart', error='AgentRestarted' WHERE state='Running';
UPDATE operations SET state='Cancelled', stage='restart', error='AgentRestarted' WHERE state='Queued';
""");
try { Execute("ALTER TABLE operations ADD COLUMN details TEXT"); }
catch (SqliteException exception) when (exception.SqliteErrorCode == 1) { }
}
public (OperationRecord Record, bool Created) Enqueue(string principal, string account, string capability,
string? idempotency, string digest, bool sideEffects, TimeSpan budget, int capacity)
string? idempotency, string digest, bool sideEffects, TimeSpan budget, int capacity, string? details = null)
{
lock (gate)
{
@@ -67,10 +69,10 @@ public sealed class OperationStore : IDisposable
throw new ServiceException("QueueFull", 429, "Agent queue is full.");
var now = DateTimeOffset.UtcNow;
var operation = new OperationRecord(Guid.NewGuid().ToString("N"), principal, account, capability,
"Queued", "queued", Guid.NewGuid().ToString("N"), now, now.Add(budget), null, sideEffects);
"Queued", "queued", Guid.NewGuid().ToString("N"), now, now.Add(budget), null, sideEffects, details);
using var insert = database.CreateCommand();
insert.Transaction = transaction;
insert.CommandText = "INSERT INTO operations VALUES ($id,$p,$a,$c,'Queued','queued',$correlation,$created,$expires,NULL,$effects,$key,$digest)";
insert.CommandText = "INSERT INTO operations (id,principal,account,capability,state,stage,correlation,created,expires,error,side_effects,idempotency,digest,details) VALUES ($id,$p,$a,$c,'Queued','queued',$correlation,$created,$expires,NULL,$effects,$key,$digest,$details)";
insert.Parameters.AddWithValue("$id", operation.Id);
insert.Parameters.AddWithValue("$p", principal);
insert.Parameters.AddWithValue("$a", account);
@@ -81,6 +83,7 @@ public sealed class OperationStore : IDisposable
insert.Parameters.AddWithValue("$effects", sideEffects ? 1 : 0);
insert.Parameters.AddWithValue("$key", (object?)idempotency ?? DBNull.Value);
insert.Parameters.AddWithValue("$digest", digest);
insert.Parameters.AddWithValue("$details", (object?)details ?? DBNull.Value);
insert.ExecuteNonQuery();
transaction.Commit();
return (operation, true);
@@ -122,16 +125,17 @@ public sealed class OperationStore : IDisposable
}
}
public void Transition(string id, string state, string stage, string? error = null)
public void Transition(string id, string state, string stage, string? error = null, string? details = null)
{
lock (gate)
{
using var command = database.CreateCommand();
command.CommandText = "UPDATE operations SET state=$state,stage=$stage,error=$error WHERE id=$id AND state IN ('Queued','Running')";
command.CommandText = "UPDATE operations SET state=$state,stage=$stage,error=$error,details=COALESCE($details,details) WHERE id=$id AND state IN ('Queued','Running')";
command.Parameters.AddWithValue("$id", id);
command.Parameters.AddWithValue("$state", state);
command.Parameters.AddWithValue("$stage", stage);
command.Parameters.AddWithValue("$error", (object?)error ?? DBNull.Value);
command.Parameters.AddWithValue("$details", (object?)details ?? DBNull.Value);
command.ExecuteNonQuery();
}
}
@@ -139,7 +143,8 @@ public sealed class OperationStore : IDisposable
private static OperationRecord Read(SqliteDataReader reader) => new(reader.GetString(0), reader.GetString(1), reader.GetString(2),
reader.GetString(3), reader.GetString(4), reader.GetString(5), reader.GetString(6),
DateTimeOffset.Parse(reader.GetString(7), System.Globalization.CultureInfo.InvariantCulture),
DateTimeOffset.Parse(reader.GetString(8), System.Globalization.CultureInfo.InvariantCulture), reader.IsDBNull(9) ? null : reader.GetString(9), reader.GetInt64(10) != 0);
DateTimeOffset.Parse(reader.GetString(8), System.Globalization.CultureInfo.InvariantCulture), reader.IsDBNull(9) ? null : reader.GetString(9), reader.GetInt64(10) != 0,
reader.IsDBNull(13) ? null : reader.GetString(13));
private void Execute(string sql)
{
@@ -5,7 +5,20 @@ public sealed record AccountInfo(string AccountId, string? DisplayName, string?
public sealed record SessionInfo(string Name, string AutomationId, bool IsCurrent);
public sealed record MessageInfo(string Fingerprint, string Type, string? Sender, string? Summary, string? Content);
public sealed record ListRequest(int Limit = 50, int Offset = 0, bool IncludeContent = false, string? AccountId = null, string? Session = null);
public sealed record OperationSubmitRequest(string? Kind, string? AccountId, string? TargetId, string? Text, string? IdempotencyKey, bool Confirmed);
public sealed record OperationSubmitRequest(
string? Kind,
string? AccountId,
string? TargetId,
string? Text,
string? IdempotencyKey,
bool Confirmed,
IReadOnlyList<string>? Targets = null,
bool StopOnError = true);
public sealed record BroadcastItemResult(string TargetId, string State, string? ErrorCode);
public sealed record BroadcastResult(IReadOnlyList<string> TargetIds, IReadOnlyList<BroadcastItemResult> Items,
bool Stopped, string? StopReason);
public sealed class ReadOnlyRequest
{
+1 -1
View File
@@ -137,7 +137,7 @@ public static class ServiceHost
return Results.Ok(await service.UploadAsync(form.Files[0], ct));
});
app.MapGet("/api/v1/files/{id}", (string id, AgentService service) => Results.File(service.Download(id), "application/octet-stream"));
app.MapPost("/api/v1/operations", (OperationSubmitRequest request, AgentService service, CancellationToken ct) => service.SubmitOperation(request, ct));
app.MapPost("/api/v1/operations", async (OperationSubmitRequest request, AgentService service, CancellationToken ct) => await service.SubmitOperationAsync(request, ct));
app.MapGet("/api/v1/operations", (string? accountId, int? limit, int? offset, AgentService service) => service.Operations(accountId, limit ?? 50, offset ?? 0));
app.MapGet("/api/v1/operations/{id}", (string id, AgentService service) => service.Operation(id));
app.MapPost("/api/v1/operations/{id}/cancel", (string id, AgentService service) => service.CancelOperation(id));
@@ -22,6 +22,9 @@ public sealed class ServiceOptions
public RemoteAgentOptions? Remote { get; init; }
public ReportingConfig Reporting { get; init; } = new();
public string? RemoteConfigurationFile { get; init; }
// Explicitly opt-in for a single, user-authorized Windows validation session.
// Production deployments remain read-only unless this local gate is enabled.
public bool EnableValidationOperations { get; init; }
// Kept only so older service.json files can be loaded and rewritten by the tray.
[JsonIgnore]