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

kmp_application/commands/
update_context.rs
