feat(page-ai): add resumable run journal descriptors

2026-06-09 18:48:46 +08:00
parent 6ef233772e
commit 922965d30f
17 changed files with 3254 additions and 58 deletions
@@ -10,15 +10,15 @@ use crate::migrations;
 use crate::model::{
    password_hash_v1, share_token_hash_v1, AiAgentProfileAccessRecord, AiAgentProfileGrantRecord,
    AiAgentProfileRecord, AiExternalConversationBindingRecord, AiPolicyRecord,
-    AiRuntimeEventRecord, AiRuntimeRunRecord, AppendAiRuntimeEventInput, AppendAuditInput,
-    AuditLogRecord, AuthSessionRecord, AuthenticatePasswordInput, CreatePasswordIdentityInput,
-    CreateSessionInput, CreateShareLinkInput, CreatedShareLink, DirectoryGrantInput,
-    DirectoryGrantLookup, DirectoryGrantRecord, NavigationRecentRecord, OutboxEventInput,
-    OutboxEventRecord, ResolvedAccess, ResolvedAuthSession, ShareLinkRecord, SidebarShortcutRecord,
-    SyncStateRecord, UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput,
-    UpsertAiRuntimeRunInput, UpsertNavigationRecentInput, UpsertSidebarShortcutInput,
-    UpsertSyncStateInput, UpsertUserInput, UpsertUserUiPreferenceInput, UserRecord,
-    UserUiPreferenceRecord, WorkspaceRecord,
+    AiRuntimeEventRecord, AiRuntimeJournalEventRecord, AiRuntimeRunRecord,
+    AppendAiRuntimeEventInput, AppendAuditInput, AuditLogRecord, AuthSessionRecord,
+    AuthenticatePasswordInput, CreatePasswordIdentityInput, CreateSessionInput,
+    CreateShareLinkInput, CreatedShareLink, DirectoryGrantInput, DirectoryGrantLookup,
+    DirectoryGrantRecord, NavigationRecentRecord, OutboxEventInput, OutboxEventRecord,
+    ResolvedAccess, ResolvedAuthSession, ShareLinkRecord, SidebarShortcutRecord, SyncStateRecord,
+    UpsertAiExternalConversationBindingInput, UpsertAiPolicyInput, UpsertAiRuntimeRunInput,
+    UpsertNavigationRecentInput, UpsertSidebarShortcutInput, UpsertSyncStateInput, UpsertUserInput,
+    UpsertUserUiPreferenceInput, UserRecord, UserUiPreferenceRecord, WorkspaceRecord,
 };
 use crate::store::ControlPlaneStore;

@@ -424,6 +424,27 @@ fn row_to_ai_runtime_event(row: &rusqlite::Row<'_>) -> rusqlite::Result<AiRuntim
    })
 }

+fn row_to_ai_runtime_journal_event(
+    row: &rusqlite::Row<'_>,
+) -> rusqlite::Result<AiRuntimeJournalEventRecord> {
+    Ok(AiRuntimeJournalEventRecord {
+        seq: row.get(0)?,
+        event: AiRuntimeEventRecord {
+            id: row.get(1)?,
+            user_id: row.get(2)?,
+            workspace_id: row.get(3)?,
+            document_id: row.get(4)?,
+            session_id: row.get(5)?,
+            run_id: row.get(6)?,
+            profile: row.get(7)?,
+            acp_runtime: row.get(8)?,
+            event_type: row.get(9)?,
+            payload_json: row.get(10)?,
+            created_at: row.get(11)?,
+        },
+    })
+}
+
 fn row_to_ai_external_conversation_binding(
    row: &rusqlite::Row<'_>,
 ) -> rusqlite::Result<AiExternalConversationBindingRecord> {
@@ -2355,6 +2376,24 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
        Ok(rows)
    }

+    fn find_ai_runtime_run(
+        &self,
+        user_id: &str,
+        run_id: &str,
+    ) -> Result<Option<AiRuntimeRunRecord>, ControlPlaneError> {
+        let conn = self.conn.lock().unwrap();
+        conn.query_row(
+            "SELECT id, user_id, workspace_id, document_id, session_id, run_id, title, profile, acp_runtime, trace_id, status, runtime_json, payload_json, deleted_at, created_at, updated_at, revision
+             FROM ai_runtime_runs
+             WHERE user_id = ?1 AND run_id = ?2 AND deleted_at IS NULL
+             LIMIT 1",
+            params![user_id, run_id],
+            row_to_ai_runtime_run,
+        )
+        .optional()
+        .map_err(ControlPlaneError::from)
+    }
+
    fn append_ai_runtime_event(
        &self,
        input: AppendAiRuntimeEventInput,
@@ -2427,6 +2466,40 @@ impl ControlPlaneStore for SqliteControlPlaneStore {
        Ok(rows)
    }

+    fn list_ai_runtime_journal_events(
+        &self,
+        user_id: &str,
+        run_id: &str,
+        after_seq: i64,
+        limit: usize,
+    ) -> Result<Vec<AiRuntimeJournalEventRecord>, ControlPlaneError> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "WITH ordered AS (
+                 SELECT
+                   ROW_NUMBER() OVER (ORDER BY created_at ASC, id ASC) AS seq,
+                   id, user_id, workspace_id, document_id, session_id, run_id, profile,
+                   acp_runtime, event_type, payload_json, created_at
+                 FROM ai_runtime_events
+                 WHERE user_id = ?1 AND run_id = ?2
+             )
+             SELECT seq, id, user_id, workspace_id, document_id, session_id, run_id, profile,
+                    acp_runtime, event_type, payload_json, created_at
+             FROM ordered
+             WHERE seq > ?3
+             ORDER BY seq ASC
+             LIMIT ?4",
+        )?;
+        let rows = stmt
+            .query_map(
+                params![user_id, run_id, after_seq.max(0), limit.max(1) as i64],
+                row_to_ai_runtime_journal_event,
+            )?
+            .collect::<Result<Vec<_>, _>>()
+            .map_err(ControlPlaneError::from)?;
+        Ok(rows)
+    }
+
    fn upsert_ai_external_conversation_binding(
        &self,
        input: UpsertAiExternalConversationBindingInput,
@@ -3666,12 +3739,39 @@ mod tests {
            })
            .expect("append runtime event");
        assert_eq!(event.event_type, "message.delta");
+        store
+            .append_ai_runtime_event(AppendAiRuntimeEventInput {
+                id: None,
+                user_id: "ai_runtime_user".to_string(),
+                workspace_id: Some("ws_1".to_string()),
+                document_id: Some("doc_1".to_string()),
+                session_id: "sess_1".to_string(),
+                run_id: "run_1".to_string(),
+                profile: "reasonix".to_string(),
+                acp_runtime: "reasonix".to_string(),
+                event_type: "run.completed".to_string(),
+                payload_json: "{\"status\":\"completed\"}".to_string(),
+            })
+            .expect("append second runtime event");

        let events = store
            .list_ai_runtime_events("ai_runtime_user", "run_1", 10)
            .expect("list runtime events");
-        assert_eq!(events.len(), 1);
+        assert_eq!(events.len(), 2);
        assert_eq!(events[0].payload_json, "{\"text\":\"hello\"}");
+
+        let found = store
+            .find_ai_runtime_run("ai_runtime_user", "run_1")
+            .expect("find runtime run")
+            .expect("runtime run exists");
+        assert_eq!(found.status, "completed");
+
+        let journal_events = store
+            .list_ai_runtime_journal_events("ai_runtime_user", "run_1", 1, 10)
+            .expect("list runtime journal events");
+        assert_eq!(journal_events.len(), 1);
+        assert_eq!(journal_events[0].seq, 2);
+        assert_eq!(journal_events[0].event.event_type, "run.completed");
    }

    #[test]