Merge pull request Stack-Cairn#228 from Stack-Cairn/fix/webui-run-lost-truncation

fix: WebUI stream-loss truncation and touch/no-hover UI visibility

Author: su-fen

crates/agent-gateway/internal/session/conversation_ingress.go

@@ -44,6 +44,23 @@ func (m *Manager) ingestChatEvent(requestID string, event *gatewayv1.ChatEvent)
 		eventType = chatwire.EventTypeName(event.GetType())
 	}
 
+	if event.GetType() == gatewayv1.ChatEvent_USER_MESSAGE {
+		if record := s.runs[runID]; record != nil && record.userMessageSeeded {
+			// The gateway already appended this run's user_message at accept
+			// time; swallow the agent echo so the message appears once.
+			return
+		}
+	}
+
+	if stream.runFinishedRecently(runID) {
+		// A live event for a run whose terminal was merely inferred proves the
+		// inference wrong — reopen the run instead of dropping its stream.
+		if !s.resurrectRunLocked(stream, runID) {
+			// Late straggler after a genuine or duplicate terminal; drop it.
+			return
+		}
+	}
+
 	switch event.GetType() {
 	case gatewayv1.ChatEvent_DONE:
 		delete(payload, "type")
@@ -57,17 +74,6 @@ func (m *Manager) ingestChatEvent(requestID string, event *gatewayv1.ChatEvent)
 		delete(payload, "message")
 		s.runFinishedLocked(stream, runID, "failed", "", strings.TrimSpace(message), payload, now)
 		return
-	case gatewayv1.ChatEvent_USER_MESSAGE:
-		if record := s.runs[runID]; record != nil && record.userMessageSeeded {
-			// The gateway already appended this run's user_message at accept
-			// time; swallow the agent echo so the message appears once.
-			return
-		}
-	}
-
-	if stream.runFinishedRecently(runID) {
-		// Late straggler after a forced or duplicate terminal; drop it.
-		return
 	}
 
 	if event.GetType() == gatewayv1.ChatEvent_USER_MESSAGE {
@@ -154,8 +160,33 @@ func (m *Manager) ingestChatControl(requestID string, control *gatewayv1.ChatCon
 
 	switch controlType {
 	case "started":
+		// A reconnect republish may re-anchor a run this store wrongly gave up
+		// on (inferred loss); resurrect before the started no-ops against the
+		// finished set.
+		if stream.runFinishedRecently(runID) && !s.resurrectRunLocked(stream, runID) {
+			return
+		}
 		s.runStartedLocked(stream, runID, "", now)
 	case "completed", "failed", "cancelled":
+		inferredLoss := controlType == "failed" && isInferredRunLossCode(errorCode)
+		if stream.runFinishedRecently(runID) {
+			if inferredLoss || !s.resurrectRunLocked(stream, runID) {
+				return
+			}
+		}
+		// The desktop ledger flushes inferred losses (desktop_run_lost & co)
+		// through this same channel. For the conversation's active run, ignore
+		// such a verdict while the run's own events are fresh or it was
+		// already falsified once — genuine terminals always pass.
+		if inferredLoss &&
+			stream.activity != nil && stream.activity.RunID == runID {
+			record := s.runs[runID]
+			eventsFresh := !stream.lastEventAt.IsZero() &&
+				now.Sub(stream.lastEventAt) < s.runReportLostTimeout
+			if eventsFresh || (record != nil && record.revived) {
+				return
+			}
+		}
 		s.runFinishedLocked(stream, runID, controlType, errorCode, message, nil, now)
 	case "queued_in_gui":
 		s.markRunQueuedInGUILocked(stream, runID, now)
@@ -228,16 +259,20 @@ func (m *Manager) ingestRuntimeSnapshot(snapshot *gatewayv1.ChatRuntimeSnapshot)
 	streamWasUnknown := existingStream == nil || (existingStream.lastSeq == 0 && existingStream.activity == nil)
 	stream := s.streamLocked(conversationID, now)
 	s.noteAgentEpochLocked(stream, epoch)
+	if stream.runFinishedRecently(runID) {
+		// Both running and terminal snapshots are authoritative runtime
+		// evidence. A terminal snapshot must be able to correct an earlier
+		// inferred loss even when no token arrived between the two verdicts.
+		if !s.resurrectRunLocked(stream, runID) {
+			return
+		}
+	}
 
 	switch state {
 	case "completed", "failed", "cancelled":
 		s.runFinishedLocked(stream, runID, state, "", "", nil, now)
 		return
 	}
-	if stream.runFinishedRecently(runID) {
-		return
-	}
-
 	next := &RunSnapshot{
 		RunID:                  runID,
 		Revision:               snapshot.GetRevision(),

crates/agent-gateway/internal/session/conversation_stream.go

@@ -180,6 +180,27 @@ type chatRunRecord struct {
 	// the agent's ref-bearing user_message, so a reconnect replay of the same
 	// event cannot seed a second truncation.
 	rebaseSeeded bool
+	// lostInferred marks a run whose terminal was inferred from missing
+	// liveness reports (desktop_run_lost and friends) rather than delivered by
+	// the run itself. Such a terminal is falsifiable: fresh events for the run
+	// prove it wrong and resurrect the run instead of being dropped as
+	// stragglers.
+	lostInferred bool
+	// revived marks a run resurrected after a wrong inferred terminal; further
+	// inferred-loss signals for it are ignored (the desktop-side ledger may
+	// keep repeating the stale verdict) until a genuine terminal arrives.
+	revived bool
+}
+
+// isInferredRunLossCode reports whether an error code represents a liveness
+// inference (nobody vouched for the run) instead of an outcome the run itself
+// produced. Inferred terminals must stay reversible: the run may well be alive.
+func isInferredRunLossCode(errorCode string) bool {
+	switch errorCode {
+	case "desktop_run_lost", "stale_run", "agent_offline", "desktop_runtime_lease_expired":
+		return true
+	}
+	return false
 }
 
 // chatCommandDedupeRecord is the process-local idempotency key for WebUI chat
@@ -645,9 +666,17 @@ func (s *conversationStreamStore) runFinishedLocked(
 			payload[key] = value
 		}
 	}
-	if record := s.runs[runID]; record != nil && record.clientRequestID != "" {
+	record := s.runRecordLocked(runID, stream.conversationID)
+	if record.clientRequestID != "" {
 		payload["client_request_id"] = record.clientRequestID
 	}
+	// Inferred terminals (nobody vouched for the run) stay falsifiable: a
+	// later event for the run resurrects it instead of being dropped. Genuine
+	// terminals settle the run for good.
+	record.lostInferred = status == "failed" && isInferredRunLossCode(errorCode)
+	if !record.lostInferred {
+		record.revived = false
+	}
 	s.appendEventLocked(stream, runID, StreamEventRunFinished, payload, now)
 	stream.finishedRuns = append(stream.finishedRuns, runID)
 	if len(stream.finishedRuns) > conversationFinishedRunMemory {
@@ -666,6 +695,41 @@ func (s *conversationStreamStore) runFinishedLocked(
 	}
 }
 
+// resurrectRunLocked reopens a run that was force-finished by a liveness
+// inference: fresh agent traffic for the run proves the inference wrong. The
+// run leaves the finished set (so runStartedLocked re-registers it), is
+// flagged to ignore repeats of the stale verdict, and the stream is marked
+// snapshot-hungry so subscribers rebuild the tail that was dropped while the
+// run was considered dead. Refuses when another run owns the conversation —
+// then the late events really are stragglers.
+func (s *conversationStreamStore) resurrectRunLocked(
+	stream *conversationStream,
+	runID string,
+) bool {
+	record := s.runs[runID]
+	if record == nil || !record.lostInferred {
+		return false
+	}
+	if stream.activity != nil {
+		return false
+	}
+	kept := stream.finishedRuns[:0]
+	for _, finished := range stream.finishedRuns {
+		if finished != runID {
+			kept = append(kept, finished)
+		}
+	}
+	stream.finishedRuns = kept
+	record.lostInferred = false
+	record.revived = true
+	// The events dropped between the wrong terminal and this resurrection are
+	// unrecoverable from the log; late joiners and current subscribers rebuild
+	// from the next runtime snapshot.
+	stream.runNeedsSnapshot = true
+	stream.snapshotDirty = true
+	return true
+}
+
 // markRunQueuedLocked records that a run's command is pending in the gateway
 // (accepted but not yet started). No log event — activity only.
 func (s *conversationStreamStore) markRunQueuedLocked(
@@ -1123,17 +1187,37 @@ func (s *conversationStreamStore) onRuntimeStatus(event *gatewayv1.RuntimeStatus
 			stream.activity.UpdatedAt = now
 			continue
 		}
+		record := s.runs[runID]
+		revived := record != nil && record.revived
+		eventsFresh := !stream.lastEventAt.IsZero() &&
+			now.Sub(stream.lastEventAt) < s.runReportLostTimeout
 		if report, ok := finished[runID]; ok {
 			state := report.GetState()
 			errorCode := report.GetErrorCode()
+			// Judged on the report's own fields: a ledger-swept loss is an
+			// inference, while an unknown state is still a desktop-asserted
+			// terminal (normalized below) and stays adopted verbatim.
+			inferred := state == "failed" && isInferredRunLossCode(errorCode)
 			switch state {
 			case "completed", "failed", "cancelled":
 			default:
 				state = "failed"
 				errorCode = "desktop_run_lost"
 			}
-			s.runFinishedLocked(stream, runID, state, errorCode, report.GetMessage(),
-				map[string]any{"reason": "desktop_reported"}, now)
+			// A loss the desktop merely inferred (its ledger starved while the
+			// run's events still flow through this relay, or the verdict was
+			// already falsified once) is not adopted — genuine terminals the
+			// run itself produced always are.
+			if !inferred || (!eventsFresh && !revived) {
+				s.runFinishedLocked(stream, runID, state, errorCode, report.GetMessage(),
+					map[string]any{"reason": "desktop_reported"}, now)
+				continue
+			}
+		}
+		if revived {
+			// Resurrected after a wrong loss verdict: liveness inferences no
+			// longer end this run; the reaper's stale-run timeout is the
+			// backstop for a genuinely dead one.
 			continue
 		}
 		// Stream events vouch too: never finalize a run whose events are still