dk_engine/
changeset.rs

1use chrono::{DateTime, Utc};
2use sqlx::PgPool;
3use uuid::Uuid;
4
5use dk_core::{RepoId, SymbolId};
6
7/// Explicit changeset states. Replaces the former ambiguous "open" state.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum ChangesetState {
10    Draft,
11    Submitted,
12    Verifying,
13    Approved,
14    Rejected,
15    Merged,
16    Closed,
17}
18
19impl ChangesetState {
20    /// Parse a state string from the database into a `ChangesetState`.
21    pub fn parse(s: &str) -> Option<Self> {
22        match s {
23            "draft" => Some(Self::Draft),
24            "submitted" => Some(Self::Submitted),
25            "verifying" => Some(Self::Verifying),
26            "approved" => Some(Self::Approved),
27            "rejected" => Some(Self::Rejected),
28            "merged" => Some(Self::Merged),
29            "closed" => Some(Self::Closed),
30            _ => None,
31        }
32    }
33
34    /// Return the database string representation.
35    pub fn as_str(&self) -> &'static str {
36        match self {
37            Self::Draft => "draft",
38            Self::Submitted => "submitted",
39            Self::Verifying => "verifying",
40            Self::Approved => "approved",
41            Self::Rejected => "rejected",
42            Self::Merged => "merged",
43            Self::Closed => "closed",
44        }
45    }
46
47    /// Check whether transitioning from `self` to `target` is valid.
48    ///
49    /// Valid transitions:
50    /// - draft      -> submitted
51    /// - submitted  -> verifying
52    /// - verifying  -> approved | rejected
53    /// - approved   -> merged
54    /// - any        -> closed
55    pub fn can_transition_to(&self, target: Self) -> bool {
56        if target == Self::Closed {
57            return true;
58        }
59        matches!(
60            (self, target),
61            (Self::Draft, Self::Submitted)
62                | (Self::Submitted, Self::Verifying)
63                | (Self::Verifying, Self::Approved)
64                | (Self::Verifying, Self::Rejected)
65                | (Self::Approved, Self::Merged)
66        )
67    }
68}
69
70impl std::fmt::Display for ChangesetState {
71    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        f.write_str(self.as_str())
73    }
74}
75
76#[derive(Debug, Clone, sqlx::FromRow)]
77pub struct Changeset {
78    pub id: Uuid,
79    pub repo_id: RepoId,
80    pub number: i32,
81    pub title: String,
82    pub intent_summary: Option<String>,
83    pub source_branch: String,
84    pub target_branch: String,
85    pub state: String,
86    pub reason: String,
87    pub session_id: Option<Uuid>,
88    pub agent_id: Option<String>,
89    pub agent_name: Option<String>,
90    pub author_id: Option<Uuid>,
91    pub base_version: Option<String>,
92    pub merged_version: Option<String>,
93    pub created_at: DateTime<Utc>,
94    pub updated_at: DateTime<Utc>,
95    pub merged_at: Option<DateTime<Utc>>,
96    /// Stacked-changeset parent. PR1 ships the column additively — nothing
97    /// populates it and no consumer reads it. PR2 will set it at submit
98    /// time and enforce merge-order on the chain.
99    pub parent_changeset_id: Option<Uuid>,
100}
101
102impl Changeset {
103    /// Parse the current state string into a typed `ChangesetState`.
104    pub fn parsed_state(&self) -> Option<ChangesetState> {
105        ChangesetState::parse(&self.state)
106    }
107
108    /// Validate and perform a state transition, recording the reason.
109    /// Returns an error if the transition is not allowed.
110    pub fn transition(
111        &mut self,
112        target: ChangesetState,
113        reason: impl Into<String>,
114    ) -> dk_core::Result<()> {
115        let current = self.parsed_state().ok_or_else(|| {
116            dk_core::Error::Internal(format!("unknown current state: '{}'", self.state))
117        })?;
118
119        if !current.can_transition_to(target) {
120            return Err(dk_core::Error::InvalidInput(format!(
121                "invalid state transition: '{}' -> '{}'",
122                current, target,
123            )));
124        }
125
126        self.state = target.as_str().to_string();
127        self.reason = reason.into();
128        Ok(())
129    }
130}
131
132#[derive(Debug, Clone)]
133pub struct ChangesetFile {
134    pub changeset_id: Uuid,
135    pub file_path: String,
136    pub operation: String,
137    pub content: Option<String>,
138}
139
140#[derive(Debug, Clone)]
141pub struct ChangesetFileMeta {
142    pub file_path: String,
143    pub operation: String,
144    pub size_bytes: i64,
145}
146
147pub struct ChangesetStore {
148    db: PgPool,
149}
150
151impl ChangesetStore {
152    pub fn new(db: PgPool) -> Self {
153        Self { db }
154    }
155
156    /// Create a changeset via the Agent Protocol path.
157    /// Auto-increments the number per repo using an advisory lock.
158    /// Sets `source_branch` to `agent/<agent_name>` and `target_branch` to `main`
159    /// so platform queries that read these NOT NULL columns always succeed.
160    /// `agent_name` is the human-readable name (e.g. "agent-1" or "feature-bot").
161    pub async fn create(
162        &self,
163        repo_id: RepoId,
164        session_id: Option<Uuid>,
165        agent_id: &str,
166        intent: &str,
167        base_version: Option<&str>,
168        agent_name: &str,
169    ) -> dk_core::Result<Changeset> {
170        let intent_slug: String = intent
171            .to_lowercase()
172            .chars()
173            .map(|c| if c.is_alphanumeric() || c == '-' { c } else { '-' })
174            .collect::<String>()
175            .trim_matches('-')
176            .to_string();
177        let slug = if intent_slug.len() > 50 {
178            let cut = intent_slug
179                .char_indices()
180                .take_while(|(i, _)| *i < 50)
181                .last()
182                .map(|(i, c)| i + c.len_utf8())
183                .unwrap_or(0);
184            intent_slug[..cut].trim_end_matches('-').to_string()
185        } else {
186            intent_slug
187        };
188        let source_branch = format!("{}/{}", slug, agent_name);
189        let target_branch = "main";
190
191        let mut tx = self.db.begin().await?;
192
193        sqlx::query("SELECT pg_advisory_xact_lock(hashtext('changeset:' || $1::text))")
194            .bind(repo_id)
195            .execute(&mut *tx)
196            .await?;
197
198        let row: (Uuid, i32, String, String, DateTime<Utc>, DateTime<Utc>) = sqlx::query_as(
199            r#"INSERT INTO changesets
200                   (repo_id, number, title, intent_summary, source_branch, target_branch,
201                    state, reason, session_id, agent_id, agent_name, base_version)
202               SELECT $1, COALESCE(MAX(number), 0) + 1, $2, $2, $3, $4,
203                    'draft', 'created via agent connect', $5, $6, $7, $8
204               FROM changesets WHERE repo_id = $1
205               RETURNING id, number, state, reason, created_at, updated_at"#,
206        )
207        .bind(repo_id)
208        .bind(intent)
209        .bind(&source_branch)
210        .bind(target_branch)
211        .bind(session_id)
212        .bind(agent_id)
213        .bind(agent_name)
214        .bind(base_version)
215        .fetch_one(&mut *tx)
216        .await?;
217
218        tx.commit().await?;
219
220        Ok(Changeset {
221            id: row.0,
222            repo_id,
223            number: row.1,
224            title: intent.to_string(),
225            intent_summary: Some(intent.to_string()),
226            source_branch,
227            target_branch: target_branch.to_string(),
228            state: row.2,
229            reason: row.3,
230            session_id,
231            agent_id: Some(agent_id.to_string()),
232            agent_name: Some(agent_name.to_string()),
233            author_id: None,
234            base_version: base_version.map(String::from),
235            merged_version: None,
236            created_at: row.4,
237            updated_at: row.5,
238            merged_at: None,
239            parent_changeset_id: None,
240        })
241    }
242
243    pub async fn get(&self, id: Uuid) -> dk_core::Result<Changeset> {
244        sqlx::query_as::<_, Changeset>(
245            r#"SELECT id, repo_id, number, title, intent_summary,
246                      source_branch, target_branch, state, reason,
247                      session_id, agent_id, agent_name, author_id,
248                      base_version, merged_version,
249                      created_at, updated_at, merged_at,
250                      parent_changeset_id
251               FROM changesets WHERE id = $1"#,
252        )
253        .bind(id)
254        .fetch_optional(&self.db)
255        .await?
256        .ok_or_else(|| dk_core::Error::Internal(format!("changeset {} not found", id)))
257    }
258
259    pub async fn update_status(&self, id: Uuid, status: &str) -> dk_core::Result<()> {
260        self.update_status_with_reason(id, status, "").await
261    }
262
263    /// Update changeset status and record the reason for the transition.
264    pub async fn update_status_with_reason(
265        &self,
266        id: Uuid,
267        status: &str,
268        reason: &str,
269    ) -> dk_core::Result<()> {
270        sqlx::query(
271            "UPDATE changesets SET state = $1, reason = $2, updated_at = now() WHERE id = $3",
272        )
273        .bind(status)
274        .bind(reason)
275        .bind(id)
276        .execute(&self.db)
277        .await?;
278        Ok(())
279    }
280
281    /// Update changeset status with optimistic locking: the transition only
282    /// succeeds when the current state matches one of `expected_states`.
283    /// Returns an error if the row was not updated (state mismatch or not found).
284    pub async fn update_status_if(
285        &self,
286        id: Uuid,
287        new_status: &str,
288        expected_states: &[&str],
289    ) -> dk_core::Result<()> {
290        self.update_status_if_with_reason(id, new_status, expected_states, "").await
291    }
292
293    /// Like `update_status_if` but also records a reason for the transition.
294    pub async fn update_status_if_with_reason(
295        &self,
296        id: Uuid,
297        new_status: &str,
298        expected_states: &[&str],
299        reason: &str,
300    ) -> dk_core::Result<()> {
301        let states: Vec<String> = expected_states.iter().map(|s| s.to_string()).collect();
302        let result = sqlx::query(
303            "UPDATE changesets SET state = $1, reason = $2, updated_at = now() WHERE id = $3 AND state = ANY($4)",
304        )
305        .bind(new_status)
306        .bind(reason)
307        .bind(id)
308        .bind(&states)
309        .execute(&self.db)
310        .await?;
311
312        if result.rows_affected() == 0 {
313            return Err(dk_core::Error::Internal(format!(
314                "changeset {} not found or not in expected state (expected one of: {:?})",
315                id, expected_states,
316            )));
317        }
318        Ok(())
319    }
320
321    pub async fn set_merged(&self, id: Uuid, commit_hash: &str) -> dk_core::Result<()> {
322        sqlx::query(
323            "UPDATE changesets SET state = 'merged', reason = 'merge completed', merged_version = $1, merged_at = now(), updated_at = now() WHERE id = $2",
324        )
325        .bind(commit_hash)
326        .bind(id)
327        .execute(&self.db)
328        .await?;
329        Ok(())
330    }
331
332    pub async fn upsert_file(
333        &self,
334        changeset_id: Uuid,
335        file_path: &str,
336        operation: &str,
337        content: Option<&str>,
338    ) -> dk_core::Result<()> {
339        sqlx::query(
340            r#"INSERT INTO changeset_files (changeset_id, file_path, operation, content)
341               VALUES ($1, $2, $3, $4)
342               ON CONFLICT (changeset_id, file_path) DO UPDATE SET
343                   operation = EXCLUDED.operation,
344                   content = EXCLUDED.content"#,
345        )
346        .bind(changeset_id)
347        .bind(file_path)
348        .bind(operation)
349        .bind(content)
350        .execute(&self.db)
351        .await?;
352        Ok(())
353    }
354
355    pub async fn get_files(&self, changeset_id: Uuid) -> dk_core::Result<Vec<ChangesetFile>> {
356        let rows: Vec<(Uuid, String, String, Option<String>)> = sqlx::query_as(
357            "SELECT changeset_id, file_path, operation, content FROM changeset_files WHERE changeset_id = $1",
358        )
359        .bind(changeset_id)
360        .fetch_all(&self.db)
361        .await?;
362
363        Ok(rows
364            .into_iter()
365            .map(|r| ChangesetFile {
366                changeset_id: r.0,
367                file_path: r.1,
368                operation: r.2,
369                content: r.3,
370            })
371            .collect())
372    }
373
374    /// Lightweight query returning only file metadata (path, operation, size)
375    /// without loading the full content column.
376    pub async fn get_files_metadata(&self, changeset_id: Uuid) -> dk_core::Result<Vec<ChangesetFileMeta>> {
377        let rows: Vec<(String, String, i64)> = sqlx::query_as(
378            "SELECT file_path, operation, COALESCE(LENGTH(content), 0)::bigint AS size_bytes FROM changeset_files WHERE changeset_id = $1",
379        )
380        .bind(changeset_id)
381        .fetch_all(&self.db)
382        .await?;
383
384        Ok(rows
385            .into_iter()
386            .map(|r| ChangesetFileMeta {
387                file_path: r.0,
388                operation: r.1,
389                size_bytes: r.2,
390            })
391            .collect())
392    }
393
394    pub async fn record_affected_symbol(
395        &self,
396        changeset_id: Uuid,
397        symbol_id: SymbolId,
398        qualified_name: &str,
399    ) -> dk_core::Result<()> {
400        sqlx::query(
401            r#"INSERT INTO changeset_symbols (changeset_id, symbol_id, symbol_qualified_name)
402               VALUES ($1, $2, $3)
403               ON CONFLICT DO NOTHING"#,
404        )
405        .bind(changeset_id)
406        .bind(symbol_id)
407        .bind(qualified_name)
408        .execute(&self.db)
409        .await?;
410        Ok(())
411    }
412
413    pub async fn get_affected_symbols(&self, changeset_id: Uuid) -> dk_core::Result<Vec<(SymbolId, String)>> {
414        let rows: Vec<(Uuid, String)> = sqlx::query_as(
415            "SELECT symbol_id, symbol_qualified_name FROM changeset_symbols WHERE changeset_id = $1",
416        )
417        .bind(changeset_id)
418        .fetch_all(&self.db)
419        .await?;
420        Ok(rows)
421    }
422
423    /// List live competitors on a file path.
424    ///
425    /// Returns every changeset in `{submitted, verifying, approved}` for
426    /// `repo_id` that has a `changeset_files` row for `path`. `draft`,
427    /// `merged`, `rejected`, and `closed` are filtered at the SQL level —
428    /// `draft` is session-local, `merged` is handled by the AST merger at
429    /// `dk_merge`, and the rest are inert. Policy (self-exclusion, stale
430    /// comparison vs. session read timestamp) is applied by the pure
431    /// `dk_protocol::stale_overlay::is_stale` helper.
432    ///
433    /// Used by the STALE_OVERLAY pre-write check in `handle_file_write`.
434    pub async fn list_path_competitors(
435        &self,
436        repo_id: RepoId,
437        path: &str,
438    ) -> dk_core::Result<Vec<(Uuid, Option<Uuid>, String, DateTime<Utc>)>> {
439        let rows: Vec<(Uuid, Option<Uuid>, String, DateTime<Utc>)> = sqlx::query_as(
440            r#"SELECT c.id, c.session_id, c.state, c.updated_at
441               FROM changesets c
442               JOIN changeset_files cf ON cf.changeset_id = c.id
443               WHERE c.repo_id = $1
444                 AND cf.file_path = $2
445                 AND c.state IN ('submitted', 'verifying', 'approved')"#,
446        )
447        .bind(repo_id)
448        .bind(path)
449        .fetch_all(&self.db)
450        .await?;
451        Ok(rows)
452    }
453
454    /// Find changesets that conflict with ours.
455    /// Only considers changesets merged AFTER our base_version —
456    /// i.e. changes the agent didn't know about when it started.
457    pub async fn find_conflicting_changesets(
458        &self,
459        repo_id: RepoId,
460        base_version: &str,
461        my_changeset_id: Uuid,
462    ) -> dk_core::Result<Vec<(Uuid, Vec<String>)>> {
463        let rows: Vec<(Uuid, String)> = sqlx::query_as(
464            r#"SELECT DISTINCT cs.changeset_id, cs.symbol_qualified_name
465               FROM changeset_symbols cs
466               JOIN changesets c ON c.id = cs.changeset_id
467               WHERE c.repo_id = $1
468                 AND c.state = 'merged'
469                 AND c.id != $2
470                 AND c.merged_version IS NOT NULL
471                 AND c.merged_version != $3
472                 AND cs.symbol_qualified_name IN (
473                     SELECT symbol_qualified_name FROM changeset_symbols WHERE changeset_id = $2
474                 )"#,
475        )
476        .bind(repo_id)
477        .bind(my_changeset_id)
478        .bind(base_version)
479        .fetch_all(&self.db)
480        .await?;
481
482        let mut map: std::collections::HashMap<Uuid, Vec<String>> = std::collections::HashMap::new();
483        for (cs_id, sym_name) in rows {
484            map.entry(cs_id).or_default().push(sym_name);
485        }
486        Ok(map.into_iter().collect())
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493
494    /// Verify the source_branch format produced by `create()`.
495    /// Branch format: `{intent_slug}/{agent_name}`.
496    fn slugify_intent(intent: &str) -> String {
497        let slug: String = intent
498            .to_lowercase()
499            .chars()
500            .map(|c| if c.is_alphanumeric() || c == '-' { c } else { '-' })
501            .collect::<String>()
502            .trim_matches('-')
503            .to_string();
504        if slug.len() > 50 {
505            slug[..50].trim_end_matches('-').to_string()
506        } else {
507            slug
508        }
509    }
510
511    #[test]
512    fn source_branch_format_uses_intent_slug() {
513        let intent = "Fix UI bugs";
514        let agent_name = "agent-1";
515        let source_branch = format!("{}/{}", slugify_intent(intent), agent_name);
516        assert_eq!(source_branch, "fix-ui-bugs/agent-1");
517    }
518
519    #[test]
520    fn source_branch_format_with_custom_name() {
521        let intent = "Add comments endpoint";
522        let agent_name = "feature-bot";
523        let source_branch = format!("{}/{}", slugify_intent(intent), agent_name);
524        assert_eq!(source_branch, "add-comments-endpoint/feature-bot");
525    }
526
527    #[test]
528    fn target_branch_is_main() {
529        // create() hardcodes target_branch to "main"
530        let target_branch = "main";
531        assert_eq!(target_branch, "main");
532    }
533
534    /// Test scaffolding fixture: a "draft" Changeset with sensible defaults
535    /// (empty/None for nullable fields, "main" as target, now() for
536    /// timestamps). Tests override only the fields they actually care about
537    /// via struct-update syntax (`..test_changeset_fixture()`), which keeps
538    /// each test focused and prevents every new struct field from forcing
539    /// an edit across every test literal.
540    fn test_changeset_fixture() -> Changeset {
541        let now = Utc::now();
542        Changeset {
543            id: Uuid::new_v4(),
544            repo_id: Uuid::new_v4(),
545            number: 1,
546            title: String::new(),
547            intent_summary: None,
548            source_branch: "agent/test".to_string(),
549            target_branch: "main".to_string(),
550            state: "draft".to_string(),
551            reason: String::new(),
552            session_id: None,
553            agent_id: None,
554            agent_name: None,
555            author_id: None,
556            base_version: None,
557            merged_version: None,
558            created_at: now,
559            updated_at: now,
560            merged_at: None,
561            parent_changeset_id: None,
562        }
563    }
564
565    /// Verify that a manually-constructed Changeset (matching the shape
566    /// returned by `create()`) has the correct branch and agent fields.
567    #[test]
568    fn changeset_create_shape_has_correct_branches() {
569        let repo_id = Uuid::new_v4();
570        let session_id = Uuid::new_v4();
571        let agent_id = "test-agent";
572        let intent = "fix all the bugs";
573
574        let source_branch = format!("agent/{}", agent_id);
575
576        let cs = Changeset {
577            repo_id,
578            title: intent.to_string(),
579            intent_summary: Some(intent.to_string()),
580            source_branch: source_branch.clone(),
581            session_id: Some(session_id),
582            agent_id: Some(agent_id.to_string()),
583            agent_name: Some(agent_id.to_string()),
584            base_version: Some("abc123".to_string()),
585            ..test_changeset_fixture()
586        };
587
588        assert_eq!(cs.source_branch, "agent/test-agent");
589        assert_eq!(cs.target_branch, "main");
590        assert_eq!(cs.agent_name.as_deref(), Some("test-agent"));
591        assert_eq!(cs.agent_id, cs.agent_name, "agent_name should equal agent_id per create()");
592        assert!(cs.merged_at.is_none());
593        assert!(cs.merged_version.is_none());
594    }
595
596    /// Verify the Changeset struct fields are all accessible and have
597    /// the expected types (compile-time check + runtime assertions).
598    #[test]
599    fn changeset_all_fields_accessible() {
600        let id = Uuid::new_v4();
601        let repo_id = Uuid::new_v4();
602
603        let cs = Changeset {
604            id,
605            repo_id,
606            number: 42,
607            title: "test".to_string(),
608            source_branch: "agent/a".to_string(),
609            ..test_changeset_fixture()
610        };
611
612        assert_eq!(cs.id, id);
613        assert_eq!(cs.repo_id, repo_id);
614        assert_eq!(cs.number, 42);
615        assert_eq!(cs.title, "test");
616        assert!(cs.intent_summary.is_none());
617        assert!(cs.session_id.is_none());
618        assert!(cs.agent_id.is_none());
619        assert!(cs.agent_name.is_none());
620        assert!(cs.author_id.is_none());
621        assert!(cs.base_version.is_none());
622        assert!(cs.merged_version.is_none());
623        assert!(cs.merged_at.is_none());
624    }
625
626    #[test]
627    fn changeset_file_meta_struct() {
628        let meta = ChangesetFileMeta {
629            file_path: "src/main.rs".to_string(),
630            operation: "modify".to_string(),
631            size_bytes: 1024,
632        };
633        assert_eq!(meta.file_path, "src/main.rs");
634        assert_eq!(meta.operation, "modify");
635        assert_eq!(meta.size_bytes, 1024);
636    }
637
638    #[test]
639    fn changeset_file_struct() {
640        let cf = ChangesetFile {
641            changeset_id: Uuid::new_v4(),
642            file_path: "lib.rs".to_string(),
643            operation: "add".to_string(),
644            content: Some("fn main() {}".to_string()),
645        };
646        assert_eq!(cf.file_path, "lib.rs");
647        assert_eq!(cf.operation, "add");
648        assert!(cf.content.is_some());
649    }
650
651    #[test]
652    fn changeset_clone_produces_equal_values() {
653        let cs = Changeset {
654            title: "clone test".to_string(),
655            intent_summary: Some("intent".to_string()),
656            source_branch: "agent/x".to_string(),
657            agent_id: Some("x".to_string()),
658            agent_name: Some("x".to_string()),
659            ..test_changeset_fixture()
660        };
661
662        let cloned = cs.clone();
663        assert_eq!(cs.id, cloned.id);
664        assert_eq!(cs.source_branch, cloned.source_branch);
665        assert_eq!(cs.target_branch, cloned.target_branch);
666        assert_eq!(cs.state, cloned.state);
667    }
668}
dk_engine/changeset.rs

dk_engine/
changeset.rs
