paliad/internal/services/aichat_paliadin.go

package services

// AichatPaliadinService — the Phase B path of the Paliadin backend
// (m/paliad#38, t-paliad-194).
//
// Design + Phase A spec: docs/design/aichat-2026-05-13.md in m/mAi
// (issue m/mAi#207). The aichat service runs on mRiver itself, owns
// the long-lived `claude` tmux session per persona (windows per user),
// and exposes a small HTTP surface to client apps:
//
//   POST /chat/turn   — synchronous one-shot turn
//   POST /chat/reset  — kill the user's window
//   GET  /chat/health — service liveness
//
// Where RemotePaliadinService shells out over SSH to a per-app shim,
// AichatPaliadinService is a thin HTTP client of the centralized
// backend. It implements the same Paliadin interface as the local and
// remote backends so the cutover is a `PALIADIN_BACKEND=aichat` env
// flip rather than a handler-layer rewrite.
//
// Wiring is gated on PALIADIN_BACKEND in cmd/server/main.go:
//   PALIADIN_BACKEND=aichat  → AichatPaliadinService
//   anything else (default)  → legacy Local/Remote/Disabled selection
//
// Per-user RLS auth: the planck branch (mai/planck/paliadin-per-user-rls,
// parked t-paliad-156) carried the per-turn HS256 mint that turns
// paliad.* queries into "RLS as the user" instead of service role. The
// mint lives in paliadin_jwt.go; this service reuses it and ships the
// signed token in the `jwt` field of /chat/turn, which aichat writes
// to a per-turn file the claude pane reads to `SET LOCAL
// request.jwt.claims` before each paliad.* query.

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/jmoiron/sqlx"
)

// AichatPaliadinConfig is the bag of knobs cmd/server/main.go passes
// when constructing an AichatPaliadinService.
type AichatPaliadinConfig struct {
	// BaseURL is the aichat service root (e.g. http://100.99.98.203:8765).
	// No trailing slash. Endpoints are derived as BaseURL + "/chat/*".
	BaseURL string

	// BearerToken is the per-app raw token aichat hashes against
	// tokens.yaml. Empty token is rejected by the aichat /chat/turn
	// auth gate as "auth_failed".
	BearerToken string

	// Persona is the aichat persona id — fixed to "paliadin" for this
	// service. Exposed as config only so tests can override.
	Persona string

	// HTTPClient is the underlying transport. cmd/server/main.go wires
	// a single shared client with a 130 s timeout (matching the Phase A
	// shim ceiling: claude cold start + skill discovery + first
	// reasoning, ~120 s, plus a few seconds of HTTP overhead). Tests
	// inject a roundtripper that doesn't hit the network.
	HTTPClient *http.Client

	// JWTSecret is paliad's SUPABASE_JWT_SECRET. When non-empty,
	// RunTurn mints a fresh per-turn HS256 token scoped to the calling
	// user (sub=userID, role=authenticated). Aichat passes the raw
	// token through to the claude pane via /tmp/aichat-jwts/<turn>.jwt
	// (mode 0600, deferred-removed). The skill reads it and `SET LOCAL
	// request.jwt.claims = …` before each paliad.* query — RLS then
	// evaluates as the user. Empty → no |jwt=…| segment; aichat sees
	// jwt:"" and skips the file write, and the skill surfaces the
	// missing-JWT bug rather than silently leaking as service role.
	JWTSecret []byte

	// JWTTTL bounds the per-turn JWT lifetime. Zero → DefaultPaliadinJWTTTL.
	JWTTTL time.Duration
}

// AichatPaliadinService implements Paliadin against the centralized
// aichat HTTP backend.
type AichatPaliadinService struct {
	paliadinDB
	cfg AichatPaliadinConfig

	// Serialise turns across all users. Same rationale as the remote
	// service: aichat runs one claude per persona session, finite
	// concurrency, paliadin turns are short.
	turnMu sync.Mutex

	// Service-wide health-check cache (NOT per-session — aichat's
	// /chat/health is service-wide, unlike the shim's per-user verb).
	// Same 10 s success cache, no failure cache.
	healthMu        sync.Mutex
	healthOK        bool
	healthCheckedAt time.Time

	// Per-user-session "have we primed this pane in this Go-process
	// lifetime?" cache. Aichat is stateless on user content; the client
	// owns the primer. Same shape as RemotePaliadinService.primed.
	primedMu sync.Mutex
	primed   map[string]bool

	// Hook for tests — when non-nil, callHTTP delegates here instead
	// of hitting the wire. Production code never sets this.
	httpHook func(ctx context.Context, method, path string, body any, out any) error
}

// ErrAichatAuthFailed signals the aichat service rejected the bearer
// token. Distinct from ErrMRiverUnreachable so the operator dashboard
// can disambiguate "service is up but our token is wrong" from "service
// is down". Friendly-error mapping in handlers/paliadin.go covers both.
var ErrAichatAuthFailed = errors.New("aichat: auth failed")

// ErrAichatPersonaUnknown signals the aichat service does not know
// this persona (or this app isn't allowed to use it). Surfaces as
// shim_error / mriver_unreachable to the user — neither is recoverable
// without a deploy-side fix.
var ErrAichatPersonaUnknown = errors.New("aichat: persona unknown")

// DefaultAichatPersona is the persona id every Paliad deploy targets.
// Exposed for tests; cmd/server/main.go does not override it.
const DefaultAichatPersona = "paliadin"

// DefaultAichatHTTPTimeout matches RemotePaliadinService.callShim's
// 130 s ceiling: aichat's persona timeout is 120 s (personas.yaml) and
// HTTP overhead adds ≤10 s.
const DefaultAichatHTTPTimeout = 130 * time.Second

// NewAichatPaliadinService wires the aichat HTTP backend.
//
// Call only when PALIADIN_BACKEND=aichat in the environment; the
// constructor does not probe aichat — first probe happens on the first
// RunTurn call via healthGate.
func NewAichatPaliadinService(db *sqlx.DB, users *UserService, cfg AichatPaliadinConfig) *AichatPaliadinService {
	if cfg.Persona == "" {
		cfg.Persona = DefaultAichatPersona
	}
	if cfg.HTTPClient == nil {
		cfg.HTTPClient = &http.Client{Timeout: DefaultAichatHTTPTimeout}
	}
	cfg.BaseURL = strings.TrimRight(cfg.BaseURL, "/")
	return &AichatPaliadinService{
		paliadinDB: paliadinDB{db: db, users: users},
		cfg:        cfg,
		primed:     make(map[string]bool),
	}
}

// RunTurn drives one Q&A round against the centralized aichat backend.
// Same audit-row contract as the local + remote services: write the row
// first, run the turn, complete on success, mark error on failure.
func (s *AichatPaliadinService) RunTurn(ctx context.Context, req TurnRequest) (*TurnResult, error) {
	s.turnMu.Lock()
	defer s.turnMu.Unlock()

	turnID := uuid.New()
	startedAt := time.Now().UTC()

	if err := s.insertTurnRow(ctx, &PaliadinTurn{
		TurnID:      turnID,
		UserID:      req.UserID,
		SessionID:   req.SessionID,
		StartedAt:   startedAt,
		UserMessage: req.UserMessage,
		PageOrigin:  optionalString(req.PageOrigin),
	}, req.Context); err != nil {
		return nil, fmt.Errorf("paliadin: insert turn row: %w", err)
	}

	// Health-gate before paying the cost of a real turn.
	if err := s.healthGate(ctx); err != nil {
		_ = s.markTurnError(ctx, turnID, "mriver_unreachable")
		return nil, err
	}

	// aichat windows are named by sanitized email_localpart (m's §13
	// Q2 pick). Look up the user's email so the window name is
	// human-readable in `tmux list-windows` on mRiver. Fall back to
	// userID-prefix if the user row is missing (e.g. fresh signups
	// pre-onboarding) — aichat's persona.SanitizeWindowName will accept
	// either.
	username := s.usernameFor(ctx, req.UserID)
	session := s.cfg.Persona + ":" + username

	// Primer pulled from paliad.paliadin_turns when this is our first
	// turn for this user-window in this Go-process lifetime. aichat is
	// stateless on user content (design §8); the client owns the
	// primer. The exchanges go in the request body; aichat injects
	// them into the envelope before the user message.
	primer := s.buildPrimerExchanges(ctx, session, req)

	// Mint the per-turn JWT (t-paliad-156). Aichat handles the file
	// write + cleanup on mRiver — we just sign and ship. When the
	// secret isn't configured, send no JWT and aichat's skill will
	// surface "JWT missing — paliad bug" rather than silently leaking
	// as service role.
	jwt, err := s.mintJWTIfConfigured(req.UserID)
	if err != nil {
		_ = s.markTurnError(ctx, turnID, "jwt_mint_failed")
		return nil, fmt.Errorf("paliadin: mint turn jwt: %w", err)
	}

	// Pass any structured TurnContext (t-paliad-161 widget payload)
	// through aichat's Meta field. Skill receives it as a [ctx …]
	// envelope segment built on the aichat side.
	meta := buildAichatMeta(req)

	body := aichatTurnRequest{
		Persona:   s.cfg.Persona,
		Username:  username,
		SessionID: req.SessionID,
		Message:   sanitiseForTmux(req.UserMessage),
		JWT:       jwt,
		Primer:    primer,
		Meta:      meta,
	}

	var resp aichatTurnResponse
	if err := s.callHTTP(ctx, http.MethodPost, "/chat/turn", body, &resp); err != nil {
		_ = s.markTurnError(ctx, turnID, classifyAichatError(err))
		return nil, err
	}

	// aichat may have just spawned the window — clear our primed-cache
	// for the session so the next turn rebuilds context. The current
	// turn already shipped its own primer block, so claude saw context
	// in this exchange.
	if resp.PaneSpawned {
		s.clearPrimed(session)
	} else {
		s.markPrimed(session)
	}

	// aichat already strips the paliadin-meta trailer (it knows the
	// persona's trailer_format). Treat resp.Response as the clean body
	// and lift Meta straight from the response envelope.
	cleanBody := resp.Response
	tokens := approxTokenCount(cleanBody)
	chipCount := countChips(cleanBody)
	finished := time.Now().UTC()
	durationMS := int(finished.Sub(startedAt) / time.Millisecond)

	tmeta := trailerMeta{
		UsedTools:     resp.Meta.UsedTools,
		ClassifierTag: resp.Meta.ClassifierTag,
		RowsSeen:      coerceAichatRowsSeen(resp.Meta.RowsSeen),
	}

	if err := s.completeTurn(ctx, turnID, finished, durationMS, cleanBody, tokens, tmeta, chipCount); err != nil {
		log.Printf("paliadin: complete turn %s: %v", turnID, err)
	}

	return &TurnResult{
		TurnID:        turnID,
		Response:      cleanBody,
		UsedTools:     tmeta.UsedTools,
		RowsSeen:      tmeta.RowsSeen,
		ChipCount:     chipCount,
		ClassifierTag: tmeta.ClassifierTag,
		DurationMS:    durationMS,
	}, nil
}

// ResetSession kills the user's window on aichat so the next RunTurn
// boots a fresh claude pane. Aichat resolves the window by sanitizing
// the same email_localpart we passed at turn time.
func (s *AichatPaliadinService) ResetSession(ctx context.Context, userID uuid.UUID) error {
	username := s.usernameFor(ctx, userID)
	session := s.cfg.Persona + ":" + username

	// Drop the cached primer flag so the next turn re-injects context
	// into the new claude pane.
	s.clearPrimed(session)

	body := aichatResetRequest{
		Persona:  s.cfg.Persona,
		Username: username,
	}
	var resp aichatResetResponse
	if err := s.callHTTP(ctx, http.MethodPost, "/chat/reset", body, &resp); err != nil {
		return fmt.Errorf("paliadin: aichat reset %s/%s: %w", s.cfg.Persona, username, err)
	}
	if !resp.OK {
		return fmt.Errorf("paliadin: aichat reset %s/%s: not ok", s.cfg.Persona, username)
	}
	return nil
}

// healthGate runs the aichat /chat/health probe at most once per 10 s.
// Returns ErrMRiverUnreachable on miss so the handler maps to the
// existing mriver_unreachable friendly-error i18n key (no new strings
// needed, per design §11).
func (s *AichatPaliadinService) healthGate(ctx context.Context) error {
	s.healthMu.Lock()
	defer s.healthMu.Unlock()

	if s.healthOK && time.Since(s.healthCheckedAt) < 10*time.Second {
		return nil
	}

	probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()

	var resp aichatHealthResponse
	if err := s.callHTTP(probeCtx, http.MethodGet, "/chat/health", nil, &resp); err != nil {
		s.healthOK = false
		return fmt.Errorf("%w: %v", ErrMRiverUnreachable, err)
	}
	if !resp.OK {
		s.healthOK = false
		return fmt.Errorf("%w: aichat health reports not ok (claude=%v tmux=%v)",
			ErrMRiverUnreachable, resp.ClaudeReachable, resp.TmuxReachable)
	}
	s.healthOK = true
	s.healthCheckedAt = time.Now()
	return nil
}

// callHTTP issues one JSON request to the aichat backend. On non-2xx
// responses it decodes the aichat error envelope into a typed error so
// classifyAichatError can map it to one of our audit codes.
//
// Tests set httpHook to bypass the network entirely.
func (s *AichatPaliadinService) callHTTP(ctx context.Context, method, path string, body any, out any) error {
	if s.httpHook != nil {
		return s.httpHook(ctx, method, path, body, out)
	}

	var reqBody io.Reader
	if body != nil {
		buf := &bytes.Buffer{}
		if err := json.NewEncoder(buf).Encode(body); err != nil {
			return fmt.Errorf("aichat: encode %s body: %w", path, err)
		}
		reqBody = buf
	}
	url := s.cfg.BaseURL + path
	httpReq, err := http.NewRequestWithContext(ctx, method, url, reqBody)
	if err != nil {
		return fmt.Errorf("aichat: build %s request: %w", path, err)
	}
	if body != nil {
		httpReq.Header.Set("Content-Type", "application/json")
	}
	if s.cfg.BearerToken != "" {
		httpReq.Header.Set("Authorization", "Bearer "+s.cfg.BearerToken)
	}

	httpResp, err := s.cfg.HTTPClient.Do(httpReq)
	if err != nil {
		return fmt.Errorf("aichat: %s %s: %w", method, path, err)
	}
	defer httpResp.Body.Close()

	respBytes, err := io.ReadAll(httpResp.Body)
	if err != nil {
		return fmt.Errorf("aichat: read %s response: %w", path, err)
	}

	if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {
		return decodeAichatError(httpResp.StatusCode, respBytes)
	}

	if out != nil {
		if err := json.Unmarshal(respBytes, out); err != nil {
			return fmt.Errorf("aichat: decode %s response: %w", path, err)
		}
	}
	return nil
}

// decodeAichatError parses aichat's wire-level error envelope. The
// envelope shape is `{"error":{"code":..., "message":..., "retryable":...}}`
// (see m/mAi internal/aichat/aierrors). We surface a typed sentinel
// error per code so classifyAichatError can map it to our audit codes.
func decodeAichatError(status int, body []byte) error {
	var env struct {
		Error struct {
			Code      string `json:"code"`
			Message   string `json:"message"`
			Retryable bool   `json:"retryable"`
		} `json:"error"`
	}
	_ = json.Unmarshal(body, &env)
	code := env.Error.Code
	msg := env.Error.Message
	if msg == "" {
		msg = strings.TrimSpace(string(body))
	}

	switch code {
	case "auth_failed":
		return fmt.Errorf("%w: %s", ErrAichatAuthFailed, msg)
	case "persona_unknown":
		return fmt.Errorf("%w: %s", ErrAichatPersonaUnknown, msg)
	case "mriver_unreachable", "bootstrap_failed":
		return fmt.Errorf("%w: %s", ErrMRiverUnreachable, msg)
	case "timeout":
		return fmt.Errorf("aichat: turn timeout: %s", msg)
	case "shim_error", "":
		return fmt.Errorf("aichat: HTTP %d: %s", status, msg)
	default:
		return fmt.Errorf("aichat: HTTP %d (%s): %s", status, code, msg)
	}
}

// classifyAichatError maps a callHTTP error onto the audit-row code
// vocabulary the frontend's friendlyErrorMessage already localises.
// Keep code strings stable — they're part of the i18n contract.
func classifyAichatError(err error) string {
	switch {
	case err == nil:
		return ""
	case errors.Is(err, ErrMRiverUnreachable):
		return "mriver_unreachable"
	case errors.Is(err, ErrAichatAuthFailed):
		return "shim_auth_failed"
	case errors.Is(err, ErrAichatPersonaUnknown):
		return "shim_error"
	case errors.Is(err, context.DeadlineExceeded):
		return "timeout"
	}
	msg := err.Error()
	switch {
	case strings.Contains(msg, "turn timeout"):
		return "timeout"
	case strings.Contains(msg, "no such host"),
		strings.Contains(msg, "connection refused"),
		strings.Contains(msg, "Connection refused"),
		strings.Contains(msg, "connect: network is unreachable"):
		return "mriver_unreachable"
	default:
		return "shim_error"
	}
}

// usernameFor resolves the aichat window name for a paliad user.
//
// Aichat windows are keyed by sanitized email_localpart per m's §13 Q2
// pick (e.g. matthias.siebels@hoganlovells.com → "matthiassiebels").
// We pass the localpart unsanitized; aichat applies persona.SanitizeWindowName
// (alphanumerics + `-`/`_`, lowercased, max 32 chars; falls back to
// "user-<uuid8>" if sanitising empties the string).
//
// Fallback when the user row is missing: userID short, which aichat
// accepts as-is. Lookup errors degrade silently — we cannot block a
// chat turn on a DB hiccup, and the worst-case window name is "user-…",
// not an outage.
func (s *AichatPaliadinService) usernameFor(ctx context.Context, userID uuid.UUID) string {
	fallback := "user-" + userID.String()[:8]
	if s.db == nil {
		return fallback
	}
	var email string
	err := s.db.QueryRowxContext(ctx,
		`SELECT email FROM paliad.users WHERE id = $1`, userID).Scan(&email)
	if err != nil || email == "" {
		return fallback
	}
	at := strings.IndexByte(email, '@')
	if at <= 0 {
		return fallback
	}
	return email[:at]
}

// buildPrimerExchanges returns up to MaxPrimerTurns prior exchanges
// from the user's paliad.paliadin_turns history, in oldest→newest
// order. Returns nil when:
//
//   - we've already primed this session in this process lifetime,
//   - the session id is empty (legacy turns predating t-paliad-161),
//   - the history lookup errors (degrade silently — the user's
//     question still ships, just without continuity).
//
// Aichat injects the returned exchanges into the envelope before the
// user message. Format details live in m/mAi internal/aichat/turn/primer.go;
// the wire payload is just a slice of {user, assistant} pairs.
func (s *AichatPaliadinService) buildPrimerExchanges(ctx context.Context, session string, req TurnRequest) []aichatPrimerExchange {
	if s.isPrimed(session) || req.SessionID == "" || s.db == nil {
		return nil
	}
	rows, err := s.ListHistoryForSession(ctx, req.UserID, req.SessionID, MaxPrimerTurns)
	if err != nil {
		log.Printf("paliadin: aichat primer history lookup: %v", err)
		return nil
	}
	if len(rows) == 0 {
		return nil
	}
	if len(rows) > MaxPrimerTurns {
		rows = rows[len(rows)-MaxPrimerTurns:]
	}
	out := make([]aichatPrimerExchange, 0, len(rows))
	for _, row := range rows {
		assistant := ""
		if row.Response != nil {
			assistant = *row.Response
		}
		out = append(out, aichatPrimerExchange{
			User:      truncateForPrimer(row.UserMessage),
			Assistant: truncateForPrimer(assistant),
		})
	}
	return out
}

// mintJWTIfConfigured signs a per-turn HS256 token for the calling
// user when JWTSecret is set. Returns "" + nil when the secret is
// unset — aichat then writes no JWT file and the SKILL.md detects the
// missing path on the next paliad.* query.
func (s *AichatPaliadinService) mintJWTIfConfigured(userID uuid.UUID) (string, error) {
	if len(s.cfg.JWTSecret) == 0 {
		return "", nil
	}
	return mintTurnJWT(userID, s.cfg.JWTTTL, s.cfg.JWTSecret)
}

// buildAichatMeta packs paliad's TurnContext into the wire-level Meta
// map aichat forwards to the envelope. Empty payload returns nil so
// aichat omits the [ctx …] segment entirely.
func buildAichatMeta(req TurnRequest) map[string]string {
	out := map[string]string{}
	if req.PageOrigin != "" {
		out["page_origin"] = req.PageOrigin
	}
	if req.Context != nil {
		c := req.Context
		if c.RouteName != "" {
			out["route"] = c.RouteName
		}
		if c.PrimaryEntityType != "" && c.PrimaryEntityID != "" {
			out["entity"] = c.PrimaryEntityType + ":" + c.PrimaryEntityID
		}
		if c.ViewMode != "" {
			out["view"] = c.ViewMode
		}
		if c.FilterSummary != "" {
			out["filter"] = c.FilterSummary
		}
		if c.UserSelectionText != "" {
			sel := c.UserSelectionText
			if len(sel) > MaxSelectionChars {
				sel = sel[:MaxSelectionChars] + "…"
			}
			out["selection"] = sel
		}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// coerceAichatRowsSeen converts aichat's wire-level RowsSeen ([]string)
// back to paliad's audit-row shape ([]int). Non-numeric entries are
// dropped — the trailer parser on the aichat side already filters but
// we guard anyway.
func coerceAichatRowsSeen(in []string) []int {
	if len(in) == 0 {
		return nil
	}
	out := make([]int, 0, len(in))
	for _, s := range in {
		var n int
		if _, err := fmt.Sscanf(strings.TrimSpace(s), "%d", &n); err == nil {
			out = append(out, n)
		}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

// =============================================================================
// primer cache — same shape as RemotePaliadinService.{is,mark,clear}Primed
// =============================================================================

func (s *AichatPaliadinService) isPrimed(session string) bool {
	s.primedMu.Lock()
	defer s.primedMu.Unlock()
	return s.primed[session]
}

func (s *AichatPaliadinService) markPrimed(session string) {
	s.primedMu.Lock()
	defer s.primedMu.Unlock()
	s.primed[session] = true
}

func (s *AichatPaliadinService) clearPrimed(session string) {
	s.primedMu.Lock()
	defer s.primedMu.Unlock()
	delete(s.primed, session)
}

// =============================================================================
// wire types — mirror m/mAi internal/aichat/api/types.go exactly so we
// can JSON-marshal directly. Kept here (rather than importing m/mAi) so
// paliad stays a self-contained module.
// =============================================================================

type aichatTurnRequest struct {
	Persona   string                 `json:"persona"`
	Username  string                 `json:"username"`
	SessionID string                 `json:"session_id,omitempty"`
	Message   string                 `json:"message"`
	JWT       string                 `json:"jwt,omitempty"`
	Primer    []aichatPrimerExchange `json:"primer,omitempty"`
	Meta      map[string]string      `json:"meta,omitempty"`
}

type aichatPrimerExchange struct {
	User      string `json:"user"`
	Assistant string `json:"assistant"`
}

type aichatTurnResponse struct {
	TurnID      string     `json:"turn_id"`
	Response    string     `json:"response"`
	Meta        aichatMeta `json:"meta"`
	DurationMs  int64      `json:"duration_ms"`
	PaneSpawned bool       `json:"pane_spawned"`
}

type aichatMeta struct {
	UsedTools     []string `json:"used_tools,omitempty"`
	RowsSeen      []string `json:"rows_seen,omitempty"`
	ClassifierTag string   `json:"classifier_tag,omitempty"`
}

type aichatResetRequest struct {
	Persona  string `json:"persona"`
	Username string `json:"username"`
}

type aichatResetResponse struct {
	OK bool `json:"ok"`
}

type aichatHealthResponse struct {
	OK              bool `json:"ok"`
	ClaudeReachable bool `json:"claude_reachable"`
	TmuxReachable   bool `json:"tmux_reachable"`
}

// Compile-time interface conformance — fail the build, not a runtime
// test, if a Paliadin method drifts off this backend.
var _ Paliadin = (*AichatPaliadinService)(nil)