Merge branch 'claude/vault-realtime-push-d64c61'

tests/DodoSSH.Api.Tests/ApiFixture.cs

@@ -1,4 +1,6 @@
 using System.Net.Http.Headers;
+using System.Net.WebSockets;
+using DodoSSH.Contracts;
 using DodoSSH.Infrastructure;
 using Microsoft.AspNetCore.Hosting;
 using Microsoft.AspNetCore.Mvc.Testing;
@@ -107,6 +109,34 @@ public sealed class ApiFixture : WebApplicationFactory<Program>, IAsyncLifetime
 
     /// <summary>Opens a database scope for arranging state and asserting on it.</summary>
     public AsyncServiceScope CreateScope() => Services.CreateAsyncScope();
+
+    /// <summary>
+    /// Opens the event socket as the given subject, through the real pipeline.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The bearer token goes on the upgrade request, which is the whole of the socket's authorization
+    /// — see ADR 0012 — so a test that stubbed it would be testing nothing. The subprotocol is offered
+    /// because the server refuses an upgrade that does not, and that refusal is itself under test.
+    /// </para>
+    /// <para>
+    /// <c>TestServer</c> speaks WebSockets in-memory with no port and no network, so these run
+    /// wherever the rest of the suite does.
+    /// </para>
+    /// </remarks>
+    public Task<WebSocket> ConnectEventsAsync(string subject, CancellationToken cancellationToken)
+    {
+        var token = IdentityProvider.MintToken(subject);
+        var client = Server.CreateWebSocketClient();
+
+        client.SubProtocols.Add(VaultEvents.SubProtocol);
+        // The server-side request, so the header is a raw string rather than a typed value.
+        client.ConfigureRequest = request => request.Headers.Authorization = $"Bearer {token}";
+
+        return client.ConnectAsync(
+            new Uri(Server.BaseAddress, VaultEvents.Path.TrimStart('/')),
+            cancellationToken);
+    }
 }
 
 /// <summary>Shares one host and container across every test class in the assembly.</summary>

tests/DodoSSH.Api.Tests/EndpointInventoryTests.cs

@@ -56,6 +56,12 @@ public sealed class EndpointInventoryTests(ApiFixture fixture)
         "POST /api/v1/vaults/{vaultId:guid}/sync/pull name=SyncPull tags=Sync policies=Enrolled anon=False",
         "POST /api/v1/vaults/{vaultId:guid}/sync/push name=SyncPush tags=Sync policies=Enrolled anon=False",
 
+        // The WebSocket, gated exactly as sync is and for the same reason — it announces changes to
+        // vaults, and a caller who could not read one has nothing to be told about. It appears here as
+        // an ordinary route because that is what it is until the upgrade: the bearer token authorises
+        // the handshake, unlike the relay's ticket. See ADR 0012.
+        "GET /api/v1/events name=VaultEvents tags=Events policies=Enrolled anon=False",
+
         // Enrolled, because the answer exists to be wrapped to and a caller with no key of their own has
         // nothing to wrap and no signature to attribute it with. There is no search here — see
         // DirectoryService for why an exact-match-only directory is a decision rather than a shortcut.

tests/DodoSSH.Api.Tests/EventsEndpointTests.cs

@@ -0,0 +1,461 @@
+using System.Net;
+using System.Net.WebSockets;
+using System.Text;
+using System.Text.Json;
+using DodoSSH.Contracts;
+using DodoSSH.Domain;
+using DodoSSH.Infrastructure;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace DodoSSH.Api.Tests;
+
+/// <summary>
+/// The push channel, over a real socket through the real authentication pipeline.
+/// </summary>
+/// <remarks>
+/// <para>
+/// The tests that matter most here are the two negatives: an unauthenticated upgrade is refused, and a
+/// change to somebody else's vault does not reach this socket. A push channel that leaked <em>which
+/// vault ids exist and when they change</em> would be a disclosure the pull path takes deliberate
+/// trouble to avoid — <c>SyncPullEndpoint</c> answers 404 rather than 403 for exactly that reason —
+/// and it would be invisible in a test that only checked that notices arrive.
+/// </para>
+/// <para>
+/// Ordering is asserted rather than absence-within-a-timeout wherever possible. "Nothing arrived in
+/// two seconds" is a test that passes on a slow machine for the wrong reason; "the first notice this
+/// socket saw was about its own vault, although another vault was written to first" is not.
+/// </para>
+/// </remarks>
+[Collection(ApiCollection.Name)]
+public sealed class EventsEndpointTests(ApiFixture fixture)
+{
+    private static readonly DateTimeOffset Now = new(2026, 8, 4, 12, 0, 0, TimeSpan.Zero);
+
+    /// <summary>
+    /// How long a test will wait for a frame before calling it a failure.
+    /// </summary>
+    /// <remarks>
+    /// Generous, because it is not a measurement: every wait here is for something already committed,
+    /// so the only thing this bounds is how long a genuinely broken build hangs before it reports.
+    /// </remarks>
+    private static readonly TimeSpan FrameTimeout = TimeSpan.FromSeconds(30);
+
+    // ---- The handshake ----
+
+    [Fact]
+    public async Task WithoutAToken_TheUpgradeIsRefused()
+    {
+        var client = fixture.Server.CreateWebSocketClient();
+        client.SubProtocols.Add(VaultEvents.SubProtocol);
+
+        var connecting = client.ConnectAsync(
+            new Uri(fixture.Server.BaseAddress, VaultEvents.Path.TrimStart('/')),
+            TestContext.Current.CancellationToken);
+
+        await Should.ThrowAsync<InvalidOperationException>(connecting);
+    }
+
+    [Fact]
+    public async Task BeforeEnrolling_Is403WithAnActionableCode()
+    {
+        // The same bar as sync: a caller with no identity key holds no vault key either, so every
+        // notice this socket could carry is about ciphertext they cannot read.
+        var client = fixture.CreateClientFor(NewSubject());
+
+        var response = await client.GetAsync(
+            new Uri(VaultEvents.Path, UriKind.Relative),
+            TestContext.Current.CancellationToken);
+
+        response.StatusCode.ShouldBe(HttpStatusCode.Forbidden);
+
+        var problem = await response.Content.ReadProblemAsync();
+        problem.ShouldNotBeNull();
+        problem.Code.ShouldBe(ProblemCodes.EnrollmentRequired);
+    }
+
+    [Fact]
+    public async Task APlainGet_SaysItIsAWebSocket()
+    {
+        // A person, or a client with the wrong URL. Answering with a problem document rather than a
+        // socket that closes is the difference between a diagnosable mistake and a mysterious one.
+        var client = fixture.CreateClientFor(await SeedEnrolledUserAsync());
+
+        var response = await client.GetAsync(
+            new Uri(VaultEvents.Path, UriKind.Relative),
+            TestContext.Current.CancellationToken);
+
+        response.StatusCode.ShouldBe(HttpStatusCode.BadRequest);
+
+        var problem = await response.Content.ReadProblemAsync();
+        problem.ShouldNotBeNull();
+        problem.Code.ShouldBe(ProblemCodes.MalformedRequest);
+        problem.Detail.ShouldNotBeNull().ShouldContain(VaultEvents.SubProtocol);
+    }
+
+    [Fact]
+    public async Task AnUpgradeWithoutTheSubprotocol_IsRefused()
+    {
+        // The subprotocol is this socket's version negotiation, so accepting an upgrade that did not
+        // offer it would mean answering a client in a dialect it never agreed to read.
+        var (subject, _) = await SeedUserWithVaultAsync();
+
+        var client = fixture.Server.CreateWebSocketClient();
+        client.ConfigureRequest = request => request.Headers.Authorization =
+            $"Bearer {fixture.IdentityProvider.MintToken(subject)}";
+
+        var connecting = client.ConnectAsync(
+            new Uri(fixture.Server.BaseAddress, VaultEvents.Path.TrimStart('/')),
+            TestContext.Current.CancellationToken);
+
+        await Should.ThrowAsync<InvalidOperationException>(connecting);
+    }
+
+    [Fact]
+    public async Task TheFirstFrameIsHello()
+    {
+        var (subject, _) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        var hello = await ReadAsync(socket, timeout.Token);
+
+        hello.Kind.ShouldBe(VaultEventKinds.Hello);
+
+        // Sent so a client knows when silence means the socket is dead rather than quiet, and so a
+        // socket following nothing — a real state, for an account with no vaults — is distinguishable
+        // from one that is broken.
+        hello.HeartbeatSeconds.ShouldNotBeNull().ShouldBeGreaterThan(0);
+        hello.VaultCount.ShouldBe(1);
+    }
+
+    [Fact]
+    public async Task MetaAdvertisesTheFeature()
+    {
+        // How a client decides whether to hold a socket open at all. Absence is not an error — it
+        // means synchronise on the timer, which is what every client did before this existed.
+        var client = fixture.CreateClient();
+
+        var meta = await (await client.GetAsync(
+                new Uri("/api/v1/meta", UriKind.Relative),
+                TestContext.Current.CancellationToken))
+            .Content.ReadContractAsync<MetaResponse>();
+
+        meta.ShouldNotBeNull();
+        meta.Features.ShouldContain(
+            feature => string.Equals(feature, VaultEvents.Feature, StringComparison.Ordinal));
+    }
+
+    // ---- Notices ----
+
+    [Fact]
+    public async Task APush_AnnouncesTheVaultToAFollowingSocket()
+    {
+        var (subject, vaultId) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        var client = fixture.CreateClientFor(subject);
+        var push = await client.PostContractAsync(PushUrl(vaultId), NewCreateBatch());
+        push.EnsureSuccessStatusCode();
+
+        var notice = await ReadUntilAsync(socket, VaultEventKinds.VaultChanged, timeout.Token);
+
+        notice.VaultId.ShouldBe(vaultId);
+
+        // A hint for logging and coalescing, never a cursor: cursors are opaque and integrity-tagged,
+        // and a client that tried to resume from this would be resuming from a number it invented.
+        notice.Sequence.ShouldNotBeNull().ShouldBeGreaterThan(0);
+    }
+
+    [Fact]
+    public async Task ANoticeCarriesNoCiphertext()
+    {
+        // The load-bearing property of the whole design. A notice says only that a vault moved; the
+        // client's answer is the delta pull it would have run on its timer anyway, which keeps exactly
+        // one code path applying changes. See ADR 0012.
+        var (subject, vaultId) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        var client = fixture.CreateClientFor(subject);
+        var batch = NewCreateBatch();
+        await client.PostContractAsync(PushUrl(vaultId), batch);
+
+        var raw = await ReadRawUntilAsync(socket, VaultEventKinds.VaultChanged, timeout.Token);
+
+        // The envelope this test pushed, as it would appear if a payload had been forwarded.
+        raw.ShouldNotContain(Convert.ToBase64String(batch.Operations[0].Payload!.Envelope));
+        raw.ShouldNotContain("payload", Case.Insensitive);
+        raw.ShouldNotContain(batch.Operations[0].EntityId.ToString());
+    }
+
+    [Fact]
+    public async Task APushToAnotherAccountsVault_IsNotAnnouncedHere()
+    {
+        // The disclosure that would matter: a socket learning that vault ids it cannot read exist,
+        // and when somebody works on them.
+        var (subject, vaultId) = await SeedUserWithVaultAsync();
+        var (stranger, strangersVaultId) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        // The stranger's write goes first, so a socket that leaked would have announced it before the
+        // one this test then waits for. Ordering, not a timeout: "nothing arrived in two seconds"
+        // passes on a slow machine for the wrong reason.
+        var strangersClient = fixture.CreateClientFor(stranger);
+        (await strangersClient.PostContractAsync(PushUrl(strangersVaultId), NewCreateBatch()))
+            .EnsureSuccessStatusCode();
+
+        var ownClient = fixture.CreateClientFor(subject);
+        (await ownClient.PostContractAsync(PushUrl(vaultId), NewCreateBatch()))
+            .EnsureSuccessStatusCode();
+
+        var notice = await ReadUntilAsync(socket, VaultEventKinds.VaultChanged, timeout.Token);
+
+        notice.VaultId.ShouldBe(vaultId);
+        notice.VaultId.ShouldNotBe(strangersVaultId);
+    }
+
+    [Fact]
+    public async Task APushThatAppliedNothing_AnnouncesNothing()
+    {
+        // A batch of pure conflicts moved no vault. Announcing one anyway would have every client on
+        // it pull for a change that is not there.
+        var (subject, vaultId) = await SeedUserWithVaultAsync();
+        var client = fixture.CreateClientFor(subject);
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        // An update to an item that does not exist: rejected as a conflict, nothing written.
+        var stale = new SyncPushRequest(
+        [
+            NewOperation(Guid.CreateVersion7(), expectedVersion: 7, envelope: [9, 9]),
+        ]);
+
+        var conflicted = await client.PostContractAsync(PushUrl(vaultId), stale);
+        conflicted.EnsureSuccessStatusCode();
+
+        var results = await conflicted.Content.ReadContractAsync<SyncPushResponse>();
+        results.ShouldNotBeNull();
+        results.Results[0].Status.ShouldBe(SyncOperationStatus.Conflict);
+
+        // Then a write that did land. The first notice must be that one.
+        (await client.PostContractAsync(PushUrl(vaultId), NewCreateBatch())).EnsureSuccessStatusCode();
+
+        var notice = await ReadUntilAsync(socket, VaultEventKinds.VaultChanged, timeout.Token);
+
+        notice.Sequence.ShouldNotBeNull().ShouldBeGreaterThan(0);
+    }
+
+    [Fact]
+    public async Task APing_IsAnswered()
+    {
+        var (subject, _) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        await SendAsync(socket, new VaultEvent(VaultEventKinds.Ping), timeout.Token);
+
+        var pong = await ReadUntilAsync(socket, VaultEventKinds.Pong, timeout.Token);
+
+        pong.Kind.ShouldBe(VaultEventKinds.Pong);
+    }
+
+    [Fact]
+    public async Task AFrameThisServerCannotRead_DoesNotEndTheSocket()
+    {
+        // A control channel whose failure mode is "the client polls instead" should tolerate a frame
+        // from a newer client rather than cost that client its push for the whole session.
+        var (subject, vaultId) = await SeedUserWithVaultAsync();
+
+        using var timeout = Timeout();
+        using var socket = await fixture.ConnectEventsAsync(subject, timeout.Token);
+
+        await ReadAsync(socket, timeout.Token);
+
+        await socket.SendAsync(
+            Encoding.UTF8.GetBytes("{ not json at all"),
+            WebSocketMessageType.Text,
+            endOfMessage: true,
+            timeout.Token);
+
+        var client = fixture.CreateClientFor(subject);
+        (await client.PostContractAsync(PushUrl(vaultId), NewCreateBatch())).EnsureSuccessStatusCode();
+
+        var notice = await ReadUntilAsync(socket, VaultEventKinds.VaultChanged, timeout.Token);
+
+        notice.VaultId.ShouldBe(vaultId);
+        socket.State.ShouldBe(WebSocketState.Open);
+    }
+
+    // ---- Helpers ----
+
+    /// <summary>A token that gives up rather than letting a broken build hang the suite.</summary>
+    private static CancellationTokenSource Timeout()
+    {
+        var source = CancellationTokenSource.CreateLinkedTokenSource(
+            TestContext.Current.CancellationToken);
+
+        source.CancelAfter(FrameTimeout);
+
+        return source;
+    }
+
+    private static async Task<VaultEvent> ReadAsync(WebSocket socket, CancellationToken cancellationToken)
+    {
+        var json = await ReadRawAsync(socket, cancellationToken);
+
+        return JsonSerializer.Deserialize(json, DodoSshJsonContext.Default.VaultEvent)
+            ?? throw new InvalidOperationException($"The server sent a null frame: {json}");
+    }
+
+    private static async Task<string> ReadRawAsync(WebSocket socket, CancellationToken cancellationToken)
+    {
+        var buffer = new byte[8 * 1024];
+
+        var received = await socket.ReceiveAsync(buffer, cancellationToken);
+
+        if (received.MessageType == WebSocketMessageType.Close)
+        {
+            throw new InvalidOperationException(
+                $"The server closed the socket: {received.CloseStatus} {received.CloseStatusDescription}");
+        }
+
+        return Encoding.UTF8.GetString(buffer, 0, received.Count);
+    }
+
+    /// <summary>Reads past the frames a test does not care about — hello, and heartbeats.</summary>
+    private static async Task<VaultEvent> ReadUntilAsync(
+        WebSocket socket,
+        string kind,
+        CancellationToken cancellationToken)
+    {
+        while (true)
+        {
+            var frame = await ReadAsync(socket, cancellationToken);
+
+            if (string.Equals(frame.Kind, kind, StringComparison.Ordinal))
+            {
+                return frame;
+            }
+        }
+    }
+
+    /// <summary>
+    /// The same, but keeping the bytes.
+    /// </summary>
+    /// <remarks>
+    /// Deserialising and asserting on the fields would prove only that this <em>record</em> has no
+    /// payload member, which is a tautology. Asserting on what actually crossed the socket is what
+    /// would catch a field added to the frame later without anybody thinking about disclosure.
+    /// </remarks>
+    private static async Task<string> ReadRawUntilAsync(
+        WebSocket socket,
+        string kind,
+        CancellationToken cancellationToken)
+    {
+        while (true)
+        {
+            var json = await ReadRawAsync(socket, cancellationToken);
+            var frame = JsonSerializer.Deserialize(json, DodoSshJsonContext.Default.VaultEvent);
+
+            if (string.Equals(frame?.Kind, kind, StringComparison.Ordinal))
+            {
+                return json;
+            }
+        }
+    }
+
+    private static Task SendAsync(WebSocket socket, VaultEvent frame, CancellationToken cancellationToken) =>
+        socket.SendAsync(
+            JsonSerializer.SerializeToUtf8Bytes(frame, DodoSshJsonContext.Default.VaultEvent),
+            WebSocketMessageType.Text,
+            endOfMessage: true,
+            cancellationToken);
+
+    private static string PushUrl(Guid vaultId) => $"/api/v1/vaults/{vaultId}/sync/push";
+
+    private static string NewSubject() => $"events-{Guid.CreateVersion7():N}";
+
+    private static SyncPushOperation NewOperation(Guid entityId, int? expectedVersion, byte[] envelope) =>
+        new(
+            Guid.CreateVersion7(),
+            SyncEntityType.Host,
+            entityId,
+            SyncOperation.Upsert,
+            expectedVersion,
+            new EncryptedPayload(envelope, [0xD, 0xE], Guid.CreateVersion7(), 1, 1),
+            new SyncPlaintextFields());
+
+    private static SyncPushRequest NewCreateBatch() =>
+        new([NewOperation(Guid.CreateVersion7(), expectedVersion: null, envelope: [1, 2, 3, 4])]);
+
+    private async Task<string> SeedEnrolledUserAsync()
+    {
+        var subject = NewSubject();
+
+        await using var scope = fixture.CreateScope();
+        var database = scope.ServiceProvider.GetRequiredService<DodoDbContext>();
+
+        var user = NewUser(subject);
+        database.Users.Add(user);
+        database.UserKeys.Add(Seed.CurrentKey(user.Id, Now));
+        await database.SaveChangesAsync();
+
+        return subject;
+    }
+
+    private async Task<(string Subject, Guid VaultId)> SeedUserWithVaultAsync()
+    {
+        var subject = NewSubject();
+
+        await using var scope = fixture.CreateScope();
+        var database = scope.ServiceProvider.GetRequiredService<DodoDbContext>();
+
+        var user = NewUser(subject);
+
+        var vault = new Vault
+        {
+            Id = Guid.CreateVersion7(),
+            Name = "Personal",
+            OwnerKind = VaultOwnerKind.Personal,
+            OwnerUserId = user.Id,
+            KeyGeneration = 1,
+            CreatedAtUtc = Now,
+            UpdatedAtUtc = Now,
+        };
+
+        database.Users.Add(user);
+        database.UserKeys.Add(Seed.CurrentKey(user.Id, Now));
+        database.Vaults.Add(vault);
+        await database.SaveChangesAsync();
+
+        return (subject, vault.Id);
+    }
+
+    private UserAccount NewUser(string subject) => new()
+    {
+        Id = Guid.CreateVersion7(),
+        Issuer = fixture.IdentityProvider.Authority,
+        Subject = subject,
+        Status = UserStatus.Active,
+        CreatedAtUtc = Now,
+        UpdatedAtUtc = Now,
+    };
+}

tests/DodoSSH.Client.Api.Tests/VaultEventStreamTests.cs

@@ -0,0 +1,388 @@
+using System.Net.WebSockets;
+using System.Text;
+using System.Text.Json;
+using System.Threading.Channels;
+using DodoSSH.Client.Api;
+using DodoSSH.Contracts;
+
+namespace DodoSSH.Client.Api.Tests;
+
+/// <summary>
+/// The reconnection policy, which is what this class actually is.
+/// </summary>
+/// <remarks>
+/// <para>
+/// A dropped socket is the ordinary case here rather than the exception — laptops sleep, proxies time
+/// out, tokens expire, servers are redeployed — so the behaviour worth covering is what happens
+/// <em>after</em> a failure, not the happy path. Driven through the injected connector, because the one
+/// thing a test cannot do to a real network is make it fail on cue.
+/// </para>
+/// <para>
+/// The backoff is configured down to milliseconds throughout. What is under test is the shape of the
+/// policy — does it try again, does it wait, does it stop waiting when told the token was the problem —
+/// and none of that depends on the intervals a shipped client uses.
+/// </para>
+/// </remarks>
+public sealed class VaultEventStreamTests
+{
+    private static readonly Uri Server = new("https://dodossh.example");
+
+    private static readonly VaultEventStreamOptions Impatient = new()
+    {
+        InitialBackoff = TimeSpan.FromMilliseconds(1),
+        MaxBackoff = TimeSpan.FromMilliseconds(5),
+        InitialSilenceTimeout = TimeSpan.FromSeconds(30),
+    };
+
+    [Fact]
+    public async Task ItDialsTheWebSocketFormOfTheServersUrl()
+    {
+        // https becomes wss, and the path is the one in the contract. Getting either wrong is a client
+        // that reconnects against a 404 for the whole session, which from outside is indistinguishable
+        // from a network that eats WebSockets.
+        var dialled = new List<Uri>();
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream(
+            (url, _, _) =>
+            {
+                dialled.Add(url);
+                return Task.FromResult<WebSocket>(socket);
+            });
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+
+        await stream.ReadAsync(Token);
+
+        dialled[0].ShouldBe(new Uri("wss://dodossh.example/api/v1/events"));
+    }
+
+    [Fact]
+    public async Task ItSendsTheBearerTokenOnTheUpgrade()
+    {
+        // The whole of this socket's authorization, unlike the relay's ticket. See ADR 0012.
+        var presented = new List<string>();
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream(
+            (_, token, _) =>
+            {
+                presented.Add(token);
+                return Task.FromResult<WebSocket>(socket);
+            });
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+
+        await stream.ReadAsync(Token);
+
+        presented[0].ShouldBe(StubTokens.Token);
+    }
+
+    [Fact]
+    public async Task ANoticeReachesTheReader()
+    {
+        var vaultId = Guid.CreateVersion7();
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream((_, _, _) => Task.FromResult<WebSocket>(socket));
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, vaultId, 42));
+
+        var notice = await stream.ReadAsync(Token);
+
+        notice.Kind.ShouldBe(VaultEventKinds.VaultChanged);
+        notice.VaultId.ShouldBe(vaultId);
+        notice.Sequence.ShouldBe(42);
+    }
+
+    [Fact]
+    public async Task AHeartbeatIsAnsweredAndNotHandedToTheReader()
+    {
+        // A ping is housekeeping between the two ends. Passing it up would wake a synchronisation loop
+        // every thirty seconds for a frame that says nothing happened.
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream((_, _, _) => Task.FromResult<WebSocket>(socket));
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.Ping, HeartbeatSeconds: 30));
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+
+        var first = await stream.ReadAsync(Token);
+
+        first.Kind.ShouldBe(VaultEventKinds.VaultChanged, "the ping should not have been forwarded");
+
+        var answered = await socket.SentAsync(Token);
+        answered.Kind.ShouldBe(VaultEventKinds.Pong);
+    }
+
+    [Fact]
+    public async Task AKindThisBuildDoesNotKnow_IsStillHandedOver()
+    {
+        // What makes the frame table extensible: this class must not decide what a newer server may
+        // say. Deciding to ignore it is the caller's, and costs that caller one redundant pass.
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream((_, _, _) => Task.FromResult<WebSocket>(socket));
+
+        socket.Deliver(new VaultEvent("session.offered"));
+
+        var notice = await stream.ReadAsync(Token);
+
+        notice.Kind.ShouldBe("session.offered");
+    }
+
+    [Fact]
+    public async Task AFailedDial_IsRetried()
+    {
+        // No server yet, or no network. Neither is an error to report: the caller's synchronisation
+        // timer is running regardless, which is what lets this stay silent and keep trying.
+        var attempts = 0;
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream((_, _, _) =>
+        {
+            if (++attempts < 3)
+            {
+                throw new WebSocketException("no route to host");
+            }
+
+            return Task.FromResult<WebSocket>(socket);
+        });
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+
+        var notice = await stream.ReadAsync(Token);
+
+        notice.Kind.ShouldBe(VaultEventKinds.VaultChanged);
+        attempts.ShouldBe(3);
+    }
+
+    [Fact]
+    public async Task ADroppedSocket_IsReplaced()
+    {
+        // The case that decides whether this feature survives a laptop lid. A stream that gave up on
+        // the first close would work all morning and be silently dead after lunch.
+        var sockets = new List<FakeWebSocket>();
+
+        await using var stream = Stream((_, _, _) =>
+        {
+            var socket = new FakeWebSocket();
+            sockets.Add(socket);
+
+            if (sockets.Count == 1)
+            {
+                socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+                socket.Close(WebSocketCloseStatus.EndpointUnavailable);
+            }
+            else
+            {
+                socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 2));
+            }
+
+            return Task.FromResult<WebSocket>(socket);
+        });
+
+        (await stream.ReadAsync(Token)).Sequence.ShouldBe(1);
+        (await stream.ReadAsync(Token)).Sequence.ShouldBe(2);
+
+        sockets.Count.ShouldBeGreaterThanOrEqualTo(2);
+    }
+
+    [Fact]
+    public async Task AnExpiredTokenClose_ReconnectsAndAsksForAFreshToken()
+    {
+        // The bound that lets a long-lived socket be authorised by a short-lived credential: the server
+        // closes at the token's expiry and the client comes straight back with a new one. The token
+        // provider being asked again is the half that matters — reconnecting with the spent token would
+        // be an unbroken loop of closes.
+        var tokens = new StubTokens();
+        var sockets = 0;
+
+        await using var stream = new VaultEventStream(
+            Server,
+            tokens,
+            TimeProvider.System,
+            (_, _, _) =>
+            {
+                var socket = new FakeWebSocket();
+
+                if (++sockets == 1)
+                {
+                    socket.Close((WebSocketCloseStatus)VaultEvents.TokenExpiredCloseCode);
+                }
+                else
+                {
+                    socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 7));
+                }
+
+                return Task.FromResult<WebSocket>(socket);
+            },
+            Impatient);
+
+        (await stream.ReadAsync(Token)).Sequence.ShouldBe(7);
+
+        tokens.Requests.ShouldBeGreaterThanOrEqualTo(2);
+    }
+
+    [Fact]
+    public async Task TryRead_TakesWhatIsWaitingAndSaysWhenNothingIs()
+    {
+        // How a caller coalesces a burst: read one, wait a moment, swallow the rest. Without this a
+        // colleague tidying a folder would produce a synchronisation pass per item.
+        var socket = new FakeWebSocket();
+
+        await using var stream = Stream((_, _, _) => Task.FromResult<WebSocket>(socket));
+
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 1));
+        socket.Deliver(new VaultEvent(VaultEventKinds.VaultChanged, Guid.CreateVersion7(), 2));
+
+        (await stream.ReadAsync(Token)).Sequence.ShouldBe(1);
+
+        // Delivery is asynchronous, so the second may not have landed yet; this is the same
+        // wait-then-drain the caller performs.
+        await Task.Delay(TimeSpan.FromMilliseconds(200), Token);
+
+        stream.TryRead(out var queued).ShouldBeTrue();
+        queued.Sequence.ShouldBe(2);
+
+        stream.TryRead(out _).ShouldBeFalse();
+    }
+
+    [Fact]
+    public async Task AnIdleStream_NeverDelivers()
+    {
+        // What a server without the feature supplies. Waiting for ever rather than completing is the
+        // point: a caller selecting between this and a timer has to fall through to the timer, and a
+        // read that returned at once would spin that loop as fast as the machine allows.
+        using var stream = IdleVaultEventStream.Instance;
+        using var giveUp = CancellationTokenSource.CreateLinkedTokenSource(Token);
+
+        giveUp.CancelAfter(TimeSpan.FromMilliseconds(100));
+
+        await Should.ThrowAsync<OperationCanceledException>(
+            async () => await stream.ReadAsync(giveUp.Token));
+
+        stream.TryRead(out _).ShouldBeFalse();
+        stream.IsConnected.ShouldBeFalse();
+    }
+
+    private static CancellationToken Token => TestContext.Current.CancellationToken;
+
+    private static VaultEventStream Stream(
+        Func<Uri, string, CancellationToken, Task<WebSocket>> connect) =>
+        new(Server, new StubTokens(), TimeProvider.System, connect, Impatient);
+
+    /// <summary>A token provider that hands out one value and counts who asked.</summary>
+    private sealed class StubTokens : IAccessTokenProvider
+    {
+        internal const string Token = "access-token";
+
+        internal int Requests { get; private set; }
+
+        public ValueTask<string> GetAccessTokenAsync(CancellationToken cancellationToken)
+        {
+            Requests++;
+
+            return ValueTask.FromResult(Token);
+        }
+    }
+
+    /// <summary>
+    /// A socket a test writes the server's half of.
+    /// </summary>
+    /// <remarks>
+    /// Frames queued with <see cref="Deliver"/> are handed out by <see cref="ReceiveAsync"/> in order;
+    /// once the queue is empty the receive waits, which is what an idle connection does. <see
+    /// cref="Close"/> queues the close instead, so a test can script "two notices and then the server
+    /// went away" as a value rather than as a race.
+    /// </remarks>
+    private sealed class FakeWebSocket : WebSocket
+    {
+        private readonly Channel<byte[]> inbound = Channel.CreateUnbounded<byte[]>();
+        private readonly Channel<VaultEvent> outbound = Channel.CreateUnbounded<VaultEvent>();
+
+        private WebSocketCloseStatus? closing;
+        private WebSocketState state = WebSocketState.Open;
+
+        public override WebSocketCloseStatus? CloseStatus => closing;
+
+        public override string? CloseStatusDescription => null;
+
+        public override WebSocketState State => state;
+
+        public override string? SubProtocol => VaultEvents.SubProtocol;
+
+        /// <summary>Queues a frame for the client to read.</summary>
+        internal void Deliver(VaultEvent frame) =>
+            inbound.Writer.TryWrite(
+                JsonSerializer.SerializeToUtf8Bytes(frame, DodoSshJsonContext.Default.VaultEvent));
+
+        /// <summary>Ends the socket, after everything already queued has been read.</summary>
+        internal void Close(WebSocketCloseStatus status)
+        {
+            closing = status;
+            inbound.Writer.TryWrite([]);
+        }
+
+        /// <summary>The next frame the client sent.</summary>
+        internal ValueTask<VaultEvent> SentAsync(CancellationToken cancellationToken) =>
+            outbound.Reader.ReadAsync(cancellationToken);
+
+        public override async Task<WebSocketReceiveResult> ReceiveAsync(
+            ArraySegment<byte> buffer,
+            CancellationToken cancellationToken)
+        {
+            var frame = await inbound.Reader.ReadAsync(cancellationToken);
+
+            // The empty frame Close queues. Reported as a close, exactly as a real socket does once the
+            // peer's close frame arrives.
+            if (frame.Length == 0)
+            {
+                state = WebSocketState.Closed;
+
+                return new WebSocketReceiveResult(
+                    0, WebSocketMessageType.Close, endOfMessage: true, closing, null);
+            }
+
+            frame.CopyTo(buffer.Array!, buffer.Offset);
+
+            return new WebSocketReceiveResult(frame.Length, WebSocketMessageType.Text, endOfMessage: true);
+        }
+
+        public override Task SendAsync(
+            ArraySegment<byte> buffer,
+            WebSocketMessageType messageType,
+            bool endOfMessage,
+            CancellationToken cancellationToken)
+        {
+            var json = Encoding.UTF8.GetString(buffer.Array!, buffer.Offset, buffer.Count);
+
+            if (JsonSerializer.Deserialize(json, DodoSshJsonContext.Default.VaultEvent) is { } frame)
+            {
+                outbound.Writer.TryWrite(frame);
+            }
+
+            return Task.CompletedTask;
+        }
+
+        public override void Abort() => state = WebSocketState.Aborted;
+
+        public override Task CloseAsync(
+            WebSocketCloseStatus closeStatus,
+            string? statusDescription,
+            CancellationToken cancellationToken) => CloseOutputAsync(
+            closeStatus, statusDescription, cancellationToken);
+
+        public override Task CloseOutputAsync(
+            WebSocketCloseStatus closeStatus,
+            string? statusDescription,
+            CancellationToken cancellationToken)
+        {
+            state = WebSocketState.Closed;
+
+            return Task.CompletedTask;
+        }
+
+        public override void Dispose() => state = WebSocketState.Closed;
+    }
+}

tests/DodoSSH.Client.App.Layout.Tests/StubTeamServer.cs

@@ -57,6 +57,17 @@ internal sealed class StubTeamServer : IVaultServer, ITeamApi, IVaultGrantApi
     /// <inheritdoc />
     public IKeyBindingAuthorizer KeyBinding => throw new NotSupportedException();
 
+    /// <summary>
+    /// A push channel that never pushes.
+    /// </summary>
+    /// <remarks>
+    /// Not <c>NotSupportedException</c> like its neighbours: the background synchronisation loop reads
+    /// this on every wait, so a layout test that opened a screen would throw from a timer thread rather
+    /// than draw anything. Waiting for ever is the honest stand-in — an offline layout test has no
+    /// server to be pushed from.
+    /// </remarks>
+    public IVaultEventStream Events => IdleVaultEventStream.Instance;
+
     /// <inheritdoc />
     public SyncOptions SyncOptions => new();
 

tests/DodoSSH.Client.App.Tests/FakeVaultEventStream.cs

@@ -0,0 +1,49 @@
+using System.Threading.Channels;
+using DodoSSH.Client.Api;
+using DodoSSH.Contracts;
+
+namespace DodoSSH.Client.App.Tests;
+
+/// <summary>
+/// A server's push channel, driven by a test rather than by a socket.
+/// </summary>
+/// <remarks>
+/// The real <c>VaultEventStream</c> is a reconnection policy wrapped round a WebSocket, and none of
+/// that is what the shell's behaviour depends on: what the shell does with a notice is the same
+/// whether it arrived over a healthy socket, after four reconnections, or from this. Driving it by
+/// hand is what makes "the loop synchronised because it was told to, not because a minute passed" a
+/// test that finishes in milliseconds and cannot flake.
+/// </remarks>
+internal sealed class FakeVaultEventStream : IVaultEventStream
+{
+    private readonly Channel<VaultEvent> notices = Channel.CreateUnbounded<VaultEvent>();
+
+    /// <inheritdoc />
+    public bool IsConnected => true;
+
+    /// <summary>How many times the shell has waited on this. Proves the loop is watching at all.</summary>
+    internal int Reads { get; private set; }
+
+    /// <summary>Delivers a notice, as a server would.</summary>
+    internal void Push(Guid vaultId, long sequence = 1) =>
+        notices.Writer.TryWrite(
+            new VaultEvent(VaultEventKinds.VaultChanged, vaultId, sequence));
+
+    /// <summary>Delivers the notice that says the caller's vault list has changed.</summary>
+    internal void PushAccessChanged() =>
+        notices.Writer.TryWrite(new VaultEvent(VaultEventKinds.VaultsChanged));
+
+    /// <inheritdoc />
+    public ValueTask<VaultEvent> ReadAsync(CancellationToken cancellationToken)
+    {
+        Reads++;
+
+        return notices.Reader.ReadAsync(cancellationToken);
+    }
+
+    /// <inheritdoc />
+    public bool TryRead(out VaultEvent notice) => notices.Reader.TryRead(out notice!);
+
+    /// <inheritdoc />
+    public void Dispose() => notices.Writer.TryComplete();
+}

tests/DodoSSH.Client.App.Tests/FakeVaultServer.cs

@@ -39,6 +39,16 @@ internal sealed partial class FakeVaultServer : IVaultServer, IAccountApi, ISync
 
     internal int PushCount { get; private set; }
 
+    /// <summary>
+    /// How many delta reads this server has served.
+    /// </summary>
+    /// <remarks>
+    /// The one observable a synchronisation pass always produces. <see cref="PushCount"/> only moves when
+    /// there is something queued, so a test asking "did a pass run" — which is what the push channel's
+    /// whole purpose comes down to — has to count pulls.
+    /// </remarks>
+    internal int PullCount { get; private set; }
+
     internal bool IsEnrolled => statement is not null;
 
     /// <summary>
@@ -99,6 +109,19 @@ internal sealed partial class FakeVaultServer : IVaultServer, IAccountApi, ISync
     /// <inheritdoc />
     public IKeyBindingAuthorizer KeyBinding => this;
 
+    /// <summary>
+    /// The push channel, which a test drives by hand.
+    /// </summary>
+    /// <remarks>
+    /// A real queue rather than an idle stand-in, because the behaviour worth covering here is the one
+    /// the socket exists for: a notice arriving makes the background loop synchronise without waiting
+    /// out its minute. See <see cref="FakeVaultEventStream.Push"/>.
+    /// </remarks>
+    internal FakeVaultEventStream Notices { get; } = new();
+
+    /// <inheritdoc />
+    public IVaultEventStream Events => Notices;
+
     /// <inheritdoc />
     public SyncOptions SyncOptions => SyncOptions.Default;
 
@@ -238,6 +261,8 @@ internal sealed partial class FakeVaultServer : IVaultServer, IAccountApi, ISync
         SyncPullRequest request,
         CancellationToken cancellationToken)
     {
+        PullCount++;
+
         if (SyncFailure is { } failure)
         {
             return Task.FromException<SyncPullResponse>(failure);

tests/DodoSSH.Client.App.Tests/ShellFlowTests.cs

@@ -523,6 +523,90 @@ public sealed class ShellFlowTests : IAsyncLifetime
         vault.Status.ShouldContain("bad day");
     }
 
+    /// <summary>
+    /// The whole point of the push channel: a pass that did not wait for the minute.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// The timing is what makes this an assertion rather than a hope. The background timer is a full
+    /// minute and the wait below gives up in ten seconds, so a pull that arrives can only have been
+    /// caused by the notice — there is no interval at which the timer could have produced it.
+    /// </para>
+    /// <para>
+    /// The vault id in the notice is arbitrary, and deliberately so: a pass synchronises every vault
+    /// this session can reach, so the loop reads the notice as "there is something to fetch" and never
+    /// as "fetch this one". A test that seeded a real id would imply a targeting this does not do.
+    /// </para>
+    /// </remarks>
+    [Fact]
+    public async Task APushedNotice_SynchronisesWithoutWaitingForTheTimer()
+    {
+        await UnlockedAsync();
+
+        // The unlock starts the loop, whose first act is a pass; waited out so the count below is a
+        // baseline rather than a race with it.
+        await EventuallyAsync(
+            () => server.PullCount > 0,
+            "the pass on open should have run");
+
+        var before = server.PullCount;
+
+        server.Notices.Push(Guid.CreateVersion7());
+
+        await EventuallyAsync(
+            () => server.PullCount > before,
+            "a notice should have woken the loop long before the one-minute timer");
+    }
+
+    /// <remarks>
+    /// The half that is easy to get wrong. The loop selects between two waits, and both have to survive
+    /// losing: <c>PeriodicTimer</c> throws if a second wait is started while one is outstanding, and an
+    /// abandoned channel read stays registered and swallows the next notice written. Either defect
+    /// leaves the first notice working and every one after it silently lost, which is why one notice is
+    /// not enough to prove this.
+    /// </remarks>
+    [Fact]
+    public async Task NoticesKeepWakingTheLoop_NotJustTheFirst()
+    {
+        await UnlockedAsync();
+
+        await EventuallyAsync(() => server.PullCount > 0, "the pass on open should have run");
+
+        for (var round = 1; round <= 3; round++)
+        {
+            var before = server.PullCount;
+
+            server.Notices.Push(Guid.CreateVersion7());
+
+            await EventuallyAsync(
+                () => server.PullCount > before,
+                $"notice {round} should have woken the loop as the first one did");
+        }
+    }
+
+    /// <summary>Waits for something a background loop is expected to do, or fails saying what.</summary>
+    /// <remarks>
+    /// Polled rather than signalled because the thing under test is a loop nobody hands a completion
+    /// source to. The bound is generous — this is not measuring latency, only proving that the timer
+    /// cannot be what caused the result.
+    /// </remarks>
+    private static async Task EventuallyAsync(Func<bool> condition, string because)
+    {
+        var deadline = TimeProvider.System.GetUtcNow().AddSeconds(10);
+
+        while (TimeProvider.System.GetUtcNow() < deadline)
+        {
+            if (condition())
+            {
+                return;
+            }
+
+            await Task.Delay(TimeSpan.FromMilliseconds(20), Token);
+        }
+
+        throw new ShouldAssertException(because);
+    }
+
     /// <remarks>
     /// <para>
     /// The page's own <c>term.focus()</c> focuses the textarea inside the document, which does nothing