diff --git a/tests/OpenCode.Sdk.Tests/Sessions/SessionLogLiveTests.cs b/tests/OpenCode.Sdk.Tests/Sessions/SessionLogLiveTests.cs index de3e3515..a54f7416 100644 --- a/tests/OpenCode.Sdk.Tests/Sessions/SessionLogLiveTests.cs +++ b/tests/OpenCode.Sdk.Tests/Sessions/SessionLogLiveTests.cs @@ -15,6 +15,26 @@ public sealed class SessionLogLiveTests(SimulatedDriveServerFixture server) private const string FollowFirstReply = "Task three follow first reply."; private const string FollowSecondPrompt = "task-three follow second prompt"; private const string FollowSecondReply = "Task three follow second reply."; + private const string PastTailPrompt = "task-three past-tail prompt"; + private const string PastTailReply = "Task three past-tail reply."; + private const string SkippedPrompt = "task-three skipped opening prompt"; + private const string SkippedReply = "Task three skipped opening reply."; + private const string SkippedFollowPrompt = "task-three skipped follow prompt"; + private const string SkippedFollowReply = "Task three skipped follow reply."; + + /// + /// How far past the tail the unreachable-cursor proof reaches. A scripted turn commits single + /// digits of events, so nothing this session does can bring the aggregate up to that cursor. + /// + private const long PastTailOffset = 1000; + + /// + /// How far past the tail the suppression proof reaches. Small enough that the following turn + /// crosses it, so the same attachment shows both halves: the events at or below the cursor are + /// skipped, and delivery resumes once the aggregate overtakes it. + /// + private const long SkippedOffset = 3; + private static readonly TimeSpan CleanupTimeout = TimeSpan.FromSeconds(15); [Test] @@ -167,6 +187,153 @@ await Assert.That(transcript.ReplayItems.OfType() } } + /// + /// A cursor the aggregate has never reached is accepted rather than refused, and the reply is + /// the marker alone carrying the real watermark, a value below the cursor that was sent. The + /// same session's settled read is the control: it returns the turn, so the empty answer is the + /// cursor being out of range rather than a server that retained nothing. + /// + [Test] + [Timeout(180_000)] + public async Task GetLogAsync_Should_Answer_A_Cursor_Past_The_Tail_With_The_Marker_Alone( + CancellationToken cancellationToken) + { + using var workspace = server.CreateWorkspace(); + using var client = server.CreateClient(new LocationSelector { Directory = workspace.Path }); + var sessionId = await CreateOwnedSessionAsync(client, workspace.Path, "session-log-past-tail", cancellationToken); + var session = client.Sessions.GetSessionClient(sessionId); + var cleanup = new OwnedSessionCleanup(session, CleanupTimeout); + Exception? primaryFailure = null; + + try + { + var transcript = new SessionLogTranscript(session); + var turn = new SimulatedSessionTurn(server, client, session, sessionId); + var terminal = await turn.CompleteAsync(PastTailPrompt, PastTailReply, cancellationToken); + cleanup.MarkTurnCompleted(); + + var settled = await transcript.ReadToEndAsync( + new SessionLogRequest { Follow = QueryBoolean.False }, cancellationToken); + await Assert.That(settled.OfType() + .Any(succeeded => succeeded.Durable.Seq == terminal.Durable.Seq)).IsTrue(); + var watermark = settled.OfType().Single().Seq; + await Assert.That(watermark).IsNotNull(); + var beyond = watermark!.Value + PastTailOffset; + + var answered = await transcript.ReadToEndAsync( + new SessionLogRequest + { + After = beyond.ToString(CultureInfo.InvariantCulture), + Follow = QueryBoolean.False, + }, + cancellationToken); + + await Assert.That(answered).Count().IsEqualTo(1); + await Assert.That(answered[0]).IsTypeOf(); + var marker = (EventLogSynced)answered[0]; + await Assert.That(marker.AggregateId).IsEqualTo(sessionId); + await Assert.That(marker.Seq).IsNotNull(); + await Assert.That(marker.Seq!.Value).IsGreaterThanOrEqualTo(watermark.Value); + await Assert.That(marker.Seq.Value).IsLessThan(beyond); + + Console.WriteLine( + "session-log-past-tail-live: requested=" + Number(beyond) + + " watermark=" + Number(marker.Seq.Value) + + " replayed-before=" + Number(settled.Count) + " answered=" + Number(answered.Count)); + } + catch (Exception exception) + { + primaryFailure = exception; + } + finally + { + await cleanup.CompleteAsync(primaryFailure); + } + } + + /// + /// A cursor a few sequences past the tail suppresses exactly what it covers and no more. The + /// following turn crosses it, so one attachment shows both halves: the execution start the + /// cursor covers never arrives, while the terminal event beyond it does. A second reader on the + /// correct cursor returns the skipped start, so the absence is a skip rather than a gap. + /// + [Test] + [Timeout(240_000)] + public async Task GetLogAsync_Should_Skip_What_A_Cursor_Past_The_Tail_Covers_And_Resume_Beyond_It( + CancellationToken cancellationToken) + { + using var workspace = server.CreateWorkspace(); + using var client = server.CreateClient(new LocationSelector { Directory = workspace.Path }); + var sessionId = await CreateOwnedSessionAsync(client, workspace.Path, "session-log-skipped", cancellationToken); + var session = client.Sessions.GetSessionClient(sessionId); + var cleanup = new OwnedSessionCleanup(session, CleanupTimeout); + var transcript = new SessionLogTranscript(session); + cleanup.Own("session log enumerator", transcript.DisposeAsync); + Exception? primaryFailure = null; + + try + { + var opening = new SimulatedSessionTurn(server, client, session, sessionId); + var openingTerminal = await opening.CompleteAsync(SkippedPrompt, SkippedReply, cancellationToken); + cleanup.MarkTurnCompleted(); + + var settled = await transcript.ReadToEndAsync( + new SessionLogRequest { Follow = QueryBoolean.False }, cancellationToken); + await Assert.That(settled.OfType() + .Any(succeeded => succeeded.Durable.Seq == openingTerminal.Durable.Seq)).IsTrue(); + var watermark = settled.OfType().Single().Seq; + await Assert.That(watermark).IsNotNull(); + var covered = watermark!.Value + SkippedOffset; + + await transcript.AttachAsync(covered.ToString(CultureInfo.InvariantCulture), cancellationToken); + await Assert.That(transcript.Marker).IsNotNull(); + await Assert.That(transcript.Marker!.Seq).IsNotNull(); + await Assert.That(transcript.Marker.Seq!.Value).IsGreaterThanOrEqualTo(watermark.Value); + await Assert.That(transcript.Marker.Seq.Value).IsLessThan(covered); + await Assert.That(transcript.ReplayItems).IsEmpty(); + + cleanup.MarkTurnStarted(); + var followed = new SimulatedSessionTurn(server, client, session, sessionId); + var terminal = await followed.CompleteAsync(SkippedFollowPrompt, SkippedFollowReply, cancellationToken); + cleanup.MarkTurnCompleted(); + + // Delivery resumed past the cursor: the attachment carried the turn to its terminal event. + await transcript.ReadTurnAsync(sessionId, SkippedFollowReply); + await Assert.That(transcript.Succeeded).IsNotNull(); + await Assert.That(transcript.Succeeded!.Durable.Seq).IsEqualTo(terminal.Durable.Seq); + + // The correct cursor returns an execution start inside the covered range on this same + // session, so what the attachment never delivered was written and is readable. + var control = await new SessionLogTranscript(session).ReadToEndAsync( + new SessionLogRequest + { + After = watermark.Value.ToString(CultureInfo.InvariantCulture), + Follow = QueryBoolean.False, + }, + cancellationToken); + var skipped = control.OfType() + .Where(started => started.Data.SessionId == sessionId && started.Durable.Seq <= covered) + .ToList(); + await Assert.That(skipped).IsNotEmpty(); + await Assert.That(transcript.LiveItems.OfType() + .Any(started => started.Durable.Seq == skipped[0].Durable.Seq)).IsFalse(); + + Console.WriteLine( + "session-log-skipped-live: cursor=" + Number(covered) + + " watermark=" + Number(watermark.Value) + + " skipped-start-seq=" + Number(skipped[0].Durable.Seq) + + " resumed-terminal-seq=" + Number(transcript.Succeeded.Durable.Seq)); + } + catch (Exception exception) + { + primaryFailure = exception; + } + finally + { + await cleanup.CompleteAsync(primaryFailure); + } + } + private static async Task CreateOwnedSessionAsync( OpenCodeClient client, string directory,