1use std::collections::HashMap;
2use std::sync::{Arc, RwLock};
3use std::time::Instant;
4
5use chrono::{DateTime, Duration, Utc};
6use surrealdb::engine::any::Any;
7use surrealdb::Surreal;
8
9use crate::application::runtime::in_memory_runtime::{JobExecutionOutcome, JobHandler};
10use crate::application::runtime::replay_report::ReplayReport;
11use crate::application::runtime::retention::{RetentionPolicy, RetentionPruneReport};
12use crate::application::runtime::runtime_diagnostics_helpers;
13use crate::application::runtime::runtime_job_identity_context::RuntimeJobIdentityContext;
14use crate::application::use_cases::investigate_runtime_lineage::{
15 InvestigateRuntimeLineage, RuntimeLineageQuery, RuntimeLineageReport,
16};
17use crate::domain::errors::{Result, StasisError};
18use crate::domain::runtime::job::{JobState, NewJob};
19use crate::domain::runtime::job_attempt::{JobAttempt, JobAttemptOutcome};
20use crate::domain::runtime::outbox::{
21 OutboxEvent, OutboxPublishPolicy, OutboxStatus, RuntimeEvent, RuntimeEventType,
22};
23use crate::domain::runtime::recurring::RecurringDefinition;
24use crate::infrastructure::runtime::atomic_id_generator::AtomicIdGenerator;
25use crate::infrastructure::runtime::noop_runtime_metrics::NoopRuntimeMetrics;
26use crate::infrastructure::runtime::surreal_job_attempt_store::SurrealJobAttemptStore;
27use crate::infrastructure::runtime::surreal_job_store::SurrealJobStore;
28use crate::infrastructure::runtime::surreal_outbox_store::SurrealOutboxStore;
29use crate::infrastructure::runtime::surreal_recurring_store::SurrealRecurringStore;
30use crate::infrastructure::runtime::system_clock::SystemClock;
31use crate::ports::outbound::runtime::clock::Clock;
32use crate::ports::outbound::runtime::event_publisher::EventPublisher;
33use crate::ports::outbound::runtime::id_generator::IdGenerator;
34use crate::ports::outbound::runtime::job_attempt_store::JobAttemptStore;
35use crate::ports::outbound::runtime::job_store::JobStore;
36use crate::ports::outbound::runtime::outbox_store::OutboxStore;
37use crate::ports::outbound::runtime::recurring_store::RecurringStore;
38use crate::ports::outbound::runtime::runtime_metrics::RuntimeMetrics;
39
40const METRIC_JOB_SUCCEEDED_TOTAL: &str = "runtime.job.succeeded.total";
41const METRIC_JOB_RETRYABLE_FAILURE_TOTAL: &str = "runtime.job.retryable_failure.total";
42const METRIC_JOB_FATAL_FAILURE_TOTAL: &str = "runtime.job.fatal_failure.total";
43const METRIC_JOB_DEAD_LETTER_TOTAL: &str = "runtime.job.dead_letter.total";
44const METRIC_JOB_RETRY_SCHEDULED_TOTAL: &str = "runtime.job.retry_scheduled.total";
45const METRIC_JOB_PROCESS_DURATION_MS: &str = "runtime.job.process.duration_ms";
46const METRIC_OUTBOX_PUBLISH_SUCCESS_TOTAL: &str = "runtime.outbox.publish.success.total";
47const METRIC_OUTBOX_PUBLISH_FAILURE_TOTAL: &str = "runtime.outbox.publish.failure.total";
48const METRIC_GRAPHEME_GUARDRAIL_FAILURE_TOTAL: &str = "runtime.grapheme.guardrail_failure.total";
49
50#[derive(Clone)]
51pub struct SurrealRuntime {
52 pub job_store: SurrealJobStore,
53 pub recurring_store: SurrealRecurringStore,
54 pub outbox_store: SurrealOutboxStore,
55 pub job_attempt_store: SurrealJobAttemptStore,
56 handlers: Arc<RwLock<HashMap<String, Arc<dyn JobHandler>>>>,
57 publisher: Arc<RwLock<Option<Arc<dyn EventPublisher>>>>,
58 publish_policy: Arc<RwLock<OutboxPublishPolicy>>,
59 clock: Arc<dyn Clock>,
60 id_generator: Arc<dyn IdGenerator>,
61 metrics: Arc<dyn RuntimeMetrics>,
62 retention_policy: Arc<RwLock<RetentionPolicy>>,
63}
64
65impl SurrealRuntime {
66 pub fn new(db: Surreal<Any>) -> Self {
67 Self::with_dependencies_and_metrics(
68 db,
69 Arc::new(SystemClock),
70 Arc::new(AtomicIdGenerator::new(1)),
71 Arc::new(NoopRuntimeMetrics),
72 )
73 }
74
75 pub fn with_dependencies(
76 db: Surreal<Any>,
77 clock: Arc<dyn Clock>,
78 id_generator: Arc<dyn IdGenerator>,
79 ) -> Self {
80 Self::with_dependencies_and_metrics(db, clock, id_generator, Arc::new(NoopRuntimeMetrics))
81 }
82
83 pub fn with_dependencies_and_metrics(
84 db: Surreal<Any>,
85 clock: Arc<dyn Clock>,
86 id_generator: Arc<dyn IdGenerator>,
87 metrics: Arc<dyn RuntimeMetrics>,
88 ) -> Self {
89 Self {
90 job_store: SurrealJobStore::new(db.clone()),
91 recurring_store: SurrealRecurringStore::new(db.clone()),
92 outbox_store: SurrealOutboxStore::new(db.clone()),
93 job_attempt_store: SurrealJobAttemptStore::new(db),
94 handlers: Arc::new(RwLock::new(HashMap::new())),
95 publisher: Arc::new(RwLock::new(None)),
96 publish_policy: Arc::new(RwLock::new(OutboxPublishPolicy::default())),
97 clock,
98 id_generator,
99 metrics,
100 retention_policy: Arc::new(RwLock::new(RetentionPolicy::default())),
101 }
102 }
103
104 pub fn configure_retention_policy(&self, policy: RetentionPolicy) -> Result<()> {
105 let mut state = self
106 .retention_policy
107 .write()
108 .map_err(|_| StasisError::PortFailure("retention policy lock poisoned".to_string()))?;
109 *state = policy;
110 Ok(())
111 }
112
113 pub fn register_handler<H: JobHandler + 'static>(&self, handler: H) -> Result<()> {
114 let mut handlers = self
115 .handlers
116 .write()
117 .map_err(|_| StasisError::PortFailure("handlers lock poisoned".to_string()))?;
118
119 handlers.insert(handler.job_type().to_string(), Arc::new(handler));
120 Ok(())
121 }
122
123 pub fn register_event_publisher<P: EventPublisher + 'static>(
124 &self,
125 publisher: P,
126 ) -> Result<()> {
127 let mut state = self
128 .publisher
129 .write()
130 .map_err(|_| StasisError::PortFailure("publisher lock poisoned".to_string()))?;
131
132 *state = Some(Arc::new(publisher));
133 Ok(())
134 }
135
136 pub fn configure_outbox_publish_policy(&self, policy: OutboxPublishPolicy) -> Result<()> {
137 let mut state = self
138 .publish_policy
139 .write()
140 .map_err(|_| StasisError::PortFailure("publish policy lock poisoned".to_string()))?;
141
142 *state = policy;
143 Ok(())
144 }
145
146 pub async fn enqueue(&self, job: NewJob) -> Result<()> {
147 self.job_store.insert(job.into_job()).await
148 }
149
150 pub async fn register_recurring(&self, definition: RecurringDefinition) -> Result<()> {
151 self.recurring_store.insert(definition).await
152 }
153
154 pub async fn list_job_attempts(&self, job_id: &str) -> Result<Vec<JobAttempt>> {
155 self.job_attempt_store.list_by_job_id(job_id).await
156 }
157
158 pub async fn list_attempts_by_guardrail_code(
159 &self,
160 guardrail_code: &str,
161 ) -> Result<Vec<JobAttempt>> {
162 self.job_attempt_store
163 .list_by_guardrail_code(guardrail_code)
164 .await
165 }
166
167 pub async fn list_attempts_by_execution_id(
168 &self,
169 execution_id: &str,
170 ) -> Result<Vec<JobAttempt>> {
171 self.job_attempt_store
172 .list_by_execution_id(execution_id)
173 .await
174 }
175
176 pub async fn list_lineage_events(&self, job_id: &str) -> Result<Vec<OutboxEvent>> {
177 self.outbox_store.list_by_job_id(job_id).await
178 }
179
180 pub async fn list_lineage_events_by_execution_id(
181 &self,
182 execution_id: &str,
183 ) -> Result<Vec<OutboxEvent>> {
184 self.outbox_store.list_by_execution_id(execution_id).await
185 }
186
187 pub async fn list_lineage_events_by_thread_id(
188 &self,
189 thread_id: &str,
190 ) -> Result<Vec<OutboxEvent>> {
191 self.outbox_store.list_by_thread_id(thread_id).await
192 }
193
194 pub async fn investigate_lineage(
195 &self,
196 query: RuntimeLineageQuery,
197 ) -> Result<RuntimeLineageReport> {
198 InvestigateRuntimeLineage::new(self.job_attempt_store.clone(), self.outbox_store.clone())
199 .execute(query)
200 .await
201 }
202
203 pub async fn get_replay_report(&self, job_id: &str) -> Result<ReplayReport> {
204 Ok(ReplayReport {
205 job_id: job_id.to_string(),
206 attempts: self.list_job_attempts(job_id).await?,
207 lineage_events: self.list_lineage_events(job_id).await?,
208 })
209 }
210
211 pub async fn process_once_now(&self, queue: &str, worker_id: &str) -> Result<Option<String>> {
212 self.process_once(queue, worker_id, self.clock.now()).await
213 }
214
215 pub async fn replay_dead_letter_now(&self, job_id: &str) -> Result<bool> {
216 self.replay_dead_letter(job_id, self.clock.now()).await
217 }
218
219 pub async fn publish_pending_events_now(&self, limit: usize) -> Result<usize> {
220 self.publish_pending_events(limit, self.clock.now()).await
221 }
222
223 pub async fn materialize_recurring_now(&self, scheduler_id: &str) -> Result<usize> {
224 self.materialize_recurring(self.clock.now(), scheduler_id)
225 .await
226 }
227
228 pub async fn prune_terminal_records(
229 &self,
230 cutoff: DateTime<Utc>,
231 ) -> Result<RetentionPruneReport> {
232 Ok(RetentionPruneReport {
233 jobs_pruned: self.job_store.prune_terminal_before(cutoff).await?,
234 attempts_pruned: self.job_attempt_store.prune_finished_before(cutoff).await?,
235 outbox_events_pruned: self.outbox_store.prune_non_pending_before(cutoff).await?,
236 })
237 }
238
239 pub async fn enforce_retention(&self, now: DateTime<Utc>) -> Result<RetentionPruneReport> {
240 let policy = self
241 .retention_policy
242 .read()
243 .map_err(|_| StasisError::PortFailure("retention policy lock poisoned".to_string()))?
244 .clone();
245 let cutoff = now - Duration::days(policy.terminal_ttl_days.max(0));
246 self.prune_terminal_records(cutoff).await
247 }
248
249 pub async fn enforce_retention_now(&self) -> Result<RetentionPruneReport> {
250 self.enforce_retention(self.clock.now()).await
251 }
252
253 pub async fn materialize_recurring(
254 &self,
255 now: DateTime<Utc>,
256 scheduler_id: &str,
257 ) -> Result<usize> {
258 let due = self
259 .recurring_store
260 .lease_due(now, scheduler_id, 30)
261 .await?;
262
263 let mut produced = 0usize;
264
265 for mut definition in due {
266 if !definition.enabled {
267 continue;
268 }
269
270 let id = self.id_generator.next_id(&definition.id).to_string();
271
272 let scheduled_at = now + Duration::seconds(definition.jitter_seconds.max(0));
273
274 let job = NewJob {
275 id,
276 queue: definition.queue.clone(),
277 job_type: definition.job_type.clone(),
278 payload_ref: definition.payload_template_ref.clone(),
279 priority: 100,
280 max_attempts: definition.max_attempts,
281 idempotency_key: format!("recurring:{}:{}", definition.id, now.timestamp()),
282 correlation_id: definition.id.clone(),
283 causation_id: definition.id.clone(),
284 trace_id: definition.id.clone(),
285 sttp_input_node_id: definition.payload_template_ref.clone(),
286 scheduled_at,
287 backoff_policy: Default::default(),
288 };
289
290 self.enqueue(job).await?;
291
292 definition.last_run_at = Some(now);
293 definition.next_run_at = definition.compute_next_run_at(now)?;
294 definition.lease_owner = None;
295 definition.lease_expires_at = None;
296 self.recurring_store.save(definition).await?;
297 produced += 1;
298 }
299
300 Ok(produced)
301 }
302
303 pub async fn process_once(
304 &self,
305 queue: &str,
306 worker_id: &str,
307 now: DateTime<Utc>,
308 ) -> Result<Option<String>> {
309 let Some(mut job) = self.job_store.lease_due(queue, worker_id, now, 30).await? else {
310 return Ok(None);
311 };
312
313 job.state = JobState::Running;
314 job.started_at = job.started_at.or(Some(now));
315 job.heartbeat_at = Some(now);
316 let job_identity = RuntimeJobIdentityContext::from(&job);
317 self.job_store.save(job.clone()).await?;
318 let processing_started = Instant::now();
319
320 let handler = {
321 let handlers = self
322 .handlers
323 .read()
324 .map_err(|_| StasisError::PortFailure("handlers lock poisoned".to_string()))?;
325 handlers.get(&job.job_type).cloned()
326 };
327
328 let outcome = if let Some(handler) = handler {
329 handler.execute(&job).await?
330 } else {
331 JobExecutionOutcome::FatalFailure {
332 message: format!("no handler registered for job_type={}", job.job_type),
333 execution_id: None,
334 diagnostics: None,
335 }
336 };
337
338 let attempt_number = job.attempts + 1;
339 let attempt_started_at = now;
340
341 match outcome {
342 JobExecutionOutcome::Success {
343 sttp_output_node_id,
344 execution_id,
345 diagnostics,
346 } => {
347 let diagnostics_envelope =
348 Self::extract_diagnostics_envelope(diagnostics.as_deref());
349 job.state = JobState::Succeeded;
350 job.sttp_output_node_id = Some(sttp_output_node_id.clone());
351 job.finished_at = Some(now);
352 job.lease_owner = None;
353 job.lease_expires_at = None;
354 job.heartbeat_at = None;
355 self.job_store.save(job).await?;
356
357 self.append_outbox(
358 RuntimeEventType::JobSucceeded,
359 &job_identity,
360 Some(sttp_output_node_id.clone()),
361 None,
362 now,
363 execution_id.clone(),
364 &diagnostics_envelope,
365 )
366 .await?;
367
368 self.append_job_attempt(
369 &job_identity.job_id,
370 worker_id,
371 attempt_number,
372 attempt_started_at,
373 now,
374 JobAttemptOutcome::Succeeded,
375 None,
376 Some(sttp_output_node_id),
377 execution_id,
378 &diagnostics_envelope,
379 diagnostics,
380 )
381 .await?;
382
383 self.metrics.incr_counter(METRIC_JOB_SUCCEEDED_TOTAL, 1);
384 self.metrics.observe_duration_ms(
385 METRIC_JOB_PROCESS_DURATION_MS,
386 processing_started.elapsed().as_millis() as u64,
387 );
388 }
389 JobExecutionOutcome::RetryableFailure {
390 message,
391 execution_id,
392 diagnostics,
393 } => {
394 let diagnostics_envelope =
395 Self::extract_diagnostics_envelope(diagnostics.as_deref());
396 let guardrail_failure = diagnostics
397 .as_deref()
398 .map(|v| v.contains("\"guardrail_code\""))
399 .unwrap_or(false);
400 job.attempts += 1;
401 job.last_error = Some(message.clone());
402 job.lease_owner = None;
403 job.lease_expires_at = None;
404 job.heartbeat_at = None;
405
406 if job.attempts >= job.max_attempts {
407 job.state = JobState::DeadLetter;
408 job.finished_at = Some(now);
409 self.append_outbox(
410 RuntimeEventType::JobDeadLettered,
411 &job_identity,
412 None,
413 Some(message.clone()),
414 now,
415 execution_id.clone(),
416 &diagnostics_envelope,
417 )
418 .await?;
419
420 self.metrics.incr_counter(METRIC_JOB_DEAD_LETTER_TOTAL, 1);
421 } else {
422 job.state = JobState::Enqueued;
423 let exponent = job.attempts - 1;
424 let mut delay = job
425 .backoff_policy
426 .base_delay_seconds
427 .saturating_mul(2_i64.saturating_pow(exponent));
428 delay = delay.min(job.backoff_policy.max_delay_seconds);
429 job.scheduled_at = now + Duration::seconds(delay.max(0));
430
431 self.append_outbox(
432 RuntimeEventType::JobRetryScheduled,
433 &job_identity,
434 None,
435 Some(message.clone()),
436 now,
437 execution_id.clone(),
438 &diagnostics_envelope,
439 )
440 .await?;
441
442 self.metrics
443 .incr_counter(METRIC_JOB_RETRY_SCHEDULED_TOTAL, 1);
444 }
445
446 self.job_store.save(job).await?;
447
448 self.append_job_attempt(
449 &job_identity.job_id,
450 worker_id,
451 attempt_number,
452 attempt_started_at,
453 now,
454 JobAttemptOutcome::RetryableFailure,
455 Some(message),
456 None,
457 execution_id,
458 &diagnostics_envelope,
459 diagnostics,
460 )
461 .await?;
462
463 self.metrics
464 .incr_counter(METRIC_JOB_RETRYABLE_FAILURE_TOTAL, 1);
465 self.metrics.observe_duration_ms(
466 METRIC_JOB_PROCESS_DURATION_MS,
467 processing_started.elapsed().as_millis() as u64,
468 );
469 if guardrail_failure {
470 self.metrics
471 .incr_counter(METRIC_GRAPHEME_GUARDRAIL_FAILURE_TOTAL, 1);
472 }
473 }
474 JobExecutionOutcome::FatalFailure {
475 message,
476 execution_id,
477 diagnostics,
478 } => {
479 let diagnostics_envelope =
480 Self::extract_diagnostics_envelope(diagnostics.as_deref());
481 let guardrail_failure = diagnostics
482 .as_deref()
483 .map(|v| v.contains("\"guardrail_code\""))
484 .unwrap_or(false);
485 job.attempts += 1;
486 job.state = JobState::DeadLetter;
487 job.last_error = Some(message.clone());
488 job.finished_at = Some(now);
489 job.lease_owner = None;
490 job.lease_expires_at = None;
491 job.heartbeat_at = None;
492 self.job_store.save(job).await?;
493
494 self.append_outbox(
495 RuntimeEventType::JobDeadLettered,
496 &job_identity,
497 None,
498 Some(message.clone()),
499 now,
500 execution_id.clone(),
501 &diagnostics_envelope,
502 )
503 .await?;
504
505 self.append_job_attempt(
506 &job_identity.job_id,
507 worker_id,
508 attempt_number,
509 attempt_started_at,
510 now,
511 JobAttemptOutcome::FatalFailure,
512 Some(message),
513 None,
514 execution_id,
515 &diagnostics_envelope,
516 diagnostics,
517 )
518 .await?;
519
520 self.metrics.incr_counter(METRIC_JOB_FATAL_FAILURE_TOTAL, 1);
521 self.metrics.incr_counter(METRIC_JOB_DEAD_LETTER_TOTAL, 1);
522 self.metrics.observe_duration_ms(
523 METRIC_JOB_PROCESS_DURATION_MS,
524 processing_started.elapsed().as_millis() as u64,
525 );
526 if guardrail_failure {
527 self.metrics
528 .incr_counter(METRIC_GRAPHEME_GUARDRAIL_FAILURE_TOTAL, 1);
529 }
530 }
531 }
532
533 Ok(Some(job_identity.job_id))
534 }
535
536 pub async fn replay_dead_letter(&self, job_id: &str, now: DateTime<Utc>) -> Result<bool> {
537 let Some(mut job) = self.job_store.get(job_id).await? else {
538 return Ok(false);
539 };
540
541 if job.state != JobState::DeadLetter {
542 return Ok(false);
543 }
544
545 job.state = JobState::Enqueued;
546 job.attempts = 0;
547 job.last_error = None;
548 job.scheduled_at = now;
549 job.lease_owner = None;
550 job.lease_expires_at = None;
551 job.heartbeat_at = None;
552 job.finished_at = None;
553
554 self.job_store.save(job).await?;
555 Ok(true)
556 }
557
558 pub async fn publish_pending_events(&self, limit: usize, now: DateTime<Utc>) -> Result<usize> {
559 let publisher = {
560 let state = self
561 .publisher
562 .read()
563 .map_err(|_| StasisError::PortFailure("publisher lock poisoned".to_string()))?;
564 state.clone()
565 };
566
567 let Some(publisher) = publisher else {
568 return Ok(0);
569 };
570
571 let policy = self
572 .publish_policy
573 .read()
574 .map_err(|_| StasisError::PortFailure("publish policy lock poisoned".to_string()))?
575 .clone();
576
577 let pending = self.outbox_store.list_pending(limit).await?;
578 let mut published = 0usize;
579
580 for mut event in pending {
581 if event
582 .next_attempt_at
583 .map(|next| next > now)
584 .unwrap_or(false)
585 {
586 continue;
587 }
588
589 match publisher.publish(&event).await {
590 Ok(()) => {
591 event.status = OutboxStatus::Published;
592 event.publish_attempts = event.publish_attempts.saturating_add(1);
593 event.published_at = Some(now);
594 event.next_attempt_at = None;
595 event.last_publish_error = None;
596 self.outbox_store.save(event).await?;
597 published += 1;
598 self.metrics
599 .incr_counter(METRIC_OUTBOX_PUBLISH_SUCCESS_TOTAL, 1);
600 }
601 Err(err) => {
602 event.publish_attempts = event.publish_attempts.saturating_add(1);
603 event.published_at = None;
604 event.last_publish_error = Some(err.to_string());
605
606 if event.publish_attempts >= policy.max_attempts {
607 event.status = OutboxStatus::Failed;
608 event.next_attempt_at = None;
609 } else {
610 let exponent = event.publish_attempts - 1;
611 let mut delay = policy
612 .base_delay_seconds
613 .saturating_mul(2_i64.saturating_pow(exponent));
614 delay = delay.min(policy.max_delay_seconds);
615 event.status = OutboxStatus::Pending;
616 event.next_attempt_at = Some(now + Duration::seconds(delay.max(0)));
617 }
618
619 self.outbox_store.save(event).await?;
620 self.metrics
621 .incr_counter(METRIC_OUTBOX_PUBLISH_FAILURE_TOTAL, 1);
622 }
623 }
624 }
625
626 Ok(published)
627 }
628
629 #[allow(clippy::too_many_arguments)]
630 async fn append_outbox(
631 &self,
632 event_type: RuntimeEventType,
633 job_identity: &RuntimeJobIdentityContext,
634 sttp_output_node_id: Option<String>,
635 message: Option<String>,
636 now: DateTime<Utc>,
637 execution_id: Option<String>,
638 diagnostics: &runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope,
639 ) -> Result<()> {
640 let event = OutboxEvent {
641 event_id: self
642 .id_generator
643 .next_id(&format!("evt-{}", job_identity.job_id)),
644 status: OutboxStatus::Pending,
645 publish_attempts: 0,
646 published_at: None,
647 next_attempt_at: None,
648 last_publish_error: None,
649 event: RuntimeEvent {
650 event_type,
651 job_id: job_identity.job_id.clone(),
652 thread_id: diagnostics.thread_id.clone(),
653 correlation_id: job_identity.correlation_id.clone(),
654 causation_id: job_identity.causation_id.clone(),
655 trace_id: job_identity.trace_id.clone(),
656 sttp_input_node_id: job_identity.sttp_input_node_id.clone(),
657 sttp_output_node_id,
658 execution_id,
659 input_memory_query_id: diagnostics.input_memory_query_id.clone(),
660 input_memory_query_fingerprint: diagnostics
661 .input_memory_query_fingerprint
662 .clone(),
663 output_memory_node_id: diagnostics.output_memory_node_id.clone(),
664 retrieval_path: diagnostics.retrieval_path.clone(),
665 occurred_at: now,
666 message,
667 },
668 };
669
670 self.outbox_store.insert(event).await
671 }
672
673 #[allow(clippy::too_many_arguments)]
674 async fn append_job_attempt(
675 &self,
676 job_id: &str,
677 worker_id: &str,
678 attempt_number: u32,
679 started_at: DateTime<Utc>,
680 finished_at: DateTime<Utc>,
681 outcome: JobAttemptOutcome,
682 error_message: Option<String>,
683 sttp_output_node_id: Option<String>,
684 execution_id: Option<String>,
685 diagnostics_envelope: &runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope,
686 diagnostics: Option<String>,
687 ) -> Result<()> {
688 let attempt = JobAttempt {
689 attempt_id: self.id_generator.next_id(&format!("attempt-{job_id}")),
690 job_id: job_id.to_string(),
691 attempt_number,
692 worker_id: worker_id.to_string(),
693 started_at,
694 finished_at,
695 outcome,
696 error_message,
697 sttp_output_node_id,
698 execution_id,
699 guardrail_code: diagnostics_envelope.guardrail_code.clone(),
700 policy_reason: diagnostics_envelope.policy_reason.clone(),
701 duration_ms: diagnostics_envelope.duration_ms,
702 diagnostics,
703 };
704
705 self.job_attempt_store.insert(attempt).await
706 }
707
708 fn extract_diagnostics_envelope(
709 diagnostics: Option<&str>,
710 ) -> runtime_diagnostics_helpers::RuntimeDiagnosticsEnvelope {
711 runtime_diagnostics_helpers::extract_runtime_diagnostics_envelope(diagnostics)
712 }
713
714}