paliad/internal/services/paliadin_streaming.go

package services

// Streaming support for the Paliadin chat surface (t-paliad-235).
//
// The legacy LocalPaliadinService.RunTurn returns the full response in
// one shot — the chat UI gets one `content` blob and the typewriter
// simulates streaming. That falls apart on long turns: the HTTP client
// hits its 130 s ceiling, paliad's SSE stream closes, the bubble shows
// "Verbindung verloren" and the response is lost.
//
// The aichat backend exposes a real streaming variant at
// /chat/turn/stream that emits incremental chunks + named heartbeat
// events while claude is thinking. AichatPaliadinService implements
// the StreamingPaliadin interface defined here; the handler probes
// for it via a type assertion and falls back to the one-shot RunTurn
// when the backend doesn't support streaming (legacy path).
//
// Recovery (a separate axis): when the transport drops mid-turn,
// the AichatRecoverer interface lets the handler ask the backend to
// look up the late response via aichat's conversation API rather than
// rely on the legacy filesystem janitor — which only knows about
// LocalPaliadinService's per-turn response files.

import (
	"context"

	"github.com/google/uuid"
)

// StreamEvent is one increment of a streaming turn. The handler
// receives these via the channel passed to RunTurnStream and forwards
// them as SSE frames to the browser.
//
// Exactly one of Kind's payloads is meaningful per event:
//
//   StreamChunk     → Content holds the next slice of assistant text
//   StreamHeartbeat → ElapsedSeconds holds upstream "still thinking" tick
//   StreamMeta      → UsedTools / RowsSeen / ClassifierTag populated
//   StreamError     → Code / Message / Retryable populated
//
// StreamDone is implicit: when the channel closes without an error
// event, the turn completed. The accompanying *TurnResult returned by
// RunTurnStream carries the final accumulated body + meta + conversation
// id for persistence and recovery.
type StreamEvent struct {
	Kind StreamEventKind

	// StreamChunk
	Content string

	// StreamHeartbeat
	ElapsedSeconds int

	// StreamMeta (terminal-side; may also be merged into final TurnResult)
	UsedTools     []string
	RowsSeen      []int
	ClassifierTag string

	// StreamError
	Code      string
	Message   string
	Retryable bool

	// StreamConversation — aichat sometimes resolves the conversation id
	// before the first chunk arrives. We surface it as soon as we have
	// it so the handler can persist it for recovery, even if the stream
	// is later interrupted.
	ConversationID string
}

// StreamEventKind enumerates the meaningful flavours.
type StreamEventKind string

const (
	StreamChunk        StreamEventKind = "chunk"
	StreamHeartbeat    StreamEventKind = "heartbeat"
	StreamMeta         StreamEventKind = "meta"
	StreamError        StreamEventKind = "error"
	StreamConversation StreamEventKind = "conversation"
)

// StreamingPaliadin is the optional extension the AichatPaliadinService
// implements. Handlers detect it via type assertion; backends that don't
// implement it (the legacy local + remote paths) fall back to the
// one-shot Paliadin.RunTurn.
//
// Contract:
//   - RunTurnStream MUST close `events` before returning, so the handler
//     loop terminates cleanly.
//   - Returning a non-nil error implies the audit row was already
//     stamped with an error_code; the handler does not double-stamp.
//   - The *TurnResult is populated even on partial failure when the
//     upstream produced any meaningful body — handlers may render it as
//     a salvaged best-effort result instead of an error.
type StreamingPaliadin interface {
	Paliadin

	// RunTurnStream drives one turn against the streaming upstream and
	// pushes StreamEvents onto `events` as they arrive. Blocks until the
	// upstream finishes or the context cancels. `events` is closed by
	// the implementation before this method returns.
	RunTurnStream(ctx context.Context, req TurnRequest, events chan<- StreamEvent) (*TurnResult, error)
}

// AichatRecoverer is the optional extension that knows how to ask the
// aichat backend "did this turn actually complete?" when paliad's local
// audit row never got a response (because the transport dropped mid
// turn). Implementations look up the persisted aichat_conversation_id,
// query aichat's GET /chat/conversations/{id}/turns, find the matching
// assistant turn, and write the response back to paliad's row.
//
// Returns (nil, nil) when aichat doesn't have the response either —
// i.e. the turn is truly lost and the UI must degrade to "verloren"
// copy rather than "wird nachgereicht".
type AichatRecoverer interface {
	RecoverTurn(ctx context.Context, callerID, turnID uuid.UUID) (*PaliadinTurn, error)
}

// safeSendStream pushes an event onto the channel, dropping on context
// cancel. Mirrors the handler-side `send` helper but works against a
// generic chan StreamEvent.
func safeSendStream(ctx context.Context, ch chan<- StreamEvent, ev StreamEvent) {
	select {
	case ch <- ev:
	case <-ctx.Done():
	}
}