Fix queued channel messages losing their input (MUL-4840)

[chengweiv5]

Summary

• seal each debounced Lark/Slack message batch onto its task at enqueue time
• keep rolling upgrades compatible by using the legacy assistant cursor for the first owned batch
• add a regression test for a predecessor replying after the next message arrives but before its task is claimed

Root cause

Channel tasks still selected input at claim time by scanning user messages after the latest assistant row. If a second message arrived while the first task was running, then the first task replied before the queued successor was claimed, that reply moved the cursor past the second message and the successor ran with an empty prompt.

Verification

go test ./internal/integrations/channel/engine ./internal/service ./internal/handler

[vercel]

Someone is attempting to deploy a commit to the IndexLabs Team on Vercel.

A member of the Team first needs to authorize it.

[Bohan-J]

Thanks for the follow-up commit. The empty-reply decoupling via IsChannelChatSession is the right call — it correctly separates "channel session, stay silent on empty output" from the direct-chat no_response contract instead of overloading chat_input_task_id — and the added TestCompleteTask_ChannelEmptyOutputWritesNoRow case for an owned channel batch covers it well.

One blocking issue remains before this can merge.

Blocking — a failed flush now permanently orphans the queued batch

EnqueueChannelChatTask seals the batch by stamping chat_input_task_id (SetChatTaskInputOwnerSelf, task.go:1499) and claiming the messages in one transaction, aborting the whole tx when len(claimed) != len(messageIDs) (task.go:1512). The gap is in the caller + recovery path:

1. scheduleRun / the flush closure calls takePendingInput(key) — which removes the batch's message IDs from pendingInput — before calling EnqueueChannelChatTask (router.go:328, router.go:335).
2. flushChatRun's error branch (router.go:358-373) never puts those IDs back: the default case only logs and clears typing; ErrChatTaskAgentNoRuntime / ErrChatTaskAgentArchived just send a notice. None re-queue.
3. The transaction is all-or-nothing, so on failure those chat_message rows keep task_id = NULL. But now every channel task carries chat_input_task_id, so the daemon always reads input via ListChatInputMessages(chat_input_task_id) and never falls back to the trailingUserMessages scan (daemon.go:2077-2080). Those task_id = NULL rows are therefore unreachable by any future task — they aren't in a later batch (already taken) and the trailing scan no longer runs for channel tasks.

Net effect: any enqueue failure permanently drops the user's queued messages. The worst case is the default branch (a transient DB/tx error, or a count mismatch): no user-visible notice, typing is cleared, and the messages silently vanish — which is exactly the class of silent message loss this PR sets out to fix. Before this change EnqueueChatTask did no claim, so a failed flush left the rows task_id = NULL and the next successful flush's trailing scan re-collected them (self-healing). This PR removes the trailing scan for channel tasks without giving the channel path an equivalent recovery.

Suggested fix: on enqueue failure in flushChatRun, put messageIDs back into pendingInput (re-append) so the next flush re-attempts the claim. Since the tx fully rolls back, those rows are still task_id = NULL, so re-queuing is safe and does not reintroduce the trailing-cursor race (ownership stays explicit). At minimum this should cover the default (transient) and count-mismatch cases; whether offline/archived should also re-queue is a product call.

Minor (non-blocking)

ClaimChannelChatInputMessages uses id = ANY(...) and returns one row per match, so a duplicate ID in messageIDs would make len(claimed) < len(messageIDs) and trip the abort (compounding the drop above). Duplicates shouldn't occur with the current two-phase dedup, but a defensive de-dup of messageIDs before the claim would harden it.

Otherwise the direction — explicit input ownership instead of a trailing-cursor scan — is a solid, more reliable model. Happy to re-review once the flush-failure recovery is in.

[chengweiv5]

Addressed the failed-flush input loss reported in review.

What changed:

• Retryable channel flush failures now restore the complete pending batch instead of orphaning its messages.
• Message order and the batch's forceFresh intent are restored atomically, including when newer messages arrive before the retry.
• Offline and archived agents remain terminal cases: users receive the existing notice and those batches are not requeued.
• Channel input message IDs are defensively deduplicated before the transactional claim, preserving first-seen order and preventing false claim-count mismatches.

Coverage added:

• A failed ForceFresh=true batch is retried with both the original and newly arrived messages while retaining fresh-session semantics.
• Duplicate message IDs are claimed exactly once.

Validation completed successfully:

• Relevant channel engine, service, and handler tests
• Channel engine race tests
• Go vet for the affected packages
• GitHub CI (all checks passed)

[Bohan-J]

Thanks for the quick turnaround — the previous blocking issue is resolved. restorePendingBatch correctly re-queues the batch on a session-reload failure and on the default / count-mismatch enqueue error, preserving arrival order and forceFresh, and folding the fresh/input state under one mutex is a nice consistency win. uniqueUUIDs also addresses the duplicate-ID concern. Nicely done.

One remaining blocking issue in the same area.

Blocking — recovery is in-memory only; a hard crash/restart still permanently orphans queued messages

The batch recovery lives entirely in the Router's in-memory pendingInput map — nothing about the pending batch is persisted. That breaks the durability contract the debouncer documents in batcher.go:

A hard crash inside the window drops the pending trigger (messages are durable; they just do not fire a run until the next message).

That guarantee held before this PR because the claim was a trailing-message scan: after a crash, the next inbound message enqueued a task whose claim-time scan re-collected every trailing task_id = NULL user row — including ones persisted before the crash. This PR removes that scan for channel tasks (they now always carry chat_input_task_id, so the daemon reads input only via ListChatInputMessages and never falls back to trailingUserMessages — daemon.go:2077-2080). So now:

1. U1 is persisted by AppendUserMessage, but the 3s window hasn't flushed; the process hard-crashes and pendingInput is lost.
2. After restart U2 arrives; the new Router only puts U2 into pendingInput — it never rediscovers U1.
3. EnqueueChannelChatTask claims only the passed IDs, so U2's task owns only U2; U1 stays task_id = NULL and is read by no future task.

The same gap applies to the new re-queue path: a transient failure restores the batch to memory, but if the process crashes before the next message flushes it, it's gone. (Secondary, related: restorePendingBatch puts the IDs back but doesn't re-arm a flush timer, so even without a crash a restored batch waits for the next inbound message before it retries.)

Net: the class of silent message loss this PR targets is still reachable across a crash/restart — which the debouncer's own design explicitly treats as a handled case.

Suggested direction (either works):

• Persist the pending-batch ownership at append time and recover it on startup; or
• Let the channel enqueue also (safely) claim the session's older un-owned user rows, not just the passed IDs — restoring the self-healing scan. That then needs per-session serialization of concurrent flushes and an explicit decision on whether offline/archived-dropped messages should be re-collected later.

Please also add regression coverage:

• Persist U1, rebuild the Router before the flush, then deliver U2 → the next task should receive U1 + U2 in order.
• After a transient enqueue failure re-queues the batch, rebuild the Router before the next message → a later successful flush should still recover the failed batch.

The direction — explicit input ownership instead of a trailing-cursor scan — remains the right model; it just needs a durable recovery path to fully replace what the scan used to provide. Happy to re-review once that's in.

[chengweiv5]

Local verification has passed.

I tested the original Feishu scenario by sending two consecutive messages to the same agent while the first task was still running. Both messages were preserved and processed in order; the second message was no longer dropped or dispatched with an empty input.

[chengweiv5]

Addressed the restart-durability review findings in commit 9fdf6b453.

What changed:

• Channel enqueue now serializes flushes per chat session and claims every older unowned user row up to the newest explicit debounce message. This recovers durable input left behind by a Router restart while keeping newer concurrent messages outside the batch.
• Explicit message IDs are still required to be part of the claim, so an unavailable ID rolls the transaction back instead of silently creating a partial task.
• Retryable reload/enqueue failures now re-arm the debounce timer. A retry does not replace a newer inbound message's already-armed flush closure.
• Offline and archived outcomes keep their existing terminal notice behavior; their durable unowned rows can be recovered by a later successful channel flush.

Regression coverage now includes:

• U1 persisted before a Router restart, then U2 arriving afterward: the next task owns U1 + U2 in order.
• A rolled-back failed batch followed by a Router restart and later input: the next successful task recovers the failed batch.
• Automatic timer retry without waiting for another inbound message.
• Retry scheduling does not overwrite a newer inbound flush.

Local validation passed for the channel engine/service unit tests, race tests, package vet, generated DB package, and handler compilation. Database-backed handler tests were compiled but skipped locally because PostgreSQL was not running; CI is now executing them.

[Bohan-J]

Thanks for the continued iteration. The per-session lockFlush, the durable-recovery claim (rediscovering older unowned rows up to the batch boundary), and re-arming the retry timer are all good moves. Reviewing 0356c79d, three blocking issues remain — the first is a regression this commit introduces.

Blocking 1 (regression in this commit) — the last_assistant lower bound strands the exact double-text message this PR fixes

ClaimChannelChatInputMessages now floors the claimed rows with (m.created_at, m.id) > last_assistant. That assumes every unclaimed user row is newer than the last assistant reply, but the PR's own motivating scenario breaks it:

1. U1 arrives → task1 claims U1 and runs.
2. U2 arrives while task1 is still generating (so U2.created_at precedes the eventual reply) and schedules its own flush.
3. task1 finishes and writes assistant reply A1 with A1.created_at > U2.created_at.
4. U2's flush runs ClaimChannelChatInputMessages: boundary = U2, last_assistant = A1. Since U2 > A1 is false, U2 — the explicit batch member — is excluded and claimed comes back empty.
5. containsAllUUIDs(claimed, messageIDs) fails (task.go:1515), the tx rolls back, and flushChatRun's default branch calls restoreAndScheduleRun, restoring U2 into pendingInput and retrying. A1 stays newer than U2, so every retry fails identically.

Net: U2 is never claimed, and because it's restored into pendingInput on each failure it also poisons every later flush for the session (each new batch now contains U2 and fails containsAllUUIDs) — the session's channel input wedges permanently. It's timing-dependent (needs A1 to commit before U2's flush claim, i.e. task1 answers within ~the debounce window), but that's exactly the "reply arrives before the queued message is processed" race this PR targets. TestChannelChat_RecoverySkipsAnsweredLegacyHistory doesn't catch it because there both pending rows are created after the assistant reply.

The lower bound's goal — not replaying already-answered legacy history on a rolling deploy — is legitimate, but it can't be applied to the explicit batch members, since a genuinely-queued follow-up can be older than a previous turn's reply. Suggest applying the > last_assistant floor only to the implicitly-recovered older rows, while always including id = ANY(message_ids) unconditionally.

Blocking 2 — graceful shutdown self-deadlocks on the first flush error

flushChatRun acquires the per-session lockFlush mutex up front (router.go:360, non-reentrant, released via defer). On a session-reload or enqueue error it calls restoreAndScheduleRun → ScheduleIfAbsent. During FlushAll (graceful shutdown) the batcher is stopped, so ScheduleIfAbsent runs the retry closure inline on the same goroutine (batcher.go:110-113); that re-enters flushChatRun and blocks on lock.mu.Lock() (router.go:403) for a mutex the outer frame still holds → self-deadlock. FlushAll → Drain() never returns and shutdown hangs until the process is killed. It fires on the first reload/enqueue failure during shutdown — precisely the DB-unavailable case. (In normal operation ScheduleIfAbsent arms a timer instead, so this is shutdown-specific.) Suggest: when the batcher is stopped, don't inline-retry — leave the messages durable (task_id NULL) for the next replica's recovery — and/or guard restoreAndScheduleRun from re-entering while the flush lock is held.

Blocking 3 — correctness depends on the FK key-share lock, which the repo forbids relying on

LockChatSessionForChannelInput's comment (chat.sql:136-138) states the "cannot absorb a newer inbound message" guarantee rests on concurrent chat_message inserts taking the parent row's FK key-share lock. The project rule is that relationships and their locking must be enforced in the application layer rather than via FK/cascade behavior. AppendUserMessage doesn't take this session lock, and lockFlush only serializes flush-vs-flush in-process — so append-vs-claim ordering is currently held together solely by the FK's implicit lock. If that FK is later dropped per the rule, batch isolation breaks silently with no compile or test signal. Relatedly, since the boundary orders by (created_at, id) with a random UUID tiebreak, a row committed in the same microsecond as the boundary with a smaller UUID sorts <= boundary and can be absorbed — today only the row lock prevents that. Suggest having both AppendUserMessage and the enqueue take the same explicit per-session lock (a Postgres advisory lock, or the existing FOR UPDATE on chat_session extended to the append path) so ordering never depends on the FK.

Tests to add

• Double-text: U1 running, U2 arrives, task1's reply commits, then U2's flush → U2 must still be claimed and run (Blocking 1).
• FlushAll with a persistently failing enqueue/reload → Drain() returns within a bound (Blocking 2).
• Concurrent append during a claim → the appended row is left for the next batch, not absorbed, asserting the explicit app lock rather than relying on the FK (Blocking 3).

Direction is right and close — these are the remaining correctness gaps. Happy to re-review once they're addressed.