starweaver_session/
background.rs

1//! Product-neutral durable background-subagent execution and delivery records.
2
3use chrono::{DateTime, Utc};
4use serde::{Deserialize, Serialize};
5use sha2::{Digest, Sha256};
6use starweaver_core::{RunId, SessionId, SubagentAttemptId, TaskId, TraceContext};
7
8use crate::{AcquireRunAdmission, InputPart, RunAdmissionReceipt, RunRecord, RunStatus};
9
10/// Current durable background-subagent record schema.
11pub const BACKGROUND_SUBAGENT_RECORD_VERSION: u32 = 1;
12/// Default retained-result lifetime used by store-side interruption reconciliation.
13pub const DEFAULT_BACKGROUND_RESULT_RETENTION_SECS: i64 = 86_400;
14
15/// Durable host-owned oversized result artifact.
16#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
17pub struct BackgroundSubagentArtifact {
18    /// Stable artifact reference stored in the terminal result record.
19    pub artifact_ref: String,
20    /// Owning durable namespace.
21    pub namespace_id: String,
22    /// Attempt whose successful output was externalized.
23    pub attempt_id: SubagentAttemptId,
24    /// Complete successful output bytes encoded as UTF-8 text.
25    pub content: String,
26    /// SHA-256 digest of the complete logical result.
27    pub digest: String,
28    /// Complete logical result size in bytes.
29    pub size_bytes: u64,
30    /// Artifact creation timestamp.
31    pub created_at: DateTime<Utc>,
32    /// Policy-controlled retention deadline.
33    pub expires_at: DateTime<Utc>,
34}
35
36impl starweaver_core::VersionedRecord for BackgroundSubagentArtifact {
37    const SCHEMA: &'static str = "starweaver.session.background_subagent_artifact";
38    const VERSION: u32 = 1;
39}
40
41impl BackgroundSubagentArtifact {
42    /// Compute the domain-separated digest used for successful result content.
43    #[must_use]
44    pub fn content_digest(content: &str) -> String {
45        background_subagent_result_digest(Some(content), None)
46    }
47
48    /// Return whether the artifact matches its durable record identity and content.
49    #[must_use]
50    pub fn is_valid(&self) -> bool {
51        !self.artifact_ref.is_empty()
52            && !self.namespace_id.is_empty()
53            && self.digest == Self::content_digest(&self.content)
54            && self.size_bytes == u64::try_from(self.content.len()).unwrap_or(u64::MAX)
55            && self.expires_at > self.created_at
56    }
57
58    /// Return whether the artifact is valid and still retained at `now`.
59    #[must_use]
60    pub fn is_available_at(&self, now: DateTime<Utc>) -> bool {
61        self.is_valid() && self.expires_at > now
62    }
63}
64
65/// Host policy limits applied atomically to retained result artifacts.
66#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
67pub struct BackgroundSubagentArtifactLimits {
68    /// Maximum bytes accepted for one artifact.
69    pub max_single_bytes: u64,
70    /// Maximum aggregate unexpired artifact bytes in one durable namespace.
71    pub max_retained_bytes: u64,
72}
73
74impl BackgroundSubagentArtifactLimits {
75    /// Return whether the configured limits are non-zero and internally ordered.
76    #[must_use]
77    pub const fn is_valid(self) -> bool {
78        self.max_single_bytes > 0
79            && self.max_retained_bytes > 0
80            && self.max_single_bytes <= self.max_retained_bytes
81    }
82}
83
84/// Atomic terminal evidence and optional oversized-result artifact commit.
85#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
86pub struct BackgroundSubagentTerminalCommit {
87    /// Immutable terminal execution and delivery record.
88    pub record: BackgroundSubagentRecord,
89    /// Optional complete successful output externalized from the bounded preview.
90    #[serde(default, skip_serializing_if = "Option::is_none")]
91    pub artifact: Option<BackgroundSubagentArtifact>,
92    /// Required host quota policy whenever an artifact payload is supplied.
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub artifact_limits: Option<BackgroundSubagentArtifactLimits>,
95}
96
97#[derive(Serialize)]
98struct CanonicalBackgroundSubagentTerminalCommit<'a> {
99    schema_version: u32,
100    attempt_id: &'a SubagentAttemptId,
101    agent_id: &'a str,
102    linked_task_id: &'a Option<TaskId>,
103    subagent_name: &'a str,
104    namespace_id: &'a str,
105    parent_session_id: &'a SessionId,
106    parent_run_id: &'a RunId,
107    child_run_id: &'a Option<RunId>,
108    profile: &'a str,
109    owner_host_instance_id: &'a str,
110    owner_fencing_generation: u64,
111    execution_status: DurableBackgroundSubagentExecutionStatus,
112    result_ref: &'a Option<DurableBackgroundSubagentResultRef>,
113    failure_category: &'a Option<String>,
114    cancellation_reason: &'a Option<String>,
115    initial_retention_status: DurableBackgroundSubagentRetentionStatus,
116    initial_retention_expires_at: &'a Option<DateTime<Utc>>,
117    trace_context: &'a Option<TraceContext>,
118    accepted_at: &'a DateTime<Utc>,
119    terminal_at: &'a Option<DateTime<Utc>>,
120    updated_at: &'a DateTime<Utc>,
121    artifact: &'a Option<BackgroundSubagentArtifact>,
122}
123
124impl BackgroundSubagentTerminalCommit {
125    /// Compute the canonical fingerprint persisted for exact terminal retries.
126    ///
127    /// Logical delivery fields and their later projections are deliberately excluded. Initial
128    /// retention evidence and the complete artifact payload remain covered even after content is
129    /// expired from the live record.
130    ///
131    /// # Errors
132    ///
133    /// Returns a serialization error when the canonical evidence cannot be encoded.
134    pub fn canonical_fingerprint(&self) -> Result<String, serde_json::Error> {
135        let record = &self.record;
136        let canonical = CanonicalBackgroundSubagentTerminalCommit {
137            schema_version: record.schema_version,
138            attempt_id: &record.attempt_id,
139            agent_id: &record.agent_id,
140            linked_task_id: &record.linked_task_id,
141            subagent_name: &record.subagent_name,
142            namespace_id: &record.namespace_id,
143            parent_session_id: &record.parent_session_id,
144            parent_run_id: &record.parent_run_id,
145            child_run_id: &record.child_run_id,
146            profile: &record.profile,
147            owner_host_instance_id: &record.owner_lease.host_instance_id,
148            owner_fencing_generation: record.owner_lease.fencing_generation,
149            execution_status: record.execution_status,
150            result_ref: &record.result_ref,
151            failure_category: &record.failure_category,
152            cancellation_reason: &record.cancellation_reason,
153            initial_retention_status: record.retention_status,
154            initial_retention_expires_at: &record.retention_expires_at,
155            trace_context: &record.trace_context,
156            accepted_at: &record.accepted_at,
157            terminal_at: &record.terminal_at,
158            updated_at: &record.updated_at,
159            artifact: &self.artifact,
160        };
161        let payload = serde_json::to_vec(&canonical)?;
162        let mut hasher = Sha256::new();
163        hasher.update(b"starweaver.session.background_subagent.terminal_commit.v1\0");
164        hasher.update(payload);
165        Ok(format!("sha256:{:x}", hasher.finalize()))
166    }
167}
168
169/// Monotonic durable execution state for one background attempt.
170#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
171#[serde(rename_all = "snake_case")]
172pub enum DurableBackgroundSubagentExecutionStatus {
173    /// Identity, quota, ownership, and durable admission exist.
174    Accepted,
175    /// Child construction is in progress.
176    Starting,
177    /// Child execution can make progress.
178    Running,
179    /// The same attempt is waiting for an explicitly resumable condition.
180    Waiting,
181    /// Child execution completed successfully.
182    Completed,
183    /// Child execution failed or was interrupted after process loss.
184    Failed,
185    /// Child execution was cancelled.
186    Cancelled,
187}
188
189impl DurableBackgroundSubagentExecutionStatus {
190    /// Return whether this is an immutable terminal outcome.
191    #[must_use]
192    pub const fn is_terminal(self) -> bool {
193        matches!(self, Self::Completed | Self::Failed | Self::Cancelled)
194    }
195
196    /// Return the stable wire-format status name.
197    #[must_use]
198    pub const fn as_str(self) -> &'static str {
199        match self {
200            Self::Accepted => "accepted",
201            Self::Starting => "starting",
202            Self::Running => "running",
203            Self::Waiting => "waiting",
204            Self::Completed => "completed",
205            Self::Failed => "failed",
206            Self::Cancelled => "cancelled",
207        }
208    }
209}
210
211/// Orthogonal durable logical-delivery state.
212#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
213#[serde(rename_all = "snake_case")]
214pub enum DurableBackgroundSubagentDeliveryStatus {
215    /// Terminal content exists but no consumer owns it.
216    #[default]
217    Undelivered,
218    /// One bounded consumer owns delivery.
219    Claimed,
220    /// A parent turn, explicit wait, or admitted continuation consumed it.
221    Delivered,
222}
223
224/// Durable content-retention state, independent from execution and delivery.
225#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
226#[serde(rename_all = "snake_case")]
227pub enum DurableBackgroundSubagentRetentionStatus {
228    /// Bounded terminal content is stored inline.
229    #[default]
230    Inline,
231    /// Content is represented by a host-owned artifact reference.
232    Artifact,
233    /// Volatile content expired while minimal audit evidence remains.
234    Expired,
235}
236
237impl DurableBackgroundSubagentRetentionStatus {
238    /// Stable storage and protocol name.
239    #[must_use]
240    pub const fn as_str(self) -> &'static str {
241        match self {
242            Self::Inline => "inline",
243            Self::Artifact => "artifact",
244            Self::Expired => "expired",
245        }
246    }
247}
248
249/// Bounded terminal result material retained by the durable host.
250#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
251pub struct DurableBackgroundSubagentResultRef {
252    /// Bounded successful text content.
253    #[serde(default, skip_serializing_if = "Option::is_none")]
254    pub content: Option<String>,
255    /// Bounded safe error text.
256    #[serde(default, skip_serializing_if = "Option::is_none")]
257    pub error: Option<String>,
258    /// Optional host artifact locator for oversized content.
259    #[serde(default, skip_serializing_if = "Option::is_none")]
260    pub artifact_ref: Option<String>,
261    /// Stable digest of the retained logical content when available.
262    #[serde(default, skip_serializing_if = "Option::is_none")]
263    pub digest: Option<String>,
264    /// Logical content size before host retention policy was applied.
265    #[serde(default)]
266    pub size_bytes: u64,
267}
268
269/// Atomic durable delivery claim.
270#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
271pub struct DurableBackgroundSubagentDeliveryClaim {
272    /// Stable consumer-generated claim identity.
273    pub claim_id: String,
274    /// Optional parent continuation admitted under this claim.
275    #[serde(default, skip_serializing_if = "Option::is_none")]
276    pub continuation_run_id: Option<RunId>,
277    /// Absolute claim expiry.
278    pub deadline: DateTime<Utc>,
279}
280
281/// Durable reason for releasing a matching result-delivery claim.
282#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
283#[serde(tag = "kind", rename_all = "snake_case")]
284pub enum DurableBackgroundSubagentDeliveryRelease {
285    /// Admission or another retryable pre-consumption step failed.
286    #[default]
287    Retryable,
288    /// A parent run claimed the result but failed or was cancelled before committing it.
289    ConsumerTerminated {
290        /// Failed or cancelled run that owned the released claim.
291        run_id: RunId,
292    },
293}
294
295/// Durable fenced owner lease for one process-local background execution.
296#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
297pub struct DurableBackgroundSubagentOwnerLease {
298    /// Service instance currently authorized to advance this attempt.
299    pub host_instance_id: String,
300    /// Monotonic fencing generation for ownership-sensitive writes.
301    pub fencing_generation: u64,
302    /// Last successful owner heartbeat.
303    pub heartbeat_at: DateTime<Utc>,
304    /// Time after which reconciliation may classify the process-local execution as lost.
305    pub lease_expires_at: DateTime<Utc>,
306}
307
308impl DurableBackgroundSubagentOwnerLease {
309    /// Return whether this owner lease is structurally valid.
310    #[must_use]
311    pub fn is_valid(&self) -> bool {
312        !self.host_instance_id.is_empty()
313            && self.fencing_generation > 0
314            && self.lease_expires_at > self.heartbeat_at
315    }
316
317    /// Return whether the lease is expired at `now`.
318    #[must_use]
319    pub fn expired_at(&self, now: DateTime<Utc>) -> bool {
320        self.lease_expires_at <= now
321    }
322}
323
324/// Durable service-host projection of one background attempt.
325#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
326pub struct BackgroundSubagentRecord {
327    /// Version of this serialized record contract.
328    pub schema_version: u32,
329    /// Per-execution attempt identity.
330    pub attempt_id: SubagentAttemptId,
331    /// Stable child conversation identity within the owning supervisor scope.
332    pub agent_id: String,
333    /// Optional independently owned task-bundle identity.
334    #[serde(default, skip_serializing_if = "Option::is_none")]
335    pub linked_task_id: Option<TaskId>,
336    /// Configured subagent name.
337    pub subagent_name: String,
338    /// Durable host namespace.
339    pub namespace_id: String,
340    /// Owning parent session.
341    pub parent_session_id: SessionId,
342    /// Parent run that accepted delegation.
343    pub parent_run_id: RunId,
344    /// Child runtime run once known.
345    #[serde(default, skip_serializing_if = "Option::is_none")]
346    pub child_run_id: Option<RunId>,
347    /// Parent continuation that consumed the result, when any.
348    #[serde(default, skip_serializing_if = "Option::is_none")]
349    pub continuation_run_id: Option<RunId>,
350    /// Resolved parent profile used for an automatic continuation.
351    pub profile: String,
352    /// Fenced process owner for all non-terminal execution writes.
353    pub owner_lease: DurableBackgroundSubagentOwnerLease,
354    /// Current monotonic execution state.
355    pub execution_status: DurableBackgroundSubagentExecutionStatus,
356    /// Terminal result reference, separate from minimal lifecycle evidence.
357    #[serde(default, skip_serializing_if = "Option::is_none")]
358    pub result_ref: Option<DurableBackgroundSubagentResultRef>,
359    /// Safe failure/interruption category.
360    #[serde(default, skip_serializing_if = "Option::is_none")]
361    pub failure_category: Option<String>,
362    /// Bounded cancellation reason.
363    #[serde(default, skip_serializing_if = "Option::is_none")]
364    pub cancellation_reason: Option<String>,
365    /// Logical result-delivery state.
366    pub delivery_status: DurableBackgroundSubagentDeliveryStatus,
367    /// Current bounded claim, when delivery is claimed.
368    #[serde(default, skip_serializing_if = "Option::is_none")]
369    pub delivery_claim: Option<DurableBackgroundSubagentDeliveryClaim>,
370    /// Claim that completed logical delivery.
371    #[serde(default, skip_serializing_if = "Option::is_none")]
372    pub delivered_claim_id: Option<String>,
373    /// Failed or cancelled consumer that suppresses automatic continuation redelivery.
374    ///
375    /// Explicit later parent runs may still claim and consume the pending result.
376    #[serde(default, skip_serializing_if = "Option::is_none")]
377    pub automatic_continuation_suppressed_by_run_id: Option<RunId>,
378    /// Independent content-retention state.
379    pub retention_status: DurableBackgroundSubagentRetentionStatus,
380    /// Policy deadline for retained inline or artifact content.
381    #[serde(default, skip_serializing_if = "Option::is_none")]
382    pub retention_expires_at: Option<DateTime<Utc>>,
383    /// Parent trace correlation copied at acceptance.
384    #[serde(default, skip_serializing_if = "Option::is_none")]
385    pub trace_context: Option<TraceContext>,
386    /// Acceptance time.
387    pub accepted_at: DateTime<Utc>,
388    /// Last durable transition time.
389    pub updated_at: DateTime<Utc>,
390    /// Immutable terminal transition time.
391    #[serde(default, skip_serializing_if = "Option::is_none")]
392    pub terminal_at: Option<DateTime<Utc>>,
393}
394
395impl starweaver_core::VersionedRecord for BackgroundSubagentRecord {
396    const SCHEMA: &'static str = "starweaver.session.background_subagent_record";
397    const VERSION: u32 = BACKGROUND_SUBAGENT_RECORD_VERSION;
398}
399
400impl BackgroundSubagentRecord {
401    /// Return whether this record is a canonical acceptance projection.
402    #[must_use]
403    pub fn is_valid_acceptance(&self) -> bool {
404        self.execution_status == DurableBackgroundSubagentExecutionStatus::Accepted
405            && self.owner_lease.is_valid()
406            && self.owner_lease.heartbeat_at == self.accepted_at
407            && self.child_run_id.is_none()
408            && self.result_ref.is_none()
409            && self.failure_category.is_none()
410            && self.cancellation_reason.is_none()
411            && self.delivery_status == DurableBackgroundSubagentDeliveryStatus::Undelivered
412            && self.delivery_claim.is_none()
413            && self.delivered_claim_id.is_none()
414            && self.continuation_run_id.is_none()
415            && self.automatic_continuation_suppressed_by_run_id.is_none()
416            && self.retention_status == DurableBackgroundSubagentRetentionStatus::Inline
417            && self.retention_expires_at.is_none()
418            && self.terminal_at.is_none()
419            && self.updated_at == self.accepted_at
420    }
421
422    /// Return whether this record is a canonical immutable terminal projection.
423    #[must_use]
424    pub fn is_valid_terminal(&self) -> bool {
425        self.execution_status.is_terminal()
426            && self.owner_lease.is_valid()
427            && self.result_ref.is_some()
428            && self.delivery_status == DurableBackgroundSubagentDeliveryStatus::Undelivered
429            && self.delivery_claim.is_none()
430            && self.delivered_claim_id.is_none()
431            && self.continuation_run_id.is_none()
432            && self.automatic_continuation_suppressed_by_run_id.is_none()
433            && match self.retention_status {
434                DurableBackgroundSubagentRetentionStatus::Expired => {
435                    self.retention_expires_at.is_none()
436                        && self.result_ref.as_ref().is_some_and(|result| {
437                            result.content.is_none()
438                                && result.error.is_none()
439                                && result.artifact_ref.is_none()
440                        })
441                }
442                _ => self
443                    .retention_expires_at
444                    .is_some_and(|deadline| deadline > self.updated_at),
445            }
446            && self.terminal_at == Some(self.updated_at)
447            && self.updated_at >= self.accepted_at
448            && (self.execution_status != DurableBackgroundSubagentExecutionStatus::Completed
449                || (self.failure_category.is_none() && self.cancellation_reason.is_none()))
450            && (self.cancellation_reason.is_none()
451                || self.execution_status == DurableBackgroundSubagentExecutionStatus::Cancelled)
452    }
453
454    /// Return whether a claim may be acquired at `now`.
455    #[must_use]
456    pub fn delivery_claimable_at(&self, now: DateTime<Utc>) -> bool {
457        self.execution_status.is_terminal()
458            && (self.delivery_status == DurableBackgroundSubagentDeliveryStatus::Undelivered
459                || (self.delivery_status == DurableBackgroundSubagentDeliveryStatus::Claimed
460                    && self
461                        .delivery_claim
462                        .as_ref()
463                        .is_some_and(|claim| claim.deadline <= now)))
464    }
465
466    /// Resolve the canonical model-visible outcome for a continuation.
467    #[must_use]
468    pub fn continuation_outcome(&self, artifact_content: Option<&str>) -> String {
469        if self.retention_status == DurableBackgroundSubagentRetentionStatus::Expired {
470            let result_ref = self.result_ref.as_ref();
471            return format!(
472                "Retained background result content expired (digest: {}, logical_size_bytes: {}).",
473                result_ref
474                    .and_then(|result| result.digest.as_deref())
475                    .unwrap_or("unknown"),
476                result_ref.map_or(0, |result| result.size_bytes),
477            );
478        }
479        if let Some(content) = artifact_content {
480            return content.to_string();
481        }
482        self.result_ref.as_ref().map_or_else(
483            || {
484                format!(
485                    "Background subagent finished with status {}.",
486                    self.execution_status.as_str()
487                )
488            },
489            |result| {
490                result
491                    .content
492                    .clone()
493                    .or_else(|| result.error.clone())
494                    .unwrap_or_else(|| {
495                        format!(
496                            "Background subagent finished with status {}.",
497                            self.execution_status.as_str()
498                        )
499                    })
500            },
501        )
502    }
503
504    /// Build the canonical model-visible continuation text.
505    #[must_use]
506    pub fn continuation_text(&self, artifact_content: Option<&str>) -> String {
507        format!(
508            "Background subagent result (subagent: {}, agent_id: {}, attempt_id: {}, child_run_id: {}):\n\n{}",
509            self.subagent_name,
510            self.agent_id,
511            self.attempt_id.as_str(),
512            self.child_run_id.as_ref().map_or("unknown", RunId::as_str),
513            self.continuation_outcome(artifact_content),
514        )
515    }
516
517    /// Build the exact durable input accepted for a result-triggered continuation.
518    #[must_use]
519    pub fn continuation_input(&self, artifact_content: Option<&str>) -> Vec<InputPart> {
520        vec![InputPart::text(self.continuation_text(artifact_content))]
521    }
522
523    /// Validate immutable causal fields on a proposed result-triggered continuation.
524    #[must_use]
525    pub fn validates_continuation_run(&self, run: &RunRecord) -> bool {
526        run.session_id == self.parent_session_id
527            && run.status == RunStatus::Queued
528            && run.parent_run_id.as_ref() == Some(&self.parent_run_id)
529            && run.trigger_type.as_deref() == Some("async_subagent_result")
530            && run.profile.as_deref() == Some(self.profile.as_str())
531            && run
532                .metadata
533                .get("starweaver.async_subagent.attempt_id")
534                .and_then(serde_json::Value::as_str)
535                == Some(self.attempt_id.as_str())
536            && run
537                .metadata
538                .get("starweaver.async_subagent.agent_id")
539                .and_then(serde_json::Value::as_str)
540                == Some(self.agent_id.as_str())
541            && run
542                .metadata
543                .get("starweaver.async_subagent.parent_run_id")
544                .and_then(serde_json::Value::as_str)
545                == Some(self.parent_run_id.as_str())
546            && self.child_run_id.as_ref().map_or_else(
547                || {
548                    !run.metadata
549                        .contains_key("starweaver.async_subagent.child_run_id")
550                },
551                |child_run_id| {
552                    run.metadata
553                        .get("starweaver.async_subagent.child_run_id")
554                        .and_then(serde_json::Value::as_str)
555                        == Some(child_run_id.as_str())
556                },
557            )
558            && run.trace_context == self.trace_context.clone().unwrap_or_default()
559    }
560
561    /// Validate the typed causal envelope and its submitted input digest.
562    #[must_use]
563    pub fn validates_continuation_cause_envelope(
564        &self,
565        cause: &BackgroundSubagentContinuationCause,
566        run: &RunRecord,
567    ) -> bool {
568        let Ok(input_digest) = background_subagent_input_digest(&run.input) else {
569            return false;
570        };
571        let result_ref = self.result_ref.as_ref();
572        self.validates_continuation_run(run)
573            && cause.attempt_id == self.attempt_id
574            && cause.agent_id == self.agent_id
575            && cause.parent_session_id == self.parent_session_id
576            && cause.parent_run_id == self.parent_run_id
577            && cause.child_run_id == self.child_run_id
578            && cause.result_digest == result_ref.and_then(|result| result.digest.clone())
579            && cause.result_size_bytes == result_ref.map_or(0, |result| result.size_bytes)
580            && cause.trace_context == self.trace_context.clone().unwrap_or_default()
581            && cause.input_digest == input_digest
582    }
583
584    /// Validate a typed cause and exact canonical input against this durable result.
585    #[must_use]
586    pub fn validates_continuation_cause(
587        &self,
588        cause: &BackgroundSubagentContinuationCause,
589        run: &RunRecord,
590        artifact_content: Option<&str>,
591    ) -> bool {
592        let artifact_shape_matches = match self.retention_status {
593            DurableBackgroundSubagentRetentionStatus::Artifact => artifact_content.is_some(),
594            DurableBackgroundSubagentRetentionStatus::Inline
595            | DurableBackgroundSubagentRetentionStatus::Expired => artifact_content.is_none(),
596        };
597        if !artifact_shape_matches || !self.validates_continuation_cause_envelope(cause, run) {
598            return false;
599        }
600        run.input == self.continuation_input(artifact_content)
601    }
602}
603
604/// Typed immutable cause for one result-triggered continuation.
605#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
606pub struct BackgroundSubagentContinuationCause {
607    /// Terminal attempt whose result caused the continuation.
608    pub attempt_id: SubagentAttemptId,
609    /// Stable child conversation identity.
610    pub agent_id: String,
611    /// Owning parent session.
612    pub parent_session_id: SessionId,
613    /// Parent run that accepted the delegation.
614    pub parent_run_id: RunId,
615    /// Child runtime run, when one was created.
616    #[serde(default, skip_serializing_if = "Option::is_none")]
617    pub child_run_id: Option<RunId>,
618    /// Digest of the complete logical terminal result, when available.
619    #[serde(default, skip_serializing_if = "Option::is_none")]
620    pub result_digest: Option<String>,
621    /// Complete logical terminal-result size.
622    pub result_size_bytes: u64,
623    /// Parent trace context inherited by the continuation.
624    #[serde(default, skip_serializing_if = "TraceContext::is_empty")]
625    pub trace_context: TraceContext,
626    /// Digest of the exact canonical durable continuation input.
627    pub input_digest: String,
628}
629
630impl BackgroundSubagentContinuationCause {
631    /// Build the canonical cause for a record and already-resolved continuation input.
632    ///
633    /// # Errors
634    ///
635    /// Returns a serialization error when the canonical durable input cannot be encoded.
636    pub fn new(
637        record: &BackgroundSubagentRecord,
638        input: &[InputPart],
639    ) -> Result<Self, serde_json::Error> {
640        Ok(Self {
641            attempt_id: record.attempt_id.clone(),
642            agent_id: record.agent_id.clone(),
643            parent_session_id: record.parent_session_id.clone(),
644            parent_run_id: record.parent_run_id.clone(),
645            child_run_id: record.child_run_id.clone(),
646            result_digest: record
647                .result_ref
648                .as_ref()
649                .and_then(|result| result.digest.clone()),
650            result_size_bytes: record
651                .result_ref
652                .as_ref()
653                .map_or(0, |result| result.size_bytes),
654            trace_context: record.trace_context.clone().unwrap_or_default(),
655            input_digest: background_subagent_input_digest(input)?,
656        })
657    }
658}
659
660/// Compute the versioned domain-separated digest for one logical terminal result.
661#[must_use]
662pub fn background_subagent_result_digest(content: Option<&str>, error: Option<&str>) -> String {
663    let mut hasher = Sha256::new();
664    hasher.update(b"starweaver.session.background_subagent.result.v1\0");
665    match (content, error) {
666        (Some(content), _) => {
667            hasher.update(b"content\0");
668            hasher.update(content.as_bytes());
669        }
670        (None, Some(error)) => {
671            hasher.update(b"error\0");
672            hasher.update(error.as_bytes());
673        }
674        (None, None) => hasher.update(b"empty\0"),
675    }
676    format!("sha256:{:x}", hasher.finalize())
677}
678
679/// Compute a domain-separated SHA-256 digest of canonical durable input parts.
680///
681/// # Errors
682///
683/// Returns a serialization error when the canonical durable input cannot be encoded.
684pub fn background_subagent_input_digest(input: &[InputPart]) -> Result<String, serde_json::Error> {
685    let payload = serde_json::to_vec(input)?;
686    let mut hasher = Sha256::new();
687    hasher.update(b"starweaver.session.background_subagent_continuation.input.v1\0");
688    hasher.update(payload);
689    Ok(format!("sha256:{:x}", hasher.finalize()))
690}
691
692/// Atomic durable continuation admission request.
693#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
694pub struct AcquireBackgroundSubagentContinuation {
695    /// Terminal attempt whose logical result becomes the continuation input.
696    pub attempt_id: SubagentAttemptId,
697    /// Stable delivery claim identity.
698    pub claim_id: String,
699    /// Claim deadline retained for audit and conflict handling.
700    pub claim_deadline: DateTime<Utc>,
701    /// Typed immutable cause bound to the exact canonical continuation input.
702    pub cause: BackgroundSubagentContinuationCause,
703    /// Ordinary single-active-run admission request containing the continuation run.
704    pub admission: AcquireRunAdmission,
705}
706
707/// Atomic durable continuation admission receipt.
708#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
709pub struct BackgroundSubagentContinuationReceipt {
710    /// Store-derived typed cause attesting the admitted canonical input.
711    pub cause: BackgroundSubagentContinuationCause,
712    /// Background record after delivery ownership is linked to the continuation.
713    pub background: BackgroundSubagentRecord,
714    /// Ordinary durable run admission receipt.
715    pub admission: RunAdmissionReceipt,
716}
starweaver_session/background.rs

starweaver_session/
background.rs
