补齐 AI 任务与历史清理的内存边界 (#206)

基于当前 master（含 #204）移植未进入 #203 的内存边界修复。

- 补齐外部生成历史清理 procedure 的 SATS tuple 解析与回归测试
- 限制 AI 流式文本每阶段 chunk 数量并分批释放明细
- 修正进程内失败收口的容量检查顺序
- 分批删除外部生成审计事件并同步更新契约文档

验证：
- cargo test -p module-ai -p spacetime-module
- cargo check -p api-server
- npm run test -- scripts/spacetime-migration-common.test.ts
- npm run check:spacetime-schema
- npm run check:production-ops
- npm run check:encoding
- git diff --check

Reviewed-on: http://192.168.35.82/git/GenarrativeAI/Genarrative/pulls/206
Co-authored-by: kdletters <kdletters@qq.com>
Co-committed-by: kdletters <kdletters@qq.com>

server-rs/crates/module-ai/src/application/store.rs

@@ -114,18 +114,25 @@ impl InMemoryAiTaskStore {
                 .insert(task_id.trim().to_string(), previous_task);
             return Err(error);
         }
+        let released_text_chunks = if snapshot.status.is_terminal() {
+            state.text_chunks.remove(task_id.trim())
+        } else {
+            None
+        };
         let retained_output_bytes = retained_output_bytes(&state);
         if retained_output_bytes > MAX_AI_TASK_RETAINED_OUTPUT_BYTES {
             state
                 .tasks
                 .insert(task_id.trim().to_string(), previous_task);
+            if let Some(text_chunks) = released_text_chunks {
+                state
+                    .text_chunks
+                    .insert(task_id.trim().to_string(), text_chunks);
+            }
             return Err(AiTaskServiceError::Store(
                 "AI 任务仓储输出工作集超过内存上限".to_string(),
             ));
         }
-        if snapshot.status.is_terminal() {
-            state.text_chunks.remove(task_id.trim());
-        }
         Ok(snapshot)
     }
 
@@ -166,6 +173,13 @@ impl InMemoryAiTaskStore {
                 .get_mut(&chunk.task_id)
                 .ok_or(AiTaskServiceError::TaskNotFound)?;
             let stage_chunks = chunks.entry(chunk.stage_kind).or_default();
+            if !stage_chunks.contains_key(&chunk.sequence)
+                && stage_chunks.len() >= crate::MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE
+            {
+                return Err(AiTaskServiceError::Store(
+                    "AI 任务文本 chunk 数量超过内存上限".to_string(),
+                ));
+            }
             let previous_chunk = stage_chunks.insert(chunk.sequence, chunk.delta_text.clone());
             let aggregated_bytes = stage_chunks
                 .values()
@@ -352,3 +366,140 @@ fn validate_task_memory_limits(task: &AiTaskSnapshot) -> Result<(), AiTaskServic
     validate_ai_task_snapshot_memory_limits(task)
         .map_err(|message| AiTaskServiceError::Store(message.to_string()))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn build_running_task(task_id: &str) -> AiTaskSnapshot {
+        AiTaskSnapshot {
+            task_id: task_id.to_string(),
+            task_kind: crate::AiTaskKind::CharacterChat,
+            owner_user_id: "user-1".to_string(),
+            request_label: "测试任务".to_string(),
+            source_module: "test".to_string(),
+            source_entity_id: None,
+            request_payload_json: None,
+            status: AiTaskStatus::Running,
+            failure_message: None,
+            stages: vec![crate::AiTaskStageSnapshot {
+                stage_kind: crate::AiTaskStageKind::RequestModel,
+                label: "请求模型".to_string(),
+                detail: "测试".to_string(),
+                order: 0,
+                status: AiTaskStageStatus::Running,
+                text_output: None,
+                structured_payload_json: None,
+                warning_messages: Vec::new(),
+                started_at_micros: Some(1),
+                completed_at_micros: None,
+            }],
+            result_references: Vec::new(),
+            latest_text_output: None,
+            latest_structured_payload_json: None,
+            version: 1,
+            created_at_micros: 1,
+            started_at_micros: Some(1),
+            completed_at_micros: None,
+            updated_at_micros: 1,
+        }
+    }
+
+    #[test]
+    fn append_text_chunk_rejects_excessive_chunk_count() {
+        let store = InMemoryAiTaskStore::default();
+        let task_id = "task-chunk-limit";
+        let task = build_running_task(task_id);
+        let mut state = store.inner.lock().expect("store lock should be available");
+        state.tasks.insert(task_id.to_string(), task);
+        state.text_chunks.insert(
+            task_id.to_string(),
+            HashMap::from([(
+                crate::AiTaskStageKind::RequestModel,
+                (1..=crate::MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE as u32)
+                    .map(|sequence| (sequence, "a".to_string()))
+                    .collect(),
+            )]),
+        );
+        drop(state);
+
+        let error = store
+            .append_text_chunk(AiTextChunkSnapshot {
+                chunk_id: "chunk-overflow".to_string(),
+                task_id: task_id.to_string(),
+                stage_kind: crate::AiTaskStageKind::RequestModel,
+                sequence: crate::MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE as u32 + 1,
+                delta_text: "b".to_string(),
+                created_at_micros: 2,
+            })
+            .expect_err("a new chunk beyond the count cap should fail");
+        assert!(matches!(
+            error,
+            AiTaskServiceError::Store(message) if message.contains("chunk 数量")
+        ));
+    }
+
+    #[test]
+    fn terminal_failure_releases_chunks_before_global_cap_check() {
+        let store = InMemoryAiTaskStore::default();
+        let target_id = "task-terminal-release";
+        let target = build_running_task(target_id);
+        let mut state = store.inner.lock().expect("store lock should be available");
+        state.tasks.insert(target_id.to_string(), target);
+        state.text_chunks.insert(
+            target_id.to_string(),
+            HashMap::from([(
+                crate::AiTaskStageKind::RequestModel,
+                BTreeMap::from([(1, "t".repeat(crate::MAX_AI_TASK_TEXT_OUTPUT_BYTES))]),
+            )]),
+        );
+
+        let desired_retained = crate::MAX_AI_TASK_RETAINED_OUTPUT_BYTES
+            .saturating_sub(crate::MAX_AI_TASK_FAILURE_MESSAGE_BYTES)
+            .saturating_add(crate::MAX_AI_TASK_FAILURE_MESSAGE_BYTES / 2)
+            .saturating_sub(4 * 1024);
+        let mut filler_index = 0_u32;
+        while retained_output_bytes(&state) < desired_retained {
+            let remaining = desired_retained.saturating_sub(retained_output_bytes(&state));
+            let bytes = remaining.min(crate::MAX_AI_TASK_TEXT_OUTPUT_BYTES);
+            if bytes == 0 {
+                break;
+            }
+            let filler_id = format!("task-filler-{filler_index}");
+            filler_index += 1;
+            state
+                .tasks
+                .insert(filler_id.clone(), build_running_task(&filler_id));
+            state.text_chunks.insert(
+                filler_id,
+                HashMap::from([(
+                    crate::AiTaskStageKind::RequestModel,
+                    BTreeMap::from([(1, "f".repeat(bytes))]),
+                )]),
+            );
+        }
+        let retained_before_failure = retained_output_bytes(&state);
+        assert!(retained_before_failure <= crate::MAX_AI_TASK_RETAINED_OUTPUT_BYTES);
+        assert!(
+            retained_before_failure + crate::MAX_AI_TASK_FAILURE_MESSAGE_BYTES
+                > crate::MAX_AI_TASK_RETAINED_OUTPUT_BYTES
+        );
+        drop(state);
+
+        let failed = store
+            .update_task(target_id, |task| {
+                task.status = AiTaskStatus::Failed;
+                task.failure_message = Some("f".repeat(crate::MAX_AI_TASK_FAILURE_MESSAGE_BYTES));
+                task.completed_at_micros = Some(2);
+                task.updated_at_micros = 2;
+                task.version += 1;
+                Ok(())
+            })
+            .expect("terminal transition should release chunks before checking the cap");
+        assert_eq!(failed.status, AiTaskStatus::Failed);
+        assert_eq!(
+            failed.failure_message.as_deref().map(str::len),
+            Some(crate::MAX_AI_TASK_FAILURE_MESSAGE_BYTES)
+        );
+    }
+}

server-rs/crates/module-ai/src/domain.rs

@@ -16,7 +16,8 @@ pub use limits::{
     MAX_AI_TASK_RESULT_REFERENCES, MAX_AI_TASK_RETAINED_OUTPUT_BYTES, MAX_AI_TASK_RETAINED_TASKS,
     MAX_AI_TASK_SOURCE_ENTITY_ID_BYTES, MAX_AI_TASK_SOURCE_MODULE_BYTES,
     MAX_AI_TASK_STAGE_DETAIL_BYTES, MAX_AI_TASK_STAGE_LABEL_BYTES,
-    MAX_AI_TASK_STRUCTURED_OUTPUT_BYTES, MAX_AI_TASK_TEXT_OUTPUT_BYTES, MAX_AI_TASK_WARNING_BYTES,
+    MAX_AI_TASK_STRUCTURED_OUTPUT_BYTES, MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE,
+    MAX_AI_TASK_TEXT_OUTPUT_BYTES, MAX_AI_TASK_WARNING_BYTES,
     validate_ai_task_snapshot_memory_limits,
 };
 pub use types::{

server-rs/crates/module-ai/src/domain/limits.rs

@@ -9,6 +9,8 @@ pub const MAX_AI_TASK_SOURCE_ENTITY_ID_BYTES: usize = 512;
 pub const MAX_AI_TASK_STAGE_LABEL_BYTES: usize = 4 * 1024;
 pub const MAX_AI_TASK_STAGE_DETAIL_BYTES: usize = 8 * 1024;
 pub const MAX_AI_TASK_TEXT_OUTPUT_BYTES: usize = 512 * 1024;
+// provider 产生大量细小流式增量时，限制行和索引开销。
+pub const MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE: usize = 8 * 1024;
 pub const MAX_AI_TASK_STRUCTURED_OUTPUT_BYTES: usize = 512 * 1024;
 pub const MAX_AI_TASK_WARNING_BYTES: usize = 64 * 1024;
 pub const MAX_AI_TASK_REQUEST_PAYLOAD_BYTES: usize = 512 * 1024;

server-rs/crates/module-ai/src/lib.rs

@@ -21,10 +21,10 @@ pub use domain::{
     MAX_AI_TASK_RETAINED_OUTPUT_BYTES, MAX_AI_TASK_RETAINED_TASKS,
     MAX_AI_TASK_SOURCE_ENTITY_ID_BYTES, MAX_AI_TASK_SOURCE_MODULE_BYTES,
     MAX_AI_TASK_STAGE_DETAIL_BYTES, MAX_AI_TASK_STAGE_LABEL_BYTES,
-    MAX_AI_TASK_STRUCTURED_OUTPUT_BYTES, MAX_AI_TASK_TEXT_OUTPUT_BYTES, MAX_AI_TASK_WARNING_BYTES,
-    generate_ai_result_ref_id, generate_ai_task_id, generate_ai_task_stage_id,
-    generate_ai_text_chunk_id, normalize_optional_text, normalize_string_list,
-    validate_ai_task_snapshot_memory_limits,
+    MAX_AI_TASK_STRUCTURED_OUTPUT_BYTES, MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE,
+    MAX_AI_TASK_TEXT_OUTPUT_BYTES, MAX_AI_TASK_WARNING_BYTES, generate_ai_result_ref_id,
+    generate_ai_task_id, generate_ai_task_stage_id, generate_ai_text_chunk_id,
+    normalize_optional_text, normalize_string_list, validate_ai_task_snapshot_memory_limits,
 };
 pub use errors::{AiTaskFieldError, AiTaskServiceError};
 pub use events::AiTaskDomainEvent;

server-rs/crates/spacetime-module/src/ai/stages.rs

@@ -1,9 +1,12 @@
 use crate::*;
 use module_ai::{
-    MAX_AI_TASK_TEXT_OUTPUT_BYTES, generate_ai_result_ref_id, generate_ai_text_chunk_id,
-    normalize_optional_text, normalize_string_list, validate_ai_task_snapshot_memory_limits,
+    MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE, MAX_AI_TASK_TEXT_OUTPUT_BYTES, generate_ai_result_ref_id,
+    generate_ai_text_chunk_id, normalize_optional_text, normalize_string_list,
+    validate_ai_task_snapshot_memory_limits,
 };
 
+const AI_TEXT_CHUNK_DELETE_BATCH_SIZE: usize = 256;
+
 #[spacetimedb::table(
     accessor = ai_task_stage,
     index(accessor = by_ai_task_stage_task_id, btree(columns = [task_id])),
@@ -178,7 +181,7 @@ pub(crate) fn append_ai_text_chunk_tx(
     if input.sequence == 0 {
         return Err("ai_text_chunk.sequence 必须大于 0".to_string());
     }
-    if input.delta_text.trim().len() > MAX_AI_TASK_TEXT_OUTPUT_BYTES {
+    if input.delta_text.len() > MAX_AI_TASK_TEXT_OUTPUT_BYTES {
         return Err("AI 任务文本输出超过内存上限".to_string());
     }
 
@@ -340,15 +343,25 @@ pub(crate) fn replace_ai_task_stages(
 }
 
 pub(crate) fn delete_ai_text_chunks_for_task(ctx: &ReducerContext, task_id: &str) {
-    let chunk_row_ids = ctx
-        .db
-        .ai_text_chunk()
-        .by_ai_text_chunk_task_id()
-        .filter(task_id)
-        .map(|row| row.text_chunk_row_id.clone())
-        .collect::<Vec<_>>();
-    for row_id in chunk_row_ids {
-        ctx.db.ai_text_chunk().text_chunk_row_id().delete(&row_id);
+    loop {
+        let chunk_row_ids = ctx
+            .db
+            .ai_text_chunk()
+            .by_ai_text_chunk_task_id()
+            .filter(task_id)
+            .take(AI_TEXT_CHUNK_DELETE_BATCH_SIZE)
+            .map(|row| row.text_chunk_row_id.clone())
+            .collect::<Vec<_>>();
+        if chunk_row_ids.is_empty() {
+            break;
+        }
+        let batch_len = chunk_row_ids.len();
+        for row_id in chunk_row_ids {
+            ctx.db.ai_text_chunk().text_chunk_row_id().delete(&row_id);
+        }
+        if batch_len < AI_TEXT_CHUNK_DELETE_BATCH_SIZE {
+            break;
+        }
     }
 }
 
@@ -358,6 +371,7 @@ pub(crate) fn collect_ai_stage_text_output(
     stage_kind: AiTaskStageKind,
 ) -> Result<Option<String>, String> {
     let mut chunks = Vec::new();
+    let mut chunk_count = 0_usize;
     let mut aggregated_bytes = 0_usize;
     for row in ctx
         .db
@@ -366,6 +380,10 @@ pub(crate) fn collect_ai_stage_text_output(
         .filter(task_id)
         .filter(|row| row.task_id == task_id && row.stage_kind == stage_kind)
     {
+        chunk_count = chunk_count.saturating_add(1);
+        if chunk_count > MAX_AI_TASK_TEXT_CHUNKS_PER_STAGE {
+            return Err("AI 任务文本 chunk 数量超过内存上限".to_string());
+        }
         aggregated_bytes = aggregated_bytes.saturating_add(row.delta_text.len());
         if aggregated_bytes > MAX_AI_TASK_TEXT_OUTPUT_BYTES {
             return Err("AI 任务文本输出超过内存上限".to_string());

server-rs/crates/spacetime-module/src/external_generation.rs

@@ -23,6 +23,7 @@ const MAX_EXTERNAL_GENERATION_REQUEST_PROMPT_CHARS: usize = 2_048;
 const MAX_EXTERNAL_GENERATION_ERROR_MESSAGE_CHARS: usize = 2_048;
 const MAX_EXTERNAL_GENERATION_WARNING_MESSAGE_CHARS: usize = 2_048;
 const MAX_EXTERNAL_GENERATION_MAINTENANCE_BATCH_SIZE: u32 = 25;
+const EXTERNAL_GENERATION_EVENT_DELETE_BATCH_SIZE: usize = 256;
 const INLINE_MEDIA_REMOVED_PLACEHOLDER: &str = "[inline-media-removed]";
 const INLINE_MEDIA_ERROR_REDACTED_MESSAGE: &str = "外部生成失败（错误详情含内联媒体引用，已省略）";
 const INLINE_MEDIA_WARNING_REDACTED_MESSAGE: &str =
@@ -1433,6 +1434,15 @@ fn prune_external_generation_job_history_tx(
         .clamp(1, MAX_EXTERNAL_GENERATION_MAINTENANCE_BATCH_SIZE) as usize;
     let cursor_range = external_generation_job_maintenance_cursor_range(cursor_job_id.as_deref());
     let cursor_to_skip = cursor_job_id.clone();
+    // 若 cursor 对应的任务仍存在，说明上一事务只删完了事件的一部分，或 dry-run
+    // 尚未执行 apply；下一次必须包含该任务继续清理，成功删除后它会自然消失。
+    let include_existing_cursor = cursor_job_id.as_deref().is_some_and(|cursor| {
+        ctx.db
+            .external_generation_job()
+            .job_id()
+            .find(&cursor.to_string())
+            .is_some()
+    });
     let rows = ctx
         .db
         .external_generation_job()
@@ -1441,7 +1451,7 @@ fn prune_external_generation_job_history_tx(
         .filter(move |row| {
             cursor_to_skip
                 .as_deref()
-                .is_none_or(|cursor| row.job_id != cursor)
+                .is_none_or(|cursor| row.job_id != cursor || include_existing_cursor)
         });
     let (job_ids, next_cursor_job_id, has_more, scanned_count) =
         select_external_generation_job_ids_for_maintenance(rows, limit, |row| {
@@ -1462,6 +1472,7 @@ fn prune_external_generation_job_history_tx(
     let mut deleted_job_count = 0u32;
     let mut deleted_summary_count = 0u32;
     let mut deleted_event_count = 0u32;
+    let mut pending_event_cursor_job_id = None;
     if !input.dry_run {
         for job_id in &job_ids {
             let Some(row) = ctx.db.external_generation_job().job_id().find(job_id) else {
@@ -1484,8 +1495,13 @@ fn prune_external_generation_job_history_tx(
                 continue;
             }
 
-            deleted_event_count = deleted_event_count
-                .saturating_add(delete_external_generation_job_events_for_job(ctx, job_id));
+            let (deleted_for_job, has_more_events) =
+                delete_external_generation_job_events_for_job(ctx, job_id);
+            deleted_event_count = deleted_event_count.saturating_add(deleted_for_job);
+            if has_more_events {
+                pending_event_cursor_job_id = Some(job_id.clone());
+                break;
+            }
             ctx.db
                 .external_generation_job_summary()
                 .job_id()
@@ -1496,6 +1512,10 @@ fn prune_external_generation_job_history_tx(
         }
     }
 
+    let (next_cursor_job_id, has_more) = pending_event_cursor_job_id
+        .map(|job_id| (Some(job_id), true))
+        .unwrap_or((next_cursor_job_id, has_more));
+
     Ok(ExternalGenerationJobRetentionProcedureResult {
         ok: true,
         dry_run: input.dry_run,
@@ -1996,22 +2016,34 @@ fn select_external_generation_job_ids_for_maintenance(
     )
 }
 
-fn delete_external_generation_job_events_for_job(ctx: &ReducerContext, job_id: &str) -> u32 {
+fn delete_external_generation_job_events_for_job(
+    ctx: &ReducerContext,
+    job_id: &str,
+) -> (u32, bool) {
+    // 每次 procedure 最多删除一个固定批次；若仍有事件，保留 job/summary，调用方
+    // 通过同一个 job cursor 重试，避免单个任务把整段审计历史塞进一个事务写集。
     let event_ids = ctx
         .db
         .external_generation_job_event()
         .by_external_generation_job_event_job_id_only()
         .filter(job_id)
+        .take(EXTERNAL_GENERATION_EVENT_DELETE_BATCH_SIZE + 1)
         .map(|event| event.event_id.clone())
         .collect::<Vec<_>>();
-    let deleted_count = event_ids.len() as u32;
-    for event_id in event_ids {
+    let has_more = event_ids.len() > EXTERNAL_GENERATION_EVENT_DELETE_BATCH_SIZE;
+    let deleted_count = event_ids
+        .len()
+        .min(EXTERNAL_GENERATION_EVENT_DELETE_BATCH_SIZE) as u32;
+    for event_id in event_ids
+        .into_iter()
+        .take(EXTERNAL_GENERATION_EVENT_DELETE_BATCH_SIZE)
+    {
         ctx.db
             .external_generation_job_event()
             .event_id()
             .delete(&event_id);
     }
-    deleted_count
+    (deleted_count, has_more)
 }
 
 fn count_external_generation_job_summaries_for_owner(