car-server-core 0.55.0

//! In-process adapter to the daemon's existing assistant checkpoint/action log.
//! Owns no journal or model state: the daemon remains the single oplog writer.

use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use car_inference::Message;
use serde_json::{json, Value};
use tokio::sync::Mutex;

use super::governance::{
    completion_matrix_from_messages, AssistantCheckpoint, AssistantDurability,
    SupervisedActionRecord,
};
use crate::sync::SyncSubsystem;

/// Bound to one conversation and repository. Connection authorization belongs
/// to the caller; this adapter prevents accidentally reusing its checkpoint
/// under a different session or repository.
pub(crate) struct LocalAssistantDurability {
    sync: Arc<Mutex<SyncSubsystem>>,
    session_id: String,
    repository_root: PathBuf,
    revoked: AtomicBool,
}

impl LocalAssistantDurability {
    pub(crate) fn new(
        sync: Arc<Mutex<SyncSubsystem>>,
        session_id: String,
        repository_root: PathBuf,
    ) -> Self {
        Self {
            sync,
            session_id,
            repository_root,
            revoked: AtomicBool::new(false),
        }
    }

    pub(crate) fn revoke(&self) {
        self.revoked.store(true, Ordering::SeqCst);
    }

    fn check_writer(&self) -> Result<(), String> {
        if self.revoked.load(Ordering::SeqCst) {
            return Err("conversation writer was closed".into());
        }
        Ok(())
    }

    fn check_session(&self, session_id: &str) -> Result<(), String> {
        if session_id != self.session_id {
            return Err("checkpoint belongs to a different conversation".into());
        }
        Ok(())
    }

    fn check_repository(&self, checkpoint: &AssistantCheckpoint) -> Result<(), String> {
        if checkpoint.repository_root != self.repository_root {
            return Err("checkpoint belongs to a different repository".into());
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl AssistantDurability for LocalAssistantDurability {
    async fn load_checkpoint(
        &self,
        session_id: &str,
    ) -> Result<Option<AssistantCheckpoint>, String> {
        self.check_session(session_id)?;
        let checkpoint = self
            .sync
            .lock()
            .await
            .assistant_checkpoint_get(session_id)?;
        if let Some(checkpoint) = &checkpoint {
            self.check_repository(checkpoint)?;
        }
        Ok(checkpoint)
    }

    async fn checkpoint(
        &self,
        session_id: &str,
        messages: &[Message],
        reason: &str,
        goal: Option<Value>,
    ) -> Result<(), String> {
        self.check_session(session_id)?;
        // Read the revision and append under the same lock. A failed append
        // cannot advance an adapter-local counter or overwrite a newer writer.
        let mut sync = self.sync.lock().await;
        self.check_writer()?;
        let prior = sync.assistant_checkpoint_get(session_id)?;
        if let Some(checkpoint) = &prior {
            self.check_repository(checkpoint)?;
        }
        let revision = prior
            .as_ref()
            .map_or(0, |c| c.revision)
            .checked_add(1)
            .ok_or("checkpoint revision exhausted")?;
        sync.assistant_checkpoint_put(AssistantCheckpoint {
            id: session_id.to_string(),
            session_id: session_id.to_string(),
            revision,
            repository_root: self.repository_root.clone(),
            messages: messages.to_vec(),
            goal,
            compaction: Some(json!({ "reason": reason })),
            completion: completion_matrix_from_messages(messages),
        })?;
        Ok(())
    }

    async fn load_action(&self, action_id: &str) -> Result<Option<SupervisedActionRecord>, String> {
        let record = self.sync.lock().await.assistant_action_get(action_id)?;
        if let Some(record) = &record {
            self.check_session(&record.session_id)?;
        }
        Ok(record)
    }

    async fn record_action(&self, record: &SupervisedActionRecord) -> Result<(), String> {
        self.check_session(&record.session_id)?;
        let mut sync = self.sync.lock().await;
        self.check_writer()?;
        sync.assistant_action_put(record.clone())?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::session::ServerState;

    #[tokio::test]
    async fn exact_checkpoint_survives_daemon_restart_and_rejects_wrong_repository() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("repo");
        let state = ServerState::standalone(dir.path().join("journals"));
        let store = LocalAssistantDurability::new(
            state.sync_subsystem().unwrap(),
            "disc-test".into(),
            root.clone(),
        );
        let messages = vec![Message::System {
            content: "repo rules".into(),
        }];
        store
            .checkpoint("disc-test", &messages, "turn", None)
            .await
            .unwrap();
        assert!(store.load_checkpoint("another-session").await.is_err());
        drop(store);
        drop(state);

        let restarted = ServerState::standalone(dir.path().join("journals"));
        let sync = restarted.sync_subsystem().unwrap();
        let store = LocalAssistantDurability::new(sync.clone(), "disc-test".into(), root);
        let checkpoint = store.load_checkpoint("disc-test").await.unwrap().unwrap();
        assert_eq!(
            serde_json::to_value(checkpoint.messages).unwrap(),
            serde_json::to_value(&messages).unwrap()
        );
        assert_eq!(checkpoint.revision, 1);
        store
            .checkpoint("disc-test", &messages, "next", None)
            .await
            .unwrap();
        assert_eq!(
            store
                .load_checkpoint("disc-test")
                .await
                .unwrap()
                .unwrap()
                .revision,
            2
        );
        let foreign =
            LocalAssistantDurability::new(sync, "disc-test".into(), dir.path().join("other"));
        assert!(foreign.load_checkpoint("disc-test").await.is_err());
        assert!(foreign
            .checkpoint("disc-test", &messages, "wrong repo", None)
            .await
            .is_err());
    }
}