kenn-io--agentsview/internal/postgres/push_fingerprint.go

package postgres

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/json"
	"fmt"
	"strings"

	"go.kenn.io/agentsview/internal/db"
)

const pushComparisonBatchSize = 900

type pushMessageAggregate struct {
	Count int
	Sum   int64
	Max   int64
	Min   int64
	SysFP string
}

type pushToolCallAggregate struct {
	Count int
	Sum   int64
}

type pushMessageComparison struct {
	MessageAggregates       map[string]pushMessageAggregate
	MessageContentHash      map[string]string
	MessageRoleTime         map[string]string
	MessageFlags            map[string]string
	MessageSystemOrdinals   map[string]string
	MessageTokenFingerprint map[string]string
	ToolCallAggregates      map[string]pushToolCallAggregate
	ToolCallFingerprint     map[string]string
	ToolResultFingerprint   map[string]string
	UsageEventFingerprint   map[string]string
}

type pushLocalMessageFingerprint struct {
	Sum           int64
	Max           int64
	Min           int64
	ContentHashFP string
	RoleTimeFP    string
	FlagsFP       string
	SystemFP      string
	ToolCallCount int
	ToolCallSum   int64
	ToolCallFP    string
	ToolResultFP  string
	TokenFP       string
	UsageEventFP  string
}

func localSessionDependencyPushFingerprint(
	ctx context.Context,
	local *db.DB,
	sessionID string,
	usageEventFingerprint string,
	usageKnown bool,
) (string, error) {
	msgFP, err := localPushMessageFingerprint(
		local, sessionID, usageEventFingerprint, usageKnown,
	)
	if err != nil {
		return "", err
	}
	findings, err := local.SessionSecretFindings(ctx, sessionID)
	if err != nil {
		return "", fmt.Errorf("reading local secret findings: %w", err)
	}
	pins, err := local.ListPinnedMessages(ctx, sessionID, "")
	if err != nil {
		return "", fmt.Errorf("reading local pinned messages: %w", err)
	}
	return hashLocalDependencyPayload(msgFP, findings, pins)
}

// hashLocalDependencyPayload is the shared final step of the per-session
// and batched dependency fingerprints. The JSON encoding is the persisted
// fingerprint format: any change re-pushes every session on the next push.
func hashLocalDependencyPayload(
	msgFP pushLocalMessageFingerprint,
	findings []db.SecretFinding,
	pins []db.PinnedMessage,
) (string, error) {
	payload := struct {
		Messages       pushLocalMessageFingerprint
		SecretFindings []db.SecretFinding
		Pins           []db.PinnedMessage
	}{
		Messages:       msgFP,
		SecretFindings: findings,
		Pins:           pins,
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return "", fmt.Errorf("encoding local dependency fingerprint: %w", err)
	}
	sum := sha256.Sum256(data)
	return fmt.Sprintf("%x", sum), nil
}

// localPushDependencyState holds one chunk's worth of prefetched local
// fingerprint inputs. The push loop fingerprints every candidate session on
// every push; reading these per session (a dozen point queries each) took
// minutes on a full push, so the loop prefetches them in batched queries.
// Values must match the per-session methods byte for byte — see the
// batched twins in internal/db for the identity contract.
type localPushDependencyState struct {
	contentAgg     map[string]db.MessageContentAggregate
	contentHashFP  map[string]string
	roleTimeFP     map[string]string
	flagsFP        map[string]string
	systemFP       map[string]string
	tokenFP        map[string]string
	toolCallCount  map[string]int
	toolCallSum    map[string]int64
	toolCallFP     map[string]string
	toolResultFP   map[string]string
	secretFindings map[string][]db.SecretFinding
	pins           map[string][]db.PinnedMessage
}

func readLocalPushDependencyState(
	ctx context.Context, local *db.DB, sessionIDs []string,
) (*localPushDependencyState, error) {
	st := &localPushDependencyState{}
	var err error
	if st.contentAgg, err = local.MessageContentFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf("computing local content fingerprints: %w", err)
	}
	if st.contentHashFP, err = local.MessageContentHashFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf("computing local content hash fingerprints: %w", err)
	}
	if st.roleTimeFP, err = local.MessageRoleTimeFingerprintsWithTimestampNormalizer(
		sessionIDs, pgPushTimestampFingerprintText,
	); err != nil {
		return nil, fmt.Errorf("computing local role/time fingerprints: %w", err)
	}
	if st.flagsFP, err = local.MessageFlagsFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf(
			"computing local message flags fingerprints: %w", err,
		)
	}
	if st.systemFP, err = local.SystemMessageFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf(
			"computing local system message fingerprints: %w", err,
		)
	}
	if st.tokenFP, err = local.MessageTokenFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf("computing local token fingerprints: %w", err)
	}
	if st.toolCallCount, err = local.ToolCallCounts(sessionIDs); err != nil {
		return nil, fmt.Errorf("counting local tool_calls: %w", err)
	}
	if st.toolCallSum, err = local.ToolCallContentFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf(
			"computing local tool_call content fingerprints: %w", err,
		)
	}
	if st.toolCallFP, err = local.ToolCallFingerprints(sessionIDs); err != nil {
		return nil, fmt.Errorf("computing local tool_call fingerprints: %w", err)
	}
	if st.toolResultFP, err = local.ToolResultEventFingerprintsWithTimestampNormalizer(
		sessionIDs, pgPushTimestampFingerprintText,
	); err != nil {
		return nil, fmt.Errorf(
			"computing local tool_result_event fingerprints: %w", err,
		)
	}
	if st.secretFindings, err = local.SessionSecretFindingsBySession(
		ctx, sessionIDs,
	); err != nil {
		return nil, fmt.Errorf("reading local secret findings: %w", err)
	}
	if st.pins, err = local.PinnedMessagesBySession(ctx, sessionIDs); err != nil {
		return nil, fmt.Errorf("reading local pinned messages: %w", err)
	}
	return st, nil
}

// dependencyFingerprint assembles the same fingerprint as
// localSessionDependencyPushFingerprint from the prefetched maps. local is
// only queried on the usageKnown=false fallback, which the push loop never
// takes (it prefetches usage fingerprints for every candidate).
func (st *localPushDependencyState) dependencyFingerprint(
	local *db.DB,
	sessionID string,
	usageEventFingerprint string,
	usageKnown bool,
) (string, error) {
	agg := st.contentAgg[sessionID]
	msgFP := pushLocalMessageFingerprint{
		Sum:           agg.Sum,
		Max:           agg.Max,
		Min:           agg.Min,
		ContentHashFP: st.contentHashFP[sessionID],
		RoleTimeFP:    st.roleTimeFP[sessionID],
		FlagsFP:       st.flagsFP[sessionID],
		SystemFP:      st.systemFP[sessionID],
		ToolCallCount: st.toolCallCount[sessionID],
		ToolCallSum:   st.toolCallSum[sessionID],
		ToolCallFP:    st.toolCallFP[sessionID],
		ToolResultFP:  st.toolResultFP[sessionID],
		TokenFP:       st.tokenFP[sessionID],
	}
	if usageKnown {
		msgFP.UsageEventFP = usageEventFingerprint
	} else {
		var err error
		msgFP.UsageEventFP, err = local.UsageEventFingerprint(sessionID)
		if err != nil {
			return "", fmt.Errorf(
				"computing local usage event fingerprint: %w", err,
			)
		}
	}
	return hashLocalDependencyPayload(
		msgFP, st.secretFindings[sessionID], st.pins[sessionID],
	)
}

func localPushMessageFingerprint(
	local *db.DB,
	sessionID string,
	usageEventFingerprint string,
	usageKnown bool,
) (pushLocalMessageFingerprint, error) {
	fp := pushLocalMessageFingerprint{}
	var err error
	fp.Sum, fp.Max, fp.Min, err = local.MessageContentFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local content fingerprint: %w", err)
	}
	fp.ContentHashFP, err = local.MessageContentHashFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local content hash fingerprint: %w", err)
	}
	fp.RoleTimeFP, err = localMessageRoleTimePGFingerprint(local, sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local role/time fingerprint: %w", err)
	}
	fp.FlagsFP, err = local.MessageFlagsFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local message flags fingerprint: %w", err)
	}
	fp.SystemFP, err = local.SystemMessageFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local system message fingerprint: %w", err)
	}
	fp.ToolCallCount, err = local.ToolCallCount(sessionID)
	if err != nil {
		return fp, fmt.Errorf("counting local tool_calls: %w", err)
	}
	fp.ToolCallSum, err = local.ToolCallContentFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local tool_call content fingerprint: %w", err)
	}
	fp.ToolCallFP, err = local.ToolCallFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local tool_call fingerprint: %w", err)
	}
	fp.ToolResultFP, err = localToolResultEventPGFingerprint(local, sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local tool_result_event fingerprint: %w", err)
	}
	fp.TokenFP, err = local.MessageTokenFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local token fingerprint: %w", err)
	}
	if usageKnown {
		fp.UsageEventFP = usageEventFingerprint
		return fp, nil
	}
	fp.UsageEventFP, err = local.UsageEventFingerprint(sessionID)
	if err != nil {
		return fp, fmt.Errorf("computing local usage event fingerprint: %w", err)
	}
	return fp, nil
}

func comparisonAggregates(
	sessionID string,
	comparisons *pushMessageComparison,
) (pushMessageAggregate, pushToolCallAggregate, bool) {
	if comparisons == nil {
		return pushMessageAggregate{}, pushToolCallAggregate{}, false
	}
	return comparisons.MessageAggregates[sessionID],
		comparisons.ToolCallAggregates[sessionID],
		true
}

func readPushSessionMessageComparisons(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
) (*pushMessageComparison, error) {
	comparisons := &pushMessageComparison{
		MessageAggregates:       make(map[string]pushMessageAggregate, len(sessionIDs)),
		MessageContentHash:      make(map[string]string, len(sessionIDs)),
		MessageRoleTime:         make(map[string]string, len(sessionIDs)),
		MessageFlags:            make(map[string]string, len(sessionIDs)),
		MessageSystemOrdinals:   make(map[string]string, len(sessionIDs)),
		MessageTokenFingerprint: make(map[string]string, len(sessionIDs)),
		ToolCallAggregates:      make(map[string]pushToolCallAggregate, len(sessionIDs)),
		ToolCallFingerprint:     make(map[string]string, len(sessionIDs)),
		ToolResultFingerprint:   make(map[string]string, len(sessionIDs)),
		UsageEventFingerprint:   make(map[string]string, len(sessionIDs)),
	}

	for i := 0; i < len(sessionIDs); i += pushComparisonBatchSize {
		end := min(i+pushComparisonBatchSize, len(sessionIDs))
		chunk := sessionIDs[i:end]

		if err := loadPushMessageAggregates(ctx, tx, chunk, comparisons.MessageAggregates); err != nil {
			return nil, err
		}
		if err := loadPushMessageContentHashFingerprints(
			ctx, tx, chunk, comparisons.MessageContentHash,
		); err != nil {
			return nil, err
		}
		if err := loadPushMessageRoleTimeFingerprints(
			ctx, tx, chunk, comparisons.MessageRoleTime,
		); err != nil {
			return nil, err
		}
		if err := loadPushMessageFlagFingerprints(
			ctx, tx, chunk, comparisons.MessageFlags,
		); err != nil {
			return nil, err
		}
		if err := loadPushMessageSystemOrdinals(
			ctx, tx, chunk, comparisons.MessageSystemOrdinals,
		); err != nil {
			return nil, err
		}
		if err := loadPushMessageTokenFingerprints(
			ctx, tx, chunk, comparisons.MessageTokenFingerprint,
		); err != nil {
			return nil, err
		}
		if err := loadPushToolCallAggregates(
			ctx, tx, chunk, comparisons.ToolCallAggregates,
		); err != nil {
			return nil, err
		}
		if err := loadPushToolCallFingerprints(
			ctx, tx, chunk, comparisons.ToolCallFingerprint,
		); err != nil {
			return nil, err
		}
		if err := loadPushToolResultEventFingerprints(
			ctx, tx, chunk, comparisons.ToolResultFingerprint,
		); err != nil {
			return nil, err
		}
		if err := loadPushUsageEventFingerprints(
			ctx, tx, chunk, comparisons.UsageEventFingerprint,
		); err != nil {
			return nil, err
		}
	}

	return comparisons, nil
}

func loadPushMessageAggregates(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]pushMessageAggregate,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, COUNT(*), COALESCE(SUM(content_length), 0),
			COALESCE(MAX(content_length), 0), COALESCE(MIN(content_length), 0),
			COALESCE(STRING_AGG(ordinal::text, ',' ORDER BY ordinal)
				FILTER (WHERE is_system), '')
		FROM messages
		WHERE session_id = ANY($1)
		GROUP BY session_id
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var sessionID string
		var count int64
		var agg pushMessageAggregate
		if err := rows.Scan(
			&sessionID, &count, &agg.Sum, &agg.Max, &agg.Min, &agg.SysFP,
		); err != nil {
			return err
		}
		agg.Count = int(count)
		out[sessionID] = agg
	}
	return rows.Err()
}

func loadPushMessageContentHashFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, ordinal, COALESCE(content, ''),
			content_length
		FROM messages
		WHERE session_id = ANY($1)
		ORDER BY session_id, ordinal ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var ordinal, contentLength int
		var content string
		if err := rows.Scan(&sessionID, &ordinal, &content, &contentLength); err != nil {
			return err
		}
		sum := sha256.Sum256([]byte(db.SanitizeUTF8(content)))
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		fmt.Fprintf(b, "%d|%d|%x;", ordinal, contentLength, sum)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushMessageRoleTimeFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, ordinal, role, timestamp
		 FROM messages
		WHERE session_id = ANY($1)
		ORDER BY session_id, ordinal ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var ordinal int
		var role string
		var timestamp sql.NullTime
		if err := rows.Scan(&sessionID, &ordinal, &role, &timestamp); err != nil {
			return err
		}
		timestampText := ""
		if timestamp.Valid {
			timestampText = pgPushTimestampFingerprintText(
				FormatISO8601(timestamp.Time),
			)
		}
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		fmt.Fprintf(
			b, "%d|%d:%s|%d:%s;",
			ordinal, len(role), role, len(timestampText), timestampText,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushMessageFlagFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, ordinal, is_system, has_thinking, has_tool_use,
			COALESCE(thinking_text, '')
		 FROM messages
		WHERE session_id = ANY($1)
		ORDER BY session_id, ordinal ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var ordinal int
		var isSystem, hasThinking, hasToolUse bool
		var thinkingText string
		if err := rows.Scan(
			&sessionID, &ordinal, &isSystem, &hasThinking, &hasToolUse,
			&thinkingText,
		); err != nil {
			return err
		}
		sum := sha256.Sum256([]byte(db.SanitizeUTF8(thinkingText)))
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		fmt.Fprintf(
			b, "%d|%t|%t|%t|%x;", ordinal, isSystem, hasThinking,
			hasToolUse, sum,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushMessageSystemOrdinals(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id,
			COALESCE(
				STRING_AGG(ordinal::text, ',' ORDER BY ordinal)
					FILTER (WHERE is_system),
				''
			)
		FROM messages
		WHERE session_id = ANY($1)
		GROUP BY session_id
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var sessionID string
		var systemOrdinals string
		if err := rows.Scan(&sessionID, &systemOrdinals); err != nil {
			return err
		}
		out[sessionID] = systemOrdinals
	}
	return rows.Err()
}

func loadPushMessageTokenFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, ordinal, model, token_usage, context_tokens,
			output_tokens, has_context_tokens, has_output_tokens,
			claude_message_id, claude_request_id,
			source_type, source_subtype, source_uuid,
			source_parent_uuid, is_sidechain, is_compact_boundary
		 FROM messages
		WHERE session_id = ANY($1)
		ORDER BY session_id, ordinal ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var ordinal, contextTokens, outputTokens int
		var model, tokenUsage string
		var hasContextTokens, hasOutputTokens bool
		var claudeMsgID, claudeReqID string
		var srcType, srcSubtype, srcUUID, srcParentUUID string
		var isSidechain, isCompactBoundary bool
		if err := rows.Scan(
			&sessionID, &ordinal, &model, &tokenUsage, &contextTokens,
			&outputTokens, &hasContextTokens, &hasOutputTokens,
			&claudeMsgID, &claudeReqID,
			&srcType, &srcSubtype, &srcUUID, &srcParentUUID,
			&isSidechain, &isCompactBoundary,
		); err != nil {
			return err
		}
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		fmt.Fprintf(
			b,
			"%d|%d:%s|%d:%s|%d|%d|%t|%t|%s|%s|"+
				"%d:%s|%d:%s|%d:%s|%d:%s|%t|%t;",
			ordinal,
			len(model), model,
			len(tokenUsage), tokenUsage,
			contextTokens, outputTokens,
			hasContextTokens, hasOutputTokens,
			claudeMsgID, claudeReqID,
			len(srcType), srcType,
			len(srcSubtype), srcSubtype,
			len(srcUUID), srcUUID,
			len(srcParentUUID), srcParentUUID,
			isSidechain, isCompactBoundary,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushToolCallAggregates(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]pushToolCallAggregate,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id,
			COUNT(*), COALESCE(SUM(result_content_length), 0)
		FROM tool_calls
		WHERE session_id = ANY($1)
		GROUP BY session_id
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var sessionID string
		var agg pushToolCallAggregate
		var count int64
		if err := rows.Scan(&sessionID, &count, &agg.Sum); err != nil {
			return err
		}
		agg.Count = int(count)
		out[sessionID] = agg
	}
	return rows.Err()
}

func loadPushToolCallFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, message_ordinal, call_index, tool_name, category,
			tool_use_id, COALESCE(input_json, ''),
			COALESCE(skill_name, ''), COALESCE(subagent_session_id, ''),
			COALESCE(result_content_length, 0),
			COALESCE(result_content, ''),
			COALESCE(file_path, '')
		 FROM tool_calls
		WHERE session_id = ANY($1)
		ORDER BY session_id, message_ordinal ASC, call_index ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var messageOrdinal, callIndex, resultContentLength int
		var toolName, category, toolUseID, inputJSON string
		var skillName, subagentSessionID, resultContent, filePath string
		if err := rows.Scan(
			&sessionID, &messageOrdinal, &callIndex, &toolName,
			&category, &toolUseID, &inputJSON,
			&skillName, &subagentSessionID, &resultContentLength,
			&resultContent, &filePath,
		); err != nil {
			return err
		}
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		fmt.Fprintf(
			b,
			"%d|%d|%d:%s|%d:%s|%d:%s|%d:%s|%d:%s|%d:%s|%d|%d:%s|%d:%s;",
			messageOrdinal, callIndex,
			len(toolName), toolName,
			len(category), category,
			len(toolUseID), toolUseID,
			len(inputJSON), inputJSON,
			len(skillName), skillName,
			len(subagentSessionID), subagentSessionID,
			resultContentLength,
			len(resultContent), resultContent,
			len(filePath), filePath,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushUsageEventFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, message_ordinal, source, model,
			input_tokens, output_tokens,
			cache_creation_input_tokens, cache_read_input_tokens,
			reasoning_tokens, cost_usd, cost_status, cost_source,
			occurred_at, dedup_key
		 FROM usage_events
		WHERE session_id = ANY($1)
		ORDER BY session_id, occurred_at NULLS FIRST, id
`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var ordinal sql.NullInt64
		var source, model, costStatus, costSource string
		var inputTokens, outputTokens int
		var cacheCreationInputTokens, cacheReadInputTokens int
		var reasoningTokens int
		var cost sql.NullFloat64
		var occurredAt sql.NullTime
		var dedupKey sql.NullString
		if err := rows.Scan(
			&sessionID, &ordinal, &source, &model,
			&inputTokens, &outputTokens,
			&cacheCreationInputTokens, &cacheReadInputTokens,
			&reasoningTokens, &cost, &costStatus, &costSource,
			&occurredAt, &dedupKey,
		); err != nil {
			return err
		}
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		occurred := ""
		if occurredAt.Valid {
			occurred = FormatISO8601(occurredAt.Time)
		}
		fmt.Fprintf(
			b,
			"%t|%d|%d:%s|%d:%s|%d|%d|%d|%d|%d|%t|%g|%d:%s|%d:%s|%d:%s|%d:%s;",
			ordinal.Valid,
			ordinal.Int64,
			len(source), source,
			len(model), model,
			inputTokens,
			outputTokens,
			cacheCreationInputTokens,
			cacheReadInputTokens,
			reasoningTokens,
			cost.Valid,
			cost.Float64,
			len(costStatus), costStatus,
			len(costSource), costSource,
			len(occurred), occurred,
			len(dedupKey.String), dedupKey.String,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func loadPushToolResultEventFingerprints(
	ctx context.Context,
	tx *sql.Tx,
	sessionIDs []string,
	out map[string]string,
) error {
	rows, err := tx.QueryContext(ctx, `
		SELECT session_id, tool_call_message_ordinal, call_index, event_index,
			COALESCE(tool_use_id, ''), COALESCE(agent_id, ''),
			COALESCE(subagent_session_id, ''), source, status,
			content, content_length, timestamp
		 FROM tool_result_events
		WHERE session_id = ANY($1)
		ORDER BY session_id, tool_call_message_ordinal ASC, call_index ASC, event_index ASC
	`, sessionIDs)
	if err != nil {
		return err
	}
	defer rows.Close()

	builders := make(map[string]*strings.Builder, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		var messageOrdinal, callIndex, eventIndex, contentLength int
		var toolUseID, agentID, subagentSessionID string
		var source, status, content string
		var timestamp sql.NullTime
		if err := rows.Scan(
			&sessionID, &messageOrdinal, &callIndex, &eventIndex,
			&toolUseID, &agentID, &subagentSessionID,
			&source, &status, &content, &contentLength, &timestamp,
		); err != nil {
			return err
		}
		b := builders[sessionID]
		if b == nil {
			b = &strings.Builder{}
			builders[sessionID] = b
		}
		timestampText := ""
		if timestamp.Valid {
			timestampText = FormatISO8601(timestamp.Time)
		}
		toolUseID = db.SanitizeUTF8(toolUseID)
		agentID = db.SanitizeUTF8(agentID)
		subagentSessionID = db.SanitizeUTF8(subagentSessionID)
		source = db.SanitizeUTF8(source)
		status = db.SanitizeUTF8(status)
		content = db.SanitizeUTF8(content)
		contentSum := sha256.Sum256([]byte(content))
		fmt.Fprintf(
			b,
			"%d|%d|%d|%d:%s|%d:%s|%d:%s|%d:%s|%d:%s|%d|%x|%d:%s;",
			messageOrdinal, callIndex, eventIndex,
			len(toolUseID), toolUseID,
			len(agentID), agentID,
			len(subagentSessionID), subagentSessionID,
			len(source), source,
			len(status), status,
			contentLength,
			contentSum,
			len(timestampText), timestampText,
		)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	for sessionID, b := range builders {
		out[sessionID] = b.String()
	}
	return nil
}

func localToolResultEventPGFingerprint(
	local *db.DB, sessionID string,
) (string, error) {
	return local.ToolResultEventFingerprintWithTimestampNormalizer(
		sessionID,
		pgPushTimestampFingerprintText,
	)
}

func pgToolResultEventFingerprint(
	ctx context.Context, tx *sql.Tx, sessionID string,
) (string, error) {
	rows, err := tx.QueryContext(ctx,
		`SELECT tool_call_message_ordinal, call_index, event_index,
			COALESCE(tool_use_id, ''), COALESCE(agent_id, ''),
			COALESCE(subagent_session_id, ''), source, status,
			content, content_length, timestamp
		 FROM tool_result_events
		 WHERE session_id = $1
		 ORDER BY tool_call_message_ordinal ASC, call_index ASC, event_index ASC`,
		sessionID,
	)
	if err != nil {
		return "", err
	}
	defer rows.Close()

	var b strings.Builder
	for rows.Next() {
		var messageOrdinal, callIndex, eventIndex, contentLength int
		var toolUseID, agentID, subagentSessionID string
		var source, status, content string
		var timestamp sql.NullTime
		if err := rows.Scan(
			&messageOrdinal, &callIndex, &eventIndex,
			&toolUseID, &agentID, &subagentSessionID,
			&source, &status, &content, &contentLength, &timestamp,
		); err != nil {
			return "", err
		}
		timestampText := ""
		if timestamp.Valid {
			timestampText = FormatISO8601(timestamp.Time)
		}
		toolUseID = db.SanitizeUTF8(toolUseID)
		agentID = db.SanitizeUTF8(agentID)
		subagentSessionID = db.SanitizeUTF8(subagentSessionID)
		source = db.SanitizeUTF8(source)
		status = db.SanitizeUTF8(status)
		content = db.SanitizeUTF8(content)
		contentSum := sha256.Sum256([]byte(content))
		fmt.Fprintf(&b,
			"%d|%d|%d|%d:%s|%d:%s|%d:%s|%d:%s|%d:%s|%d|%x|%d:%s;",
			messageOrdinal, callIndex, eventIndex,
			len(toolUseID), toolUseID,
			len(agentID), agentID,
			len(subagentSessionID), subagentSessionID,
			len(source), source,
			len(status), status,
			contentLength,
			contentSum,
			len(timestampText), timestampText,
		)
	}
	return b.String(), rows.Err()
}

func shouldSkipSessionMessages(
	sessionID string,
	localCount int,
	localFP pushLocalMessageFingerprint,
	full bool,
	comparisons *pushMessageComparison,
) bool {
	if full || localCount == 0 || comparisons == nil {
		return false
	}
	pgAgg := comparisons.MessageAggregates[sessionID]
	if pgAgg.Count != localCount || pgAgg.Count == 0 {
		return false
	}

	return localFP.Sum == pgAgg.Sum &&
		localFP.Max == pgAgg.Max &&
		localFP.Min == pgAgg.Min &&
		localFP.ContentHashFP == comparisons.MessageContentHash[sessionID] &&
		localFP.RoleTimeFP == comparisons.MessageRoleTime[sessionID] &&
		localFP.FlagsFP == comparisons.MessageFlags[sessionID] &&
		localFP.SystemFP == comparisons.MessageSystemOrdinals[sessionID] &&
		localFP.ToolCallCount == comparisons.ToolCallAggregates[sessionID].Count &&
		localFP.ToolCallSum == comparisons.ToolCallAggregates[sessionID].Sum &&
		localFP.ToolCallFP == comparisons.ToolCallFingerprint[sessionID] &&
		localFP.ToolResultFP == comparisons.ToolResultFingerprint[sessionID] &&
		localFP.TokenFP == comparisons.MessageTokenFingerprint[sessionID] &&
		localFP.UsageEventFP == comparisons.UsageEventFingerprint[sessionID]
}