diff --git a/crates/app/ironclaw_composition/src/runtime.rs b/crates/app/ironclaw_composition/src/runtime.rs index 716ef52b044..40604bf7648 100644 --- a/crates/app/ironclaw_composition/src/runtime.rs +++ b/crates/app/ironclaw_composition/src/runtime.rs @@ -37,7 +37,8 @@ use ironclaw_assistant::{ ApprovalResolverPort, ApprovalTurnRunLocator, AuthInteractionService, DefaultApprovalInteractionService, DefaultAuthInteractionService, OutboundPreferencesProductService, PersistentApprovalGranteeResolver, - RunStateApprovalInteractionReadModel, SuggestionsProcessCommitObserver, + RunOutcomeProcessCommitObserver, RunStateApprovalInteractionReadModel, + SuggestionsProcessCommitObserver, }; use ironclaw_event_log::{ DurableAuditLog, DurableEventLog, EventError, NonBlockingEventSink, RuntimeEvent, @@ -3241,6 +3242,14 @@ pub(crate) async fn build_runtime_with_resource_governor( .map_err(|error| RebornRuntimeError::MalformedConfig { reason: format!("suggestion generation observer wiring failed: {error}"), })?; + processes + .subscribe_process_observer(Arc::new(RunOutcomeProcessCommitObserver::new( + Arc::clone(&services.notification_inbox), + Arc::clone(&thread_service), + ))) + .map_err(|error| RebornRuntimeError::MalformedConfig { + reason: format!("run outcome notification observer wiring failed: {error}"), + })?; let filesystem_skill_context_runtime = filesystem_skill_context_runtime(&services); let (skill_context_source, skill_activation_source, skill_execution_adapter) = match ( configured_skill_context_source, diff --git a/crates/app/ironclaw_composition/tests/trigger_poller_e2e.rs b/crates/app/ironclaw_composition/tests/trigger_poller_e2e.rs index daa9dc0adbd..dfb918c5425 100644 --- a/crates/app/ironclaw_composition/tests/trigger_poller_e2e.rs +++ b/crates/app/ironclaw_composition/tests/trigger_poller_e2e.rs @@ -61,7 +61,12 @@ use ironclaw_outbound::{ CommunicationPreferenceRecord, DeliveryDefaultScope, TriggeredRunDeliveryOutcomeKind, TriggeredRunDeliveryStore, }; -use ironclaw_product_contracts::surface::{ProductSurfaceCaller, ProductSurfaceInvokeRequest}; +use ironclaw_product_contracts::{ + notification_inbox::{ + NOTIFICATIONS_VIEW, ProductListNotificationsResponse, ProductNotificationKind, + }, + surface::{ProductSurfaceCaller, ProductSurfaceInvokeRequest, ProductSurfaceQueryRequest}, +}; use ironclaw_triggers::{ TRIGGER_TRUSTED_ADAPTER_INSTALLATION_ID, TRIGGER_TRUSTED_ADAPTER_KIND, TRIGGER_TRUSTED_EXTERNAL_ACTOR_NAMESPACE, TriggerDeliveryTargetId, TriggerExecutionSpec, @@ -1548,8 +1553,8 @@ where /// model (spec §8): a scheduled fire's RESULT is never pushed to a channel. /// /// due trigger -> trusted ingress -> real Reborn run -> persisted final reply -/// in the fire's own run thread -> background-run notifier -> **nothing on any -/// channel**. +/// in the fire's own run thread -> durable Inbox outcome, while the background +/// delivery notifier still sends **nothing on any channel**. /// /// The two arms are the two ways the retired routing used to pick a /// destination: QA-9B had none (it inherited the creator's default), QA-9D @@ -1601,7 +1606,7 @@ async fn scheduled_trigger_results_are_never_pushed_to_a_channel_across_restart( ) .await; - wait_for_recorded_outcome( + let default_run_id = wait_for_recorded_outcome( &repository, &delivery_store, default_target_trigger, @@ -1615,7 +1620,7 @@ async fn scheduled_trigger_results_are_never_pushed_to_a_channel_across_restart( TriggeredRunDeliveryOutcomeKind::Suppressed, ) .await; - wait_for_recorded_outcome( + let explicit_run_id = wait_for_recorded_outcome( &repository, &delivery_store, explicit_target_trigger, @@ -1682,6 +1687,71 @@ async fn scheduled_trigger_results_are_never_pushed_to_a_channel_across_restart( "the suppressible routine must perform prior tool work before its typed result" ); + let caller = ProductSurfaceCaller::new( + TenantId::new(TENANT).expect("tenant"), + UserId::new(USER).expect("user"), + Some(AgentId::new(AGENT).expect("agent")), + None, + ); + let surface = runtime.product_surface(None).expect("product surface"); + let inbox = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let page = surface + .query( + caller.clone(), + ProductSurfaceQueryRequest { + view_id: NOTIFICATIONS_VIEW.id.to_string(), + input: json!({ "limit": 10 }), + cursor: None, + limit: None, + }, + ) + .await + .expect("query notification Inbox"); + let response: ProductListNotificationsResponse = serde_json::from_value( + page.items + .into_iter() + .next() + .expect("notification response payload"), + ) + .expect("decode notification response"); + if response.notifications.len() == 2 { + break response; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("background completion notifications arrive"); + assert!( + inbox + .notifications + .iter() + .all(|notification| notification.kind == ProductNotificationKind::RunCompleted), + "only selected completed results are materialized: {:?}", + inbox.notifications + ); + let notified_runs = inbox + .notifications + .iter() + .filter_map(|notification| notification.turn_run_id.clone()) + .collect::>(); + assert_eq!( + notified_runs, + std::collections::BTreeSet::from( + [default_run_id.to_string(), explicit_run_id.to_string(),] + ), + "ordinary scheduled completions notify, while NothingToReport stays suppressed" + ); + + // The identities themselves are what dedupe on replay. + let record_count_before_restart = inbox.notifications.len(); + let identities_before_restart = inbox + .notifications + .iter() + .map(|notification| notification.id.clone()) + .collect::>(); + tokio::time::sleep(Duration::from_millis(250)).await; assert!( slack_provider.provider_messages().is_empty(), @@ -1696,7 +1766,79 @@ async fn scheduled_trigger_results_are_never_pushed_to_a_channel_across_restart( ) .await; register_delivery_targets(&restarted); - tokio::time::sleep(Duration::from_millis(300)).await; + let restarted_surface = restarted.product_surface(None).expect("restarted surface"); + let replayed = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let replayed_page = restarted_surface + .query( + caller.clone(), + ProductSurfaceQueryRequest { + view_id: NOTIFICATIONS_VIEW.id.to_string(), + input: json!({ "limit": 10 }), + cursor: None, + limit: None, + }, + ) + .await + .expect("query Inbox while observer replay settles"); + let replayed: ProductListNotificationsResponse = serde_json::from_value( + replayed_page + .items + .into_iter() + .next() + .expect("replayed notification payload"), + ) + .expect("decode replayed Inbox"); + if replayed.notifications.len() >= record_count_before_restart { + break replayed; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("observer replay reaches the pre-restart notification count"); + assert_eq!( + replayed.notifications.len(), + record_count_before_restart, + "restart/replay must not add a second record under an already-existing id" + ); + + tokio::time::sleep(Duration::from_millis(100)).await; + let settled_page = restarted_surface + .query( + caller, + ProductSurfaceQueryRequest { + view_id: NOTIFICATIONS_VIEW.id.to_string(), + input: json!({ "limit": 10 }), + cursor: None, + limit: None, + }, + ) + .await + .expect("query Inbox after observer replay settles"); + let settled: ProductListNotificationsResponse = serde_json::from_value( + settled_page + .items + .into_iter() + .next() + .expect("settled notification payload"), + ) + .expect("decode settled Inbox"); + assert_eq!( + settled.notifications.len(), + record_count_before_restart, + "restart/replay must remain at the pre-restart count after a settle window" + ); + assert_eq!( + settled + .notifications + .iter() + .map(|notification| notification.id.clone()) + .collect::>(), + identities_before_restart, + "restart/replay resolves to the same notification identities, so a replayed \ + commit is absorbed by its stable id instead of arriving twice" + ); restarted .shutdown() .await diff --git a/crates/domains/ironclaw_notifications/README.md b/crates/domains/ironclaw_notifications/README.md index a0aad11f342..7d1b74cc16c 100644 --- a/crates/domains/ironclaw_notifications/README.md +++ b/crates/domains/ironclaw_notifications/README.md @@ -10,6 +10,12 @@ publication, and the filesystem-backed store contract. attempt; those remain in `ironclaw_outbound`. Product-specific projection and read policy remain in `ironclaw_assistant`. +Run outcome records are produced by `ironclaw_assistant` from committed process +journal transitions. Successful scheduled runs require an exact durable +finalized-reply lookup; external delivery failures remain separate from the run +outcome and come from the outbound delivery caller. The domain store does not +infer either fact. + The store persists through `ScopedFilesystem`, so composition selects the real backend and provides tenant/user-rewritten mounts. Records contain metadata and typed references only; prompts, replies, tool payloads, secrets, and backend diff --git a/crates/kernel/ironclaw_processes/Cargo.toml b/crates/kernel/ironclaw_processes/Cargo.toml index 88f31b500b2..56b2c60580d 100644 --- a/crates/kernel/ironclaw_processes/Cargo.toml +++ b/crates/kernel/ironclaw_processes/Cargo.toml @@ -37,6 +37,6 @@ ironclaw_event_store = { path = "../../events/ironclaw_event_store" } libsql = { version = "0.9", default-features = false, features = ["core", "replication", "remote", "tls"] } deadpool-postgres = "0.14" tempfile = "3" -tokio = { version = "1", features = ["macros", "rt", "time"] } +tokio = { version = "1", features = ["macros", "rt", "test-util", "time"] } tokio-postgres = { version = "0.7", features = ["with-uuid-1", "with-chrono-0_4", "with-serde_json-1"] } uuid = { version = "1", features = ["v4", "serde"] } diff --git a/crates/kernel/ironclaw_processes/src/journal.rs b/crates/kernel/ironclaw_processes/src/journal.rs index 2335704dfa5..3564722f7a6 100644 --- a/crates/kernel/ironclaw_processes/src/journal.rs +++ b/crates/kernel/ironclaw_processes/src/journal.rs @@ -513,6 +513,10 @@ pub struct ProcessJournalEntry { pub struct ProcessJournalCommit { pub state: JournaledProcessSnapshot, pub kind: ProcessJournalKind, + /// Timestamp of the journal transition that produced this committed + /// state. Observers use it for replay-stable materialized records. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub occurred_at: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub sanitized_reason: Option, } @@ -1489,6 +1493,52 @@ mod tests { assert_eq!(absent.checkpoint_kind, None); } + #[test] + fn process_journal_commit_without_occurred_at_remains_readable() { + let commit = ProcessJournalCommit { + state: JournaledProcessSnapshot { + process_id: ProcessId::new(), + process_kind: ProcessKind::Internal, + scope: ResourceScope { + tenant_id: TenantId::new("tenant-legacy-commit").expect("tenant"), + user_id: UserId::new("user-legacy-commit").expect("user"), + agent_id: None, + project_id: None, + mission_id: None, + thread_id: None, + invocation_id: ironclaw_host_api::ids::InvocationId::new(), + }, + status: ProcessLifecycleStatus::Completed, + suspension: None, + checkpoint_ref: None, + checkpoint_kind: None, + input_ref: None, + failure: None, + journal_cursor: ProcessJournalCursor(1), + lease: None, + crash_reclaim_count: 0, + created_at: chrono::Utc::now(), + owner_user_id: None, + concurrency_class: None, + parent_process_id: None, + root_process_id: None, + metadata: Value::Null, + }, + kind: ProcessJournalKind::Completed, + occurred_at: Some(chrono::Utc::now()), + sanitized_reason: None, + }; + let mut legacy = serde_json::to_value(commit).expect("serialize current commit"); + legacy + .as_object_mut() + .expect("commit object") + .remove("occurred_at"); + + let decoded: ProcessJournalCommit = + serde_json::from_value(legacy).expect("legacy commit remains readable"); + assert_eq!(decoded.occurred_at, None); + } + #[test] fn opaque_refs_reject_empty_oversized_and_control_values() { assert!(ProcessCheckpointRef::new("").is_err()); diff --git a/crates/kernel/ironclaw_processes/src/journal_store/flusher.rs b/crates/kernel/ironclaw_processes/src/journal_store/flusher.rs index ec40fee908d..958fabc8d18 100644 --- a/crates/kernel/ironclaw_processes/src/journal_store/flusher.rs +++ b/crates/kernel/ironclaw_processes/src/journal_store/flusher.rs @@ -548,6 +548,7 @@ fn committed_batch(entries: Vec) -> Option entry.committed_state.map(|state| ProcessJournalCommit { state: *state, kind: entry.kind, + occurred_at: entry.occurred_at, sanitized_reason: entry.sanitized_reason, }) }) diff --git a/crates/kernel/ironclaw_processes/src/journal_store/observer.rs b/crates/kernel/ironclaw_processes/src/journal_store/observer.rs index 3359f45ecc4..9d874fe267d 100644 --- a/crates/kernel/ironclaw_processes/src/journal_store/observer.rs +++ b/crates/kernel/ironclaw_processes/src/journal_store/observer.rs @@ -16,6 +16,8 @@ use crate::{ ProcessJournalObserverRegistry, }; +const MAX_OBSERVER_REPLAY_ATTEMPTS: u32 = 8; + #[derive(Clone)] pub(super) struct RegisteredProcessObserver { pub(super) id: String, @@ -266,6 +268,7 @@ where commits.push(ProcessJournalCommit { state: state.clone(), kind: entry.kind, + occurred_at: entry.occurred_at, sanitized_reason: entry.sanitized_reason.clone(), }); } @@ -340,12 +343,25 @@ where tokio::spawn(async move { let _replay_guard = ObserverReplayGuard(Arc::clone(&observer.replay_running)); let mut delay = Duration::from_millis(250); + let mut attempt = 0_u32; loop { + attempt += 1; match store.replay_durable_observer_once(&observer, None).await { Ok(()) => break, + Err(error) if attempt >= MAX_OBSERVER_REPLAY_ATTEMPTS => { + tracing::error!( + observer_id = %observer.id, + attempts = attempt, + %error, + "durable process observer replay exhausted its retry budget; \ + cursor remains unacknowledged" + ); + break; + } Err(error) => { tracing::debug!( observer_id = %observer.id, + attempt, %error, "durable process observer delivery will retry" ); diff --git a/crates/kernel/ironclaw_processes/tests/process_journal_store_contract.rs b/crates/kernel/ironclaw_processes/tests/process_journal_store_contract.rs index 5b4c7051c8c..9ef1953c9e8 100644 --- a/crates/kernel/ironclaw_processes/tests/process_journal_store_contract.rs +++ b/crates/kernel/ironclaw_processes/tests/process_journal_store_contract.rs @@ -122,6 +122,22 @@ impl ProcessJournalCommitObserver for FailingProcessObserver { } } +struct CountingFailingProcessObserver { + attempts: AtomicUsize, +} + +#[async_trait] +impl ProcessJournalCommitObserver for CountingFailingProcessObserver { + fn process_observer_id(&self) -> &'static str { + "counting-failing-process-observer" + } + + async fn observe_process_commit(&self, _commit: ProcessJournalCommit) -> Result<(), String> { + self.attempts.fetch_add(1, Ordering::SeqCst); + Err("permanent observer failure".to_string()) + } +} + struct FailOnceProcessObserver { attempts: AtomicUsize, } @@ -1797,6 +1813,7 @@ async fn process_observer_receives_commits_once_not_idempotency_replays() { created_at: Utc::now(), metadata: serde_json::Value::Null, }; + let process_id = request.process_id; store .submit_process(request.clone()) @@ -1807,12 +1824,28 @@ async fn process_observer_receives_commits_once_not_idempotency_replays() { .await .expect("replay process submission"); + let journal = store + .read_process_journal_log_after(None, 16) + .await + .expect("read source journal entry"); + let expected_occurred_at = journal + .entries + .iter() + .find(|entry| entry.process_id == process_id) + .and_then(|entry| entry.occurred_at) + .expect("source journal entry carries its timestamp"); + let commits = observer.commits.lock().expect("observer commits"); assert_eq!(commits.len(), 1); assert_eq!( commits[0].kind, ironclaw_processes::ProcessJournalKind::Submitted ); + assert_eq!( + commits[0].occurred_at, + Some(expected_occurred_at), + "live observer delivery preserves the source journal timestamp" + ); } #[tokio::test] @@ -1874,6 +1907,15 @@ async fn observer_registration_replays_commits_durably_after_restart() { }) .await .expect("submit before observer registration"); + let expected_occurred_at = store + .read_process_journal_log_after(None, 16) + .await + .expect("read source journal entry") + .entries + .into_iter() + .find(|entry| entry.process_id == process_id) + .and_then(|entry| entry.occurred_at) + .expect("source journal entry carries its timestamp"); let reopened = ProcessJournalStore::new(filesystem); let observer = Arc::new(RecordingProcessObserver::default()); @@ -1897,6 +1939,15 @@ async fn observer_registration_replays_commits_durably_after_restart() { }) .await .expect("durable observer replay"); + let commits = observer.commits.lock().expect("observer commits"); + assert_eq!( + commits + .iter() + .find(|commit| commit.state.process_id == process_id) + .and_then(|commit| commit.occurred_at), + Some(expected_occurred_at), + "replayed observer delivery preserves the source journal timestamp" + ); } /// Two store instances can share one observer id across a rolling restart, and @@ -2232,7 +2283,11 @@ async fn observer_replays_when_a_second_writer_leaves_a_cursor_gap() { } fn observer_cursor_path() -> ScopedPath { - let digest = blake3::hash(b"recording-process-observer").to_hex(); + observer_cursor_path_for("recording-process-observer") +} + +fn observer_cursor_path_for(observer_id: &str) -> ScopedPath { + let digest = blake3::hash(observer_id.as_bytes()).to_hex(); ScopedPath::new(format!("/processes/materialized/observer-cursor/{digest}")) .expect("cursor path") } @@ -2317,6 +2372,62 @@ async fn observer_failure_retries_until_durable_cursor_is_acknowledged() { assert_eq!(after_restart.attempts.load(Ordering::SeqCst), 0); } +#[tokio::test(start_paused = true)] +async fn permanent_observer_failure_exhausts_replay_without_acknowledging_cursor() { + let filesystem = in_memory_backed_processes_filesystem(); + let store = ProcessJournalStore::new(Arc::clone(&filesystem)); + store + .submit_process(SubmitProcessRequest { + process_id: ProcessId::new(), + process_kind: ProcessKind::Internal, + scope: scope(), + exclusive_within_scope: false, + operation_id: None, + owner_user_id: None, + concurrency_class: None, + parent_process_id: None, + root_process_id: None, + spawn_tree_descendant_cap: None, + dependency: None, + checkpoint_ref: None, + input: None, + created_at: Utc::now(), + metadata: serde_json::Value::Null, + }) + .await + .expect("submit before observer registration"); + + let observer = Arc::new(CountingFailingProcessObserver { + attempts: AtomicUsize::new(0), + }); + store + .subscribe_process_observer(observer.clone()) + .expect("subscribe permanently failing observer"); + + tokio::time::sleep(Duration::from_secs(120)).await; + let exhausted_attempts = observer.attempts.load(Ordering::SeqCst); + assert_eq!( + exhausted_attempts, 9, + "the initial registration delivery plus eight replay attempts must exhaust the budget" + ); + assert_eq!( + read_observer_cursor( + &filesystem, + &observer_cursor_path_for(observer.process_observer_id()), + ) + .await, + None, + "exhausting retries must preserve the unacknowledged durable cursor" + ); + + tokio::time::sleep(Duration::from_secs(120)).await; + assert_eq!( + observer.attempts.load(Ordering::SeqCst), + exhausted_attempts, + "a permanently failing observer must not keep retrying forever" + ); +} + #[tokio::test] async fn group_committed_submissions_deliver_every_entry_once_before_returning() { let store = Arc::new(ProcessJournalStore::new( diff --git a/crates/kernel/ironclaw_turns/src/process_projection/runtime.rs b/crates/kernel/ironclaw_turns/src/process_projection/runtime.rs index af559f515da..8f1b8c58909 100644 --- a/crates/kernel/ironclaw_turns/src/process_projection/runtime.rs +++ b/crates/kernel/ironclaw_turns/src/process_projection/runtime.rs @@ -91,7 +91,7 @@ impl ProcessJournalCommitObserver for AgentTurnProcessCommitObserver { process_id: commit.state.process_id, process_kind: commit.state.process_kind, scope: commit.state.scope, - occurred_at: Some(Utc::now()), + occurred_at: commit.occurred_at, owner_user_id: commit.state.owner_user_id, status: commit.state.status, kind: commit.kind, diff --git a/crates/loop/ironclaw_turn_runner/src/steering_reconcile.rs b/crates/loop/ironclaw_turn_runner/src/steering_reconcile.rs index 36aee334bff..82907237b0d 100644 --- a/crates/loop/ironclaw_turn_runner/src/steering_reconcile.rs +++ b/crates/loop/ironclaw_turn_runner/src/steering_reconcile.rs @@ -1085,6 +1085,7 @@ mod observer_tests { metadata: serde_json::Value::Null, }, kind, + occurred_at: None, sanitized_reason: None, }; @@ -1150,6 +1151,7 @@ mod observer_tests { metadata: serde_json::Value::Null, }, kind: ProcessJournalKind::Failed, + occurred_at: None, sanitized_reason: None, }; diff --git a/crates/product/ironclaw_assistant/README.md b/crates/product/ironclaw_assistant/README.md index 420c6206a52..704571b0219 100644 --- a/crates/product/ironclaw_assistant/README.md +++ b/crates/product/ironclaw_assistant/README.md @@ -28,6 +28,10 @@ shares this one implementation instead of reimplementing it. `IdempotencyLedger`/`ProductInboundAction`, `ProductCommandAdmissionService`. - `DeliveryCoordinator` + run-delivery drivers (delivery *semantics*; the at-most-once reservation itself is `ironclaw_outbound`'s). +- `RunOutcomeProcessCommitObserver` — materializes selected scheduled-run + completion/failure facts from the authoritative process journal; completion + requires the exact finalized assistant reply, while delivery failure remains + a separate Inbox fact from the external-delivery caller. - `ApprovalInteractionService` / `AuthInteractionService` and their redacted read models. - The frozen command/view/capability descriptor constants (`*_COMMAND`, diff --git a/crates/product/ironclaw_assistant/src/lib.rs b/crates/product/ironclaw_assistant/src/lib.rs index 937e083c336..b0823a8e2a6 100644 --- a/crates/product/ironclaw_assistant/src/lib.rs +++ b/crates/product/ironclaw_assistant/src/lib.rs @@ -64,6 +64,7 @@ mod project_create_capability; pub mod projection; mod reborn_services; mod run_delivery; +mod run_outcome_observer; mod scoped_fs; mod steering; mod suggestions; @@ -123,6 +124,7 @@ pub use channel_workflow::{ ChannelWorkflowDeliveryServices, ChannelWorkflowIdentity, RebornChannelWorkflowFactory, RebornChannelWorkflowServices, build_session_inbound_ledger, channel_conversation_services, }; +pub use run_outcome_observer::RunOutcomeProcessCommitObserver; // The conversation-binding family moved to // `ironclaw_product_contracts::binding` (§12.11 D-A): the channel host's // workflow factory hands a live binding service back to a caller that sits diff --git a/crates/product/ironclaw_assistant/src/run_delivery.rs b/crates/product/ironclaw_assistant/src/run_delivery.rs index 14df70e69c1..e80e04b5c76 100644 --- a/crates/product/ironclaw_assistant/src/run_delivery.rs +++ b/crates/product/ironclaw_assistant/src/run_delivery.rs @@ -689,7 +689,12 @@ impl RunDeliveryServices { } } -fn run_notification_inbox_id( +/// Lifecycle reference for the actionable block a delivery timeout leaves +/// behind. Publisher and resolver must derive the same stable id from it, so +/// it is named once rather than spelled at each site. +pub(crate) const TIMEOUT_LIFECYCLE_REF: &str = "timeout"; + +pub(crate) fn run_notification_inbox_id( run_id: TurnRunId, kind: NotificationKind, lifecycle_ref: Option<&str>, diff --git a/crates/product/ironclaw_assistant/src/run_delivery/triggered.rs b/crates/product/ironclaw_assistant/src/run_delivery/triggered.rs index 05c578f4a2c..797abbeaddf 100644 --- a/crates/product/ironclaw_assistant/src/run_delivery/triggered.rs +++ b/crates/product/ironclaw_assistant/src/run_delivery/triggered.rs @@ -749,7 +749,9 @@ async fn notify_background_run( notifications: vec![TriggeredNotification { event_kind: RunNotificationEventKind::RunBlocked, intent: DeliveryIntent::BackgroundRunNotice, - notice_discriminator: Some("timeout".to_string()), + notice_discriminator: Some( + crate::run_delivery::TIMEOUT_LIFECYCLE_REF.to_string(), + ), text: format!( "{}{}", prompts::DELIVERY_TIMEOUT_MESSAGE, @@ -767,7 +769,7 @@ async fn notify_background_run( &scope, run_id, NotificationKind::RunBlocked, - Some("timeout"), + Some(crate::run_delivery::TIMEOUT_LIFECYCLE_REF), ) .await; let fan = @@ -1353,6 +1355,17 @@ async fn fan_out_plan( } } } + if !targets.is_empty() && !out.any_delivered { + services + .publish_inbox_notification( + ¬ification_context.actor.user_id, + notification_context.scope, + notification_context.run_id, + NotificationKind::DeliveryFailed, + Some("external-delivery"), + ) + .await; + } out } diff --git a/crates/product/ironclaw_assistant/src/run_outcome_observer.rs b/crates/product/ironclaw_assistant/src/run_outcome_observer.rs new file mode 100644 index 00000000000..d200a6dc98d --- /dev/null +++ b/crates/product/ironclaw_assistant/src/run_outcome_observer.rs @@ -0,0 +1,876 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use ironclaw_host_api::{ + Timestamp, + execution_policy::{ResultDeliveryPolicy, TurnExecutionPolicy}, + output::OutputContract, + turn::{TurnExecutionOutcome, TurnOriginKind, TurnRunId}, +}; +use ironclaw_notifications::{ + LifecycleRef, NotificationAction, NotificationId, NotificationInboxError, + NotificationInboxStorePort, NotificationKind, NotificationMutationRequest, + NotificationRecipient, NotificationSeverity, NotificationSource, PublishNotificationRequest, +}; +use ironclaw_processes::{ + JournaledProcessSnapshot, ProcessJournalCommit, ProcessJournalCommitObserver, + ProcessJournalKind, ProcessKind, ProcessLifecycleStatus, +}; +use ironclaw_threads::{FinalizedAssistantMessageByRunRequest, SessionThreadService, ThreadScope}; +use serde::Deserialize; + +/// Materializes durable background-run outcomes from the authoritative +/// process journal. It does not observe delivery watchers or poll run state. +pub struct RunOutcomeProcessCommitObserver { + inbox: Arc, + thread_service: Arc, +} + +impl RunOutcomeProcessCommitObserver { + pub fn new( + inbox: Arc, + thread_service: Arc, + ) -> Self { + Self { + inbox, + thread_service, + } + } + + async fn publish( + &self, + snapshot: &JournaledProcessSnapshot, + run_id: TurnRunId, + kind: NotificationKind, + occurred_at: Timestamp, + ) -> Result<(), String> { + let thread_id = snapshot + .scope + .thread_id + .clone() + .ok_or_else(|| "eligible run outcome has no thread id".to_string())?; + let owner_user_id = snapshot + .owner_user_id + .clone() + .ok_or_else(|| "eligible run outcome has no owner".to_string())?; + let severity = match kind { + NotificationKind::RunCompleted => NotificationSeverity::Success, + NotificationKind::RunFailed | NotificationKind::DeliveryFailed => { + NotificationSeverity::Error + } + NotificationKind::ApprovalRequired + | NotificationKind::AuthenticationRequired + | NotificationKind::RunBlocked => NotificationSeverity::Warning, + }; + let notification_id = outcome_notification_id(run_id, kind)?; + self.inbox + .publish(PublishNotificationRequest { + id: notification_id, + recipient: NotificationRecipient { + tenant_id: snapshot.scope.tenant_id.clone(), + user_id: owner_user_id, + }, + kind, + severity, + source: NotificationSource { + thread_id: thread_id.clone(), + turn_run_id: Some(run_id), + lifecycle_ref: Some(outcome_lifecycle_ref("process-terminal")?), + }, + action: NotificationAction::OpenThread { thread_id }, + occurred_at, + }) + .await + .map_err(|error| format!("publish run outcome notification failed: {error}"))?; + Ok(()) + } +} + +impl RunOutcomeProcessCommitObserver { + /// Retire the actionable block a delivery timeout left behind. That + /// publisher returns straight after recording it, so no watcher observes + /// the run again — a terminal fact is the only thing that can close it. + /// A run that never timed out simply has no such record. + async fn resolve_timed_out_block( + &self, + snapshot: &JournaledProcessSnapshot, + run_id: TurnRunId, + occurred_at: Timestamp, + ) -> Result<(), String> { + let Some(owner_user_id) = snapshot.owner_user_id.clone() else { + return Ok(()); + }; + let notification_id = crate::run_delivery::run_notification_inbox_id( + run_id, + NotificationKind::RunBlocked, + Some(crate::run_delivery::TIMEOUT_LIFECYCLE_REF), + ) + .map_err(|error| format!("build timed-out run block id failed: {error}"))?; + match self + .inbox + .resolve(NotificationMutationRequest { + recipient: NotificationRecipient { + tenant_id: snapshot.scope.tenant_id.clone(), + user_id: owner_user_id, + }, + notification_id, + occurred_at, + }) + .await + { + // Most runs never time out, so there is usually nothing to retire; + // the outcome itself is not interesting here, only that the store + // was reached. + Ok(_) | Err(NotificationInboxError::NotificationNotFound) => Ok(()), + Err(error) => Err(format!("resolve timed-out run block failed: {error}")), + } + } +} + +#[async_trait] +impl ProcessJournalCommitObserver for RunOutcomeProcessCommitObserver { + fn process_observer_id(&self) -> &'static str { + "run-outcome-inbox-commit-observer-v1" + } + + async fn observe_process_commit(&self, commit: ProcessJournalCommit) -> Result<(), String> { + let Some(metadata) = eligible_background_run(&commit.state) else { + return Ok(()); + }; + let run_id = TurnRunId::from_uuid(commit.state.process_id.as_uuid()); + let occurred_at = commit.occurred_at.unwrap_or(commit.state.created_at); + if commit.state.status.is_terminal() { + self.resolve_timed_out_block(&commit.state, run_id, occurred_at) + .await?; + } + match (commit.kind, commit.state.status) { + (ProcessJournalKind::Completed, ProcessLifecycleStatus::Completed) => { + if !metadata.output_contract.is_assistant_message() { + return Ok(()); + } + if metadata.execution_outcome == Some(TurnExecutionOutcome::NothingToReport) + && metadata + .product_context + .as_ref() + .and_then(|context| context.execution_policy.as_ref()) + .is_some_and(|policy| { + policy.result_delivery + == ResultDeliveryPolicy::SuppressWhenNothingToReport + }) + { + return Ok(()); + } + let Some(thread_scope) = thread_scope_for_snapshot(&commit.state) else { + return Ok(()); + }; + let thread_id = commit + .state + .scope + .thread_id + .clone() + .ok_or_else(|| "eligible completed run has no thread id".to_string())?; + let final_reply = self + .thread_service + .finalized_assistant_message_by_run(FinalizedAssistantMessageByRunRequest { + scope: thread_scope, + thread_id, + turn_run_id: run_id.to_string(), + }) + .await + .map_err(|error| format!("read finalized run reply failed: {error}"))?; + let Some(_final_reply) = final_reply else { + // The loop host appends the finalized reply, with retry, + // before the turn can report completion, so an absent one + // here is a contract miss rather than a routine race — + // returning an error retains the durable observer cursor + // so replay can try again after the reply is persisted. + return Err(format!( + "completed background run {run_id} has no finalized assistant reply" + )); + }; + self.publish( + &commit.state, + run_id, + NotificationKind::RunCompleted, + occurred_at, + ) + .await?; + } + (ProcessJournalKind::Failed, ProcessLifecycleStatus::Failed) + | (ProcessJournalKind::RecoveryRequired, ProcessLifecycleStatus::RecoveryRequired) => { + self.publish( + &commit.state, + run_id, + NotificationKind::RunFailed, + occurred_at, + ) + .await?; + } + _ => {} + } + Ok(()) + } +} + +#[derive(Debug, Deserialize)] +struct OutcomeMetadataEnvelope { + agent_turn: OutcomeMetadata, +} + +#[derive(Debug, Deserialize)] +struct OutcomeMetadata { + #[serde(default)] + output_contract: OutputContract, + #[serde(default)] + execution_outcome: Option, + #[serde(default)] + subagent_depth: u32, + #[serde(default)] + ownerless_thread: bool, + product_context: Option, +} + +#[derive(Debug, Deserialize)] +struct OutcomeProductContext { + origin: TurnOriginKind, + #[serde(default)] + execution_policy: Option, +} + +fn eligible_background_run(snapshot: &JournaledProcessSnapshot) -> Option { + if snapshot.process_kind != ProcessKind::AgentTurn + || snapshot.parent_process_id.is_some() + || snapshot.owner_user_id.is_none() + || snapshot.scope.thread_id.is_none() + || snapshot.scope.agent_id.is_none() + { + return None; + } + // silent-ok: eligibility screening. Journal metadata this observer cannot + // read describes a process it does not own, so the absence of a notification + // is the correct outcome rather than a failure to report; the cause is + // recorded above before it is dropped. + let envelope = serde_json::from_value::(snapshot.metadata.clone()) + .map_err(|error| { + tracing::debug!( + process_id = %snapshot.process_id, + %error, + "run outcome observer ignored malformed agent-turn metadata" + ); + }) + .ok()?; + let metadata = envelope.agent_turn; + if metadata.subagent_depth != 0 + || metadata.ownerless_thread + || metadata + .product_context + .as_ref() + .is_none_or(|context| context.origin != TurnOriginKind::ScheduledTrigger) + { + return None; + } + Some(metadata) +} + +fn thread_scope_for_snapshot(snapshot: &JournaledProcessSnapshot) -> Option { + Some(ThreadScope { + tenant_id: snapshot.scope.tenant_id.clone(), + agent_id: snapshot.scope.agent_id.clone()?, + project_id: snapshot.scope.project_id.clone(), + owner_user_id: snapshot.owner_user_id.clone(), + mission_id: snapshot.scope.mission_id.clone(), + }) +} + +fn outcome_lifecycle_ref(value: &'static str) -> Result { + LifecycleRef::new(value) + .map_err(|error| format!("build run outcome lifecycle reference failed: {error}")) +} + +fn outcome_notification_id( + run_id: TurnRunId, + kind: NotificationKind, +) -> Result { + let kind = match kind { + NotificationKind::RunCompleted => "completed", + NotificationKind::RunFailed => "failed", + NotificationKind::DeliveryFailed => "delivery-failed", + NotificationKind::ApprovalRequired => "approval", + NotificationKind::AuthenticationRequired => "authentication", + NotificationKind::RunBlocked => "blocked", + }; + NotificationId::new(format!("run:{run_id}:{kind}")) + .map_err(|error| format!("build run outcome notification id failed: {error}")) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use chrono::Utc; + use ironclaw_filesystem::{InMemoryBackend, ScopedFilesystem}; + use ironclaw_host_api::{ + ids::{AgentId, ProcessId, TenantId, ThreadId, UserId}, + mount::{MountGrant, MountPermissions, MountView}, + path::{MountAlias, VirtualPath}, + resource::ResourceScope, + turn::TurnRunId, + }; + use ironclaw_notifications::{ + LifecycleRef, ListNotificationsRequest, NotificationAction, NotificationInboxStore, + NotificationInboxStorePort, NotificationKind, NotificationRecipient, NotificationSeverity, + NotificationSource, PublishNotificationRequest, + }; + use ironclaw_processes::{ + ClaimProcessesRequest, JournaledProcessSnapshot, ProcessJournalCommit, + ProcessJournalCommitObserver, ProcessJournalCursor, ProcessJournalKind, + ProcessJournalObserverRegistry, ProcessJournalSource, ProcessJournalStore, ProcessKind, + ProcessLeaseRequest, ProcessLifecycleStatus, ProcessStateTransitionRequest, + ProcessSubmissionPort, ProcessTransitionPort, ProcessWorkerId, SubmitProcessRequest, + }; + use ironclaw_threads::{ + AppendFinalizedAssistantMessageRequest, EnsureThreadRequest, InMemorySessionThreadService, + MessageContent, SessionThreadService, ThreadScope, + }; + use serde_json::json; + + use super::RunOutcomeProcessCommitObserver; + + fn tenant() -> TenantId { + TenantId::new("outcome-observer-tenant").expect("tenant") + } + + fn user() -> UserId { + UserId::new("outcome-observer-user").expect("user") + } + + fn agent() -> AgentId { + AgentId::new("outcome-observer-agent").expect("agent") + } + + fn thread() -> ThreadId { + ThreadId::new("outcome-observer-thread").expect("thread") + } + + fn thread_scope() -> ThreadScope { + ThreadScope { + tenant_id: tenant(), + agent_id: agent(), + project_id: None, + owner_user_id: Some(user()), + mission_id: None, + } + } + + fn inbox() -> Arc> { + let mounts = MountView::new(vec![MountGrant::new( + MountAlias::new("/notifications").expect("notification mount alias"), + VirtualPath::new("/engine/test/run-outcome-notifications") + .expect("notification mount target"), + MountPermissions::read_write_list_delete(), + )]) + .expect("notification mount view"); + Arc::new(NotificationInboxStore::new( + Arc::new(ScopedFilesystem::with_fixed_view( + Arc::new(InMemoryBackend::new()), + mounts, + )), + ironclaw_notifications::NOTIFICATION_INBOX_MAX_RECORDS, + )) + } + + fn process_filesystem() -> Arc> { + let mounts = MountView::new(vec![MountGrant::new( + MountAlias::new("/processes").expect("process mount alias"), + VirtualPath::new("/engine/test/run-outcome-processes").expect("process mount target"), + MountPermissions::read_write_list_delete(), + )]) + .expect("process mount view"); + Arc::new(ScopedFilesystem::with_fixed_view( + Arc::new(InMemoryBackend::new()), + mounts, + )) + } + + fn commit( + run_id: TurnRunId, + status: ProcessLifecycleStatus, + kind: ProcessJournalKind, + origin: &str, + ) -> ProcessJournalCommit { + let now = Utc::now(); + ProcessJournalCommit { + state: JournaledProcessSnapshot { + process_id: ProcessId::from_uuid(run_id.as_uuid()), + process_kind: ProcessKind::AgentTurn, + scope: ResourceScope { + tenant_id: tenant(), + user_id: user(), + agent_id: Some(agent()), + project_id: None, + mission_id: None, + thread_id: Some(thread()), + invocation_id: ironclaw_host_api::ids::InvocationId::new(), + }, + status, + suspension: None, + checkpoint_ref: None, + checkpoint_kind: None, + input_ref: None, + failure: None, + journal_cursor: ProcessJournalCursor(1), + lease: None, + crash_reclaim_count: 0, + created_at: now, + owner_user_id: Some(user()), + concurrency_class: None, + parent_process_id: None, + root_process_id: None, + metadata: json!({ + "agent_turn": { + "product_context": { "origin": origin }, + "execution_outcome": "result_available" + } + }), + }, + kind, + sanitized_reason: None, + occurred_at: Some(now), + } + } + + /// A commit whose agent-turn metadata carries the exclusions the observer + /// screens on, so a future edit to either predicate fails a test instead of + /// quietly publishing for a child or ownerless run. + fn excluded_commit( + run_id: TurnRunId, + subagent_depth: u32, + ownerless_thread: bool, + ) -> ProcessJournalCommit { + let mut commit = commit( + run_id, + ProcessLifecycleStatus::Failed, + ProcessJournalKind::Failed, + "scheduled_trigger", + ); + commit.state.metadata = json!({ + "agent_turn": { + "product_context": { "origin": "scheduled_trigger" }, + "execution_outcome": "result_available", + "subagent_depth": subagent_depth, + "ownerless_thread": ownerless_thread, + } + }); + commit + } + + async fn records(inbox: &dyn NotificationInboxStorePort) -> Vec { + inbox + .list(ListNotificationsRequest { + recipient: NotificationRecipient { + tenant_id: tenant(), + user_id: user(), + }, + limit: 10, + cursor: None, + include_archived: false, + }) + .await + .expect("list notifications") + .notifications + .into_iter() + .map(|record| record.kind) + .collect() + } + + /// A timed-out fire publishes an actionable block and the delivery watcher + /// then returns, so nothing else ever observes that run again. Only a + /// terminal fact can retire the record. + #[tokio::test] + async fn a_terminal_run_resolves_the_block_left_behind_by_a_delivery_timeout() { + for status in [ + ProcessLifecycleStatus::Completed, + ProcessLifecycleStatus::Failed, + ProcessLifecycleStatus::Cancelled, + ] { + let inbox = inbox(); + let threads = Arc::new(InMemorySessionThreadService::default()); + threads + .ensure_thread(EnsureThreadRequest { + scope: thread_scope(), + thread_id: Some(thread()), + created_by_actor_id: user().to_string(), + title: None, + metadata_json: None, + }) + .await + .expect("thread"); + let run_id = TurnRunId::new(); + let recipient = NotificationRecipient { + tenant_id: tenant(), + user_id: user(), + }; + + inbox + .publish(PublishNotificationRequest { + id: crate::run_delivery::run_notification_inbox_id( + run_id, + NotificationKind::RunBlocked, + Some(crate::run_delivery::TIMEOUT_LIFECYCLE_REF), + ) + .expect("timeout block id"), + recipient: recipient.clone(), + kind: NotificationKind::RunBlocked, + severity: NotificationSeverity::Warning, + source: NotificationSource { + thread_id: thread(), + turn_run_id: Some(run_id), + lifecycle_ref: Some( + LifecycleRef::new(crate::run_delivery::TIMEOUT_LIFECYCLE_REF) + .expect("timeout lifecycle ref"), + ), + }, + action: NotificationAction::OpenThread { + thread_id: thread(), + }, + occurred_at: Utc::now(), + }) + .await + .expect("seed the timed-out block"); + + if status == ProcessLifecycleStatus::Completed { + threads + .append_finalized_assistant_message(AppendFinalizedAssistantMessageRequest { + scope: thread_scope(), + thread_id: thread(), + turn_run_id: run_id.to_string(), + content: MessageContent::text("durable final reply"), + }) + .await + .expect("completed run final reply"); + } + + let observer = RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + Arc::clone(&threads) as Arc, + ); + observer + .observe_process_commit(commit( + run_id, + status, + ProcessJournalKind::Completed, + "scheduled_trigger", + )) + .await + .expect("terminal commit"); + + let page = inbox + .list(ListNotificationsRequest { + recipient, + limit: 10, + cursor: None, + include_archived: true, + }) + .await + .expect("list"); + let block = page + .notifications + .iter() + .find(|record| record.kind == NotificationKind::RunBlocked) + .expect("the block record survives"); + assert!( + block.resolved_at.is_some(), + "{status:?} must retire the block a delivery timeout left open", + ); + } + } + + #[tokio::test] + async fn completed_background_run_publishes_only_after_exact_reply_is_finalized() { + let inbox = inbox(); + let threads = Arc::new(InMemorySessionThreadService::default()); + let run_id = TurnRunId::new(); + threads + .ensure_thread(EnsureThreadRequest { + scope: thread_scope(), + thread_id: Some(thread()), + created_by_actor_id: user().to_string(), + title: None, + metadata_json: None, + }) + .await + .expect("thread"); + let observer = RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + Arc::clone(&threads) as Arc, + ); + + let completion = commit( + run_id, + ProcessLifecycleStatus::Completed, + ProcessJournalKind::Completed, + "scheduled_trigger", + ); + let committed_at = completion.occurred_at.expect("committed timestamp"); + let error = observer + .observe_process_commit(completion.clone()) + .await + .expect_err("missing final reply must retain the observer cursor for retry"); + assert!(error.contains("no finalized assistant reply"), "{error}"); + assert!(records(inbox.as_ref()).await.is_empty()); + + threads + .append_finalized_assistant_message(AppendFinalizedAssistantMessageRequest { + scope: thread_scope(), + thread_id: thread(), + turn_run_id: run_id.to_string(), + content: MessageContent::text("durable final reply"), + }) + .await + .expect("final reply"); + observer + .observe_process_commit(completion) + .await + .expect("completion notification"); + + let page = inbox + .list(ListNotificationsRequest { + recipient: NotificationRecipient { + tenant_id: tenant(), + user_id: user(), + }, + limit: 10, + cursor: None, + include_archived: false, + }) + .await + .expect("list completion notification"); + assert_eq!(page.notifications.len(), 1); + assert_eq!(page.notifications[0].kind, NotificationKind::RunCompleted); + assert_eq!( + page.notifications[0].created_at, committed_at, + "completion ordering uses the committed journal transition timestamp" + ); + } + + #[tokio::test] + async fn durable_journal_retries_completion_after_the_final_reply_is_persisted() { + let inbox = inbox(); + let threads = Arc::new(InMemorySessionThreadService::default()); + threads + .ensure_thread(EnsureThreadRequest { + scope: thread_scope(), + thread_id: Some(thread()), + created_by_actor_id: user().to_string(), + title: None, + metadata_json: None, + }) + .await + .expect("thread"); + let process_filesystem = process_filesystem(); + let store = ProcessJournalStore::new(Arc::clone(&process_filesystem)); + store + .subscribe_process_observer(Arc::new(RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + Arc::clone(&threads) as Arc, + ))) + .expect("subscribe outcome observer"); + let run_id = TurnRunId::new(); + let process_id = ProcessId::from_uuid(run_id.as_uuid()); + let resource_scope = ResourceScope { + tenant_id: tenant(), + user_id: user(), + agent_id: Some(agent()), + project_id: None, + mission_id: None, + thread_id: Some(thread()), + invocation_id: ironclaw_host_api::ids::InvocationId::new(), + }; + store + .submit_process(SubmitProcessRequest { + process_id, + process_kind: ProcessKind::AgentTurn, + scope: resource_scope.clone(), + exclusive_within_scope: false, + operation_id: None, + owner_user_id: Some(user()), + concurrency_class: None, + parent_process_id: None, + root_process_id: None, + spawn_tree_descendant_cap: None, + dependency: None, + checkpoint_ref: None, + input: None, + created_at: Utc::now(), + metadata: json!({ + "agent_turn": { + "product_context": { "origin": "scheduled_trigger" }, + "execution_outcome": "result_available" + } + }), + }) + .await + .expect("submit process"); + let claimed = store + .claim_next_processes(ClaimProcessesRequest { + worker_id: ProcessWorkerId::from_trusted("outcome-observer-worker"), + scope_filter: Some(resource_scope), + process_id_filter: Some(process_id), + process_kind_filter: Some(ProcessKind::AgentTurn), + max_processes: 1, + }) + .await + .expect("claim process") + .pop() + .expect("process is claimable"); + store + .complete_process(ProcessStateTransitionRequest { + lease: ProcessLeaseRequest { + process_id, + worker_id: claimed.worker_id, + lease_token: claimed.lease_token, + }, + metadata: None, + }) + .await + .expect("terminal process commit remains successful"); + let committed_at = store + .read_process_journal_log_after(None, 16) + .await + .expect("read committed journal row") + .entries + .into_iter() + .find(|entry| { + entry.process_id == process_id && entry.kind == ProcessJournalKind::Completed + }) + .and_then(|entry| entry.occurred_at) + .expect("committed journal timestamp"); + assert!( + records(inbox.as_ref()).await.is_empty(), + "the first delivery attempt cannot publish before the exact reply exists" + ); + + threads + .append_finalized_assistant_message(AppendFinalizedAssistantMessageRequest { + scope: thread_scope(), + thread_id: thread(), + turn_run_id: run_id.to_string(), + content: MessageContent::text("durable final reply"), + }) + .await + .expect("persist final reply before observer retry"); + + let page = tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + let page = inbox + .list(ListNotificationsRequest { + recipient: NotificationRecipient { + tenant_id: tenant(), + user_id: user(), + }, + limit: 10, + cursor: None, + include_archived: false, + }) + .await + .expect("list completion notification"); + if !page.notifications.is_empty() { + break page; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("durable observer retries the commit"); + assert_eq!(page.notifications.len(), 1); + assert_eq!(page.notifications[0].kind, NotificationKind::RunCompleted); + assert_eq!(page.notifications[0].created_at, committed_at); + } + + #[tokio::test] + async fn failed_background_run_is_deduplicated_and_foreground_completion_is_ignored() { + let inbox = inbox(); + let threads = Arc::new(InMemorySessionThreadService::default()); + let observer = RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + threads as Arc, + ); + let failed_run = TurnRunId::new(); + let failed = commit( + failed_run, + ProcessLifecycleStatus::Failed, + ProcessJournalKind::Failed, + "scheduled_trigger", + ); + + observer + .observe_process_commit(failed.clone()) + .await + .expect("failed notification"); + observer + .observe_process_commit(failed) + .await + .expect("replayed failure"); + observer + .observe_process_commit(commit( + TurnRunId::new(), + ProcessLifecycleStatus::Completed, + ProcessJournalKind::Completed, + "web_ui", + )) + .await + .expect("foreground completion is ignored"); + + assert_eq!( + records(inbox.as_ref()).await, + vec![NotificationKind::RunFailed] + ); + } + #[tokio::test] + async fn a_recovery_required_run_publishes_the_failure_outcome() { + let inbox = inbox(); + let observer = RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + Arc::new(InMemorySessionThreadService::default()) as Arc, + ); + + observer + .observe_process_commit(commit( + TurnRunId::new(), + ProcessLifecycleStatus::RecoveryRequired, + ProcessJournalKind::RecoveryRequired, + "scheduled_trigger", + )) + .await + .expect("recovery-required commit"); + + assert_eq!( + records(inbox.as_ref()).await, + vec![NotificationKind::RunFailed], + "a run needing recovery is reported as a failed outcome" + ); + } + + #[tokio::test] + async fn a_child_or_ownerless_run_publishes_nothing() { + let inbox = inbox(); + let observer = RunOutcomeProcessCommitObserver::new( + Arc::clone(&inbox) as Arc, + Arc::new(InMemorySessionThreadService::default()) as Arc, + ); + + observer + .observe_process_commit(excluded_commit(TurnRunId::new(), 1, false)) + .await + .expect("a subagent turn is screened out, not an error"); + observer + .observe_process_commit(excluded_commit(TurnRunId::new(), 0, true)) + .await + .expect("an ownerless thread is screened out, not an error"); + + assert!( + records(inbox.as_ref()).await.is_empty(), + "neither a child run nor an ownerless thread reaches a user's inbox" + ); + } +} diff --git a/crates/product/ironclaw_assistant/src/suggestions_observer.rs b/crates/product/ironclaw_assistant/src/suggestions_observer.rs index 1e373813bca..a1c26413f81 100644 --- a/crates/product/ironclaw_assistant/src/suggestions_observer.rs +++ b/crates/product/ironclaw_assistant/src/suggestions_observer.rs @@ -353,6 +353,7 @@ mod tests { .observe_process_commit(ProcessJournalCommit { state: snapshot, kind: ProcessJournalKind::Completed, + occurred_at: Some(now), sanitized_reason: None, }) .await diff --git a/crates/product/ironclaw_assistant/tests/run_delivery_contract.rs b/crates/product/ironclaw_assistant/tests/run_delivery_contract.rs index ffd9bb4f22f..efd0f9d411e 100644 --- a/crates/product/ironclaw_assistant/tests/run_delivery_contract.rs +++ b/crates/product/ironclaw_assistant/tests/run_delivery_contract.rs @@ -3470,15 +3470,23 @@ async fn triggered_failed_gate_fanout_still_resolves_the_inbox_after_resume() { ) .await; let inbox = inbox_records(harness.notification_inbox.as_ref()).await; - assert_eq!(inbox.notifications.len(), 1); - assert_eq!( - inbox.notifications[0].kind, - NotificationKind::ApprovalRequired - ); + assert_eq!(inbox.notifications.len(), 2); + let approval = inbox + .notifications + .iter() + .find(|notification| notification.kind == NotificationKind::ApprovalRequired) + .expect("approval gate remains in the Inbox"); assert!( - inbox.notifications[0].resolved_at.is_some(), + approval.resolved_at.is_some(), "external fan-out failure must not strand the durable Inbox gate" ); + assert!( + inbox + .notifications + .iter() + .any(|notification| notification.kind == NotificationKind::DeliveryFailed), + "the failed external fan-out remains visible as a separate Inbox fact" + ); assert_eq!( harness.adapter.texts().len(), 1, diff --git a/docs/internal/reborn/contracts/notification-inbox.md b/docs/internal/reborn/contracts/notification-inbox.md new file mode 100644 index 00000000000..3746ffc341c --- /dev/null +++ b/docs/internal/reborn/contracts/notification-inbox.md @@ -0,0 +1,38 @@ +# Notification Inbox contract + +The notification Inbox is durable, metadata-only product state. Record grammar, +recipient isolation, bounded retention, pagination, and orthogonal read, +resolved, and archived timestamps are owned by `ironclaw_notifications`. +Product eligibility and production belong to `ironclaw_assistant`; external +delivery attempts remain owned by `ironclaw_outbound`. + +## Run outcome production + +- The durable process journal is the source of truth. Delivery watchers do not + manufacture run completion or failure facts. +- Only top-level, user-owned, thread-backed scheduled-trigger AgentTurn runs are + eligible. Foreground WebUI runs, child runs, and ownerless runs do not create + outcome Inbox items. Structured-output completions have no ordinary final + assistant reply and are excluded; their terminal failures remain eligible. +- A completed run produces `run_completed` only after the thread service finds + a finalized assistant message for the exact `turn_run_id`. +- `SuppressWhenNothingToReport` plus a durable `NothingToReport` outcome creates + no completion item. +- Failed and recovery-required scheduled runs produce `run_failed` from their + matching committed terminal transition. +- A failed external notification attempt may produce `delivery_failed`; it is + not a run failure and never changes the process journal. +- IDs are derived from the run and outcome kind. Replay, retry, and restart + therefore converge on the same record. +- Records are retained. Lifecycle cleanup means resolving or archiving, never + deleting history. + +## Verification + +```bash +cargo test -p ironclaw_assistant run_outcome_observer +cargo test -p ironclaw_assistant --test run_delivery_contract triggered_timeout_notice_delivery_failure_records_failed +cargo test -p ironclaw_composition --features test-support --test trigger_poller_e2e scheduled_trigger_results_are_never_pushed_to_a_channel_across_restart +cargo test -p ironclaw_notifications +cargo test -p ironclaw_architecture_tests +``` diff --git a/docs/internal/reborn/contracts/processes.md b/docs/internal/reborn/contracts/processes.md index 154d6309cac..74d32a4a22c 100644 --- a/docs/internal/reborn/contracts/processes.md +++ b/docs/internal/reborn/contracts/processes.md @@ -27,6 +27,11 @@ Observers and transport projections do not own state. A committed journal mutation is successful even if an in-process wake hint fails; required consumers must be replayable from durable journal cursors. +Background observer replay uses a bounded exponential retry budget. Exhausting +that budget emits an operator-visible error and stops without acknowledging the +cursor; a later commit or observer registration resumes from the same durable +position instead of losing the stalled commit. + ## Durable layout The store uses `RootFilesystem` multi-key transactions under: diff --git a/scripts/reborn-e2e-rust.sh b/scripts/reborn-e2e-rust.sh index 9f4d7ed6daa..f28843c14c0 100755 --- a/scripts/reborn-e2e-rust.sh +++ b/scripts/reborn-e2e-rust.sh @@ -172,6 +172,11 @@ run_runtimes() { run_test ironclaw_processes process_journal_store_contract run_test ironclaw_processes legacy_migration_backend_contract run_test ironclaw_processes process_services_contract + # Pins docs/internal/reborn/contracts/notification-inbox.md: scheduled-run + # completion is materialized only from the committed process transition and + # its exact finalized assistant reply, never from a delivery watcher. + run_lib_test_exact ironclaw_assistant \ + run_outcome_observer::tests::completed_background_run_publishes_only_after_exact_reply_is_finalized } run_substrates() {