From 272e1ef59bcb2d7f4285ec453db7a1e69013c39e Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 13 Sep 2026 22:08:24 -0400 Subject: [PATCH] test(e2e): prove offline messages redeliver on session start, durably (RIG-3532) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RIG-3473 T5. The offline-redelivery design records had no positive e2e coverage: the start-sweep appeared only as a hazard other legs acked around (legfive_test.go). `TestCommsOfflineRedeliveryOnSessionStart` makes the redelivery the asserted behavior. The frozen record's prescribed mechanism does not match the code, so this test takes a different shape on two axes. Both were verified against the tree; a follow-up amends the record's T5 section. **The durable surface is `UndeliveredMessages`, not `owed_mentions`.** The record names `CountOwedMentions`/`ClearOwedMention`, but `owed_mentions` rows are written only by `RecordOwedMention`, whose callsites require the message to `@`-mention the agent AND the agent to be outside the channel's sweep set. `InSweepSet` is satisfied by `cm.channel_id = aa.home_channel_id`, so a home-channel post never records a row — `CountOwedMentions` would read 0 both before and after the sweep, and the test would be green while asserting nothing. `UndeliveredMessages` is what the sweep itself consumes (`drainStarts` -> `sweepSession`), so it is the surface that actually moves. **"Offline" has to be a despawn, not a never-started agent.** The record's "post to a provisioned-but-not-started agent, then StartSession" is unreachable: the post wakes the agent and fresh-starts it, because `fanOut` finds no live session, calls `wake`, and `freshStart` resolves the placement `Provision` already recorded — and the waker is wired in the real assembly the fixture runs (`sinks.go` `SetAgentWaker`). `StartSession` would then be a second, racing start. Posting after `RemoveWorkspace` deletes the placement makes the wake a benign no-placement no-op, so the message stays owed. The assertion is durable rather than on the live tail, which is the point of the task: the sweep is enqueued inside `Resume` and drained asynchronously, so a tail opened after `Resume` returns can miss the redelivery. That is the RIG-3044 flake shape. Event-gated throughout, no sleeps and no retries. A pre-condition check asserts the message IS owed before the resume, so the "no longer owed" post-condition cannot pass vacuously. Verified: PASS in 60.45s. Mutation-tested against production — disabling `sweepSession` in `drainStarts` reddens it at the redelivery assertion ("delivery cursor did not advance past message ... within 15s"); production restored byte-identical. Co-authored-by: Matt Wilkinson --- go/e2e/legcomms_redeliver_test.go | 148 ++++++++++++++++++++++++++++++ 1 file changed, 148 insertions(+) create mode 100644 go/e2e/legcomms_redeliver_test.go diff --git a/go/e2e/legcomms_redeliver_test.go b/go/e2e/legcomms_redeliver_test.go new file mode 100644 index 00000000..f961a8ab --- /dev/null +++ b/go/e2e/legcomms_redeliver_test.go @@ -0,0 +1,148 @@ +//go:build podman + +package e2e + +import ( + "context" + "testing" + + "github.com/RigelBuild/compass/go/internal/store" +) + +// TestCommsOfflineRedeliveryOnSessionStart proves the session-start sweep +// redelivers a message that accrued while the agent was offline (RIG-3532 T5). +// "Offline" is a DESPAWN, not a never-started agent: a post to a +// provisioned-but-idle agent wakes and fresh-starts it (the placement recorded +// at Provision makes the wake an auto-start), so the message would never be +// owed. Posting AFTER RemoveWorkspace deletes the placement makes the wake a +// benign no-placement no-op, so the message stays owed for the next start. +func TestCommsOfflineRedeliveryOnSessionStart(t *testing.T) { + if !podmanUsable() { + t.Skip("rootless podman cannot run compass-agent:latest here; skipping the real-stack e2e") + } + + ctx := context.Background() // test root, threaded into NewFixture + every primitive + + // Slot 0 settles lifetime 1's warm turn, driven only so the resume has a + // persisted transcript to reconstruct; slot 1 settles the resumed turn the + // sweep's redelivery of the offline message drives. + const warmReply = "canned lifetime-1 warm turn settled OK" + const redeliverReply = "canned redelivered turn settled OK" + f := NewFixture(ctx, t, WithCannedScript( + CannedText(warmReply), + CannedText(redeliverReply), + )) + + accountID, err := f.CreateAgent(ctx, "comms-redeliver", "Comms Offline Redelivery") + if err != nil { + t.Fatalf("CreateAgent: %v", err) + } + + // ── Lifetime 1: provision, drive one turn so the session is resumable ── + + container1, err := f.Provision(ctx, accountID, "comms-redeliver-provision-1") + if err != nil { + t.Fatalf("Provision (container1): %v", err) + } + // Reap net: the reparented rootless conmon outlives stack Down, so an explicit + // RemoveWorkspace is the reliable reap. Registered before StartSession so a + // later failure still tears it down; the mid-test despawn below no-ops this + // one idempotently. Best-effort teardown, so the error is deliberately dropped. + t.Cleanup(func() { + _ = f.RemoveWorkspace(ctx, container1, "comms-redeliver-teardown-1") + }) + + sessionID, err := f.StartSession(ctx, container1) + if err != nil { + t.Fatalf("StartSession (container1): %v", err) + } + + st, err := store.Open(ctx, f.DSN()) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + defer st.Close() + + acc, err := adminAgentByHandle(ctx, st, "comms-redeliver") + if err != nil { + t.Fatalf("AgentByHandle: %v", err) + } + home := acc.Agent.HomeChannelID + + tail1, err := f.OpenSessionTail(ctx, sessionID) + if err != nil { + t.Fatalf("OpenSessionTail (lifetime 1): %v", err) + } + defer tail1.Close() + + warmID, err := f.PostMessage(ctx, string(home), "general", "warm the session so it persists a transcript") + if err != nil { + t.Fatalf("PostMessage (warm): %v", err) + } + if err := f.AwaitTurnSettled(ctx, tail1); err != nil { + t.Fatalf("AwaitTurnSettled (lifetime 1): %v", err) + } + + // Ack the warm message BEFORE teardown so the resume sweep does not redeliver + // it too — only the offline post below must be owed at resume time. + if err := f.waitDeliveryCursorPast(ctx, st, acc.ID, home, store.MessageID(warmID)); err != nil { + t.Fatalf("waitDeliveryCursorPast (warm): %v", err) + } + // The resume reconstructs the session body from the durable transcript, so the + // warm turn must be committed before container1 is torn down. + if _, err := f.awaitTranscriptPersisted(ctx, st, sessionID, warmReply); err != nil { + t.Fatalf("awaitTranscriptPersisted (warm reply): %v", err) + } + + // Despawn boundary: RemoveWorkspace deletes the placement, which is what makes + // the offline post's wake a no-placement no-op instead of an auto-start. + if err := f.RemoveWorkspace(ctx, container1, "comms-redeliver-teardown-1"); err != nil { + t.Fatalf("RemoveWorkspace (despawn boundary): %v", err) + } + + offlineID, err := f.PostMessage(ctx, string(home), "general", "message accrued while the agent is offline") + if err != nil { + t.Fatalf("PostMessage (offline): %v", err) + } + + // Pre-condition: the offline message IS owed. Without it the post-condition + // ("no longer owed") would pass vacuously against a never-owed message. + owed, err := st.UndeliveredMessages(ctx, acc.ID) + if err != nil { + t.Fatalf("UndeliveredMessages (pre-condition): %v", err) + } + preOwed := false + for _, m := range owed[home] { + if m.ID == store.MessageID(offlineID) { + preOwed = true + break + } + } + if !preOwed { + t.Fatalf("offline message %s is not owed before session start; the pre-condition the redelivery proof rests on does not hold", offlineID) + } + + // ── Lifetime 2: provision a fresh container, RESUME the logical session ── + + container2, err := f.Provision(ctx, accountID, "comms-redeliver-provision-2") + if err != nil { + t.Fatalf("Provision (container2): %v", err) + } + // Reap net for container2, same reparented-conmon reason; registered before + // Resume so a Resume failure still tears it down. Best-effort, error dropped. + t.Cleanup(func() { + _ = f.RemoveWorkspace(ctx, container2, "comms-redeliver-teardown-2") + }) + + if _, err := f.Resume(ctx, container2, sessionID); err != nil { + t.Fatalf("Resume (container2): %v", err) + } + + // Post-condition read DURABLY from the store, not a live tail: the start-sweep + // is enqueued INSIDE Resume and drained async, so a tail opened after Resume + // can miss the redelivery (the RIG-3044 flake shape). The cursor advancing past + // the offline message proves the sweep redelivered it and the session acked. + if err := f.waitDeliveryCursorPast(ctx, st, acc.ID, home, store.MessageID(offlineID)); err != nil { + t.Fatalf("waitDeliveryCursorPast (offline redelivery): %v", err) + } +}