kandev/apps/backend/internal/github/store.go

package github

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jmoiron/sqlx"

	dbutil "github.com/kandev/kandev/internal/db"
	"github.com/kandev/kandev/internal/db/dialect"
	"github.com/kandev/kandev/internal/persistence"
)

// Store provides SQLite persistence for GitHub integration data.
type Store struct {
	db                         *sqlx.DB // writer
	ro                         *sqlx.DB // reader
	freshInstall               bool
	deploymentAppPersistenceMu sync.Mutex
	appLifecycleLocksMu        sync.Mutex
	appLifecycleLocks          map[string]*appRegistrationLifecycleLock
	prWatchMigration           *PRWatchMigrationStats
}

type appRegistrationLifecycleLock struct {
	mu   sync.Mutex
	refs int
}

type taskIssueMetadataRow struct {
	TaskID    string `db:"task_id"`
	TaskTitle string `db:"task_title"`
	Metadata  string `db:"metadata"`
}

// NewStore creates a new GitHub store and initializes the schema.
func NewStore(writer, reader *sqlx.DB) (*Store, error) {
	s := &Store{
		db: writer, ro: reader,
		appLifecycleLocks: make(map[string]*appRegistrationLifecycleLock),
	}
	settingsExists, err := dbutil.TableExists(s.db, "github_workspace_settings")
	if err != nil {
		return nil, fmt.Errorf("probe GitHub workspace settings schema: %w", err)
	}
	connectionsExists, err := dbutil.TableExists(s.db, "github_workspace_connections")
	if err != nil {
		return nil, fmt.Errorf("probe GitHub workspace connections schema: %w", err)
	}
	s.freshInstall = !settingsExists && !connectionsExists
	legacyUpgrade := settingsExists && !connectionsExists
	if err := s.initSchema(legacyUpgrade); err != nil {
		return nil, fmt.Errorf("github schema init: %w", err)
	}
	return s, nil
}

func (s *Store) lockAppRegistrationLifecycle(registrationID string) func() {
	s.appLifecycleLocksMu.Lock()
	lock := s.appLifecycleLocks[registrationID]
	if lock == nil {
		lock = &appRegistrationLifecycleLock{}
		s.appLifecycleLocks[registrationID] = lock
	}
	lock.refs++
	s.appLifecycleLocksMu.Unlock()

	lock.mu.Lock()
	return func() {
		lock.mu.Unlock()
		s.appLifecycleLocksMu.Lock()
		lock.refs--
		if lock.refs == 0 {
			delete(s.appLifecycleLocks, registrationID)
		}
		s.appLifecycleLocksMu.Unlock()
	}
}

// createTablesSQL holds the DDL for all GitHub integration tables.
//
// Multi-repo: both `github_pr_watches` and `github_task_prs` carry a
// `repository_id` that names the per-task repository (when the task spans
// multiple repos). The uniqueness constraints include `repository_id` so
// each repo can have its own watch/PR for the same session/task. Existing
// installs migrated from the single-repo schema get the column dropped to
// `”` (empty) and the constraints rebuilt by `migratePRTablesForMultiRepo`.
const createTablesSQL = `
	-- github_pr_watches is task-owned (ADR 2026-08-31-task-owned-pr-watch-identity):
	-- a searching watch (pr_number = 0) is canonical by (task_id, repository_id,
	-- branch) and a discovered watch (pr_number != 0) is canonical by
	-- (task_id, repository_id, pr_number). Both are enforced by the partial
	-- unique indexes in applyIdempotentSchemaIndexes, not an inline UNIQUE
	-- constraint here (SQLite can't express a partial UNIQUE inline).
	-- session_id is optional provenance only - never a uniqueness component.
	CREATE TABLE IF NOT EXISTS github_pr_watches (
		id TEXT PRIMARY KEY,
		workspace_id TEXT NOT NULL DEFAULT '',
		session_id TEXT NOT NULL DEFAULT '',
		task_id TEXT NOT NULL,
		repository_id TEXT NOT NULL DEFAULT '',
		owner TEXT NOT NULL,
		repo TEXT NOT NULL,
		pr_number INTEGER NOT NULL,
		branch TEXT NOT NULL,
		last_checked_at DATETIME,
		last_comment_at DATETIME,
		last_check_status TEXT DEFAULT '',
		last_review_state TEXT DEFAULT '',
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL
	);

	CREATE TABLE IF NOT EXISTS github_task_prs (
		id TEXT PRIMARY KEY,
		workspace_id TEXT NOT NULL DEFAULT '',
		task_id TEXT NOT NULL,
		repository_id TEXT NOT NULL DEFAULT '',
		owner TEXT NOT NULL,
		repo TEXT NOT NULL,
		pr_number INTEGER NOT NULL,
		pr_url TEXT NOT NULL,
		pr_title TEXT NOT NULL,
		head_branch TEXT NOT NULL,
		base_branch TEXT NOT NULL,
		head_sha TEXT NOT NULL DEFAULT '',
		author_login TEXT NOT NULL,
		state TEXT NOT NULL DEFAULT 'open',
		review_state TEXT NOT NULL DEFAULT '',
		checks_state TEXT NOT NULL DEFAULT '',
		mergeable_state TEXT NOT NULL DEFAULT '',
		merge_queue_state TEXT NOT NULL DEFAULT '',
		merge_queue_position INTEGER,
		merge_queue_entry_id TEXT NOT NULL DEFAULT '',
		merge_queue_entry_head_sha TEXT NOT NULL DEFAULT '',
		merge_queue_estimated_time_to_merge_seconds INTEGER,
		merge_queue_last_removal_id TEXT NOT NULL DEFAULT '',
		merge_queue_last_removed_at DATETIME,
		merge_queue_last_removal_reason TEXT NOT NULL DEFAULT '',
		merge_queue_last_removal_before_sha TEXT NOT NULL DEFAULT '',
		review_count INTEGER DEFAULT 0,
		pending_review_count INTEGER DEFAULT 0,
		required_reviews INTEGER,
		comment_count INTEGER DEFAULT 0,
		unresolved_review_threads INTEGER DEFAULT 0,
		checks_total INTEGER DEFAULT 0,
		checks_passing INTEGER DEFAULT 0,
		additions INTEGER DEFAULT 0,
		deletions INTEGER DEFAULT 0,
		created_at DATETIME NOT NULL,
		merged_at DATETIME,
		closed_at DATETIME,
		last_synced_at DATETIME,
		detached_at DATETIME,
		updated_at DATETIME NOT NULL,
		is_draft BOOLEAN,
		changed_files INTEGER,
		merged_by_login TEXT,
		closed_by_login TEXT,
		auto_merge_observed_at DATETIME,
		source TEXT NOT NULL DEFAULT '',
		workflow_attention TEXT NOT NULL DEFAULT '',
		UNIQUE(task_id, repository_id, pr_number)
	);

	CREATE TABLE IF NOT EXISTS github_review_watches (
		id TEXT PRIMARY KEY,
		workspace_id TEXT NOT NULL,
		workflow_id TEXT NOT NULL,
		workflow_step_id TEXT NOT NULL,
		repos TEXT NOT NULL DEFAULT '[]',
		agent_profile_id TEXT NOT NULL,
		executor_profile_id TEXT NOT NULL,
		prompt TEXT DEFAULT '',
		review_scope TEXT NOT NULL DEFAULT 'user_and_teams',
		custom_query TEXT NOT NULL DEFAULT '',
		target_login TEXT NOT NULL DEFAULT '',
		enabled BOOLEAN DEFAULT 1,
		poll_interval_seconds INTEGER DEFAULT 300,
		cleanup_policy TEXT NOT NULL DEFAULT 'auto',
		last_polled_at DATETIME,
		last_error TEXT NOT NULL DEFAULT '',
		last_error_at DATETIME,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL
	);

	CREATE TABLE IF NOT EXISTS github_review_pr_tasks (
		id TEXT PRIMARY KEY,
		review_watch_id TEXT NOT NULL,
		repo_owner TEXT NOT NULL DEFAULT '',
		repo_name TEXT NOT NULL DEFAULT '',
		pr_number INTEGER NOT NULL,
		pr_url TEXT NOT NULL,
		task_id TEXT NOT NULL,
		created_at DATETIME NOT NULL,
		UNIQUE(review_watch_id, repo_owner, repo_name, pr_number)
	);

	CREATE TABLE IF NOT EXISTS github_issue_watches (
		id TEXT PRIMARY KEY,
		workspace_id TEXT NOT NULL,
		workflow_id TEXT NOT NULL,
		workflow_step_id TEXT NOT NULL,
		repos TEXT NOT NULL DEFAULT '[]',
		agent_profile_id TEXT NOT NULL,
		executor_profile_id TEXT NOT NULL,
		prompt TEXT DEFAULT '',
		labels TEXT NOT NULL DEFAULT '[]',
		custom_query TEXT NOT NULL DEFAULT '',
		enabled BOOLEAN DEFAULT 1,
		poll_interval_seconds INTEGER DEFAULT 300,
		cleanup_policy TEXT NOT NULL DEFAULT 'auto',
		last_polled_at DATETIME,
		last_error TEXT NOT NULL DEFAULT '',
		last_error_at DATETIME,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL
	);

	CREATE TABLE IF NOT EXISTS github_issue_watch_tasks (
		id TEXT PRIMARY KEY,
		issue_watch_id TEXT NOT NULL,
		repo_owner TEXT NOT NULL DEFAULT '',
		repo_name TEXT NOT NULL DEFAULT '',
		issue_number INTEGER NOT NULL,
		issue_url TEXT NOT NULL,
		task_id TEXT NOT NULL,
		created_at DATETIME NOT NULL,
		UNIQUE(issue_watch_id, repo_owner, repo_name, issue_number)
	);

	CREATE TABLE IF NOT EXISTS github_action_presets (
		workspace_id TEXT PRIMARY KEY,
		pr_presets TEXT NOT NULL DEFAULT '[]',
		issue_presets TEXT NOT NULL DEFAULT '[]',
		updated_at DATETIME NOT NULL
	);

	CREATE TABLE IF NOT EXISTS github_workspace_settings (
		workspace_id TEXT PRIMARY KEY,
		task_git_credentials_mode TEXT NOT NULL DEFAULT 'managed',
		repo_scope_mode TEXT NOT NULL DEFAULT 'all',
		repo_scope_orgs TEXT NOT NULL DEFAULT '[]',
		repo_scope_repos TEXT NOT NULL DEFAULT '[]',
		saved_presets TEXT NOT NULL DEFAULT '[]',
		default_query_presets TEXT,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL
	);

	CREATE TABLE IF NOT EXISTS github_workspace_connections (
		workspace_id TEXT PRIMARY KEY,
		source TEXT NOT NULL CHECK (source IN ('legacy_shared', 'pat', 'gh_cli', 'github_app_installation')),
		github_host TEXT NOT NULL CHECK (github_host <> ''),
		login TEXT,
		installation_id BIGINT,
		installation_account_login TEXT,
		installation_account_type TEXT CHECK (
			installation_account_type IS NULL OR installation_account_type IN ('User', 'Organization')
		),
		app_registration_id TEXT,
		status TEXT NOT NULL CHECK (status IN ('active', 'invalid', 'suspended', 'revoked')),
		credential_generation BIGINT NOT NULL DEFAULT 1 CHECK (credential_generation > 0),
		last_error TEXT,
		created_at TIMESTAMP NOT NULL,
		updated_at TIMESTAMP NOT NULL,
		CHECK (
			(source = 'legacy_shared' AND login IS NULL AND installation_id IS NULL AND app_registration_id IS NULL) OR
			(source IN ('pat', 'gh_cli') AND login IS NOT NULL AND login <> '' AND installation_id IS NULL AND app_registration_id IS NULL) OR
			(source = 'github_app_installation' AND installation_id IS NOT NULL AND installation_id > 0 AND
			 installation_account_login IS NOT NULL AND installation_account_login <> '' AND
			 installation_account_type IS NOT NULL AND app_registration_id IS NOT NULL)
		),
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE,
		FOREIGN KEY (app_registration_id) REFERENCES github_app_registrations(id) ON DELETE RESTRICT
	);

	CREATE TABLE IF NOT EXISTS github_user_connections (
		workspace_id TEXT NOT NULL,
		user_id TEXT NOT NULL,
		app_registration_id TEXT NOT NULL,
		github_user_id BIGINT NOT NULL CHECK (github_user_id > 0),
		login TEXT NOT NULL CHECK (login <> ''),
		status TEXT NOT NULL CHECK (status IN ('active', 'invalid', 'revoked')),
		access_expires_at TIMESTAMP NOT NULL,
		refresh_expires_at TIMESTAMP,
		credential_generation BIGINT NOT NULL DEFAULT 1 CHECK (credential_generation > 0),
		last_error TEXT,
		created_at TIMESTAMP NOT NULL,
		updated_at TIMESTAMP NOT NULL,
		PRIMARY KEY (workspace_id, user_id),
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE,
		FOREIGN KEY (app_registration_id) REFERENCES github_app_registrations(id) ON DELETE RESTRICT
	);

	CREATE TRIGGER IF NOT EXISTS github_user_connections_registration_insert
	BEFORE INSERT ON github_user_connections
	BEGIN
		SELECT CASE WHEN NOT EXISTS (
			SELECT 1 FROM github_workspace_connections workspace
			WHERE workspace.workspace_id = NEW.workspace_id
				AND workspace.source = 'github_app_installation'
				AND workspace.app_registration_id = NEW.app_registration_id
		) THEN RAISE(ABORT, 'personal GitHub App registration must match workspace') END;
	END;

	CREATE TRIGGER IF NOT EXISTS github_user_connections_registration_update
	BEFORE UPDATE OF workspace_id, app_registration_id ON github_user_connections
	BEGIN
		SELECT CASE WHEN NOT EXISTS (
			SELECT 1 FROM github_workspace_connections workspace
			WHERE workspace.workspace_id = NEW.workspace_id
				AND workspace.source = 'github_app_installation'
				AND workspace.app_registration_id = NEW.app_registration_id
		) THEN RAISE(ABORT, 'personal GitHub App registration must match workspace') END;
	END;

	CREATE TRIGGER IF NOT EXISTS github_workspace_connections_registration_update
	BEFORE UPDATE OF source, app_registration_id ON github_workspace_connections
	BEGIN
		SELECT CASE WHEN EXISTS (
			SELECT 1 FROM github_user_connections personal
			WHERE personal.workspace_id = OLD.workspace_id
				AND (
					NEW.source <> 'github_app_installation' OR
					NEW.app_registration_id IS NULL OR
					personal.app_registration_id <> NEW.app_registration_id
				)
		) THEN RAISE(ABORT, 'workspace GitHub App registration must match personal connections') END;
	END;

	CREATE TRIGGER IF NOT EXISTS github_workspace_connections_registration_delete
	BEFORE DELETE ON github_workspace_connections
	BEGIN
		SELECT CASE WHEN EXISTS (
			SELECT 1 FROM github_user_connections personal
			WHERE personal.workspace_id = OLD.workspace_id
		) THEN RAISE(ABORT, 'workspace GitHub App connection still has personal connections') END;
	END;

	CREATE TABLE IF NOT EXISTS github_user_connection_versions (
		workspace_id TEXT NOT NULL,
		user_id TEXT NOT NULL,
		credential_generation BIGINT NOT NULL DEFAULT 0 CHECK (credential_generation >= 0),
		updated_at TIMESTAMP NOT NULL,
		PRIMARY KEY (workspace_id, user_id),
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE
	);

	CREATE TABLE IF NOT EXISTS github_auth_flows (
		state_hash TEXT PRIMARY KEY,
		workspace_id TEXT NOT NULL,
		user_id TEXT NOT NULL,
		app_registration_id TEXT NOT NULL,
		kind TEXT NOT NULL CHECK (kind IN ('app_installation', 'personal')),
		pkce_verifier TEXT NOT NULL DEFAULT '',
		expected_workspace_source TEXT NOT NULL DEFAULT '',
		expected_workspace_generation BIGINT NOT NULL DEFAULT 0,
		expected_installation_id BIGINT,
		expected_workspace_app_registration_id TEXT,
		expected_personal_generation BIGINT NOT NULL DEFAULT 0,
		expires_at TIMESTAMP NOT NULL,
		consumed_at TIMESTAMP,
		created_at TIMESTAMP NOT NULL,
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE,
		FOREIGN KEY (app_registration_id) REFERENCES github_app_registrations(id) ON DELETE RESTRICT
	);

	CREATE TABLE IF NOT EXISTS github_webhook_deliveries (
		app_registration_id TEXT NOT NULL,
		delivery_id TEXT NOT NULL,
		event TEXT NOT NULL,
		status TEXT NOT NULL CHECK (status IN ('received', 'processed', 'ignored', 'failed')),
		result TEXT NOT NULL DEFAULT '',
		received_at TIMESTAMP NOT NULL,
		processed_at TIMESTAMP,
		PRIMARY KEY (app_registration_id, delivery_id),
		FOREIGN KEY (app_registration_id) REFERENCES github_app_registrations(id) ON DELETE RESTRICT
	);

	CREATE TABLE IF NOT EXISTS github_task_ci_options (
		task_id TEXT PRIMARY KEY,
		auto_fix_enabled BOOLEAN NOT NULL DEFAULT 0,
		auto_merge_enabled BOOLEAN NOT NULL DEFAULT 0,
		auto_fix_prompt_override TEXT,
		prompt_on_review_requested BOOLEAN NOT NULL DEFAULT 0,
		prompt_on_merged BOOLEAN NOT NULL DEFAULT 0,
		prompt_on_closed BOOLEAN NOT NULL DEFAULT 0,
		review_reviewer_login TEXT NOT NULL DEFAULT '',
		review_prompt_override TEXT,
		merged_prompt_override TEXT,
		closed_prompt_override TEXT,
		pr_scope_migrated_at DATETIME,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL
	);

	-- Per-PR automation switches. Source of truth for the five automation
	-- switches (AutoFixEnabled, AutoMergeEnabled, PromptOnReviewRequested,
	-- PromptOnMerged, PromptOnClosed) formerly stored task-wide on
	-- github_task_ci_options. See migrateTaskCIOptionsToPRScope.
	CREATE TABLE IF NOT EXISTS github_task_pr_automation_options (
		task_id TEXT NOT NULL,
		repository_id TEXT NOT NULL DEFAULT '',
		pr_number INTEGER NOT NULL,
		auto_fix_enabled BOOLEAN NOT NULL DEFAULT 0,
		auto_merge_enabled BOOLEAN NOT NULL DEFAULT 0,
		prompt_on_review_requested BOOLEAN NOT NULL DEFAULT 0,
		prompt_on_merged BOOLEAN NOT NULL DEFAULT 0,
		prompt_on_closed BOOLEAN NOT NULL DEFAULT 0,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL,
		PRIMARY KEY (task_id, repository_id, pr_number)
	);

	CREATE TABLE IF NOT EXISTS github_task_ci_pr_state (
		task_id TEXT NOT NULL,
		repository_id TEXT NOT NULL DEFAULT '',
		pr_number INTEGER NOT NULL,
		last_fix_signature TEXT NOT NULL DEFAULT '',
		last_fix_checkpoint_json TEXT NOT NULL DEFAULT '',
		last_fix_enqueued_at DATETIME,
		last_fix_session_id TEXT,
		auto_fix_round_count INTEGER NOT NULL DEFAULT 0,
		auto_fix_exhausted_at DATETIME,
		last_merge_signature TEXT NOT NULL DEFAULT '',
		last_merge_attempt_at DATETIME,
		last_merge_result TEXT NOT NULL DEFAULT '',
		merge_retry_pending BOOLEAN NOT NULL DEFAULT 0,
		last_queue_attempt_head_sha TEXT NOT NULL DEFAULT '',
		last_queue_fix_event_id TEXT NOT NULL DEFAULT '',
		last_queue_removal_cause TEXT NOT NULL DEFAULT '',
		review_request_initialized BOOLEAN NOT NULL DEFAULT 0,
		last_review_requested BOOLEAN NOT NULL DEFAULT 0,
		last_observed_pr_state TEXT NOT NULL DEFAULT '',
		last_lifecycle_event TEXT NOT NULL DEFAULT '',
		last_lifecycle_prompt_at DATETIME,
		last_lifecycle_session_id TEXT,
		last_error TEXT,
		last_error_kind TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_state TEXT NOT NULL DEFAULT 'acknowledged',
		auto_fix_attempt_queue_entry_id TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_session_id TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_turn_id TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_signature TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_provider_generation TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_outcome TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_summary TEXT NOT NULL DEFAULT '',
		auto_fix_attempt_started_at DATETIME,
		auto_fix_attempt_outcome_at DATETIME,
		auto_fix_attempt_progress_deadline DATETIME,
		created_at DATETIME NOT NULL,
		updated_at DATETIME NOT NULL,
		PRIMARY KEY (task_id, repository_id, pr_number)
	);
`

const appRegistrationTablesSQL = `
	CREATE TABLE IF NOT EXISTS github_app_registrations (
		id TEXT PRIMARY KEY CHECK (id <> ''),
		source TEXT NOT NULL CHECK (source IN ('managed', 'imported')),
		display_name TEXT NOT NULL CHECK (display_name <> ''),
		github_host TEXT NOT NULL CHECK (github_host <> ''),
		app_id BIGINT NOT NULL CHECK (app_id > 0),
		client_id TEXT NOT NULL CHECK (client_id <> ''),
		slug TEXT NOT NULL CHECK (slug <> ''),
		owner_login TEXT NOT NULL CHECK (owner_login <> ''),
		owner_type TEXT NOT NULL CHECK (owner_type IN ('User', 'Organization')),
		visibility TEXT NOT NULL CHECK (visibility IN ('private', 'public')),
		public_base_url TEXT NOT NULL CHECK (public_base_url <> ''),
		created_for_workspace_id TEXT,
		credential_generation BIGINT NOT NULL CHECK (credential_generation > 0),
		credential_secret_id TEXT NOT NULL CHECK (credential_secret_id <> ''),
		status TEXT NOT NULL CHECK (status IN ('active', 'invalid')),
		webhook_status TEXT NOT NULL CHECK (webhook_status IN ('unverified', 'verified', 'failing')),
		last_webhook_at TIMESTAMP,
		last_error TEXT,
		created_at TIMESTAMP NOT NULL,
		updated_at TIMESTAMP NOT NULL,
		UNIQUE (github_host, app_id)
	);

	CREATE TABLE IF NOT EXISTS github_app_registration_flows (
		state_hash TEXT PRIMARY KEY,
		registration_id TEXT NOT NULL CHECK (registration_id <> ''),
		workspace_id TEXT NOT NULL CHECK (workspace_id <> ''),
		user_id TEXT NOT NULL CHECK (user_id <> ''),
		owner_type TEXT NOT NULL CHECK (owner_type IN ('User', 'Organization')),
		owner_login TEXT NOT NULL CHECK (owner_login <> ''),
		display_name TEXT NOT NULL CHECK (display_name <> ''),
		visibility TEXT NOT NULL CHECK (visibility IN ('private', 'public')),
		public_base_url TEXT NOT NULL CHECK (public_base_url <> ''),
		manifest_revision INTEGER NOT NULL CHECK (manifest_revision > 0),
		expires_at TIMESTAMP NOT NULL,
		consumed_at TIMESTAMP,
		created_at TIMESTAMP NOT NULL,
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE
	);

	CREATE TABLE IF NOT EXISTS github_app_import_preparations (
		registration_id TEXT PRIMARY KEY CHECK (registration_id <> ''),
		workspace_id TEXT NOT NULL CHECK (workspace_id <> ''),
		user_id TEXT NOT NULL CHECK (user_id <> ''),
		public_base_url TEXT NOT NULL CHECK (public_base_url <> ''),
		expires_at TIMESTAMP NOT NULL,
		consumed_at TIMESTAMP,
		created_at TIMESTAMP NOT NULL,
		FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE
	);
`

func (s *Store) initSchema(legacyUpgrade bool) error {
	if err := s.initSchemaFoundations(); err != nil {
		return err
	}
	if err := s.applyIdempotentSchemaColumns(); err != nil {
		return err
	}
	if err := s.initSchemaUpgrades(); err != nil {
		return err
	}
	if err := s.activateTaskPROutcomeTracking(); err != nil {
		return err
	}
	if err := s.initSchemaData(legacyUpgrade); err != nil {
		return err
	}
	if err := s.applyIdempotentSchemaIndexes(); err != nil {
		return err
	}
	return s.ensureWorkspaceOwnershipIndexes()
}

// taskPROutcomeActivatedAtMetaKey is the kandev_meta key under which the
// PR-outcome-attribution feature's one-time activation instant is recorded
// (AC-05). Any point-in-time extract over the five outcome columns must
// read this key to scope its window and distinguish "not yet activated"
// from "writer broke" (spec: Persistence guarantees).
const taskPROutcomeActivatedAtMetaKey = "github_task_pr_outcome_activated_at"

// activateTaskPROutcomeTracking stamps the activation instant exactly once
// per database (AC-05, AC-06). WriteMetaKeyIfAbsent's ON CONFLICT DO NOTHING
// makes this safe to call on every boot rather than gate it behind whether
// this boot's migration literally ran an ALTER TABLE: a fresh install
// receives the five columns inline via createTablesSQL, not an ALTER, but
// still needs the instant stamped on its very first boot. A write failure
// aborts startup — a database with the columns but no activation instant is
// unreportable and must not be shipped (spec: Failure modes).
func (s *Store) activateTaskPROutcomeTracking() error {
	if err := persistence.EnsureMetaTable(s.db); err != nil {
		return fmt.Errorf("ensure kandev_meta table: %w", err)
	}
	if _, err := persistence.WriteMetaKeyIfAbsent(
		s.db, taskPROutcomeActivatedAtMetaKey, time.Now().UTC().Format(time.RFC3339),
	); err != nil {
		return fmt.Errorf("activate task PR outcome tracking: %w", err)
	}
	return nil
}

func (s *Store) initSchemaFoundations() error {
	if err := s.resetUnpublishedGitHubAuthSchema(); err != nil {
		return err
	}
	if err := s.initCoreSchema(); err != nil {
		return err
	}
	if err := s.addWorkspaceOwnershipColumns(); err != nil {
		return err
	}
	if err := s.addReviewWatchTargetLogin(); err != nil {
		return err
	}
	return nil
}

func (s *Store) applyIdempotentSchemaColumns() error {
	// Idempotent migrations for existing databases.
	exec := func(name, statement string) error {
		if _, err := s.db.Exec(schemaSQLForDriver(statement, s.db.DriverName())); err != nil && !dbutil.IsDuplicateColumnError(err) {
			return fmt.Errorf("add %s: %w", name, err)
		}
		return nil
	}
	if err := exec("github_pr_watches.last_review_state", `ALTER TABLE github_pr_watches ADD COLUMN last_review_state TEXT DEFAULT ''`); err != nil {
		return err
	}
	if err := exec("github_task_prs.mergeable_state", `ALTER TABLE github_task_prs ADD COLUMN mergeable_state TEXT NOT NULL DEFAULT ''`); err != nil {
		return err
	}
	// Phase 4 (multi-repo): per-repo PR association on github_task_prs.
	if err := exec("github_task_prs.repository_id", `ALTER TABLE github_task_prs ADD COLUMN repository_id TEXT NOT NULL DEFAULT ''`); err != nil {
		return err
	}
	if err := exec("github_pr_watches.repository_id", `ALTER TABLE github_pr_watches ADD COLUMN repository_id TEXT NOT NULL DEFAULT ''`); err != nil {
		return err
	}
	// CI popover: aggregate counts + branch protection's required_approving_review_count
	// + unresolved review-threads, surfaced in the PR top-bar hover popover so the
	// frontend can render the counts row without a second round-trip.
	if err := exec("github_task_prs.required_reviews", `ALTER TABLE github_task_prs ADD COLUMN required_reviews INTEGER`); err != nil {
		return err
	}
	if err := exec("github_task_prs.unresolved_review_threads", `ALTER TABLE github_task_prs ADD COLUMN unresolved_review_threads INTEGER DEFAULT 0`); err != nil {
		return err
	}
	if err := exec("github_task_prs.checks_total", `ALTER TABLE github_task_prs ADD COLUMN checks_total INTEGER DEFAULT 0`); err != nil {
		return err
	}
	if err := exec("github_task_prs.checks_passing", `ALTER TABLE github_task_prs ADD COLUMN checks_passing INTEGER DEFAULT 0`); err != nil {
		return err
	}
	if err := exec("github_task_prs.detached_at", `ALTER TABLE github_task_prs ADD COLUMN detached_at DATETIME`); err != nil {
		return err
	}
	// Per-watch cleanup policy for review/issue watches: controls whether the
	// poller deletes auto-created tasks when the underlying PR/issue reaches
	// a terminal state. Values: 'auto' (default — preserve only when user
	// engaged), 'always' (delete on terminal state), 'never' (manual only).
	if err := exec("github_review_watches.cleanup_policy", `ALTER TABLE github_review_watches ADD COLUMN cleanup_policy TEXT NOT NULL DEFAULT 'auto'`); err != nil {
		return err
	}
	if err := exec("github_issue_watches.cleanup_policy", `ALTER TABLE github_issue_watches ADD COLUMN cleanup_policy TEXT NOT NULL DEFAULT 'auto'`); err != nil {
		return err
	}
	return nil
}

func (s *Store) initSchemaUpgrades() error {
	if err := s.addTaskGitCredentialsMode(); err != nil {
		return err
	}
	// Watcher self-heal columns: when the dispatch pipeline detects an
	// orphaned watcher (e.g. its agent profile has been soft-deleted), it
	// disables the row and stamps a human-readable cause + timestamp here
	// for the settings page to surface. Unlike the cleanup_policy column
	// above, the readers (IssueWatch.LastError / LastErrorAt) scan these
	// columns unconditionally — a driver-level ALTER failure here would
	// turn into a confusing scan panic on the next poll instead of a
	// clear boot error. Use the same fail-loud column-precheck idiom the
	// sibling jira/linear stores already use.
	if err := s.addWatchSelfHealColumns(); err != nil {
		return err
	}
	if err := s.addTaskCIRoundColumns(); err != nil {
		return err
	}
	if err := s.addTaskPRAgentAutomationColumns(); err != nil {
		return err
	}
	if err := s.addPRScopeMigrationColumn(); err != nil {
		return err
	}
	if err := s.addGitHubAuthFlowExpectationColumns(); err != nil {
		return err
	}
	if err := s.addAppRegistrationReferenceColumns(); err != nil {
		return err
	}
	if err := s.addTaskPROutcomeColumns(); err != nil {
		return err
	}
	if err := s.addTaskPRMergeQueueColumns(); err != nil {
		return err
	}
	if err := s.addTaskPRSourceColumn(); err != nil {
		return err
	}
	if err := s.addTaskPRWorkflowAttentionColumn(); err != nil {
		return err
	}
	return nil
}

var taskPRMergeQueueColumnDDL = []struct {
	name string
	ddl  string
}{
	{"head_sha", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_state", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_position", "INTEGER"},
	{"merge_queue_entry_id", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_entry_head_sha", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_estimated_time_to_merge_seconds", "INTEGER"},
	{"merge_queue_last_removal_id", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_last_removed_at", "DATETIME"},
	{"merge_queue_last_removal_reason", "TEXT NOT NULL DEFAULT ''"},
	{"merge_queue_last_removal_before_sha", "TEXT NOT NULL DEFAULT ''"},
}

func (s *Store) addTaskPRMergeQueueColumns() error {
	columns, err := s.tableColumns("github_task_prs")
	if err != nil {
		return fmt.Errorf("read github_task_prs columns: %w", err)
	}
	for _, column := range taskPRMergeQueueColumnDDL {
		if _, ok := columns[column.name]; ok {
			continue
		}
		stmt := schemaSQLForDriver(
			"ALTER TABLE github_task_prs ADD COLUMN "+column.name+" "+column.ddl,
			s.db.DriverName(),
		)
		if _, err := s.db.Exec(stmt); err != nil && !dbutil.IsDuplicateColumnError(err) {
			return fmt.Errorf("add github_task_prs.%s: %w", column.name, err)
		}
	}
	return nil
}

// addTaskPRSourceColumn adds github_task_prs.source, which records which
// write path created the association ("watch" or "url_link"; empty for rows
// written before this column existed). It joins taskPRColumns like the
// outcome-attribution columns above, so a driver-level ALTER failure here
// must also abort startup rather than surface as a scan error on the next
// read. Only dbutil.IsDuplicateColumnError is tolerated, per ADR 0027; no
// backfill runs against existing rows.
func (s *Store) addTaskPRSourceColumn() error {
	cols, err := s.tableColumns("github_task_prs")
	if err != nil {
		return fmt.Errorf("read github_task_prs columns: %w", err)
	}
	if _, ok := cols["source"]; ok {
		return nil
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`ALTER TABLE github_task_prs ADD COLUMN source TEXT NOT NULL DEFAULT ''`, s.db.DriverName(),
	)); err != nil &&
		!dbutil.IsDuplicateColumnError(err) {
		return fmt.Errorf("add github_task_prs.source: %w", err)
	}
	return nil
}

// addTaskPRWorkflowAttentionColumn adds the serialized, head-scoped Actions
// observation. Empty is the legacy/unobserved value and is not an assertion
// that no workflow needs attention.
func (s *Store) addTaskPRWorkflowAttentionColumn() error {
	cols, err := s.tableColumns("github_task_prs")
	if err != nil {
		return fmt.Errorf("read github_task_prs columns: %w", err)
	}
	if _, ok := cols["workflow_attention"]; ok {
		return nil
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`ALTER TABLE github_task_prs ADD COLUMN workflow_attention TEXT NOT NULL DEFAULT ''`, s.db.DriverName(),
	)); err != nil && !dbutil.IsDuplicateColumnError(err) {
		return fmt.Errorf("add github_task_prs.workflow_attention: %w", err)
	}
	return nil
}

// taskPROutcomeColumnDDL lists the five nullable PR-outcome-attribution
// columns and their type fragments. Every column is nullable with no
// DEFAULT (AC-01): NULL means "never observed" and must never be confused
// with a zero value or an empty string.
var taskPROutcomeColumnDDL = []struct {
	name string
	ddl  string
}{
	{"is_draft", "BOOLEAN"},
	{"changed_files", "INTEGER"},
	{"merged_by_login", "TEXT"},
	{"closed_by_login", "TEXT"},
	{"auto_merge_observed_at", "DATETIME"},
}

// addTaskPROutcomeColumns adds the five PR-outcome-attribution columns to
// github_task_prs. These columns join taskPRColumns, so every existing read
// scans them unconditionally; unlike applyIdempotentSchemaColumns, a
// driver-level ALTER failure here must abort startup rather than silently
// turn into a scan error on the next read (AC-03). Only
// dbutil.IsDuplicateColumnError is tolerated, per ADR 0027 — no local
// error-string classifier, and no UPDATE/backfill runs against
// github_task_prs anywhere in this path (AC-04).
func (s *Store) addTaskPROutcomeColumns() error {
	cols, err := s.tableColumns("github_task_prs")
	if err != nil {
		return fmt.Errorf("read github_task_prs columns: %w", err)
	}
	for _, col := range taskPROutcomeColumnDDL {
		if _, ok := cols[col.name]; ok {
			continue
		}
		stmt := schemaSQLForDriver(
			"ALTER TABLE github_task_prs ADD COLUMN "+col.name+" "+col.ddl,
			s.db.DriverName(),
		)
		if _, err := s.db.Exec(stmt); err != nil && !dbutil.IsDuplicateColumnError(err) {
			return fmt.Errorf("add github_task_prs.%s: %w", col.name, err)
		}
	}
	return nil
}

func (s *Store) addTaskGitCredentialsMode() error {
	columns, err := s.tableColumns("github_workspace_settings")
	if err != nil {
		return fmt.Errorf("read github_workspace_settings columns: %w", err)
	}
	if _, ok := columns["task_git_credentials_mode"]; ok {
		return nil
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`ALTER TABLE github_workspace_settings ADD COLUMN task_git_credentials_mode TEXT NOT NULL DEFAULT 'managed'`,
		s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("add github_workspace_settings.task_git_credentials_mode: %w", err)
	}
	return nil
}

func (s *Store) initSchemaData(legacyUpgrade bool) error {
	if err := s.backfillGitHubUserConnectionVersions(); err != nil {
		return err
	}
	if err := s.clearLifecyclePromptOverrides(); err != nil {
		return err
	}
	if err := s.migratePRTablesForMultiRepo(); err != nil {
		return fmt.Errorf("migrate PR tables for multi-repo: %w", err)
	}
	if err := s.backfillTaskPRsRepositoryID(); err != nil {
		return fmt.Errorf("backfill github_task_prs.repository_id: %w", err)
	}
	if err := s.backfillPRWatchesRepositoryID(); err != nil {
		return fmt.Errorf("backfill github_pr_watches.repository_id: %w", err)
	}
	if err := s.healTaskOwnedOrphans(); err != nil {
		return err
	}
	if err := s.migratePRWatchesToTaskOwnership(); err != nil {
		return fmt.Errorf("migrate PR watches to task ownership: %w", err)
	}
	if err := s.migrateTaskCIOptionsToPRScope(); err != nil {
		return fmt.Errorf("migrate task CI options to PR scope: %w", err)
	}
	if err := s.backfillGitHubWorkspaceOwnership(); err != nil {
		return err
	}
	if legacyUpgrade {
		if err := s.seedLegacyWorkspaceConnections(); err != nil {
			return err
		}
	}
	return nil
}

// migrateTaskCIOptionsToPRScope seeds github_task_pr_automation_options rows
// from each pre-upgrade github_task_ci_options row's legacy booleans, fanning
// each task's values out onto every github_task_prs row currently linked to
// it. Guarded by pr_scope_migrated_at, stamped in the same transaction as the
// fan-out insert: without the marker, replaying this on every boot would
// re-enable a switch a user has since turned off for one PR (R2), and a PR
// linked to the task after migration would incorrectly inherit the legacy
// value instead of starting all-off (AC17) via ON CONFLICT DO NOTHING alone.
func (s *Store) migrateTaskCIOptionsToPRScope() error {
	type legacyOptions struct {
		taskID                                                       string
		autoFix, autoMerge, promptReview, promptMerged, promptClosed bool
	}
	rows, err := s.db.Query(`
		SELECT task_id, auto_fix_enabled, auto_merge_enabled, prompt_on_review_requested,
			prompt_on_merged, prompt_on_closed
		FROM github_task_ci_options
		WHERE pr_scope_migrated_at IS NULL`)
	if err != nil {
		return fmt.Errorf("list unmigrated task CI options: %w", err)
	}
	var legacy []legacyOptions
	for rows.Next() {
		var row legacyOptions
		if err := rows.Scan(
			&row.taskID, &row.autoFix, &row.autoMerge, &row.promptReview, &row.promptMerged, &row.promptClosed,
		); err != nil {
			_ = rows.Close()
			return fmt.Errorf("scan unmigrated task CI options: %w", err)
		}
		legacy = append(legacy, row)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate unmigrated task CI options: %w", err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("close unmigrated task CI options rows: %w", err)
	}
	for _, row := range legacy {
		if err := s.fanOutTaskCIOptionsToPRScope(
			row.taskID, row.autoFix, row.autoMerge, row.promptReview, row.promptMerged, row.promptClosed,
		); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) fanOutTaskCIOptionsToPRScope(
	taskID string, autoFix, autoMerge, promptReview, promptMerged, promptClosed bool,
) error {
	tx, err := s.db.Beginx()
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	now := time.Now().UTC()
	if _, err := tx.Exec(tx.Rebind(`
		INSERT INTO github_task_pr_automation_options (
			task_id, repository_id, pr_number, auto_fix_enabled, auto_merge_enabled,
			prompt_on_review_requested, prompt_on_merged, prompt_on_closed, created_at, updated_at
		)
		SELECT task_id, repository_id, pr_number, ?, ?, ?, ?, ?, ?, ?
		FROM github_task_prs
		WHERE task_id = ? AND detached_at IS NULL
		ON CONFLICT(task_id, repository_id, pr_number) DO NOTHING`),
		autoFix, autoMerge, promptReview, promptMerged, promptClosed, now, now, taskID); err != nil {
		return fmt.Errorf("fan out task CI options for %s: %w", taskID, err)
	}
	if _, err := tx.Exec(
		tx.Rebind(`UPDATE github_task_ci_options SET pr_scope_migrated_at = ? WHERE task_id = ?`), now, taskID,
	); err != nil {
		return fmt.Errorf("stamp pr_scope_migrated_at for %s: %w", taskID, err)
	}
	return tx.Commit()
}

func (s *Store) clearLifecyclePromptOverrides() error {
	_, err := s.db.Exec(`
		UPDATE github_task_ci_options
		SET review_prompt_override = NULL,
			merged_prompt_override = NULL,
			closed_prompt_override = NULL
		WHERE review_prompt_override IS NOT NULL
			OR merged_prompt_override IS NOT NULL
			OR closed_prompt_override IS NOT NULL`)
	return err
}

func (s *Store) applyIdempotentSchemaIndexes() error {
	// pr_number is the 3rd column of UNIQUE(task_id, repository_id, pr_number),
	// so SQLite can't use that index for the PR-number task search. Add a
	// dedicated leading-key index so lookups by PR number stay index-backed.
	if _, err := s.db.Exec(schemaSQLForDriver(
		`CREATE INDEX IF NOT EXISTS idx_github_task_prs_pr_number ON github_task_prs (pr_number)`,
		s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("create idx_github_task_prs_pr_number: %w", err)
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`CREATE INDEX IF NOT EXISTS idx_github_task_ci_pr_state_task ON github_task_ci_pr_state (task_id)`,
		s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("create idx_github_task_ci_pr_state_task: %w", err)
	}
	// Canonical PR-watch identity (ADR 2026-08-31-task-owned-pr-watch-identity):
	// a searching watch is unique per (task, repository, branch); a discovered
	// watch is unique per (task, repository, pr_number). session_id never
	// participates. Partial indexes because SQLite can't express a
	// conditional UNIQUE inline in CREATE TABLE. Applied unconditionally on
	// every boot (fresh installs and post-migration upgrades alike) - once
	// migratePRWatchesToTaskOwnership has deduplicated any legacy rows, both
	// creations are no-ops that just confirm the invariant holds.
	if _, err := s.db.Exec(schemaSQLForDriver(
		`CREATE UNIQUE INDEX IF NOT EXISTS idx_github_pr_watches_searching
			ON github_pr_watches (task_id, repository_id, branch) WHERE pr_number = 0`,
		s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("create idx_github_pr_watches_searching: %w", err)
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`CREATE UNIQUE INDEX IF NOT EXISTS idx_github_pr_watches_discovered
			ON github_pr_watches (task_id, repository_id, pr_number) WHERE pr_number != 0`,
		s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("create idx_github_pr_watches_discovered: %w", err)
	}
	return nil
}

func (s *Store) resetUnpublishedGitHubAuthSchema() error {
	reset, err := s.unpublishedGitHubAuthSchemaNeedsReset()
	if err != nil || !reset {
		return err
	}
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	for _, object := range []string{
		"github_user_connections_registration_insert",
		"github_user_connections_registration_update",
		"github_workspace_connections_registration_update",
		"github_workspace_connections_registration_delete",
	} {
		if _, err := tx.Exec(`DROP TRIGGER IF EXISTS ` + object); err != nil {
			return err
		}
	}
	for _, table := range []string{
		"github_webhook_deliveries",
		"github_auth_flows",
		"github_user_connection_versions",
		"github_user_connections",
		"github_workspace_connections",
		"github_app_import_preparations",
		"github_app_registration_flows",
		"github_app_registrations",
		"github_app_registration_flow_head",
		"github_app_registration",
	} {
		if _, err := tx.Exec(`DROP TABLE IF EXISTS ` + table); err != nil {
			return err
		}
	}
	return tx.Commit()
}

func (s *Store) unpublishedGitHubAuthSchemaNeedsReset() (bool, error) {
	if dialect.IsPostgres(s.db.DriverName()) {
		return false, nil
	}
	for _, singleton := range []string{"github_app_registration", "github_app_registration_flow_head"} {
		if s.tableExists(singleton) {
			return true, nil
		}
	}
	required := map[string][]string{
		"github_app_registrations":     {"display_name TEXT NOT NULL", "UNIQUE (github_host, app_id)"},
		"github_workspace_connections": {"FOREIGN KEY (app_registration_id)", "source = 'github_app_installation'"},
		"github_user_connections":      {"app_registration_id TEXT NOT NULL", "FOREIGN KEY (app_registration_id)"},
		"github_auth_flows":            {"app_registration_id TEXT NOT NULL", "expected_workspace_app_registration_id TEXT"},
		"github_webhook_deliveries":    {"PRIMARY KEY (app_registration_id, delivery_id)"},
	}
	for table, fragments := range required {
		var schema string
		err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?`, table).Scan(&schema)
		if errors.Is(err, sql.ErrNoRows) {
			continue
		}
		if err != nil {
			return false, err
		}
		for _, fragment := range fragments {
			if !strings.Contains(schema, fragment) {
				return true, nil
			}
		}
	}
	return false, nil
}

func (s *Store) addAppRegistrationReferenceColumns() error {
	for _, migration := range []struct {
		table     string
		statement string
	}{
		{"github_workspace_connections", `ALTER TABLE github_workspace_connections ADD COLUMN app_registration_id TEXT`},
		{"github_user_connections", `ALTER TABLE github_user_connections ADD COLUMN app_registration_id TEXT`},
		{"github_auth_flows", `ALTER TABLE github_auth_flows ADD COLUMN app_registration_id TEXT`},
	} {
		columns, err := s.tableColumns(migration.table)
		if err != nil {
			return fmt.Errorf("read %s columns: %w", migration.table, err)
		}
		if _, exists := columns["app_registration_id"]; exists {
			continue
		}
		if _, err := s.db.Exec(schemaSQLForDriver(migration.statement, s.db.DriverName())); err != nil {
			return fmt.Errorf("add %s.app_registration_id: %w", migration.table, err)
		}
	}
	return nil
}

func (s *Store) initCoreSchema() error {
	if err := s.initAppRegistrationSchema(); err != nil {
		return err
	}
	_, err := s.db.Exec(schemaSQLForDriver(createTablesSQL, s.db.DriverName()))
	return err
}

func (s *Store) initAppRegistrationSchema() error {
	if _, err := s.db.Exec(schemaSQLForDriver(appRegistrationTablesSQL, s.db.DriverName())); err != nil {
		return fmt.Errorf("initialize GitHub App registration schema: %w", err)
	}
	return nil
}

func schemaSQLForDriver(schema, driver string) string {
	schema = dialect.MustRenderSchema(driver, schema)
	if dialect.IsPostgres(driver) {
		if strings.Contains(schema, "CREATE TRIGGER IF NOT EXISTS github_user_connections_registration_insert") {
			schema = withoutSQLiteGitHubAuthTriggers(schema)
			schema += postgresGitHubAuthTriggers()
		}
	}
	return schema
}

func withoutSQLiteGitHubAuthTriggers(schema string) string {
	start := strings.Index(schema, "\n\tCREATE TRIGGER IF NOT EXISTS github_user_connections_registration_insert")
	end := strings.Index(schema, "\n\tCREATE TABLE IF NOT EXISTS github_user_connection_versions")
	if start < 0 || end <= start {
		return schema
	}
	return schema[:start] + schema[end:]
}

func postgresGitHubAuthTriggers() string {
	return `
CREATE OR REPLACE FUNCTION github_validate_user_connection_registration()
RETURNS trigger
LANGUAGE plpgsql
AS $kandev$
BEGIN
	IF NOT EXISTS (
		SELECT 1 FROM github_workspace_connections workspace
		WHERE workspace.workspace_id = NEW.workspace_id
			AND workspace.source = 'github_app_installation'
			AND workspace.app_registration_id = NEW.app_registration_id
	) THEN
		RAISE EXCEPTION 'personal GitHub App registration must match workspace';
	END IF;
	RETURN NEW;
END;
$kandev$;

CREATE OR REPLACE FUNCTION github_validate_workspace_connection_registration()
RETURNS trigger
LANGUAGE plpgsql
AS $kandev$
BEGIN
	IF TG_OP = 'DELETE' THEN
		IF EXISTS (
			SELECT 1 FROM github_user_connections personal
			WHERE personal.workspace_id = OLD.workspace_id
		) THEN
			RAISE EXCEPTION 'workspace GitHub App connection still has personal connections';
		END IF;
		RETURN OLD;
	END IF;

	IF EXISTS (
		SELECT 1 FROM github_user_connections personal
		WHERE personal.workspace_id = OLD.workspace_id
			AND (
				NEW.source <> 'github_app_installation'
				OR NEW.app_registration_id IS NULL
				OR personal.app_registration_id <> NEW.app_registration_id
			)
	) THEN
		RAISE EXCEPTION 'workspace GitHub App registration must match personal connections';
	END IF;
	RETURN NEW;
END;
$kandev$;

DROP TRIGGER IF EXISTS github_user_connections_registration_insert ON github_user_connections;
CREATE TRIGGER github_user_connections_registration_insert
BEFORE INSERT ON github_user_connections
FOR EACH ROW EXECUTE FUNCTION github_validate_user_connection_registration();

DROP TRIGGER IF EXISTS github_user_connections_registration_update ON github_user_connections;
CREATE TRIGGER github_user_connections_registration_update
BEFORE UPDATE OF workspace_id, app_registration_id ON github_user_connections
FOR EACH ROW EXECUTE FUNCTION github_validate_user_connection_registration();

DROP TRIGGER IF EXISTS github_workspace_connections_registration_update ON github_workspace_connections;
CREATE TRIGGER github_workspace_connections_registration_update
BEFORE UPDATE OF source, app_registration_id ON github_workspace_connections
FOR EACH ROW EXECUTE FUNCTION github_validate_workspace_connection_registration();

DROP TRIGGER IF EXISTS github_workspace_connections_registration_delete ON github_workspace_connections;
CREATE TRIGGER github_workspace_connections_registration_delete
BEFORE DELETE ON github_workspace_connections
FOR EACH ROW EXECUTE FUNCTION github_validate_workspace_connection_registration();
`
}

func (s *Store) backfillGitHubUserConnectionVersions() error {
	// Package-local tests may initialize the GitHub store before the shared
	// workspace schema. A database with existing user connections necessarily
	// has the workspace table, so skipping this empty backfill is lossless.
	if !s.tableExists("workspaces") {
		return nil
	}
	if _, err := s.db.Exec(`
		INSERT INTO github_user_connection_versions (workspace_id, user_id, credential_generation, updated_at)
		SELECT workspace_id, user_id, credential_generation, updated_at FROM github_user_connections
		WHERE true
		ON CONFLICT(workspace_id, user_id) DO UPDATE SET
			credential_generation = CASE
				WHEN github_user_connection_versions.credential_generation > excluded.credential_generation
				THEN github_user_connection_versions.credential_generation
				ELSE excluded.credential_generation
			END,
			updated_at = excluded.updated_at`); err != nil {
		return fmt.Errorf("backfill GitHub user connection versions: %w", err)
	}
	return nil
}

func (s *Store) addGitHubAuthFlowExpectationColumns() error {
	columns, err := s.tableColumns("github_auth_flows")
	if err != nil {
		return fmt.Errorf("read github_auth_flows columns: %w", err)
	}
	for name, statement := range map[string]string{
		"expected_workspace_source":              `ALTER TABLE github_auth_flows ADD COLUMN expected_workspace_source TEXT NOT NULL DEFAULT ''`,
		"expected_workspace_generation":          `ALTER TABLE github_auth_flows ADD COLUMN expected_workspace_generation BIGINT NOT NULL DEFAULT 0`,
		"expected_installation_id":               `ALTER TABLE github_auth_flows ADD COLUMN expected_installation_id BIGINT`,
		"expected_workspace_app_registration_id": `ALTER TABLE github_auth_flows ADD COLUMN expected_workspace_app_registration_id TEXT`,
		"expected_personal_generation":           `ALTER TABLE github_auth_flows ADD COLUMN expected_personal_generation BIGINT NOT NULL DEFAULT 0`,
	} {
		if _, ok := columns[name]; ok {
			continue
		}
		if _, err := s.db.Exec(schemaSQLForDriver(statement, s.db.DriverName())); err != nil {
			return fmt.Errorf("add github_auth_flows.%s: %w", name, err)
		}
	}
	return nil
}

func (s *Store) ensureWorkspaceOwnershipIndexes() error {
	for name, statement := range map[string]string{
		"github_task_prs":   `CREATE INDEX IF NOT EXISTS idx_github_task_prs_workspace ON github_task_prs (workspace_id)`,
		"github_pr_watches": `CREATE INDEX IF NOT EXISTS idx_github_pr_watches_workspace ON github_pr_watches (workspace_id)`,
	} {
		if _, err := s.db.Exec(statement); err != nil {
			return fmt.Errorf("index %s workspace ownership: %w", name, err)
		}
	}
	return nil
}

func (s *Store) addWorkspaceOwnershipColumns() error {
	for _, migration := range []struct {
		table string
		stmt  string
	}{
		{"github_pr_watches", `ALTER TABLE github_pr_watches ADD COLUMN workspace_id TEXT NOT NULL DEFAULT ''`},
		{"github_task_prs", `ALTER TABLE github_task_prs ADD COLUMN workspace_id TEXT NOT NULL DEFAULT ''`},
	} {
		if _, err := s.db.Exec(schemaSQLForDriver(migration.stmt, s.db.DriverName())); err != nil && !dbutil.IsDuplicateColumnError(err) {
			return fmt.Errorf("add %s.workspace_id: %w", migration.table, err)
		}
	}
	return nil
}

func (s *Store) addReviewWatchTargetLogin() error {
	_, err := s.db.Exec(schemaSQLForDriver(
		`ALTER TABLE github_review_watches ADD COLUMN target_login TEXT NOT NULL DEFAULT ''`,
		s.db.DriverName(),
	))
	if err != nil && !dbutil.IsDuplicateColumnError(err) {
		return fmt.Errorf("add github_review_watches.target_login: %w", err)
	}
	return nil
}

func (s *Store) backfillGitHubWorkspaceOwnership() error {
	if !s.tableExists("tasks") {
		return nil
	}
	taskColumns, err := s.tableColumns("tasks")
	if err != nil {
		return fmt.Errorf("read tasks columns for github ownership backfill: %w", err)
	}
	if _, ok := taskColumns["workspace_id"]; !ok {
		return nil
	}
	for _, table := range []string{"github_pr_watches", "github_task_prs"} {
		query := `UPDATE ` + table + `
			SET workspace_id = (
				SELECT tasks.workspace_id FROM tasks WHERE tasks.id = ` + table + `.task_id
			)
			WHERE workspace_id = ''
			  AND EXISTS (SELECT 1 FROM tasks WHERE tasks.id = ` + table + `.task_id)
			  AND (SELECT tasks.workspace_id FROM tasks WHERE tasks.id = ` + table + `.task_id) <> ''`
		if _, err := s.db.Exec(query); err != nil {
			return fmt.Errorf("backfill %s.workspace_id: %w", table, err)
		}
	}
	return nil
}

func (s *Store) seedLegacyWorkspaceConnections() error {
	if !s.tableExists("workspaces") {
		return nil
	}
	_, err := s.db.Exec(`
		INSERT INTO github_workspace_connections (
			workspace_id, source, github_host, status, credential_generation, created_at, updated_at
		)
		SELECT id, 'legacy_shared', 'github.com', 'active', 1, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
		FROM workspaces
		WHERE TRUE
		ON CONFLICT(workspace_id) DO NOTHING`)
	if err != nil {
		return fmt.Errorf("seed legacy github workspace connections: %w", err)
	}
	return nil
}

func (s *Store) addTaskPRAgentAutomationColumns() error {
	migrations := map[string][]struct {
		name string
		sql  string
	}{
		"github_task_ci_options": {
			{"prompt_on_review_requested", "ALTER TABLE github_task_ci_options ADD COLUMN prompt_on_review_requested BOOLEAN NOT NULL DEFAULT 0"},
			{"prompt_on_merged", "ALTER TABLE github_task_ci_options ADD COLUMN prompt_on_merged BOOLEAN NOT NULL DEFAULT 0"},
			{"prompt_on_closed", "ALTER TABLE github_task_ci_options ADD COLUMN prompt_on_closed BOOLEAN NOT NULL DEFAULT 0"},
			{"review_reviewer_login", "ALTER TABLE github_task_ci_options ADD COLUMN review_reviewer_login TEXT NOT NULL DEFAULT ''"},
			{"review_prompt_override", "ALTER TABLE github_task_ci_options ADD COLUMN review_prompt_override TEXT"},
			{"merged_prompt_override", "ALTER TABLE github_task_ci_options ADD COLUMN merged_prompt_override TEXT"},
			{"closed_prompt_override", "ALTER TABLE github_task_ci_options ADD COLUMN closed_prompt_override TEXT"},
		},
		"github_task_ci_pr_state": {
			{"last_merge_result", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_merge_result TEXT NOT NULL DEFAULT ''"},
			{"merge_retry_pending", "ALTER TABLE github_task_ci_pr_state ADD COLUMN merge_retry_pending BOOLEAN NOT NULL DEFAULT 0"},
			{"last_error_kind", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_error_kind TEXT NOT NULL DEFAULT ''"},
			{"last_queue_attempt_head_sha", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_queue_attempt_head_sha TEXT NOT NULL DEFAULT ''"},
			{"last_queue_fix_event_id", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_queue_fix_event_id TEXT NOT NULL DEFAULT ''"},
			{"last_queue_removal_cause", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_queue_removal_cause TEXT NOT NULL DEFAULT ''"},
			{"review_request_initialized", "ALTER TABLE github_task_ci_pr_state ADD COLUMN review_request_initialized BOOLEAN NOT NULL DEFAULT 0"},
			{"last_review_requested", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_review_requested BOOLEAN NOT NULL DEFAULT 0"},
			{"last_observed_pr_state", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_observed_pr_state TEXT NOT NULL DEFAULT ''"},
			{"last_lifecycle_event", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_lifecycle_event TEXT NOT NULL DEFAULT ''"},
			{"last_lifecycle_prompt_at", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_lifecycle_prompt_at DATETIME"},
			{"last_lifecycle_session_id", "ALTER TABLE github_task_ci_pr_state ADD COLUMN last_lifecycle_session_id TEXT"},
			{"auto_fix_attempt_state", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_state TEXT NOT NULL DEFAULT 'acknowledged'"},
			{"auto_fix_attempt_queue_entry_id", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_queue_entry_id TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_session_id", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_session_id TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_turn_id", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_turn_id TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_signature", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_signature TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_provider_generation", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_provider_generation TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_outcome", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_outcome TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_summary", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_summary TEXT NOT NULL DEFAULT ''"},
			{"auto_fix_attempt_started_at", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_started_at DATETIME"},
			{"auto_fix_attempt_outcome_at", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_outcome_at DATETIME"},
			{"auto_fix_attempt_progress_deadline", "ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_attempt_progress_deadline DATETIME"},
		},
	}
	for table, fields := range migrations {
		columns, err := s.tableColumns(table)
		if err != nil {
			return fmt.Errorf("read %s columns: %w", table, err)
		}
		for _, field := range fields {
			if _, exists := columns[field.name]; exists {
				continue
			}
			if _, err := s.db.Exec(schemaSQLForDriver(field.sql, s.db.DriverName())); err != nil {
				return fmt.Errorf("add %s.%s: %w", table, field.name, err)
			}
		}
	}
	return s.backfillTaskCIAutoMergeErrorKinds()
}

func (s *Store) backfillTaskCIAutoMergeErrorKinds() error {
	_, err := s.db.Exec(s.db.Rebind(`
		UPDATE github_task_ci_pr_state
		SET last_error_kind = ?
		WHERE last_error_kind = ''
		  AND (last_error LIKE 'merge PR:%'
		    OR last_error LIKE 'PR status is not freshly synced for auto-merge%')`),
		TaskCIErrorKindAutoMerge)
	if err != nil {
		return fmt.Errorf("backfill CI auto-merge error kinds: %w", err)
	}
	return nil
}

// backfillTaskPRsRepositoryID heals github_task_prs rows that pre-date the
// per-repo schema (i.e. have repository_id = ”). Two passes:
//
//  1. Dedup: for each (task_id, owner, repo, pr_number) tuple, if a legacy
//     row (repository_id = ”) AND a newer per-repo row coexist, drop the
//     legacy one. This happens when an old single-repo install upgraded and
//     a subsequent sync inserted a new row under the resolved repository_id
//     instead of updating the legacy row, leaving two entries for one PR
//     (and so two badges on the kanban card).
//
//  2. Backfill: for any remaining empty-repo rows, look up the task's
//     primary repository (from `task_repositories` ordered by position) and
//     stamp its id. Skipped silently when the task has zero per-repo rows
//     (e.g. quick-chat tasks) — the row keeps its empty repository_id.
//
// Idempotent: re-running on a healed db is a no-op.
func (s *Store) backfillTaskPRsRepositoryID() error {
	if _, err := s.db.Exec(`
		DELETE FROM github_task_prs
		WHERE repository_id = ''
		  AND EXISTS (
		    SELECT 1 FROM github_task_prs other
		    WHERE other.task_id = github_task_prs.task_id
		      AND other.owner   = github_task_prs.owner
		      AND other.repo    = github_task_prs.repo
		      AND other.pr_number = github_task_prs.pr_number
		      AND other.repository_id != ''
		  )
	`); err != nil {
		return fmt.Errorf("dedup legacy task PR rows: %w", err)
	}
	// `task_repositories` lives in the task package's schema, not ours.
	// Skip the backfill when it doesn't exist (e.g. github-store unit tests
	// that init only this package's schema). The dedup pass above is the
	// load-bearing fix for the "two PRs on a single-repo task" symptom; the
	// backfill is a courtesy that converts orphan legacy rows in real
	// deployments.
	if !s.tableExists("task_repositories") {
		return nil
	}
	_, err := s.db.Exec(`
		UPDATE github_task_prs
		SET repository_id = (
		  SELECT tr.repository_id
		  FROM task_repositories tr
		  WHERE tr.task_id = github_task_prs.task_id
		  ORDER BY tr.position
		  LIMIT 1
		)
		WHERE repository_id = ''
		  AND EXISTS (
		    SELECT 1 FROM task_repositories tr
		    WHERE tr.task_id = github_task_prs.task_id
		  )
	`)
	if err != nil {
		return fmt.Errorf("backfill task PR repository_id: %w", err)
	}
	return nil
}

// backfillPRWatchesRepositoryID heals github_pr_watches rows that pre-date
// the per-repo schema (repository_id = ”). Same two-pass shape as
// backfillTaskPRsRepositoryID — without this the orchestrator's reconciler
// (which keys its existence-check by (session_id, repository_id)) would see
// the legacy `(sess, ”)` row as foreign and insert a SECOND watch row
// under the resolved repository_id. Two watches → two AssociatePRWithTask
// calls when the user opens a PR → two github_task_prs rows for the same
// PR, which is the "PR appears twice on a single-repo task" symptom we hit
// after the multi-repo rollout.
//
//  1. Dedup: drop legacy `”` rows whose session already has a non-empty
//     row — the reconciler-inserted row supersedes the legacy one.
//
//  2. Backfill: stamp the remaining `”` rows with the task's primary
//     repository_id from `task_repositories`. Skipped silently when the
//     table is absent (unit tests that init only this package's schema).
//
// Idempotent: re-running on a healed db is a no-op.
func (s *Store) backfillPRWatchesRepositoryID() error {
	if _, err := s.db.Exec(`
		DELETE FROM github_pr_watches
		WHERE repository_id = ''
		  AND EXISTS (
		    SELECT 1 FROM github_pr_watches other
		    WHERE other.session_id = github_pr_watches.session_id
		      AND other.repository_id != ''
		  )
	`); err != nil {
		return fmt.Errorf("dedup legacy PR watch rows: %w", err)
	}
	if !s.tableExists("task_repositories") {
		return nil
	}
	_, err := s.db.Exec(`
		UPDATE github_pr_watches
		SET repository_id = (
		  SELECT tr.repository_id
		  FROM task_repositories tr
		  WHERE tr.task_id = github_pr_watches.task_id
		  ORDER BY tr.position
		  LIMIT 1
		)
		WHERE repository_id = ''
		  AND EXISTS (
		    SELECT 1 FROM task_repositories tr
		    WHERE tr.task_id = github_pr_watches.task_id
		  )
	`)
	if err != nil {
		return fmt.Errorf("backfill PR watch repository_id: %w", err)
	}
	return nil
}

// addWatchSelfHealColumns adds last_error / last_error_at to the issue and
// review watch tables using a column-precheck (mirroring the jira and linear
// stores). Unlike the cleanup_policy ALTER above, the readers
// (IssueWatch.LastError / LastErrorAt) scan these columns unconditionally,
// so a driver-level failure must bubble up at boot rather than turn into
// a scan panic on the next poll.
func (s *Store) addWatchSelfHealColumns() error {
	for _, table := range []string{"github_review_watches", "github_issue_watches"} {
		cols, err := s.tableColumns(table)
		if err != nil {
			return fmt.Errorf("read %s columns: %w", table, err)
		}
		if _, ok := cols["last_error"]; !ok {
			if _, err := s.db.Exec("ALTER TABLE " + table + " ADD COLUMN last_error TEXT NOT NULL DEFAULT ''"); err != nil {
				return fmt.Errorf("add %s.last_error: %w", table, err)
			}
		}
		if _, ok := cols["last_error_at"]; !ok {
			if _, err := s.db.Exec(schemaSQLForDriver(
				"ALTER TABLE "+table+" ADD COLUMN last_error_at DATETIME", s.db.DriverName(),
			)); err != nil {
				return fmt.Errorf("add %s.last_error_at: %w", table, err)
			}
		}
	}
	return nil
}

func (s *Store) addTaskCIRoundColumns() error {
	cols, err := s.tableColumns("github_task_ci_pr_state")
	if err != nil {
		return fmt.Errorf("read github_task_ci_pr_state columns: %w", err)
	}
	if _, ok := cols["auto_fix_round_count"]; !ok {
		if _, err := s.db.Exec("ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_round_count INTEGER NOT NULL DEFAULT 0"); err != nil {
			return fmt.Errorf("add github_task_ci_pr_state.auto_fix_round_count: %w", err)
		}
	}
	if _, ok := cols["auto_fix_exhausted_at"]; !ok {
		if _, err := s.db.Exec(schemaSQLForDriver(
			"ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_exhausted_at DATETIME", s.db.DriverName(),
		)); err != nil {
			return fmt.Errorf("add github_task_ci_pr_state.auto_fix_exhausted_at: %w", err)
		}
	}
	return nil
}

// addPRScopeMigrationColumn adds the marker column that guards the one-time
// fan-out of legacy task-level automation switches onto per-PR rows (see
// migrateTaskCIOptionsToPRScope). Column-precheck idiom, mirroring
// addTaskCIRoundColumns.
func (s *Store) addPRScopeMigrationColumn() error {
	cols, err := s.tableColumns("github_task_ci_options")
	if err != nil {
		return fmt.Errorf("read github_task_ci_options columns: %w", err)
	}
	if _, ok := cols["pr_scope_migrated_at"]; ok {
		return nil
	}
	if _, err := s.db.Exec(schemaSQLForDriver(
		`ALTER TABLE github_task_ci_options ADD COLUMN pr_scope_migrated_at DATETIME`, s.db.DriverName(),
	)); err != nil {
		return fmt.Errorf("add github_task_ci_options.pr_scope_migrated_at: %w", err)
	}
	return nil
}

// tableColumns returns the set of column names declared on `table`. Cheap
// SQLite PRAGMA lookup; used by addWatchSelfHealColumns to skip ALTERs on a
// fresh install whose createTablesSQL already includes the columns. Mirrors
// the helper in jira/store.go.
func (s *Store) tableColumns(table string) (map[string]struct{}, error) {
	columns, err := dbutil.TableColumns(s.db, table)
	if err != nil {
		return nil, err
	}
	result := make(map[string]struct{}, len(columns))
	for column := range columns {
		result[column] = struct{}{}
	}
	return result, nil
}

// tableExists returns true when the named table is present in the active schema.
// Used by the multi-repo backfill to skip cross-package healing in unit
// tests that don't bring up the task schema.
func (s *Store) tableExists(name string) bool {
	exists, err := dbutil.TableExists(s.db, name)
	return err == nil && exists
}

// migratePRTablesForMultiRepo rebuilds `github_pr_watches` and
// `github_task_prs` to drop the legacy single-repo unique constraints
// (`UNIQUE(session_id)` and `UNIQUE(task_id, pr_number)`) and replace them
// with the multi-repo / multi-branch variants. SQLite can't ALTER TABLE
// DROP CONSTRAINT, so each table is rebuilt via the recommended
// copy-and-rename pattern. The migration is idempotent: it inspects
// `sqlite_master.sql` for the legacy constraint string and only runs the
// rebuild when found. The watch rebuild fires twice — once for the original
// single-repo shape, once for the interim multi-repo shape — so DBs caught
// in either state upgrade cleanly to the multi-branch shape.
func (s *Store) migratePRTablesForMultiRepo() error {
	if dialect.IsPostgres(s.db.DriverName()) {
		return nil
	}
	for _, trigger := range []string{
		"session_id TEXT NOT NULL UNIQUE",
		"UNIQUE(session_id, repository_id)\n",
	} {
		if err := s.rebuildIfHasLegacyConstraint(
			"github_pr_watches",
			trigger,
			`CREATE TABLE github_pr_watches_new (
				id TEXT PRIMARY KEY,
				workspace_id TEXT NOT NULL DEFAULT '',
				session_id TEXT NOT NULL,
				task_id TEXT NOT NULL,
				repository_id TEXT NOT NULL DEFAULT '',
				owner TEXT NOT NULL,
				repo TEXT NOT NULL,
				pr_number INTEGER NOT NULL,
				branch TEXT NOT NULL,
				last_checked_at DATETIME,
				last_comment_at DATETIME,
				last_check_status TEXT DEFAULT '',
				last_review_state TEXT DEFAULT '',
				created_at DATETIME NOT NULL,
				updated_at DATETIME NOT NULL,
				UNIQUE(session_id, repository_id, branch)
			)`,
			`INSERT INTO github_pr_watches_new (
				id, workspace_id, session_id, task_id, repository_id, owner, repo, pr_number, branch,
				last_checked_at, last_comment_at, last_check_status, last_review_state,
				created_at, updated_at
			) SELECT
				id, COALESCE(workspace_id, ''), session_id, task_id, COALESCE(repository_id, ''), owner, repo, pr_number, branch,
				last_checked_at, last_comment_at, last_check_status, last_review_state,
				created_at, updated_at
			FROM github_pr_watches`,
		); err != nil {
			return err
		}
	}
	return s.rebuildIfHasLegacyConstraint(
		"github_task_prs",
		"UNIQUE(task_id, pr_number)",
		`CREATE TABLE github_task_prs_new (
			id TEXT PRIMARY KEY,
			workspace_id TEXT NOT NULL DEFAULT '',
			task_id TEXT NOT NULL,
			repository_id TEXT NOT NULL DEFAULT '',
			owner TEXT NOT NULL,
			repo TEXT NOT NULL,
			pr_number INTEGER NOT NULL,
			pr_url TEXT NOT NULL,
			pr_title TEXT NOT NULL,
			head_branch TEXT NOT NULL,
			base_branch TEXT NOT NULL,
			head_sha TEXT NOT NULL DEFAULT '',
			author_login TEXT NOT NULL,
			state TEXT NOT NULL DEFAULT 'open',
			review_state TEXT NOT NULL DEFAULT '',
			checks_state TEXT NOT NULL DEFAULT '',
			mergeable_state TEXT NOT NULL DEFAULT '',
			merge_queue_state TEXT NOT NULL DEFAULT '',
			merge_queue_position INTEGER,
			merge_queue_entry_id TEXT NOT NULL DEFAULT '',
			merge_queue_entry_head_sha TEXT NOT NULL DEFAULT '',
			merge_queue_estimated_time_to_merge_seconds INTEGER,
			merge_queue_last_removal_id TEXT NOT NULL DEFAULT '',
			merge_queue_last_removed_at DATETIME,
			merge_queue_last_removal_reason TEXT NOT NULL DEFAULT '',
			merge_queue_last_removal_before_sha TEXT NOT NULL DEFAULT '',
			review_count INTEGER DEFAULT 0,
			pending_review_count INTEGER DEFAULT 0,
			required_reviews INTEGER,
			comment_count INTEGER DEFAULT 0,
			unresolved_review_threads INTEGER DEFAULT 0,
			checks_total INTEGER DEFAULT 0,
			checks_passing INTEGER DEFAULT 0,
			additions INTEGER DEFAULT 0,
			deletions INTEGER DEFAULT 0,
			created_at DATETIME NOT NULL,
			merged_at DATETIME,
			closed_at DATETIME,
			last_synced_at DATETIME,
			detached_at DATETIME,
			updated_at DATETIME NOT NULL,
			is_draft BOOLEAN,
			changed_files INTEGER,
			merged_by_login TEXT,
			closed_by_login TEXT,
			auto_merge_observed_at DATETIME,
			source TEXT NOT NULL DEFAULT '',
			workflow_attention TEXT NOT NULL DEFAULT '',
			UNIQUE(task_id, repository_id, pr_number)
		)`,
		// The five outcome-attribution columns and source are selected
		// directly, not via COALESCE: addTaskPROutcomeColumns and
		// addTaskPRSourceColumn both run earlier in initSchemaUpgrades,
		// before this data-migration step, so the old table being rebuilt
		// here already has them (fail-loud — startup would have aborted
		// otherwise).
		`INSERT INTO github_task_prs_new (
			id, workspace_id, task_id, repository_id, owner, repo, pr_number, pr_url, pr_title,
			head_branch, base_branch, head_sha, author_login, state, review_state, checks_state,
			mergeable_state, merge_queue_state, merge_queue_position, merge_queue_entry_id, merge_queue_entry_head_sha,
			merge_queue_estimated_time_to_merge_seconds, merge_queue_last_removal_id, merge_queue_last_removed_at,
			merge_queue_last_removal_reason, merge_queue_last_removal_before_sha,
			review_count, pending_review_count, comment_count,
			additions, deletions, created_at, merged_at, closed_at, last_synced_at, detached_at, updated_at,
			is_draft, changed_files, merged_by_login, closed_by_login, auto_merge_observed_at, source, workflow_attention
		) SELECT
			id, COALESCE(workspace_id, ''), task_id, COALESCE(repository_id, ''), owner, repo, pr_number, pr_url, pr_title,
			head_branch, base_branch, COALESCE(head_sha, ''), author_login, state, review_state, checks_state,
			mergeable_state, merge_queue_state, merge_queue_position, COALESCE(merge_queue_entry_id, ''), COALESCE(merge_queue_entry_head_sha, ''),
			merge_queue_estimated_time_to_merge_seconds, COALESCE(merge_queue_last_removal_id, ''), merge_queue_last_removed_at,
			COALESCE(merge_queue_last_removal_reason, ''), COALESCE(merge_queue_last_removal_before_sha, ''),
			review_count, pending_review_count, comment_count,
			additions, deletions, created_at, merged_at, closed_at, last_synced_at, detached_at, updated_at,
			is_draft, changed_files, merged_by_login, closed_by_login, auto_merge_observed_at, source, workflow_attention
		FROM github_task_prs`,
	)
}

// rebuildIfHasLegacyConstraint checks the table's stored CREATE statement in
// `sqlite_master` for the literal `legacyConstraint` substring; if present,
// runs the table rebuild (create new, copy data, drop old, rename) inside a
// transaction. No-op when the legacy substring is absent — fresh installs
// already use the new schema and previously-migrated databases skip too.
func (s *Store) rebuildIfHasLegacyConstraint(table, legacyConstraint, createNew, copyData string) error {
	if dialect.IsPostgres(s.db.DriverName()) {
		return nil
	}
	var existingSQL string
	row := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type='table' AND name=?`, table)
	if err := row.Scan(&existingSQL); err != nil {
		// Table missing entirely shouldn't happen after createTablesSQL ran;
		// treat as no-op to keep the migration robust under unexpected drift.
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		return err
	}
	if !strings.Contains(existingSQL, legacyConstraint) {
		return nil
	}
	tx, err := s.db.Begin()
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	for _, stmt := range []string{
		createNew,
		copyData,
		"DROP TABLE " + table,
		"ALTER TABLE " + table + "_new RENAME TO " + table,
	} {
		if _, err := tx.Exec(stmt); err != nil {
			return fmt.Errorf("rebuild %s: %w", table, err)
		}
	}
	return tx.Commit()
}

// PRWatchMigrationStats reports the outcome of migratePRWatchesToTaskOwnership.
// Populated only when the legacy session-scoped schema was detected; nil on a
// fresh install or a database that has already been migrated. Callers (tests,
// future health/metrics wiring) read this via Store.PRWatchMigrationStats
// instead of relying on log scraping.
type PRWatchMigrationStats struct {
	RowsBefore        int
	RowsAfter         int
	OrphansRemoved    int
	DuplicatesRemoved int
}

// PRWatchMigrationStats returns the aggregate counts from the most recent
// canonical PR-watch migration performed during store initialization, or nil
// when no legacy schema was found (fresh install / already migrated).
func (s *Store) PRWatchMigrationStats() *PRWatchMigrationStats {
	return s.prWatchMigration
}

// legacyPRWatchesUniqueConstraint is the pre-ADR-2026-08-31 constraint that
// scoped watch identity to session_id instead of task_id. Its presence in
// sqlite_master.sql is the idempotency trigger for
// migratePRWatchesToTaskOwnership: a fresh install's createTablesSQL no
// longer includes it, and a database that has already been migrated has had
// it rebuilt away, so both cases skip the migration entirely.
const legacyPRWatchesUniqueConstraint = "UNIQUE(session_id, repository_id, branch)"

// migratePRWatchesToTaskOwnership makes github_pr_watches task-owned per
// ADR-2026-08-31-task-owned-pr-watch-identity: a searching watch (pr_number =
// 0) becomes canonical by (task_id, repository_id, branch) and a discovered
// watch (pr_number != 0) becomes canonical by (task_id, repository_id,
// pr_number), instead of the legacy (session_id, repository_id, branch).
//
// Idempotent and transactional: it only runs when the legacy UNIQUE
// constraint is still present (checked via sqlite_master.sql), and either
// fully commits or leaves every row untouched. It must run after this
// process's pre-migration SnapshotSQLite backup (persistence.Provide runs
// that backup before any repository store, including this one, opens its
// schema - see internal/persistence/provider.go) so a failure here can
// always be rolled back from the snapshot.
//
// Steps, in order: (1) clear session_id provenance for sessions that no
// longer exist, (2) remove watches for tasks that no longer exist or whose
// repository has been detached from the task (never polled - see
// ListActivePRWatches - so safe to drop), (3) deduplicate remaining rows by
// (task_id, repository_id, branch), preferring any discovered row and
// merging the newest check/comment/review watermarks into the survivor, (4)
// a second pass deduplicates by (task_id, repository_id, pr_number) in case
// two different branches independently resolved to the same PR, and (5) the
// table is rebuilt without the legacy constraint so the caller can install
// the new partial unique indexes (applyIdempotentSchemaIndexes) without a
// uniqueness violation.
func (s *Store) migratePRWatchesToTaskOwnership() error {
	if dialect.IsPostgres(s.db.DriverName()) {
		return s.migratePRWatchesToTaskOwnershipRows(false)
	}
	hasLegacy, err := s.tableSQLContains("github_pr_watches", legacyPRWatchesUniqueConstraint)
	if err != nil {
		return fmt.Errorf("inspect github_pr_watches schema: %w", err)
	}
	if !hasLegacy {
		return nil
	}
	return s.migratePRWatchesToTaskOwnershipRows(true)
}

func (s *Store) migratePRWatchesToTaskOwnershipRows(rebuild bool) error {
	// Resolve table existence up front, on the shared pool, before opening
	// the transaction below: the SQLite writer pool is capped to a single
	// connection, so calling s.tableExists (which borrows a pool connection
	// via s.db) once a transaction already holds that one connection
	// deadlocks forever waiting for a connection that will never free up.
	hasSessions := s.tableExists("sessions")
	hasTasks := s.tableExists("tasks")
	hasTaskRepositories := s.tableExists("task_repositories")

	tx, err := s.db.Beginx()
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	stats := &PRWatchMigrationStats{}
	if err := tx.Get(&stats.RowsBefore, `SELECT COUNT(*) FROM github_pr_watches`); err != nil {
		return fmt.Errorf("count PR watches before migration: %w", err)
	}

	if err := clearStalePRWatchSessionProvenance(tx, hasSessions); err != nil {
		return err
	}

	orphansRemoved, err := removeOrphanedPRWatches(tx, hasTasks, hasTaskRepositories)
	if err != nil {
		return err
	}
	stats.OrphansRemoved = int(orphansRemoved)

	var all []*PRWatch
	if err := tx.Select(&all, `SELECT * FROM github_pr_watches`); err != nil {
		return fmt.Errorf("load PR watches for dedup: %w", err)
	}
	toDelete, survivors := dedupPRWatchesForTaskOwnership(all)

	if err := applyPRWatchDedup(tx, toDelete, survivors); err != nil {
		return err
	}
	stats.DuplicatesRemoved = len(toDelete)

	if rebuild {
		if err := rebuildPRWatchesTableForTaskOwnership(tx); err != nil {
			return err
		}
	}

	noop, err := prWatchMigrationNoop(tx, rebuild, stats)
	if err != nil {
		return err
	}
	if noop {
		return nil
	}

	if err := tx.Get(&stats.RowsAfter, `SELECT COUNT(*) FROM github_pr_watches`); err != nil {
		return fmt.Errorf("count PR watches after migration: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit PR watch task-ownership migration: %w", err)
	}
	s.prWatchMigration = stats
	return nil
}

func prWatchMigrationNoop(tx *sqlx.Tx, rebuild bool, stats *PRWatchMigrationStats) (bool, error) {
	if rebuild || stats.DuplicatesRemoved != 0 || stats.OrphansRemoved != 0 {
		return false, nil
	}
	if err := tx.Get(&stats.RowsAfter, `SELECT COUNT(*) FROM github_pr_watches`); err != nil {
		return false, fmt.Errorf("count PR watches after migration: %w", err)
	}
	if stats.RowsBefore != stats.RowsAfter {
		return false, nil
	}
	if err := tx.Commit(); err != nil {
		return false, fmt.Errorf("commit PR watch task-ownership migration: %w", err)
	}
	return true, nil
}

// clearStalePRWatchSessionProvenance clears session_id on rows whose session
// no longer exists. session_id is provenance-only (ADR
// 2026-08-31-task-owned-pr-watch-identity) so this never removes a row, it
// only stops an already-gone session from being reported as the owner.
func clearStalePRWatchSessionProvenance(tx *sqlx.Tx, hasSessions bool) error {
	if !hasSessions {
		return nil
	}
	if _, err := tx.Exec(`
		UPDATE github_pr_watches SET session_id = ''
		WHERE session_id <> '' AND NOT EXISTS (
			SELECT 1 FROM sessions WHERE sessions.id = github_pr_watches.session_id
		)`); err != nil {
		return fmt.Errorf("clear stale PR watch session provenance: %w", err)
	}
	return nil
}

// removeOrphanedPRWatches deletes rows whose task no longer exists and rows
// whose repository has been detached from their task, returning the total
// number of rows removed. Either table may be absent in unit tests that
// don't bring up the task package's schema, in which case that half of the
// sweep is skipped.
func removeOrphanedPRWatches(tx *sqlx.Tx, hasTasks, hasTaskRepositories bool) (int64, error) {
	var removed int64
	if hasTasks {
		res, err := tx.Exec(`
			DELETE FROM github_pr_watches
			WHERE NOT EXISTS (SELECT 1 FROM tasks WHERE tasks.id = github_pr_watches.task_id)`)
		if err != nil {
			return 0, fmt.Errorf("remove orphaned-task PR watches: %w", err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return 0, fmt.Errorf("count orphaned-task PR watches removed: %w", err)
		}
		removed += n
	}
	if hasTaskRepositories {
		res, err := tx.Exec(`
			DELETE FROM github_pr_watches
			WHERE repository_id <> '' AND NOT EXISTS (
				SELECT 1 FROM task_repositories tr
				WHERE tr.task_id = github_pr_watches.task_id AND tr.repository_id = github_pr_watches.repository_id
			)`)
		if err != nil {
			return 0, fmt.Errorf("remove detached-repository PR watches: %w", err)
		}
		n, err := res.RowsAffected()
		if err != nil {
			return 0, fmt.Errorf("count detached-repository PR watches removed: %w", err)
		}
		removed += n
	}
	return removed, nil
}

// dedupPRWatchesForTaskOwnership groups the given rows into their canonical
// searching/discovered identities and decides which row in each group
// survives. Pass 1 dedups by tracked branch (preferring a discovered row
// over a still-searching one so a stale duplicate that never got promoted
// does not outlive the row that actually found the PR). Pass 2 then
// guarantees discovered uniqueness across branch groups - e.g. two
// different branches (a rename) independently resolving to the same PR
// number.
func dedupPRWatchesForTaskOwnership(all []*PRWatch) (toDelete map[string]bool, survivors map[string]*PRWatch) {
	toDelete = map[string]bool{}
	survivors = map[string]*PRWatch{}

	type branchKey struct{ taskID, repositoryID, branch string }
	branchGroups := map[branchKey][]*PRWatch{}
	for _, w := range all {
		if w.PRNumber != 0 {
			survivors[w.ID] = w
			continue
		}
		k := branchKey{w.TaskID, w.RepositoryID, w.Branch}
		branchGroups[k] = append(branchGroups[k], w)
	}
	for _, group := range branchGroups {
		mergePRWatchGroup(group, toDelete, survivors)
	}

	type prKey struct {
		taskID, repositoryID string
		prNumber             int
	}
	prGroups := map[prKey][]*PRWatch{}
	for id, w := range survivors {
		if toDelete[id] || w.PRNumber == 0 {
			continue
		}
		k := prKey{w.TaskID, w.RepositoryID, w.PRNumber}
		prGroups[k] = append(prGroups[k], w)
	}
	for _, group := range prGroups {
		if len(group) < 2 {
			continue
		}
		mergePRWatchGroup(group, toDelete, survivors)
	}

	return toDelete, survivors
}

// mergePRWatchGroup picks the survivor for one duplicate group (preferring a
// discovered row when present), merges the group's watermarks onto it, and
// marks every other member of the group for deletion.
func mergePRWatchGroup(group []*PRWatch, toDelete map[string]bool, survivors map[string]*PRWatch) {
	candidates := discoveredWatches(group)
	if len(candidates) == 0 {
		candidates = group
	}
	survivor := pickPRWatchSurvivor(candidates)
	mergePRWatchWatermarks(survivor, group)
	survivors[survivor.ID] = survivor
	for _, w := range group {
		if w.ID != survivor.ID {
			toDelete[w.ID] = true
		}
	}
}

// applyPRWatchDedup deletes the losing rows from a dedup pass and persists
// the winning rows' merged watermarks.
func applyPRWatchDedup(tx *sqlx.Tx, toDelete map[string]bool, survivors map[string]*PRWatch) error {
	for id := range toDelete {
		if _, err := tx.Exec(tx.Rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id); err != nil {
			return fmt.Errorf("remove duplicate PR watch %s: %w", id, err)
		}
	}
	for id, w := range survivors {
		if toDelete[id] {
			continue
		}
		if _, err := tx.Exec(tx.Rebind(`
			UPDATE github_pr_watches
			SET last_checked_at = ?, last_comment_at = ?, last_check_status = ?, last_review_state = ?
			WHERE id = ?`),
			w.LastCheckedAt, w.LastCommentAt, w.LastCheckStatus, w.LastReviewState, id); err != nil {
			return fmt.Errorf("merge PR watch watermark %s: %w", id, err)
		}
	}
	return nil
}

// rebuildPRWatchesTableForTaskOwnership rebuilds github_pr_watches without
// the legacy UNIQUE(session_id, repository_id, branch) constraint. SQLite
// can't ALTER TABLE DROP CONSTRAINT, so this uses the standard
// create-copy-drop-rename sequence (mirrors migratePRTablesForMultiRepo).
func rebuildPRWatchesTableForTaskOwnership(tx *sqlx.Tx) error {
	for _, stmt := range []string{
		`CREATE TABLE github_pr_watches_new (
			id TEXT PRIMARY KEY,
			workspace_id TEXT NOT NULL DEFAULT '',
			session_id TEXT NOT NULL DEFAULT '',
			task_id TEXT NOT NULL,
			repository_id TEXT NOT NULL DEFAULT '',
			owner TEXT NOT NULL,
			repo TEXT NOT NULL,
			pr_number INTEGER NOT NULL,
			branch TEXT NOT NULL,
			last_checked_at DATETIME,
			last_comment_at DATETIME,
			last_check_status TEXT DEFAULT '',
			last_review_state TEXT DEFAULT '',
			created_at DATETIME NOT NULL,
			updated_at DATETIME NOT NULL
		)`,
		`INSERT INTO github_pr_watches_new (
			id, workspace_id, session_id, task_id, repository_id, owner, repo, pr_number, branch,
			last_checked_at, last_comment_at, last_check_status, last_review_state, created_at, updated_at
		) SELECT
			id, COALESCE(workspace_id, ''), COALESCE(session_id, ''), task_id, COALESCE(repository_id, ''),
			owner, repo, pr_number, branch,
			last_checked_at, last_comment_at, last_check_status, last_review_state, created_at, updated_at
		FROM github_pr_watches`,
		`DROP TABLE github_pr_watches`,
		`ALTER TABLE github_pr_watches_new RENAME TO github_pr_watches`,
	} {
		if _, err := tx.Exec(stmt); err != nil {
			return fmt.Errorf("rebuild github_pr_watches for task ownership: %w", err)
		}
	}
	return nil
}

// tableSQLContains reports whether `table`'s stored CREATE statement in
// sqlite_master contains the literal substring. Returns false (not an error)
// when the table doesn't exist, matching rebuildIfHasLegacyConstraint's
// tolerance for unexpected schema drift in tests.
func (s *Store) tableSQLContains(table, substr string) (bool, error) {
	var existingSQL string
	err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE type='table' AND name=?`, table).Scan(&existingSQL)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return strings.Contains(existingSQL, substr), nil
}

// discoveredWatches returns the subset of watches that have found a PR
// (pr_number != 0).
func discoveredWatches(group []*PRWatch) []*PRWatch {
	var out []*PRWatch
	for _, w := range group {
		if w.PRNumber != 0 {
			out = append(out, w)
		}
	}
	return out
}

// pickPRWatchSurvivor deterministically picks the most-recently-updated
// watch from a duplicate group as the canonical row to keep.
func pickPRWatchSurvivor(group []*PRWatch) *PRWatch {
	best := group[0]
	for _, w := range group[1:] {
		if w.UpdatedAt.After(best.UpdatedAt) {
			best = w
		}
	}
	return best
}

// mergePRWatchWatermarks folds the newest last_comment_at across the group
// into survivor, and the newest last_checked_at together with its paired
// last_check_status/last_review_state (both observed at that same check) so
// a duplicate that happened to poll more recently isn't silently discarded.
func mergePRWatchWatermarks(survivor *PRWatch, group []*PRWatch) {
	for _, w := range group {
		if w.LastCommentAt != nil && (survivor.LastCommentAt == nil || w.LastCommentAt.After(*survivor.LastCommentAt)) {
			survivor.LastCommentAt = w.LastCommentAt
		}
		if w.LastCheckedAt != nil && (survivor.LastCheckedAt == nil || w.LastCheckedAt.After(*survivor.LastCheckedAt)) {
			survivor.LastCheckedAt = w.LastCheckedAt
			survivor.LastCheckStatus = w.LastCheckStatus
			survivor.LastReviewState = w.LastReviewState
		}
	}
}

// --- PR Watch operations ---

// CreatePRWatch creates a new PR watch.
func (s *Store) CreatePRWatch(ctx context.Context, w *PRWatch) error {
	if w.ID == "" {
		w.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	w.CreatedAt = now
	w.UpdatedAt = now
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_pr_watches (id, workspace_id, session_id, task_id, repository_id, owner, repo, pr_number, branch, last_check_status, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
		w.ID, w.WorkspaceID, w.SessionID, w.TaskID, w.RepositoryID, w.Owner, w.Repo, w.PRNumber, w.Branch, w.LastCheckStatus, w.CreatedAt, w.UpdatedAt)
	return err
}

// GetPRWatchBySession returns the first PR watch for a session. For
// multi-repo sessions the result is non-deterministic across repos — use
// GetPRWatchBySessionAndRepo or ListPRWatchesBySession instead.
func (s *Store) GetPRWatchBySession(ctx context.Context, sessionID string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE session_id = ? LIMIT 1`), sessionID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatch returns a PR watch by ID.
func (s *Store) GetPRWatch(ctx context.Context, id string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w, s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE id = ?`), id)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatchBySessionAndRepo returns the PR watch for a (session, repository)
// pair, or nil. Used by per-repo branch-switch / commit handlers so each
// repo's watch is reset independently.
//
// Multi-branch caveat: a task can hold multiple watches for the same
// (session, repository) on different branches. This lookup returns the
// most-recently-updated row — callers that need branch-specific lookup
// must use GetPRWatchBySessionRepoAndBranch.
func (s *Store) GetPRWatchBySessionAndRepo(ctx context.Context, sessionID, repositoryID string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE session_id = ? AND repository_id = ?
		 ORDER BY updated_at DESC LIMIT 1`,
		), sessionID, repositoryID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatchBySessionRepoAndBranch returns the PR watch for the precise
// (session, repository, branch) triple. Required for multi-branch tasks
// where each branch needs its own watch — querying by (session, repo)
// alone would collapse the secondary branch's push detection onto the
// primary's watch and the secondary PR would never land in github_task_prs.
func (s *Store) GetPRWatchBySessionRepoAndBranch(ctx context.Context, sessionID, repositoryID, branch string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches
		 WHERE session_id = ? AND repository_id = ? AND branch = ? LIMIT 1`,
		), sessionID, repositoryID, branch)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatchByTaskRepoBranch returns the canonical searching watch (pr_number
// = 0) for a (task, repository, branch) triple, or nil. This is the task-owned
// identity lookup (ADR 2026-08-31-task-owned-pr-watch-identity): unlike
// GetPRWatchBySessionRepoAndBranch, session identity plays no part, so it
// finds an existing watch regardless of which session originally created it.
func (s *Store) GetPRWatchByTaskRepoBranch(ctx context.Context, taskID, repositoryID, branch string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches
		 WHERE task_id = ? AND repository_id = ? AND branch = ? AND pr_number = 0 LIMIT 1`,
		),
		taskID, repositoryID, branch)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatchByTaskRepoBranchAny returns the canonical watch for a task,
// repository, and branch regardless of whether PR discovery has completed.
// A discovered watch must prevent reconciliation from creating a second
// searching row for the same branch.
func (s *Store) GetPRWatchByTaskRepoBranchAny(ctx context.Context, taskID, repositoryID, branch string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches
			WHERE task_id = ? AND repository_id = ? AND branch = ?
			ORDER BY pr_number DESC, updated_at DESC LIMIT 1`,
		), taskID, repositoryID, branch)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// GetPRWatchByTaskRepoPRNumber returns the canonical discovered watch for a
// (task, repository, pr_number) triple, or nil. Task-owned counterpart to
// GetPRWatchByTaskRepoBranch for watches that have already found their PR.
func (s *Store) GetPRWatchByTaskRepoPRNumber(ctx context.Context, taskID, repositoryID string, prNumber int) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w,
		s.ro.Rebind(`SELECT * FROM github_pr_watches
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ? LIMIT 1`,
		),
		taskID, repositoryID, prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// ListPRWatchesBySession returns every PR watch for a session (one per repo
// in multi-repo workspaces). Empty slice when no watches exist.
func (s *Store) ListPRWatchesBySession(ctx context.Context, sessionID string) ([]*PRWatch, error) {
	var watches []*PRWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE session_id = ? ORDER BY created_at ASC`), sessionID)
	return watches, err
}

// GetPRWatchByTask returns the first PR watch for a task. For multi-repo
// tasks the result is non-deterministic across repos — use
// ListPRWatchesByTask when every repo's watch is needed.
func (s *Store) GetPRWatchByTask(ctx context.Context, taskID string) (*PRWatch, error) {
	var w PRWatch
	err := s.ro.GetContext(ctx, &w, s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE task_id = ? ORDER BY updated_at DESC LIMIT 1`), taskID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &w, err
}

// ListPRWatchesByTask returns every PR watch for a task (one per repo in
// multi-repo workspaces).
func (s *Store) ListPRWatchesByTask(ctx context.Context, taskID string) ([]*PRWatch, error) {
	var watches []*PRWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_pr_watches WHERE task_id = ? ORDER BY created_at ASC`), taskID)
	return watches, err
}

// ListActivePRWatches returns all active PR watches whose task is not archived.
// Watches for archived tasks (and orphaned watches whose task row was hard-deleted)
// are excluded so the poller stops making GitHub API calls for them. An INNER JOIN
// on `tasks` is used so orphans are dropped automatically.
func (s *Store) ListActivePRWatches(ctx context.Context) ([]*PRWatch, error) {
	var watches []*PRWatch
	err := s.ro.SelectContext(ctx, &watches, s.ro.Rebind(`
		SELECT w.* FROM github_pr_watches w
		INNER JOIN tasks t ON t.id = w.task_id
		WHERE t.archived_at IS NULL
		ORDER BY w.created_at`))
	return watches, err
}

// PRWatchCardinality is the aggregate, identity-free state exported for
// operational metrics. Counts never contain task, repository, branch, or
// workspace labels.
type PRWatchCardinality struct {
	Active     int64
	Searching  int64
	Duplicates int64
	Orphans    int64
}

// PRWatchCardinality returns current watch health across the installation.
// The duplicate count is the number of rows beyond one canonical survivor;
// unique indexes normally keep it at zero, making a non-zero value a direct
// corruption/migration signal.
func (s *Store) PRWatchCardinality(ctx context.Context) (PRWatchCardinality, error) {
	var out PRWatchCardinality
	activeQuery := `
		SELECT COUNT(*), COALESCE(SUM(CASE WHEN w.pr_number = 0 THEN 1 ELSE 0 END), 0)
		FROM github_pr_watches w
		INNER JOIN tasks t ON t.id = w.task_id
		WHERE t.archived_at IS NULL`
	if err := s.ro.QueryRowContext(ctx, activeQuery).Scan(&out.Active, &out.Searching); err != nil {
		return PRWatchCardinality{}, fmt.Errorf("count active PR watches: %w", err)
	}

	duplicateQueries := []string{
		`SELECT COALESCE(SUM(group_count - 1), 0) FROM (
			SELECT COUNT(*) AS group_count FROM github_pr_watches
			WHERE pr_number = 0 GROUP BY task_id, repository_id, branch HAVING COUNT(*) > 1
		) duplicate_groups`,
		`SELECT COALESCE(SUM(group_count - 1), 0) FROM (
			SELECT COUNT(*) AS group_count FROM github_pr_watches
			WHERE pr_number <> 0 GROUP BY task_id, repository_id, pr_number HAVING COUNT(*) > 1
		) duplicate_groups`,
	}
	for _, query := range duplicateQueries {
		var duplicates int64
		if err := s.ro.QueryRowContext(ctx, query).Scan(&duplicates); err != nil {
			return PRWatchCardinality{}, fmt.Errorf("count duplicate PR watches: %w", err)
		}
		out.Duplicates += duplicates
	}

	orphanQuery := `
		SELECT COUNT(*) FROM github_pr_watches w
		LEFT JOIN tasks t ON t.id = w.task_id
		LEFT JOIN task_repositories tr ON tr.repository_id = w.repository_id AND tr.task_id = w.task_id
		WHERE t.id IS NULL OR (w.repository_id <> '' AND tr.repository_id IS NULL)`
	if err := s.ro.QueryRowContext(ctx, orphanQuery).Scan(&out.Orphans); err != nil {
		fallback := `SELECT COUNT(*) FROM github_pr_watches w LEFT JOIN tasks t ON t.id = w.task_id WHERE t.id IS NULL`
		if fallbackErr := s.ro.QueryRowContext(ctx, fallback).Scan(&out.Orphans); fallbackErr != nil {
			return PRWatchCardinality{}, fmt.Errorf("count orphaned PR watches: %w", err)
		}
	}
	return out, nil
}

// ListActivePRWatchesForWorkspace returns active watches only for one workspace.
func (s *Store) ListActivePRWatchesForWorkspace(ctx context.Context, workspaceID string) ([]*PRWatch, error) {
	var watches []*PRWatch
	err := s.ro.SelectContext(ctx, &watches, s.ro.Rebind(`
		SELECT w.* FROM github_pr_watches w
		INNER JOIN tasks t ON t.id = w.task_id
		WHERE w.workspace_id = ? AND t.archived_at IS NULL
		ORDER BY w.created_at`), workspaceID)
	return watches, err
}

// UpdatePRWatchTimestamps updates the last checked timestamps and status fields.
func (s *Store) UpdatePRWatchTimestamps(ctx context.Context, id string, checkedAt time.Time, commentAt *time.Time, checkStatus, reviewState string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_pr_watches SET last_checked_at = ?, last_comment_at = ?, last_check_status = ?, last_review_state = ?, updated_at = ?
		WHERE id = ?`),
		checkedAt, commentAt, checkStatus, reviewState, time.Now().UTC(), id)
	return err
}

// DeletePRWatch deletes a PR watch by ID.
func (s *Store) DeletePRWatch(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id)
	return err
}

// DeletePRWatchesByTaskID deletes all PR watches for a task. Returns the number
// of rows removed so callers can log meaningful diagnostics.
func (s *Store) DeletePRWatchesByTaskID(ctx context.Context, taskID string) (int64, error) {
	res, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_pr_watches WHERE task_id = ?`), taskID)
	if err != nil {
		return 0, err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return 0, err
	}
	return n, nil
}

// UpdatePRWatchPRNumber updates a PR watch's PR number after discovery (or
// resets it to 0 to re-search). Promoting to a non-zero PR number could
// collide with the canonical discovered-watch identity (task_id,
// repository_id, pr_number) if a sibling watch already tracks that PR - e.g.
// two branches (a rename) independently resolving to the same PR. In that
// case the source watch is redundant (the sibling already owns the PR's
// state) and is dropped rather than left to trip the unique index.
func (s *Store) UpdatePRWatchPRNumber(ctx context.Context, id string, prNumber int) error {
	if prNumber == 0 {
		_, err := s.db.ExecContext(ctx,
			s.db.Rebind(`UPDATE github_pr_watches SET pr_number = 0, updated_at = ? WHERE id = ?`),
			time.Now().UTC(), id)
		return err
	}

	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	var taskID, repositoryID string
	err = tx.QueryRowContext(ctx,
		s.db.Rebind(`SELECT task_id, repository_id FROM github_pr_watches WHERE id = ?`), id).
		Scan(&taskID, &repositoryID)
	if errors.Is(err, sql.ErrNoRows) {
		return tx.Commit()
	}
	if err != nil {
		return err
	}

	var probe int // existence probe only; value unused
	err = tx.QueryRowContext(ctx,
		s.db.Rebind(`SELECT 1 FROM github_pr_watches
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ? AND id <> ?`,
		),
		taskID, repositoryID, prNumber, id).Scan(&probe)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if err == nil {
		if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id); err != nil {
			return err
		}
		return tx.Commit()
	}

	const savepoint = "update_pr_watch_pr_number"
	if _, err := tx.ExecContext(ctx, s.db.Rebind("SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx,
		s.db.Rebind(`UPDATE github_pr_watches SET pr_number = ?, updated_at = ? WHERE id = ?`),
		prNumber, time.Now().UTC(), id); err != nil {
		return recoverPRWatchDiscoveredUpdate(ctx, tx, s.db.Rebind, savepoint, id, err, tx.Commit)
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind("RELEASE SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	return tx.Commit()
}

// UpdatePRWatchRepository repairs the provider repository identity after PR
// discovery. A watch can start on a contributor fork while the PR targets the
// canonical parent repository.
func (s *Store) UpdatePRWatchRepository(ctx context.Context, id, owner, repo string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_pr_watches SET owner = ?, repo = ?, updated_at = ? WHERE id = ?`),
		owner, repo, time.Now().UTC(), id)
	return err
}

// ResetPRWatch atomically resets a watch to the searching state: updates the
// tracked branch and clears pr_number in a single statement. Used when the
// session's active branch changes (rename, checkout) so the poller re-searches
// for a PR on the new branch without leaving an inconsistent intermediate
// state.
func (s *Store) ResetPRWatch(ctx context.Context, id, branch string) error {
	// SQLite's deferred transactions let two sessions both probe an empty
	// destination before either writer acquires the lock. BEGIN IMMEDIATE
	// serializes the probe and the coalescing update as one operation.
	var tx prWatchTx
	var commit func() error
	var rollback func()
	if dialect.IsPostgres(s.db.DriverName()) {
		pgTx, err := s.db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		tx = pgTx
		commit = pgTx.Commit
		rollback = func() { _ = pgTx.Rollback() }
	} else {
		conn, err := s.db.Conn(ctx)
		if err != nil {
			return err
		}
		if _, err := conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
			_ = conn.Close()
			return err
		}
		tx = conn
		commit = func() error {
			_, err := conn.ExecContext(ctx, "COMMIT")
			return err
		}
		rollback = func() {
			_, _ = conn.ExecContext(context.WithoutCancel(ctx), "ROLLBACK")
			_ = conn.Close()
		}
	}
	defer rollback()

	var taskID, repositoryID string
	err := tx.QueryRowContext(ctx, s.db.Rebind(
		`SELECT task_id, repository_id FROM github_pr_watches WHERE id = ?`), id).
		Scan(&taskID, &repositoryID)
	if errors.Is(err, sql.ErrNoRows) {
		return commit()
	}
	if err != nil {
		return err
	}

	var probe int
	err = tx.QueryRowContext(ctx, s.db.Rebind(
		`SELECT 1 FROM github_pr_watches
			 WHERE task_id = ? AND repository_id = ? AND branch = ? AND id <> ?`,
	), taskID, repositoryID, branch, id).Scan(&probe)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if err == nil {
		if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id); err != nil {
			return err
		}
		return commit()
	}

	const savepoint = "reset_pr_watch_update"
	if _, err := tx.ExecContext(ctx, s.db.Rebind("SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_pr_watches SET branch = ?, pr_number = 0, updated_at = ? WHERE id = ?`,
	), branch, time.Now().UTC(), id); err != nil {
		return recoverPRWatchUniqueUpdate(ctx, tx, s.db.Rebind, savepoint, id, err, commit)
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind("RELEASE SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	return commit()
}

const prWatchSearchingIndexName = "idx_github_pr_watches_searching"
const prWatchDiscoveredIndexName = "idx_github_pr_watches_discovered"

func isPRWatchUniqueViolation(err error) bool {
	if err == nil {
		return false
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) {
		return pgErr.Code == "23505" && pgErr.ConstraintName == prWatchSearchingIndexName
	}
	return strings.Contains(err.Error(), "UNIQUE constraint failed")
}

func isPRWatchDiscoveredUniqueViolation(err error) bool {
	if err == nil {
		return false
	}
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) {
		return pgErr.Code == "23505" && pgErr.ConstraintName == prWatchDiscoveredIndexName
	}
	return strings.Contains(err.Error(), "UNIQUE constraint failed")
}

func recoverPRWatchUniqueUpdate(
	ctx context.Context,
	tx prWatchTx,
	rebind func(string) string,
	savepoint, id string,
	updateErr error,
	commit func() error,
) error {
	if _, err := tx.ExecContext(ctx, rebind("ROLLBACK TO SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, rebind("RELEASE SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if !isPRWatchUniqueViolation(updateErr) {
		return updateErr
	}
	if _, err := tx.ExecContext(ctx, rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id); err != nil {
		return err
	}
	return commit()
}

func recoverPRWatchDiscoveredUpdate(
	ctx context.Context,
	tx prWatchTx,
	rebind func(string) string,
	savepoint, id string,
	updateErr error,
	commit func() error,
) error {
	if _, err := tx.ExecContext(ctx, rebind("ROLLBACK TO SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, rebind("RELEASE SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	// An external writer can still race between the preflight probe and the
	// UPDATE. Treat that discovered-index collision like the probe-found path.
	if !isPRWatchDiscoveredUniqueViolation(updateErr) {
		return updateErr
	}
	if _, err := tx.ExecContext(ctx, rebind(`DELETE FROM github_pr_watches WHERE id = ?`), id); err != nil {
		return err
	}
	return commit()
}

// UpdatePRWatchBranchIfSearching atomically updates branch only when pr_number = 0,
// preventing races with concurrent PR association.
//
// Collision semantics: a sibling watch may already own the destination
// (task_id, repository_id, branch) triple — e.g. multi-branch task where the
// agent's live branch collapsed onto a peer watch's branch. In that case the
// raw UPDATE would trip the UNIQUE constraint. We instead drop the source row
// (which is still searching, pr_number=0, so it owns no PR state) and let the
// sibling continue to track the branch.
func (s *Store) UpdatePRWatchBranchIfSearching(ctx context.Context, id, branch string) error {
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	var taskID, repositoryID string
	var prNumber int
	err = tx.QueryRowContext(ctx,
		s.db.Rebind(`SELECT task_id, repository_id, pr_number FROM github_pr_watches WHERE id = ?`), id).
		Scan(&taskID, &repositoryID, &prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		return tx.Commit()
	}
	if err != nil {
		return err
	}
	if prNumber != 0 {
		return tx.Commit()
	}

	var probe int // existence probe only; value unused
	err = tx.QueryRowContext(ctx,
		s.db.Rebind(`SELECT 1 FROM github_pr_watches
			 WHERE task_id = ? AND repository_id = ? AND branch = ? AND id <> ?`),
		taskID, repositoryID, branch, id).Scan(&probe)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	if err == nil {
		return dropSourceAndCommit(ctx, tx, id)
	}

	const savepoint = "update_pr_watch_branch"
	if _, err := tx.ExecContext(ctx, s.db.Rebind("SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_pr_watches SET branch = ?, updated_at = ? WHERE id = ? AND pr_number = 0`,
	), branch, time.Now().UTC(), id); err != nil {
		return recoverPRWatchUniqueUpdate(ctx, tx, s.db.Rebind, savepoint, id, err, tx.Commit)
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind("RELEASE SAVEPOINT "+savepoint)); err != nil {
		return err
	}
	return tx.Commit()
}

type prWatchTx interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

// dropSourceAndCommit removes a still-searching source watch (pr_number=0)
// whose destination branch is already owned by a sibling row, then commits.
func dropSourceAndCommit(ctx context.Context, tx *sqlx.Tx, id string) error {
	if _, err := tx.ExecContext(ctx, tx.Rebind(
		`DELETE FROM github_pr_watches WHERE id = ? AND pr_number = 0`), id); err != nil {
		return err
	}
	return tx.Commit()
}

// --- TaskPR operations ---

// CreateTaskPR associates a PR with a task. RepositoryID may be empty for
// single-repo tasks; multi-repo task launches set it so each repo's PR is
// distinguishable.
func (s *Store) CreateTaskPR(ctx context.Context, tp *TaskPR) error {
	if tp.ID == "" {
		tp.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	tp.UpdatedAt = now
	tp.WorkflowAttentionJSON = marshalWorkflowAttention(tp.WorkflowAttention)
	return insertTaskPR(ctx, s.db, tp)
}

// taskPRColumns is the explicit column list for every github_task_prs read.
// Using `SELECT *` here is unsafe: sqlx's StructScan errors out
// ("missing destination name X in *github.TaskPR") the moment the table has
// a column the current TaskPR struct doesn't declare a field for. That drift
// is not hypothetical — a self-update that applies a newer release's
// migration (adding a column) followed by a rollback to an older binary
// leaves exactly this mismatch permanently on disk, since SQLite migrations
// are additive and never reverted. Projecting the known columns keeps every
// read working regardless of what the table has picked up beyond them.
const taskPRColumns = `id, workspace_id, task_id, repository_id, owner, repo, pr_number, pr_url,
	pr_title, head_branch, base_branch, author_login, state, review_state, checks_state,
	mergeable_state, merge_queue_state, merge_queue_position, merge_queue_estimated_time_to_merge_seconds, review_count, pending_review_count, required_reviews, comment_count,
	unresolved_review_threads, checks_total, checks_passing, additions, deletions,
	created_at, merged_at, closed_at, last_synced_at, detached_at, updated_at,
	is_draft, changed_files, merged_by_login, closed_by_login, auto_merge_observed_at,
	head_sha, merge_queue_entry_id, merge_queue_entry_head_sha, merge_queue_last_removal_id,
	merge_queue_last_removed_at, merge_queue_last_removal_reason, merge_queue_last_removal_before_sha,
	source, workflow_attention`

// taskPRColumnsQualified is taskPRColumns with each column qualified by the
// `gtp` alias, for queries that join github_task_prs against another table.
const taskPRColumnsQualified = `gtp.id, gtp.workspace_id, gtp.task_id, gtp.repository_id, gtp.owner, gtp.repo,
	gtp.pr_number, gtp.pr_url, gtp.pr_title, gtp.head_branch, gtp.base_branch, gtp.author_login,
	gtp.state, gtp.review_state, gtp.checks_state, gtp.mergeable_state, gtp.merge_queue_state, gtp.merge_queue_position, gtp.merge_queue_estimated_time_to_merge_seconds, gtp.review_count,
	gtp.pending_review_count, gtp.required_reviews, gtp.comment_count, gtp.unresolved_review_threads,
	gtp.checks_total, gtp.checks_passing, gtp.additions, gtp.deletions,
	gtp.created_at, gtp.merged_at, gtp.closed_at, gtp.last_synced_at, gtp.detached_at, gtp.updated_at,
	gtp.is_draft, gtp.changed_files, gtp.merged_by_login, gtp.closed_by_login, gtp.auto_merge_observed_at,
	gtp.head_sha, gtp.merge_queue_entry_id, gtp.merge_queue_entry_head_sha, gtp.merge_queue_last_removal_id,
	gtp.merge_queue_last_removed_at, gtp.merge_queue_last_removal_reason, gtp.merge_queue_last_removal_before_sha,
	gtp.source, gtp.workflow_attention`

type taskPRWriter interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	Rebind(string) string
}

func insertTaskPR(ctx context.Context, writer taskPRWriter, tp *TaskPR) error {
	values := taskPRValues(tp)
	columnCount := taskPRColumnCount()
	if len(values) != columnCount {
		return fmt.Errorf("task PR insert arity mismatch: %d columns, %d values", columnCount, len(values))
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", columnCount), ",")
	_, err := writer.ExecContext(ctx, writer.Rebind(
		`INSERT INTO github_task_prs (`+taskPRColumns+`) VALUES (`+placeholders+`)`),
		values...,
	)
	return err
}

func taskPRColumnCount() int {
	count := 0
	for _, column := range strings.Split(taskPRColumns, ",") {
		if strings.TrimSpace(column) != "" {
			count++
		}
	}
	return count
}

func taskPRValues(tp *TaskPR) []any {
	return []any{
		tp.ID, tp.WorkspaceID, tp.TaskID, tp.RepositoryID, tp.Owner, tp.Repo, tp.PRNumber, tp.PRURL,
		tp.PRTitle, tp.HeadBranch, tp.BaseBranch, tp.AuthorLogin, tp.State, tp.ReviewState,
		tp.ChecksState, tp.MergeableState, tp.MergeQueueState, tp.MergeQueuePosition,
		tp.MergeQueueEstimatedTimeToMergeSeconds, tp.ReviewCount, tp.PendingReviewCount,
		tp.RequiredReviews, tp.CommentCount, tp.UnresolvedReviewThreads, tp.ChecksTotal,
		tp.ChecksPassing, tp.Additions, tp.Deletions, tp.CreatedAt, tp.MergedAt, tp.ClosedAt,
		tp.LastSyncedAt, tp.DetachedAt, tp.UpdatedAt, tp.IsDraft, tp.ChangedFiles,
		tp.MergedByLogin, tp.ClosedByLogin, tp.AutoMergeObservedAt, tp.HeadSHA,
		tp.MergeQueueEntryID, tp.MergeQueueEntryHeadSHA, tp.MergeQueueLastRemovalID,
		tp.MergeQueueLastRemovedAt, tp.MergeQueueLastRemovalReason, tp.MergeQueueLastRemovalBeforeSHA,
		tp.Source, marshalWorkflowAttention(tp.WorkflowAttention),
	}
}

func marshalWorkflowAttention(attention *WorkflowAttention) string {
	if attention == nil {
		return ""
	}
	encoded, err := json.Marshal(attention)
	if err != nil {
		return ""
	}
	return string(encoded)
}

func hydrateTaskPRWorkflowAttention(tp *TaskPR) error {
	if tp == nil || strings.TrimSpace(tp.WorkflowAttentionJSON) == "" {
		if tp != nil {
			tp.WorkflowAttention = nil
		}
		return nil
	}
	var attention WorkflowAttention
	if err := json.Unmarshal([]byte(tp.WorkflowAttentionJSON), &attention); err != nil {
		return fmt.Errorf("decode task PR workflow attention: %w", err)
	}
	if attention.Runs == nil {
		attention.Runs = []WorkflowAttentionRun{}
	}
	tp.WorkflowAttention = &attention
	return nil
}

func hydrateTaskPRsWorkflowAttention(prs []TaskPR) error {
	for i := range prs {
		if err := hydrateTaskPRWorkflowAttention(&prs[i]); err != nil {
			return err
		}
	}
	return nil
}

// GetTaskPR returns the first PR association for a task. For multi-repo tasks
// the result is non-deterministic across repos — use ListTaskPRsByTask instead.
func (s *Store) GetTaskPR(ctx context.Context, taskID string) (*TaskPR, error) {
	var tp TaskPR
	err := s.ro.GetContext(ctx, &tp, s.ro.Rebind(
		`SELECT `+taskPRColumns+` FROM github_task_prs WHERE task_id = ? AND detached_at IS NULL LIMIT 1`), taskID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err == nil {
		err = hydrateTaskPRWorkflowAttention(&tp)
	}
	return &tp, err
}

// GetTaskPRByID returns an association regardless of whether it was detached.
// Mutation paths use this exact lookup so detached rows remain durable
// tombstones and can be explicitly restored by a later link action.
func (s *Store) GetTaskPRByID(ctx context.Context, associationID string) (*TaskPR, error) {
	var tp TaskPR
	err := s.ro.GetContext(ctx, &tp,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs WHERE id = ? LIMIT 1`), associationID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err == nil {
		err = hydrateTaskPRWorkflowAttention(&tp)
	}
	return &tp, err
}

// GetTaskPRByRepository returns the PR association for a (task, repository)
// pair, or nil if none. Use this for multi-repo tasks.
//
// Multi-branch caveat: a task can hold N rows per (task, repo) — one per
// PR number. This lookup returns the most-recently-updated row so callers
// that need a deterministic single value still get one. Callers that need
// the row for a specific PR number must use GetTaskPRByRepoAndNumber.
func (s *Store) GetTaskPRByRepository(ctx context.Context, taskID, repositoryID string) (*TaskPR, error) {
	var tp TaskPR
	err := s.ro.GetContext(ctx, &tp,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs WHERE task_id = ? AND repository_id = ? AND detached_at IS NULL
		 ORDER BY updated_at DESC LIMIT 1`),
		taskID, repositoryID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err == nil {
		err = hydrateTaskPRWorkflowAttention(&tp)
	}
	return &tp, err
}

// GetTaskPRByRepoAndNumber returns the exact PR row matching the
// (task, repository, pr_number) triple. Required for multi-branch tasks
// where AssociatePRWithTask's "already-current" short-circuit must check
// the same PR number, not a sibling PR that happens to be the first
// row returned by the legacy by-repo query.
func (s *Store) GetTaskPRByRepoAndNumber(ctx context.Context, taskID, repositoryID string, prNumber int) (*TaskPR, error) {
	var tp TaskPR
	err := s.ro.GetContext(ctx, &tp,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ? AND detached_at IS NULL LIMIT 1`),
		taskID, repositoryID, prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err == nil {
		err = hydrateTaskPRWorkflowAttention(&tp)
	}
	return &tp, err
}

// GetTaskPRByRepoAndNumberIncludingDetached returns the exact association
// row, including a detached tombstone, for automatic-versus-explicit link
// decisions.
func (s *Store) GetTaskPRByRepoAndNumberIncludingDetached(ctx context.Context, taskID, repositoryID string, prNumber int) (*TaskPR, error) {
	var tp TaskPR
	err := s.ro.GetContext(ctx, &tp,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ? LIMIT 1`),
		taskID, repositoryID, prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err == nil {
		err = hydrateTaskPRWorkflowAttention(&tp)
	}
	return &tp, err
}

// ListTaskPRsByTask returns every PR association for a task (one per repo
// when multi-repo). Empty slice when no PRs exist.
func (s *Store) ListTaskPRsByTask(ctx context.Context, taskID string) ([]*TaskPR, error) {
	var prs []TaskPR
	if err := s.ro.SelectContext(ctx, &prs,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs WHERE task_id = ? AND detached_at IS NULL ORDER BY created_at ASC`), taskID); err != nil {
		return nil, err
	}
	if err := hydrateTaskPRsWorkflowAttention(prs); err != nil {
		return nil, err
	}
	out := make([]*TaskPR, 0, len(prs))
	for i := range prs {
		out = append(out, &prs[i])
	}
	return out, nil
}

// ListTaskPRsByTaskIncludingDetached returns every PR association for a task,
// including detached tombstones. Callers that expose active task surfaces
// should use ListTaskPRsByTask; this variant is for cleanup of state keyed by
// an association that may have been detached.
func (s *Store) ListTaskPRsByTaskIncludingDetached(ctx context.Context, taskID string) ([]*TaskPR, error) {
	var prs []TaskPR
	if err := s.ro.SelectContext(ctx, &prs,
		s.ro.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs WHERE task_id = ? ORDER BY created_at ASC`), taskID); err != nil {
		return nil, err
	}
	if err := hydrateTaskPRsWorkflowAttention(prs); err != nil {
		return nil, err
	}
	out := make([]*TaskPR, 0, len(prs))
	for i := range prs {
		out = append(out, &prs[i])
	}
	return out, nil
}

// DetachTaskPR marks an association as removed from active task surfaces while
// retaining the row so automatic PR discovery cannot silently resurrect it.
// The bool return reports whether this call performed the transition.
func (s *Store) DetachTaskPR(ctx context.Context, associationID string) (*TaskPR, bool, error) {
	now := time.Now().UTC()
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, false, err
	}
	defer func() { _ = tx.Rollback() }()
	var identity struct {
		TaskID       string `db:"task_id"`
		RepositoryID string `db:"repository_id"`
		PRNumber     int    `db:"pr_number"`
	}
	if err := tx.GetContext(ctx, &identity, tx.Rebind(`SELECT task_id, repository_id, pr_number FROM github_task_prs WHERE id = ? AND detached_at IS NULL`), associationID); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			_ = tx.Rollback()
			tp, getErr := s.GetTaskPRByID(ctx, associationID)
			return tp, false, getErr
		}
		return nil, false, err
	}
	result, err := tx.ExecContext(ctx, tx.Rebind(
		`UPDATE github_task_prs SET detached_at = ?, updated_at = ?
		 WHERE id = ? AND detached_at IS NULL`), now, now, associationID)
	if err != nil {
		return nil, false, err
	}
	count, err := result.RowsAffected()
	if err != nil {
		return nil, false, err
	}
	if count > 0 {
		for _, table := range []string{"github_task_pr_automation_options", "github_task_ci_pr_state"} {
			if _, err := tx.ExecContext(ctx, tx.Rebind(`DELETE FROM `+table+` WHERE task_id = ? AND repository_id = ? AND pr_number = ?`), identity.TaskID, identity.RepositoryID, identity.PRNumber); err != nil {
				return nil, false, err
			}
		}
	}
	var tp TaskPR
	if err := tx.GetContext(ctx, &tp, tx.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs WHERE id = ? LIMIT 1`), associationID); err != nil {
		return nil, false, err
	}
	if err := tx.Commit(); err != nil {
		return nil, false, err
	}
	return &tp, count > 0, nil
}

// RestoreTaskPR clears a detached tombstone for an explicit link action and
// refreshes the persisted fields available from the fetched GitHub PR.
// status carries the caller's raw PR observation and its populated-ness
// flags (AC-43p) — RestoreTaskPR resolves the five outcome-attribution
// columns itself, inside this call's own transaction, against the outgoing
// row it is about to overwrite (AC-43, AC-43a). The caller must not
// pre-resolve them via resolveTaskPROutcomeFields: a value resolved against
// an earlier read is stale by the time this statement executes, and on a
// row that has just reached a terminal state there is no later poll to
// repair a value clobbered that way (AC-36).
func (s *Store) RestoreTaskPR(ctx context.Context, taskID, repositoryID string, status *PRStatus) (*TaskPR, error) {
	if status == nil || status.PR == nil {
		return nil, errors.New("restore task PR: missing PR data")
	}
	pr := status.PR

	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	var outgoing TaskPR
	err = tx.GetContext(ctx, &outgoing,
		tx.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs
			 WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		taskID, repositoryID, pr.Number)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return nil, err
	}
	if err == nil {
		if err := hydrateTaskPRWorkflowAttention(&outgoing); err != nil {
			return nil, err
		}
	}

	isDraft, changedFiles, mergedByLogin, closedByLogin, autoMergeObservedAt :=
		resolveTaskPROutcomeFields(&outgoing, status)
	queue := resolveTaskPRMergeQueueState(&outgoing, status)
	headSHA := outgoing.HeadSHA
	if pr.HeadSHA != "" {
		headSHA = pr.HeadSHA
	}
	workflowAttention := resolveTaskPRWorkflowAttention(&outgoing, status, headSHA)

	if _, err := tx.ExecContext(ctx, tx.Rebind(
		`UPDATE github_task_prs SET owner = ?, repo = ?, pr_url = ?, pr_title = ?,
			head_branch = ?, base_branch = ?, head_sha = ?, author_login = ?, state = ?, mergeable_state = ?,
			merge_queue_state = ?, merge_queue_position = ?, merge_queue_entry_id = ?, merge_queue_entry_head_sha = ?, merge_queue_estimated_time_to_merge_seconds = ?,
			merge_queue_last_removal_id = ?, merge_queue_last_removed_at = ?, merge_queue_last_removal_reason = ?, merge_queue_last_removal_before_sha = ?,
			additions = ?, deletions = ?, merged_at = ?, closed_at = ?, detached_at = NULL, updated_at = ?,
			is_draft = ?, changed_files = ?, merged_by_login = ?, closed_by_login = ?,
			auto_merge_observed_at = COALESCE(auto_merge_observed_at, ?), workflow_attention = ?
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		pr.RepoOwner, pr.RepoName, pr.HTMLURL, pr.Title, pr.HeadBranch, pr.BaseBranch, headSHA, pr.AuthorLogin,
		pr.State, pr.MergeableState, queue.state, queue.position, queue.entryID, queue.entryHeadSHA, queue.estimate,
		queue.lastRemovalID, queue.lastRemovedAt, queue.lastRemovalReason, queue.lastRemovalBeforeSHA,
		pr.Additions, pr.Deletions, pr.MergedAt, pr.ClosedAt, time.Now().UTC(),
		isDraft, changedFiles, mergedByLogin, closedByLogin, autoMergeObservedAt,
		marshalWorkflowAttention(workflowAttention),
		taskID, repositoryID, pr.Number); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetTaskPRByRepoAndNumber(ctx, taskID, repositoryID, pr.Number)
}

// ListTaskPRsByTaskIDs returns PR associations for multiple tasks. Each task
// may have multiple PRs (one per repository for multi-repo tasks); rows are
// returned grouped by task_id, ordered by created_at ascending within a group.
func (s *Store) ListTaskPRsByTaskIDs(ctx context.Context, taskIDs []string) (map[string][]*TaskPR, error) {
	if len(taskIDs) == 0 {
		return make(map[string][]*TaskPR), nil
	}
	query, args, err := sqlx.In(
		`SELECT `+taskPRColumns+` FROM github_task_prs WHERE task_id IN (?) AND detached_at IS NULL ORDER BY created_at ASC`,
		taskIDs,
	)
	if err != nil {
		return nil, err
	}
	query = s.ro.Rebind(query)
	var prs []TaskPR
	if err := s.ro.SelectContext(ctx, &prs, query, args...); err != nil {
		return nil, err
	}
	if err := hydrateTaskPRsWorkflowAttention(prs); err != nil {
		return nil, err
	}
	return groupTaskPRsByTask(prs), nil
}

// ListTaskPRsByWorkspaceID returns all PR associations for tasks in a workspace.
// Each task may have multiple PRs (one per repository for multi-repo tasks);
// rows are returned grouped by task_id, ordered by created_at ascending.
func (s *Store) ListTaskPRsByWorkspaceID(ctx context.Context, workspaceID string) (map[string][]*TaskPR, error) {
	var prs []TaskPR
	if err := s.ro.SelectContext(ctx, &prs,
		s.ro.Rebind(`SELECT `+taskPRColumnsQualified+` FROM github_task_prs gtp
		 INNER JOIN tasks t ON gtp.task_id = t.id
		 WHERE t.workspace_id = ? AND gtp.detached_at IS NULL
			ORDER BY gtp.created_at ASC`), workspaceID); err != nil {
		return nil, err
	}
	if err := hydrateTaskPRsWorkflowAttention(prs); err != nil {
		return nil, err
	}
	return groupTaskPRsByTask(prs), nil
}

// ListTaskIssueMetadataByWorkspaceID projects workspace-bounded task metadata for issue-link
// normalization. Issue links are persisted in metadata, so parsing the two supported shapes in
// Go keeps the SQL query independent of SQLite's JSON capabilities.
func (s *Store) ListTaskIssueMetadataByWorkspaceID(ctx context.Context, workspaceID string) ([]taskIssueMetadataRow, error) {
	var rows []taskIssueMetadataRow
	err := s.ro.SelectContext(ctx, &rows, s.ro.Rebind(
		`SELECT id AS task_id, title AS task_title, COALESCE(metadata, '{}') AS metadata
		 FROM tasks
		 WHERE workspace_id = ? AND archived_at IS NULL
		 ORDER BY created_at ASC, id ASC`,
	), workspaceID)
	return rows, err
}

// ListTaskIDsByPRNumber returns the IDs of tasks in a workspace that have a PR
// association with the given PR number. Workspace-scoped via the JOIN on tasks
// so a PR number shared across workspaces never leaks results. A task with
// multiple PR rows for the same number (multi-repo) is returned once.
func (s *Store) ListTaskIDsByPRNumber(ctx context.Context, workspaceID string, prNumber int) ([]string, error) {
	var ids []string
	if err := s.ro.SelectContext(ctx, &ids,
		s.ro.Rebind(`SELECT DISTINCT gtp.task_id FROM github_task_prs gtp
		 INNER JOIN tasks t ON gtp.task_id = t.id
			 WHERE t.workspace_id = ? AND gtp.pr_number = ? AND gtp.detached_at IS NULL`), workspaceID, prNumber); err != nil {
		return nil, err
	}
	return ids, nil
}

// ListTaskPRsByPRNumber returns the non-detached PR associations in a workspace
// that point at one exact (owner, repo, pr_number). Workspace-scoped via the
// JOIN on tasks — the same guard ListTaskIDsByPRNumber uses — so a PR number
// shared across workspaces never reaches a caller holding only one workspace's
// credentials. More than one row is normal: several tasks in a workspace can
// legitimately link the same PR.
func (s *Store) ListTaskPRsByPRNumber(
	ctx context.Context, workspaceID, owner, repo string, prNumber int,
) ([]*TaskPR, error) {
	var prs []TaskPR
	if err := s.ro.SelectContext(ctx, &prs,
		s.ro.Rebind(`SELECT `+taskPRColumnsQualified+` FROM github_task_prs gtp
		 INNER JOIN tasks t ON gtp.task_id = t.id
		 WHERE t.workspace_id = ? AND gtp.owner = ? AND gtp.repo = ? AND gtp.pr_number = ?
			 AND gtp.detached_at IS NULL
			ORDER BY gtp.created_at ASC`), workspaceID, owner, repo, prNumber); err != nil {
		return nil, err
	}
	if err := hydrateTaskPRsWorkflowAttention(prs); err != nil {
		return nil, err
	}
	out := make([]*TaskPR, 0, len(prs))
	for i := range prs {
		out = append(out, &prs[i])
	}
	return out, nil
}

func groupTaskPRsByTask(prs []TaskPR) map[string][]*TaskPR {
	result := make(map[string][]*TaskPR)
	for i := range prs {
		taskID := prs[i].TaskID
		result[taskID] = append(result[taskID], &prs[i])
	}
	return result
}

// ReplaceTaskPR atomically associates a PR with a task, replacing only the
// row that matches the exact (task_id, repository_id, pr_number) triple.
// Multi-branch tasks may hold multiple PR rows per (task, repo) — one per
// branch — so the delete MUST NOT wipe sibling PR rows. Single-repo
// callers (RepositoryID == "") only delete legacy untagged rows for the
// same PR number.
//
// status carries the caller's raw PR observation and its populated-ness
// flags (AC-43p): ReplaceTaskPR resolves the five outcome-attribution
// columns itself, inside this transaction, against the outgoing row it is
// about to replace (AC-43, AC-43a) — callers must not pre-resolve them.
// When no outgoing row exists (this method's ordinary production path,
// since its caller only reaches here after confirming no row exists for
// this PR number), resolution runs against a zero-value outgoing row: a
// populating observation writes what it observed, a non-populating one
// writes NULL in all five (AC-43).
//
// The DELETE+INSERT pair inside one transaction is the upsert form; an
// ON CONFLICT would also work but the per-row delete pattern matches the
// existing migration layout (rebuilds are easier to reason about) and
// avoids leaking SQLite-specific syntax into the service layer. Returns
// the row actually written, since the five resolved outcome fields may
// differ from what the caller's tp carried in.
func (s *Store) ReplaceTaskPR(ctx context.Context, tp *TaskPR, status *PRStatus) (*TaskPR, error) {
	if tp.ID == "" {
		tp.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	tp.UpdatedAt = now

	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()

	outgoing, err := replaceTaskPROutgoingRow(ctx, tx, tp)
	if err != nil {
		return nil, err
	}
	tp.IsDraft, tp.ChangedFiles, tp.MergedByLogin, tp.ClosedByLogin, tp.AutoMergeObservedAt =
		resolveTaskPROutcomeFields(outgoing, status)
	queueSource := tp
	if outgoing.ID != "" {
		queueSource = outgoing
	}
	queue := resolveTaskPRMergeQueueState(queueSource, status)
	tp.MergeQueueState, tp.MergeQueuePosition, tp.MergeQueueEstimatedTimeToMergeSeconds = queue.state, queue.position, queue.estimate
	tp.MergeQueueEntryID, tp.MergeQueueEntryHeadSHA = queue.entryID, queue.entryHeadSHA
	tp.MergeQueueLastRemovalID, tp.MergeQueueLastRemovedAt = queue.lastRemovalID, queue.lastRemovedAt
	tp.MergeQueueLastRemovalReason, tp.MergeQueueLastRemovalBeforeSHA = queue.lastRemovalReason, queue.lastRemovalBeforeSHA
	if status != nil && status.PR != nil && status.PR.HeadSHA != "" {
		tp.HeadSHA = status.PR.HeadSHA
	} else if tp.HeadSHA == "" && outgoing.ID != "" {
		tp.HeadSHA = outgoing.HeadSHA
	}
	tp.WorkflowAttention = resolveTaskPRWorkflowAttention(outgoing, status, tp.HeadSHA)
	tp.WorkflowAttentionJSON = marshalWorkflowAttention(tp.WorkflowAttention)

	if tp.RepositoryID != "" {
		if _, err := tx.ExecContext(ctx, tx.Rebind(
			`DELETE FROM github_task_prs
			 WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
			tp.TaskID, tp.RepositoryID, tp.PRNumber); err != nil {
			return nil, err
		}
	} else {
		if _, err := tx.ExecContext(ctx, tx.Rebind(
			`DELETE FROM github_task_prs
			 WHERE task_id = ? AND repository_id = '' AND pr_number = ?`),
			tp.TaskID, tp.PRNumber); err != nil {
			return nil, err
		}
	}
	if err := insertTaskPR(ctx, tx, tp); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return tp, nil
}

// replaceTaskPROutgoingRow reads, inside tx, the row ReplaceTaskPR is about
// to delete-and-replace for tp's (task_id, repository_id, pr_number), or a
// zero-value TaskPR when none exists (AC-43's "no outgoing row" case, the
// ordinary production path). Reading via tx means this observes the same
// snapshot the subsequent DELETE/INSERT commits against (AC-43a).
func replaceTaskPROutgoingRow(ctx context.Context, tx *sqlx.Tx, tp *TaskPR) (*TaskPR, error) {
	var outgoing TaskPR
	var err error
	if tp.RepositoryID != "" {
		err = tx.GetContext(ctx, &outgoing,
			tx.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs
			 WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
			tp.TaskID, tp.RepositoryID, tp.PRNumber)
	} else {
		err = tx.GetContext(ctx, &outgoing,
			tx.Rebind(`SELECT `+taskPRColumns+` FROM github_task_prs
			 WHERE task_id = ? AND repository_id = '' AND pr_number = ?`),
			tp.TaskID, tp.PRNumber)
	}
	if errors.Is(err, sql.ErrNoRows) {
		return &TaskPR{}, nil
	}
	if err != nil {
		return nil, err
	}
	if err := hydrateTaskPRWorkflowAttention(&outgoing); err != nil {
		return nil, err
	}
	return &outgoing, nil
}

// UpdateTaskPR updates a task-PR association. This is the sync writer's
// exclusive write path.
// UpdateTaskPR writes queue-removal evidence in a separate guarded statement
// so a delayed observation cannot replace a newer event already committed by
// another sync. It writes auto_merge_observed_at through a SQL-level
// COALESCE(auto_merge_observed_at, ?) rather than a direct SET. The Go-side
// latch check in resolveTaskPROutcomeFields (tp.AutoMergeObservedAt == nil)
// reads a snapshot that can be stale by the time this statement executes —
// two concurrent syncs can both observe NULL and each compute their own
// "now" timestamp. COALESCE evaluates the row's *current* value at
// UPDATE-execution time, inside this single statement, so whichever write
// actually lands first wins atomically and the second's differing timestamp
// is silently discarded instead of overwriting it (AC-16/AC-17).
func (s *Store) UpdateTaskPR(ctx context.Context, tp *TaskPR) error {
	tp.UpdatedAt = time.Now().UTC()
	removalAfter := dialect.DurationMs(s.db.DriverName(), "?", "merge_queue_last_removed_at")
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_prs SET state = ?, review_state = ?, checks_state = ?, mergeable_state = ?,
			head_sha = ?, merge_queue_state = ?, merge_queue_position = ?, merge_queue_entry_id = ?, merge_queue_entry_head_sha = ?, merge_queue_estimated_time_to_merge_seconds = ?,
			review_count = ?, pending_review_count = ?, required_reviews = ?, comment_count = ?,
			unresolved_review_threads = ?, checks_total = ?, checks_passing = ?,
			additions = ?, deletions = ?, pr_title = ?, base_branch = ?,
			merged_at = ?, closed_at = ?, last_synced_at = ?, updated_at = ?,
			is_draft = ?, changed_files = ?, merged_by_login = ?, closed_by_login = ?,
			auto_merge_observed_at = COALESCE(auto_merge_observed_at, ?), workflow_attention = ?
		WHERE id = ?`),
		tp.State, tp.ReviewState, tp.ChecksState, tp.MergeableState, tp.HeadSHA, tp.MergeQueueState, tp.MergeQueuePosition, tp.MergeQueueEntryID, tp.MergeQueueEntryHeadSHA, tp.MergeQueueEstimatedTimeToMergeSeconds,
		tp.ReviewCount, tp.PendingReviewCount, tp.RequiredReviews, tp.CommentCount,
		tp.UnresolvedReviewThreads, tp.ChecksTotal, tp.ChecksPassing,
		tp.Additions, tp.Deletions, tp.PRTitle, tp.BaseBranch,
		tp.MergedAt, tp.ClosedAt, tp.LastSyncedAt, tp.UpdatedAt,
		tp.IsDraft, tp.ChangedFiles, tp.MergedByLogin, tp.ClosedByLogin,
		tp.AutoMergeObservedAt, marshalWorkflowAttention(tp.WorkflowAttention), tp.ID); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_prs SET
			merge_queue_last_removal_id = ?, merge_queue_last_removed_at = ?,
			merge_queue_last_removal_reason = ?, merge_queue_last_removal_before_sha = ?
		WHERE id = ?
			AND ? <> ''
			AND merge_queue_last_removal_id <> ?
			AND (
				merge_queue_last_removal_id = ''
				OR merge_queue_last_removed_at IS NULL
				OR `+removalAfter+` > 0
			)`),
		tp.MergeQueueLastRemovalID, tp.MergeQueueLastRemovedAt, tp.MergeQueueLastRemovalReason,
		tp.MergeQueueLastRemovalBeforeSHA, tp.ID, tp.MergeQueueLastRemovalID,
		tp.MergeQueueLastRemovalID, tp.MergeQueueLastRemovedAt); err != nil {
		return err
	}
	return tx.Commit()
}

// --- Task CI automation operations ---

// GetTaskCIOptions returns persisted task CI automation options, or disabled defaults.
func (s *Store) GetTaskCIOptions(ctx context.Context, taskID string) (*TaskCIOptions, error) {
	var opts TaskCIOptions
	err := s.ro.GetContext(ctx, &opts,
		s.ro.Rebind(`SELECT * FROM github_task_ci_options WHERE task_id = ?`), taskID)
	if errors.Is(err, sql.ErrNoRows) {
		now := time.Now().UTC()
		return &TaskCIOptions{TaskID: taskID, CreatedAt: now, UpdatedAt: now}, nil
	}
	return &opts, err
}

// advanceTaskCIOptionsVersion advances the version carried by the complete CI
// automation payload. It creates disabled defaults for state-first tasks so a
// later WebSocket update can always be ordered against an earlier payload.
func (s *Store) advanceTaskCIOptionsVersion(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID string,
	candidate time.Time,
) (time.Time, error) {
	candidate = candidate.UTC()
	var current time.Time
	err := tx.GetContext(ctx, &current,
		tx.Rebind(`SELECT updated_at FROM github_task_ci_options WHERE task_id = ?`), taskID)
	if errors.Is(err, sql.ErrNoRows) {
		if _, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_options (
				task_id, auto_fix_prompt_override, created_at, updated_at
			) VALUES (?, NULL, ?, ?)`),
			taskID, candidate, candidate); err != nil {
			return time.Time{}, err
		}
		return candidate, nil
	}
	if err != nil {
		return time.Time{}, err
	}

	version := candidate
	if !version.After(current) {
		version = current.Add(time.Nanosecond)
	}
	if _, err := tx.ExecContext(ctx, tx.Rebind(
		`UPDATE github_task_ci_options SET updated_at = ? WHERE task_id = ?`), version, taskID); err != nil {
		return time.Time{}, err
	}
	return version, nil
}

func (s *Store) mutateTaskCIPRState(
	ctx context.Context,
	taskID string,
	mutate func(context.Context, *sqlx.Tx, time.Time) error,
) error {
	writeCtx := context.WithoutCancel(ctx)
	tx, err := s.db.BeginTxx(writeCtx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()

	version, err := s.advanceTaskCIOptionsVersion(writeCtx, tx, taskID, time.Now().UTC())
	if err != nil {
		return err
	}
	if err := mutate(writeCtx, tx, version); err != nil {
		return err
	}
	return tx.Commit()
}

// UpdateTaskCIOptions applies a partial update to task CI automation options.
func (s *Store) UpdateTaskCIOptions(ctx context.Context, taskID string, patch TaskCIOptionsPatch) (*TaskCIOptions, error) {
	writeCtx := context.WithoutCancel(ctx)
	tx, err := s.db.BeginTxx(writeCtx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()
	if err := s.updateTaskCIOptionsTx(writeCtx, tx, taskID, patch); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetTaskCIOptions(writeCtx, taskID)
}

// UpdateTaskCIOptionsWithPRAutomation applies a task update and all selected
// per-PR switch updates in one transaction. The service resolves and
// validates targets before calling this method, so a rejected identity cannot
// leave task-level fields or a subset of fan-out rows persisted.
func (s *Store) UpdateTaskCIOptionsWithPRAutomation(
	ctx context.Context,
	taskID string,
	patch TaskCIOptionsPatch,
	targets []*TaskPR,
	prPatch TaskPRAutomationOptionsPatch,
	reviewerChanged bool,
) (*TaskCIOptions, error) {
	writeCtx := context.WithoutCancel(ctx)
	tx, err := s.db.BeginTxx(writeCtx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()
	if err := s.updateTaskCIOptionsTx(writeCtx, tx, taskID, patch); err != nil {
		return nil, err
	}
	for _, target := range targets {
		if target == nil {
			continue
		}
		if err := s.updateTaskPRAutomationOptionsTx(
			writeCtx, tx, taskID, target.RepositoryID, target.PRNumber, prPatch, reviewerChanged,
		); err != nil {
			return nil, err
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetTaskCIOptions(writeCtx, taskID)
}

func (s *Store) updateTaskCIOptionsTx(
	ctx context.Context, tx *sqlx.Tx, taskID string, patch TaskCIOptionsPatch,
) error {
	now := time.Now().UTC()
	version, err := s.advanceTaskCIOptionsVersion(ctx, tx, taskID, now)
	if err != nil {
		return err
	}
	var previous TaskCIOptions
	if err := tx.GetContext(ctx, &previous,
		tx.Rebind(`SELECT * FROM github_task_ci_options WHERE task_id = ?`), taskID); err != nil {
		return err
	}
	autoFixSet, autoFixValue := boolPatchValue(patch.AutoFixEnabled)
	autoMergeSet, autoMergeValue := boolPatchValue(patch.AutoMergeEnabled)
	reviewSet, reviewValue := boolPatchValue(patch.PromptOnReviewRequested)
	mergedSet, mergedValue := boolPatchValue(patch.PromptOnMerged)
	closedSet, closedValue := boolPatchValue(patch.PromptOnClosed)
	promptSet := patch.AutoFixPromptOverride != nil
	promptValue := normalizedPromptOverride(patch.AutoFixPromptOverride)
	reviewerLoginSet := patch.ReviewReviewerLogin != nil
	reviewerChanged := reviewerLoginSet && !strings.EqualFold(
		previous.ReviewReviewerLogin, normalizedString(patch.ReviewReviewerLogin),
	)
	if _, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_ci_options SET
			auto_fix_enabled = CASE WHEN ? THEN ? ELSE auto_fix_enabled END,
			auto_merge_enabled = CASE WHEN ? THEN ? ELSE auto_merge_enabled END,
			auto_fix_prompt_override = CASE WHEN ? THEN ? ELSE auto_fix_prompt_override END,
			prompt_on_review_requested = CASE WHEN ? THEN ? ELSE prompt_on_review_requested END,
			prompt_on_merged = CASE WHEN ? THEN ? ELSE prompt_on_merged END,
			prompt_on_closed = CASE WHEN ? THEN ? ELSE prompt_on_closed END,
			review_reviewer_login = CASE WHEN ? THEN ? ELSE review_reviewer_login END,
			updated_at = ?
		WHERE task_id = ?`),
		autoFixSet, autoFixValue, autoMergeSet, autoMergeValue, promptSet, promptValue,
		reviewSet, reviewValue, mergedSet, mergedValue, closedSet, closedValue,
		reviewerLoginSet, normalizedString(patch.ReviewReviewerLogin),
		version, taskID); err != nil {
		return err
	}
	return applyTaskCIOptionResets(ctx, tx, taskID, version, previous, patch, reviewerChanged)
}

func applyTaskCIOptionResets(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID string,
	now time.Time,
	previous TaskCIOptions,
	patch TaskCIOptionsPatch,
	reviewerChanged bool,
) error {
	if shouldResetAutoFix(patch.AutoFixEnabled, previous.AutoFixEnabled) {
		if err := resetTaskCIAutoFixStateForTask(ctx, tx, taskID, now); err != nil {
			return err
		}
	}
	if shouldResetReviewRequests(
		patch.PromptOnReviewRequested, previous.PromptOnReviewRequested, reviewerChanged,
	) {
		if _, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET review_request_initialized = FALSE, last_review_requested = FALSE, updated_at = ?
			WHERE task_id = ?`), now, taskID); err != nil {
			return err
		}
	}
	if shouldResetTerminalPrompt(patch.PromptOnMerged, previous.PromptOnMerged) {
		if err := resetTaskCITerminalCheckpointForTask(ctx, tx, taskID, "merged", now); err != nil {
			return err
		}
	}
	if shouldResetTerminalPrompt(patch.PromptOnClosed, previous.PromptOnClosed) {
		return resetTaskCITerminalCheckpointForTask(ctx, tx, taskID, "closed", now)
	}
	return nil
}

func resetTaskCIAutoFixStateForTask(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID string,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_ci_pr_state
		SET auto_fix_round_count = 0,
		    last_fix_signature = '',
		    last_fix_checkpoint_json = '',
		    last_fix_enqueued_at = NULL,
		    last_fix_session_id = NULL,
		    auto_fix_attempt_state = 'acknowledged',
		    auto_fix_attempt_queue_entry_id = '',
		    auto_fix_attempt_session_id = '',
		    auto_fix_attempt_turn_id = '',
		    auto_fix_attempt_signature = '',
		    auto_fix_attempt_provider_generation = '',
		    auto_fix_attempt_outcome = '',
		    auto_fix_attempt_summary = '',
		    auto_fix_attempt_started_at = NULL,
		    auto_fix_attempt_outcome_at = NULL,
		    auto_fix_attempt_progress_deadline = NULL,
		    last_error = NULL,
		    last_error_kind = '',
		    auto_fix_exhausted_at = NULL,
		    updated_at = ?
		WHERE task_id = ?`), now, taskID)
	return err
}

func resetTaskCITerminalCheckpointForTask(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID, state string,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_ci_pr_state
		SET last_observed_pr_state = '',
		    last_lifecycle_event = '',
		    last_lifecycle_prompt_at = NULL,
		    last_lifecycle_session_id = NULL,
		    updated_at = ?
		WHERE task_id = ? AND (last_observed_pr_state = ? OR last_lifecycle_event = ?)`),
		now, taskID, state, state)
	return err
}

// GetTaskPRAutomationOptions returns one PR's automation switches, or disabled defaults.
func (s *Store) GetTaskPRAutomationOptions(
	ctx context.Context, taskID, repositoryID string, prNumber int,
) (*TaskPRAutomationOptions, error) {
	var opts TaskPRAutomationOptions
	err := s.ro.GetContext(ctx, &opts,
		s.ro.Rebind(`SELECT * FROM github_task_pr_automation_options WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		taskID, repositoryID, prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		now := time.Now().UTC()
		return &TaskPRAutomationOptions{
			TaskID: taskID, RepositoryID: repositoryID, PRNumber: prNumber,
			CreatedAt: now, UpdatedAt: now,
		}, nil
	}
	return &opts, err
}

// ListTaskPRAutomationOptions returns every stored per-PR automation row for a task.
func (s *Store) ListTaskPRAutomationOptions(ctx context.Context, taskID string) ([]*TaskPRAutomationOptions, error) {
	var rows []TaskPRAutomationOptions
	if err := s.ro.SelectContext(ctx, &rows,
		s.ro.Rebind(`SELECT * FROM github_task_pr_automation_options WHERE task_id = ? ORDER BY repository_id ASC, pr_number ASC`),
		taskID); err != nil {
		return nil, err
	}
	out := make([]*TaskPRAutomationOptions, 0, len(rows))
	for i := range rows {
		out = append(out, &rows[i])
	}
	return out, nil
}

// ListTaskPRAutomationOptionsByTaskIDs returns stored per-PR automation rows
// grouped by task. It is used by bounded task-summary hydration so a task list
// does not issue one options query for each visible row.
func (s *Store) ListTaskPRAutomationOptionsByTaskIDs(
	ctx context.Context, taskIDs []string,
) (map[string][]*TaskPRAutomationOptions, error) {
	if len(taskIDs) == 0 {
		return make(map[string][]*TaskPRAutomationOptions), nil
	}
	query, args, err := sqlx.In(
		`SELECT * FROM github_task_pr_automation_options WHERE task_id IN (?) ORDER BY task_id ASC, repository_id ASC, pr_number ASC`,
		taskIDs,
	)
	if err != nil {
		return nil, err
	}
	query = s.ro.Rebind(query)
	var rows []TaskPRAutomationOptions
	if err := s.ro.SelectContext(ctx, &rows, query, args...); err != nil {
		return nil, err
	}
	result := make(map[string][]*TaskPRAutomationOptions)
	for i := range rows {
		result[rows[i].TaskID] = append(result[rows[i].TaskID], &rows[i])
	}
	return result, nil
}

// UpdateTaskPRAutomationOptions applies a partial update to one PR's
// automation switches, upserting the row if absent. reviewerChanged mirrors
// the task-wide reviewer rebind: the caller resolves it once (comparing the
// patch's task-level ReviewReviewerLogin against the previously stored
// value) and passes it into every PR targeted by the same update, so a
// changed connected account re-baselines the review-request checkpoint even
// when the switch itself did not change value.
func (s *Store) UpdateTaskPRAutomationOptions(
	ctx context.Context, taskID, repositoryID string, prNumber int,
	patch TaskPRAutomationOptionsPatch, reviewerChanged bool,
) (*TaskPRAutomationOptions, error) {
	writeCtx := context.WithoutCancel(ctx)
	tx, err := s.db.BeginTxx(writeCtx, nil)
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback() }()
	if err := s.updateTaskPRAutomationOptionsTx(
		writeCtx, tx, taskID, repositoryID, prNumber, patch, reviewerChanged,
	); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetTaskPRAutomationOptions(writeCtx, taskID, repositoryID, prNumber)
}

func (s *Store) updateTaskPRAutomationOptionsTx(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID, repositoryID string,
	prNumber int,
	patch TaskPRAutomationOptionsPatch,
	reviewerChanged bool,
) error {
	now := time.Now().UTC()
	if _, err := tx.ExecContext(ctx, tx.Rebind(`
		INSERT INTO github_task_pr_automation_options (
			task_id, repository_id, pr_number, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?)
		ON CONFLICT(task_id, repository_id, pr_number) DO NOTHING`),
		taskID, repositoryID, prNumber, now, now); err != nil {
		return err
	}
	var previous TaskPRAutomationOptions
	if err := tx.GetContext(ctx, &previous,
		tx.Rebind(`SELECT * FROM github_task_pr_automation_options WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		taskID, repositoryID, prNumber); err != nil {
		return err
	}
	autoFixSet, autoFixValue := boolPatchValue(patch.AutoFixEnabled)
	autoMergeSet, autoMergeValue := boolPatchValue(patch.AutoMergeEnabled)
	reviewSet, reviewValue := boolPatchValue(patch.PromptOnReviewRequested)
	mergedSet, mergedValue := boolPatchValue(patch.PromptOnMerged)
	closedSet, closedValue := boolPatchValue(patch.PromptOnClosed)
	if _, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_pr_automation_options SET
			auto_fix_enabled = CASE WHEN ? THEN ? ELSE auto_fix_enabled END,
			auto_merge_enabled = CASE WHEN ? THEN ? ELSE auto_merge_enabled END,
			prompt_on_review_requested = CASE WHEN ? THEN ? ELSE prompt_on_review_requested END,
			prompt_on_merged = CASE WHEN ? THEN ? ELSE prompt_on_merged END,
			prompt_on_closed = CASE WHEN ? THEN ? ELSE prompt_on_closed END,
			updated_at = ?
		WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		autoFixSet, autoFixValue, autoMergeSet, autoMergeValue,
		reviewSet, reviewValue, mergedSet, mergedValue, closedSet, closedValue,
		now, taskID, repositoryID, prNumber); err != nil {
		return err
	}
	if err := applyTaskPRAutomationOptionResets(
		ctx, tx, taskID, repositoryID, prNumber, now, previous, patch, reviewerChanged,
	); err != nil {
		return err
	}
	return nil
}

func applyTaskPRAutomationOptionResets(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID, repositoryID string,
	prNumber int,
	now time.Time,
	previous TaskPRAutomationOptions,
	patch TaskPRAutomationOptionsPatch,
	reviewerChanged bool,
) error {
	if shouldResetAutoFix(patch.AutoFixEnabled, previous.AutoFixEnabled) {
		if err := resetTaskCIAutoFixState(ctx, tx, taskID, repositoryID, prNumber, now); err != nil {
			return err
		}
	}
	if shouldResetReviewRequests(
		patch.PromptOnReviewRequested, previous.PromptOnReviewRequested, reviewerChanged,
	) {
		if _, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET review_request_initialized = FALSE, last_review_requested = FALSE, updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
			now, taskID, repositoryID, prNumber); err != nil {
			return err
		}
	}
	if shouldResetTerminalPrompt(patch.PromptOnMerged, previous.PromptOnMerged) {
		if err := resetTaskCITerminalCheckpoint(ctx, tx, taskID, repositoryID, prNumber, "merged", now); err != nil {
			return err
		}
	}
	if shouldResetTerminalPrompt(patch.PromptOnClosed, previous.PromptOnClosed) {
		return resetTaskCITerminalCheckpoint(ctx, tx, taskID, repositoryID, prNumber, "closed", now)
	}
	return nil
}

func shouldResetAutoFix(patchValue *bool, wasEnabled bool) bool {
	return patchValue != nil && *patchValue && !wasEnabled
}

func shouldResetReviewRequests(patchValue *bool, wasEnabled, reviewerChanged bool) bool {
	return patchValue != nil && *patchValue && (!wasEnabled || reviewerChanged)
}

func shouldResetTerminalPrompt(patchValue *bool, wasEnabled bool) bool {
	return patchValue != nil && *patchValue && !wasEnabled
}

func resetTaskCIAutoFixState(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID, repositoryID string,
	prNumber int,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_ci_pr_state
		SET auto_fix_round_count = 0,
		    last_fix_signature = '',
		    last_fix_checkpoint_json = '',
		    last_fix_enqueued_at = NULL,
		    last_fix_session_id = NULL,
		    auto_fix_attempt_state = ?,
		    auto_fix_attempt_queue_entry_id = '',
		    auto_fix_attempt_session_id = '',
		    auto_fix_attempt_turn_id = '',
		    auto_fix_attempt_signature = '',
		    auto_fix_attempt_provider_generation = '',
		    auto_fix_attempt_outcome = '',
		    auto_fix_attempt_summary = '',
		    auto_fix_attempt_started_at = NULL,
		    auto_fix_attempt_outcome_at = NULL,
		    auto_fix_attempt_progress_deadline = NULL,
		    last_error = NULL,
		    last_error_kind = '',
		    auto_fix_exhausted_at = NULL,
			updated_at = ?
		WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
		string(TaskCIAutoFixAttemptAcknowledged), now, taskID, repositoryID, prNumber)
	return err
}

func resetTaskCITerminalCheckpoint(
	ctx context.Context,
	tx *sqlx.Tx,
	taskID, repositoryID string,
	prNumber int,
	state string,
	now time.Time,
) error {
	_, err := tx.ExecContext(ctx, tx.Rebind(`
		UPDATE github_task_ci_pr_state
		SET last_observed_pr_state = '',
		    last_lifecycle_event = '',
		    last_lifecycle_prompt_at = NULL,
		    last_lifecycle_session_id = NULL,
		    updated_at = ?
		WHERE task_id = ? AND repository_id = ? AND pr_number = ?
		  AND (last_observed_pr_state = ? OR last_lifecycle_event = ?)`),
		now, taskID, repositoryID, prNumber, state, state)
	return err
}

// RebindTaskPRReviewer atomically updates the task's authenticated reviewer
// login and quietly resets only its review-request baselines when it changes.
func (s *Store) RebindTaskPRReviewer(ctx context.Context, taskID, login string) (bool, error) {
	ctx = context.WithoutCancel(ctx)
	tx, err := s.db.BeginTxx(ctx, nil)
	if err != nil {
		return false, err
	}
	defer func() { _ = tx.Rollback() }()

	var current TaskCIOptions
	err = tx.GetContext(ctx, &current,
		tx.Rebind(`SELECT * FROM github_task_ci_options WHERE task_id = ?`), taskID)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	if strings.EqualFold(current.ReviewReviewerLogin, login) {
		return false, tx.Commit()
	}
	version, err := s.advanceTaskCIOptionsVersion(ctx, tx, taskID, time.Now().UTC())
	if err != nil {
		return false, err
	}
	if _, err := tx.ExecContext(ctx, tx.Rebind(
		`UPDATE github_task_ci_options SET review_reviewer_login = ?, updated_at = ? WHERE task_id = ?`), login, version, taskID); err != nil {
		return false, err
	}
	if _, err := tx.ExecContext(ctx, tx.Rebind(
		`UPDATE github_task_ci_pr_state SET review_request_initialized = FALSE, last_review_requested = FALSE, updated_at = ? WHERE task_id = ?`), version, taskID); err != nil {
		return false, err
	}
	if err := tx.Commit(); err != nil {
		return false, err
	}
	return true, nil
}

// ListTaskCIPRStates returns CI automation state rows for a task.
func (s *Store) ListTaskCIPRStates(ctx context.Context, taskID string) ([]*TaskCIPRAutomationState, error) {
	var rows []TaskCIPRAutomationState
	if err := s.ro.SelectContext(ctx, &rows,
		s.ro.Rebind(`SELECT * FROM github_task_ci_pr_state WHERE task_id = ? ORDER BY repository_id ASC, pr_number ASC`),
		taskID); err != nil {
		return nil, err
	}
	out := make([]*TaskCIPRAutomationState, 0, len(rows))
	for i := range rows {
		out = append(out, &rows[i])
	}
	return out, nil
}

// ListAllTaskCIPRStates returns every persisted PR automation state row. It is
// used only by bounded startup reconciliation; watcher evaluation remains
// task-scoped and uses ListTaskCIPRStates.
func (s *Store) ListAllTaskCIPRStates(ctx context.Context) ([]*TaskCIPRAutomationState, error) {
	var rows []TaskCIPRAutomationState
	if err := s.ro.SelectContext(ctx, &rows,
		s.ro.Rebind(`SELECT * FROM github_task_ci_pr_state ORDER BY task_id ASC, repository_id ASC, pr_number ASC`),
	); err != nil {
		return nil, err
	}
	out := make([]*TaskCIPRAutomationState, 0, len(rows))
	for i := range rows {
		out = append(out, &rows[i])
	}
	return out, nil
}

// GetTaskCIPRState returns one task/PR automation state row, or nil.
func (s *Store) GetTaskCIPRState(ctx context.Context, taskID, repositoryID string, prNumber int) (*TaskCIPRAutomationState, error) {
	var state TaskCIPRAutomationState
	err := s.ro.GetContext(ctx, &state,
		s.ro.Rebind(`SELECT * FROM github_task_ci_pr_state
		 WHERE task_id = ? AND repository_id = ? AND pr_number = ?`,
		), taskID, repositoryID, prNumber)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return &state, err
}

var (
	ErrTaskCIAutoFixAttemptNotFound = errors.New("CI auto-fix attempt not found or no longer matches")
	ErrTaskCIAutoFixOutcomeInvalid  = errors.New("invalid CI auto-fix outcome")
)

const taskCIAutoFixProviderProgressWindow = 2 * time.Minute

// RecordTaskCIFixAttempt records the feedback checkpoint that produced an
// auto-fix prompt and starts or replaces its durable attempt reservation.
func (s *Store) RecordTaskCIFixAttempt(ctx context.Context, attempt TaskCIFixAttempt) error {
	when := attempt.EnqueuedAt
	if when.IsZero() {
		when = time.Now().UTC()
	}
	attemptState := attempt.State
	if attemptState == "" {
		// Callers from before the explicit outcome protocol recorded the
		// checkpoint after dispatch completed. Treat those rows as already
		// acknowledged during the compatibility window.
		attemptState = TaskCIAutoFixAttemptAcknowledged
	}
	roundCount := 0
	if attempt.IncrementRound {
		roundCount = 1
	}
	return s.mutateTaskCIPRState(ctx, attempt.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_fix_signature, last_fix_checkpoint_json,
				last_fix_enqueued_at, last_fix_session_id, auto_fix_round_count, auto_fix_exhausted_at,
				last_queue_fix_event_id, last_queue_removal_cause,
				auto_fix_attempt_state, auto_fix_attempt_queue_entry_id,
				auto_fix_attempt_session_id, auto_fix_attempt_turn_id,
				auto_fix_attempt_signature, auto_fix_attempt_provider_generation,
				auto_fix_attempt_outcome, auto_fix_attempt_summary,
				auto_fix_attempt_started_at, auto_fix_attempt_outcome_at,
				auto_fix_attempt_progress_deadline,
				created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, '', '', ?, NULL, NULL, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_fix_signature = excluded.last_fix_signature,
				last_fix_checkpoint_json = excluded.last_fix_checkpoint_json,
				last_fix_enqueued_at = excluded.last_fix_enqueued_at,
				last_fix_session_id = excluded.last_fix_session_id,
				auto_fix_round_count = github_task_ci_pr_state.auto_fix_round_count + excluded.auto_fix_round_count,
				auto_fix_attempt_state = excluded.auto_fix_attempt_state,
				auto_fix_attempt_queue_entry_id = excluded.auto_fix_attempt_queue_entry_id,
				auto_fix_attempt_session_id = excluded.auto_fix_attempt_session_id,
				auto_fix_attempt_turn_id = excluded.auto_fix_attempt_turn_id,
				auto_fix_attempt_signature = excluded.auto_fix_attempt_signature,
				auto_fix_attempt_provider_generation = excluded.auto_fix_attempt_provider_generation,
				auto_fix_attempt_outcome = excluded.auto_fix_attempt_outcome,
				auto_fix_attempt_summary = excluded.auto_fix_attempt_summary,
				auto_fix_attempt_started_at = excluded.auto_fix_attempt_started_at,
				auto_fix_attempt_outcome_at = excluded.auto_fix_attempt_outcome_at,
				auto_fix_attempt_progress_deadline = excluded.auto_fix_attempt_progress_deadline,
				last_queue_fix_event_id = CASE
					WHEN excluded.last_queue_fix_event_id <> '' THEN excluded.last_queue_fix_event_id
					ELSE github_task_ci_pr_state.last_queue_fix_event_id END,
				last_queue_removal_cause = CASE
					WHEN excluded.last_queue_removal_cause <> '' THEN excluded.last_queue_removal_cause
					ELSE github_task_ci_pr_state.last_queue_removal_cause END,
				last_error = NULL,
				last_error_kind = '',
				updated_at = excluded.updated_at`),
			attempt.TaskID, attempt.RepositoryID, attempt.PRNumber, attempt.Signature,
			attempt.CheckpointJSON, when, nullableString(attempt.SessionID), roundCount,
			attempt.QueueRemovalEventID, attempt.QueueRemovalCause,
			string(attemptState), strings.TrimSpace(attempt.QueueEntryID), strings.TrimSpace(attempt.SessionID),
			strings.TrimSpace(attempt.TurnID), attempt.Signature, strings.TrimSpace(attempt.ProviderGeneration),
			when, now, now)
		return err
	})
}

// BindTaskCIAutoFixAttemptTurn binds a queued or direct auto-fix reservation
// to the exact turn accepted by the agent runtime. Every identity component is
// part of the compare-and-set predicate so a replaced or stale queue entry
// cannot claim a newer attempt.
func (s *Store) BindTaskCIAutoFixAttemptTurn(ctx context.Context, binding TaskCIAutoFixAttemptBinding) error {
	if strings.TrimSpace(binding.TaskID) == "" || strings.TrimSpace(binding.SessionID) == "" ||
		strings.TrimSpace(binding.Signature) == "" || strings.TrimSpace(binding.TurnID) == "" {
		return ErrTaskCIAutoFixAttemptNotFound
	}
	return s.mutateTaskCIPRState(ctx, binding.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET auto_fix_attempt_state = ?,
				auto_fix_attempt_turn_id = ?,
				updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND auto_fix_attempt_session_id = ?
			  AND auto_fix_attempt_queue_entry_id = ?
			  AND auto_fix_attempt_signature = ?
			  AND auto_fix_attempt_state IN (?, ?, ?)
			  AND (auto_fix_attempt_turn_id = '' OR auto_fix_attempt_turn_id = ? OR auto_fix_attempt_state = ?)`),
			string(TaskCIAutoFixAttemptRunning), binding.TurnID, now,
			binding.TaskID, binding.RepositoryID, binding.PRNumber,
			binding.SessionID, binding.QueueEntryID, binding.Signature,
			string(TaskCIAutoFixAttemptQueued), string(TaskCIAutoFixAttemptRunning),
			string(TaskCIAutoFixAttemptRetryable), binding.TurnID, string(TaskCIAutoFixAttemptRetryable))
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIAutoFixAttemptNotFound
		}
		return nil
	})
}

// ReportTaskCIAutoFixOutcome accepts the first explicit disposition from a
// matching auto-fix turn. The task/session/turn identity is server-owned; the
// caller never selects a repository or PR.
func (s *Store) ReportTaskCIAutoFixOutcome(ctx context.Context, report TaskCIAutoFixOutcomeReport) error {
	if !validTaskCIAutoFixOutcome(report.Outcome) {
		return ErrTaskCIAutoFixOutcomeInvalid
	}
	if strings.TrimSpace(report.TaskID) == "" || strings.TrimSpace(report.SessionID) == "" ||
		strings.TrimSpace(report.TurnID) == "" {
		return ErrTaskCIAutoFixAttemptNotFound
	}
	summary := strings.TrimSpace(report.Summary)
	if len(summary) > 4096 {
		return fmt.Errorf("CI auto-fix outcome summary exceeds 4096 bytes")
	}
	return s.mutateTaskCIPRState(ctx, report.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		state := TaskCIAutoFixAttemptAcknowledged
		deadline := (*time.Time)(nil)
		if report.Outcome == TaskCIAutoFixOutcomeActionTaken {
			state = TaskCIAutoFixAttemptAwaitingProviderProgress
			progressDeadline := now.Add(taskCIAutoFixProviderProgressWindow)
			deadline = &progressDeadline
		}
		lastError := interface{}(nil)
		lastErrorKind := ""
		if report.Outcome == TaskCIAutoFixOutcomeBlocked && summary != "" {
			lastError = summary
			lastErrorKind = TaskCIErrorKindAutoFix
		}
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET auto_fix_attempt_state = ?,
				auto_fix_attempt_outcome = ?,
				auto_fix_attempt_summary = ?,
				auto_fix_attempt_outcome_at = ?,
				auto_fix_attempt_progress_deadline = ?,
				last_error = ?,
				last_error_kind = ?,
				updated_at = ?
			WHERE task_id = ?
			  AND auto_fix_attempt_session_id = ?
			  AND auto_fix_attempt_turn_id = ?
			  AND auto_fix_attempt_state = ?
			  AND auto_fix_attempt_outcome = ''`),
			string(state), string(report.Outcome), summary, now, deadline,
			lastError, lastErrorKind, now,
			report.TaskID, report.SessionID, report.TurnID,
			string(TaskCIAutoFixAttemptRunning))
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIAutoFixAttemptNotFound
		}
		return nil
	})
}

func validTaskCIAutoFixOutcome(outcome TaskCIAutoFixOutcome) bool {
	switch outcome {
	case TaskCIAutoFixOutcomeActionTaken, TaskCIAutoFixOutcomeNonActionable, TaskCIAutoFixOutcomeBlocked:
		return true
	default:
		return false
	}
}

// ReconcileTaskCIAutoFixTurnCompletion makes a matching undispositioned turn
// retryable. It is safe to call for every turn completion: non-auto-fix turns
// and already-dispositioned attempts simply return ErrTaskCIAutoFixAttemptNotFound.
func (s *Store) ReconcileTaskCIAutoFixTurnCompletion(ctx context.Context, taskID, sessionID, turnID string) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET auto_fix_attempt_state = ?, updated_at = ?
			WHERE task_id = ? AND auto_fix_attempt_session_id = ? AND auto_fix_attempt_turn_id = ?
			  AND auto_fix_attempt_state = ? AND auto_fix_attempt_outcome = ''`),
			string(TaskCIAutoFixAttemptRetryable), now, taskID, sessionID, turnID,
			string(TaskCIAutoFixAttemptRunning))
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIAutoFixAttemptNotFound
		}
		return nil
	})
}

// ReconcileTaskCIAutoFixQueuedDispatchFailure releases a queued reservation
// when the queue worker cannot reach the agent acceptance boundary.
func (s *Store) ReconcileTaskCIAutoFixQueuedDispatchFailure(ctx context.Context, binding TaskCIAutoFixAttemptBinding) error {
	return s.mutateTaskCIPRState(ctx, binding.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET auto_fix_attempt_state = ?, updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND auto_fix_attempt_session_id = ?
			  AND auto_fix_attempt_queue_entry_id = ?
			  AND auto_fix_attempt_signature = ?
			  AND auto_fix_attempt_state = ?`),
			string(TaskCIAutoFixAttemptRetryable), now,
			binding.TaskID, binding.RepositoryID, binding.PRNumber,
			binding.SessionID, binding.QueueEntryID, binding.Signature,
			string(TaskCIAutoFixAttemptQueued))
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIAutoFixAttemptNotFound
		}
		return nil
	})
}

// ReconcileTaskCIAutoFixProviderProgress acknowledges action_taken after a
// provider generation changes, or makes it retryable after the bounded
// progress deadline expires.
func (s *Store) ReconcileTaskCIAutoFixProviderProgress(ctx context.Context, progress TaskCIAutoFixProviderProgress) error {
	observedAt := progress.ObservedAt
	if observedAt.IsZero() {
		observedAt = time.Now().UTC()
	}
	return s.mutateTaskCIPRState(ctx, progress.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET auto_fix_attempt_state = CASE
					WHEN auto_fix_attempt_provider_generation <> ? THEN ?
					WHEN auto_fix_attempt_progress_deadline <= ? THEN ?
					ELSE auto_fix_attempt_state END,
				auto_fix_attempt_progress_deadline = NULL,
				updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND auto_fix_attempt_signature = ?
			  AND auto_fix_attempt_state = ?
			  AND (
				auto_fix_attempt_provider_generation <> ?
				OR auto_fix_attempt_progress_deadline <= ?
			  )`),
			progress.ProviderGeneration, string(TaskCIAutoFixAttemptAcknowledged), observedAt,
			string(TaskCIAutoFixAttemptRetryable), now,
			progress.TaskID, progress.RepositoryID, progress.PRNumber, progress.Signature,
			string(TaskCIAutoFixAttemptAwaitingProviderProgress),
			progress.ProviderGeneration, observedAt)
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIAutoFixAttemptNotFound
		}
		return nil
	})
}

// RefreshTaskCIFixCheckpoint updates the current feedback checkpoint without recording a new prompt dispatch.
func (s *Store) RefreshTaskCIFixCheckpoint(ctx context.Context, taskID, repositoryID string, prNumber int, signature, checkpointJSON string) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_fix_signature, last_fix_checkpoint_json, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_fix_signature = excluded.last_fix_signature,
				last_fix_checkpoint_json = excluded.last_fix_checkpoint_json,
				last_fix_enqueued_at = NULL,
				last_error = CASE
					WHEN auto_fix_attempt_state = ?
					 AND auto_fix_attempt_outcome = ?
					 AND last_fix_signature = ? THEN last_error
					ELSE NULL END,
				last_error_kind = CASE
					WHEN auto_fix_attempt_state = ?
					 AND auto_fix_attempt_outcome = ?
					 AND last_fix_signature = ? THEN last_error_kind
					ELSE '' END,
				updated_at = excluded.updated_at`),
			taskID, repositoryID, prNumber, signature, checkpointJSON, now, now,
			string(TaskCIAutoFixAttemptAcknowledged), string(TaskCIAutoFixOutcomeBlocked), signature,
			string(TaskCIAutoFixAttemptAcknowledged), string(TaskCIAutoFixOutcomeBlocked), signature)
		return err
	})
}

var ErrTaskCIMergeAttemptAlreadyReserved = errors.New("CI auto-merge attempt already reserved")
var ErrTaskCIMergeAttemptNotFound = errors.New("CI auto-merge attempt not found")
var ErrTaskCIMergeRetryNotAllowed = errors.New("CI auto-merge retry is not available")

const taskCIMergeRetryInFlightTTL = 2 * time.Minute

// AuthorizeTaskCIMergeRetry persists one single-use authorization to
// reevaluate a failed or expired automatic merge attempt.
func (s *Store) AuthorizeTaskCIMergeRetry(
	ctx context.Context, taskID, repositoryID string, prNumber int, requestedAt time.Time,
) error {
	if requestedAt.IsZero() {
		requestedAt = time.Now().UTC()
	}
	expiredBefore := requestedAt.Add(-taskCIMergeRetryInFlightTTL)
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state SET
				merge_retry_pending = TRUE,
				updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND merge_retry_pending = FALSE
			  AND (
				(last_merge_result = ? AND last_error_kind = ?)
				OR (last_merge_result = ? AND last_merge_attempt_at <= ?)
			  )`),
			now, taskID, repositoryID, prNumber,
			TaskCIMergeResultFailed, TaskCIErrorKindAutoMerge,
			TaskCIMergeResultInFlight, expiredBefore)
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIMergeRetryNotAllowed
		}
		return nil
	})
}

// ClearTaskCIMergeRetryAuthorization rolls back a one-shot retry authorization
// when its follow-up event could not be published. This keeps a failed publish
// from leaving the PR permanently blocked behind merge_retry_pending.
func (s *Store) ClearTaskCIMergeRetryAuthorization(
	ctx context.Context, taskID, repositoryID string, prNumber int,
) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET merge_retry_pending = FALSE, updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND merge_retry_pending = TRUE`),
			now, taskID, repositoryID, prNumber)
		return err
	})
}

// RecordTaskCIMergeAttempt reserves an auto-merge attempt signature before
// the provider side effect. An unchanged signature cannot replace an existing
// reservation or terminal result.
func (s *Store) RecordTaskCIMergeAttempt(ctx context.Context, attempt TaskCIMergeAttempt) error {
	when := attempt.AttemptedAt
	if when.IsZero() {
		when = time.Now().UTC()
	}
	return s.mutateTaskCIPRState(ctx, attempt.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		result, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_merge_signature, last_merge_attempt_at,
				last_merge_result, merge_retry_pending, last_queue_attempt_head_sha, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, FALSE, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_merge_signature = excluded.last_merge_signature,
				last_merge_attempt_at = excluded.last_merge_attempt_at,
				last_merge_result = excluded.last_merge_result,
				merge_retry_pending = FALSE,
				last_queue_attempt_head_sha = CASE
					WHEN excluded.last_queue_attempt_head_sha <> '' THEN excluded.last_queue_attempt_head_sha
					ELSE github_task_ci_pr_state.last_queue_attempt_head_sha END,
				last_error = CASE WHEN github_task_ci_pr_state.last_error_kind = ?
					THEN NULL ELSE github_task_ci_pr_state.last_error END,
				last_error_kind = CASE WHEN github_task_ci_pr_state.last_error_kind = ?
					THEN '' ELSE github_task_ci_pr_state.last_error_kind END,
				updated_at = excluded.updated_at
			WHERE github_task_ci_pr_state.last_merge_signature <> excluded.last_merge_signature
			   OR github_task_ci_pr_state.merge_retry_pending = TRUE`),
			attempt.TaskID, attempt.RepositoryID, attempt.PRNumber, attempt.Signature, when,
			TaskCIMergeResultInFlight, attempt.AttemptedHeadSHA, now, now,
			TaskCIErrorKindAutoMerge, TaskCIErrorKindAutoMerge)
		if err != nil {
			return err
		}
		rows, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIMergeAttemptAlreadyReserved
		}
		return nil
	})
}

// RecordTaskCIMergeAttemptResult completes a reserved attempt and updates its
// typed error in the same transaction.
func (s *Store) RecordTaskCIMergeAttemptResult(
	ctx context.Context,
	taskID, repositoryID string,
	prNumber int,
	signature, result, message string,
) error {
	if result != TaskCIMergeResultFailed && result != TaskCIMergeResultAccepted {
		return fmt.Errorf("invalid CI auto-merge result %q", result)
	}
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		outcome, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state SET
				last_merge_result = ?,
				last_error = CASE
					WHEN ? = ? THEN ?
					WHEN ? = ? AND last_error_kind = ? THEN NULL
					ELSE last_error END,
				last_error_kind = CASE
					WHEN ? = ? THEN ?
					WHEN ? = ? AND last_error_kind = ? THEN ''
					ELSE last_error_kind END,
				updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND last_merge_signature = ?`),
			result,
			result, TaskCIMergeResultFailed, strings.TrimSpace(message),
			result, TaskCIMergeResultAccepted, TaskCIErrorKindAutoMerge,
			result, TaskCIMergeResultFailed, TaskCIErrorKindAutoMerge,
			result, TaskCIMergeResultAccepted, TaskCIErrorKindAutoMerge,
			now, taskID, repositoryID, prNumber, signature)
		if err != nil {
			return err
		}
		rows, err := outcome.RowsAffected()
		if err != nil {
			return err
		}
		if rows == 0 {
			return ErrTaskCIMergeAttemptNotFound
		}
		return nil
	})
}

// RecordTaskCIMergeQueueObservation persists an active queue attempt or a
// conservative current-head baseline when a removal is observed first. The
// baseline is written only when no queue attempt has been recorded yet, so a
// later poll cannot move the automatic requeue guard to a newer head. The
// passive baseline is not auto-fix provenance; queue-removal auto-fix also
// requires the durable merge-attempt signature.
func (s *Store) RecordTaskCIMergeQueueObservation(ctx context.Context, observation TaskCIMergeQueueObservation) error {
	return s.mutateTaskCIPRState(ctx, observation.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		baselineHead := ""
		mergeResult := ""
		if observation.ActiveQueueHeadSHA != "" || observation.Accepted {
			mergeResult = TaskCIMergeResultAccepted
		}
		if observation.ActiveQueueHeadSHA == "" {
			baselineHead = observation.RemovalObservedHeadSHA
		}
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_merge_signature, last_merge_result,
				last_queue_attempt_head_sha, last_queue_removal_cause, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_merge_signature = CASE
					WHEN excluded.last_merge_signature <> '' THEN excluded.last_merge_signature
					ELSE github_task_ci_pr_state.last_merge_signature END,
				last_merge_result = CASE
					WHEN excluded.last_merge_result <> '' THEN excluded.last_merge_result
					ELSE github_task_ci_pr_state.last_merge_result END,
				merge_retry_pending = CASE
					WHEN excluded.last_merge_result = ? THEN FALSE
					ELSE github_task_ci_pr_state.merge_retry_pending END,
				last_queue_attempt_head_sha = CASE
					WHEN excluded.last_queue_attempt_head_sha <> ''
						THEN excluded.last_queue_attempt_head_sha
					ELSE github_task_ci_pr_state.last_queue_attempt_head_sha END,
				last_queue_removal_cause = CASE
					WHEN excluded.last_queue_removal_cause <> ''
						THEN excluded.last_queue_removal_cause
					ELSE github_task_ci_pr_state.last_queue_removal_cause END,
				last_error = CASE
					WHEN excluded.last_merge_result = ? AND github_task_ci_pr_state.last_error_kind = ?
						THEN NULL ELSE github_task_ci_pr_state.last_error END,
				last_error_kind = CASE
					WHEN excluded.last_merge_result = ? AND github_task_ci_pr_state.last_error_kind = ?
						THEN '' ELSE github_task_ci_pr_state.last_error_kind END,
				updated_at = excluded.updated_at`),
			observation.TaskID, observation.RepositoryID, observation.PRNumber,
			observation.MergeSignature, mergeResult,
			observation.ActiveQueueHeadSHA, observation.RemovalCause, now, now,
			TaskCIMergeResultAccepted,
			TaskCIMergeResultAccepted, TaskCIErrorKindAutoMerge,
			TaskCIMergeResultAccepted, TaskCIErrorKindAutoMerge)
		if err != nil {
			return err
		}
		if baselineHead == "" {
			return nil
		}
		_, err = tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state
			SET last_queue_attempt_head_sha = ?, updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?
			  AND last_queue_attempt_head_sha = ''`),
			baselineHead, now, observation.TaskID, observation.RepositoryID, observation.PRNumber)
		return err
	})
}

// RecordTaskCIError stores the latest user-visible CI automation error for a task PR.
func (s *Store) RecordTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error {
	return s.recordTaskCIError(ctx, taskID, repositoryID, prNumber, "", message)
}

// RecordTaskCIAutoMergeError stores an error produced by the automatic merge path.
func (s *Store) RecordTaskCIAutoMergeError(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error {
	return s.recordTaskCIError(ctx, taskID, repositoryID, prNumber, TaskCIErrorKindAutoMerge, message)
}

func (s *Store) recordTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int, kind, message string) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_error, last_error_kind, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_error = excluded.last_error,
				last_error_kind = excluded.last_error_kind,
				updated_at = excluded.updated_at`),
			taskID, repositoryID, prNumber, strings.TrimSpace(message), kind, now, now)
		return err
	})
}

// MarkTaskCIAutoFixExhausted records that auto-fix reached its per-PR round cap.
func (s *Store) MarkTaskCIAutoFixExhausted(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, auto_fix_exhausted_at, last_error, last_error_kind, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				auto_fix_exhausted_at = excluded.auto_fix_exhausted_at,
				last_error = excluded.last_error,
				last_error_kind = excluded.last_error_kind,
				updated_at = excluded.updated_at`),
			taskID, repositoryID, prNumber, now, strings.TrimSpace(message), TaskCIErrorKindAutoFix, now, now)
		return err
	})
}

// ClearTaskCIError clears the latest CI automation error for a task PR.
func (s *Store) ClearTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			UPDATE github_task_ci_pr_state SET last_error = NULL, last_error_kind = '', updated_at = ?
			WHERE task_id = ? AND repository_id = ? AND pr_number = ?`),
			now, taskID, repositoryID, prNumber)
		return err
	})
}

// SetTaskPRReviewRequestState records a complete reviewer-request observation.
func (s *Store) SetTaskPRReviewRequestState(
	ctx context.Context, taskID, repositoryID string, prNumber int, requested bool,
) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, review_request_initialized,
				last_review_requested, created_at, updated_at
			) VALUES (?, ?, ?, TRUE, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				review_request_initialized = TRUE,
				last_review_requested = excluded.last_review_requested,
				updated_at = excluded.updated_at`),
			taskID, repositoryID, prNumber, requested, now, now)
		return err
	})
}

// SetTaskPRObservedState records the current PR state used to detect terminal entry.
func (s *Store) SetTaskPRObservedState(
	ctx context.Context, taskID, repositoryID string, prNumber int, state string,
) error {
	return s.mutateTaskCIPRState(ctx, taskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, last_observed_pr_state, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				last_observed_pr_state = excluded.last_observed_pr_state,
				last_lifecycle_event = CASE
					WHEN excluded.last_observed_pr_state IN ('merged', 'closed')
					THEN github_task_ci_pr_state.last_lifecycle_event
					ELSE '' END,
				updated_at = excluded.updated_at`),
			taskID, repositoryID, prNumber, state, now, now)
		return err
	})
}

// RecordTaskPRLifecyclePrompt stamps an accepted or durably queued lifecycle prompt.
func (s *Store) RecordTaskPRLifecyclePrompt(ctx context.Context, prompt TaskPRLifecyclePrompt) error {
	when := prompt.PromptedAt
	if when.IsZero() {
		when = time.Now().UTC()
	}
	return s.mutateTaskCIPRState(ctx, prompt.TaskID, func(ctx context.Context, tx *sqlx.Tx, now time.Time) error {
		_, err := tx.ExecContext(ctx, tx.Rebind(`
			INSERT INTO github_task_ci_pr_state (
				task_id, repository_id, pr_number, review_request_initialized,
				last_review_requested, last_observed_pr_state, last_lifecycle_event,
				last_lifecycle_prompt_at, last_lifecycle_session_id, created_at, updated_at
			) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
			ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
				review_request_initialized = CASE
					WHEN excluded.last_lifecycle_event = 'review_requested' THEN TRUE
					ELSE github_task_ci_pr_state.review_request_initialized END,
				last_review_requested = CASE
					WHEN excluded.last_lifecycle_event = 'review_requested' THEN excluded.last_review_requested
					ELSE github_task_ci_pr_state.last_review_requested END,
				last_observed_pr_state = CASE
					WHEN excluded.last_observed_pr_state <> '' THEN excluded.last_observed_pr_state
					ELSE github_task_ci_pr_state.last_observed_pr_state END,
				last_lifecycle_event = excluded.last_lifecycle_event,
				last_lifecycle_prompt_at = excluded.last_lifecycle_prompt_at,
				last_lifecycle_session_id = excluded.last_lifecycle_session_id,
				last_error = NULL,
				last_error_kind = '',
				updated_at = excluded.updated_at`),
			prompt.TaskID, prompt.RepositoryID, prompt.PRNumber,
			prompt.Event == "review_requested", prompt.ReviewRequested,
			prompt.ObservedState, prompt.Event, when, nullableString(prompt.SessionID), now, now)
		return err
	})
}

func nullableString(value string) *string {
	if value == "" {
		return nil
	}
	return &value
}

func boolPatchValue(value *bool) (bool, bool) {
	if value == nil {
		return false, false
	}
	return true, *value
}

func normalizedPromptOverride(value *string) *string {
	if value == nil {
		return nil
	}
	trimmed := strings.TrimSpace(*value)
	if trimmed == "" {
		return nil
	}
	return &trimmed
}

func normalizedString(value *string) string {
	if value == nil {
		return ""
	}
	return strings.TrimSpace(*value)
}

// --- Review Watch operations ---

// CreateReviewWatch creates a new review watch configuration.
func (s *Store) CreateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
	if rw.ID == "" {
		rw.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	rw.CreatedAt = now
	rw.UpdatedAt = now
	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
	reposJSON, err := json.Marshal(rw.Repos)
	if err != nil {
		return fmt.Errorf("marshal repos: %w", err)
	}
	rw.ReposJSON = string(reposJSON)
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_review_watches (id, workspace_id, workflow_id, workflow_step_id, repos,
			agent_profile_id, executor_profile_id, prompt, review_scope, custom_query, target_login,
			enabled, poll_interval_seconds, cleanup_policy, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
		rw.ID, rw.WorkspaceID, rw.WorkflowID, rw.WorkflowStepID, rw.ReposJSON,
		rw.AgentProfileID, rw.ExecutorProfileID, rw.Prompt, rw.ReviewScope, rw.CustomQuery, rw.TargetLogin,
		rw.Enabled, rw.PollIntervalSeconds, rw.CleanupPolicy, rw.CreatedAt, rw.UpdatedAt)
	return err
}

// hydrateReviewWatchRepos unmarshals the ReposJSON field into the Repos slice
// and normalizes the cleanup policy so legacy rows (or zero values) surface
// as the documented default.
func hydrateReviewWatchRepos(rw *ReviewWatch) {
	if rw.ReposJSON != "" {
		if err := json.Unmarshal([]byte(rw.ReposJSON), &rw.Repos); err != nil {
			// Log but don't fail — the watch can still function with no repo filter.
			fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal repos JSON for review watch %s: %v\n", rw.ID, err)
		}
	}
	if rw.Repos == nil {
		rw.Repos = []RepoFilter{}
	}
	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
}

// GetReviewWatch returns a review watch by ID.
func (s *Store) GetReviewWatch(ctx context.Context, id string) (*ReviewWatch, error) {
	var rw ReviewWatch
	err := s.ro.GetContext(ctx, &rw, s.ro.Rebind(`SELECT * FROM github_review_watches WHERE id = ?`), id)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	hydrateReviewWatchRepos(&rw)
	return &rw, nil
}

// ListReviewWatches returns all review watches for a workspace.
func (s *Store) ListReviewWatches(ctx context.Context, workspaceID string) ([]*ReviewWatch, error) {
	var watches []*ReviewWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_review_watches WHERE workspace_id = ? ORDER BY created_at`), workspaceID)
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateReviewWatchRepos(w)
	}
	return watches, nil
}

// ListAllReviewWatches returns every review watch across all workspaces. Used
// by the install-wide settings UI when no workspace filter is supplied.
func (s *Store) ListAllReviewWatches(ctx context.Context) ([]*ReviewWatch, error) {
	var watches []*ReviewWatch
	err := s.ro.SelectContext(ctx, &watches,
		`SELECT * FROM github_review_watches ORDER BY workspace_id, created_at`)
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateReviewWatchRepos(w)
	}
	return watches, nil
}

// ListEnabledReviewWatches returns all enabled review watches.
func (s *Store) ListEnabledReviewWatches(ctx context.Context) ([]*ReviewWatch, error) {
	var watches []*ReviewWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_review_watches WHERE enabled = TRUE ORDER BY created_at`))
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateReviewWatchRepos(w)
	}
	return watches, nil
}

// UpdateReviewWatch updates a review watch.
func (s *Store) UpdateReviewWatch(ctx context.Context, rw *ReviewWatch) error {
	rw.UpdatedAt = time.Now().UTC()
	rw.CleanupPolicy = NormalizeCleanupPolicy(rw.CleanupPolicy)
	reposJSON, err := json.Marshal(rw.Repos)
	if err != nil {
		return fmt.Errorf("marshal repos: %w", err)
	}
	rw.ReposJSON = string(reposJSON)
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_review_watches SET workflow_id = ?, workflow_step_id = ?, repos = ?,
			agent_profile_id = ?, executor_profile_id = ?,
			prompt = ?, review_scope = ?, custom_query = ?, target_login = ?,
			enabled = ?, poll_interval_seconds = ?, cleanup_policy = ?, last_polled_at = ?, updated_at = ?
		WHERE id = ?`),
		rw.WorkflowID, rw.WorkflowStepID, rw.ReposJSON,
		rw.AgentProfileID, rw.ExecutorProfileID,
		rw.Prompt, rw.ReviewScope, rw.CustomQuery, rw.TargetLogin,
		rw.Enabled, rw.PollIntervalSeconds, rw.CleanupPolicy, rw.LastPolledAt, rw.UpdatedAt, rw.ID)
	return err
}

// DeleteReviewWatch deletes a review watch and all its associated dedup rows
// in one transaction. Dedup rows have no foreign key (SQLite never enforced
// one for this table), so the explicit cascade is required — otherwise the
// rows survive after the watch is gone, become invisible to the per-watch
// poller, and the tasks they reference leak forever.
func (s *Store) DeleteReviewWatch(ctx context.Context, id string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_review_pr_tasks WHERE review_watch_id = ?`), id); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_review_watches WHERE id = ?`), id); err != nil {
		return err
	}
	return tx.Commit()
}

// DisableReviewWatchWithError is the self-heal write: it disables the watch
// and stamps a human-readable cause + timestamp so the settings UI can show
// a "disabled because ..." banner. Called by the orchestrator when the
// watcher's bound agent profile is detected as soft-deleted.
func (s *Store) DisableReviewWatchWithError(ctx context.Context, id, cause string) error {
	now := time.Now().UTC()
	_, err := s.db.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_review_watches
		   SET enabled = FALSE, last_error = ?, last_error_at = ?, updated_at = ?
		 WHERE id = ?`),
		cause, now, now, id)
	return err
}

// --- Review PR Task deduplication ---

// CreateReviewPRTask records that a task was created for a review PR.
func (s *Store) CreateReviewPRTask(ctx context.Context, rpt *ReviewPRTask) error {
	if rpt.ID == "" {
		rpt.ID = uuid.New().String()
	}
	rpt.CreatedAt = time.Now().UTC()
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_review_pr_tasks (id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)`),
		rpt.ID, rpt.ReviewWatchID, rpt.RepoOwner, rpt.RepoName, rpt.PRNumber, rpt.PRURL, rpt.TaskID, rpt.CreatedAt)
	return err
}

// HasReviewPRTask checks if a task was already created for a PR in a review watch.
func (s *Store) HasReviewPRTask(ctx context.Context, reviewWatchID, repoOwner, repoName string, prNumber int) (bool, error) {
	var count int
	err := s.ro.GetContext(ctx, &count, s.ro.Rebind(
		`SELECT COUNT(*) FROM github_review_pr_tasks WHERE review_watch_id = ? AND repo_owner = ? AND repo_name = ? AND pr_number = ?`),
		reviewWatchID, repoOwner, repoName, prNumber)
	return count > 0, err
}

// ReserveReviewPRTask atomically claims a slot for a (watch, repo, PR) tuple
// using the UNIQUE constraint. Returns true if this
// caller won the race and should proceed to create the task, false if another
// caller already holds the slot. The caller is expected to call
// AssignReviewPRTaskID once the task is created, or ReleaseReviewPRTask if
// task creation fails.
func (s *Store) ReserveReviewPRTask(ctx context.Context, reviewWatchID, repoOwner, repoName string, prNumber int, prURL string) (bool, error) {
	res, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_review_pr_tasks (id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(review_watch_id, repo_owner, repo_name, pr_number) DO NOTHING`),
		uuid.New().String(), reviewWatchID, repoOwner, repoName, prNumber, prURL, "", time.Now().UTC())
	if err != nil {
		return false, err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return rows == 1, nil
}

// AssignReviewPRTaskID sets the task_id on a reserved dedup row. Called after
// the task has been created so cleanup logic can locate and delete it later.
// Returns an error if no row was updated, which surfaces the narrow race where
// the reservation was removed (e.g. by a concurrent cleanup sweep) between
// Reserve and Assign — otherwise the task would leak with no dedup record.
func (s *Store) AssignReviewPRTaskID(ctx context.Context, reviewWatchID, repoOwner, repoName string, prNumber int, taskID string) error {
	res, err := s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_review_pr_tasks SET task_id = ?
		WHERE review_watch_id = ? AND repo_owner = ? AND repo_name = ? AND pr_number = ?`),
		taskID, reviewWatchID, repoOwner, repoName, prNumber)
	if err != nil {
		return err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if rows == 0 {
		return fmt.Errorf("assign task ID: reservation row not found for watch=%s pr=%d", reviewWatchID, prNumber)
	}
	return nil
}

// ReleaseReviewPRTask removes a reservation for a (watch, repo, PR) tuple.
// Used when task creation fails so a later poll can retry instead of the PR
// being permanently blocked by an orphan reservation.
func (s *Store) ReleaseReviewPRTask(ctx context.Context, reviewWatchID, repoOwner, repoName string, prNumber int) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		DELETE FROM github_review_pr_tasks
		WHERE review_watch_id = ? AND repo_owner = ? AND repo_name = ? AND pr_number = ?`),
		reviewWatchID, repoOwner, repoName, prNumber)
	return err
}

// ListReviewPRTasksByWatch lists all dedup records for a given review watch.
func (s *Store) ListReviewPRTasksByWatch(ctx context.Context, watchID string) ([]*ReviewPRTask, error) {
	var tasks []*ReviewPRTask
	err := s.ro.SelectContext(ctx, &tasks, s.ro.Rebind(
		`SELECT id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at
		 FROM github_review_pr_tasks WHERE review_watch_id = ?`), watchID)
	return tasks, err
}

// ListAllReviewPRTasks lists every dedup record across all watches. Used by
// the global cleanup sweep so orphaned rows (whose watch was deleted or
// disabled) still get evaluated for terminal-state cleanup.
func (s *Store) ListAllReviewPRTasks(ctx context.Context) ([]*ReviewPRTask, error) {
	var tasks []*ReviewPRTask
	err := s.ro.SelectContext(ctx, &tasks,
		`SELECT id, review_watch_id, repo_owner, repo_name, pr_number, pr_url, task_id, created_at
		 FROM github_review_pr_tasks`)
	return tasks, err
}

// ListScheduledReviewPRTasksByWatch lists review-task records that routine
// cleanup may inspect. Archived tasks stay in the complete inventory but do
// not spend provider requests on each poll. Empty reservations and records for
// hard-deleted tasks remain eligible for recovery.
func (s *Store) ListScheduledReviewPRTasksByWatch(ctx context.Context, watchID string) ([]*ReviewPRTask, error) {
	var tasks []*ReviewPRTask
	err := s.ro.SelectContext(ctx, &tasks, s.ro.Rebind(`
		SELECT rpt.id, rpt.review_watch_id, rpt.repo_owner, rpt.repo_name, rpt.pr_number,
		       rpt.pr_url, rpt.task_id, rpt.created_at
		FROM github_review_pr_tasks rpt
		LEFT JOIN tasks t ON t.id = rpt.task_id
		WHERE rpt.review_watch_id = ?
		  AND (rpt.task_id = '' OR t.id IS NULL OR t.archived_at IS NULL)`), watchID)
	return tasks, err
}

// ListScheduledReviewPRTasks lists routine-cleanup candidates across all
// watches, preserving reservations and records for hard-deleted tasks.
func (s *Store) ListScheduledReviewPRTasks(ctx context.Context) ([]*ReviewPRTask, error) {
	var tasks []*ReviewPRTask
	err := s.ro.SelectContext(ctx, &tasks, `
		SELECT rpt.id, rpt.review_watch_id, rpt.repo_owner, rpt.repo_name, rpt.pr_number,
		       rpt.pr_url, rpt.task_id, rpt.created_at
		FROM github_review_pr_tasks rpt
		LEFT JOIN tasks t ON t.id = rpt.task_id
		WHERE rpt.task_id = '' OR t.id IS NULL OR t.archived_at IS NULL`)
	return tasks, err
}

// DeleteReviewPRTask deletes a dedup record by ID.
func (s *Store) DeleteReviewPRTask(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_review_pr_tasks WHERE id = ?`), id)
	return err
}

// ListReviewPRTaskIDsByWatch returns every task_id recorded against a
// review watch, including empty-string reservations. Used by the watch
// reset flow to enumerate the tasks to cascade-delete.
func (s *Store) ListReviewPRTaskIDsByWatch(ctx context.Context, watchID string) ([]string, error) {
	var ids []string
	err := s.ro.SelectContext(ctx, &ids,
		s.ro.Rebind(`SELECT task_id FROM github_review_pr_tasks WHERE review_watch_id = ?`), watchID)
	return ids, err
}

// ResetReviewWatchState wipes a review watch's dedup rows and nulls its
// last_polled_at in a single transaction. Used by the reset flow after
// the cascade-delete loop so the next poll re-imports every currently
// matching PR as if the watch were freshly created.
func (s *Store) ResetReviewWatchState(ctx context.Context, watchID string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`DELETE FROM github_review_pr_tasks WHERE review_watch_id = ?`), watchID); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_review_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`),
		time.Now().UTC(), watchID); err != nil {
		return err
	}
	return tx.Commit()
}

// --- Stats queries ---

// prStatsQuery builds parameterised SELECT queries against the github_task_prs table.
type prStatsQuery struct {
	from  string
	where string
	args  []interface{}
}

func newPRStatsQuery(req *PRStatsRequest) *prStatsQuery {
	q := &prStatsQuery{
		from:  "github_task_prs gtp",
		where: "1=1",
	}
	if req.WorkspaceID != "" {
		q.from += " INNER JOIN tasks t ON gtp.task_id = t.id"
		q.where += " AND t.workspace_id = ?"
		q.args = append(q.args, req.WorkspaceID)
	}
	if req.StartDate != nil {
		q.where += " AND gtp.created_at >= ?"
		q.args = append(q.args, req.StartDate)
	}
	if req.EndDate != nil {
		q.where += " AND gtp.created_at <= ?"
		q.args = append(q.args, req.EndDate)
	}
	return q
}

func (q *prStatsQuery) build(sel, extraWhere string) string {
	w := q.where
	if extraWhere != "" {
		w += " AND " + extraWhere
	}
	return fmt.Sprintf(`SELECT %s FROM %s WHERE %s`, sel, q.from, w)
}

// GetPRStats returns aggregated PR statistics.
func (s *Store) GetPRStats(ctx context.Context, req *PRStatsRequest) (*PRStats, error) {
	return s.runPRStatsQueries(ctx, newPRStatsQuery(req))
}

func (s *Store) runPRStatsQueries(ctx context.Context, q *prStatsQuery) (*PRStats, error) {
	stats := &PRStats{}

	if err := s.ro.GetContext(ctx, &stats.TotalPRsCreated, s.ro.Rebind(q.build("COUNT(*)", "")), q.args...); err != nil {
		return nil, err
	}
	if err := s.ro.GetContext(ctx, &stats.TotalComments,
		s.ro.Rebind(q.build("COALESCE(SUM(gtp.comment_count), 0)", "")), q.args...); err != nil {
		return nil, err
	}
	if err := s.fetchCIPassRate(ctx, q, stats); err != nil {
		return nil, err
	}
	if err := s.fetchApprovalRate(ctx, q, stats); err != nil {
		return nil, err
	}

	var avgMerge sql.NullFloat64
	avgDuration := fmt.Sprintf("(%s) / 3600000.0", dialect.DurationMs(s.ro.DriverName(), "gtp.merged_at", "gtp.created_at"))
	avgQ := q.build("AVG("+avgDuration+")", "gtp.merged_at IS NOT NULL")
	if err := s.ro.GetContext(ctx, &avgMerge, s.ro.Rebind(avgQ), q.args...); err != nil {
		return nil, err
	}
	if avgMerge.Valid {
		stats.AvgTimeToMergeHours = avgMerge.Float64
	}

	dailyDate := dialect.DateOf(s.ro.DriverName(), "gtp.created_at")
	dailyQ := q.build(dailyDate+" as date, COUNT(*) as count", "") +
		" GROUP BY " + dailyDate + " ORDER BY date"
	if err := s.ro.SelectContext(ctx, &stats.PRsByDay, s.ro.Rebind(dailyQ), q.args...); err != nil {
		return nil, err
	}
	return stats, nil
}

func (s *Store) fetchCIPassRate(ctx context.Context, q *prStatsQuery, stats *PRStats) error {
	var totalWithChecks, passed int
	if err := s.ro.GetContext(ctx, &totalWithChecks,
		s.ro.Rebind(q.build("COUNT(*)", "gtp.checks_state != ''")), q.args...); err != nil {
		return err
	}
	if err := s.ro.GetContext(ctx, &passed,
		s.ro.Rebind(q.build("COUNT(*)", "gtp.checks_state = 'success'")), q.args...); err != nil {
		return err
	}
	if totalWithChecks > 0 {
		stats.CIPassRate = float64(passed) / float64(totalWithChecks)
	}
	return nil
}

func (s *Store) fetchApprovalRate(ctx context.Context, q *prStatsQuery, stats *PRStats) error {
	var totalReviewed, approved int
	if err := s.ro.GetContext(ctx, &totalReviewed,
		s.ro.Rebind(q.build("COUNT(*)", "gtp.review_state != ''")), q.args...); err != nil {
		return err
	}
	if err := s.ro.GetContext(ctx, &approved,
		s.ro.Rebind(q.build("COUNT(*)", "gtp.review_state = 'approved'")), q.args...); err != nil {
		return err
	}
	stats.TotalPRsReviewed = totalReviewed
	if totalReviewed > 0 {
		stats.ApprovalRate = float64(approved) / float64(totalReviewed)
	}
	return nil
}

// --- Issue Watch operations ---

// hydrateIssueWatch unmarshals JSON fields into their Go slices and
// normalizes the cleanup policy so legacy rows (or zero values) surface as
// the documented default.
func hydrateIssueWatch(iw *IssueWatch) {
	if iw.ReposJSON != "" {
		if err := json.Unmarshal([]byte(iw.ReposJSON), &iw.Repos); err != nil {
			fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal repos JSON for issue watch %s: %v\n", iw.ID, err)
		}
	}
	if iw.Repos == nil {
		iw.Repos = []RepoFilter{}
	}
	if iw.LabelsJSON != "" {
		if err := json.Unmarshal([]byte(iw.LabelsJSON), &iw.Labels); err != nil {
			fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal labels JSON for issue watch %s: %v\n", iw.ID, err)
		}
	}
	if iw.Labels == nil {
		iw.Labels = []string{}
	}
	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
}

// CreateIssueWatch creates a new issue watch configuration.
func (s *Store) CreateIssueWatch(ctx context.Context, iw *IssueWatch) error {
	if iw.ID == "" {
		iw.ID = uuid.New().String()
	}
	now := time.Now().UTC()
	iw.CreatedAt = now
	iw.UpdatedAt = now
	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
	reposJSON, err := json.Marshal(iw.Repos)
	if err != nil {
		return fmt.Errorf("marshal repos: %w", err)
	}
	iw.ReposJSON = string(reposJSON)
	labelsJSON, err := json.Marshal(iw.Labels)
	if err != nil {
		return fmt.Errorf("marshal labels: %w", err)
	}
	iw.LabelsJSON = string(labelsJSON)
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_issue_watches (id, workspace_id, workflow_id, workflow_step_id, repos,
			agent_profile_id, executor_profile_id, prompt, labels, custom_query,
			enabled, poll_interval_seconds, cleanup_policy, created_at, updated_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`),
		iw.ID, iw.WorkspaceID, iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON,
		iw.AgentProfileID, iw.ExecutorProfileID, iw.Prompt, iw.LabelsJSON, iw.CustomQuery,
		iw.Enabled, iw.PollIntervalSeconds, iw.CleanupPolicy, iw.CreatedAt, iw.UpdatedAt)
	return err
}

// GetIssueWatch returns an issue watch by ID.
func (s *Store) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) {
	var iw IssueWatch
	err := s.ro.GetContext(ctx, &iw, s.ro.Rebind(`SELECT * FROM github_issue_watches WHERE id = ?`), id)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	hydrateIssueWatch(&iw)
	return &iw, nil
}

// ListIssueWatches returns all issue watches for a workspace.
func (s *Store) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) {
	var watches []*IssueWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_issue_watches WHERE workspace_id = ? ORDER BY created_at`), workspaceID)
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateIssueWatch(w)
	}
	return watches, nil
}

// ListAllIssueWatches returns every issue watch across all workspaces.
func (s *Store) ListAllIssueWatches(ctx context.Context) ([]*IssueWatch, error) {
	var watches []*IssueWatch
	err := s.ro.SelectContext(ctx, &watches,
		`SELECT * FROM github_issue_watches ORDER BY workspace_id, created_at`)
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateIssueWatch(w)
	}
	return watches, nil
}

// ListEnabledIssueWatches returns all enabled issue watches.
func (s *Store) ListEnabledIssueWatches(ctx context.Context) ([]*IssueWatch, error) {
	var watches []*IssueWatch
	err := s.ro.SelectContext(ctx, &watches,
		s.ro.Rebind(`SELECT * FROM github_issue_watches WHERE enabled = TRUE ORDER BY created_at`))
	if err != nil {
		return nil, err
	}
	for _, w := range watches {
		hydrateIssueWatch(w)
	}
	return watches, nil
}

// UpdateIssueWatch updates an issue watch.
func (s *Store) UpdateIssueWatch(ctx context.Context, iw *IssueWatch) error {
	iw.UpdatedAt = time.Now().UTC()
	iw.CleanupPolicy = NormalizeCleanupPolicy(iw.CleanupPolicy)
	reposJSON, err := json.Marshal(iw.Repos)
	if err != nil {
		return fmt.Errorf("marshal repos: %w", err)
	}
	iw.ReposJSON = string(reposJSON)
	labelsJSON, err := json.Marshal(iw.Labels)
	if err != nil {
		return fmt.Errorf("marshal labels: %w", err)
	}
	iw.LabelsJSON = string(labelsJSON)
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_issue_watches SET workflow_id = ?, workflow_step_id = ?, repos = ?,
			agent_profile_id = ?, executor_profile_id = ?,
			prompt = ?, labels = ?, custom_query = ?,
			enabled = ?, poll_interval_seconds = ?, cleanup_policy = ?, last_polled_at = ?, updated_at = ?
		WHERE id = ?`),
		iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON,
		iw.AgentProfileID, iw.ExecutorProfileID,
		iw.Prompt, iw.LabelsJSON, iw.CustomQuery,
		iw.Enabled, iw.PollIntervalSeconds, iw.CleanupPolicy, iw.LastPolledAt, iw.UpdatedAt, iw.ID)
	return err
}

// DeleteIssueWatch deletes an issue watch and all its associated dedup task rows.
func (s *Store) DeleteIssueWatch(ctx context.Context, id string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_issue_watch_tasks WHERE issue_watch_id = ?`), id); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_issue_watches WHERE id = ?`), id); err != nil {
		return err
	}
	return tx.Commit()
}

// DisableIssueWatchWithError is the self-heal write: disables the watch and
// stamps a human-readable cause + timestamp. Symmetric with
// DisableReviewWatchWithError; called by the orchestrator when the
// watcher's bound agent profile is detected as soft-deleted.
func (s *Store) DisableIssueWatchWithError(ctx context.Context, id, cause string) error {
	now := time.Now().UTC()
	_, err := s.db.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_issue_watches
		   SET enabled = FALSE, last_error = ?, last_error_at = ?, updated_at = ?
		 WHERE id = ?`),
		cause, now, now, id)
	return err
}

// --- Issue Watch Task deduplication ---

// ReserveIssueWatchTask atomically claims a slot for a (watch, repo, issue) tuple.
// Returns true if this caller won the race and should proceed to create the task.
func (s *Store) ReserveIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int, issueURL string) (bool, error) {
	res, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_issue_watch_tasks (id, issue_watch_id, repo_owner, repo_name, issue_number, issue_url, task_id, created_at)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(issue_watch_id, repo_owner, repo_name, issue_number) DO NOTHING`),
		uuid.New().String(), issueWatchID, repoOwner, repoName, issueNumber, issueURL, "", time.Now().UTC())
	if err != nil {
		return false, err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return false, err
	}
	return rows == 1, nil
}

// AssignIssueWatchTaskID sets the task_id on a reserved dedup row.
func (s *Store) AssignIssueWatchTaskID(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int, taskID string) error {
	res, err := s.db.ExecContext(ctx, s.db.Rebind(`
		UPDATE github_issue_watch_tasks SET task_id = ?
		WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`),
		taskID, issueWatchID, repoOwner, repoName, issueNumber)
	if err != nil {
		return err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if rows == 0 {
		return fmt.Errorf("assign task ID: reservation row not found for watch=%s issue=%d", issueWatchID, issueNumber)
	}
	return nil
}

// ReleaseIssueWatchTask removes a reservation for a (watch, repo, issue) tuple.
func (s *Store) ReleaseIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		DELETE FROM github_issue_watch_tasks
		WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`),
		issueWatchID, repoOwner, repoName, issueNumber)
	return err
}

// HasIssueWatchTask checks if a task was already created for an issue in an issue watch.
func (s *Store) HasIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int) (bool, error) {
	var count int
	err := s.ro.GetContext(ctx, &count, s.ro.Rebind(
		`SELECT COUNT(*) FROM github_issue_watch_tasks WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`),
		issueWatchID, repoOwner, repoName, issueNumber)
	return count > 0, err
}

// ListIssueWatchTasksByWatch lists all dedup records for a given issue watch.
func (s *Store) ListIssueWatchTasksByWatch(ctx context.Context, watchID string) ([]*IssueWatchTask, error) {
	var tasks []*IssueWatchTask
	err := s.ro.SelectContext(ctx, &tasks, s.ro.Rebind(
		`SELECT d.id, d.issue_watch_id, d.repo_owner, d.repo_name, d.issue_number, d.issue_url, d.task_id, d.created_at
		 FROM github_issue_watch_tasks d
		 WHERE d.issue_watch_id = ?
		   AND (d.task_id = '' OR EXISTS (
			SELECT 1 FROM tasks t
			WHERE t.id = d.task_id AND t.archived_at IS NULL
		   ))`), watchID)
	return tasks, err
}

// ListAllIssueWatchTasks lists every dedup record across all watches. Used by
// the global cleanup sweep so orphaned rows still get evaluated.
func (s *Store) ListAllIssueWatchTasks(ctx context.Context) ([]*IssueWatchTask, error) {
	var tasks []*IssueWatchTask
	err := s.ro.SelectContext(ctx, &tasks,
		`SELECT d.id, d.issue_watch_id, d.repo_owner, d.repo_name, d.issue_number, d.issue_url, d.task_id, d.created_at
		 FROM github_issue_watch_tasks d
		 WHERE d.task_id = '' OR EXISTS (
			SELECT 1 FROM tasks t
			WHERE t.id = d.task_id AND t.archived_at IS NULL
		 )`)
	return tasks, err
}

// DeleteIssueWatchTask deletes a dedup record by ID.
func (s *Store) DeleteIssueWatchTask(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_issue_watch_tasks WHERE id = ?`), id)
	return err
}

// ListIssueWatchTaskIDsByWatch returns every task_id recorded against an
// issue watch, including empty-string reservations. Used by the watch
// reset flow to enumerate the tasks to cascade-delete.
func (s *Store) ListIssueWatchTaskIDsByWatch(ctx context.Context, watchID string) ([]string, error) {
	var ids []string
	err := s.ro.SelectContext(ctx, &ids,
		s.ro.Rebind(`SELECT task_id FROM github_issue_watch_tasks WHERE issue_watch_id = ?`), watchID)
	return ids, err
}

// ResetIssueWatchState wipes an issue watch's dedup rows and nulls its
// last_polled_at in a single transaction. Used by the reset flow after
// the cascade-delete loop so the next poll re-imports every currently
// matching issue as if the watch were freshly created.
func (s *Store) ResetIssueWatchState(ctx context.Context, watchID string) error {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`DELETE FROM github_issue_watch_tasks WHERE issue_watch_id = ?`), watchID); err != nil {
		return err
	}
	if _, err := tx.ExecContext(ctx, s.db.Rebind(
		`UPDATE github_issue_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`),
		time.Now().UTC(), watchID); err != nil {
		return err
	}
	return tx.Commit()
}

// --- Workspace settings operations ---

func defaultWorkspaceSettings(workspaceID string) *WorkspaceSettings {
	now := time.Now().UTC()
	return &WorkspaceSettings{
		WorkspaceID:            workspaceID,
		TaskGitCredentialsMode: TaskGitCredentialsModeManaged,
		RepoScopeMode:          RepoScopeModeAll,
		RepoScopeOrgs:          []string{},
		RepoScopeRepos:         []RepoFilter{},
		SavedPresets:           json.RawMessage("[]"),
		DefaultQueryPresets:    nil,
		CreatedAt:              now,
		UpdatedAt:              now,
	}
}

func cloneRawMessage(raw json.RawMessage) json.RawMessage {
	if len(raw) == 0 {
		return nil
	}
	out := make([]byte, len(raw))
	copy(out, raw)
	return out
}

const maxWorkspaceSettingsJSONBytes = 64 * 1024

func normalizeRepoScopeMode(mode string) string {
	mode = strings.TrimSpace(strings.ToLower(mode))
	switch mode {
	case RepoScopeModeOrgs, RepoScopeModeRepos:
		return mode
	default:
		return RepoScopeModeAll
	}
}

func normalizeTaskGitCredentialsMode(mode string) string {
	if strings.EqualFold(strings.TrimSpace(mode), TaskGitCredentialsModeExecutor) {
		return TaskGitCredentialsModeExecutor
	}
	return TaskGitCredentialsModeManaged
}

func normalizeWorkspaceSettings(settings *WorkspaceSettings) *WorkspaceSettings {
	if settings == nil {
		return nil
	}
	out := *settings
	out.WorkspaceID = strings.TrimSpace(out.WorkspaceID)
	out.TaskGitCredentialsMode = normalizeTaskGitCredentialsMode(out.TaskGitCredentialsMode)
	out.RepoScopeMode = normalizeRepoScopeMode(out.RepoScopeMode)
	if out.RepoScopeMode != RepoScopeModeOrgs {
		out.RepoScopeOrgs = nil
	}
	if out.RepoScopeMode != RepoScopeModeRepos {
		out.RepoScopeRepos = nil
	}
	orgs := make([]string, 0, len(out.RepoScopeOrgs))
	seenOrgs := make(map[string]struct{}, len(out.RepoScopeOrgs))
	for _, org := range out.RepoScopeOrgs {
		org = strings.TrimSpace(org)
		if org == "" {
			continue
		}
		key := strings.ToLower(org)
		if _, ok := seenOrgs[key]; ok {
			continue
		}
		seenOrgs[key] = struct{}{}
		orgs = append(orgs, org)
	}
	out.RepoScopeOrgs = orgs
	repos := make([]RepoFilter, 0, len(out.RepoScopeRepos))
	seenRepos := make(map[string]struct{}, len(out.RepoScopeRepos))
	for _, repo := range out.RepoScopeRepos {
		owner := strings.TrimSpace(repo.Owner)
		name := strings.TrimSpace(repo.Name)
		if owner == "" || name == "" {
			continue
		}
		key := strings.ToLower(owner + "/" + name)
		if _, ok := seenRepos[key]; ok {
			continue
		}
		seenRepos[key] = struct{}{}
		repos = append(repos, RepoFilter{Owner: owner, Name: name})
	}
	out.RepoScopeRepos = repos
	if len(out.SavedPresets) == 0 {
		out.SavedPresets = json.RawMessage("[]")
	} else {
		out.SavedPresets = cloneRawMessage(out.SavedPresets)
	}
	out.DefaultQueryPresets = cloneRawMessage(out.DefaultQueryPresets)
	return &out
}

// GetWorkspaceSettings returns per-workspace GitHub settings. Missing rows
// resolve to the backwards-compatible All repos defaults.
func (s *Store) GetWorkspaceSettings(ctx context.Context, workspaceID string) (*WorkspaceSettings, error) {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return nil, fmt.Errorf("workspace_id is required")
	}
	var row struct {
		WorkspaceID            string         `db:"workspace_id"`
		TaskGitCredentialsMode string         `db:"task_git_credentials_mode"`
		RepoScopeMode          string         `db:"repo_scope_mode"`
		RepoScopeOrgsJSON      string         `db:"repo_scope_orgs"`
		RepoScopeReposJSON     string         `db:"repo_scope_repos"`
		SavedPresets           string         `db:"saved_presets"`
		DefaultQueryPresets    sql.NullString `db:"default_query_presets"`
		CreatedAt              time.Time      `db:"created_at"`
		UpdatedAt              time.Time      `db:"updated_at"`
	}
	err := s.ro.GetContext(ctx, &row, s.ro.Rebind(`
		SELECT workspace_id, task_git_credentials_mode, repo_scope_mode, repo_scope_orgs, repo_scope_repos,
		       saved_presets, default_query_presets, created_at, updated_at
		FROM github_workspace_settings
		WHERE workspace_id = ?`), workspaceID)
	if errors.Is(err, sql.ErrNoRows) {
		return defaultWorkspaceSettings(workspaceID), nil
	}
	if err != nil {
		return nil, err
	}
	settings := &WorkspaceSettings{
		WorkspaceID:            row.WorkspaceID,
		TaskGitCredentialsMode: row.TaskGitCredentialsMode,
		RepoScopeMode:          row.RepoScopeMode,
		RepoScopeOrgsJSON:      row.RepoScopeOrgsJSON,
		RepoScopeReposJSON:     row.RepoScopeReposJSON,
		SavedPresets:           json.RawMessage(row.SavedPresets),
		CreatedAt:              row.CreatedAt,
		UpdatedAt:              row.UpdatedAt,
	}
	if row.DefaultQueryPresets.Valid {
		settings.DefaultQueryPresets = json.RawMessage(row.DefaultQueryPresets.String)
	}
	if err := json.Unmarshal([]byte(row.RepoScopeOrgsJSON), &settings.RepoScopeOrgs); err != nil {
		return nil, fmt.Errorf("unmarshal repo scope orgs: %w", err)
	}
	if err := json.Unmarshal([]byte(row.RepoScopeReposJSON), &settings.RepoScopeRepos); err != nil {
		return nil, fmt.Errorf("unmarshal repo scope repos: %w", err)
	}
	return normalizeWorkspaceSettings(settings), nil
}

// EnsureWorkspaceExecutorDefaults creates the initial task Git policy without
// rewriting any settings that already exist. This is intentionally separate
// from GetWorkspaceSettings' managed fallback so upgrades do not change the
// behavior of existing workspaces that have never stored a settings row.
func (s *Store) EnsureWorkspaceExecutorDefaults(ctx context.Context, workspaceID string) error {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return fmt.Errorf("workspace_id is required")
	}
	now := time.Now().UTC()
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_workspace_settings (
			workspace_id, task_git_credentials_mode, repo_scope_mode, repo_scope_orgs, repo_scope_repos,
			saved_presets, default_query_presets, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(workspace_id) DO NOTHING`),
		workspaceID, TaskGitCredentialsModeExecutor, RepoScopeModeAll, "[]", "[]", "[]", nil, now, now)
	return err
}

// DeleteWorkspaceSettings removes the non-secret GitHub settings owned by a
// workspace after the task repository has deleted the workspace row.
func (s *Store) DeleteWorkspaceSettings(ctx context.Context, workspaceID string) error {
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return fmt.Errorf("workspace_id is required")
	}
	_, err := s.db.ExecContext(ctx, s.db.Rebind(
		`DELETE FROM github_workspace_settings WHERE workspace_id = ?`), workspaceID)
	return err
}

func (s *Store) listWorkspaceIDs(ctx context.Context) ([]string, error) {
	if !s.tableExists("workspaces") {
		return nil, nil
	}
	var ids []string
	if err := s.ro.SelectContext(ctx, &ids, `
		SELECT id FROM workspaces WHERE TRIM(id) <> '' ORDER BY id`); err != nil {
		return nil, err
	}
	return ids, nil
}

// UpsertWorkspaceSettings stores per-workspace GitHub settings.
func (s *Store) UpsertWorkspaceSettings(ctx context.Context, settings *WorkspaceSettings) error {
	settings = normalizeWorkspaceSettings(settings)
	if settings == nil || settings.WorkspaceID == "" {
		return fmt.Errorf("workspace_id is required")
	}
	orgsJSON, err := json.Marshal(settings.RepoScopeOrgs)
	if err != nil {
		return fmt.Errorf("marshal repo scope orgs: %w", err)
	}
	reposJSON, err := json.Marshal(settings.RepoScopeRepos)
	if err != nil {
		return fmt.Errorf("marshal repo scope repos: %w", err)
	}
	now := time.Now().UTC()
	defaults := sql.NullString{}
	if len(settings.DefaultQueryPresets) > 0 {
		defaults.Valid = true
		defaults.String = string(settings.DefaultQueryPresets)
	}
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_workspace_settings (
			workspace_id, task_git_credentials_mode, repo_scope_mode, repo_scope_orgs, repo_scope_repos,
			saved_presets, default_query_presets, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(workspace_id) DO UPDATE SET
			task_git_credentials_mode = excluded.task_git_credentials_mode,
			repo_scope_mode = excluded.repo_scope_mode,
			repo_scope_orgs = excluded.repo_scope_orgs,
			repo_scope_repos = excluded.repo_scope_repos,
			saved_presets = excluded.saved_presets,
			default_query_presets = excluded.default_query_presets,
			updated_at = excluded.updated_at`),
		settings.WorkspaceID, settings.TaskGitCredentialsMode, settings.RepoScopeMode, string(orgsJSON), string(reposJSON),
		string(settings.SavedPresets), defaults, now, now)
	return err
}

// PatchWorkspaceSettings applies only the fields present in the request. This
// avoids lost updates when independent preference migrations run concurrently.
func (s *Store) PatchWorkspaceSettings(ctx context.Context, req *UpdateWorkspaceSettingsRequest) (*WorkspaceSettings, error) {
	if req == nil || strings.TrimSpace(req.WorkspaceID) == "" {
		return nil, fmt.Errorf("%w: workspace_id is required", ErrWorkspaceSettingsValidation)
	}
	workspaceID := strings.TrimSpace(req.WorkspaceID)
	now := time.Now().UTC()
	if _, err := s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_workspace_settings (
			workspace_id, task_git_credentials_mode, repo_scope_mode, repo_scope_orgs, repo_scope_repos,
			saved_presets, default_query_presets, created_at, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(workspace_id) DO NOTHING`),
		workspaceID, TaskGitCredentialsModeManaged, RepoScopeModeAll, "[]", "[]", "[]", nil, now, now); err != nil {
		return nil, err
	}

	patch := workspaceSettingsPatch{
		sets: make([]string, 0, 8),
		args: make([]any, 0, 10),
	}
	if err := appendWorkspaceScopePatch(&patch, req); err != nil {
		return nil, err
	}
	if req.TaskGitCredentialsMode != nil {
		patch.add("task_git_credentials_mode = ?", normalizeTaskGitCredentialsMode(*req.TaskGitCredentialsMode))
	}
	if err := appendWorkspacePresetPatch(&patch, req); err != nil {
		return nil, err
	}
	if len(patch.sets) > 0 {
		patch.add("updated_at = ?", now)
		patch.args = append(patch.args, workspaceID)
		query := "UPDATE github_workspace_settings SET " + strings.Join(patch.sets, ", ") + " WHERE workspace_id = ?"
		if _, err := s.db.ExecContext(ctx, s.db.Rebind(query), patch.args...); err != nil {
			return nil, err
		}
	}
	return s.GetWorkspaceSettings(ctx, workspaceID)
}

type workspaceSettingsPatch struct {
	sets []string
	args []any
}

func (p *workspaceSettingsPatch) add(set string, arg any) {
	p.sets = append(p.sets, set)
	p.args = append(p.args, arg)
}

func appendWorkspaceScopePatch(patch *workspaceSettingsPatch, req *UpdateWorkspaceSettingsRequest) error {
	if err := validateWorkspaceScopePatch(req); err != nil {
		return err
	}
	if req.RepoScopeMode != nil {
		if err := appendWorkspaceScopeModePatch(patch, *req.RepoScopeMode); err != nil {
			return err
		}
	}
	if req.RepoScopeOrgs != nil {
		raw, err := marshalNormalizedRepoScopeOrgs(*req.RepoScopeOrgs)
		if err != nil {
			return err
		}
		patch.add("repo_scope_orgs = ?", string(raw))
	}
	if req.RepoScopeRepos != nil {
		raw, err := marshalNormalizedRepoScopeRepos(*req.RepoScopeRepos)
		if err != nil {
			return err
		}
		patch.add("repo_scope_repos = ?", string(raw))
	}
	return nil
}

func validateWorkspaceScopePatch(req *UpdateWorkspaceSettingsRequest) error {
	if req.RepoScopeMode == nil {
		return nil
	}
	mode := normalizeRepoScopeMode(*req.RepoScopeMode)
	hasOrgs := req.RepoScopeOrgs != nil && len(*req.RepoScopeOrgs) > 0
	hasRepos := req.RepoScopeRepos != nil && len(*req.RepoScopeRepos) > 0
	if mode == RepoScopeModeAll && (hasOrgs || hasRepos) {
		return fmt.Errorf("%w: repo_scope_mode all cannot be patched with repo scope filters", ErrWorkspaceSettingsValidation)
	}
	if mode == RepoScopeModeOrgs && hasRepos {
		return fmt.Errorf("%w: repo_scope_mode orgs cannot be patched with repo_scope_repos", ErrWorkspaceSettingsValidation)
	}
	if mode == RepoScopeModeRepos && hasOrgs {
		return fmt.Errorf("%w: repo_scope_mode repos cannot be patched with repo_scope_orgs", ErrWorkspaceSettingsValidation)
	}
	return nil
}

func appendWorkspaceScopeModePatch(patch *workspaceSettingsPatch, rawMode string) error {
	if !isValidRepoScopeMode(rawMode) {
		return fmt.Errorf("%w: invalid repo_scope_mode %q", ErrWorkspaceSettingsValidation, rawMode)
	}
	mode := normalizeRepoScopeMode(rawMode)
	patch.add("repo_scope_mode = ?", mode)
	switch mode {
	case RepoScopeModeAll:
		patch.add("repo_scope_orgs = ?", "[]")
		patch.add("repo_scope_repos = ?", "[]")
	case RepoScopeModeOrgs:
		patch.add("repo_scope_repos = ?", "[]")
	case RepoScopeModeRepos:
		patch.add("repo_scope_orgs = ?", "[]")
	}
	return nil
}

func appendWorkspacePresetPatch(patch *workspaceSettingsPatch, req *UpdateWorkspaceSettingsRequest) error {
	if req.SavedPresetsSet {
		raw := json.RawMessage("[]")
		if req.SavedPresets != nil {
			raw = cloneRawMessage(*req.SavedPresets)
		}
		if err := validateWorkspaceSettingsJSON("saved_presets", raw, '['); err != nil {
			return err
		}
		patch.add("saved_presets = ?", string(raw))
	}
	if !req.DefaultQueriesSet {
		return nil
	}
	if req.DefaultQueryPresets == nil {
		patch.add("default_query_presets = ?", sql.NullString{})
		return nil
	}
	raw := cloneRawMessage(*req.DefaultQueryPresets)
	if err := validateWorkspaceSettingsJSON("default_query_presets", raw, '{'); err != nil {
		return err
	}
	patch.add("default_query_presets = ?", sql.NullString{String: string(raw), Valid: true})
	return nil
}

func marshalNormalizedRepoScopeOrgs(orgs []string) ([]byte, error) {
	settings := normalizeWorkspaceSettings(&WorkspaceSettings{
		RepoScopeMode: RepoScopeModeOrgs,
		RepoScopeOrgs: orgs,
	})
	raw, err := json.Marshal(settings.RepoScopeOrgs)
	if err != nil {
		return nil, fmt.Errorf("marshal repo scope orgs: %w", err)
	}
	return raw, nil
}

func marshalNormalizedRepoScopeRepos(repos []RepoFilter) ([]byte, error) {
	settings := normalizeWorkspaceSettings(&WorkspaceSettings{
		RepoScopeMode:  RepoScopeModeRepos,
		RepoScopeRepos: repos,
	})
	raw, err := json.Marshal(settings.RepoScopeRepos)
	if err != nil {
		return nil, fmt.Errorf("marshal repo scope repos: %w", err)
	}
	return raw, nil
}

func validateWorkspaceSettingsJSON(field string, raw json.RawMessage, wantFirst byte) error {
	raw = json.RawMessage(strings.TrimSpace(string(raw)))
	if len(raw) == 0 || len(raw) > maxWorkspaceSettingsJSONBytes || !json.Valid(raw) {
		return fmt.Errorf("%w: invalid %s", ErrWorkspaceSettingsValidation, field)
	}
	if raw[0] != wantFirst {
		return fmt.Errorf("%w: invalid %s", ErrWorkspaceSettingsValidation, field)
	}
	return nil
}

// --- Action preset operations ---

// GetActionPresets returns stored PR/Issue presets for a workspace. Returns
// (nil, nil) when no row exists yet so the caller can apply defaults.
func (s *Store) GetActionPresets(ctx context.Context, workspaceID string) (*ActionPresets, error) {
	var row struct {
		PRJSON    string `db:"pr_presets"`
		IssueJSON string `db:"issue_presets"`
	}
	err := s.ro.GetContext(ctx, &row, s.ro.Rebind(
		`SELECT pr_presets, issue_presets FROM github_action_presets WHERE workspace_id = ?`), workspaceID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	presets := &ActionPresets{WorkspaceID: workspaceID}
	if err := json.Unmarshal([]byte(row.PRJSON), &presets.PR); err != nil {
		return nil, fmt.Errorf("unmarshal pr presets: %w", err)
	}
	if err := json.Unmarshal([]byte(row.IssueJSON), &presets.Issue); err != nil {
		return nil, fmt.Errorf("unmarshal issue presets: %w", err)
	}
	return presets, nil
}

// UpsertActionPresets stores PR/Issue presets for a workspace, replacing any
// existing row.
func (s *Store) UpsertActionPresets(ctx context.Context, presets *ActionPresets) error {
	prJSON, err := json.Marshal(presets.PR)
	if err != nil {
		return fmt.Errorf("marshal pr presets: %w", err)
	}
	issueJSON, err := json.Marshal(presets.Issue)
	if err != nil {
		return fmt.Errorf("marshal issue presets: %w", err)
	}
	_, err = s.db.ExecContext(ctx, s.db.Rebind(`
		INSERT INTO github_action_presets (workspace_id, pr_presets, issue_presets, updated_at)
		VALUES (?, ?, ?, ?)
		ON CONFLICT(workspace_id) DO UPDATE SET
			pr_presets = excluded.pr_presets,
			issue_presets = excluded.issue_presets,
			updated_at = excluded.updated_at`),
		presets.WorkspaceID, string(prJSON), string(issueJSON), time.Now().UTC())
	return err
}

// DeleteActionPresets removes the stored overrides for a workspace so defaults
// apply again.
func (s *Store) DeleteActionPresets(ctx context.Context, workspaceID string) error {
	_, err := s.db.ExecContext(ctx, s.db.Rebind(`DELETE FROM github_action_presets WHERE workspace_id = ?`), workspaceID)
	return err
}