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::RetryableConflict(format!(
146                "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::RetryableConflict(format!(
159                "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
198            .event_store
199            .append(event, current_revision)
200            .await
201            .map_err(|error| match error {
202                // The revision was read above and changed before append. No
203                // part of this event was committed, so replaying the same
204                // logical command under its existing idempotency key is safe.
205                PortError::Conflict(message) => ApplicationError::RetryableConflict(message),
206                other => ApplicationError::Ports(other),
207            })?;
208
209        let projection_mutations =
210            memory_projection_mutations(&command, new_revision, &content_hash)?;
211        if !projection_mutations.is_empty() {
212            self.projection_writer
213                .apply_mutations(projection_mutations)
214                .await?;
215        }
216
217        Ok(UpdateContextOutcome {
218            accepted_version: AcceptedVersion {
219                revision: new_revision,
220                content_hash,
221                generator_version: self.generator_version.to_string(),
222            },
223            warnings,
224        })
225    }
226}
227
228/// Deterministic SHA-256 hash of context changes for optimistic concurrency.
229/// Stable across process restarts and machines.
230fn compute_content_hash(changes: &[UpdateContextChange]) -> String {
231    use sha2::{Digest, Sha256};
232    let mut hasher = Sha256::new();
233    for change in changes {
234        hasher.update(change.operation.as_bytes());
235        hasher.update(change.entity_kind.as_bytes());
236        hasher.update(change.entity_id.as_bytes());
237        hasher.update(change.payload_json.as_bytes());
238    }
239    format!("{:064x}", hasher.finalize())
240}
241
242#[cfg(test)]
243mod tests {
244    use std::sync::Arc;
245
246    use kmp_domain::ProjectionMutation;
247    use kmp_testkit::{InMemoryContextEventStore, InMemoryProjectionWriter};
248
249    use super::*;
250
251    fn event_store() -> Arc<InMemoryContextEventStore> {
252        Arc::new(InMemoryContextEventStore::new())
253    }
254
255    fn use_case(
256        store: Arc<InMemoryContextEventStore>,
257    ) -> UpdateContextUseCase<InMemoryContextEventStore> {
258        UpdateContextUseCase::new(store, "0.1.0")
259    }
260
261    fn sample_change() -> UpdateContextChange {
262        UpdateContextChange {
263            operation: "UPDATE".to_string(),
264            entity_kind: "node_detail".to_string(),
265            entity_id: "node-1".to_string(),
266            payload_json: r#"{"status":"ACTIVE"}"#.to_string(),
267            reason: "test".to_string(),
268            scopes: vec!["graph".to_string()],
269        }
270    }
271
272    fn memory_changes() -> Vec<UpdateContextChange> {
273        vec![
274            UpdateContextChange {
275                operation: "UPSERT".to_string(),
276                entity_kind: "memory_dimension".to_string(),
277                entity_id: "conversation:mcp".to_string(),
278                payload_json: r#"{"id":"conversation:mcp","kind":"conversation","title":"MCP smoke"}"#.to_string(),
279                reason: "test dimension".to_string(),
280                scopes: vec!["memory".to_string()],
281            },
282            UpdateContextChange {
283                operation: "UPSERT".to_string(),
284                entity_kind: "memory_entry".to_string(),
285                entity_id: "claim:mcp".to_string(),
286                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(),
287                reason: "test entry".to_string(),
288                scopes: vec!["memory".to_string()],
289            },
290            UpdateContextChange {
291                operation: "UPSERT".to_string(),
292                entity_kind: "memory_relation".to_string(),
293                entity_id: "relation:mcp".to_string(),
294                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(),
295                reason: "test relation".to_string(),
296                scopes: vec!["memory".to_string()],
297            },
298            UpdateContextChange {
299                operation: "UPSERT".to_string(),
300                entity_kind: "memory_evidence".to_string(),
301                entity_id: "evidence:mcp".to_string(),
302                payload_json: r#"{"id":"evidence:mcp","supports":["claim:mcp"],"text":"Projection writer recorded the detail.","source":"unit-test"}"#.to_string(),
303                reason: "test evidence".to_string(),
304                scopes: vec!["memory".to_string()],
305            },
306        ]
307    }
308
309    #[tokio::test]
310    async fn execute_appends_event_and_returns_revision_one() {
311        let store = event_store();
312        let uc = use_case(Arc::clone(&store));
313
314        let outcome = uc
315            .execute(UpdateContextCommand {
316                root_node_id: "node-1".to_string(),
317                role: "developer".to_string(),
318                work_item_id: "task-1".to_string(),
319                changes: vec![sample_change()],
320                expected_revision: None,
321                expected_content_hash: None,
322                idempotency_key: None,
323                logical_digest: None,
324                requested_by: None,
325            })
326            .await
327            .expect("should succeed");
328
329        assert_eq!(outcome.accepted_version.revision, 1);
330        assert!(!outcome.accepted_version.content_hash.is_empty());
331        assert!(outcome.warnings.is_empty());
332    }
333
334    #[tokio::test]
335    async fn execute_rejects_wrong_expected_revision() {
336        let store = event_store();
337        let uc = use_case(Arc::clone(&store));
338
339        let err = uc
340            .execute(UpdateContextCommand {
341                root_node_id: "node-1".to_string(),
342                role: "developer".to_string(),
343                work_item_id: "task-1".to_string(),
344                changes: vec![sample_change()],
345                expected_revision: Some(99),
346                expected_content_hash: None,
347                idempotency_key: None,
348                logical_digest: None,
349                requested_by: None,
350            })
351            .await;
352
353        let error = err.expect_err("should fail");
354        assert!(matches!(&error, ApplicationError::RetryableConflict(_)));
355        let msg = error.to_string();
356        assert!(msg.contains("expected revision 99"));
357    }
358
359    #[tokio::test]
360    async fn execute_returns_cached_outcome_for_duplicate_idempotency_key() {
361        let store = event_store();
362        let uc = use_case(Arc::clone(&store));
363
364        let first = uc
365            .execute(UpdateContextCommand {
366                root_node_id: "node-1".to_string(),
367                role: "developer".to_string(),
368                work_item_id: "task-1".to_string(),
369                changes: vec![sample_change()],
370                expected_revision: None,
371                expected_content_hash: None,
372                idempotency_key: Some("idem-1".to_string()),
373                logical_digest: None,
374                requested_by: None,
375            })
376            .await
377            .expect("first call should succeed");
378
379        let second = uc
380            .execute(UpdateContextCommand {
381                root_node_id: "node-1".to_string(),
382                role: "developer".to_string(),
383                work_item_id: "task-1".to_string(),
384                changes: vec![sample_change()],
385                expected_revision: None,
386                expected_content_hash: None,
387                idempotency_key: Some("idem-1".to_string()),
388                logical_digest: None,
389                requested_by: None,
390            })
391            .await
392            .expect("second call should return cached outcome");
393
394        assert_eq!(
395            first.accepted_version.revision,
396            second.accepted_version.revision
397        );
398        assert_eq!(
399            first.accepted_version.content_hash,
400            second.accepted_version.content_hash
401        );
402    }
403
404    #[tokio::test]
405    async fn execute_rejects_duplicate_idempotency_key_with_different_content() {
406        let store = event_store();
407        let uc = use_case(Arc::clone(&store));
408
409        uc.execute(UpdateContextCommand {
410            root_node_id: "node-1".to_string(),
411            role: "developer".to_string(),
412            work_item_id: "task-1".to_string(),
413            changes: vec![sample_change()],
414            expected_revision: None,
415            expected_content_hash: None,
416            idempotency_key: Some("idem-1".to_string()),
417            logical_digest: None,
418            requested_by: None,
419        })
420        .await
421        .expect("first call should succeed");
422
423        let mut changed = sample_change();
424        changed.payload_json = r#"{"status":"CHANGED"}"#.to_string();
425        let error = uc
426            .execute(UpdateContextCommand {
427                root_node_id: "node-1".to_string(),
428                role: "developer".to_string(),
429                work_item_id: "task-1".to_string(),
430                changes: vec![changed],
431                expected_revision: None,
432                expected_content_hash: None,
433                idempotency_key: Some("idem-1".to_string()),
434                logical_digest: None,
435                requested_by: None,
436            })
437            .await
438            .expect_err("duplicate idempotency key with changed payload should fail");
439
440        assert!(error.to_string().contains("different content"));
441    }
442
443    #[tokio::test]
444    async fn execute_warns_on_empty_changes() {
445        let store = event_store();
446        let uc = use_case(Arc::clone(&store));
447
448        let outcome = uc
449            .execute(UpdateContextCommand {
450                root_node_id: "node-1".to_string(),
451                role: "developer".to_string(),
452                work_item_id: "task-1".to_string(),
453                changes: vec![],
454                expected_revision: None,
455                expected_content_hash: None,
456                idempotency_key: None,
457                logical_digest: None,
458                requested_by: None,
459            })
460            .await
461            .expect("empty changes should succeed with warning");
462
463        assert_eq!(outcome.warnings.len(), 1);
464        assert!(outcome.warnings[0].contains("no changes supplied"));
465    }
466
467    #[tokio::test]
468    async fn execute_increments_revision_on_sequential_calls() {
469        let store = event_store();
470        let uc = use_case(Arc::clone(&store));
471
472        let first = uc
473            .execute(UpdateContextCommand {
474                root_node_id: "node-1".to_string(),
475                role: "developer".to_string(),
476                work_item_id: "task-1".to_string(),
477                changes: vec![sample_change()],
478                expected_revision: None,
479                expected_content_hash: None,
480                idempotency_key: None,
481                logical_digest: None,
482                requested_by: None,
483            })
484            .await
485            .expect("first should succeed");
486
487        let second = uc
488            .execute(UpdateContextCommand {
489                root_node_id: "node-1".to_string(),
490                role: "developer".to_string(),
491                work_item_id: "task-2".to_string(),
492                changes: vec![sample_change()],
493                expected_revision: Some(1),
494                expected_content_hash: None,
495                idempotency_key: None,
496                logical_digest: None,
497                requested_by: None,
498            })
499            .await
500            .expect("second should succeed");
501
502        assert_eq!(first.accepted_version.revision, 1);
503        assert_eq!(second.accepted_version.revision, 2);
504    }
505
506    #[tokio::test]
507    async fn execute_rejects_wrong_content_hash_precondition() {
508        let store = event_store();
509        let uc = use_case(Arc::clone(&store));
510
511        // First command establishes a hash
512        let first = uc
513            .execute(UpdateContextCommand {
514                root_node_id: "node-1".to_string(),
515                role: "developer".to_string(),
516                work_item_id: "task-1".to_string(),
517                changes: vec![sample_change()],
518                expected_revision: None,
519                expected_content_hash: None,
520                idempotency_key: None,
521                logical_digest: None,
522                requested_by: None,
523            })
524            .await
525            .expect("first should succeed");
526
527        // Second command with wrong expected_content_hash must fail
528        let err = uc
529            .execute(UpdateContextCommand {
530                root_node_id: "node-1".to_string(),
531                role: "developer".to_string(),
532                work_item_id: "task-2".to_string(),
533                changes: vec![sample_change()],
534                expected_revision: Some(1),
535                expected_content_hash: Some("wrong-hash".to_string()),
536                idempotency_key: None,
537                logical_digest: None,
538                requested_by: None,
539            })
540            .await;
541
542        let error = err.expect_err("should fail");
543        assert!(matches!(&error, ApplicationError::RetryableConflict(_)));
544        let msg = error.to_string();
545        assert!(msg.contains("expected content hash"));
546
547        // Same command with correct hash succeeds
548        let ok = uc
549            .execute(UpdateContextCommand {
550                root_node_id: "node-1".to_string(),
551                role: "developer".to_string(),
552                work_item_id: "task-2".to_string(),
553                changes: vec![sample_change()],
554                expected_revision: Some(1),
555                expected_content_hash: Some(first.accepted_version.content_hash.clone()),
556                idempotency_key: None,
557                logical_digest: None,
558                requested_by: None,
559            })
560            .await
561            .expect("correct hash should succeed");
562
563        assert_eq!(ok.accepted_version.revision, 2);
564    }
565
566    #[tokio::test]
567    async fn execute_projects_memory_changes_into_read_model_mutations() {
568        let store = event_store();
569        let writer = InMemoryProjectionWriter::default();
570        let uc = UpdateContextUseCase::new_with_projection_writer(
571            Arc::clone(&store),
572            writer.clone(),
573            "0.1.0",
574        );
575
576        uc.execute(UpdateContextCommand {
577            root_node_id: "question:mcp".to_string(),
578            role: "memory".to_string(),
579            work_item_id: "ingest:mcp:1".to_string(),
580            changes: memory_changes(),
581            expected_revision: None,
582            expected_content_hash: None,
583            idempotency_key: Some("ingest:mcp:1".to_string()),
584            logical_digest: None,
585            requested_by: Some("mcp-test".to_string()),
586        })
587        .await
588        .expect("memory ingest should succeed");
589
590        let mutations = writer.mutations().await;
591        assert!(mutations.iter().any(|mutation| matches!(
592            mutation,
593            ProjectionMutation::UpsertNode(node)
594                if node.node_id == "claim:mcp"
595                    && node.node_kind == "claim"
596                    && node.summary == "MCP ingest materializes into the read model."
597        )));
598        assert!(mutations.iter().any(|mutation| matches!(
599            mutation,
600            ProjectionMutation::UpsertNodeDetail(detail)
601                if detail.node_id == "claim:mcp"
602                    && detail.detail == "MCP ingest materializes into the read model."
603        )));
604        assert!(mutations.iter().any(|mutation| matches!(
605            mutation,
606            ProjectionMutation::UpsertNode(node)
607                if node.node_id == "evidence:mcp"
608                    && node.node_kind == "memory_evidence"
609                    && node.summary == "Projection writer recorded the detail."
610        )));
611        assert!(mutations.iter().any(|mutation| matches!(
612            mutation,
613            ProjectionMutation::UpsertNodeRelation(relation)
614                if relation.source_node_id == "question:mcp"
615                    && relation.target_node_id == "claim:mcp"
616                    && relation.relation_type == "records"
617        )));
618        assert!(mutations.iter().any(|mutation| matches!(
619            mutation,
620            ProjectionMutation::UpsertNodeRelation(relation)
621                if relation.source_node_id == "conversation:mcp"
622                    && relation.target_node_id == "claim:mcp"
623                    && relation.relation_type == "contains_entry"
624                    && relation.explanation.dimension() == Some("conversation")
625                    && relation.explanation.scope_id() == Some("conversation:mcp")
626                    && relation.explanation.sequence() == Some(2)
627                    && relation.explanation.occurred_at() == Some("2026-05-04T10:00:00Z")
628                    && relation.explanation.valid_from() == Some("2026-05-04T10:00:00Z")
629        )));
630        assert!(mutations.iter().any(|mutation| matches!(
631            mutation,
632            ProjectionMutation::UpsertNodeRelation(relation)
633                if relation.source_node_id == "evidence:mcp"
634                    && relation.target_node_id == "claim:mcp"
635                    && relation.relation_type == "supports"
636        )));
637        assert!(mutations.iter().any(|mutation| matches!(
638            mutation,
639            ProjectionMutation::UpsertNodeRelation(relation)
640                if relation.source_node_id == "claim:old"
641                    && relation.target_node_id == "claim:mcp"
642                    && relation.relation_type == "supersedes"
643        )));
644        assert!(mutations.iter().any(|mutation| matches!(
645            mutation,
646            ProjectionMutation::UpdateNodeStatus { node_id, status }
647                if node_id == "claim:mcp" && status == "SUPERSEDED"
648        )));
649    }
650
651    #[tokio::test]
652    async fn execute_does_not_reproject_duplicate_idempotency_key() {
653        let store = event_store();
654        let writer = InMemoryProjectionWriter::default();
655        let uc = UpdateContextUseCase::new_with_projection_writer(
656            Arc::clone(&store),
657            writer.clone(),
658            "0.1.0",
659        );
660
661        for _ in 0..2 {
662            uc.execute(UpdateContextCommand {
663                root_node_id: "question:mcp".to_string(),
664                role: "memory".to_string(),
665                work_item_id: "ingest:mcp:1".to_string(),
666                changes: memory_changes(),
667                expected_revision: None,
668                expected_content_hash: None,
669                idempotency_key: Some("ingest:mcp:1".to_string()),
670                logical_digest: None,
671                requested_by: Some("mcp-test".to_string()),
672            })
673            .await
674            .expect("idempotent memory ingest should succeed");
675        }
676
677        let mutations = writer.mutations().await;
678        let claim_nodes = mutations
679            .iter()
680            .filter(|mutation| {
681                matches!(
682                    mutation,
683                    ProjectionMutation::UpsertNode(node) if node.node_id == "claim:mcp"
684                )
685            })
686            .count();
687        assert_eq!(claim_nodes, 1);
688    }
689}
kmp_application/commands/update_context.rs

kmp_application/commands/
update_context.rs
