tutti/packages/agent/store-sqlite/activity_delete.go

package storesqlite

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"
)

var ErrDeleteSessionsPlanChanged = errors.New("workspace agent session deletion plan changed")

// PlanClearSessions resolves the exact canonical session set owned by one
// workspace. Host uses this snapshot to close live runtimes before issuing the
// same atomic batch deletion command used by scoped deletes.
func (s *Store) PlanClearSessions(
	ctx context.Context,
	workspaceID string,
) (DeleteSessionsPlan, error) {
	if s == nil || s.db == nil {
		return DeleteSessionsPlan{}, errors.New("workspace database is not initialized")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return DeleteSessionsPlan{}, nil
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return DeleteSessionsPlan{}, fmt.Errorf("begin plan workspace agent sessions clear: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	sessionIDs, err := listAgentSessionIDsTx(ctx, tx, workspaceID)
	if err != nil {
		return DeleteSessionsPlan{}, err
	}
	if err := requireSessionForkDeleteAllowedTx(ctx, tx, workspaceID, sessionIDs); err != nil {
		return DeleteSessionsPlan{}, err
	}
	return DeleteSessionsPlan{
		WorkspaceID: workspaceID,
		SessionIDs:  normalizedSessionIDs(sessionIDs),
	}, nil
}

func (s *Store) PlanDeleteSessions(
	ctx context.Context,
	input DeleteSessionsBatchInput,
) (DeleteSessionsPlan, error) {
	if s == nil || s.db == nil {
		return DeleteSessionsPlan{}, errors.New("workspace database is not initialized")
	}
	workspaceID := strings.TrimSpace(input.WorkspaceID)
	sessionIDs := normalizedSessionIDs(input.SessionIDs)
	if workspaceID == "" || len(sessionIDs) == 0 {
		return DeleteSessionsPlan{}, nil
	}
	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
	if err != nil {
		return DeleteSessionsPlan{}, fmt.Errorf("begin plan workspace agent sessions delete: %w", err)
	}
	defer func() { _ = tx.Rollback() }()
	eligibleRootIDs, err := filterDeleteSessionRootsTx(ctx, tx, workspaceID, sessionIDs, input)
	if err != nil {
		return DeleteSessionsPlan{}, err
	}
	resolved, err := expandSessionTreeIDsTx(ctx, tx, workspaceID, eligibleRootIDs)
	if err != nil {
		return DeleteSessionsPlan{}, err
	}
	if err := requireSessionForkDeleteAllowedTx(ctx, tx, workspaceID, append(eligibleRootIDs, resolved...)); err != nil {
		return DeleteSessionsPlan{}, err
	}
	return DeleteSessionsPlan{WorkspaceID: workspaceID, SessionIDs: normalizedSessionIDs(resolved)}, nil
}

func (s *Store) DeleteSession(
	ctx context.Context,
	workspaceID string,
	agentSessionID string,
) (bool, error) {
	result, err := s.DeleteSessionWithCommit(ctx, workspaceID, agentSessionID)
	return result.RemovedSessions > 0, err
}

func (s *Store) DeleteSessionWithCommit(
	ctx context.Context,
	workspaceID string,
	agentSessionID string,
) (DeleteSessionResult, error) {
	if s == nil || s.db == nil {
		return DeleteSessionResult{}, errors.New("workspace database is not initialized")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	agentSessionID = strings.TrimSpace(agentSessionID)
	if workspaceID == "" || agentSessionID == "" {
		return DeleteSessionResult{}, nil
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return DeleteSessionResult{}, fmt.Errorf("begin delete workspace agent session: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	now := unixMs(time.Now().UTC())
	sessionClosureIDs, err := expandSessionTreeIDsTx(ctx, tx, workspaceID, []string{agentSessionID})
	if err != nil {
		return DeleteSessionResult{}, err
	}
	if err := requireSessionForkDeleteAllowedTx(ctx, tx, workspaceID, append([]string{agentSessionID}, sessionClosureIDs...)); err != nil {
		return DeleteSessionResult{}, err
	}
	mutations, err := sessionDeleteMutationsTx(ctx, tx, workspaceID, sessionClosureIDs, now)
	if err != nil {
		return DeleteSessionResult{}, err
	}
	removedMessages, removedSessions, removedSessionIDs, terminalMutations, err := deleteSessionTreeRowsTx(ctx, tx, workspaceID, sessionClosureIDs, now)
	if err != nil {
		return DeleteSessionResult{}, err
	}
	mutations = append(mutations, terminalMutations...)
	delta, err := s.commitTransaction(ctx, tx, workspaceID, mutations)
	if err != nil {
		return DeleteSessionResult{}, fmt.Errorf("commit delete workspace agent session: %w", err)
	}
	committed = true
	return DeleteSessionResult{
		TransactionID: delta.TransactionID, CommitDelta: delta,
		RemovedMessages: removedMessages, RemovedSessions: removedSessions, RemovedSessionIDs: removedSessionIDs,
	}, nil
}

func (s *Store) DeleteSessionsBatch(
	ctx context.Context,
	input DeleteSessionsBatchInput,
) (DeleteSessionsBatchResult, error) {
	if s == nil || s.db == nil {
		return DeleteSessionsBatchResult{}, errors.New("workspace database is not initialized")
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return DeleteSessionsBatchResult{}, fmt.Errorf("begin delete workspace agent sessions batch: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	result, err := s.DeleteSessionsBatchTx(ctx, tx, input)
	if err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	if err := tx.Commit(); err != nil {
		return DeleteSessionsBatchResult{}, fmt.Errorf("commit delete workspace agent sessions batch: %w", err)
	}
	committed = true
	return result, nil
}

// DeleteSessionsBatchTx deletes the full root/child closure inside a host-owned
// transaction. It lets an embedding host atomically remove domain rows that
// are keyed by agent session without introducing package-specific foreign
// keys into this reusable store.
func (s *Store) DeleteSessionsBatchTx(ctx context.Context, tx *sql.Tx, input DeleteSessionsBatchInput) (DeleteSessionsBatchResult, error) {
	if s == nil || tx == nil {
		return DeleteSessionsBatchResult{}, errors.New("workspace database is not initialized")
	}
	workspaceID := strings.TrimSpace(input.WorkspaceID)
	sessionIDs := normalizedSessionIDs(input.SessionIDs)
	if workspaceID == "" || len(sessionIDs) == 0 {
		return DeleteSessionsBatchResult{}, nil
	}
	eligibleRootIDs, err := filterDeleteSessionRootsTx(ctx, tx, workspaceID, sessionIDs, input)
	if err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	sessionClosureIDs, err := expandSessionTreeIDsTx(ctx, tx, workspaceID, eligibleRootIDs)
	if err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	if err := requireSessionForkDeleteAllowedTx(ctx, tx, workspaceID, append(eligibleRootIDs, sessionClosureIDs...)); err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	if expected := normalizedSessionIDs(input.ExpectedSessionIDs); len(expected) > 0 && !equalStrings(expected, normalizedSessionIDs(sessionClosureIDs)) {
		return DeleteSessionsBatchResult{}, ErrDeleteSessionsPlanChanged
	}
	now := unixMs(time.Now().UTC())
	mutations, err := sessionDeleteMutationsTx(ctx, tx, workspaceID, sessionClosureIDs, now)
	if err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	removedMessages, removedSessions, removedSessionIDs, terminalMutations, err := deleteSessionTreeRowsTx(ctx, tx, workspaceID, sessionClosureIDs, now)
	if err != nil {
		return DeleteSessionsBatchResult{}, err
	}
	mutations = append(mutations, terminalMutations...)
	delta, err := s.participateTransaction(ctx, tx, workspaceID, mutations)
	if err != nil {
		return DeleteSessionsBatchResult{}, fmt.Errorf("participate in delete workspace agent sessions batch: %w", err)
	}
	return DeleteSessionsBatchResult{
		TransactionID:     delta.TransactionID,
		CommitDelta:       delta,
		RemovedMessages:   removedMessages,
		RemovedSessions:   removedSessions,
		RemovedSessionIDs: removedSessionIDs,
	}, nil
}

// filterDeleteSessionRootsTx applies optional ownership predicates to the
// requested root set. Project removal uses these predicates as a commit-time
// fence: a root that is pinned or moved after discovery is excluded, causing
// the expected deletion plan to change instead of deleting stale candidates.
func filterDeleteSessionRootsTx(
	ctx context.Context,
	tx *sql.Tx,
	workspaceID string,
	sessionIDs []string,
	input DeleteSessionsBatchInput,
) ([]string, error) {
	sectionKey := strings.TrimSpace(input.RequiredRootRailSectionKey)
	if sectionKey == "" && !input.ExcludePinnedRoots {
		return sessionIDs, nil
	}
	if len(sessionIDs) == 0 {
		return nil, nil
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(sessionIDs)), ",")
	args := make([]any, 0, len(sessionIDs)+2)
	args = append(args, workspaceID)
	for _, sessionID := range sessionIDs {
		args = append(args, sessionID)
	}
	query := `
SELECT agent_session_id
FROM workspace_agent_sessions
WHERE workspace_id = ?
  AND agent_session_id IN (` + placeholders + `)
  AND session_kind = 'root'
  AND deleted_at_unix_ms = 0`
	if sectionKey != "" {
		query += " AND rail_section_key = ?"
		args = append(args, sectionKey)
	}
	if input.ExcludePinnedRoots {
		query += " AND pinned_at_unix_ms = 0"
	}
	rows, err := tx.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("filter workspace agent session deletion roots: %w", err)
	}
	defer rows.Close()
	eligible := make([]string, 0, len(sessionIDs))
	for rows.Next() {
		var sessionID string
		if err := rows.Scan(&sessionID); err != nil {
			return nil, fmt.Errorf("scan workspace agent session deletion root: %w", err)
		}
		eligible = append(eligible, sessionID)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workspace agent session deletion roots: %w", err)
	}
	return normalizedSessionIDs(eligible), nil
}

func normalizedSessionIDs(values []string) []string {
	result := make([]string, 0, len(values))
	seen := make(map[string]struct{}, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, exists := seen[value]; exists {
			continue
		}
		seen[value] = struct{}{}
		result = append(result, value)
	}
	sort.Strings(result)
	return result
}

func equalStrings(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if left[index] != right[index] {
			return false
		}
	}
	return true
}

func (s *Store) ClearSessions(
	ctx context.Context,
	workspaceID string,
) (ClearSessionsResult, error) {
	if s == nil || s.db == nil {
		return ClearSessionsResult{}, errors.New("workspace database is not initialized")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return ClearSessionsResult{}, nil
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("begin clear workspace agent sessions: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	removedSessionIDs, err := listAgentSessionIDsTx(ctx, tx, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	mutations, err := sessionDeleteMutationsTx(ctx, tx, workspaceID, removedSessionIDs, unixMs(time.Now().UTC()))
	if err != nil {
		return ClearSessionsResult{}, err
	}
	result, err := s.clearSessionsTx(ctx, tx, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	delta, err := s.commitTransaction(ctx, tx, workspaceID, mutations)
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("commit clear workspace agent sessions: %w", err)
	}
	committed = true
	result.TransactionID = delta.TransactionID
	result.CommitDelta = delta
	return result, nil
}

// ClearSessionsTx hard-deletes a workspace's sessions and messages within
// the caller's transaction. Hosts that delete a workspace of their own and
// need the agent-row cascade to be atomic with that deletion should run
// both through one transaction via this method; the caller owns commit and
// rollback. The configured participant runs before return, so its marker is
// governed by that same caller-owned commit.
func (s *Store) ClearSessionsTx(
	ctx context.Context,
	tx *sql.Tx,
	workspaceID string,
) (ClearSessionsResult, error) {
	removedSessionIDs, err := listAgentSessionIDsTx(ctx, tx, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	mutations, err := sessionDeleteMutationsTx(ctx, tx, workspaceID, removedSessionIDs, unixMs(time.Now().UTC()))
	if err != nil {
		return ClearSessionsResult{}, err
	}
	result, err := s.clearSessionsTx(ctx, tx, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	delta, err := s.participateTransaction(ctx, tx, workspaceID, mutations)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	result.TransactionID = delta.TransactionID
	result.CommitDelta = delta
	return result, nil
}

func (s *Store) clearSessionsTx(
	ctx context.Context,
	tx *sql.Tx,
	workspaceID string,
) (ClearSessionsResult, error) {
	if s == nil || tx == nil {
		return ClearSessionsResult{}, errors.New("workspace database is not initialized")
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return ClearSessionsResult{}, nil
	}

	removedSessionIDs, err := listAgentSessionIDsTx(ctx, tx, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, err
	}
	if err := requireSessionForkDeleteAllowedTx(ctx, tx, workspaceID, removedSessionIDs); err != nil {
		return ClearSessionsResult{}, err
	}
	messageResult, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_messages
WHERE workspace_id = ?
`, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent messages: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_submit_claims WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent submit claims: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_runtime_operation_events WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent runtime operation events: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_runtime_operations WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent runtime operations: %w", err)
	}
	// Goal cleanup is explicit and ordered because hosts may run with foreign
	// keys disabled; a reused session id must never inherit an old Goal saga.
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_goal_provenance_ledger WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal provenance ledger: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_goal_reconcile_inbox WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal reconcile inbox: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_goal_repair_incidents WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal repair incidents: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_goal_generation_fences WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal generation fences: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_goal_control_operations WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal operations: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `DELETE FROM workspace_agent_session_goals WHERE workspace_id = ?`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent goal state: %w", err)
	}
	// Explicit deletes rather than FK cascades: SQLite only cascades with
	// PRAGMA foreign_keys enabled, which hosts do not guarantee.
	if _, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_interactions
WHERE workspace_id = ?
`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent interactions: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_turn_submissions
WHERE workspace_id = ?
`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent turn submissions: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_turn_history
WHERE workspace_id = ?
`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent turn history: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_turns
WHERE workspace_id = ?
`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent turns: %w", err)
	}
	if _, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_session_history
WHERE workspace_id = ?
`, workspaceID); err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent session history: %w", err)
	}
	sessionResult, err := tx.ExecContext(ctx, `
DELETE FROM workspace_agent_sessions
WHERE workspace_id = ?
`, workspaceID)
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent sessions: %w", err)
	}
	removedMessages, err := messageResult.RowsAffected()
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent messages rows affected: %w", err)
	}
	removedSessions, err := sessionResult.RowsAffected()
	if err != nil {
		return ClearSessionsResult{}, fmt.Errorf("clear workspace agent sessions rows affected: %w", err)
	}
	return ClearSessionsResult{
		RemovedMessages:   int(removedMessages),
		RemovedSessions:   int(removedSessions),
		RemovedSessionIDs: removedSessionIDs,
	}, nil
}

func sessionDeleteMutations(workspaceID string, sessionIDs []string, version int64) []TransactionMutation {
	mutations := make([]TransactionMutation, 0, len(sessionIDs))
	for _, agentSessionID := range sessionIDs {
		mutations = append(mutations, transactionMutation(
			workspaceID, agentSessionID, MutationEntitySession, agentSessionID, "delete", version,
		))
	}
	return mutations
}

func listAgentSessionIDsTx(ctx context.Context, tx *sql.Tx, workspaceID string) ([]string, error) {
	rows, err := tx.QueryContext(ctx, `
SELECT agent_session_id
FROM workspace_agent_sessions
WHERE workspace_id = ?
ORDER BY updated_at_unix_ms DESC, agent_session_id ASC
`, workspaceID)
	if err != nil {
		return nil, fmt.Errorf("list workspace agent sessions for clear: %w", err)
	}
	defer rows.Close()

	sessionIDs := make([]string, 0)
	for rows.Next() {
		var sessionID string
		if err := rows.Scan(&sessionID); err != nil {
			return nil, fmt.Errorf("scan workspace agent session id for clear: %w", err)
		}
		sessionID = strings.TrimSpace(sessionID)
		if sessionID != "" {
			sessionIDs = append(sessionIDs, sessionID)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate workspace agent session ids for clear: %w", err)
	}
	return sessionIDs, nil
}