| 1 | //! The only write path for the work graph. |
| 2 | //! |
| 3 | //! [`apply`] is pure and deterministic: no clock reads, no RNG, no I/O — the |
| 4 | //! caller-supplied [`ChangeCtx`] carries timestamps and session identity, and |
| 5 | //! every derived ID comes from SHA-256 over `(session_id, discriminator)`. |
| 6 | //! The same snapshot + change + ctx always produce byte-identical results. |
| 7 | //! |
| 8 | //! Contract per change: |
| 9 | //! 1. idempotency-key duplicates are acknowledged as no-op receipts; |
| 10 | //! 2. the change is applied to a copy; |
| 11 | //! 3. the candidate is validated fail-closed — on any violation the input |
| 12 | //! snapshot is untouched and the caller gets the full report; |
| 13 | //! 4. the revision increments exactly once; |
| 14 | //! 5. the receipt is pushed onto the bounded history. |
| 15 | //! |
| 16 | //! Callers on `Ok`: derive compat snapshots → validate combined → persist → |
| 17 | //! publish projections (publish AFTER the persist enqueue, never before). |
| 18 | //! On `Err`: surface the report; no state changed. |
| 19 | |
| 20 | use super::events::{ |
| 21 | CancelOutcome, ChangeCtx, ChangeReceipt, ObservationSummary, OperationObservation, OwnerState, |
| 22 | WorkGraphChange, WorkGraphProposal, WorkNodePatch, |
| 23 | }; |
| 24 | use super::ids::{ChangeId, WorkEdgeId, WorkNodeId}; |
| 25 | use super::model::{ |
| 26 | EdgeKind, EvidenceRef, NodeKind, NodeState, OperationBinding, Provenance, WorkActivityEvent, |
| 27 | WorkEdge, WorkGraphSnapshot, WorkNode, |
| 28 | }; |
| 29 | use super::validate::{ValidationCode, ValidationReport, validate}; |
| 30 | |
| 31 | /// Apply one change, producing the next snapshot and a receipt, or a |
| 32 | /// validation report with the input snapshot untouched. |
| 33 | pub fn apply( |
| 34 | g: &WorkGraphSnapshot, |
| 35 | change: WorkGraphChange, |
| 36 | ctx: ChangeCtx, |
| 37 | ) -> Result<(WorkGraphSnapshot, ChangeReceipt), ValidationReport> { |
| 38 | if let Some(key) = &ctx.idempotency_key |
| 39 | && g.seen_keys.contains(key) |
| 40 | { |
| 41 | // Duplicate runtime event: acknowledge without effect. |
| 42 | let receipt = ChangeReceipt { |
| 43 | change_id: ChangeId::derive( |
| 44 | &ctx.session_id, |
| 45 | &format!("noop:{}:{}", key.binding.as_str(), key.seq), |
| 46 | ), |
| 47 | revision: g.revision, |
| 48 | summary: format!("{} (duplicate)", change.kind_name()), |
| 49 | applied_at: ctx.now, |
| 50 | idempotency_key: Some(key.clone()), |
| 51 | no_op: true, |
| 52 | }; |
| 53 | return Ok((g.clone(), receipt)); |
| 54 | } |
| 55 | |
| 56 | let mut next = apply_pure(g, &change, &ctx)?; |
| 57 | validate(&next)?; // fail closed: `g` unchanged on Err |
| 58 | next.revision = g.revision + 1; // exactly once |
| 59 | let receipt = ChangeReceipt::of(&change, next.revision, &ctx); |
| 60 | next.history.push_bounded(receipt.clone()); |
| 61 | if let Some(key) = ctx.idempotency_key { |
| 62 | next.seen_keys.insert(key); |
| 63 | } |
| 64 | Ok((next, receipt)) |
| 65 | } |
| 66 | |
| 67 | fn structural(message: impl Into<String>) -> ValidationReport { |
| 68 | ValidationReport::single(ValidationCode::Structural, message) |
| 69 | } |
| 70 | |
| 71 | fn apply_pure( |
| 72 | g: &WorkGraphSnapshot, |
| 73 | change: &WorkGraphChange, |
| 74 | ctx: &ChangeCtx, |
| 75 | ) -> Result<WorkGraphSnapshot, ValidationReport> { |
| 76 | let mut next = g.clone(); |
| 77 | match change { |
| 78 | WorkGraphChange::AddNode { node } => { |
| 79 | add_node(&mut next, node.clone())?; |
| 80 | } |
| 81 | WorkGraphChange::UpdateNode { id, patch } => { |
| 82 | patch_node(&mut next, id, patch, ctx.now)?; |
| 83 | } |
| 84 | WorkGraphChange::AddEdge { edge } => { |
| 85 | add_edge(&mut next, edge.clone())?; |
| 86 | } |
| 87 | WorkGraphChange::RemoveEdge { id } => { |
| 88 | let before = next.edges.len(); |
| 89 | next.edges.retain(|e| &e.id != id); |
| 90 | if next.edges.len() == before { |
| 91 | return Err(structural(format!("edge {id} not found"))); |
| 92 | } |
| 93 | } |
| 94 | WorkGraphChange::BindOperation { node, binding } => { |
| 95 | let now = ctx.now; |
| 96 | let target = next |
| 97 | .node_mut(node) |
| 98 | .ok_or_else(|| structural(format!("node {node} not found")))?; |
| 99 | target.binding = Some(binding.clone()); |
| 100 | target.updated_at = now; |
| 101 | } |
| 102 | WorkGraphChange::ReconcileOperation { node, obs } => { |
| 103 | reconcile(&mut next, node, obs, ctx)?; |
| 104 | } |
| 105 | WorkGraphChange::AttachEvidence { node, evidence } => { |
| 106 | attach_evidence(&mut next, node, evidence, ctx)?; |
| 107 | } |
| 108 | WorkGraphChange::ProposePlanDiff { proposal } => { |
| 109 | if next.proposals.iter().any(|p| p.id == proposal.id) { |
| 110 | return Err(structural(format!("duplicate proposal {}", proposal.id))); |
| 111 | } |
| 112 | validate_proposal(&next, proposal, ctx)?; |
| 113 | next.proposals.push(proposal.clone()); |
| 114 | } |
| 115 | WorkGraphChange::WithdrawPlanDiff { proposal_id } => { |
| 116 | let before = next.proposals.len(); |
| 117 | next.proposals |
| 118 | .retain(|proposal| proposal.id != *proposal_id); |
| 119 | if next.proposals.len() == before { |
| 120 | return Err(structural(format!("proposal {proposal_id} not found"))); |
| 121 | } |
| 122 | } |
| 123 | WorkGraphChange::AcceptPlanDiff { |
| 124 | proposal_id, |
| 125 | approval, |
| 126 | } => { |
| 127 | let index = next |
| 128 | .proposals |
| 129 | .iter() |
| 130 | .position(|p| p.id == *proposal_id) |
| 131 | .ok_or_else(|| structural(format!("proposal {proposal_id} not found")))?; |
| 132 | let proposal = next.proposals.remove(index); |
| 133 | apply_proposal(&mut next, &proposal, ctx)?; |
| 134 | // Record the acceptance as an Approval node so the review action |
| 135 | // itself is part of the graph's history. |
| 136 | let approval_node = WorkNode { |
| 137 | id: WorkNodeId::derive( |
| 138 | &ctx.session_id, |
| 139 | &format!("approval:{}", proposal_id.as_str()), |
| 140 | ), |
| 141 | kind: NodeKind::Approval, |
| 142 | title: format!("plan diff approved: {}", approval.reference), |
| 143 | state: NodeState::Completed, |
| 144 | acceptance: Vec::new(), |
| 145 | binding: None, |
| 146 | evidence: None, |
| 147 | provenance: Provenance::UserEdit { |
| 148 | proposal_id: proposal_id.clone(), |
| 149 | }, |
| 150 | created_at: ctx.now, |
| 151 | updated_at: ctx.now, |
| 152 | }; |
| 153 | add_node(&mut next, approval_node)?; |
| 154 | } |
| 155 | WorkGraphChange::Supersede { old, replacement } => { |
| 156 | if old == replacement { |
| 157 | return Err(structural("node cannot supersede itself")); |
| 158 | } |
| 159 | if next.node(replacement).is_none() { |
| 160 | return Err(structural(format!( |
| 161 | "replacement node {replacement} not found" |
| 162 | ))); |
| 163 | } |
| 164 | let now = ctx.now; |
| 165 | { |
| 166 | let old_node = next |
| 167 | .node_mut(old) |
| 168 | .ok_or_else(|| structural(format!("node {old} not found")))?; |
| 169 | // Explicit supersede is the sanctioned way past V9's |
| 170 | // terminal-state protection. |
| 171 | old_node.state = NodeState::Superseded; |
| 172 | old_node.updated_at = now; |
| 173 | } |
| 174 | let edge = WorkEdge { |
| 175 | id: WorkEdgeId::derive( |
| 176 | &ctx.session_id, |
| 177 | &format!("supersedes:{}:{}", replacement.as_str(), old.as_str()), |
| 178 | ), |
| 179 | kind: EdgeKind::Supersedes, |
| 180 | from: replacement.clone(), |
| 181 | to: old.clone(), |
| 182 | }; |
| 183 | add_edge(&mut next, edge)?; |
| 184 | } |
| 185 | WorkGraphChange::ReplaceCompatProjection { compat } => { |
| 186 | next.compat = compat.clone(); |
| 187 | } |
| 188 | WorkGraphChange::SetImportDigest { digest } => { |
| 189 | if digest.is_empty() { |
| 190 | return Err(structural("legacy import digest cannot be empty")); |
| 191 | } |
| 192 | next.import_digest = Some(digest.clone()); |
| 193 | } |
| 194 | WorkGraphChange::RecordActivity { event } => { |
| 195 | let operation = match event { |
| 196 | WorkActivityEvent::ReasoningEffortChanged { operation, .. } => operation, |
| 197 | }; |
| 198 | if let Some(operation) = operation { |
| 199 | let node = next.node(operation).ok_or_else(|| { |
| 200 | structural(format!("activity references missing operation {operation}")) |
| 201 | })?; |
| 202 | if node.kind != NodeKind::Operation || !node.state.is_live() { |
| 203 | return Err(structural(format!( |
| 204 | "activity operation {operation} is not live" |
| 205 | ))); |
| 206 | } |
| 207 | } |
| 208 | next.activities.push_bounded(event.clone()); |
| 209 | } |
| 210 | } |
| 211 | Ok(next) |
| 212 | } |
| 213 | |
| 214 | fn add_node(next: &mut WorkGraphSnapshot, node: WorkNode) -> Result<(), ValidationReport> { |
| 215 | if next.node(&node.id).is_some() { |
| 216 | return Err(structural(format!("duplicate node {}", node.id))); |
| 217 | } |
| 218 | next.nodes.push(node); |
| 219 | Ok(()) |
| 220 | } |
| 221 | |
| 222 | fn add_edge(next: &mut WorkGraphSnapshot, edge: WorkEdge) -> Result<(), ValidationReport> { |
| 223 | if next.edge(&edge.id).is_some() { |
| 224 | return Err(structural(format!("duplicate edge {}", edge.id))); |
| 225 | } |
| 226 | for endpoint in [&edge.from, &edge.to] { |
| 227 | if next.node(endpoint).is_none() { |
| 228 | return Err(structural(format!( |
| 229 | "edge {} references missing node {endpoint}", |
| 230 | edge.id |
| 231 | ))); |
| 232 | } |
| 233 | } |
| 234 | next.edges.push(edge); |
| 235 | Ok(()) |
| 236 | } |
| 237 | |
| 238 | /// V9 at the write path: terminal states are never overwritten by patches; |
| 239 | /// only the explicit `Supersede` change (or a reconcile-rule change) may move |
| 240 | /// a node out of a terminal state. |
| 241 | fn patch_node( |
| 242 | next: &mut WorkGraphSnapshot, |
| 243 | id: &WorkNodeId, |
| 244 | patch: &WorkNodePatch, |
| 245 | now: i64, |
| 246 | ) -> Result<(), ValidationReport> { |
| 247 | let node = next |
| 248 | .node_mut(id) |
| 249 | .ok_or_else(|| structural(format!("node {id} not found")))?; |
| 250 | // Only an actual transition OUT of a terminal state is forbidden. |
| 251 | // Re-asserting the state a node already holds is a no-op, and rejecting it |
| 252 | // broke the only tool that writes here: `work_update` replaces the whole |
| 253 | // todo list on every call, so once an item is cancelled every later call |
| 254 | // re-sends it as cancelled and the entire update was refused. The model was |
| 255 | // then told to "use Supersede", which `work_update` does not expose — a |
| 256 | // dead end whose only escape was silently dropping the item from the list. |
| 257 | if node.state.is_terminal() && patch.state.is_some_and(|s| s != node.state) { |
| 258 | return Err(ValidationReport::single( |
| 259 | ValidationCode::V9, |
| 260 | format!( |
| 261 | "node {id} is terminal ({:?}) and cannot move to {:?}. \ |
| 262 | Re-sending its current state is fine; changing it needs Supersede.", |
| 263 | node.state, |
| 264 | patch.state.expect("checked above") |
| 265 | ), |
| 266 | )); |
| 267 | } |
| 268 | if let Some(title) = &patch.title { |
| 269 | node.title = title.clone(); |
| 270 | } |
| 271 | if let Some(state) = patch.state { |
| 272 | node.state = state; |
| 273 | } |
| 274 | if let Some(acceptance) = &patch.acceptance { |
| 275 | node.acceptance = acceptance.clone(); |
| 276 | } |
| 277 | if let Some(provenance) = &patch.provenance { |
| 278 | node.provenance = provenance.clone(); |
| 279 | } |
| 280 | node.updated_at = now; |
| 281 | Ok(()) |
| 282 | } |
| 283 | |
| 284 | /// Pure application of an owner observation. The owner is authoritative for |
| 285 | /// lifecycle; the graph never invents liveness: |
| 286 | /// - a missing owner maps to `Stale` — NEVER to Active or Completed; |
| 287 | /// - a terminal owner lifecycle maps to `Completed` — never `Verified` |
| 288 | /// (verification only ever comes from evidence, V4); |
| 289 | /// - nodes already in a terminal state keep it (V9); only the observation |
| 290 | /// summary is updated. |
| 291 | fn reconcile( |
| 292 | next: &mut WorkGraphSnapshot, |
| 293 | id: &WorkNodeId, |
| 294 | obs: &OperationObservation, |
| 295 | ctx: &ChangeCtx, |
| 296 | ) -> Result<(), ValidationReport> { |
| 297 | let now = ctx.now; |
| 298 | let node = next |
| 299 | .node_mut(id) |
| 300 | .ok_or_else(|| structural(format!("node {id} not found")))?; |
| 301 | let binding = node |
| 302 | .binding |
| 303 | .as_mut() |
| 304 | .ok_or_else(|| structural(format!("node {id} has no operation binding")))?; |
| 305 | |
| 306 | let new_state = match obs { |
| 307 | OperationObservation::OwnerReported { |
| 308 | state, |
| 309 | seq, |
| 310 | at, |
| 311 | output, |
| 312 | } => { |
| 313 | binding.last_observation = Some(ObservationSummary { |
| 314 | owner_state: *state, |
| 315 | seq: *seq, |
| 316 | observed_at: *at, |
| 317 | output: output.clone(), |
| 318 | }); |
| 319 | Some(match state { |
| 320 | OwnerState::Initializing => NodeState::Initializing, |
| 321 | OwnerState::Running => NodeState::Active, |
| 322 | OwnerState::Waiting => NodeState::Waiting, |
| 323 | OwnerState::Completed => NodeState::Completed, |
| 324 | OwnerState::Failed => NodeState::Failed, |
| 325 | OwnerState::Cancelled => NodeState::Cancelled, |
| 326 | }) |
| 327 | } |
| 328 | OperationObservation::OwnerMissing { .. } => Some(NodeState::Stale), |
| 329 | OperationObservation::CancelUpdate { outcome, .. } => match outcome { |
| 330 | // In-flight acknowledgements: record only, no state claim yet. |
| 331 | CancelOutcome::Requested |
| 332 | | CancelOutcome::Acknowledged |
| 333 | | CancelOutcome::AlreadyFinished => None, |
| 334 | CancelOutcome::Forced => Some(NodeState::Cancelled), |
| 335 | CancelOutcome::NotFound | CancelOutcome::StaleUnknown => Some(NodeState::Stale), |
| 336 | }, |
| 337 | }; |
| 338 | if let Some(state) = new_state |
| 339 | && !node.state.is_terminal() |
| 340 | { |
| 341 | node.state = state; |
| 342 | } |
| 343 | node.updated_at = now; |
| 344 | Ok(()) |
| 345 | } |
| 346 | |
| 347 | /// Materialize evidence as an Evidence node plus a `Verifies` edge onto the |
| 348 | /// target, both with deterministically derived IDs (so the same evidence |
| 349 | /// reference attaches at most once — a repeat is a structural rejection, not |
| 350 | /// a duplicate node). |
| 351 | fn attach_evidence( |
| 352 | next: &mut WorkGraphSnapshot, |
| 353 | target: &WorkNodeId, |
| 354 | evidence: &EvidenceRef, |
| 355 | ctx: &ChangeCtx, |
| 356 | ) -> Result<(), ValidationReport> { |
| 357 | if next.node(target).is_none() { |
| 358 | return Err(structural(format!("node {target} not found"))); |
| 359 | } |
| 360 | let evidence_id = WorkNodeId::derive( |
| 361 | &ctx.session_id, |
| 362 | &format!("evidence:{}:{}", target.as_str(), evidence.reference()), |
| 363 | ); |
| 364 | let node = WorkNode { |
| 365 | id: evidence_id.clone(), |
| 366 | kind: NodeKind::Evidence, |
| 367 | title: format!("evidence: {}", evidence.reference()), |
| 368 | state: NodeState::Completed, |
| 369 | acceptance: Vec::new(), |
| 370 | binding: None, |
| 371 | evidence: Some(evidence.clone()), |
| 372 | provenance: Provenance::RuntimeReconcile { |
| 373 | source: "attach_evidence".to_string(), |
| 374 | observed_at: ctx.now, |
| 375 | }, |
| 376 | created_at: ctx.now, |
| 377 | updated_at: ctx.now, |
| 378 | }; |
| 379 | add_node(next, node)?; |
| 380 | let edge = WorkEdge { |
| 381 | id: WorkEdgeId::derive( |
| 382 | &ctx.session_id, |
| 383 | &format!("verifies:{}:{}", evidence_id.as_str(), target.as_str()), |
| 384 | ), |
| 385 | kind: EdgeKind::Verifies, |
| 386 | from: evidence_id, |
| 387 | to: target.clone(), |
| 388 | }; |
| 389 | add_edge(next, edge) |
| 390 | } |
| 391 | |
| 392 | /// Apply an accepted proposal atomically: nodes first (so added edges may |
| 393 | /// reference them), then edges, then patches, then removals. Any failure |
| 394 | /// rejects the whole acceptance (the caller's snapshot stays untouched). |
| 395 | fn apply_proposal( |
| 396 | next: &mut WorkGraphSnapshot, |
| 397 | proposal: &WorkGraphProposal, |
| 398 | ctx: &ChangeCtx, |
| 399 | ) -> Result<(), ValidationReport> { |
| 400 | for node in &proposal.added_nodes { |
| 401 | add_node(next, node.clone())?; |
| 402 | } |
| 403 | for edge in &proposal.added_edges { |
| 404 | add_edge(next, edge.clone())?; |
| 405 | } |
| 406 | for update in &proposal.updated_nodes { |
| 407 | patch_node(next, &update.id, &update.patch, ctx.now)?; |
| 408 | } |
| 409 | for edge_id in &proposal.removed_edges { |
| 410 | let before = next.edges.len(); |
| 411 | next.edges.retain(|e| &e.id != edge_id); |
| 412 | if next.edges.len() == before { |
| 413 | return Err(structural(format!("edge {edge_id} not found"))); |
| 414 | } |
| 415 | } |
| 416 | for node_id in &proposal.removed_nodes { |
| 417 | if next |
| 418 | .edges |
| 419 | .iter() |
| 420 | .any(|edge| edge.from == *node_id || edge.to == *node_id) |
| 421 | { |
| 422 | return Err(structural(format!( |
| 423 | "node {node_id} still has edges after proposed removals" |
| 424 | ))); |
| 425 | } |
| 426 | let before = next.nodes.len(); |
| 427 | next.nodes.retain(|node| node.id != *node_id); |
| 428 | if next.nodes.len() == before { |
| 429 | return Err(structural(format!("node {node_id} not found"))); |
| 430 | } |
| 431 | } |
| 432 | if let Some(compat) = &proposal.replacement_compat { |
| 433 | next.compat.clone_from(compat); |
| 434 | } |
| 435 | Ok(()) |
| 436 | } |
| 437 | |
| 438 | /// Reject malformed or invariant-breaking plan edits before they become |
| 439 | /// user-reviewable. Acceptance reruns the same atomic application and the |
| 440 | /// outer reducer validation, so reviewed and accepted semantics cannot drift. |
| 441 | fn validate_proposal( |
| 442 | current: &WorkGraphSnapshot, |
| 443 | proposal: &WorkGraphProposal, |
| 444 | ctx: &ChangeCtx, |
| 445 | ) -> Result<(), ValidationReport> { |
| 446 | preview_plan_diff(current, proposal, ctx).map(|_| ()) |
| 447 | } |
| 448 | |
| 449 | pub(super) fn preview_plan_diff( |
| 450 | current: &WorkGraphSnapshot, |
| 451 | proposal: &WorkGraphProposal, |
| 452 | ctx: &ChangeCtx, |
| 453 | ) -> Result<WorkGraphSnapshot, ValidationReport> { |
| 454 | let mut candidate = current.clone(); |
| 455 | apply_proposal(&mut candidate, proposal, ctx)?; |
| 456 | validate(&candidate)?; |
| 457 | Ok(candidate) |
| 458 | } |
| 459 |