fix(core): preserve queued messages after recall

[AaronZ345]

Summary

• preserve messages queued after an active message is recalled
• resume the oldest surviving message in a replacement agent session while retaining FIFO order
• avoid treating an intentionally closed agent event channel as a crash that drops the queue

Root cause

The silent recall path cleared the entire pendingMessages queue when stopping the recalled turn. Messages sent after the recalled message were therefore accepted and acknowledged, then discarded without a response.

Tests

• go test -race ./core -run 'TestCUJ_A8_RecallPreservesLaterQueuedMessage|TestHandleMessageRecallPreservesLaterQueueWhenAgentChannelCloses' -count=1
• go test ./core -count=1
• go test -tags no_web ./... -count=1

[chenhg5]

Conclusion: Approve

Overall assessment:

• This is a focused, well-scoped fix for the CUJ-A8 "messages queued after a recall are silently dropped" bug. The author splits the recovery into two complementary pieces — resumePendingAfterStoppedTurn (new) recovers queued successors when the turn ends in a Stopped state, and the isStopped() short-circuit in processInteractiveEvents prevents the engine from mis-treating that same intentional close as a crash and draining the queue a second time. Lock ownership is explicitly transferred to the nested processor via unlocked = true, which mirrors the existing pattern in the file. Tests are strong: a CUJ-A8 scenario with three messages (active, recalled, survivor) plus a targeted engine test using a controllableAgent whose session can be swapped between oldSession and replacementSession so we observe the replacement session receiving only the survivor.

Review scope:

• Reviewed core/engine.go (the new resumePendingAfterStoppedTurn, the channelClosed short-circuit, and the two removed pendingMessages = nil blocks in stopInteractiveSessionWithOptions) plus the new tests in core/cuj_test.go and core/engine_test.go.
• Focused on race safety, lock ownership handoff, FIFO ordering under the placeholder pattern, and backward compatibility of the silent-recall path.

✅ What looks good:

• FIFO preservation via the placeholder pattern. resumePendingAfterStoppedTurn takes [P0..Pn] from the stopped state, moves Pn into a placeholder entry, and starts the nested processor on P0. The freshly-created state in getOrCreateInteractiveStateWith then re-uses the existing adoptPendingFromPlaceholder to inherit the remainder. New messages that arrive between the unlock of the stopped state and the re-creation of the placeholder are appended after Pn, preserving order.
• Lock handoff is explicit. unlocked = true is set immediately before the recursive processInteractiveMessageWith call; the deferred fallback if !unlocked { session.Unlock() } then skips the unlock on the outer call, and the inner call's defer releases the lock exactly once. This matches the established handoff pattern in drainOrphanedQueue / drainPendingMessages.
• Deduplication of "channel close" handling. Removing the state.pendingMessages = nil block from the non-notifyQueued branches of stopInteractiveSessionWithOptions means the silent-recall path now relies on resumePendingAfterStoppedTurn to recover the queue — and the isStopped() short-circuit at channelClosed: correctly returns early before notifyDroppedQueuedMessages / cleanupInteractiveState would otherwise drain the queue. The two changes are independent but they compose: removing the wipe without the short-circuit would leak a queue; adding the short-circuit without removing the wipe would lose the queue. Together they work.
• Tests cover both unit and CUJ angles. TestCUJ_A8_RecallPreservesLaterQueuedMessage exercises the full Engine+CUJ path, and TestHandleMessageRecallPreservesLaterQueueWhenAgentChannelCloses exercises the fast path (channel close mid-recall). Together they lock in the FIFO invariant and the no-double-drain invariant.
• slog.Info("resuming queued messages after stopped turn", …) gives operators a breadcrumb to grep when triaging — the line we wanted in production is now there.

🚨/🔴 Must fix:

• None.

🟠 Should improve:

• Minor: log a parallel breadcrumb on the channelClosed: short-circuit. Right now the only signal that "the channel close was a recall, not a crash" is the absence of the previous "agent process exited" warn line. Adding slog.Debug("agent channel closed after intentional stop; deferring queue recovery", "session_key", sessionKey) (or Info, depending on volume) would make the connection between the two pieces of this PR obvious in logs. Pure observability nit.
• Minor: consider using state.markStopped() consistently. The new resumePendingAfterStoppedTurn only reads state.stopped; it does not call markStopped() itself. That's correct because the caller (stopInteractiveSessionWithOptions) already marks the state stopped and removes it from the map. But it's worth a one-line comment at the top of the new function pointing at the only call site, so future readers understand the invariant ("caller must have already called markStopped() and delete(e.interactiveStates, …)").
• Minor: When the placeholder is freshly allocated (placeholder == nil || placeholder == state), it is eventsNeedResync: true. After processInteractiveMessageWith replaces it with a real state via adoptPendingFromPlaceholder, the new state's eventsNeedResync is also true (from getOrCreateInteractiveStateWith's constructor) — so the flag round-trip is correct. Worth a one-line comment so a future reader doesn't "optimize" one of them away.

🔵 Optional:

• A test that places a NEW message between e.stopInteractiveSessionWithOptions returning and resumePendingAfterStoppedTurn acquiring interactiveMu would exercise the "messages appended to placeholder" branch explicitly. The current CUJ test only checks the case where the placeholder is freshly created with no pre-existing pending; the merge-with-prior-placeholder path is exercised by the implementation but not asserted on. Not blocking.
• A test for the eventsNeedResync: true invariant on the replacement session would close the small "did we forget to drain stale events from the old turn?" gap. Again, not blocking.

❓ Questions:

• Did the silent-recall path used to notify the user of dropped queued messages? Looking at the diff, the old code in the else branch (when notifyQueued == false) dropped without notification, and the new code preserves those messages instead. So this is a net-positive behavior change — but it would be good to confirm with PM that no caller was relying on the old "silently drop queue on recall" semantics.

Testing / Risk:

• Verified: go test -race ./core/... -count=1 -timeout 180s (48.4s, all green); new tests pass; the full CUJ test suite (TestCUJ_A6_A7…, TestCUJ_A8…) is green.
• Blast radius is core/engine.go only. No agent-runtime, no platform, no CLI touched.
• Backward compatible for the recall target (the recalled message is still silently stopped); the only observable behavior change is "messages queued after the recall now get a response instead of being silently dropped." This is exactly the contract issue #1499 / #1498 callers have been asking for.

Next step:

• Merge. The two 🟠 minor comments are nice-to-have for the next maintainer; they are not required for correctness.

[AaronZ345]

Superseded by #1706.