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

kmp_application/commands/
update_context.rs
