kmp_application/commands/
update_context.rs

1use std::sync::Arc;
2use std::time::SystemTime;
3
4use kmp_domain::{
5    CaseId, ContextEventChange, ContextEventStore, ContextUpdatedEvent, PortError,
6    ProjectionMutation, ProjectionWriter, Role,
7};
8
9use crate::ApplicationError;
10use crate::commands::memory_projection::memory_projection_mutations;
11
12#[derive(Debug, Clone, PartialEq, Eq)]
13pub struct UpdateContextChange {
14    pub operation: String,
15    pub entity_kind: String,
16    pub entity_id: String,
17    pub payload_json: String,
18    pub reason: String,
19    pub scopes: Vec<String>,
20}
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct UpdateContextCommand {
24    pub root_node_id: String,
25    pub role: String,
26    pub work_item_id: String,
27    pub changes: Vec<UpdateContextChange>,
28    pub expected_revision: Option<u64>,
29    pub expected_content_hash: Option<String>,
30    pub idempotency_key: Option<String>,
31    /// Digest of the logical command, computed by the caller *before* any
32    /// state-dependent translation. Optional: a caller that supplies none gets
33    /// idempotency compared on the translated changes, which refuses a replay
34    /// whenever translation saw different state — safe, but stricter than the
35    /// caller may mean.
36    pub logical_digest: Option<String>,
37    pub requested_by: Option<String>,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq)]
41pub struct AcceptedVersion {
42    pub revision: u64,
43    pub content_hash: String,
44    pub generator_version: String,
45}
46
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct UpdateContextOutcome {
49    pub accepted_version: AcceptedVersion,
50    /// Determined at the idempotency check, never inferred from a caller retry.
51    pub replayed: bool,
52    pub replayed_receipt: Option<kmp_domain::StoredCommandReceipt>,
53    pub warnings: Vec<String>,
54}
55
56#[derive(Debug)]
57pub struct UpdateContextUseCase<E, W = NoopProjectionWriter> {
58    event_store: Arc<E>,
59    projection_writer: W,
60    generator_version: &'static str,
61}
62
63#[derive(Debug, Clone, Copy, Default)]
64pub struct NoopProjectionWriter;
65
66impl ProjectionWriter for NoopProjectionWriter {
67    async fn apply_mutations(&self, _mutations: Vec<ProjectionMutation>) -> Result<(), PortError> {
68        Ok(())
69    }
70}
71
72impl<E> UpdateContextUseCase<E, NoopProjectionWriter>
73where
74    E: ContextEventStore + Send + Sync,
75{
76    pub fn new(event_store: Arc<E>, generator_version: &'static str) -> Self {
77        Self {
78            event_store,
79            projection_writer: NoopProjectionWriter,
80            generator_version,
81        }
82    }
83}
84
85impl<E, W> UpdateContextUseCase<E, W>
86where
87    E: ContextEventStore + Send + Sync,
88    W: ProjectionWriter + Send + Sync,
89{
90    pub fn new_with_projection_writer(
91        event_store: Arc<E>,
92        projection_writer: W,
93        generator_version: &'static str,
94    ) -> Self {
95        Self {
96            event_store,
97            projection_writer,
98            generator_version,
99        }
100    }
101
102    /// What an idempotency key was already accepted with, if anything: the
103    /// question a write asks before translating against a store its own
104    /// first apply has already changed.
105    pub async fn accepted_outcome(
106        &self,
107        idempotency_key: &str,
108    ) -> Result<Option<kmp_domain::IdempotentOutcome>, ApplicationError> {
109        Ok(self
110            .event_store
111            .find_by_idempotency_key(idempotency_key)
112            .await?)
113    }
114
115    pub async fn execute(
116        &self,
117        command: UpdateContextCommand,
118    ) -> Result<UpdateContextOutcome, ApplicationError> {
119        self.execute_after_read(command, &std::collections::BTreeMap::new())
120            .await
121    }
122
123    pub(crate) async fn execute_after_read(
124        &self,
125        command: UpdateContextCommand,
126        read_revisions: &std::collections::BTreeMap<String, u64>,
127    ) -> Result<UpdateContextOutcome, ApplicationError> {
128        let case_id = CaseId::new(&command.root_node_id)?;
129        let role = Role::new(&command.role)?;
130
131        // Compute content hash from actual change payloads. Duplicate idempotency keys
132        // must replay the same logical command, not mask a conflicting payload.
133        let content_hash = compute_content_hash(&command.changes);
134
135        // Idempotency check. The logical digest, when both sides carry one,
136        // is the honest comparison: the same logical command translated after
137        // its own first apply produces different changes (creates become
138        // updates), and comparing those would refuse a legitimate replay.
139        if let Some(ref key) = command.idempotency_key
140            && let Some(outcome) = self.event_store.find_by_idempotency_key(key).await?
141        {
142            let same_logical_command = match (&command.logical_digest, &outcome.logical_digest) {
143                (Some(ours), Some(stored)) => ours == stored,
144                _ => outcome.content_hash == content_hash,
145            };
146            if !same_logical_command {
147                return Err(ApplicationError::Ports(kmp_domain::PortError::Conflict(
148                    format!("idempotency key '{key}' was already accepted with different content"),
149                )));
150            }
151            return Ok(UpdateContextOutcome {
152                replayed: true,
153                replayed_receipt: outcome.receipt,
154                accepted_version: AcceptedVersion {
155                    revision: outcome.revision,
156                    content_hash: outcome.content_hash,
157                    generator_version: self.generator_version.to_string(),
158                },
159                warnings: vec![],
160            });
161        }
162
163        // Load current revision
164        let current_revision = self
165            .event_store
166            .current_revision(case_id.as_str(), role.as_str())
167            .await?;
168
169        // Validate revision precondition
170        let expected_revision = command.expected_revision.unwrap_or(current_revision);
171        if expected_revision != current_revision {
172            return Err(ApplicationError::RetryableConflict(format!(
173                "expected revision {expected_revision}, current is {current_revision}"
174            )));
175        }
176
177        // Validate content hash precondition
178        if let Some(ref expected_hash) = command.expected_content_hash
179            && let Some(ref current_hash) = self
180                .event_store
181                .current_content_hash(case_id.as_str(), role.as_str())
182                .await?
183            && expected_hash != current_hash
184        {
185            return Err(ApplicationError::RetryableConflict(format!(
186                "expected content hash '{expected_hash}', current is '{current_hash}'"
187            )));
188        }
189
190        let mut warnings = Vec::new();
191        if command.changes.is_empty() {
192            warnings.push("no changes supplied; update was accepted as a no-op".to_string());
193        }
194
195        // Build domain event
196        let event = ContextUpdatedEvent {
197            root_node_id: case_id.as_str().to_string(),
198            role: role.as_str().to_string(),
199            revision: current_revision + 1,
200            content_hash: content_hash.clone(),
201            changes: command
202                .changes
203                .iter()
204                .map(|c| ContextEventChange {
205                    operation: c.operation.clone(),
206                    entity_kind: c.entity_kind.clone(),
207                    entity_id: c.entity_id.clone(),
208                    payload_json: c.payload_json.clone(),
209                    reason: if c.reason.is_empty() {
210                        None
211                    } else {
212                        Some(c.reason.clone())
213                    },
214                    scopes: c.scopes.clone(),
215                })
216                .collect(),
217            idempotency_key: command.idempotency_key.clone(),
218            logical_digest: command.logical_digest.clone(),
219            requested_by: command.requested_by.clone(),
220            occurred_at: SystemTime::now(),
221        };
222
223        let mut projection_mutations =
224            memory_projection_mutations(&command, current_revision + 1, &content_hash)?;
225        let atomic = self.event_store.commits_projections_atomically();
226        let appended = if atomic {
227            let revisions = read_revisions
228                .iter()
229                .map(|(about, revision)| kmp_domain::ContextRevision {
230                    root_node_id: about.clone(),
231                    role: "memory".into(),
232                    revision: *revision,
233                })
234                .collect();
235            self.event_store
236                .append_projected(
237                    event,
238                    current_revision,
239                    revisions,
240                    std::mem::take(&mut projection_mutations),
241                )
242                .await
243        } else {
244            self.event_store.append(event, current_revision).await
245        };
246        let new_revision = appended.map_err(|error| match error {
247            // The revision was read above and changed before append. No
248            // part of this event was committed, so replaying the same
249            // logical command under its existing idempotency key is safe.
250            PortError::Conflict(message) => ApplicationError::RetryableConflict(message),
251            other => ApplicationError::Ports(other),
252        })?;
253
254        if !atomic && !projection_mutations.is_empty() {
255            self.projection_writer
256                .apply_mutations(projection_mutations)
257                .await?;
258        }
259
260        Ok(UpdateContextOutcome {
261            replayed: false,
262            replayed_receipt: None,
263            accepted_version: AcceptedVersion {
264                revision: new_revision,
265                content_hash,
266                generator_version: self.generator_version.to_string(),
267            },
268            warnings,
269        })
270    }
271
272    pub(crate) async fn memory_revision(&self, about: &str) -> Result<u64, ApplicationError> {
273        Ok(self.event_store.current_revision(about, "memory").await?)
274    }
275}
276
277/// Deterministic SHA-256 hash of context changes for optimistic concurrency.
278/// Stable across process restarts and machines.
279fn compute_content_hash(changes: &[UpdateContextChange]) -> String {
280    use sha2::{Digest, Sha256};
281    let mut hasher = Sha256::new();
282    for change in changes {
283        hasher.update(change.operation.as_bytes());
284        hasher.update(change.entity_kind.as_bytes());
285        hasher.update(change.entity_id.as_bytes());
286        hasher.update(change.payload_json.as_bytes());
287    }
288    format!("{:064x}", hasher.finalize())
289}
290
291#[cfg(test)]
292mod tests {
293    use std::sync::Arc;
294
295    use kmp_domain::ProjectionMutation;
296    use kmp_testkit::{InMemoryContextEventStore, InMemoryProjectionWriter};
297
298    use super::*;
299
300    fn event_store() -> Arc<InMemoryContextEventStore> {
301        Arc::new(InMemoryContextEventStore::new())
302    }
303
304    fn use_case(
305        store: Arc<InMemoryContextEventStore>,
306    ) -> UpdateContextUseCase<InMemoryContextEventStore> {
307        UpdateContextUseCase::new(store, "0.1.0")
308    }
309
310    fn sample_change() -> UpdateContextChange {
311        UpdateContextChange {
312            operation: "UPDATE".to_string(),
313            entity_kind: "node_detail".to_string(),
314            entity_id: "node-1".to_string(),
315            payload_json: r#"{"status":"ACTIVE"}"#.to_string(),
316            reason: "test".to_string(),
317            scopes: vec!["graph".to_string()],
318        }
319    }
320
321    fn memory_changes() -> Vec<UpdateContextChange> {
322        vec![
323            UpdateContextChange {
324                operation: "UPSERT".to_string(),
325                entity_kind: "memory_dimension".to_string(),
326                entity_id: "conversation:mcp".to_string(),
327                payload_json: r#"{"id":"conversation:mcp","kind":"conversation","title":"MCP smoke"}"#.to_string(),
328                reason: "test dimension".to_string(),
329                scopes: vec!["memory".to_string()],
330            },
331            UpdateContextChange {
332                operation: "UPSERT".to_string(),
333                entity_kind: "memory_entry".to_string(),
334                entity_id: "claim:mcp".to_string(),
335                payload_json: r#"{"id":"claim:mcp","kind":"claim","text":"MCP ingest materializes into the read model.","coordinates":[{"dimension":"conversation","scope_id":"conversation:mcp","sequence":2,"occurred_at":"2026-05-04T10:00:00Z","valid_from":"2026-05-04T10:00:00Z"}]}"#.to_string(),
336                reason: "test entry".to_string(),
337                scopes: vec!["memory".to_string()],
338            },
339            UpdateContextChange {
340                operation: "UPSERT".to_string(),
341                entity_kind: "memory_relation".to_string(),
342                entity_id: "relation:mcp".to_string(),
343                payload_json: r#"{"from":"claim:old","to":"claim:mcp","rel":"supersedes","class":"evidential","why":"The new memory updates the previous claim.","confidence":"high"}"#.to_string(),
344                reason: "test relation".to_string(),
345                scopes: vec!["memory".to_string()],
346            },
347            UpdateContextChange {
348                operation: "UPSERT".to_string(),
349                entity_kind: "memory_evidence".to_string(),
350                entity_id: "evidence:mcp".to_string(),
351                payload_json: r#"{"id":"evidence:mcp","supports":["claim:mcp"],"text":"Projection writer recorded the detail.","source":"unit-test"}"#.to_string(),
352                reason: "test evidence".to_string(),
353                scopes: vec!["memory".to_string()],
354            },
355        ]
356    }
357
358    #[tokio::test]
359    async fn execute_appends_event_and_returns_revision_one() {
360        let store = event_store();
361        let uc = use_case(Arc::clone(&store));
362
363        let outcome = uc
364            .execute(UpdateContextCommand {
365                root_node_id: "node-1".to_string(),
366                role: "developer".to_string(),
367                work_item_id: "task-1".to_string(),
368                changes: vec![sample_change()],
369                expected_revision: None,
370                expected_content_hash: None,
371                idempotency_key: None,
372                logical_digest: None,
373                requested_by: None,
374            })
375            .await
376            .expect("should succeed");
377
378        assert_eq!(outcome.accepted_version.revision, 1);
379        assert!(!outcome.accepted_version.content_hash.is_empty());
380        assert!(outcome.warnings.is_empty());
381    }
382
383    #[tokio::test]
384    async fn execute_rejects_wrong_expected_revision() {
385        let store = event_store();
386        let uc = use_case(Arc::clone(&store));
387
388        let err = uc
389            .execute(UpdateContextCommand {
390                root_node_id: "node-1".to_string(),
391                role: "developer".to_string(),
392                work_item_id: "task-1".to_string(),
393                changes: vec![sample_change()],
394                expected_revision: Some(99),
395                expected_content_hash: None,
396                idempotency_key: None,
397                logical_digest: None,
398                requested_by: None,
399            })
400            .await;
401
402        let error = err.expect_err("should fail");
403        assert!(matches!(&error, ApplicationError::RetryableConflict(_)));
404        let msg = error.to_string();
405        assert!(msg.contains("expected revision 99"));
406    }
407
408    #[tokio::test]
409    async fn execute_returns_cached_outcome_for_duplicate_idempotency_key() {
410        let store = event_store();
411        let uc = use_case(Arc::clone(&store));
412
413        let first = uc
414            .execute(UpdateContextCommand {
415                root_node_id: "node-1".to_string(),
416                role: "developer".to_string(),
417                work_item_id: "task-1".to_string(),
418                changes: vec![sample_change()],
419                expected_revision: None,
420                expected_content_hash: None,
421                idempotency_key: Some("idem-1".to_string()),
422                logical_digest: None,
423                requested_by: None,
424            })
425            .await
426            .expect("first call should succeed");
427
428        let second = uc
429            .execute(UpdateContextCommand {
430                root_node_id: "node-1".to_string(),
431                role: "developer".to_string(),
432                work_item_id: "task-1".to_string(),
433                changes: vec![sample_change()],
434                expected_revision: None,
435                expected_content_hash: None,
436                idempotency_key: Some("idem-1".to_string()),
437                logical_digest: None,
438                requested_by: None,
439            })
440            .await
441            .expect("second call should return cached outcome");
442
443        assert_eq!(
444            first.accepted_version.revision,
445            second.accepted_version.revision
446        );
447        assert_eq!(
448            first.accepted_version.content_hash,
449            second.accepted_version.content_hash
450        );
451    }
452
453    #[tokio::test]
454    async fn execute_rejects_duplicate_idempotency_key_with_different_content() {
455        let store = event_store();
456        let uc = use_case(Arc::clone(&store));
457
458        uc.execute(UpdateContextCommand {
459            root_node_id: "node-1".to_string(),
460            role: "developer".to_string(),
461            work_item_id: "task-1".to_string(),
462            changes: vec![sample_change()],
463            expected_revision: None,
464            expected_content_hash: None,
465            idempotency_key: Some("idem-1".to_string()),
466            logical_digest: None,
467            requested_by: None,
468        })
469        .await
470        .expect("first call should succeed");
471
472        let mut changed = sample_change();
473        changed.payload_json = r#"{"status":"CHANGED"}"#.to_string();
474        let error = uc
475            .execute(UpdateContextCommand {
476                root_node_id: "node-1".to_string(),
477                role: "developer".to_string(),
478                work_item_id: "task-1".to_string(),
479                changes: vec![changed],
480                expected_revision: None,
481                expected_content_hash: None,
482                idempotency_key: Some("idem-1".to_string()),
483                logical_digest: None,
484                requested_by: None,
485            })
486            .await
487            .expect_err("duplicate idempotency key with changed payload should fail");
488
489        assert!(error.to_string().contains("different content"));
490    }
491
492    #[tokio::test]
493    async fn execute_warns_on_empty_changes() {
494        let store = event_store();
495        let uc = use_case(Arc::clone(&store));
496
497        let outcome = uc
498            .execute(UpdateContextCommand {
499                root_node_id: "node-1".to_string(),
500                role: "developer".to_string(),
501                work_item_id: "task-1".to_string(),
502                changes: vec![],
503                expected_revision: None,
504                expected_content_hash: None,
505                idempotency_key: None,
506                logical_digest: None,
507                requested_by: None,
508            })
509            .await
510            .expect("empty changes should succeed with warning");
511
512        assert_eq!(outcome.warnings.len(), 1);
513        assert!(outcome.warnings[0].contains("no changes supplied"));
514    }
515
516    #[tokio::test]
517    async fn execute_increments_revision_on_sequential_calls() {
518        let store = event_store();
519        let uc = use_case(Arc::clone(&store));
520
521        let first = uc
522            .execute(UpdateContextCommand {
523                root_node_id: "node-1".to_string(),
524                role: "developer".to_string(),
525                work_item_id: "task-1".to_string(),
526                changes: vec![sample_change()],
527                expected_revision: None,
528                expected_content_hash: None,
529                idempotency_key: None,
530                logical_digest: None,
531                requested_by: None,
532            })
533            .await
534            .expect("first should succeed");
535
536        let second = uc
537            .execute(UpdateContextCommand {
538                root_node_id: "node-1".to_string(),
539                role: "developer".to_string(),
540                work_item_id: "task-2".to_string(),
541                changes: vec![sample_change()],
542                expected_revision: Some(1),
543                expected_content_hash: None,
544                idempotency_key: None,
545                logical_digest: None,
546                requested_by: None,
547            })
548            .await
549            .expect("second should succeed");
550
551        assert_eq!(first.accepted_version.revision, 1);
552        assert_eq!(second.accepted_version.revision, 2);
553    }
554
555    #[tokio::test]
556    async fn execute_rejects_wrong_content_hash_precondition() {
557        let store = event_store();
558        let uc = use_case(Arc::clone(&store));
559
560        // First command establishes a hash
561        let first = uc
562            .execute(UpdateContextCommand {
563                root_node_id: "node-1".to_string(),
564                role: "developer".to_string(),
565                work_item_id: "task-1".to_string(),
566                changes: vec![sample_change()],
567                expected_revision: None,
568                expected_content_hash: None,
569                idempotency_key: None,
570                logical_digest: None,
571                requested_by: None,
572            })
573            .await
574            .expect("first should succeed");
575
576        // Second command with wrong expected_content_hash must fail
577        let err = uc
578            .execute(UpdateContextCommand {
579                root_node_id: "node-1".to_string(),
580                role: "developer".to_string(),
581                work_item_id: "task-2".to_string(),
582                changes: vec![sample_change()],
583                expected_revision: Some(1),
584                expected_content_hash: Some("wrong-hash".to_string()),
585                idempotency_key: None,
586                logical_digest: None,
587                requested_by: None,
588            })
589            .await;
590
591        let error = err.expect_err("should fail");
592        assert!(matches!(&error, ApplicationError::RetryableConflict(_)));
593        let msg = error.to_string();
594        assert!(msg.contains("expected content hash"));
595
596        // Same command with correct hash succeeds
597        let ok = uc
598            .execute(UpdateContextCommand {
599                root_node_id: "node-1".to_string(),
600                role: "developer".to_string(),
601                work_item_id: "task-2".to_string(),
602                changes: vec![sample_change()],
603                expected_revision: Some(1),
604                expected_content_hash: Some(first.accepted_version.content_hash.clone()),
605                idempotency_key: None,
606                logical_digest: None,
607                requested_by: None,
608            })
609            .await
610            .expect("correct hash should succeed");
611
612        assert_eq!(ok.accepted_version.revision, 2);
613    }
614
615    #[tokio::test]
616    async fn execute_projects_memory_changes_into_read_model_mutations() {
617        let store = event_store();
618        let writer = InMemoryProjectionWriter::default();
619        let uc = UpdateContextUseCase::new_with_projection_writer(
620            Arc::clone(&store),
621            writer.clone(),
622            "0.1.0",
623        );
624
625        uc.execute(UpdateContextCommand {
626            root_node_id: "question:mcp".to_string(),
627            role: "memory".to_string(),
628            work_item_id: "ingest:mcp:1".to_string(),
629            changes: memory_changes(),
630            expected_revision: None,
631            expected_content_hash: None,
632            idempotency_key: Some("ingest:mcp:1".to_string()),
633            logical_digest: None,
634            requested_by: Some("mcp-test".to_string()),
635        })
636        .await
637        .expect("memory ingest should succeed");
638
639        let mutations = writer.mutations().await;
640        assert!(mutations.iter().any(|mutation| matches!(
641            mutation,
642            ProjectionMutation::UpsertNode(node)
643                if node.node_id == "claim:mcp"
644                    && node.node_kind == "claim"
645                    && node.summary == "MCP ingest materializes into the read model."
646        )));
647        assert!(mutations.iter().any(|mutation| matches!(
648            mutation,
649            ProjectionMutation::UpsertNodeDetail(detail)
650                if detail.node_id == "claim:mcp"
651                    && detail.detail == "MCP ingest materializes into the read model."
652        )));
653        assert!(mutations.iter().any(|mutation| matches!(
654            mutation,
655            ProjectionMutation::UpsertNode(node)
656                if node.node_id == "evidence:mcp"
657                    && node.node_kind == "memory_evidence"
658                    && node.summary == "Projection writer recorded the detail."
659        )));
660        assert!(mutations.iter().any(|mutation| matches!(
661            mutation,
662            ProjectionMutation::UpsertNodeRelation(relation)
663                if relation.source_node_id == "question:mcp"
664                    && relation.target_node_id == "claim:mcp"
665                    && relation.relation_type == "records"
666        )));
667        assert!(mutations.iter().any(|mutation| matches!(
668            mutation,
669            ProjectionMutation::UpsertNodeRelation(relation)
670                if relation.source_node_id == "conversation:mcp"
671                    && relation.target_node_id == "claim:mcp"
672                    && relation.relation_type == "contains_entry"
673                    && relation.explanation.dimension() == Some("conversation")
674                    && relation.explanation.scope_id() == Some("conversation:mcp")
675                    && relation.explanation.sequence() == Some(2)
676                    && relation.explanation.occurred_at() == Some("2026-05-04T10:00:00Z")
677                    && relation.explanation.valid_from() == Some("2026-05-04T10:00:00Z")
678        )));
679        assert!(mutations.iter().any(|mutation| matches!(
680            mutation,
681            ProjectionMutation::UpsertNodeRelation(relation)
682                if relation.source_node_id == "evidence:mcp"
683                    && relation.target_node_id == "claim:mcp"
684                    && relation.relation_type == "supports"
685        )));
686        assert!(mutations.iter().any(|mutation| matches!(
687            mutation,
688            ProjectionMutation::UpsertNodeRelation(relation)
689                if relation.source_node_id == "claim:old"
690                    && relation.target_node_id == "claim:mcp"
691                    && relation.relation_type == "supersedes"
692        )));
693        assert!(mutations.iter().any(|mutation| matches!(
694            mutation,
695            ProjectionMutation::UpdateNodeStatus { node_id, status }
696                if node_id == "claim:mcp" && status == "SUPERSEDED"
697        )));
698    }
699
700    #[tokio::test]
701    async fn execute_does_not_reproject_duplicate_idempotency_key() {
702        let store = event_store();
703        let writer = InMemoryProjectionWriter::default();
704        let uc = UpdateContextUseCase::new_with_projection_writer(
705            Arc::clone(&store),
706            writer.clone(),
707            "0.1.0",
708        );
709
710        for _ in 0..2 {
711            uc.execute(UpdateContextCommand {
712                root_node_id: "question:mcp".to_string(),
713                role: "memory".to_string(),
714                work_item_id: "ingest:mcp:1".to_string(),
715                changes: memory_changes(),
716                expected_revision: None,
717                expected_content_hash: None,
718                idempotency_key: Some("ingest:mcp:1".to_string()),
719                logical_digest: None,
720                requested_by: Some("mcp-test".to_string()),
721            })
722            .await
723            .expect("idempotent memory ingest should succeed");
724        }
725
726        let mutations = writer.mutations().await;
727        let claim_nodes = mutations
728            .iter()
729            .filter(|mutation| {
730                matches!(
731                    mutation,
732                    ProjectionMutation::UpsertNode(node) if node.node_id == "claim:mcp"
733                )
734            })
735            .count();
736        assert_eq!(claim_nodes, 1);
737    }
738}
kmp_application/commands/update_context.rs

kmp_application/commands/
update_context.rs
