diff --git a/samples/eve-browser-network/Program.cs b/samples/eve-browser-network/Program.cs index 21933bda..25809744 100644 --- a/samples/eve-browser-network/Program.cs +++ b/samples/eve-browser-network/Program.cs @@ -208,7 +208,29 @@ static async Task RunHeadlessAsync(Args arguments) Console.WriteLine("HEADLESS_READY " + JsonSerializer.Serialize(new { count = initial.Count })); } if (initial.Count >= arguments.ExpectedCount) completion.TrySetResult(); - await completion.Task.WaitAsync(TimeSpan.FromSeconds(45)); + var commandTask = Task.Run(async () => + { + var line = await Console.In.ReadLineAsync(); + if (line == null || !line.StartsWith("INVOKE ", StringComparison.Ordinal)) + throw new InvalidOperationException("The headless sample expected 'INVOKE ' on stdin."); + var commandId = line["INVOKE ".Length..].Trim(); + var result = await mesh.InvokeAsync( + arguments.VerseId, + "sample.counter", + "sample.counter.increment", + "sample.increment.v1", + "gamecult.eve.command_receipt.v1", + new IncrementRequest { Amount = 1 }, + sourceRuntimeId: "sample.csharp", + idempotencyKey: commandId); + Console.WriteLine("HEADLESS_RECEIPT " + JsonSerializer.Serialize(new + { + receiptId = result.Value.ReceiptId, + count = result.Value.Count, + status = result.Status + })); + }); + await Task.WhenAll(completion.Task, commandTask).WaitAsync(TimeSpan.FromSeconds(45)); } static EveSurfaceDocument CreateCounterSurface() diff --git a/samples/eve-browser-network/README.md b/samples/eve-browser-network/README.md index 5068ce0d..69720f7e 100644 --- a/samples/eve-browser-network/README.md +++ b/samples/eve-browser-network/README.md @@ -7,8 +7,9 @@ This is the real-process continuation of `samples/eve-two-runtime`: - one real Chromium page leases both documents through `cultmesh-browser` and lowers the provider-owned Eve surface; - one retained C# headless client discovers the provider through the same Odin - identity, observes the typed counter and canonical receipt ids, and survives - the provider route replacement without application reconnect code; + identity, observes the typed counter and canonical receipt ids, survives the + provider route replacement, and invokes through the generic CultMesh client + without application transport or reconnect code; - one separate local Odin fixture answers the canonical CultMesh Verse catalog over binary CultNet WebSocket messages; - the browser and C# clients speak the same binary CultNet schema-v0 WebSocket @@ -26,12 +27,12 @@ directly only to diagnose the network layer. Browser discovery is portable: `CHROME_PATH` wins when configured, then common Chrome/Chromium/Edge locations and Playwright's installed Chromium are checked. -The verifier clicks the Eve button twice with one idempotency key, requires one -state transition, restarts the provider on a different physical port, updates -the local Odin fixture, and requires the still-open browser to rediscover the -route, resubscribe, execute a second command, and preserve the durable count. -The retained C# lease independently rediscovers the same replacement route and -observes both canonical receipt ids. +The verifier clicks the Eve button twice with one idempotency key and requires +one state transition. It then restarts the provider on a different physical +port, updates the local Odin fixture, and requires both retained clients to +rediscover and resubscribe. The C# client invokes the second command through +`CultMeshClient.InvokeAsync`; Chromium observes the resulting count and both +clients observe both canonical receipt ids. The fixture speaks the production Verse-catalog contract; a deployment smoke against the full Odin daemon remains a separate infrastructure gate. diff --git a/scripts/verify-eve-browser-network.mjs b/scripts/verify-eve-browser-network.mjs index 7f80cb96..c82b7c50 100644 --- a/scripts/verify-eve-browser-network.mjs +++ b/scripts/verify-eve-browser-network.mjs @@ -99,10 +99,12 @@ try { await page.waitForFunction(() => window.__sampleConnectionStates?.includes("reconnecting") && window.__sampleConnectionStates.at(-1) === "connected"); - await page.evaluate(() => { window.__sampleCommandId = "browser-click-2"; }); - await page.locator("button").click(); - await page.waitForFunction(() => window.__sampleReceipt?.count === 2 && window.__sampleCount === 2); - const secondReceipt = await page.evaluate(() => window.__sampleReceipt?.receiptId); + headless.sendLine("INVOKE headless-command-2"); + const headlessReceipt = JSON.parse(await headless.waitFor("HEADLESS_RECEIPT ")); + assert.equal(headlessReceipt.status, "accepted"); + assert.equal(headlessReceipt.count, 2); + await page.waitForFunction(() => window.__sampleCount === 2); + const secondReceipt = headlessReceipt.receiptId; const secondHeadlessUpdate = JSON.parse(await headless.waitFor("HEADLESS_UPDATE_2 ")); assert.equal(secondHeadlessUpdate.count, 2); assert.ok(secondHeadlessUpdate.receiptIds.includes(firstReceipt)); @@ -151,7 +153,7 @@ async function startOdin(port, providerEndpoint) { function startDotnet(arguments_) { const process = spawn("dotnet", [ "run", "--no-build", "--project", join(sampleRoot, "EveBrowserNetworkSample.csproj"), "--", ...arguments_, - ], { cwd: repoRoot, stdio: ["ignore", "pipe", "pipe"] }); + ], { cwd: repoRoot, stdio: ["pipe", "pipe", "pipe"] }); let output = ""; let error = ""; const waiters = []; @@ -177,6 +179,9 @@ function startDotnet(arguments_) { } return { process, + sendLine(line) { + process.stdin.write(`${line}\n`); + }, waitFor(prefix, timeoutMs = 30_000) { const existing = output.split(/\r?\n/u).find(value => value.startsWith(prefix)); if (existing) return Promise.resolve(existing.slice(prefix.length)); diff --git a/src/GameCult.Mesh/CultMeshClient.cs b/src/GameCult.Mesh/CultMeshClient.cs index 783bb3e0..970e0f2f 100644 --- a/src/GameCult.Mesh/CultMeshClient.cs +++ b/src/GameCult.Mesh/CultMeshClient.cs @@ -6,6 +6,7 @@ using System.Threading; using System.Threading.Tasks; using GameCult.Caching; using GameCult.Networking; +using MessagePack; using R3; namespace GameCult.Mesh @@ -40,6 +41,45 @@ namespace GameCult.Mesh /// Gets or sets the deadline before an unanswered subscription intent is replayed. public TimeSpan SubscriptionResponseTimeout { get; set; } = TimeSpan.FromSeconds(2); + + /// Gets or sets the deadline for one correlated typed operation response. + public TimeSpan OperationResponseTimeout { get; set; } = TimeSpan.FromSeconds(10); + } + + /// + /// Carries one provider-authored operation response. Transport delivery does not by itself + /// establish that the provider accepted or committed the requested gameplay change. + /// + public sealed class CultMeshOperationResult + { + internal CultMeshOperationResult(CultNetOperationResponseMessage envelope, TResponse value) + { + MessageId = envelope.MessageId; + ServiceId = envelope.ServiceId; + Operation = envelope.Operation; + Status = envelope.Status; + PayloadSchema = envelope.PayloadSchema; + SourceRuntimeId = envelope.SourceRuntimeId; + Diagnostics = envelope.Diagnostics ?? Array.Empty(); + Value = value; + } + + /// Gets the durable caller-supplied idempotency key. + public string MessageId { get; } + /// Gets the provider service that answered. + public string ServiceId { get; } + /// Gets the operation that answered. + public string Operation { get; } + /// Gets the provider-authored operation status. + public string Status { get; } + /// Gets the portable response payload schema. + public string PayloadSchema { get; } + /// Gets the responding runtime identity, when supplied. + public string? SourceRuntimeId { get; } + /// Gets provider or transport diagnostics carried by the response. + public IReadOnlyList Diagnostics { get; } + /// Gets the decoded typed response payload. + public TResponse Value { get; } } /// Owns one shared live document subscription until disposed. @@ -85,6 +125,7 @@ namespace GameCult.Mesh private readonly CultMeshDiscoveryService _discovery; private readonly CultMeshSessionManager _sessions; private readonly TimeSpan _subscriptionResponseTimeout; + private readonly TimeSpan _operationResponseTimeout; private readonly ConcurrentDictionary _documents = new(StringComparer.Ordinal); private readonly ConcurrentDictionary _collections = new(StringComparer.Ordinal); private bool _disposed; @@ -94,6 +135,8 @@ namespace GameCult.Mesh if (options == null) throw new ArgumentNullException(nameof(options)); if (options.SubscriptionResponseTimeout <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(options.SubscriptionResponseTimeout)); + if (options.OperationResponseTimeout <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(options.OperationResponseTimeout)); var endpoints = options.RendezvousEndpoints .Where(endpoint => !string.IsNullOrWhiteSpace(endpoint)) .Distinct(StringComparer.Ordinal) @@ -123,6 +166,7 @@ namespace GameCult.Mesh options.RealtimeConnectors, options.Sessions); _subscriptionResponseTimeout = options.SubscriptionResponseTimeout; + _operationResponseTimeout = options.OperationResponseTimeout; } /// Connects to a stable endpoint identity using one application protocol. @@ -296,6 +340,98 @@ namespace GameCult.Mesh session.OpenSchemaClient().SendCultNet(message); } + /// + /// Invokes one typed provider operation through the reusable identity-owned session. + /// The caller supplies a durable idempotency key; the returned status and payload remain + /// provider-authored evidence rather than client-side authority over canonical state. + /// + public async Task> InvokeAsync( + string endpointId, + string serviceId, + string operation, + string requestSchema, + string responseSchema, + TRequest request, + string sourceRuntimeId, + string idempotencyKey, + CancellationToken cancellationToken = default) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(endpointId)) throw new ArgumentException("Endpoint identity is required.", nameof(endpointId)); + if (string.IsNullOrWhiteSpace(serviceId)) throw new ArgumentException("Service identity is required.", nameof(serviceId)); + if (string.IsNullOrWhiteSpace(operation)) throw new ArgumentException("Operation identity is required.", nameof(operation)); + if (string.IsNullOrWhiteSpace(requestSchema)) throw new ArgumentException("Request schema is required.", nameof(requestSchema)); + if (string.IsNullOrWhiteSpace(responseSchema)) throw new ArgumentException("Response schema is required.", nameof(responseSchema)); + if (request is null) throw new ArgumentNullException(nameof(request)); + if (string.IsNullOrWhiteSpace(sourceRuntimeId)) throw new ArgumentException("Source runtime identity is required.", nameof(sourceRuntimeId)); + if (string.IsNullOrWhiteSpace(idempotencyKey)) throw new ArgumentException("A durable idempotency key is required.", nameof(idempotencyKey)); + + var identity = endpointId.Trim(); + var expectedService = serviceId.Trim(); + var expectedOperation = operation.Trim(); + var expectedResponseSchema = responseSchema.Trim(); + var messageId = idempotencyKey.Trim(); + var session = await ConnectAsync(identity, CultMeshProtocols.Documents, cancellationToken).ConfigureAwait(false); + var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = session.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + completion.TrySetResult(response); + }); + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(_operationResponseTimeout); + using var cancellation = timeout.Token.Register(() => completion.TrySetCanceled(timeout.Token)); + + session.SendCultNet(new CultNetOperationRequestMessage + { + MessageId = messageId, + ServiceId = expectedService, + Operation = expectedOperation, + PayloadSchema = requestSchema.Trim(), + PayloadEncoding = "messagepack-base64", + Payload = Convert.ToBase64String(MessagePackSerializer.Serialize(request, CultNetSchemaMessageSerialization.Options)), + SourceRuntimeId = sourceRuntimeId.Trim() + }); + + CultNetOperationResponseMessage envelope; + try + { + envelope = await completion.Task.ConfigureAwait(false); + } + catch (TaskCanceledException) when (!cancellationToken.IsCancellationRequested) + { + throw new TimeoutException( + $"CultMesh operation '{expectedService}/{expectedOperation}' did not answer idempotency key '{messageId}' " + + $"within {_operationResponseTimeout}."); + } + cancellationToken.ThrowIfCancellationRequested(); + if (!string.Equals(envelope.ServiceId, expectedService, StringComparison.Ordinal) || + !string.Equals(envelope.Operation, expectedOperation, StringComparison.Ordinal)) + throw new InvalidOperationException("CultMesh operation response did not match the requested service and operation."); + if (!string.Equals(envelope.PayloadSchema, expectedResponseSchema, StringComparison.Ordinal)) + throw new InvalidOperationException( + $"CultMesh operation returned payload schema '{envelope.PayloadSchema}', expected '{expectedResponseSchema}'."); + if (!string.Equals(envelope.PayloadEncoding, "messagepack-base64", StringComparison.Ordinal)) + throw new InvalidOperationException( + $"CultMesh operation returned unsupported payload encoding '{envelope.PayloadEncoding}'."); + if (string.IsNullOrWhiteSpace(envelope.Status)) + throw new InvalidOperationException("CultMesh operation response omitted its provider-authored status."); + + TResponse value; + try + { + value = MessagePackSerializer.Deserialize( + Convert.FromBase64String(envelope.Payload), + CultNetSchemaMessageSerialization.Options); + } + catch (FormatException error) + { + throw new InvalidOperationException("CultMesh operation response payload was not valid base64.", error); + } + if (value is null) throw new InvalidOperationException("CultMesh operation response decoded to null."); + return new CultMeshOperationResult(envelope, value); + } + /// Gets the number of currently leased shared document subscriptions. public int ActiveDocumentResourceCount => _documents.Count; diff --git a/src/GameCult.Mesh/docs/browser-verse-transport.md b/src/GameCult.Mesh/docs/browser-verse-transport.md index 92c10844..58a00cae 100644 --- a/src/GameCult.Mesh/docs/browser-verse-transport.md +++ b/src/GameCult.Mesh/docs/browser-verse-transport.md @@ -178,10 +178,11 @@ Current executable coverage from `scripts/verify-eve-browser-network.mjs`: - Eve button to typed operation to provider receipt to shared state; - duplicate idempotency key produces one effect; - provider restart on a different physical port rehydrates the `.cc` state; -- the retained browser lease rediscovers through Odin, resubscribes, and - commits a second provider-authored receipt; -- the retained C# lease independently rediscovers, resubscribes, and observes - both canonical receipt ids without application reconnect code; +- the retained browser lease rediscovers through Odin and resubscribes; +- the retained C# client independently rediscovers, resubscribes, invokes the + second command through `CultMeshClient.InvokeAsync`, and receives the + provider-authored receipt without application transport or reconnect code; +- Chromium observes the C# command's canonical state and receipt chronology; - generated browser bundle contains no `node:` builtin import. The full public Verse gate still requires the deployed Odin daemon rather than @@ -199,10 +200,12 @@ From an empty temporary consumer using packed/released artifacts only: 6. observe the same state version and receipt id in both clients; 7. disconnect or rotate the physical WebSocket endpoint, then require both leases to resume without application reconnect code; -8. restart the provider and prove state plus receipt persistence; -9. reject malformed schemas, direct document writes, duplicate command +8. invoke from C# after route replacement and require Chromium to observe the + same provider-authored state and receipt; +9. restart the provider and prove state plus receipt persistence; +10. reject malformed schemas, direct document writes, duplicate command effects, stale subscription generations, and unsupported authority; -10. inspect the browser bundle and fail if it contains Node builtin imports. +11. inspect the browser bundle and fail if it contains Node builtin imports. The gate runs on Windows and Linux. A screenshot is useful presentation evidence, but the receipt/state chronology is the authority proof. diff --git a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md index 06547e74..8e7a8869 100644 --- a/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md +++ b/src/GameCult.Mesh/docs/getting-started/03-publish-an-eve-surface.md @@ -61,7 +61,9 @@ The second layer runs a durable C# provider, a separate local Odin fixture, a real Chromium Eve lowerer, and an independent C# headless observer over binary CultNet WebSocket lanes. The open browser resolves provider identity through the canonical Verse catalog, survives a provider restart on a different route, -resubscribes, and commits another canonical receipt. +and resubscribes. The retained C# client then invokes a second typed operation +through `CultMeshClient.InvokeAsync`; Chromium and C# observe the same +provider-authored state and receipt chronology. Current verification for the Unity consumer remains: diff --git a/src/GameCult.Mesh/docs/getting-started/README.md b/src/GameCult.Mesh/docs/getting-started/README.md index 44bd5e2c..e0bd2074 100644 --- a/src/GameCult.Mesh/docs/getting-started/README.md +++ b/src/GameCult.Mesh/docs/getting-started/README.md @@ -16,8 +16,9 @@ TypeScript packages into an empty temporary consumer, validates the Eve surface and receipt, executes one idempotent counter command, proves a DOM lowerer and headless observer converge, and reopens the durable `.cc` store. Then it boots a durable C# provider, local Odin fixture, real Chromium lowerer, and independent -C# observer; it proves discovery by identity, commands, canonical receipts, -provider restart on another route, resubscription, and durable convergence. +C# client. Chromium invokes before provider replacement; C# invokes afterward. +Both retained clients prove discovery by identity, canonical receipts, provider +restart on another route, resubscription, and durable convergence. Set `CHROME_PATH` only when no installed Chrome, Chromium, Edge, or Playwright Chromium can be found automatically. diff --git a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs index f16ee060..10af2417 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshClientTests.cs @@ -257,6 +257,37 @@ public sealed class CultMeshClientTests submitted.Document.SourceRole.Should().Be("pilot"); } + [Test] + public async Task InvokeOperation_CorrelatesProviderReceiptThroughStableIdentitySession() + { + var connector = new DocumentConnector(); + using var mesh = new CultMeshClient(new CultMeshClientOptions + { + RendezvousEndpoints = new[] { "rudp://odin:3076" }, + Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() }, + Connectors = new[] { connector } + }); + + var result = await mesh.InvokeAsync( + "aetheria", + "tests.counter", + "tests.counter.increment", + "tests.counter.increment_request.v1", + "tests.counter.increment_receipt.v1", + new ClientOperationRequest { Amount = 2 }, + sourceRuntimeId: "pilot-one", + idempotencyKey: "pilot-command-1"); + + result.MessageId.Should().Be("pilot-command-1"); + result.Status.Should().Be("accepted"); + result.SourceRuntimeId.Should().Be("provider-daemon"); + result.Value.ReceiptId.Should().Be("receipt:pilot-command-1"); + result.Value.Count.Should().Be(2); + connector.Clients.Should().ContainSingle(); + connector.Clients[0].Operations.Should().ContainSingle() + .Which.SourceRuntimeId.Should().Be("pilot-one"); + } + [Test] public async Task Dispose_TerminatesDocumentWaitingForInitialSnapshot() { @@ -360,6 +391,7 @@ public sealed class CultMeshClientTests private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth { private readonly List> _snapshots = new(); + private readonly List> _operationResponses = new(); private readonly CultNetDocumentRegistry _documents; private readonly string _text; private readonly TaskCompletionSource _failure = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -376,6 +408,7 @@ public sealed class CultMeshClientTests public int SubscribeCount { get; private set; } public int HandlerRegistrationCount { get; private set; } public List Submitted { get; } = new(); + public List Operations { get; } = new(); public void Connect(string host, int port) { } public void SendCultNet(T message) where T : ICultNetSchemaMessage { @@ -384,6 +417,32 @@ public sealed class CultMeshClientTests Submitted.Add(submitted); return; } + if (message is CultNetOperationRequestMessage operation) + { + Operations.Add(operation); + var decodedRequest = MessagePackSerializer.Deserialize( + Convert.FromBase64String(operation.Payload), + CultNetSchemaMessageSerialization.Options); + var receipt = new ClientOperationReceipt + { + ReceiptId = "receipt:" + operation.MessageId, + Count = decodedRequest.Amount + }; + var operationResponse = new CultNetOperationResponseMessage + { + MessageId = operation.MessageId, + ServiceId = operation.ServiceId, + Operation = operation.Operation, + Status = "accepted", + PayloadSchema = "tests.counter.increment_receipt.v1", + Payload = Convert.ToBase64String(MessagePackSerializer.Serialize( + receipt, + CultNetSchemaMessageSerialization.Options)), + SourceRuntimeId = "provider-daemon" + }; + foreach (var handler in _operationResponses.ToArray()) handler(operationResponse); + return; + } if (message is not CultNetDatabaseSubscribeMessage request) return; SubscribeCount++; if (_snapshotsToSuppress > 0) @@ -408,6 +467,8 @@ public sealed class CultMeshClientTests HandlerRegistrationCount++; if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage)) _snapshots.Add(response => callback((T)(object)response)); + if (typeof(T) == typeof(CultNetOperationResponseMessage)) + _operationResponses.Add(response => callback((T)(object)response)); } public void Dispose() { } public void Fail(Exception error) => _failure.TrySetResult(error); @@ -440,4 +501,17 @@ public sealed class CultMeshClientTests [Key(0), CultName] public string Id { get; set; } = string.Empty; [Key(1)] public string Text { get; set; } = string.Empty; } + + [MessagePackObject] + public sealed class ClientOperationRequest + { + [Key("amount")] public int Amount { get; set; } + } + + [MessagePackObject] + public sealed class ClientOperationReceipt + { + [Key("receiptId")] public string ReceiptId { get; set; } = string.Empty; + [Key("count")] public int Count { get; set; } + } }