a3s_code_core/
execution_identity.rs

1//! Stable identities for replayable execution boundaries.
2//!
3//! The identity is deliberately content-addressed and domain-separated. It
4//! is suitable for deduplication and fencing, but it does not itself claim a
5//! lease or persist an outcome; those responsibilities remain with the
6//! caller's ledger.
7
8use serde::{Deserialize, Serialize};
9use sha2::{Digest, Sha256};
10use std::collections::BTreeMap;
11use thiserror::Error;
12
13pub const EXECUTION_IDENTITY_SCHEMA_V1: &str = "a3s.code.execution-identity.v1";
14pub const MODEL_CALL_IDENTITY_DOMAIN_V1: &str = "a3s.code.model-call.identity.v1";
15pub const TOOL_INVOCATION_IDENTITY_DOMAIN_V1: &str = "a3s.code.tool-invocation.identity.v1";
16pub const FLOW_DECISION_IDENTITY_DOMAIN_V1: &str = "a3s.code.flow-decision.identity.v1";
17/// Identity domain for a dynamically admitted A3S Flow step.
18///
19/// Dynamic Flow steps are intentionally separate from delegated Agent steps:
20/// the former are identified by the Flow run/step/name and JSON input, while
21/// the latter are identified by an [`AgentStepSpec`](crate::orchestration::AgentStepSpec).
22pub const FLOW_STEP_IDENTITY_DOMAIN_V1: &str = "a3s.code.flow-step.identity.v1";
23/// Identity domain for a process-local scheduler admission scope.
24///
25/// Scope identities are derived from a run/host boundary and only their
26/// digest is retained by the scheduler. They are capacity keys, not durable
27/// workflow claims or result identities.
28pub const TASK_ADMISSION_SCOPE_IDENTITY_DOMAIN_V1: &str =
29    "a3s.code.task-admission-scope.identity.v1";
30/// Identity domain for a provider/model generation-capacity pool.
31///
32/// Pool identities deliberately bind only non-secret routing facts. API keys,
33/// session tokens, prompts, and request payloads must never influence or
34/// appear in a scheduler capacity key.
35pub const MODEL_GENERATION_POOL_IDENTITY_DOMAIN_V1: &str =
36    "a3s.code.model-generation-pool.identity.v1";
37/// Identity domain for the immutable input portion of a dynamic workflow.
38pub const DYNAMIC_WORKFLOW_INPUT_IDENTITY_DOMAIN_V1: &str =
39    "a3s.code.dynamic-workflow.input.identity.v1";
40/// Identity domain for a dynamic workflow continuation reconstructed from its
41/// durable immutable facts.
42pub const DYNAMIC_WORKFLOW_CONTINUATION_IDENTITY_DOMAIN_V1: &str =
43    "a3s.code.dynamic-workflow.continuation.identity.v1";
44/// Identity domain for the stable root claim that fences one dynamic
45/// workflow continuation while workers replay its evolving step history.
46pub const DYNAMIC_WORKFLOW_CLAIM_IDENTITY_DOMAIN_V1: &str =
47    "a3s.code.dynamic-workflow.claim.identity.v1";
48/// Identity domain for the immutable definition of a projected execution plan.
49pub const EXECUTION_PLAN_IDENTITY_DOMAIN_V1: &str = "a3s.code.execution-plan.identity.v1";
50pub const WORKFLOW_STEP_IDENTITY_DOMAIN_V1: &str = "a3s.code.workflow-step.identity.v1";
51pub const WORKFLOW_STEP_EVIDENCE_DOMAIN_V1: &str = "a3s.code.workflow-step.evidence.v1";
52pub const WORKFLOW_STEP_RESULT_DOMAIN_V1: &str = "a3s.code.workflow-step.result.v1";
53pub const EVALUATION_DISPATCH_IDENTITY_DOMAIN_V1: &str = "a3s.code.evaluation-dispatch.identity.v1";
54pub const EVALUATION_DISPATCH_REQUEST_DOMAIN_V1: &str = "a3s.code.evaluation-dispatch.request.v1";
55pub const EXECUTION_RESULT_RECEIPT_SCHEMA_V1: &str = "a3s.code.execution-result-receipt.v1";
56pub const EXECUTION_RESULT_MAX_BYTES: u64 = 1024 * 1024;
57
58#[derive(Debug, Clone, PartialEq, Eq, Error)]
59pub enum ExecutionIdentityError {
60    #[error("execution identity domain is empty or invalid")]
61    InvalidDomain,
62    #[error("execution identity serialization failed: {0}")]
63    Serialization(String),
64    #[error("execution identity digest is invalid")]
65    InvalidDigest,
66    #[error("execution identity digest does not match the value")]
67    DigestMismatch,
68    #[error("execution claim field `{0}` is empty")]
69    InvalidClaimField(&'static str),
70    #[error("execution result receipt field `{0}` is invalid")]
71    InvalidReceiptField(&'static str),
72    #[error("execution result receipt exceeds its byte limit")]
73    ReceiptSizeLimit,
74}
75
76/// A portable, domain-separated content identity.
77#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
78#[serde(deny_unknown_fields)]
79pub struct ExecutionIdentityV1 {
80    pub schema: String,
81    pub domain: String,
82    pub digest: String,
83}
84
85impl ExecutionIdentityV1 {
86    pub fn derive<T: Serialize>(
87        domain: impl Into<String>,
88        value: &T,
89    ) -> Result<Self, ExecutionIdentityError> {
90        let domain = domain.into();
91        validate_domain(&domain)?;
92        let canonical = serde_json::to_value(value)
93            .map(canonicalize)
94            .map_err(|error| ExecutionIdentityError::Serialization(error.to_string()))?;
95        let bytes = serde_json::to_vec(&canonical)
96            .map_err(|error| ExecutionIdentityError::Serialization(error.to_string()))?;
97        let mut hasher = Sha256::new();
98        hasher.update(domain.as_bytes());
99        hasher.update([0]);
100        hasher.update(bytes);
101        let digest = format!("sha256:{:x}", hasher.finalize());
102        Ok(Self {
103            schema: EXECUTION_IDENTITY_SCHEMA_V1.to_string(),
104            domain,
105            digest,
106        })
107    }
108
109    pub fn validate(&self) -> Result<(), ExecutionIdentityError> {
110        if self.schema != EXECUTION_IDENTITY_SCHEMA_V1 {
111            return Err(ExecutionIdentityError::InvalidDomain);
112        }
113        validate_domain(&self.domain)?;
114        let Some(hex) = self.digest.strip_prefix("sha256:") else {
115            return Err(ExecutionIdentityError::InvalidDigest);
116        };
117        if hex.len() != 64
118            || !hex
119                .bytes()
120                .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
121        {
122            return Err(ExecutionIdentityError::InvalidDigest);
123        }
124        Ok(())
125    }
126
127    pub fn key(&self) -> &str {
128        &self.digest
129    }
130
131    pub fn validate_for<T: Serialize>(&self, value: &T) -> Result<(), ExecutionIdentityError> {
132        self.validate()?;
133        let expected = Self::derive(&self.domain, value)?;
134        if expected.digest != self.digest {
135            return Err(ExecutionIdentityError::DigestMismatch);
136        }
137        Ok(())
138    }
139}
140
141#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
142#[serde(rename_all = "snake_case")]
143pub enum ExecutionResultOutcomeV1 {
144    Succeeded,
145    Failed,
146    Cancelled,
147    TimedOut,
148}
149
150/// A bounded, digest-only terminal result bound to one execution claim and
151/// the evidence snapshot consumed by that execution.
152#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
153#[serde(deny_unknown_fields)]
154pub struct ExecutionResultReceiptV1 {
155    pub schema: String,
156    pub identity: ExecutionIdentityV1,
157    pub evidence_digest: String,
158    pub outcome: ExecutionResultOutcomeV1,
159    #[serde(default, skip_serializing_if = "Option::is_none")]
160    pub result_digest: Option<String>,
161    pub result_bytes: u64,
162}
163
164impl ExecutionResultReceiptV1 {
165    pub fn new(
166        identity: ExecutionIdentityV1,
167        evidence_digest: impl Into<String>,
168        outcome: ExecutionResultOutcomeV1,
169        result_digest: Option<String>,
170        result_bytes: u64,
171    ) -> Result<Self, ExecutionIdentityError> {
172        let receipt = Self {
173            schema: EXECUTION_RESULT_RECEIPT_SCHEMA_V1.to_string(),
174            identity,
175            evidence_digest: evidence_digest.into(),
176            outcome,
177            result_digest,
178            result_bytes,
179        };
180        receipt.validate()?;
181        Ok(receipt)
182    }
183
184    pub fn validate(&self) -> Result<(), ExecutionIdentityError> {
185        if self.schema != EXECUTION_RESULT_RECEIPT_SCHEMA_V1 {
186            return Err(ExecutionIdentityError::InvalidReceiptField("schema"));
187        }
188        self.identity.validate()?;
189        validate_digest(&self.evidence_digest)
190            .map_err(|_| ExecutionIdentityError::InvalidReceiptField("evidence_digest"))?;
191        if self.result_bytes > EXECUTION_RESULT_MAX_BYTES {
192            return Err(ExecutionIdentityError::ReceiptSizeLimit);
193        }
194        match (&self.result_digest, self.result_bytes, self.outcome) {
195            (Some(digest), bytes, ExecutionResultOutcomeV1::Succeeded) => {
196                validate_digest(digest)
197                    .map_err(|_| ExecutionIdentityError::InvalidReceiptField("result_digest"))?;
198                if bytes == 0 {
199                    return Err(ExecutionIdentityError::InvalidReceiptField("result_bytes"));
200                }
201            }
202            (None, 0, ExecutionResultOutcomeV1::Failed)
203            | (None, 0, ExecutionResultOutcomeV1::Cancelled)
204            | (None, 0, ExecutionResultOutcomeV1::TimedOut) => {}
205            _ => return Err(ExecutionIdentityError::InvalidReceiptField("outcome")),
206        }
207        Ok(())
208    }
209}
210
211/// Binds a semantic execution identity to the key used by an existing claim
212/// ledger. The ledger key is kept separate so old persisted receipts remain
213/// replay-compatible while new code can carry one typed identity through all
214/// claim, renewal, completion, and release operations.
215#[derive(Debug, Clone, PartialEq, Eq)]
216pub(crate) struct ExecutionClaimV1 {
217    identity: ExecutionIdentityV1,
218    record_id: String,
219    ledger_key: String,
220    owner_id: String,
221}
222
223impl ExecutionClaimV1 {
224    pub(crate) fn new(
225        identity: ExecutionIdentityV1,
226        record_id: impl Into<String>,
227        ledger_key: impl Into<String>,
228        owner_id: impl Into<String>,
229    ) -> Result<Self, ExecutionIdentityError> {
230        identity.validate()?;
231        let record_id = record_id.into();
232        let ledger_key = ledger_key.into();
233        let owner_id = owner_id.into();
234        if record_id.is_empty() {
235            return Err(ExecutionIdentityError::InvalidClaimField("record_id"));
236        }
237        if ledger_key.is_empty() {
238            return Err(ExecutionIdentityError::InvalidClaimField("ledger_key"));
239        }
240        if owner_id.is_empty() {
241            return Err(ExecutionIdentityError::InvalidClaimField("owner_id"));
242        }
243        Ok(Self {
244            identity,
245            record_id,
246            ledger_key,
247            owner_id,
248        })
249    }
250
251    pub(crate) fn identity(&self) -> &ExecutionIdentityV1 {
252        &self.identity
253    }
254
255    pub(crate) fn record_id(&self) -> &str {
256        &self.record_id
257    }
258
259    pub(crate) fn ledger_key(&self) -> &str {
260        &self.ledger_key
261    }
262
263    pub(crate) fn owner_id(&self) -> &str {
264        &self.owner_id
265    }
266
267    pub(crate) fn result_receipt(
268        &self,
269        evidence_digest: impl Into<String>,
270        outcome: ExecutionResultOutcomeV1,
271        result_digest: Option<String>,
272        result_bytes: u64,
273    ) -> Result<ExecutionResultReceiptV1, ExecutionIdentityError> {
274        ExecutionResultReceiptV1::new(
275            self.identity.clone(),
276            evidence_digest,
277            outcome,
278            result_digest,
279            result_bytes,
280        )
281    }
282}
283
284fn canonicalize(value: serde_json::Value) -> serde_json::Value {
285    match value {
286        serde_json::Value::Array(values) => {
287            serde_json::Value::Array(values.into_iter().map(canonicalize).collect())
288        }
289        serde_json::Value::Object(values) => {
290            let values = values
291                .into_iter()
292                .map(|(key, value)| (key, canonicalize(value)))
293                .collect::<BTreeMap<_, _>>();
294            serde_json::Value::Object(values.into_iter().collect())
295        }
296        value => value,
297    }
298}
299
300fn validate_digest(value: &str) -> Result<(), ExecutionIdentityError> {
301    let Some(hex) = value.strip_prefix("sha256:") else {
302        return Err(ExecutionIdentityError::InvalidDigest);
303    };
304    if hex.len() != 64
305        || !hex
306            .bytes()
307            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
308    {
309        return Err(ExecutionIdentityError::InvalidDigest);
310    }
311    Ok(())
312}
313
314fn validate_domain(domain: &str) -> Result<(), ExecutionIdentityError> {
315    if domain.is_empty()
316        || domain.len() > 128
317        || domain.contains('\0')
318        || domain.lines().count() != 1
319        || !domain.bytes().all(|byte| {
320            byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'-' | b'_')
321        })
322    {
323        return Err(ExecutionIdentityError::InvalidDomain);
324    }
325    Ok(())
326}
327
328#[cfg(test)]
329mod tests {
330    use super::*;
331
332    #[test]
333    fn identity_is_domain_separated_and_replay_stable() {
334        let value = serde_json::json!({"name": "read", "args": {"path": "src/lib.rs"}});
335        let first =
336            ExecutionIdentityV1::derive(TOOL_INVOCATION_IDENTITY_DOMAIN_V1, &value).unwrap();
337        let second =
338            ExecutionIdentityV1::derive(TOOL_INVOCATION_IDENTITY_DOMAIN_V1, &value).unwrap();
339        let other = ExecutionIdentityV1::derive(MODEL_CALL_IDENTITY_DOMAIN_V1, &value).unwrap();
340
341        assert_eq!(first, second);
342        assert_ne!(first.digest, other.digest);
343        first.validate().unwrap();
344        first.validate_for(&value).unwrap();
345
346        let left = serde_json::json!({"a": 1, "b": 2});
347        let right = serde_json::json!({"b": 2, "a": 1});
348        assert_eq!(
349            ExecutionIdentityV1::derive("a3s.test", &left).unwrap(),
350            ExecutionIdentityV1::derive("a3s.test", &right).unwrap()
351        );
352    }
353
354    #[test]
355    fn identity_rejects_malformed_domains_and_digests() {
356        assert!(matches!(
357            ExecutionIdentityV1::derive("Bad Domain", &"value"),
358            Err(ExecutionIdentityError::InvalidDomain)
359        ));
360        let mut identity = ExecutionIdentityV1::derive("a3s.test", &"value").unwrap();
361        identity.digest = "sha256:ABC".to_string();
362        assert!(matches!(
363            identity.validate(),
364            Err(ExecutionIdentityError::InvalidDigest)
365        ));
366    }
367
368    #[test]
369    fn identity_validation_detects_content_tampering() {
370        let value = serde_json::json!({"a": 1, "b": {"c": true}});
371        let identity = ExecutionIdentityV1::derive("a3s.test", &value).unwrap();
372        let changed = serde_json::json!({"a": 2, "b": {"c": true}});
373        assert!(matches!(
374            identity.validate_for(&changed),
375            Err(ExecutionIdentityError::DigestMismatch)
376        ));
377    }
378
379    #[test]
380    fn claim_binds_canonical_identity_to_a_legacy_ledger_key() {
381        let payload = serde_json::json!({"secret": "do-not-persist"});
382        let identity = ExecutionIdentityV1::derive("a3s.test", &payload).unwrap();
383        let claim = ExecutionClaimV1::new(identity.clone(), "dispatch-1", "legacy-hash", "owner-1")
384            .unwrap();
385
386        assert_eq!(claim.identity(), &identity);
387        assert_eq!(claim.record_id(), "dispatch-1");
388        assert_eq!(claim.ledger_key(), "legacy-hash");
389        assert_eq!(claim.owner_id(), "owner-1");
390        let debug = format!("{claim:?}");
391        assert!(!debug.contains("do-not-persist"));
392    }
393
394    #[test]
395    fn result_receipt_is_digest_only_and_bounded() {
396        let identity = ExecutionIdentityV1::derive("a3s.test", &"request").unwrap();
397        let claim =
398            ExecutionClaimV1::new(identity.clone(), "record-1", "legacy-hash", "owner-1").unwrap();
399        let receipt = claim
400            .result_receipt(
401                "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
402                ExecutionResultOutcomeV1::Succeeded,
403                Some(
404                    "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
405                        .into(),
406                ),
407                12,
408            )
409            .unwrap();
410        receipt.validate().unwrap();
411        assert_eq!(receipt.identity, identity);
412        assert!(!format!("{receipt:?}").contains("request"));
413
414        let invalid = ExecutionResultReceiptV1 {
415            schema: EXECUTION_RESULT_RECEIPT_SCHEMA_V1.into(),
416            identity: receipt.identity,
417            evidence_digest: receipt.evidence_digest,
418            outcome: ExecutionResultOutcomeV1::Succeeded,
419            result_digest: None,
420            result_bytes: 0,
421        };
422        assert!(matches!(
423            invalid.validate(),
424            Err(ExecutionIdentityError::InvalidReceiptField("outcome"))
425        ));
426    }
427}
a3s_code_core/execution_identity.rs

a3s_code_core/
execution_identity.rs
