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

kmp_application/commands/
update_context.rs
