ag_store/
session.rs

1//! Session-scoped persistence adapters and query helpers.
2
3use std::sync::Arc;
4
5use ag_agent::{
6    self as agent, AgentKind, AgentModel, PermissionMode, ReasoningLevel, SessionStats, SpeedMode,
7};
8use ag_session::{FocusedReviewStatus, SessionMessageKind};
9use async_trait::async_trait;
10use sqlx::SqlitePool;
11use tracing::warn;
12
13use super::review::{
14    NewSessionReviewCommentResolution, SessionReviewRequestRow, insert_review_comment_resolutions,
15};
16use super::session_message::SessionMessageStore;
17use super::session_snapshot::SessionSnapshotStore;
18use super::status;
19use crate::DbError;
20use crate::timestamp::TimestampSource;
21
22/// Transactional turn-metadata payload persisted after one completed agent
23/// turn.
24///
25/// Owns its fields so the persistence trait method stays lifetime-free. A
26/// borrowed variant (`SessionTurnMetadata<'a>`) forced the persist method to
27/// carry a generic lifetime, which `mockall::automock` drops in the generated
28/// mock and newer `clippy` then rejects via `extra_unused_lifetimes`. Owning
29/// the data is allocation-cheap on this once-per-turn path and keeps the trait
30/// signature stable across toolchains.
31pub struct SessionTurnMetadata {
32    /// Personality id successfully delivered for this turn, or `None` when
33    /// the turn cleared or had no personality.
34    pub applied_personality_id: Option<String>,
35    /// Fingerprint of the personality prompt successfully delivered for this
36    /// turn.
37    pub applied_personality_prompt_hash: Option<String>,
38    /// Session-scoped instruction bootstrap marker for app-server providers.
39    pub instruction_conversation_id: Option<String>,
40    /// Model identifier used for per-model usage aggregation.
41    pub model: String,
42    /// Persisted provider-native conversation identifier for future resumes.
43    pub provider_conversation_id: Option<String>,
44    /// Serialized clarification-question payload stored on the session row.
45    pub questions_json: String,
46    /// Review-comment operations committed with this completed turn.
47    pub review_comment_resolutions: Vec<NewSessionReviewCommentResolution>,
48    /// Serialized structured summary payload stored on the session row.
49    pub summary: String,
50    /// Token-usage delta attributed to the completed turn.
51    pub token_usage_delta: SessionStats,
52}
53
54/// Borrowed values used to persist a newly created session with explicit
55/// provider identity and reasoning configuration.
56pub struct PersistedSessionCreation<'a> {
57    /// Persisted agent provider kind for the session.
58    pub agent: &'a str,
59    /// Base branch or parent branch used for future worktree materialization.
60    pub base_branch: &'a str,
61    /// Stable session identifier.
62    pub id: &'a str,
63    /// Whether the row was created through explicit draft staging.
64    pub is_draft: bool,
65    /// Persisted model identifier for the session.
66    pub model: &'a str,
67    /// Orchestration task that owns this child session, when applicable.
68    pub orchestration_task_id: Option<i64>,
69    /// Optional parent session id for one-level stacked drafts.
70    pub parent_session_id: Option<&'a str>,
71    /// Provider permission mode captured for the session.
72    pub permission_mode: PermissionMode,
73    /// Workspace personality selected for future turns, when present.
74    pub personality_id: Option<&'a str>,
75    /// Owning project identifier.
76    pub project_id: i64,
77    /// Reasoning level captured from the project default at creation.
78    pub reasoning_level: ReasoningLevel,
79    /// Persisted session role, or `None` for the default worker role.
80    pub role: Option<&'a str>,
81    /// Response-speed preference captured for the session.
82    pub speed_mode: SpeedMode,
83    /// Initial lifecycle status string.
84    pub status: &'a str,
85}
86
87/// Borrowed identifiers used to persist one forked session snapshot.
88pub struct ForkSessionSnapshot<'a> {
89    /// Stable id assigned to the newly forked session.
90    pub new_session_id: &'a str,
91    /// Stable id of the source session whose metadata and transcript are
92    /// copied.
93    pub source_session_id: &'a str,
94    /// Initial lifecycle status for the forked session.
95    pub status: &'a str,
96}
97
98/// Row returned when loading a session from the `session` table.
99///
100/// Includes optional normalized forge review-request linkage metadata loaded
101/// through the `session_review_request` table when the session has been
102/// published for remote review.
103pub struct SessionRow {
104    /// Persisted added-line count from the latest diff stats refresh.
105    pub added_lines: i64,
106    /// Persisted agent provider kind selected for this session.
107    pub agent: String,
108    /// Base branch used to create the session worktree.
109    pub base_branch: String,
110    /// Session creation timestamp in Unix seconds.
111    pub created_at: i64,
112    /// Persisted deleted-line count from the latest diff stats refresh.
113    pub deleted_lines: i64,
114    /// Whether the latest successful diff refresh returned content, or
115    /// `None` when diff availability is unknown.
116    pub has_diff: Option<bool>,
117    /// Stable session identifier.
118    pub id: String,
119    /// Open active-work interval start timestamp, if any.
120    pub in_progress_started_at: Option<i64>,
121    /// Completed active-work duration in whole seconds.
122    pub in_progress_total_seconds: i64,
123    /// Total input tokens accumulated for the session.
124    pub input_tokens: i64,
125    /// Whether the session is still an explicit draft.
126    pub is_draft: bool,
127    /// Persisted agent model identifier.
128    pub model: String,
129    /// Total output tokens accumulated for the session.
130    pub output_tokens: i64,
131    /// Parent session id when this is a one-level stacked draft.
132    pub parent_session_id: Option<String>,
133    /// Persisted provider permission mode for future turns.
134    pub permission_mode: String,
135    /// Workspace personality selected for future turns, when present.
136    pub personality_id: Option<String>,
137    /// Owning project identifier, when present.
138    pub project_id: Option<i64>,
139    /// Initial or staged prompt text.
140    pub prompt: String,
141    /// Published upstream branch reference, when present.
142    pub published_upstream_ref: Option<String>,
143    /// Serialized clarification-question payload, when present.
144    pub questions: Option<String>,
145    /// Persisted session-specific reasoning override, when present.
146    pub reasoning_level_override: Option<String>,
147    /// Joined forge review-request metadata, when present and complete.
148    pub review_request: Option<SessionReviewRequestRow>,
149    /// Persisted session role string, or `None` for the default worker role.
150    pub role: Option<String>,
151    /// Persisted size bucket string.
152    pub size: String,
153    /// Persisted session response-speed preference.
154    pub speed_mode: String,
155    /// Persisted lifecycle status string.
156    pub status: String,
157    /// Persisted structured summary text, when present.
158    pub summary: Option<String>,
159    /// Optional display title.
160    pub title: Option<String>,
161    /// Last update timestamp in Unix seconds.
162    pub updated_at: i64,
163}
164
165/// Lightweight row returned when loading session-list metadata.
166///
167/// Omits transcript-scale fields (`prompt`, `questions`, and `summary`) so
168/// list refreshes scale with visible metadata instead of the cumulative size
169/// of every saved conversation.
170pub struct SessionListRow {
171    /// Persisted added-line count from the latest diff stats refresh.
172    pub added_lines: i64,
173    /// Persisted agent provider kind for this session.
174    pub agent: String,
175    /// Base branch used to create the session worktree.
176    pub base_branch: String,
177    /// Session creation timestamp in Unix seconds.
178    pub created_at: i64,
179    /// Persisted deleted-line count from the latest diff stats refresh.
180    pub deleted_lines: i64,
181    /// Whether the latest successful diff refresh returned content, or
182    /// `None` when diff availability is unknown.
183    pub has_diff: Option<bool>,
184    /// Stable session identifier.
185    pub id: String,
186    /// Open active-work interval start timestamp, if any.
187    pub in_progress_started_at: Option<i64>,
188    /// Completed active-work duration in whole seconds.
189    pub in_progress_total_seconds: i64,
190    /// Total input tokens accumulated for the session.
191    pub input_tokens: i64,
192    /// Whether the session is still an explicit draft.
193    pub is_draft: bool,
194    /// Persisted agent model identifier.
195    pub model: String,
196    /// Total output tokens accumulated for the session.
197    pub output_tokens: i64,
198    /// Parent session id when this row is a one-level stacked draft.
199    pub parent_session_id: Option<String>,
200    /// Persisted provider permission mode for future turns.
201    pub permission_mode: String,
202    /// Workspace personality selected for future turns, when present.
203    pub personality_id: Option<String>,
204    /// Owning project identifier, when present.
205    pub project_id: Option<i64>,
206    /// Published upstream branch reference, when present.
207    pub published_upstream_ref: Option<String>,
208    /// Persisted session-specific reasoning override, when present.
209    pub reasoning_level_override: Option<String>,
210    /// Joined forge review-request metadata, when present and complete.
211    pub review_request: Option<SessionReviewRequestRow>,
212    /// Persisted session role string, or `None` for the default worker role.
213    pub role: Option<String>,
214    /// Persisted size bucket string.
215    pub size: String,
216    /// Persisted session response-speed preference.
217    pub speed_mode: String,
218    /// Persisted lifecycle status string.
219    pub status: String,
220    /// Optional display title.
221    pub title: Option<String>,
222    /// Last update timestamp in Unix seconds.
223    pub updated_at: i64,
224}
225
226/// Minimal provider/model row used to migrate active sessions across projects.
227#[derive(sqlx::FromRow)]
228pub struct SessionAgentModelRow {
229    /// Persisted agent provider kind for this session.
230    pub agent: String,
231    /// Stable session identifier.
232    pub id: String,
233    /// Persisted agent model identifier.
234    pub model: String,
235    /// Persisted lifecycle status string.
236    pub status: String,
237}
238
239/// Transcript-detail row loaded lazily for the session being viewed.
240pub struct SessionDetailRow {
241    /// Initial or staged prompt text.
242    pub prompt: String,
243    /// Serialized clarification-question payload, when present.
244    pub questions: Option<String>,
245    /// Persisted structured summary text, when present.
246    pub summary: Option<String>,
247}
248
249/// Row returned when loading one persisted `session_message`.
250#[derive(Clone, Debug, Eq, PartialEq)]
251pub struct SessionMessageRow {
252    /// Canonical transcript text for this message.
253    pub content: String,
254    /// Stable message-kind string.
255    pub kind: String,
256    /// Monotonic position within the owning session transcript.
257    pub position: i64,
258}
259
260/// Row returned when hydrating persisted focused-review cache entries.
261#[derive(Clone, Debug, Eq, PartialEq)]
262pub struct SessionFocusedReviewRow {
263    /// Diff-content hash captured when the focused review was generated.
264    pub diff_hash: String,
265    /// Stable session identifier.
266    pub session_id: String,
267    /// Generated focused-review markdown text.
268    pub text: String,
269}
270
271/// Persisted selected and successfully applied personality state.
272#[derive(Clone, Debug, Eq, PartialEq)]
273pub struct SessionPersonalityState {
274    /// Personality id delivered during the latest successful turn.
275    pub applied_personality_id: Option<String>,
276    /// Fingerprint of the personality prompt delivered during that turn.
277    pub applied_personality_prompt_hash: Option<String>,
278    /// Personality id selected for the session's next turn.
279    pub personality_id: Option<String>,
280}
281
282/// Session-focused persistence boundary used by app orchestration and tests.
283#[async_trait]
284pub trait SessionRepository: Send + Sync {
285    /// Appends one typed transcript message and refreshes session ordering
286    /// metadata.
287    async fn append_session_message(
288        &self,
289        id: &str,
290        kind: SessionMessageKind,
291        content: &str,
292    ) -> Result<(), DbError>;
293
294    /// Sets `project_id` for sessions that do not yet reference a project.
295    async fn backfill_session_project(&self, project_id: i64) -> Result<(), DbError>;
296
297    /// Persists an automatic focused-review trigger when the session still
298    /// exists and is eligible for a worker review.
299    async fn defer_session_focused_review(&self, id: &str) -> Result<bool, DbError>;
300
301    /// Deletes a session row by identifier.
302    async fn delete_session(&self, id: &str) -> Result<(), DbError>;
303
304    /// Returns the persisted base branch for a session, when present.
305    async fn get_session_base_branch(&self, id: &str) -> Result<Option<String>, DbError>;
306
307    /// Returns the parent session id for a stacked session, when present.
308    async fn get_session_parent_session_id(&self, id: &str) -> Result<Option<String>, DbError>;
309
310    /// Returns the parent/base commit hash that a stacked child branch was
311    /// last known to contain.
312    async fn get_session_stack_base_commit_hash(&self, id: &str)
313    -> Result<Option<String>, DbError>;
314
315    /// Returns the persisted app-server instruction bootstrap marker for a
316    /// session, when present.
317    async fn get_session_instruction_conversation_id(
318        &self,
319        id: &str,
320    ) -> Result<Option<String>, DbError>;
321
322    /// Returns the provider conversation identifier for a session, when
323    /// present.
324    async fn get_session_provider_conversation_id(
325        &self,
326        id: &str,
327    ) -> Result<Option<String>, DbError>;
328
329    /// Inserts a newly created draft-session row.
330    async fn insert_draft_session(
331        &self,
332        id: &str,
333        model: &str,
334        base_branch: &str,
335        status: &str,
336        project_id: i64,
337    ) -> Result<(), DbError>;
338
339    /// Inserts a newly created stacked draft-session row.
340    async fn insert_stacked_draft_session(
341        &self,
342        id: &str,
343        model: &str,
344        base_branch: &str,
345        status: &str,
346        parent_session_id: &str,
347        project_id: i64,
348    ) -> Result<(), DbError>;
349
350    /// Inserts a newly created session row.
351    async fn insert_session(
352        &self,
353        id: &str,
354        model: &str,
355        base_branch: &str,
356        status: &str,
357        project_id: i64,
358    ) -> Result<(), DbError>;
359
360    /// Inserts a newly created session row with explicit provider identity.
361    async fn insert_session_with_agent(
362        &self,
363        session: PersistedSessionCreation<'_>,
364    ) -> Result<(), DbError>;
365
366    /// Inserts a new session by snapshotting source metadata and ordered
367    /// transcript messages while clearing source-specific runtime linkage.
368    async fn fork_session_snapshot(&self, snapshot: ForkSessionSnapshot<'_>)
369    -> Result<(), DbError>;
370
371    /// Loads one complete persisted session row by stable identifier.
372    async fn load_session(&self, session_id: &str) -> Result<Option<SessionRow>, DbError>;
373
374    /// Loads provider/model metadata for every non-terminal session across
375    /// projects.
376    async fn load_active_session_agent_models(&self) -> Result<Vec<SessionAgentModelRow>, DbError>;
377
378    #[cfg(any(test, feature = "test-utils"))]
379    /// Loads all sessions ordered by most recent update.
380    async fn load_sessions(&self) -> Result<Vec<SessionRow>, DbError>;
381
382    /// Loads lightweight session-list metadata ordered by most recent update
383    /// for one project.
384    async fn load_sessions_for_project(
385        &self,
386        project_id: i64,
387    ) -> Result<Vec<SessionListRow>, DbError>;
388
389    /// Loads transcript-scale detail for one session when it becomes active.
390    async fn load_session_detail(
391        &self,
392        session_id: &str,
393    ) -> Result<Option<SessionDetailRow>, DbError>;
394
395    /// Loads ordered transcript messages for one session.
396    async fn load_session_messages(
397        &self,
398        session_id: &str,
399    ) -> Result<Vec<SessionMessageRow>, DbError>;
400
401    /// Loads persisted focused-review cache rows for one project.
402    async fn load_session_focused_reviews_for_project(
403        &self,
404        project_id: i64,
405    ) -> Result<Vec<SessionFocusedReviewRow>, DbError>;
406
407    /// Loads lightweight session metadata used for cheap change detection.
408    async fn load_sessions_metadata(&self) -> Result<(i64, i64), DbError>;
409
410    /// Loads the project identifier associated with one session.
411    async fn load_session_project_id(&self, session_id: &str) -> Result<Option<i64>, DbError>;
412
413    /// Loads selected and last-applied personality state for one session.
414    async fn load_session_personality_state(
415        &self,
416        session_id: &str,
417    ) -> Result<Option<SessionPersonalityState>, DbError>;
418
419    /// Loads parentless review-ready sessions that still need their recorded
420    /// stack-base commit replayed onto their current base branch.
421    async fn load_pending_stack_restack_session_ids(
422        &self,
423        project_id: i64,
424    ) -> Result<Vec<String>, DbError>;
425
426    /// Loads eligible worker sessions with a durable automatic focused-review
427    /// trigger for one project.
428    async fn load_pending_focused_review_session_ids(
429        &self,
430        project_id: i64,
431    ) -> Result<Vec<String>, DbError>;
432
433    /// Returns the persisted upstream reference for a published session
434    /// branch, when present.
435    async fn load_session_published_upstream_ref(
436        &self,
437        id: &str,
438    ) -> Result<Option<String>, DbError>;
439
440    /// Loads the persisted merged commit hash for one session, when present.
441    async fn load_session_merged_commit_hash(
442        &self,
443        session_id: &str,
444    ) -> Result<Option<String>, DbError>;
445
446    /// Loads the immutable diff archived before managed-session cleanup.
447    async fn load_session_archived_diff(&self, session_id: &str)
448    -> Result<Option<String>, DbError>;
449
450    /// Clears parent links for children after their parent session merges
451    /// into its base branch, returning materialized children that may need a
452    /// follow-up branch restack.
453    async fn restack_child_sessions_after_parent_merge(
454        &self,
455        parent_session_id: &str,
456        base_branch: &str,
457        parent_commit_hash: Option<String>,
458    ) -> Result<Vec<String>, DbError>;
459
460    /// Loads the persisted session reasoning level.
461    async fn load_session_reasoning_level(
462        &self,
463        session_id: &str,
464    ) -> Result<ReasoningLevel, DbError>;
465
466    /// Loads the persisted provider permission mode for future turns.
467    async fn load_session_permission_mode(
468        &self,
469        session_id: &str,
470    ) -> Result<PermissionMode, DbError>;
471
472    /// Loads the persisted session response-speed preference.
473    async fn load_session_speed_mode(&self, session_id: &str) -> Result<SpeedMode, DbError>;
474
475    /// Loads the persisted summary text associated with one session.
476    async fn load_session_summary(&self, session_id: &str) -> Result<Option<String>, DbError>;
477
478    /// Returns `(created_at, updated_at)` timestamps for a session.
479    async fn load_session_timestamps(
480        &self,
481        session_id: &str,
482    ) -> Result<Option<(i64, i64)>, DbError>;
483
484    /// Persists all canonical turn metadata for one completed agent turn in a
485    /// single transaction.
486    async fn persist_session_turn_metadata(
487        &self,
488        session_id: &str,
489        turn_metadata: &SessionTurnMetadata,
490    ) -> Result<(), DbError>;
491
492    /// Marks persisted diff availability unknown while retaining the last
493    /// known size and line counts.
494    async fn mark_session_diff_unknown(&self, id: &str) -> Result<(), DbError>;
495
496    /// Updates persisted diff-derived presence, size, and line-count fields
497    /// for a session row.
498    async fn update_session_diff_stats(
499        &self,
500        added_lines: u64,
501        deleted_lines: u64,
502        has_diff: bool,
503        id: &str,
504        size: &str,
505    ) -> Result<(), DbError>;
506
507    /// Updates the persisted app-server instruction bootstrap marker for a
508    /// session.
509    async fn update_session_instruction_conversation_id(
510        &self,
511        id: &str,
512        provider_conversation_id: Option<String>,
513    ) -> Result<(), DbError>;
514
515    /// Updates the persisted model for a session.
516    async fn update_session_model(&self, id: &str, model: &str) -> Result<(), DbError>;
517
518    /// Updates or clears the personality selected for future turns.
519    async fn update_session_personality_id(
520        &self,
521        id: &str,
522        personality_id: Option<String>,
523    ) -> Result<(), DbError>;
524
525    /// Updates the persisted agent provider and model for a session.
526    async fn update_session_agent_model(
527        &self,
528        id: &str,
529        agent: &str,
530        model: &str,
531    ) -> Result<(), DbError>;
532
533    /// Updates the persisted agent provider and model only while the session
534    /// remains non-terminal, without changing its activity timestamp.
535    async fn update_active_session_agent_model(
536        &self,
537        id: &str,
538        agent: &str,
539        model: &str,
540    ) -> Result<(), DbError>;
541
542    /// Clears the draft flag for a session row once its staged draft bundle
543    /// starts the first live turn.
544    async fn clear_session_draft_flag(&self, id: &str) -> Result<(), DbError>;
545
546    /// Updates the persisted merged commit hash for a session row.
547    async fn update_session_merged_commit_hash(
548        &self,
549        id: &str,
550        merged_commit_hash: Option<String>,
551    ) -> Result<(), DbError>;
552
553    /// Persists or clears the immutable diff retained for archived sessions.
554    async fn update_session_archived_diff(
555        &self,
556        id: &str,
557        archived_diff: Option<String>,
558    ) -> Result<(), DbError>;
559
560    /// Persists or clears the parent/base commit hash used for deterministic
561    /// stacked-child rebases.
562    async fn update_session_stack_base_commit_hash(
563        &self,
564        id: &str,
565        stack_base_commit_hash: Option<String>,
566    ) -> Result<(), DbError>;
567
568    /// Updates the saved prompt for a session row.
569    async fn update_session_prompt(&self, id: &str, prompt: &str) -> Result<(), DbError>;
570
571    /// Updates the persisted provider conversation identifier for a session.
572    async fn update_session_provider_conversation_id(
573        &self,
574        id: &str,
575        provider_conversation_id: Option<String>,
576    ) -> Result<(), DbError>;
577
578    /// Updates the model clarification questions for a session row.
579    async fn update_session_questions(&self, id: &str, questions: &str) -> Result<(), DbError>;
580
581    /// Updates the persisted session reasoning level.
582    async fn update_session_reasoning_level(
583        &self,
584        id: &str,
585        reasoning_level: ReasoningLevel,
586    ) -> Result<(), DbError>;
587
588    /// Updates the persisted provider permission mode for future turns.
589    async fn update_session_permission_mode(
590        &self,
591        id: &str,
592        permission_mode: PermissionMode,
593    ) -> Result<(), DbError>;
594
595    /// Updates the persisted session response-speed preference.
596    async fn update_session_speed_mode(
597        &self,
598        id: &str,
599        speed_mode: SpeedMode,
600    ) -> Result<(), DbError>;
601
602    /// Updates the persisted upstream reference for a published session
603    /// branch.
604    async fn update_session_published_upstream_ref(
605        &self,
606        id: &str,
607        published_upstream_ref: Option<String>,
608    ) -> Result<(), DbError>;
609
610    /// Accumulates token statistics for a session.
611    async fn update_session_stats(&self, id: &str, stats: &SessionStats) -> Result<(), DbError>;
612
613    /// Updates the status for a session row and opens or closes the persisted
614    /// cumulative active-work interval when crossing the `InProgress`
615    /// boundary.
616    async fn update_session_status_with_timing_at(
617        &self,
618        id: &str,
619        status: &str,
620        timestamp_seconds: i64,
621    ) -> Result<(), DbError>;
622
623    /// Updates the persisted session summary text for a session row.
624    async fn update_session_summary(&self, id: &str, summary: &str) -> Result<(), DbError>;
625
626    /// Updates or clears the persisted focused-review cache for a session.
627    async fn update_session_focused_review(
628        &self,
629        id: &str,
630        status: Option<FocusedReviewStatus>,
631        diff_hash: Option<String>,
632        text: Option<String>,
633    ) -> Result<(), DbError>;
634
635    /// Updates the display title for a session row.
636    async fn update_session_title(&self, id: &str, title: &str) -> Result<(), DbError>;
637
638    /// Stores a fallback title that can be refined by a later substantive
639    /// user prompt.
640    async fn update_session_provisional_title(&self, id: &str, title: &str) -> Result<(), DbError>;
641
642    /// Claims the next ordered title candidate for a session.
643    ///
644    /// When `requires_provisional_title` is true, no candidate is claimed
645    /// after a generated or commit-derived title becomes authoritative.
646    async fn begin_session_title_generation(
647        &self,
648        id: &str,
649        requires_provisional_title: bool,
650    ) -> Result<Option<i64>, DbError>;
651
652    /// Applies one generated title unless a newer candidate or authoritative
653    /// title has already been accepted.
654    async fn update_session_title_for_generation(
655        &self,
656        id: &str,
657        expected_generation: i64,
658        title: &str,
659    ) -> Result<bool, DbError>;
660
661    /// Overrides the `created_at` timestamp for one session row.
662    #[cfg(any(test, feature = "test-utils"))]
663    async fn update_session_created_at(&self, id: &str, created_at: i64) -> Result<(), DbError>;
664
665    #[cfg(any(test, feature = "test-utils"))]
666    /// Overrides the `updated_at` timestamp for one session row.
667    async fn update_session_updated_at(&self, id: &str, updated_at: i64) -> Result<(), DbError>;
668}
669
670/// `SQLite` implementation of [`SessionRepository`].
671#[derive(Clone)]
672pub(crate) struct SqliteSessionRepository(
673    SqlitePool,
674    Arc<dyn TimestampSource>,
675    SessionMessageStore,
676    SessionSnapshotStore,
677);
678
679impl SqliteSessionRepository {
680    /// Creates a session repository backed by the provided pool.
681    pub(crate) fn new(pool: SqlitePool, timestamp_source: Arc<dyn TimestampSource>) -> Self {
682        Self(
683            pool.clone(),
684            Arc::clone(&timestamp_source),
685            SessionMessageStore::new(pool.clone(), Arc::clone(&timestamp_source)),
686            SessionSnapshotStore::new(pool, timestamp_source),
687        )
688    }
689
690    /// Returns the shared persistence timestamp in Unix seconds.
691    fn now(&self) -> i64 {
692        self.1.now_timestamp_seconds()
693    }
694}
695
696/// Row returned when loading a required string scalar value.
697struct RequiredStringValueRow {
698    value: String,
699}
700
701/// Row returned when loading session count and latest-update metadata.
702struct SessionStatsMetadataRow {
703    /// Latest `session.updated_at` timestamp across rows.
704    max_updated_at: i64,
705    /// Total number of persisted sessions.
706    session_count: i64,
707}
708
709/// Row returned when loading an optional `i64` scalar value.
710struct OptionalI64ValueRow {
711    value: Option<i64>,
712}
713
714/// Row returned when loading the persisted instruction bootstrap marker for
715/// one session.
716struct SessionInstructionStateRow {
717    app_server_instruction_provider_conversation_id: Option<String>,
718}
719
720impl SessionInstructionStateRow {
721    /// Converts the optional stored provider conversation id into one
722    /// normalized bootstrap conversation id when present and non-empty.
723    fn into_instruction_conversation_id(self) -> Option<String> {
724        agent::normalize_instruction_conversation_id(
725            self.app_server_instruction_provider_conversation_id
726                .as_deref(),
727        )
728    }
729}
730
731/// Row returned when loading both persisted timestamps for one session.
732struct SessionTimestampsRow {
733    created_at: i64,
734    updated_at: i64,
735}
736
737/// Shared columns for session metadata rows used by both session and
738/// session-list mappings.
739struct SessionRowMetadata {
740    added_lines: i64,
741    agent: String,
742    base_branch: String,
743    created_at: i64,
744    deleted_lines: i64,
745    has_diff: Option<bool>,
746    id: String,
747    in_progress_started_at: Option<i64>,
748    in_progress_total_seconds: i64,
749    input_tokens: i64,
750    is_draft: bool,
751    model: String,
752    output_tokens: i64,
753    parent_session_id: Option<String>,
754    permission_mode: String,
755    personality_id: Option<String>,
756    project_id: Option<i64>,
757    published_upstream_ref: Option<String>,
758    reasoning_level_override: Option<String>,
759    role: Option<String>,
760    size: String,
761    speed_mode: String,
762    status: String,
763    title: Option<String>,
764    updated_at: i64,
765}
766
767impl SessionRowMetadata {
768    /// Converts shared metadata fields into a complete session row.
769    fn into_session_row(
770        self,
771        prompt: String,
772        questions: Option<String>,
773        summary: Option<String>,
774        review_request: Option<SessionReviewRequestRow>,
775    ) -> SessionRow {
776        SessionRow {
777            added_lines: self.added_lines,
778            agent: self.agent,
779            base_branch: self.base_branch,
780            created_at: self.created_at,
781            deleted_lines: self.deleted_lines,
782            has_diff: self.has_diff,
783            id: self.id,
784            in_progress_started_at: self.in_progress_started_at,
785            in_progress_total_seconds: self.in_progress_total_seconds,
786            input_tokens: self.input_tokens,
787            is_draft: self.is_draft,
788            model: self.model,
789            output_tokens: self.output_tokens,
790            parent_session_id: self.parent_session_id,
791            permission_mode: self.permission_mode,
792            personality_id: self.personality_id,
793            project_id: self.project_id,
794            prompt,
795            published_upstream_ref: self.published_upstream_ref,
796            questions,
797            reasoning_level_override: self.reasoning_level_override,
798            review_request,
799            role: self.role,
800            size: self.size,
801            speed_mode: self.speed_mode,
802            status: self.status,
803            summary,
804            title: self.title,
805            updated_at: self.updated_at,
806        }
807    }
808
809    /// Converts shared metadata fields into a session-list row.
810    fn into_session_list_row(
811        self,
812        review_request: Option<SessionReviewRequestRow>,
813    ) -> SessionListRow {
814        SessionListRow {
815            added_lines: self.added_lines,
816            agent: self.agent,
817            base_branch: self.base_branch,
818            created_at: self.created_at,
819            deleted_lines: self.deleted_lines,
820            has_diff: self.has_diff,
821            id: self.id,
822            in_progress_started_at: self.in_progress_started_at,
823            in_progress_total_seconds: self.in_progress_total_seconds,
824            input_tokens: self.input_tokens,
825            is_draft: self.is_draft,
826            model: self.model,
827            output_tokens: self.output_tokens,
828            parent_session_id: self.parent_session_id,
829            permission_mode: self.permission_mode,
830            personality_id: self.personality_id,
831            project_id: self.project_id,
832            published_upstream_ref: self.published_upstream_ref,
833            reasoning_level_override: self.reasoning_level_override,
834            review_request,
835            role: self.role,
836            size: self.size,
837            speed_mode: self.speed_mode,
838            status: self.status,
839            title: self.title,
840            updated_at: self.updated_at,
841        }
842    }
843}
844
845/// Row returned when loading one complete `session` plus aliased
846/// `session_review_request` join columns.
847#[derive(sqlx::FromRow)]
848struct SessionJoinRow {
849    added_lines: i64,
850    agent: String,
851    base_branch: String,
852    created_at: i64,
853    deleted_lines: i64,
854    has_diff: Option<bool>,
855    id: String,
856    in_progress_started_at: Option<i64>,
857    in_progress_total_seconds: i64,
858    input_tokens: i64,
859    is_draft: bool,
860    model: String,
861    output_tokens: i64,
862    parent_session_id: Option<String>,
863    permission_mode: String,
864    personality_id: Option<String>,
865    project_id: Option<i64>,
866    prompt: String,
867    published_upstream_ref: Option<String>,
868    questions: Option<String>,
869    reasoning_level_override: Option<String>,
870    review_request_display_id: Option<String>,
871    review_request_forge_kind: Option<String>,
872    review_request_last_refreshed_at: Option<i64>,
873    review_request_source_branch: Option<String>,
874    review_request_state: Option<String>,
875    review_request_status_summary: Option<String>,
876    review_request_target_branch: Option<String>,
877    review_request_title: Option<String>,
878    review_request_web_url: Option<String>,
879    role: Option<String>,
880    size: String,
881    speed_mode: String,
882    status: String,
883    summary: Option<String>,
884    title: Option<String>,
885    updated_at: i64,
886}
887
888impl SessionJoinRow {
889    /// Returns whether this row can be included in a collection load.
890    ///
891    /// Invalid persisted statuses are logged and omitted so one corrupt row
892    /// cannot hide every otherwise valid session in the project list.
893    fn has_loadable_status(&self) -> bool {
894        if let Err(error) = status::validate_session(&self.status) {
895            warn!(
896                session_id = %self.id,
897                %error,
898                "Skipping session with invalid persisted status"
899            );
900
901            return false;
902        }
903
904        true
905    }
906
907    /// Converts the query-mapped join row into a complete [`SessionRow`].
908    fn into_session_row(self) -> SessionRow {
909        let (metadata, detail, review_request) = self.into_parts();
910
911        metadata.into_session_row(
912            detail.prompt,
913            detail.questions,
914            detail.summary,
915            review_request,
916        )
917    }
918
919    /// Converts placeholder-detail query rows into a lightweight
920    /// [`SessionListRow`].
921    fn into_session_list_row(self) -> SessionListRow {
922        let (metadata, _, review_request) = self.into_parts();
923
924        metadata.into_session_list_row(review_request)
925    }
926
927    /// Splits the flat query row into shared metadata, transcript detail, and
928    /// normalized review-request data.
929    fn into_parts(
930        self,
931    ) -> (
932        SessionRowMetadata,
933        SessionDetailRow,
934        Option<SessionReviewRequestRow>,
935    ) {
936        let Self {
937            added_lines,
938            agent,
939            base_branch,
940            created_at,
941            deleted_lines,
942            has_diff,
943            id,
944            in_progress_started_at,
945            in_progress_total_seconds,
946            input_tokens,
947            is_draft,
948            model,
949            output_tokens,
950            parent_session_id,
951            permission_mode,
952            personality_id,
953            project_id,
954            prompt,
955            published_upstream_ref,
956            questions,
957            reasoning_level_override,
958            review_request_display_id,
959            review_request_forge_kind,
960            review_request_last_refreshed_at,
961            review_request_source_branch,
962            review_request_state,
963            review_request_status_summary,
964            review_request_target_branch,
965            review_request_title,
966            review_request_web_url,
967            role,
968            size,
969            speed_mode,
970            status,
971            summary,
972            title,
973            updated_at,
974        } = self;
975
976        let metadata = SessionRowMetadata {
977            added_lines,
978            agent,
979            base_branch,
980            created_at,
981            deleted_lines,
982            has_diff,
983            id,
984            in_progress_started_at,
985            in_progress_total_seconds,
986            input_tokens,
987            is_draft,
988            model,
989            output_tokens,
990            parent_session_id,
991            permission_mode,
992            personality_id,
993            project_id,
994            published_upstream_ref,
995            reasoning_level_override,
996            role,
997            size,
998            speed_mode,
999            status,
1000            title,
1001            updated_at,
1002        };
1003        let detail = SessionDetailRow {
1004            prompt,
1005            questions,
1006            summary,
1007        };
1008        let review_request = SessionReviewRequestJoinRow {
1009            display_id: review_request_display_id,
1010            forge_kind: review_request_forge_kind,
1011            last_refreshed_at: review_request_last_refreshed_at,
1012            source_branch: review_request_source_branch,
1013            state: review_request_state,
1014            status_summary: review_request_status_summary,
1015            target_branch: review_request_target_branch,
1016            title: review_request_title,
1017            web_url: review_request_web_url,
1018        }
1019        .into_review_request_row();
1020
1021        (metadata, detail, review_request)
1022    }
1023}
1024
1025/// Aliased nullable `session_review_request` columns loaded through a joined
1026/// session query.
1027struct SessionReviewRequestJoinRow {
1028    display_id: Option<String>,
1029    forge_kind: Option<String>,
1030    last_refreshed_at: Option<i64>,
1031    source_branch: Option<String>,
1032    state: Option<String>,
1033    status_summary: Option<String>,
1034    target_branch: Option<String>,
1035    title: Option<String>,
1036    web_url: Option<String>,
1037}
1038
1039impl SessionReviewRequestJoinRow {
1040    /// Converts the joined nullable columns into a review-request row only
1041    /// when every required field is present.
1042    fn into_review_request_row(self) -> Option<SessionReviewRequestRow> {
1043        let Self {
1044            display_id,
1045            forge_kind,
1046            last_refreshed_at,
1047            source_branch,
1048            state,
1049            status_summary,
1050            target_branch,
1051            title,
1052            web_url,
1053        } = self;
1054
1055        Some(SessionReviewRequestRow {
1056            display_id: display_id?,
1057            forge_kind: forge_kind?,
1058            last_refreshed_at: last_refreshed_at?,
1059            source_branch: source_branch?,
1060            state: state?,
1061            status_summary,
1062            target_branch: target_branch?,
1063            title: title?,
1064            web_url: web_url?,
1065        })
1066    }
1067}
1068
1069#[async_trait]
1070impl SessionRepository for SqliteSessionRepository {
1071    async fn append_session_message(
1072        &self,
1073        id: &str,
1074        kind: SessionMessageKind,
1075        content: &str,
1076    ) -> Result<(), DbError> {
1077        self.2.append(id, kind, content).await
1078    }
1079
1080    async fn backfill_session_project(&self, project_id: i64) -> Result<(), DbError> {
1081        let now = self.now();
1082
1083        sqlx::query!(
1084            r"
1085UPDATE session
1086SET project_id = ?,
1087    updated_at = ?
1088WHERE project_id IS NULL
1089",
1090            project_id,
1091            now
1092        )
1093        .execute(&self.0)
1094        .await?;
1095
1096        Ok(())
1097    }
1098
1099    async fn defer_session_focused_review(&self, id: &str) -> Result<bool, DbError> {
1100        let now = self.now();
1101        let result = sqlx::query!(
1102            r#"
1103UPDATE session
1104SET focused_review_status = 'Pending',
1105    focused_review_diff_hash = NULL,
1106    focused_review_text = NULL,
1107    updated_at = ?
1108WHERE id = ?
1109  AND status IN ('InProgress', 'Review', 'AgentReview')
1110  AND (role IS NULL OR role <> 'Orchestrator')
1111"#,
1112            now,
1113            id
1114        )
1115        .execute(&self.0)
1116        .await?;
1117
1118        Ok(result.rows_affected() > 0)
1119    }
1120
1121    async fn delete_session(&self, id: &str) -> Result<(), DbError> {
1122        let now = self.now();
1123        let mut transaction = self.0.begin().await?;
1124
1125        // Retarget any stacked children onto this session's base branch before
1126        // the row is removed. The `ON DELETE SET NULL` foreign key clears the
1127        // child parent link automatically, but it leaves children pointing at
1128        // the deleted parent's worktree branch, which no longer exists. Mirror
1129        // the post-merge restack so a surviving child rebases against the
1130        // parent's base branch instead of an orphaned `wt/<parent>` ref.
1131        sqlx::query!(
1132            r"
1133UPDATE session
1134SET parent_session_id = NULL,
1135    base_branch = COALESCE((SELECT base_branch FROM session WHERE id = ?), base_branch),
1136    updated_at = ?
1137WHERE parent_session_id = ?
1138  AND status <> 'Canceled'
1139",
1140            id,
1141            now,
1142            id
1143        )
1144        .execute(&mut *transaction)
1145        .await?;
1146
1147        sqlx::query!(
1148            r"
1149DELETE FROM session
1150WHERE id = ?
1151",
1152            id
1153        )
1154        .execute(&mut *transaction)
1155        .await?;
1156
1157        transaction.commit().await?;
1158
1159        Ok(())
1160    }
1161
1162    async fn get_session_base_branch(&self, id: &str) -> Result<Option<String>, DbError> {
1163        let row = sqlx::query_as!(
1164            RequiredStringValueRow,
1165            r#"
1166SELECT base_branch AS "value!: _"
1167FROM session
1168WHERE id = ?
1169"#,
1170            id
1171        )
1172        .fetch_optional(&self.0)
1173        .await?;
1174
1175        Ok(row.map(|row| row.value))
1176    }
1177
1178    async fn get_session_parent_session_id(&self, id: &str) -> Result<Option<String>, DbError> {
1179        let value = sqlx::query_scalar!(
1180            r"
1181SELECT parent_session_id
1182FROM session
1183WHERE id = ?
1184",
1185            id
1186        )
1187        .fetch_optional(&self.0)
1188        .await?
1189        .flatten();
1190
1191        Ok(value)
1192    }
1193
1194    async fn get_session_stack_base_commit_hash(
1195        &self,
1196        id: &str,
1197    ) -> Result<Option<String>, DbError> {
1198        let value = sqlx::query_scalar!(
1199            r"
1200SELECT stack_base_commit_hash
1201FROM session
1202WHERE id = ?
1203",
1204            id
1205        )
1206        .fetch_optional(&self.0)
1207        .await?
1208        .flatten();
1209
1210        Ok(value)
1211    }
1212
1213    async fn get_session_instruction_conversation_id(
1214        &self,
1215        id: &str,
1216    ) -> Result<Option<String>, DbError> {
1217        let row = sqlx::query_as!(
1218            SessionInstructionStateRow,
1219            r"
1220SELECT app_server_instruction_provider_conversation_id
1221FROM session
1222WHERE id = ?
1223",
1224            id
1225        )
1226        .fetch_optional(&self.0)
1227        .await?;
1228
1229        Ok(row.and_then(SessionInstructionStateRow::into_instruction_conversation_id))
1230    }
1231
1232    async fn get_session_provider_conversation_id(
1233        &self,
1234        id: &str,
1235    ) -> Result<Option<String>, DbError> {
1236        let value = sqlx::query_scalar!(
1237            r"SELECT provider_conversation_id FROM session WHERE id = ?",
1238            id
1239        )
1240        .fetch_optional(&self.0)
1241        .await?
1242        .flatten();
1243
1244        Ok(value)
1245    }
1246
1247    async fn insert_draft_session(
1248        &self,
1249        id: &str,
1250        model: &str,
1251        base_branch: &str,
1252        status: &str,
1253        project_id: i64,
1254    ) -> Result<(), DbError> {
1255        let agent = persisted_agent_for_model(model);
1256
1257        insert_session_with_draft_mode(
1258            &self.0,
1259            self.now(),
1260            InsertSessionRow {
1261                agent: &agent,
1262                base_branch,
1263                id,
1264                is_draft: true,
1265                model,
1266                orchestration_task_id: None,
1267                parent_session_id: None,
1268                permission_mode: PermissionMode::AutoEdit,
1269                personality_id: None,
1270                project_id,
1271                reasoning_level: ReasoningLevel::default(),
1272                role: None,
1273                speed_mode: SpeedMode::Normal,
1274                status,
1275            },
1276        )
1277        .await
1278    }
1279
1280    async fn insert_stacked_draft_session(
1281        &self,
1282        id: &str,
1283        model: &str,
1284        base_branch: &str,
1285        status: &str,
1286        parent_session_id: &str,
1287        project_id: i64,
1288    ) -> Result<(), DbError> {
1289        let agent = persisted_agent_for_model(model);
1290
1291        insert_session_with_draft_mode(
1292            &self.0,
1293            self.now(),
1294            InsertSessionRow {
1295                agent: &agent,
1296                base_branch,
1297                id,
1298                is_draft: true,
1299                model,
1300                orchestration_task_id: None,
1301                parent_session_id: Some(parent_session_id),
1302                permission_mode: PermissionMode::AutoEdit,
1303                personality_id: None,
1304                project_id,
1305                reasoning_level: ReasoningLevel::default(),
1306                role: None,
1307                speed_mode: SpeedMode::Normal,
1308                status,
1309            },
1310        )
1311        .await
1312    }
1313
1314    async fn insert_session(
1315        &self,
1316        id: &str,
1317        model: &str,
1318        base_branch: &str,
1319        status: &str,
1320        project_id: i64,
1321    ) -> Result<(), DbError> {
1322        let agent = persisted_agent_for_model(model);
1323
1324        insert_session_with_draft_mode(
1325            &self.0,
1326            self.now(),
1327            InsertSessionRow {
1328                agent: &agent,
1329                base_branch,
1330                id,
1331                is_draft: false,
1332                model,
1333                orchestration_task_id: None,
1334                parent_session_id: None,
1335                permission_mode: PermissionMode::AutoEdit,
1336                personality_id: None,
1337                project_id,
1338                reasoning_level: ReasoningLevel::default(),
1339                role: None,
1340                speed_mode: SpeedMode::Normal,
1341                status,
1342            },
1343        )
1344        .await
1345    }
1346
1347    async fn insert_session_with_agent(
1348        &self,
1349        session: PersistedSessionCreation<'_>,
1350    ) -> Result<(), DbError> {
1351        let PersistedSessionCreation {
1352            agent,
1353            base_branch,
1354            id,
1355            is_draft,
1356            model,
1357            orchestration_task_id,
1358            parent_session_id,
1359            permission_mode,
1360            personality_id,
1361            project_id,
1362            reasoning_level,
1363            role,
1364            speed_mode,
1365            status,
1366        } = session;
1367
1368        insert_session_with_draft_mode(
1369            &self.0,
1370            self.now(),
1371            InsertSessionRow {
1372                agent,
1373                base_branch,
1374                id,
1375                is_draft,
1376                model,
1377                orchestration_task_id,
1378                parent_session_id,
1379                permission_mode,
1380                personality_id,
1381                project_id,
1382                reasoning_level,
1383                role,
1384                speed_mode,
1385                status,
1386            },
1387        )
1388        .await
1389    }
1390
1391    async fn fork_session_snapshot(
1392        &self,
1393        snapshot: ForkSessionSnapshot<'_>,
1394    ) -> Result<(), DbError> {
1395        self.3.fork(snapshot).await
1396    }
1397
1398    async fn load_session(&self, session_id: &str) -> Result<Option<SessionRow>, DbError> {
1399        let row = sqlx::query_as::<_, SessionJoinRow>(
1400            r"
1401SELECT session.base_branch AS base_branch,
1402       session.added_lines AS added_lines,
1403       session.agent AS agent,
1404       session.created_at AS created_at,
1405       session.deleted_lines AS deleted_lines,
1406       session.has_diff AS has_diff,
1407       session.id AS id,
1408       session.in_progress_started_at,
1409       session.in_progress_total_seconds AS in_progress_total_seconds,
1410       session.input_tokens AS input_tokens,
1411       session.is_draft AS is_draft,
1412       session.model AS model,
1413       session.output_tokens AS output_tokens,
1414       session.parent_session_id,
1415       session.permission_mode AS permission_mode,
1416       session.personality_id,
1417       session.project_id,
1418       session.prompt AS prompt,
1419       session.reasoning_level AS reasoning_level_override,
1420       session.speed_mode AS speed_mode,
1421       session.published_upstream_ref,
1422       session.questions,
1423       session_review_request.display_id AS review_request_display_id,
1424       session_review_request.forge_kind AS review_request_forge_kind,
1425       session_review_request.last_refreshed_at AS review_request_last_refreshed_at,
1426       session_review_request.source_branch AS review_request_source_branch,
1427       session_review_request.state AS review_request_state,
1428       session_review_request.status_summary AS review_request_status_summary,
1429       session_review_request.target_branch AS review_request_target_branch,
1430       session_review_request.title AS review_request_title,
1431       session_review_request.web_url AS review_request_web_url,
1432       session.role,
1433       session.size AS size,
1434       session.status AS status,
1435       session.summary,
1436       session.title,
1437       session.updated_at AS updated_at
1438FROM session
1439LEFT JOIN session_review_request
1440ON session_review_request.session_id = session.id
1441WHERE session.id = ?
1442",
1443        )
1444        .bind(session_id)
1445        .fetch_optional(&self.0)
1446        .await?;
1447
1448        let row = row.map(SessionJoinRow::into_session_row);
1449        if let Some(row) = &row {
1450            status::validate_session(&row.status)?;
1451        }
1452
1453        Ok(row)
1454    }
1455
1456    async fn load_active_session_agent_models(&self) -> Result<Vec<SessionAgentModelRow>, DbError> {
1457        let rows = sqlx::query_as::<_, SessionAgentModelRow>(
1458            r"
1459SELECT agent,
1460       id,
1461       model,
1462       status
1463FROM session
1464WHERE status NOT IN ('Merged', 'Done', 'Canceled')
1465ORDER BY id
1466",
1467        )
1468        .fetch_all(&self.0)
1469        .await?;
1470
1471        Ok(rows)
1472    }
1473
1474    #[cfg(any(test, feature = "test-utils"))]
1475    async fn load_sessions(&self) -> Result<Vec<SessionRow>, DbError> {
1476        let rows = sqlx::query_as!(
1477            SessionJoinRow,
1478            r#"
1479SELECT session.base_branch AS base_branch,
1480       session.added_lines AS added_lines,
1481       session.agent AS agent,
1482       session.created_at AS created_at,
1483       session.deleted_lines AS deleted_lines,
1484       session.has_diff AS "has_diff: bool",
1485       session.id AS id,
1486       session.in_progress_started_at,
1487       session.in_progress_total_seconds AS in_progress_total_seconds,
1488       session.input_tokens AS input_tokens,
1489       session.is_draft AS "is_draft: bool",
1490       session.model AS model,
1491       session.output_tokens AS output_tokens,
1492       session.parent_session_id,
1493       session.permission_mode AS permission_mode,
1494       session.personality_id,
1495       session.project_id,
1496       session.prompt AS prompt,
1497       session.reasoning_level AS reasoning_level_override,
1498       session.speed_mode AS speed_mode,
1499       session.published_upstream_ref,
1500       session.questions,
1501       session_review_request.display_id AS review_request_display_id,
1502       session_review_request.forge_kind AS review_request_forge_kind,
1503       session_review_request.last_refreshed_at AS review_request_last_refreshed_at,
1504       session_review_request.source_branch AS review_request_source_branch,
1505       session_review_request.state AS review_request_state,
1506       session_review_request.status_summary AS review_request_status_summary,
1507       session_review_request.target_branch AS review_request_target_branch,
1508       session_review_request.title AS review_request_title,
1509       session_review_request.web_url AS review_request_web_url,
1510       session.role,
1511       session.size AS size,
1512       session.status AS status,
1513       session.summary,
1514       session.title,
1515       session.updated_at AS updated_at
1516FROM session
1517LEFT JOIN session_review_request
1518ON session_review_request.session_id = session.id
1519ORDER BY session.updated_at DESC, session.created_at DESC, session.id
1520"#
1521        )
1522        .fetch_all(&self.0)
1523        .await?;
1524
1525        let rows = rows
1526            .into_iter()
1527            .filter(SessionJoinRow::has_loadable_status)
1528            .map(SessionJoinRow::into_session_row)
1529            .collect::<Vec<_>>();
1530
1531        Ok(rows)
1532    }
1533
1534    async fn load_sessions_for_project(
1535        &self,
1536        project_id: i64,
1537    ) -> Result<Vec<SessionListRow>, DbError> {
1538        let rows = sqlx::query_as!(
1539            SessionJoinRow,
1540            r#"
1541SELECT session.base_branch AS base_branch,
1542       session.added_lines AS added_lines,
1543       session.agent AS agent,
1544       session.created_at AS created_at,
1545       session.deleted_lines AS deleted_lines,
1546       session.has_diff AS "has_diff: bool",
1547       session.id AS id,
1548       session.in_progress_started_at,
1549       session.in_progress_total_seconds AS in_progress_total_seconds,
1550       session.input_tokens AS input_tokens,
1551       session.is_draft AS "is_draft: bool",
1552       session.model AS model,
1553       session.output_tokens AS output_tokens,
1554       session.parent_session_id,
1555       session.permission_mode AS permission_mode,
1556       session.personality_id,
1557       session.project_id,
1558       '' AS "prompt!: String",
1559       session.reasoning_level AS reasoning_level_override,
1560       session.speed_mode AS speed_mode,
1561       session.published_upstream_ref,
1562       NULL AS "questions: String",
1563       session_review_request.display_id AS review_request_display_id,
1564       session_review_request.forge_kind AS review_request_forge_kind,
1565       session_review_request.last_refreshed_at AS review_request_last_refreshed_at,
1566       session_review_request.source_branch AS review_request_source_branch,
1567       session_review_request.state AS review_request_state,
1568       session_review_request.status_summary AS review_request_status_summary,
1569       session_review_request.target_branch AS review_request_target_branch,
1570       session_review_request.title AS review_request_title,
1571       session_review_request.web_url AS review_request_web_url,
1572       session.role,
1573       session.size AS size,
1574       session.status AS status,
1575       NULL AS "summary: String",
1576       session.title,
1577       session.updated_at AS updated_at
1578FROM session
1579LEFT JOIN session_review_request
1580ON session_review_request.session_id = session.id
1581WHERE session.project_id = ?
1582ORDER BY session.updated_at DESC, session.created_at DESC, session.id
1583"#,
1584            project_id
1585        )
1586        .fetch_all(&self.0)
1587        .await?;
1588
1589        let rows = rows
1590            .into_iter()
1591            .filter(SessionJoinRow::has_loadable_status)
1592            .map(SessionJoinRow::into_session_list_row)
1593            .collect::<Vec<_>>();
1594
1595        Ok(rows)
1596    }
1597
1598    async fn load_session_detail(
1599        &self,
1600        session_id: &str,
1601    ) -> Result<Option<SessionDetailRow>, DbError> {
1602        let row = sqlx::query_as!(
1603            SessionDetailRow,
1604            r"
1605SELECT prompt,
1606       questions,
1607       summary
1608FROM session
1609WHERE id = ?
1610",
1611            session_id
1612        )
1613        .fetch_optional(&self.0)
1614        .await?;
1615
1616        Ok(row)
1617    }
1618
1619    async fn load_session_messages(
1620        &self,
1621        session_id: &str,
1622    ) -> Result<Vec<SessionMessageRow>, DbError> {
1623        let rows = sqlx::query_as!(
1624            SessionMessageRow,
1625            r"
1626SELECT content,
1627       kind,
1628       position
1629FROM session_message
1630WHERE session_id = ?
1631ORDER BY position, id
1632",
1633            session_id
1634        )
1635        .fetch_all(&self.0)
1636        .await?;
1637
1638        Ok(rows)
1639    }
1640
1641    async fn load_session_focused_reviews_for_project(
1642        &self,
1643        project_id: i64,
1644    ) -> Result<Vec<SessionFocusedReviewRow>, DbError> {
1645        let rows = sqlx::query_as!(
1646            SessionFocusedReviewRow,
1647            r#"
1648SELECT id AS session_id,
1649       focused_review_diff_hash AS "diff_hash!: String",
1650       focused_review_text AS "text!: String"
1651FROM session
1652WHERE project_id = ?
1653  AND focused_review_diff_hash IS NOT NULL
1654  AND focused_review_text IS NOT NULL
1655  AND focused_review_text <> ''
1656ORDER BY updated_at DESC, id
1657"#,
1658            project_id
1659        )
1660        .fetch_all(&self.0)
1661        .await?;
1662
1663        Ok(rows)
1664    }
1665
1666    async fn load_sessions_metadata(&self) -> Result<(i64, i64), DbError> {
1667        let row = sqlx::query_as!(
1668            SessionStatsMetadataRow,
1669            r#"
1670SELECT (SELECT COUNT(*) FROM session) AS "session_count!: _",
1671       COALESCE(
1672           (
1673               SELECT updated_at
1674               FROM session
1675               ORDER BY updated_at DESC, id
1676               LIMIT 1
1677           ),
1678           0
1679       ) AS "max_updated_at!: _"
1680"#
1681        )
1682        .fetch_one(&self.0)
1683        .await?;
1684
1685        Ok((row.session_count, row.max_updated_at))
1686    }
1687
1688    async fn load_session_project_id(&self, session_id: &str) -> Result<Option<i64>, DbError> {
1689        let row = sqlx::query_as!(
1690            OptionalI64ValueRow,
1691            r#"
1692SELECT project_id AS "value: _"
1693FROM session
1694WHERE id = ?
1695"#,
1696            session_id
1697        )
1698        .fetch_optional(&self.0)
1699        .await?;
1700
1701        Ok(row.and_then(|row| row.value))
1702    }
1703
1704    async fn load_session_personality_state(
1705        &self,
1706        session_id: &str,
1707    ) -> Result<Option<SessionPersonalityState>, DbError> {
1708        let state = sqlx::query_as!(
1709            SessionPersonalityState,
1710            r"
1711SELECT applied_personality_id,
1712       applied_personality_prompt_hash,
1713       personality_id
1714FROM session
1715WHERE id = ?
1716",
1717            session_id
1718        )
1719        .fetch_optional(&self.0)
1720        .await?;
1721
1722        Ok(state)
1723    }
1724
1725    async fn load_pending_stack_restack_session_ids(
1726        &self,
1727        project_id: i64,
1728    ) -> Result<Vec<String>, DbError> {
1729        let session_ids = sqlx::query_scalar!(
1730            r"
1731SELECT id
1732FROM session
1733WHERE project_id = ?
1734  AND parent_session_id IS NULL
1735  AND stack_base_commit_hash IS NOT NULL
1736  AND status IN ('Review', 'AgentReview')
1737ORDER BY updated_at ASC, id ASC
1738",
1739            project_id
1740        )
1741        .fetch_all(&self.0)
1742        .await?;
1743
1744        Ok(session_ids)
1745    }
1746
1747    async fn load_pending_focused_review_session_ids(
1748        &self,
1749        project_id: i64,
1750    ) -> Result<Vec<String>, DbError> {
1751        let session_ids = sqlx::query_scalar!(
1752            r#"
1753SELECT id
1754FROM session
1755WHERE project_id = ?
1756  AND focused_review_status = 'Pending'
1757  AND status IN ('Review', 'AgentReview')
1758  AND (role IS NULL OR role <> 'Orchestrator')
1759ORDER BY updated_at DESC, id
1760"#,
1761            project_id
1762        )
1763        .fetch_all(&self.0)
1764        .await?;
1765
1766        Ok(session_ids)
1767    }
1768
1769    async fn load_session_published_upstream_ref(
1770        &self,
1771        id: &str,
1772    ) -> Result<Option<String>, DbError> {
1773        let value = sqlx::query_scalar!(
1774            r"SELECT published_upstream_ref FROM session WHERE id = ?",
1775            id
1776        )
1777        .fetch_optional(&self.0)
1778        .await?
1779        .flatten();
1780
1781        Ok(value)
1782    }
1783
1784    async fn load_session_merged_commit_hash(
1785        &self,
1786        session_id: &str,
1787    ) -> Result<Option<String>, DbError> {
1788        let row = sqlx::query_scalar!(
1789            r"
1790SELECT merged_commit_hash
1791FROM session
1792WHERE id = ?
1793",
1794            session_id
1795        )
1796        .fetch_optional(&self.0)
1797        .await?;
1798
1799        Ok(row.flatten())
1800    }
1801
1802    async fn load_session_archived_diff(
1803        &self,
1804        session_id: &str,
1805    ) -> Result<Option<String>, DbError> {
1806        let row = sqlx::query_scalar!(
1807            r"
1808SELECT archived_diff
1809FROM session
1810WHERE id = ?
1811",
1812            session_id
1813        )
1814        .fetch_optional(&self.0)
1815        .await?;
1816
1817        Ok(row.flatten())
1818    }
1819
1820    async fn load_session_reasoning_level(
1821        &self,
1822        session_id: &str,
1823    ) -> Result<ReasoningLevel, DbError> {
1824        let value = sqlx::query_scalar!(
1825            r"SELECT reasoning_level FROM session WHERE id = ?",
1826            session_id
1827        )
1828        .fetch_optional(&self.0)
1829        .await?
1830        .flatten();
1831
1832        Ok(value
1833            .and_then(|value| value.parse::<ReasoningLevel>().ok())
1834            .unwrap_or_default())
1835    }
1836
1837    async fn load_session_permission_mode(
1838        &self,
1839        session_id: &str,
1840    ) -> Result<PermissionMode, DbError> {
1841        let value = sqlx::query_scalar!(
1842            r"SELECT permission_mode FROM session WHERE id = ?",
1843            session_id
1844        )
1845        .fetch_optional(&self.0)
1846        .await?;
1847
1848        Ok(value
1849            .and_then(|value| value.parse::<PermissionMode>().ok())
1850            .unwrap_or_default())
1851    }
1852
1853    async fn load_session_speed_mode(&self, session_id: &str) -> Result<SpeedMode, DbError> {
1854        let value = sqlx::query_scalar!(r"SELECT speed_mode FROM session WHERE id = ?", session_id)
1855            .fetch_optional(&self.0)
1856            .await?;
1857
1858        Ok(value
1859            .and_then(|value| value.parse::<SpeedMode>().ok())
1860            .unwrap_or_default())
1861    }
1862
1863    async fn restack_child_sessions_after_parent_merge(
1864        &self,
1865        parent_session_id: &str,
1866        base_branch: &str,
1867        parent_commit_hash: Option<String>,
1868    ) -> Result<Vec<String>, DbError> {
1869        let now = self.now();
1870        let mut transaction = self.0.begin().await?;
1871        let materialized_child_ids = sqlx::query_scalar!(
1872            r"
1873SELECT id
1874FROM session
1875WHERE parent_session_id = ?
1876  AND status NOT IN ('Canceled', 'Draft')
1877ORDER BY created_at ASC, id ASC
1878",
1879            parent_session_id
1880        )
1881        .fetch_all(&mut *transaction)
1882        .await?;
1883
1884        sqlx::query!(
1885            r"
1886UPDATE session
1887SET parent_session_id = NULL,
1888    base_branch = ?,
1889    stack_base_commit_hash = CASE
1890        WHEN status = 'Draft' THEN NULL
1891        ELSE COALESCE(stack_base_commit_hash, ?)
1892    END,
1893    updated_at = ?
1894WHERE parent_session_id = ?
1895  AND status <> 'Canceled'
1896",
1897            base_branch,
1898            parent_commit_hash,
1899            now,
1900            parent_session_id
1901        )
1902        .execute(&mut *transaction)
1903        .await?;
1904
1905        transaction.commit().await?;
1906
1907        Ok(materialized_child_ids)
1908    }
1909
1910    async fn load_session_summary(&self, session_id: &str) -> Result<Option<String>, DbError> {
1911        let row = sqlx::query_scalar!(
1912            r"
1913SELECT summary
1914FROM session
1915WHERE id = ?
1916",
1917            session_id
1918        )
1919        .fetch_optional(&self.0)
1920        .await?;
1921
1922        Ok(row.flatten())
1923    }
1924
1925    async fn load_session_timestamps(
1926        &self,
1927        session_id: &str,
1928    ) -> Result<Option<(i64, i64)>, DbError> {
1929        let row = sqlx::query_as!(
1930            SessionTimestampsRow,
1931            r#"
1932SELECT created_at, updated_at
1933FROM session
1934WHERE id = ?
1935            "#,
1936            session_id
1937        )
1938        .fetch_optional(&self.0)
1939        .await?;
1940
1941        Ok(row.map(|row| (row.created_at, row.updated_at)))
1942    }
1943
1944    async fn persist_session_turn_metadata(
1945        &self,
1946        session_id: &str,
1947        turn_metadata: &SessionTurnMetadata,
1948    ) -> Result<(), DbError> {
1949        let now = self.now();
1950        let mut transaction = self.0.begin().await?;
1951
1952        let session_update = sqlx::query!(
1953            r"
1954UPDATE session
1955SET questions = ?,
1956    summary = ?,
1957    provider_conversation_id = ?,
1958    app_server_instruction_provider_conversation_id = ?,
1959    applied_personality_id = ?,
1960    applied_personality_prompt_hash = ?,
1961    updated_at = ?
1962WHERE id = ?
1963",
1964            turn_metadata.questions_json.as_str(),
1965            turn_metadata.summary.as_str(),
1966            turn_metadata.provider_conversation_id.as_deref(),
1967            turn_metadata.instruction_conversation_id.as_deref(),
1968            turn_metadata.applied_personality_id.as_deref(),
1969            turn_metadata.applied_personality_prompt_hash.as_deref(),
1970            now,
1971            session_id
1972        )
1973        .execute(&mut *transaction)
1974        .await?;
1975        if session_update.rows_affected() != 1 {
1976            return Err(sqlx::Error::RowNotFound.into());
1977        }
1978
1979        if turn_metadata.token_usage_delta.input_tokens != 0
1980            || turn_metadata.token_usage_delta.output_tokens != 0
1981        {
1982            sqlx::query!(
1983                r"
1984UPDATE session
1985SET input_tokens = input_tokens + ?,
1986    output_tokens = output_tokens + ?,
1987    updated_at = ?
1988WHERE id = ?
1989",
1990                turn_metadata.token_usage_delta.input_tokens.cast_signed(),
1991                turn_metadata.token_usage_delta.output_tokens.cast_signed(),
1992                now,
1993                session_id
1994            )
1995            .execute(&mut *transaction)
1996            .await?;
1997
1998            sqlx::query!(
1999                r"
2000INSERT INTO session_usage (
2001    session_id, model, created_at, input_tokens, output_tokens, invocation_count
2002)
2003VALUES (?, ?, ?, ?, ?, 1)
2004ON CONFLICT(session_id, model) DO UPDATE SET
2005    input_tokens = input_tokens + excluded.input_tokens,
2006    output_tokens = output_tokens + excluded.output_tokens,
2007    invocation_count = invocation_count + 1
2008",
2009                session_id,
2010                turn_metadata.model.as_str(),
2011                now,
2012                turn_metadata.token_usage_delta.input_tokens.cast_signed(),
2013                turn_metadata.token_usage_delta.output_tokens.cast_signed()
2014            )
2015            .execute(&mut *transaction)
2016            .await?;
2017        }
2018
2019        insert_review_comment_resolutions(
2020            &mut transaction,
2021            session_id,
2022            &turn_metadata.review_comment_resolutions,
2023        )
2024        .await?;
2025
2026        transaction.commit().await?;
2027
2028        Ok(())
2029    }
2030
2031    async fn update_session_diff_stats(
2032        &self,
2033        added_lines: u64,
2034        deleted_lines: u64,
2035        has_diff: bool,
2036        id: &str,
2037        size: &str,
2038    ) -> Result<(), DbError> {
2039        let now = self.now();
2040
2041        sqlx::query!(
2042            r"
2043UPDATE session
2044SET added_lines = ?,
2045    deleted_lines = ?,
2046    has_diff = ?,
2047    size = ?,
2048    updated_at = ?
2049WHERE id = ?
2050  AND (
2051      added_lines <> ?
2052      OR deleted_lines <> ?
2053      OR has_diff IS NOT ?
2054      OR size <> ?
2055  )
2056",
2057            added_lines.cast_signed(),
2058            deleted_lines.cast_signed(),
2059            has_diff,
2060            size,
2061            now,
2062            id,
2063            added_lines.cast_signed(),
2064            deleted_lines.cast_signed(),
2065            has_diff,
2066            size
2067        )
2068        .execute(&self.0)
2069        .await?;
2070
2071        Ok(())
2072    }
2073
2074    async fn mark_session_diff_unknown(&self, id: &str) -> Result<(), DbError> {
2075        let now = self.now();
2076
2077        sqlx::query!(
2078            r"
2079UPDATE session
2080SET has_diff = NULL,
2081    updated_at = ?
2082WHERE id = ?
2083  AND has_diff IS NOT NULL
2084",
2085            now,
2086            id
2087        )
2088        .execute(&self.0)
2089        .await?;
2090
2091        Ok(())
2092    }
2093
2094    async fn update_session_instruction_conversation_id(
2095        &self,
2096        id: &str,
2097        provider_conversation_id: Option<String>,
2098    ) -> Result<(), DbError> {
2099        let now = self.now();
2100
2101        sqlx::query!(
2102            r"
2103UPDATE session
2104SET app_server_instruction_provider_conversation_id = ?,
2105    updated_at = ?
2106WHERE id = ?
2107",
2108            provider_conversation_id.as_deref(),
2109            now,
2110            id
2111        )
2112        .execute(&self.0)
2113        .await?;
2114
2115        Ok(())
2116    }
2117
2118    async fn update_session_model(&self, id: &str, model: &str) -> Result<(), DbError> {
2119        let agent = persisted_agent_for_model(model);
2120        let now = self.now();
2121
2122        sqlx::query!(
2123            r"
2124UPDATE session
2125SET agent = ?,
2126    model = ?,
2127    updated_at = ?
2128WHERE id = ?
2129",
2130            agent,
2131            model,
2132            now,
2133            id
2134        )
2135        .execute(&self.0)
2136        .await?;
2137
2138        Ok(())
2139    }
2140
2141    async fn update_session_personality_id(
2142        &self,
2143        id: &str,
2144        personality_id: Option<String>,
2145    ) -> Result<(), DbError> {
2146        let now = self.now();
2147
2148        sqlx::query!(
2149            r"
2150UPDATE session
2151SET personality_id = ?,
2152    updated_at = ?
2153WHERE id = ?
2154",
2155            personality_id.as_deref(),
2156            now,
2157            id
2158        )
2159        .execute(&self.0)
2160        .await?;
2161
2162        Ok(())
2163    }
2164
2165    async fn update_session_agent_model(
2166        &self,
2167        id: &str,
2168        agent: &str,
2169        model: &str,
2170    ) -> Result<(), DbError> {
2171        let now = self.now();
2172
2173        sqlx::query!(
2174            r"
2175UPDATE session
2176SET agent = ?,
2177    model = ?,
2178    updated_at = ?
2179WHERE id = ?
2180",
2181            agent,
2182            model,
2183            now,
2184            id
2185        )
2186        .execute(&self.0)
2187        .await?;
2188
2189        Ok(())
2190    }
2191
2192    async fn update_active_session_agent_model(
2193        &self,
2194        id: &str,
2195        agent: &str,
2196        model: &str,
2197    ) -> Result<(), DbError> {
2198        sqlx::query!(
2199            r"
2200UPDATE session
2201SET agent = ?,
2202    model = ?
2203WHERE id = ?
2204  AND status NOT IN ('Merged', 'Done', 'Canceled')
2205",
2206            agent,
2207            model,
2208            id
2209        )
2210        .execute(&self.0)
2211        .await?;
2212
2213        Ok(())
2214    }
2215
2216    async fn clear_session_draft_flag(&self, id: &str) -> Result<(), DbError> {
2217        let now = self.now();
2218
2219        sqlx::query!(
2220            r"
2221UPDATE session
2222SET is_draft = 0,
2223    updated_at = ?
2224WHERE id = ?
2225",
2226            now,
2227            id
2228        )
2229        .execute(&self.0)
2230        .await?;
2231
2232        Ok(())
2233    }
2234
2235    async fn update_session_merged_commit_hash(
2236        &self,
2237        id: &str,
2238        merged_commit_hash: Option<String>,
2239    ) -> Result<(), DbError> {
2240        let now = self.now();
2241
2242        sqlx::query!(
2243            r"
2244UPDATE session
2245SET merged_commit_hash = ?,
2246    updated_at = ?
2247WHERE id = ?
2248",
2249            merged_commit_hash.as_deref(),
2250            now,
2251            id
2252        )
2253        .execute(&self.0)
2254        .await?;
2255
2256        Ok(())
2257    }
2258
2259    async fn update_session_archived_diff(
2260        &self,
2261        id: &str,
2262        archived_diff: Option<String>,
2263    ) -> Result<(), DbError> {
2264        let now = self.now();
2265
2266        sqlx::query!(
2267            r"
2268UPDATE session
2269SET archived_diff = ?,
2270    updated_at = ?
2271WHERE id = ?
2272",
2273            archived_diff.as_deref(),
2274            now,
2275            id
2276        )
2277        .execute(&self.0)
2278        .await?;
2279
2280        Ok(())
2281    }
2282
2283    async fn update_session_stack_base_commit_hash(
2284        &self,
2285        id: &str,
2286        stack_base_commit_hash: Option<String>,
2287    ) -> Result<(), DbError> {
2288        let now = self.now();
2289
2290        sqlx::query(
2291            r"
2292UPDATE session
2293SET stack_base_commit_hash = ?,
2294    updated_at = ?
2295WHERE id = ?
2296",
2297        )
2298        .bind(stack_base_commit_hash)
2299        .bind(now)
2300        .bind(id)
2301        .execute(&self.0)
2302        .await?;
2303
2304        Ok(())
2305    }
2306
2307    async fn update_session_prompt(&self, id: &str, prompt: &str) -> Result<(), DbError> {
2308        let now = self.now();
2309
2310        sqlx::query!(
2311            r"
2312UPDATE session
2313SET prompt = ?,
2314    updated_at = ?
2315WHERE id = ?
2316",
2317            prompt,
2318            now,
2319            id
2320        )
2321        .execute(&self.0)
2322        .await?;
2323
2324        Ok(())
2325    }
2326
2327    async fn update_session_provider_conversation_id(
2328        &self,
2329        id: &str,
2330        provider_conversation_id: Option<String>,
2331    ) -> Result<(), DbError> {
2332        let now = self.now();
2333
2334        sqlx::query!(
2335            r"
2336UPDATE session
2337SET provider_conversation_id = ?,
2338    updated_at = ?
2339WHERE id = ?
2340",
2341            provider_conversation_id.as_deref(),
2342            now,
2343            id
2344        )
2345        .execute(&self.0)
2346        .await?;
2347
2348        Ok(())
2349    }
2350
2351    async fn update_session_questions(&self, id: &str, questions: &str) -> Result<(), DbError> {
2352        let now = self.now();
2353
2354        sqlx::query!(
2355            r"
2356UPDATE session
2357SET questions = ?,
2358    updated_at = ?
2359WHERE id = ?
2360",
2361            questions,
2362            now,
2363            id
2364        )
2365        .execute(&self.0)
2366        .await?;
2367
2368        Ok(())
2369    }
2370
2371    async fn update_session_reasoning_level(
2372        &self,
2373        id: &str,
2374        reasoning_level: ReasoningLevel,
2375    ) -> Result<(), DbError> {
2376        let now = self.now();
2377
2378        sqlx::query!(
2379            r#"
2380UPDATE session
2381SET reasoning_level = ?,
2382    updated_at = ?
2383WHERE id = ?
2384            "#,
2385            reasoning_level.as_str(),
2386            now,
2387            id
2388        )
2389        .execute(&self.0)
2390        .await?;
2391
2392        Ok(())
2393    }
2394
2395    async fn update_session_permission_mode(
2396        &self,
2397        id: &str,
2398        permission_mode: PermissionMode,
2399    ) -> Result<(), DbError> {
2400        let now = self.now();
2401
2402        sqlx::query!(
2403            r#"
2404UPDATE session
2405SET permission_mode = ?,
2406    updated_at = ?
2407WHERE id = ?
2408            "#,
2409            permission_mode.label(),
2410            now,
2411            id
2412        )
2413        .execute(&self.0)
2414        .await?;
2415
2416        Ok(())
2417    }
2418
2419    async fn update_session_speed_mode(
2420        &self,
2421        id: &str,
2422        speed_mode: SpeedMode,
2423    ) -> Result<(), DbError> {
2424        let now = self.now();
2425
2426        sqlx::query!(
2427            r#"
2428UPDATE session
2429SET speed_mode = ?,
2430    updated_at = ?
2431WHERE id = ?
2432            "#,
2433            speed_mode.as_str(),
2434            now,
2435            id
2436        )
2437        .execute(&self.0)
2438        .await?;
2439
2440        Ok(())
2441    }
2442
2443    async fn update_session_published_upstream_ref(
2444        &self,
2445        id: &str,
2446        published_upstream_ref: Option<String>,
2447    ) -> Result<(), DbError> {
2448        let now = self.now();
2449
2450        sqlx::query!(
2451            r"
2452UPDATE session
2453SET published_upstream_ref = ?,
2454    updated_at = ?
2455WHERE id = ?
2456",
2457            published_upstream_ref.as_deref(),
2458            now,
2459            id
2460        )
2461        .execute(&self.0)
2462        .await?;
2463
2464        Ok(())
2465    }
2466
2467    async fn update_session_stats(&self, id: &str, stats: &SessionStats) -> Result<(), DbError> {
2468        if stats.input_tokens == 0 && stats.output_tokens == 0 {
2469            return Ok(());
2470        }
2471
2472        let now = self.now();
2473
2474        sqlx::query!(
2475            r"
2476UPDATE session
2477SET input_tokens = input_tokens + ?,
2478    output_tokens = output_tokens + ?,
2479    updated_at = ?
2480WHERE id = ?
2481",
2482            stats.input_tokens.cast_signed(),
2483            stats.output_tokens.cast_signed(),
2484            now,
2485            id
2486        )
2487        .execute(&self.0)
2488        .await?;
2489
2490        Ok(())
2491    }
2492
2493    async fn update_session_status_with_timing_at(
2494        &self,
2495        id: &str,
2496        status: &str,
2497        timestamp_seconds: i64,
2498    ) -> Result<(), DbError> {
2499        status::validate_session(status)?;
2500        let now = self.now();
2501
2502        sqlx::query!(
2503            r"
2504UPDATE session
2505SET status = ?,
2506    in_progress_total_seconds = CASE
2507        WHEN ? = 'InProgress' OR in_progress_started_at IS NULL THEN in_progress_total_seconds
2508        ELSE in_progress_total_seconds + MAX(0, ? - in_progress_started_at)
2509    END,
2510    in_progress_started_at = CASE
2511        WHEN ? = 'InProgress' THEN COALESCE(in_progress_started_at, ?)
2512        ELSE NULL
2513    END,
2514    updated_at = ?
2515WHERE id = ?
2516",
2517            status,
2518            status,
2519            timestamp_seconds,
2520            status,
2521            timestamp_seconds,
2522            now,
2523            id
2524        )
2525        .execute(&self.0)
2526        .await?;
2527
2528        Ok(())
2529    }
2530
2531    async fn update_session_summary(&self, id: &str, summary: &str) -> Result<(), DbError> {
2532        let now = self.now();
2533
2534        sqlx::query!(
2535            r"
2536UPDATE session
2537SET summary = ?,
2538    updated_at = ?
2539WHERE id = ?
2540",
2541            summary,
2542            now,
2543            id
2544        )
2545        .execute(&self.0)
2546        .await?;
2547
2548        Ok(())
2549    }
2550
2551    async fn update_session_focused_review(
2552        &self,
2553        id: &str,
2554        status: Option<FocusedReviewStatus>,
2555        diff_hash: Option<String>,
2556        text: Option<String>,
2557    ) -> Result<(), DbError> {
2558        let now = self.now();
2559
2560        sqlx::query!(
2561            r"
2562UPDATE session
2563SET focused_review_status = ?,
2564    focused_review_diff_hash = ?,
2565    focused_review_text = ?,
2566    updated_at = ?
2567WHERE id = ?
2568",
2569            status.map(|status| status.to_string()),
2570            diff_hash.as_deref(),
2571            text.as_deref(),
2572            now,
2573            id
2574        )
2575        .execute(&self.0)
2576        .await?;
2577
2578        Ok(())
2579    }
2580
2581    async fn update_session_title(&self, id: &str, title: &str) -> Result<(), DbError> {
2582        let now = self.now();
2583
2584        sqlx::query!(
2585            r#"
2586UPDATE session
2587SET title = ?,
2588    is_title_provisional = 0,
2589    title_generation = title_generation + 1,
2590    applied_title_generation = title_generation + 1,
2591    updated_at = ?
2592WHERE id = ?
2593"#,
2594            title,
2595            now,
2596            id,
2597        )
2598        .execute(&self.0)
2599        .await?;
2600
2601        Ok(())
2602    }
2603
2604    async fn update_session_provisional_title(&self, id: &str, title: &str) -> Result<(), DbError> {
2605        let now = self.now();
2606
2607        sqlx::query!(
2608            r#"
2609UPDATE session
2610SET title = ?,
2611    is_title_provisional = 1,
2612    title_generation = title_generation + 1,
2613    applied_title_generation = title_generation + 1,
2614    updated_at = ?
2615WHERE id = ?
2616"#,
2617            title,
2618            now,
2619            id,
2620        )
2621        .execute(&self.0)
2622        .await?;
2623
2624        Ok(())
2625    }
2626
2627    async fn begin_session_title_generation(
2628        &self,
2629        id: &str,
2630        requires_provisional_title: bool,
2631    ) -> Result<Option<i64>, DbError> {
2632        let now = self.now();
2633        let generation = if requires_provisional_title {
2634            sqlx::query_scalar!(
2635                r#"
2636UPDATE session
2637SET is_title_provisional = 1,
2638    title_generation = title_generation + 1,
2639    updated_at = ?
2640WHERE id = ?
2641  AND is_title_provisional = 1
2642RETURNING title_generation
2643"#,
2644                now,
2645                id,
2646            )
2647            .fetch_optional(&self.0)
2648            .await?
2649        } else {
2650            sqlx::query_scalar!(
2651                r#"
2652UPDATE session
2653SET is_title_provisional = 1,
2654    title_generation = title_generation + 1,
2655    updated_at = ?
2656WHERE id = ?
2657RETURNING title_generation
2658"#,
2659                now,
2660                id,
2661            )
2662            .fetch_optional(&self.0)
2663            .await?
2664        };
2665
2666        Ok(generation)
2667    }
2668
2669    async fn update_session_title_for_generation(
2670        &self,
2671        id: &str,
2672        expected_generation: i64,
2673        title: &str,
2674    ) -> Result<bool, DbError> {
2675        let now = self.now();
2676
2677        let result = sqlx::query!(
2678            r#"
2679UPDATE session
2680SET title = ?,
2681    is_title_provisional = 0,
2682    applied_title_generation = ?,
2683    updated_at = ?
2684WHERE id = ?
2685  AND title_generation >= ?
2686  AND applied_title_generation < ?
2687"#,
2688            title,
2689            expected_generation,
2690            now,
2691            id,
2692            expected_generation,
2693            expected_generation,
2694        )
2695        .execute(&self.0)
2696        .await?;
2697
2698        Ok(result.rows_affected() > 0)
2699    }
2700
2701    #[cfg(any(test, feature = "test-utils"))]
2702    async fn update_session_created_at(&self, id: &str, created_at: i64) -> Result<(), DbError> {
2703        sqlx::query!(
2704            r"
2705UPDATE session
2706SET created_at = ?
2707WHERE id = ?
2708",
2709            created_at,
2710            id
2711        )
2712        .execute(&self.0)
2713        .await?;
2714
2715        Ok(())
2716    }
2717
2718    #[cfg(any(test, feature = "test-utils"))]
2719    async fn update_session_updated_at(&self, id: &str, updated_at: i64) -> Result<(), DbError> {
2720        sqlx::query!(
2721            r"
2722UPDATE session
2723SET updated_at = ?
2724WHERE id = ?
2725",
2726            updated_at,
2727            id
2728        )
2729        .execute(&self.0)
2730        .await?;
2731
2732        Ok(())
2733    }
2734}
2735
2736/// Borrowed values used to insert one newly created session row.
2737struct InsertSessionRow<'a> {
2738    /// Agent provider kind persisted alongside the model for this session.
2739    agent: &'a str,
2740    /// Base branch or parent branch used for future worktree materialization.
2741    base_branch: &'a str,
2742    /// Stable session identifier.
2743    id: &'a str,
2744    /// Whether the row was created through explicit draft staging.
2745    is_draft: bool,
2746    /// Agent model identifier persisted for the session.
2747    model: &'a str,
2748    /// Orchestration task that owns this child session, when applicable.
2749    orchestration_task_id: Option<i64>,
2750    /// Optional parent session id for one-level stacked drafts.
2751    parent_session_id: Option<&'a str>,
2752    /// Provider permission mode captured for the session.
2753    permission_mode: PermissionMode,
2754    /// Workspace personality selected for future turns, when present.
2755    personality_id: Option<&'a str>,
2756    /// Owning project identifier.
2757    project_id: i64,
2758    /// Reasoning level captured from the project default at creation.
2759    reasoning_level: ReasoningLevel,
2760    /// Persisted session role, or `None` for the default worker role.
2761    role: Option<&'a str>,
2762    /// Response-speed preference captured for the session.
2763    speed_mode: SpeedMode,
2764    /// Initial lifecycle status string.
2765    status: &'a str,
2766}
2767
2768/// Inserts one newly created session row with explicit draft-mode
2769/// persistence.
2770async fn insert_session_with_draft_mode(
2771    pool: &SqlitePool,
2772    timestamp_seconds: i64,
2773    row: InsertSessionRow<'_>,
2774) -> Result<(), DbError> {
2775    let InsertSessionRow {
2776        agent,
2777        base_branch,
2778        id,
2779        is_draft,
2780        model,
2781        orchestration_task_id,
2782        parent_session_id,
2783        permission_mode,
2784        personality_id,
2785        project_id,
2786        reasoning_level,
2787        role,
2788        speed_mode,
2789        status,
2790    } = row;
2791    status::validate_session(status)?;
2792
2793    sqlx::query(
2794        r"
2795INSERT INTO session (
2796    id,
2797    agent,
2798    model,
2799    base_branch,
2800    status,
2801    has_diff,
2802    is_draft,
2803    parent_session_id,
2804    permission_mode,
2805    personality_id,
2806    project_id,
2807    reasoning_level,
2808    role,
2809    speed_mode,
2810    orchestration_task_id,
2811    prompt,
2812    created_at,
2813    updated_at
2814)
2815VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
2816",
2817    )
2818    .bind(id)
2819    .bind(agent)
2820    .bind(model)
2821    .bind(base_branch)
2822    .bind(status)
2823    // Diff availability remains unknown until the worktree is refreshed.
2824    .bind(Option::<bool>::None)
2825    .bind(is_draft)
2826    .bind(parent_session_id)
2827    .bind(permission_mode.label())
2828    .bind(personality_id)
2829    .bind(project_id)
2830    .bind(reasoning_level.as_str())
2831    .bind(role)
2832    .bind(speed_mode.as_str())
2833    .bind(orchestration_task_id)
2834    .bind("")
2835    .bind(timestamp_seconds)
2836    .bind(timestamp_seconds)
2837    .execute(pool)
2838    .await?;
2839
2840    Ok(())
2841}
2842
2843/// Returns the persisted agent value paired with a newly saved model string.
2844fn persisted_agent_for_model(model: &str) -> String {
2845    AgentModel::parse_persisted(model).map_or_else(
2846        |_| persisted_agent_for_unknown_model(model).to_string(),
2847        |agent_model| persisted_agent_for_known_model(model, agent_model).to_string(),
2848    )
2849}
2850
2851/// Returns a compatibility agent value for known model strings passed through
2852/// model-only legacy persistence helpers.
2853fn persisted_agent_for_known_model(model: &str, agent_model: AgentModel) -> AgentKind {
2854    if model.starts_with("claude-") {
2855        return AgentKind::Claude;
2856    }
2857
2858    if model.starts_with("gpt-") {
2859        return AgentKind::Codex;
2860    }
2861
2862    if model.starts_with("gemini-") {
2863        return AgentKind::Antigravity;
2864    }
2865
2866    AgentKind::ALL
2867        .iter()
2868        .copied()
2869        .find(|agent_kind| agent_kind.supports_model(agent_model))
2870        .unwrap_or(AgentKind::Antigravity)
2871}
2872
2873/// Returns a compatibility agent value for tests or older callers that pass
2874/// model strings outside the current curated model set.
2875fn persisted_agent_for_unknown_model(model: &str) -> AgentKind {
2876    if model.starts_with("claude-") {
2877        return AgentKind::Claude;
2878    }
2879
2880    if model.starts_with("gpt-") {
2881        return AgentKind::Codex;
2882    }
2883
2884    if model.starts_with("gemini-") {
2885        return AgentKind::Antigravity;
2886    }
2887
2888    AgentKind::Antigravity
2889}
2890
2891#[cfg(test)]
2892mod tests {
2893    use ag_session::{ForgeKind, ReviewRequest, ReviewRequestState, ReviewRequestSummary};
2894
2895    use super::*;
2896    use crate::AppRepositories;
2897
2898    /// Session columns that must be reset when snapshotting a fork.
2899    struct ForkResetRow {
2900        applied_personality_id: Option<String>,
2901        applied_personality_prompt_hash: Option<String>,
2902        app_server_instruction_provider_conversation_id: Option<String>,
2903        focused_review_diff_hash: Option<String>,
2904        focused_review_text: Option<String>,
2905        in_progress_started_at: Option<i64>,
2906        in_progress_total_seconds: i64,
2907        is_draft: bool,
2908        merged_commit_hash: Option<String>,
2909        parent_session_id: Option<String>,
2910        provider_conversation_id: Option<String>,
2911        published_upstream_ref: Option<String>,
2912        questions: Option<String>,
2913        stack_base_commit_hash: Option<String>,
2914    }
2915
2916    impl SessionJoinRow {
2917        /// Builds a deterministic joined-session row fixture for conversion
2918        /// tests.
2919        fn fixture_for_test() -> Self {
2920            Self {
2921                added_lines: 14,
2922                agent: "codex".to_string(),
2923                base_branch: "main".to_string(),
2924                created_at: 100,
2925                deleted_lines: 6,
2926                has_diff: Some(true),
2927                id: "session-a".to_string(),
2928                in_progress_started_at: None,
2929                in_progress_total_seconds: 0,
2930                input_tokens: 11,
2931                is_draft: false,
2932                model: "gpt-5.6-sol".to_string(),
2933                output_tokens: 29,
2934                parent_session_id: Some("parent-session".to_string()),
2935                permission_mode: "read_only".to_string(),
2936                personality_id: Some("reviewer".to_string()),
2937                project_id: Some(7),
2938                prompt: "Implement feature".to_string(),
2939                published_upstream_ref: Some("origin/session-a".to_string()),
2940                questions: Some("Question text".to_string()),
2941                reasoning_level_override: None,
2942                review_request_display_id: Some("#42".to_string()),
2943                review_request_forge_kind: Some("GitHub".to_string()),
2944                review_request_last_refreshed_at: Some(456),
2945                review_request_source_branch: Some("feature/forge".to_string()),
2946                review_request_state: Some("Open".to_string()),
2947                review_request_status_summary: Some("2 approvals, checks passing".to_string()),
2948                review_request_target_branch: Some("main".to_string()),
2949                review_request_title: Some("Add forge review support".to_string()),
2950                review_request_web_url: Some(
2951                    "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
2952                ),
2953                role: Some("Orchestrator".to_string()),
2954                size: "M".to_string(),
2955                speed_mode: "normal".to_string(),
2956                status: "Review".to_string(),
2957                summary: Some("Summary text".to_string()),
2958                title: Some("Review session".to_string()),
2959                updated_at: 200,
2960            }
2961        }
2962    }
2963
2964    /// Builds the fully populated review-request row expected by join-row
2965    /// conversion tests.
2966    fn expected_review_request_row() -> SessionReviewRequestRow {
2967        SessionReviewRequestRow {
2968            display_id: "#42".to_string(),
2969            forge_kind: "GitHub".to_string(),
2970            last_refreshed_at: 456,
2971            source_branch: "feature/forge".to_string(),
2972            state: "Open".to_string(),
2973            status_summary: Some("2 approvals, checks passing".to_string()),
2974            target_branch: "main".to_string(),
2975            title: "Add forge review support".to_string(),
2976            web_url: "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
2977        }
2978    }
2979
2980    /// Builds the review-request domain fixture used by fork snapshot tests.
2981    fn review_request_fixture() -> ReviewRequest {
2982        ReviewRequest {
2983            last_refreshed_at: 456,
2984            summary: ReviewRequestSummary {
2985                display_id: "#42".to_string(),
2986                forge_kind: ForgeKind::GitHub,
2987                source_branch: "feature/forge".to_string(),
2988                state: ReviewRequestState::Open,
2989                status_summary: Some("2 approvals, checks passing".to_string()),
2990                target_branch: "main".to_string(),
2991                title: "Add forge review support".to_string(),
2992                web_url: "https://github.com/agentty-xyz/agentty/pull/42".to_string(),
2993            },
2994        }
2995    }
2996
2997    /// Loads reset-sensitive fork columns that are not exposed by public
2998    /// session row projections.
2999    async fn load_fork_reset_row(pool: &SqlitePool, session_id: &str) -> ForkResetRow {
3000        sqlx::query_as!(
3001            ForkResetRow,
3002            r#"
3003SELECT app_server_instruction_provider_conversation_id,
3004       applied_personality_id,
3005       applied_personality_prompt_hash,
3006       focused_review_diff_hash,
3007       focused_review_text,
3008       in_progress_started_at,
3009       in_progress_total_seconds,
3010       is_draft AS "is_draft: bool",
3011       merged_commit_hash,
3012       parent_session_id,
3013       provider_conversation_id,
3014       published_upstream_ref,
3015       questions,
3016       stack_base_commit_hash
3017FROM session
3018WHERE id = ?
3019"#,
3020            session_id
3021        )
3022        .fetch_one(pool)
3023        .await
3024        .expect("failed to load fork reset row")
3025    }
3026
3027    /// Seeds a forkable source session with every source-only field that the
3028    /// snapshot insert is expected to clear.
3029    async fn seed_fork_snapshot_source(
3030        database: &AppRepositories,
3031        pool: &SqlitePool,
3032    ) -> (ForkResetRow, Option<SessionReviewRequestRow>) {
3033        let project_id = database
3034            .projects()
3035            .upsert_project("/tmp/project", None)
3036            .await
3037            .expect("failed to upsert project");
3038        database
3039            .sessions()
3040            .insert_session(
3041                "parent-session",
3042                "gpt-5.6-sol",
3043                "main",
3044                "Review",
3045                project_id,
3046            )
3047            .await
3048            .expect("failed to insert parent session");
3049        database
3050            .sessions()
3051            .insert_stacked_draft_session(
3052                "source-session",
3053                "gpt-5.6-sol",
3054                "wt/parent",
3055                "Review",
3056                "parent-session",
3057                project_id,
3058            )
3059            .await
3060            .expect("failed to insert source session");
3061
3062        seed_fork_snapshot_source_linkage(database).await;
3063        seed_fork_snapshot_source_timing(database, pool).await;
3064
3065        let source_reset_row = load_fork_reset_row(pool, "source-session").await;
3066        let source_review_request = database
3067            .reviews()
3068            .load_session_review_request("source-session")
3069            .await
3070            .expect("failed to load source review request");
3071
3072        (source_reset_row, source_review_request)
3073    }
3074
3075    /// Persists source-only linkage and counters on the fork source row.
3076    async fn seed_fork_snapshot_source_linkage(database: &AppRepositories) {
3077        seed_fork_snapshot_source_settings(database).await;
3078        database
3079            .sessions()
3080            .persist_session_turn_metadata(
3081                "source-session",
3082                &SessionTurnMetadata {
3083                    applied_personality_id: Some("reviewer".to_string()),
3084                    applied_personality_prompt_hash: Some("personality-hash".to_string()),
3085                    instruction_conversation_id: None,
3086                    model: "gpt-5.6-sol".to_string(),
3087                    provider_conversation_id: None,
3088                    questions_json: "[]".to_string(),
3089                    review_comment_resolutions: Vec::new(),
3090                    summary: String::new(),
3091                    token_usage_delta: SessionStats::default(),
3092                },
3093            )
3094            .await
3095            .expect("failed to persist applied personality");
3096        database
3097            .sessions()
3098            .update_session_provider_conversation_id(
3099                "source-session",
3100                Some("provider-thread".to_string()),
3101            )
3102            .await
3103            .expect("failed to update provider conversation id");
3104        database
3105            .sessions()
3106            .update_session_instruction_conversation_id(
3107                "source-session",
3108                Some("instruction-thread".to_string()),
3109            )
3110            .await
3111            .expect("failed to update instruction conversation id");
3112        database
3113            .sessions()
3114            .update_session_questions("source-session", r#"["Need detail?"]"#)
3115            .await
3116            .expect("failed to update questions");
3117        database
3118            .sessions()
3119            .update_session_published_upstream_ref(
3120                "source-session",
3121                Some("origin/wt/source-session".to_string()),
3122            )
3123            .await
3124            .expect("failed to update published upstream ref");
3125        database
3126            .sessions()
3127            .update_session_merged_commit_hash("source-session", Some("merged123".to_string()))
3128            .await
3129            .expect("failed to update merged commit hash");
3130        database
3131            .sessions()
3132            .update_session_focused_review(
3133                "source-session",
3134                Some(FocusedReviewStatus::Ready),
3135                Some("diff123".to_string()),
3136                Some("Focused review text".to_string()),
3137            )
3138            .await
3139            .expect("failed to update focused review");
3140        database
3141            .sessions()
3142            .update_session_stack_base_commit_hash(
3143                "source-session",
3144                Some("stackbase123".to_string()),
3145            )
3146            .await
3147            .expect("failed to update stack base commit hash");
3148        database
3149            .sessions()
3150            .update_session_stats(
3151                "source-session",
3152                &SessionStats {
3153                    added_lines: 0,
3154                    deleted_lines: 0,
3155                    diff_state: agent::SessionDiffState::Unknown,
3156                    input_tokens: 11,
3157                    output_tokens: 29,
3158                },
3159            )
3160            .await
3161            .expect("failed to update token stats");
3162        database
3163            .sessions()
3164            .update_session_diff_stats(7, 3, true, "source-session", "S")
3165            .await
3166            .expect("failed to update source diff stats");
3167        database
3168            .reviews()
3169            .update_session_review_request("source-session", Some(review_request_fixture()))
3170            .await
3171            .expect("failed to update review request");
3172    }
3173
3174    /// Persists the session settings that a fork must inherit.
3175    async fn seed_fork_snapshot_source_settings(database: &AppRepositories) {
3176        database
3177            .sessions()
3178            .update_session_permission_mode("source-session", PermissionMode::ReadOnly)
3179            .await
3180            .expect("failed to update permission mode");
3181        database
3182            .sessions()
3183            .update_session_personality_id("source-session", Some("reviewer".to_string()))
3184            .await
3185            .expect("failed to update personality id");
3186    }
3187
3188    /// Persists active-work timing fields on the fork source row.
3189    async fn seed_fork_snapshot_source_timing(database: &AppRepositories, pool: &SqlitePool) {
3190        database
3191            .sessions()
3192            .update_session_status_with_timing_at("source-session", "InProgress", 100)
3193            .await
3194            .expect("failed to open timing interval");
3195        sqlx::query!(
3196            r"
3197UPDATE session
3198SET in_progress_total_seconds = ?
3199WHERE id = ?
3200",
3201            75_i64,
3202            "source-session"
3203        )
3204        .execute(pool)
3205        .await
3206        .expect("failed to seed elapsed timing");
3207    }
3208
3209    /// Asserts the fixture source row actually had source-only state before
3210    /// the snapshot was taken.
3211    fn assert_source_reset_state(
3212        source_row: &SessionRow,
3213        source_reset_row: &ForkResetRow,
3214        source_review_request: Option<&SessionReviewRequestRow>,
3215    ) {
3216        assert_eq!(source_row.added_lines, 7);
3217        assert_eq!(source_row.deleted_lines, 3);
3218        assert_eq!(source_row.has_diff, Some(true));
3219        assert_eq!(source_row.size, "S");
3220        assert_eq!(source_row.permission_mode, "read_only");
3221        assert_eq!(
3222            source_row.permission_mode.parse::<PermissionMode>(),
3223            Ok(PermissionMode::ReadOnly)
3224        );
3225        assert!(source_reset_row.is_draft);
3226        assert_eq!(source_row.personality_id.as_deref(), Some("reviewer"));
3227        assert_eq!(
3228            source_reset_row.applied_personality_id.as_deref(),
3229            Some("reviewer")
3230        );
3231        assert_eq!(
3232            source_reset_row.applied_personality_prompt_hash.as_deref(),
3233            Some("personality-hash")
3234        );
3235        assert_eq!(
3236            source_reset_row.parent_session_id.as_deref(),
3237            Some("parent-session")
3238        );
3239        assert_eq!(
3240            source_reset_row.provider_conversation_id.as_deref(),
3241            Some("provider-thread")
3242        );
3243        assert_eq!(
3244            source_reset_row
3245                .app_server_instruction_provider_conversation_id
3246                .as_deref(),
3247            Some("instruction-thread")
3248        );
3249        assert_eq!(
3250            source_reset_row.published_upstream_ref.as_deref(),
3251            Some("origin/wt/source-session")
3252        );
3253        assert_eq!(
3254            source_reset_row.questions.as_deref(),
3255            Some(r#"["Need detail?"]"#)
3256        );
3257        assert_eq!(
3258            source_reset_row.merged_commit_hash.as_deref(),
3259            Some("merged123")
3260        );
3261        assert_eq!(
3262            source_reset_row.focused_review_diff_hash.as_deref(),
3263            Some("diff123")
3264        );
3265        assert_eq!(
3266            source_reset_row.focused_review_text.as_deref(),
3267            Some("Focused review text")
3268        );
3269        assert_eq!(
3270            source_reset_row.stack_base_commit_hash.as_deref(),
3271            Some("stackbase123")
3272        );
3273        assert_eq!(source_reset_row.in_progress_started_at, Some(100));
3274        assert_eq!(source_reset_row.in_progress_total_seconds, 75);
3275        assert_eq!(
3276            source_review_request.map(|review_request| review_request.display_id.as_str()),
3277            Some("#42")
3278        );
3279    }
3280
3281    /// Asserts the forked row kept durable snapshot state while clearing
3282    /// source-only linkage.
3283    fn assert_fork_reset_state(
3284        fork_row: &SessionRow,
3285        fork_reset_row: &ForkResetRow,
3286        fork_review_request: Option<&SessionReviewRequestRow>,
3287    ) {
3288        assert_eq!(fork_row.status, "Review");
3289        assert!(!fork_row.is_draft);
3290        assert_eq!(fork_row.parent_session_id, None);
3291        assert_eq!(fork_row.personality_id.as_deref(), Some("reviewer"));
3292        assert_eq!(fork_row.input_tokens, 0);
3293        assert_eq!(fork_row.output_tokens, 0);
3294        assert_eq!(fork_row.added_lines, 0);
3295        assert_eq!(fork_row.deleted_lines, 0);
3296        assert_eq!(fork_row.has_diff, None);
3297        assert_eq!(fork_row.size, "XS");
3298        assert_eq!(fork_row.permission_mode, "read_only");
3299        assert_eq!(fork_row.questions, None);
3300        assert_eq!(fork_row.published_upstream_ref, None);
3301        assert_eq!(fork_row.review_request, None);
3302        assert_eq!(fork_reset_row.provider_conversation_id, None);
3303        assert_eq!(fork_reset_row.applied_personality_id, None);
3304        assert_eq!(fork_reset_row.applied_personality_prompt_hash, None);
3305        assert_eq!(
3306            fork_reset_row.app_server_instruction_provider_conversation_id,
3307            None
3308        );
3309        assert_eq!(fork_reset_row.merged_commit_hash, None);
3310        assert_eq!(fork_reset_row.focused_review_diff_hash, None);
3311        assert_eq!(fork_reset_row.focused_review_text, None);
3312        assert_eq!(fork_reset_row.questions, None);
3313        assert_eq!(fork_reset_row.stack_base_commit_hash, None);
3314        assert_eq!(fork_reset_row.in_progress_started_at, None);
3315        assert_eq!(fork_reset_row.in_progress_total_seconds, 0);
3316        assert_eq!(fork_review_request, None);
3317    }
3318
3319    #[tokio::test]
3320    async fn test_load_session_rejects_unknown_status() {
3321        // Arrange
3322        let (database, pool) = AppRepositories::in_memory_with_pool()
3323            .await
3324            .expect("db should open");
3325        let project_id = database
3326            .projects()
3327            .upsert_project("/tmp/invalid-session", None)
3328            .await
3329            .expect("failed to upsert project");
3330        database
3331            .sessions()
3332            .insert_session("session-a", "gpt-5.6-sol", "main", "Draft", project_id)
3333            .await
3334            .expect("failed to insert session");
3335        sqlx::query("UPDATE session SET status = 'Unknown' WHERE id = 'session-a'")
3336            .execute(&pool)
3337            .await
3338            .expect("failed to corrupt session status");
3339
3340        // Act
3341        let result = database.sessions().load_session("session-a").await;
3342
3343        // Assert
3344        assert!(matches!(
3345            result,
3346            Err(DbError::InvalidStatus {
3347                entity: "session",
3348                value,
3349            }) if value == "Unknown"
3350        ));
3351    }
3352
3353    #[tokio::test]
3354    async fn test_load_session_collections_skip_unknown_status() {
3355        // Arrange
3356        let (database, pool) = AppRepositories::in_memory_with_pool()
3357            .await
3358            .expect("db should open");
3359        let project_id = database
3360            .projects()
3361            .upsert_project("/tmp/invalid-session-list", None)
3362            .await
3363            .expect("failed to upsert project");
3364        for session_id in ["session-valid", "session-invalid"] {
3365            database
3366                .sessions()
3367                .insert_session(session_id, "gpt-5.6-sol", "main", "Draft", project_id)
3368                .await
3369                .expect("failed to insert session");
3370        }
3371        sqlx::query("UPDATE session SET status = 'Unknown' WHERE id = 'session-invalid'")
3372            .execute(&pool)
3373            .await
3374            .expect("failed to corrupt session status");
3375
3376        // Act
3377        let all_sessions = database
3378            .sessions()
3379            .load_sessions()
3380            .await
3381            .expect("failed to load all sessions");
3382        let project_sessions = database
3383            .sessions()
3384            .load_sessions_for_project(project_id)
3385            .await
3386            .expect("failed to load project sessions");
3387
3388        // Assert
3389        assert_eq!(
3390            all_sessions
3391                .iter()
3392                .map(|session| session.id.as_str())
3393                .collect::<Vec<_>>(),
3394            ["session-valid"]
3395        );
3396        assert_eq!(
3397            project_sessions
3398                .iter()
3399                .map(|session| session.id.as_str())
3400                .collect::<Vec<_>>(),
3401            ["session-valid"]
3402        );
3403    }
3404
3405    #[tokio::test]
3406    async fn test_insert_session_starts_with_unknown_diff() {
3407        // Arrange
3408        let (database, _) = AppRepositories::in_memory_with_pool()
3409            .await
3410            .expect("db should open");
3411        let project_id = database
3412            .projects()
3413            .upsert_project("/tmp/project", None)
3414            .await
3415            .expect("failed to upsert project");
3416
3417        // Act
3418        database
3419            .sessions()
3420            .insert_session("session-a", "gpt-5.6-sol", "main", "Draft", project_id)
3421            .await
3422            .expect("failed to insert session");
3423        let session = database
3424            .sessions()
3425            .load_sessions()
3426            .await
3427            .expect("failed to load sessions")
3428            .into_iter()
3429            .next()
3430            .expect("missing inserted session");
3431
3432        // Assert
3433        assert_eq!(session.has_diff, None);
3434    }
3435
3436    #[tokio::test]
3437    async fn test_load_sessions_uses_created_at_to_break_updated_at_ties() {
3438        // Arrange
3439        let (database, pool) = AppRepositories::in_memory_with_pool()
3440            .await
3441            .expect("db should open");
3442        let project_id = database
3443            .projects()
3444            .upsert_project("/tmp/project", None)
3445            .await
3446            .expect("failed to upsert project");
3447        for session_id in ["a-older", "z-newer"] {
3448            database
3449                .sessions()
3450                .insert_session(session_id, "gpt-5.6-sol", "main", "Review", project_id)
3451                .await
3452                .expect("failed to insert session");
3453        }
3454        sqlx::query!(
3455            r"
3456UPDATE session
3457SET created_at = CASE id WHEN 'a-older' THEN 100 ELSE 200 END,
3458    updated_at = 300
3459WHERE id IN ('a-older', 'z-newer')
3460"
3461        )
3462        .execute(&pool)
3463        .await
3464        .expect("failed to set session timestamps");
3465
3466        // Act
3467        let all_session_ids = database
3468            .sessions()
3469            .load_sessions()
3470            .await
3471            .expect("failed to load sessions")
3472            .into_iter()
3473            .map(|session| session.id)
3474            .collect::<Vec<_>>();
3475        let project_session_ids = database
3476            .sessions()
3477            .load_sessions_for_project(project_id)
3478            .await
3479            .expect("failed to load project sessions")
3480            .into_iter()
3481            .map(|session| session.id)
3482            .collect::<Vec<_>>();
3483
3484        // Assert
3485        assert_eq!(all_session_ids, ["z-newer", "a-older"]);
3486        assert_eq!(project_session_ids, ["z-newer", "a-older"]);
3487    }
3488
3489    #[tokio::test]
3490    async fn test_fork_session_snapshot_resets_source_specific_state() {
3491        // Arrange
3492        let (database, pool) = AppRepositories::in_memory_with_pool()
3493            .await
3494            .expect("db should open");
3495        let (source_reset_row, source_review_request) =
3496            seed_fork_snapshot_source(&database, &pool).await;
3497
3498        // Act
3499        database
3500            .sessions()
3501            .fork_session_snapshot(ForkSessionSnapshot {
3502                new_session_id: "fork-session",
3503                source_session_id: "source-session",
3504                status: "Review",
3505            })
3506            .await
3507            .expect("failed to fork session snapshot");
3508
3509        // Assert
3510        let session_rows = database
3511            .sessions()
3512            .load_sessions()
3513            .await
3514            .expect("failed to load sessions");
3515        let source_row = session_rows
3516            .iter()
3517            .find(|session_row| session_row.id == "source-session")
3518            .expect("missing source session row");
3519        let fork_row = session_rows
3520            .iter()
3521            .find(|session_row| session_row.id == "fork-session")
3522            .expect("missing forked session row");
3523        let fork_reset_row = load_fork_reset_row(&pool, "fork-session").await;
3524        let fork_review_request = database
3525            .reviews()
3526            .load_session_review_request("fork-session")
3527            .await
3528            .expect("failed to load fork review request");
3529        let fork_permission_mode = database
3530            .sessions()
3531            .load_session_permission_mode("fork-session")
3532            .await
3533            .expect("failed to load fork permission mode");
3534
3535        assert_source_reset_state(
3536            source_row,
3537            &source_reset_row,
3538            source_review_request.as_ref(),
3539        );
3540        assert_fork_reset_state(fork_row, &fork_reset_row, fork_review_request.as_ref());
3541        assert_eq!(fork_permission_mode, PermissionMode::ReadOnly);
3542    }
3543
3544    #[tokio::test]
3545    async fn test_clear_session_draft_flag_marks_draft_session_live() {
3546        // Arrange
3547        let (database, _pool) = AppRepositories::in_memory_with_pool()
3548            .await
3549            .expect("db should open");
3550        let project_id = database
3551            .projects()
3552            .upsert_project("/tmp/project", None)
3553            .await
3554            .expect("failed to upsert project");
3555        database
3556            .sessions()
3557            .insert_draft_session("draft-session", "gpt-5.6-sol", "main", "Draft", project_id)
3558            .await
3559            .expect("failed to insert draft session");
3560
3561        // Act
3562        database
3563            .sessions()
3564            .clear_session_draft_flag("draft-session")
3565            .await
3566            .expect("failed to clear session draft flag");
3567
3568        // Assert
3569        let session_row = database
3570            .sessions()
3571            .load_sessions()
3572            .await
3573            .expect("failed to load sessions")
3574            .into_iter()
3575            .find(|session_row| session_row.id == "draft-session")
3576            .expect("missing draft session row");
3577        assert!(!session_row.is_draft);
3578    }
3579
3580    /// Verifies `SessionJoinRow::into_session_row()` drops partially
3581    /// populated review-request columns instead of surfacing an invalid row
3582    /// model.
3583    #[test]
3584    fn test_session_join_row_ignores_partial_review_request_columns() {
3585        // Arrange
3586        let mut session_join_row = SessionJoinRow::fixture_for_test();
3587        session_join_row.review_request_last_refreshed_at = None;
3588
3589        // Act
3590        let session_row = session_join_row.into_session_row();
3591
3592        // Assert
3593        assert_eq!(session_row.id, "session-a");
3594        assert_eq!(session_row.project_id, Some(7));
3595        assert_eq!(
3596            session_row.parent_session_id.as_deref(),
3597            Some("parent-session")
3598        );
3599        assert_eq!(session_row.status, "Review");
3600        assert_eq!(session_row.added_lines, 14);
3601        assert_eq!(session_row.deleted_lines, 6);
3602        assert_eq!(session_row.review_request, None);
3603    }
3604
3605    /// Verifies `SessionJoinRow::into_session_row()` maps a fully populated
3606    /// review-request into the public session row model.
3607    #[test]
3608    fn test_session_join_row_maps_review_request_columns() {
3609        // Arrange
3610        let session_join_row = SessionJoinRow::fixture_for_test();
3611
3612        // Act
3613        let session_row = session_join_row.into_session_row();
3614
3615        // Assert
3616        assert_eq!(session_row.id, "session-a");
3617        assert_eq!(session_row.added_lines, 14);
3618        assert_eq!(session_row.deleted_lines, 6);
3619        assert_eq!(session_row.project_id, Some(7));
3620        assert_eq!(session_row.personality_id.as_deref(), Some("reviewer"));
3621        assert_eq!(
3622            session_row.parent_session_id.as_deref(),
3623            Some("parent-session")
3624        );
3625        assert_eq!(
3626            session_row.published_upstream_ref.as_deref(),
3627            Some("origin/session-a")
3628        );
3629        assert_eq!(session_row.questions.as_deref(), Some("Question text"));
3630        assert_eq!(session_row.summary.as_deref(), Some("Summary text"));
3631        assert_eq!(session_row.title.as_deref(), Some("Review session"));
3632        assert_eq!(
3633            session_row.review_request,
3634            Some(expected_review_request_row())
3635        );
3636    }
3637}
ag_store/session.rs

ag_store/
session.rs
