1use std::sync::{Arc, Mutex, PoisonError};
13use std::time::Instant;
14
15use lean_ctx_protocol::context_gateway::{
16 ContextDecisionReceiptV1, ContextDecisionV1, ContextDispositionV1, DeliveryOutcomeV1,
17 DestinationLocalityV1, DestinationV1, GatewayModeV1, MAX_DECISIONS, PolicyRefV1,
18 PrincipalKindV1, PrincipalV1, RECEIPT_V2_SCHEMA_VERSION, ReasonCodeV1, SecurityCountsV1,
19 SourceCountsV1, TokenAccountV1,
20};
21use lean_ctx_protocol::{PolicyId, ProtocolReference, Sha256Digest, TaskId};
22
23use super::AdmissionCounts;
24
25tokio::task_local! {
26 pub static ADMISSIONS: Option<Arc<AdmissionCapture>>;
27}
28
29pub fn current() -> Option<Arc<AdmissionCapture>> {
31 ADMISSIONS.try_with(Clone::clone).ok().flatten()
32}
33
34#[derive(Debug, Clone)]
35struct Recorded {
36 decision: ContextDecisionV1,
37 counts: AdmissionCounts,
38 external: bool,
41}
42
43pub const UNTRUSTED_ORIGIN: &str = "origin.untrusted";
46
47pub const MAX_LEDGER_REFS: usize = 256;
49
50#[derive(Debug, Clone, Default, PartialEq, Eq)]
54pub struct OriginLedger {
55 pub count: u64,
57 pub refs: std::collections::BTreeSet<super::origin::InputRef>,
59 pub overflow: bool,
61}
62
63impl OriginLedger {
64 fn note(&mut self, input: Option<super::origin::InputRef>) {
65 self.count = self.count.saturating_add(1);
66 match input {
67 Some(input) if self.refs.contains(&input) => {}
68 Some(input) if self.refs.len() < MAX_LEDGER_REFS => {
69 self.refs.insert(input);
70 }
71 _ => self.overflow = true,
72 }
73 }
74
75 fn merge(&mut self, other: &Self) {
76 self.count = self.count.saturating_add(other.count);
77 for input in &other.refs {
78 if !self.refs.contains(input) {
79 if self.refs.len() < MAX_LEDGER_REFS {
80 self.refs.insert(input.clone());
81 } else {
82 self.overflow = true;
83 }
84 }
85 }
86 self.overflow |= other.overflow;
87 }
88
89 pub fn is_empty(&self) -> bool {
91 self.count == 0 && !self.overflow
92 }
93}
94
95pub static ORPHAN_EXTERNAL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
99
100thread_local! {
101 static OUTSIDE_CALL: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
103 static IN_SPAN: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
105 static EGRESS: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
107}
108
109pub fn outside_call_active() -> bool {
111 OUTSIDE_CALL.with(std::cell::Cell::get) > 0
112}
113
114pub fn egress_check_active() -> bool {
118 EGRESS.with(std::cell::Cell::get) > 0
119}
120
121pub fn egress_check<T>(check: impl FnOnce() -> T) -> T {
124 struct Leave;
125 impl Drop for Leave {
126 fn drop(&mut self) {
127 EGRESS.with(|depth| depth.set(depth.get().saturating_sub(1)));
128 }
129 }
130 EGRESS.with(|depth| depth.set(depth.get().saturating_add(1)));
131 let _leave = Leave;
132 outside_call(check)
133}
134
135pub fn outside_call<T>(operation: impl FnOnce() -> T) -> T {
141 struct Leave;
142 impl Drop for Leave {
143 fn drop(&mut self) {
144 OUTSIDE_CALL.with(|depth| depth.set(depth.get().saturating_sub(1)));
145 }
146 }
147 OUTSIDE_CALL.with(|depth| depth.set(depth.get().saturating_add(1)));
148 let _leave = Leave;
149 operation()
150}
151
152pub fn note_origin(class: &super::origin::SourceClass) {
157 if !class.is_external_or_unknown() || outside_call_active() {
158 return;
159 }
160 let input = class.input_ref();
161 match current() {
162 Some(capture) => capture.note_origin_ref(input),
163 None => {
164 ORPHAN_EXTERNAL.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
165 }
166 }
167}
168
169#[derive(Clone)]
172pub struct Carried {
173 capture: Option<Arc<AdmissionCapture>>,
174 outside: bool,
175 project: Option<std::path::PathBuf>,
176}
177
178pub fn carry() -> Carried {
180 Carried {
181 capture: current(),
182 outside: outside_call_active(),
183 project: crate::core::policy::runtime::REQUEST_PROJECT
184 .try_with(|slot| slot.borrow().clone())
185 .ok()
186 .flatten(),
187 }
188}
189
190impl Carried {
191 pub fn enter<T>(&self, operation: impl FnOnce() -> T) -> T {
193 let inner = || {
194 if self.outside {
195 outside_call(operation)
196 } else {
197 operation()
198 }
199 };
200 let with_project = || match &self.project {
201 Some(project) if crate::core::policy::diagnostics::request_project().is_none() => {
202 crate::core::policy::runtime::in_project(project.clone(), inner)
203 }
204 _ => inner(),
205 };
206 if self.capture.is_some() && current().is_none() {
207 ADMISSIONS.sync_scope(self.capture.clone(), with_project)
208 } else {
209 with_project()
210 }
211 }
212}
213
214pub fn external_span(
220 class: &super::origin::SourceClass,
221 admit: impl FnOnce() -> Option<String>,
222) -> Option<String> {
223 let Some(capture) =
224 current().filter(|_| class.is_external_or_unknown() && !outside_call_active())
225 else {
226 return admit();
227 };
228 struct Leave;
229 impl Drop for Leave {
230 fn drop(&mut self) {
231 IN_SPAN.with(|depth| depth.set(depth.get().saturating_sub(1)));
232 }
233 }
234 IN_SPAN.with(|depth| depth.set(depth.get().saturating_add(1)));
235 let admitted = {
236 let _leave = Leave;
237 admit()
238 };
239 if let Some(text) = &admitted
240 && !text.is_empty()
241 {
242 capture
243 .spans
244 .lock()
245 .unwrap_or_else(PoisonError::into_inner)
246 .push((class.input_ref(), text.clone()));
247 }
248 admitted
249}
250
251fn in_span() -> bool {
252 IN_SPAN.with(std::cell::Cell::get) > 0
253}
254
255#[derive(Debug, Clone)]
264pub struct CallSnapshot {
265 admission: super::stores::StoreAdmission,
266 protected: bool,
267 pack: Option<(String, String)>,
268 org_policy: Option<OrgObservation>,
271}
272
273pub(crate) type OrgObservation = crate::core::policy::org::Observation;
275
276#[derive(Debug, Clone, PartialEq, Eq, Default)]
278pub enum OrgEvidence {
279 #[default]
281 None,
282 One(OrgObservation),
284 Diverged,
289 Unavailable,
294}
295
296impl OrgEvidence {
297 fn of(observation: Option<OrgObservation>) -> Self {
298 observation.map_or(Self::None, Self::One)
299 }
300}
301
302impl CallSnapshot {
303 pub fn take() -> Self {
308 let admission = super::carrier::gateway_snapshot();
309 let protected =
310 super::stored::source_authority_required_under(admission.admission_policy());
311 let (active, org_policy) = crate::core::policy::runtime::active_observed();
314 let pack = active.map(|active| {
315 (
316 active.resolved.name.clone(),
317 active.resolved.version.clone(),
318 )
319 });
320 Self {
321 admission,
322 protected,
323 pack,
324 org_policy,
325 }
326 }
327
328 pub fn admission(&self) -> &super::stores::StoreAdmission {
330 &self.admission
331 }
332
333 pub(crate) fn protected(&self) -> bool {
336 self.protected
337 }
338
339 pub(crate) fn org_policy(&self) -> OrgEvidence {
341 OrgEvidence::of(self.org_policy.clone())
342 }
343
344 pub(crate) fn pack(&self) -> Option<&(String, String)> {
346 self.pack.as_ref()
347 }
348}
349
350pub(crate) const MAX_SOURCE_REFERENCES: usize = 64;
352pub(crate) const MAX_SOURCE_REFERENCE_BYTES: usize = 256;
354
355#[derive(Debug, Clone, PartialEq, Eq)]
357pub enum SourceName {
358 Path(String),
360 Id(String),
363}
364
365#[derive(Debug, Default)]
367struct SourceNotes {
368 named: Vec<(Sha256Digest, SourceName)>,
371 omitted: u32,
373}
374
375thread_local! {
376 static UNNAMED: std::cell::RefCell<Option<(usize, Sha256Digest)>> =
382 const { std::cell::RefCell::new(None) };
383}
384
385pub(crate) fn note_source(project: &std::path::Path, canonical_path: &str) {
393 let Some((capture, object)) = unnamed_decision() else {
394 return;
395 };
396 let path = std::path::Path::new(canonical_path);
397 let relative = path
398 .strip_prefix(project)
399 .ok()
400 .map(std::path::Path::to_path_buf)
401 .or_else(|| {
402 let project = crate::core::pathutil::canonicalize_secure(project).ok()?;
403 path.strip_prefix(&project)
404 .ok()
405 .map(std::path::Path::to_path_buf)
406 });
407 let reference = relative
408 .as_deref()
409 .and_then(std::path::Path::to_str)
410 .filter(|reference| !reference.is_empty() && reference.len() <= MAX_SOURCE_REFERENCE_BYTES)
411 .map(|reference| SourceName::Path(reference.replace('\\', "/")));
412 capture.name_source(object, reference);
413}
414
415pub fn note_source_id(id: &str) {
422 let Some((capture, object)) = unnamed_decision() else {
423 return;
424 };
425 let reference = (!id.is_empty() && id.len() <= MAX_SOURCE_REFERENCE_BYTES)
426 .then(|| SourceName::Id(id.to_owned()));
427 capture.name_source(object, reference);
428}
429
430fn unnamed_decision() -> Option<(Arc<AdmissionCapture>, Sha256Digest)> {
434 let capture = current()?;
435 let owner = std::ptr::from_ref::<AdmissionCapture>(&capture).addr();
436 let (recorded_by, object) = UNNAMED.with(|slot| slot.borrow_mut().take())?;
437 (recorded_by == owner).then_some((capture, object))
438}
439
440#[derive(Debug)]
442pub struct AdmissionCapture {
443 started: Instant,
444 records: Mutex<Vec<Recorded>>,
445 overflow: Mutex<Overflow>,
447 original_tokens: Mutex<Option<u64>>,
449 policy: Mutex<Option<(GatewayModeV1, Sha256Digest)>>,
450 task: Mutex<Option<(String, String)>>,
453 snapshot: Option<CallSnapshot>,
457 diverged: std::sync::atomic::AtomicBool,
460 sources: Mutex<SourceNotes>,
461 pack: Mutex<Option<(String, String)>>,
463 org_policy: Mutex<Option<OrgEvidence>>,
466 origin: Mutex<OriginLedger>,
468 spans: Mutex<Vec<(Option<super::origin::InputRef>, String)>>,
470 orphan_mark: u64,
472 output_class: Mutex<Option<super::origin::SourceClass>>,
475 released: Mutex<std::collections::BTreeSet<[u8; 32]>>,
480}
481
482const MAX_RELEASED_BLOCKS: usize = 4096;
485
486impl Default for AdmissionCapture {
487 fn default() -> Self {
488 Self {
489 released: Mutex::new(std::collections::BTreeSet::new()),
490 origin: Mutex::new(OriginLedger::default()),
491 spans: Mutex::new(Vec::new()),
492 orphan_mark: ORPHAN_EXTERNAL.load(std::sync::atomic::Ordering::SeqCst),
493 output_class: Mutex::new(None),
494 started: Instant::now(),
495 records: Mutex::new(Vec::new()),
496 overflow: Mutex::new(Overflow::default()),
497 original_tokens: Mutex::new(None),
498 policy: Mutex::new(None),
499 task: Mutex::new(None),
500 snapshot: None,
501 diverged: std::sync::atomic::AtomicBool::new(false),
502 sources: Mutex::new(SourceNotes::default()),
503 pack: Mutex::new(None),
504 org_policy: Mutex::new(None),
505 }
506 }
507}
508
509impl AdmissionCapture {
510 pub fn new() -> Arc<Self> {
511 Arc::new(Self::default())
512 }
513
514 pub fn for_call(snapshot: CallSnapshot) -> Arc<Self> {
517 Arc::new(Self {
518 snapshot: Some(snapshot),
519 ..Self::default()
520 })
521 }
522
523 pub fn snapshot(&self) -> Option<&CallSnapshot> {
525 self.snapshot.as_ref()
526 }
527
528 pub fn set_output_class(&self, class: super::origin::SourceClass) {
531 *self
532 .output_class
533 .lock()
534 .unwrap_or_else(PoisonError::into_inner) = Some(class);
535 }
536
537 pub fn output_class(&self) -> Option<super::origin::SourceClass> {
539 self.output_class
540 .lock()
541 .unwrap_or_else(PoisonError::into_inner)
542 .clone()
543 }
544
545 fn note_origin_ref(&self, input: Option<super::origin::InputRef>) {
546 if in_span() {
547 return;
548 }
549 self.origin
550 .lock()
551 .unwrap_or_else(PoisonError::into_inner)
552 .note(input);
553 }
554
555 pub fn origin_ledger(&self) -> OriginLedger {
558 let mut ledger = self
559 .origin
560 .lock()
561 .unwrap_or_else(PoisonError::into_inner)
562 .clone();
563 if self.orphan_external() {
564 ledger.overflow = true;
565 }
566 ledger
567 }
568
569 pub fn orphan_external(&self) -> bool {
572 ORPHAN_EXTERNAL.load(std::sync::atomic::Ordering::SeqCst) != self.orphan_mark
573 }
574
575 pub fn external_spans(&self) -> Vec<(Option<super::origin::InputRef>, String)> {
577 self.spans
578 .lock()
579 .unwrap_or_else(PoisonError::into_inner)
580 .clone()
581 }
582
583 pub fn mark_released(&self, text: &str) {
585 let mut released = self.released.lock().unwrap_or_else(PoisonError::into_inner);
586 if released.len() < MAX_RELEASED_BLOCKS {
587 released.insert(*blake3::hash(text.as_bytes()).as_bytes());
588 }
589 }
590
591 pub fn was_released(&self, text: &str) -> bool {
594 self.released
595 .lock()
596 .unwrap_or_else(PoisonError::into_inner)
597 .contains(blake3::hash(text.as_bytes()).as_bytes())
598 }
599
600 pub fn output_origins(&self) -> OriginLedger {
604 let mut ledger = self.origin_ledger();
605 for (input, _) in self.external_spans() {
606 ledger.note(input);
607 }
608 ledger
609 }
610
611 pub fn renote_origins(&self, origins: &OriginLedger) {
615 self.origin
616 .lock()
617 .unwrap_or_else(PoisonError::into_inner)
618 .merge(origins);
619 }
620
621 pub fn spans_into_ledger(&self) {
624 let spans = std::mem::take(&mut *self.spans.lock().unwrap_or_else(PoisonError::into_inner));
625 let mut ledger = self.origin.lock().unwrap_or_else(PoisonError::into_inner);
626 for (input, _) in spans {
627 ledger.note(input);
628 }
629 }
630
631 pub(crate) fn policy_diverged(&self) -> bool {
634 self.diverged.load(std::sync::atomic::Ordering::SeqCst)
635 }
636
637 pub fn source_references(&self) -> (Vec<(Sha256Digest, SourceName)>, u32) {
640 let sources = self.sources.lock().unwrap_or_else(PoisonError::into_inner);
641 (sources.named.clone(), sources.omitted)
642 }
643
644 fn name_source(&self, object: Sha256Digest, name: Option<SourceName>) {
647 let mut sources = self.sources.lock().unwrap_or_else(PoisonError::into_inner);
648 match name {
649 Some(name) if sources.named.len() < MAX_SOURCE_REFERENCES => {
650 sources.named.push((object, name));
651 }
652 _ => sources.omitted = sources.omitted.saturating_add(1),
653 }
654 }
655
656 pub(super) fn note_pack(&self, name: &str, version: &str) {
659 *self.pack.lock().unwrap_or_else(PoisonError::into_inner) =
660 Some((name.to_owned(), version.to_owned()));
661 }
662
663 pub(super) fn note_org_policy(&self, observation: Option<OrgObservation>) {
668 self.merge_org_policy(OrgEvidence::of(observation));
669 }
670
671 pub(super) fn note_org_policy_unavailable(&self) {
674 self.merge_org_policy(OrgEvidence::Unavailable);
675 }
676
677 fn merge_org_policy(&self, incoming: OrgEvidence) {
678 let mut noted = self
679 .org_policy
680 .lock()
681 .unwrap_or_else(PoisonError::into_inner);
682 *noted = Some(match noted.take() {
683 None => incoming,
684 Some(present) if present == incoming => present,
685 Some(_) => OrgEvidence::Diverged,
686 });
687 }
688
689 pub(crate) fn org_policy(&self) -> OrgEvidence {
692 match &self.snapshot {
693 Some(snapshot) => OrgEvidence::of(snapshot.org_policy.clone()),
694 None => self
695 .org_policy
696 .lock()
697 .unwrap_or_else(PoisonError::into_inner)
698 .clone()
699 .unwrap_or_default(),
700 }
701 }
702
703 pub(crate) fn pack(&self) -> Option<(String, String)> {
706 self.snapshot
707 .as_ref()
708 .and_then(|snapshot| snapshot.pack.clone())
709 .or_else(|| {
710 self.pack
711 .lock()
712 .unwrap_or_else(PoisonError::into_inner)
713 .clone()
714 })
715 }
716
717 pub(super) fn record(&self, decision: &ContextDecisionV1, counts: AdmissionCounts) {
718 self.record_as(decision, counts, false);
719 }
720
721 pub(super) fn record_as(
724 &self,
725 decision: &ContextDecisionV1,
726 counts: AdmissionCounts,
727 external: bool,
728 ) {
729 if let Some(envelope) = crate::core::task_spine::TaskSpine::current() {
730 let mut task = self.task.lock().unwrap_or_else(PoisonError::into_inner);
731 if task.is_none() {
732 *task = Some((
733 envelope.task_id.as_str().to_owned(),
734 crate::core::context_store::task_scope(
735 envelope.tenant_id.as_ref(),
736 &envelope.project_id,
737 ),
738 ));
739 }
740 }
741 let nameable = decision.object != note_object() && decision.disposition.delivers_content();
744 UNNAMED.with(|slot| {
745 *slot.borrow_mut() =
746 nameable.then(|| (std::ptr::from_ref(self).addr(), decision.object.clone()));
747 });
748 let mut decision = decision.clone();
749 if external {
752 let code =
753 ReasonCodeV1::new(UNTRUSTED_ORIGIN).expect("built-in reason codes are valid");
754 if !decision.reason_codes.contains(&code) {
755 decision.reason_codes.push(code);
756 }
757 }
758 let incoming = Recorded {
759 decision,
760 counts,
761 external,
762 };
763 let mut records = self.records.lock().unwrap_or_else(PoisonError::into_inner);
764 if records.len() < MAX_DECISIONS {
765 records.push(incoming);
766 return;
767 }
768 let displaced = if incoming.is_clean() {
773 incoming
774 } else if let Some(position) = records.iter().position(Recorded::is_clean) {
775 std::mem::replace(&mut records[position], incoming)
776 } else if incoming.is_note() {
777 let Some(displaced) = mark_notes_not_itemized(&mut records) else {
781 return;
782 };
783 displaced
784 } else {
785 incoming
786 };
787 self.overflow
788 .lock()
789 .unwrap_or_else(PoisonError::into_inner)
790 .add(&displaced);
791 }
792
793 pub(super) fn bind_policy(&self, mode: GatewayModeV1, digest: Sha256Digest) {
794 if let Some(snapshot) = &self.snapshot
795 && snapshot.admission.digest() != digest.hex()
796 {
797 self.diverged
798 .store(true, std::sync::atomic::Ordering::SeqCst);
799 }
800 *self.policy.lock().unwrap_or_else(PoisonError::into_inner) = Some((mode, digest));
801 }
802
803 pub(crate) fn task_scope(&self) -> Option<String> {
805 self.task
806 .lock()
807 .unwrap_or_else(PoisonError::into_inner)
808 .as_ref()
809 .map(|(_, scope)| scope.clone())
810 }
811
812 pub fn set_original_tokens(&self, tokens: u64) {
814 *self
815 .original_tokens
816 .lock()
817 .unwrap_or_else(PoisonError::into_inner) = Some(tokens);
818 }
819
820 pub fn is_empty(&self) -> bool {
821 self.records
822 .lock()
823 .unwrap_or_else(PoisonError::into_inner)
824 .is_empty()
825 }
826
827 pub(crate) fn describes_sources(&self) -> bool {
831 let note = note_object();
832 let itemized = self
833 .records
834 .lock()
835 .unwrap_or_else(PoisonError::into_inner)
836 .iter()
837 .any(|record| record.decision.object != note);
838 itemized
839 || self
840 .overflow
841 .lock()
842 .unwrap_or_else(PoisonError::into_inner)
843 .objects
844 > 0
845 }
846
847 pub(super) fn record_unsourced(
854 &self,
855 output: Sha256Digest,
856 mode: GatewayModeV1,
857 policy: Sha256Digest,
858 ) {
859 if self.bound_policy().is_none() {
860 self.bind_policy(mode, policy);
861 }
862 let decision = ContextDecisionV1 {
863 object: output,
864 disposition: ContextDispositionV1::Allow,
865 reason_codes: vec![
866 ReasonCodeV1::new(NO_SOURCE_DECISION).expect("built-in reason codes are valid"),
867 ],
868 signals: Vec::new(),
869 required_transformations: Vec::new(),
870 };
871 self.record(&decision, AdmissionCounts::default());
872 }
873
874 pub fn itemized(&self) -> Vec<(ContextDecisionV1, AdmissionCounts)> {
877 self.records
878 .lock()
879 .unwrap_or_else(PoisonError::into_inner)
880 .iter()
881 .map(|record| (record.decision.clone(), record.counts))
882 .collect()
883 }
884
885 pub(crate) fn counted_only(&self) -> u32 {
887 self.overflow
888 .lock()
889 .unwrap_or_else(PoisonError::into_inner)
890 .objects
891 }
892
893 pub(super) fn bound_policy(&self) -> Option<(GatewayModeV1, Sha256Digest)> {
895 self.policy
896 .lock()
897 .unwrap_or_else(PoisonError::into_inner)
898 .clone()
899 }
900
901 pub(super) fn absorb(&self, earlier: &Self) {
906 if let Some((name, version)) = earlier.pack()
907 && self.pack().is_none()
908 {
909 self.note_pack(&name, &version);
910 }
911 let theirs = earlier
915 .org_policy
916 .lock()
917 .unwrap_or_else(PoisonError::into_inner)
918 .clone();
919 if let Some(theirs) = theirs {
920 self.merge_org_policy(theirs);
921 }
922 let records = earlier
923 .records
924 .lock()
925 .unwrap_or_else(PoisonError::into_inner)
926 .clone();
927 for record in &records {
928 self.record_as(&record.decision, record.counts, record.external);
929 }
930 let ledger = earlier
931 .origin
932 .lock()
933 .unwrap_or_else(PoisonError::into_inner)
934 .clone();
935 self.origin
936 .lock()
937 .unwrap_or_else(PoisonError::into_inner)
938 .merge(&ledger);
939 let spans = earlier.external_spans();
940 self.spans
941 .lock()
942 .unwrap_or_else(PoisonError::into_inner)
943 .extend(spans);
944 let released = earlier
945 .released
946 .lock()
947 .unwrap_or_else(PoisonError::into_inner)
948 .clone();
949 {
950 let mut ours = self.released.lock().unwrap_or_else(PoisonError::into_inner);
951 for digest in released {
952 if ours.len() >= MAX_RELEASED_BLOCKS {
953 break;
954 }
955 ours.insert(digest);
956 }
957 }
958 let theirs = earlier
959 .overflow
960 .lock()
961 .unwrap_or_else(PoisonError::into_inner);
962 let mut ours = self.overflow.lock().unwrap_or_else(PoisonError::into_inner);
963 ours.objects = ours.objects.saturating_add(theirs.objects);
964 ours.permitted = ours.permitted.saturating_add(theirs.permitted);
965 ours.blocked = ours.blocked.saturating_add(theirs.blocked);
966 ours.redactions = ours.redactions.saturating_add(theirs.redactions);
967 ours.injection = ours.injection.saturating_add(theirs.injection);
968 ours.incomplete = ours.incomplete.saturating_add(theirs.incomplete);
969 ours.security.merge(&theirs.security);
970 ours.injection_flags
971 .extend(theirs.injection_flags.iter().cloned());
972 }
973
974 pub(super) fn deny_reason_codes(
977 &self,
978 ) -> Vec<lean_ctx_protocol::context_gateway::ReasonCodeV1> {
979 let mut codes = Vec::new();
980 for record in self
981 .records
982 .lock()
983 .unwrap_or_else(PoisonError::into_inner)
984 .iter()
985 .filter(|record| record.decision.disposition == ContextDispositionV1::Deny)
986 {
987 for code in &record.decision.reason_codes {
988 if !codes.contains(code) {
989 codes.push(code.clone());
990 }
991 }
992 }
993 codes
994 }
995
996 pub fn flagged_injection(&self) -> bool {
999 self.records
1000 .lock()
1001 .unwrap_or_else(PoisonError::into_inner)
1002 .iter()
1003 .any(|record| record.counts.injection > 0)
1004 }
1005
1006 pub(crate) fn security_tally(&self) -> crate::core::security_events::SecurityCounts {
1010 let mut tally = self
1011 .overflow
1012 .lock()
1013 .unwrap_or_else(PoisonError::into_inner)
1014 .security;
1015 for record in self
1016 .records
1017 .lock()
1018 .unwrap_or_else(PoisonError::into_inner)
1019 .iter()
1020 {
1021 tally.merge(&record.security());
1022 }
1023 tally
1024 }
1025
1026 fn injection_flags(&self) -> Vec<crate::core::security_events::InjectionFlag> {
1028 let mut flags = self
1029 .overflow
1030 .lock()
1031 .unwrap_or_else(PoisonError::into_inner)
1032 .injection_flags
1033 .clone();
1034 flags.extend(
1035 self.records
1036 .lock()
1037 .unwrap_or_else(PoisonError::into_inner)
1038 .iter()
1039 .filter_map(Recorded::injection_flag),
1040 );
1041 flags
1042 }
1043
1044 pub(crate) fn record_security(
1049 &self,
1050 tool: &str,
1051 agent_id: &str,
1052 ) -> crate::core::security_events::SecurityCounts {
1053 use crate::core::security_events::{self, SecurityKind};
1054 let mut tally = self.security_tally();
1055 tally.injection_flagged = 0;
1056 security_events::record(tool, agent_id, &tally);
1057 let flagged = security_events::record_injections(tool, agent_id, &self.injection_flags());
1058 tally.add(SecurityKind::InjectionFlagged, flagged);
1059 tally
1060 }
1061}
1062
1063pub(super) const NOTES_NOT_ITEMIZED: &str = "recovery.notes_not_itemized";
1068
1069pub(crate) const NO_SOURCE_DECISION: &str = "delivery.no_source_decision";
1073
1074fn mark_notes_not_itemized(records: &mut [Recorded]) -> Option<Recorded> {
1082 if records.iter().any(Recorded::states_notes_not_itemized) {
1083 return None;
1084 }
1085 let marker = Recorded {
1086 decision: ContextDecisionV1 {
1087 object: note_object(),
1088 disposition: ContextDispositionV1::Allow,
1089 reason_codes: vec![
1090 ReasonCodeV1::new(NOTES_NOT_ITEMIZED).expect("built-in reason codes are valid"),
1091 ],
1092 signals: Vec::new(),
1093 required_transformations: Vec::new(),
1094 },
1095 counts: AdmissionCounts::default(),
1096 external: false,
1097 };
1098 let slot = records
1099 .iter()
1100 .rposition(Recorded::is_note)
1101 .or_else(|| {
1102 records
1103 .iter()
1104 .rposition(|record| record.decision.disposition.delivers_content())
1105 })
1106 .unwrap_or(records.len().saturating_sub(1));
1107 let displaced = std::mem::replace(records.get_mut(slot)?, marker);
1108 (!displaced.is_note()).then_some(displaced)
1109}
1110
1111impl Recorded {
1112 fn injection_flag(&self) -> Option<crate::core::security_events::InjectionFlag> {
1113 (self.counts.injection > 0).then(|| crate::core::security_events::InjectionFlag {
1114 object_sha256: self.decision.object.hex().to_owned(),
1115 patterns: self.counts.injection_patterns,
1116 })
1117 }
1118
1119 fn is_clean(&self) -> bool {
1121 self.decision.disposition == ContextDispositionV1::Allow
1122 && self.decision.reason_codes.is_empty()
1123 && !self.external
1124 }
1125
1126 fn is_note(&self) -> bool {
1128 self.decision.object == note_object()
1129 }
1130
1131 fn states_notes_not_itemized(&self) -> bool {
1132 self.is_note()
1133 && self
1134 .decision
1135 .reason_codes
1136 .iter()
1137 .any(|reason| reason.as_str() == NOTES_NOT_ITEMIZED)
1138 }
1139
1140 fn security(&self) -> crate::core::security_events::SecurityCounts {
1143 use crate::core::security_events::{SecurityCounts, SecurityKind};
1144 let mut tally = SecurityCounts::default();
1145 let has = |code: &str| {
1146 self.decision
1147 .reason_codes
1148 .iter()
1149 .any(|reason| reason.as_str() == code)
1150 };
1151 if has("secret.redacted") {
1152 tally.add(SecurityKind::SecretRedacted, u64::from(self.counts.secrets));
1153 }
1154 if has("pii.redacted") {
1155 tally.add(SecurityKind::PiiRedacted, u64::from(self.counts.pii));
1156 }
1157 if self.counts.injection > 0 {
1158 tally.add(
1159 SecurityKind::InjectionFlagged,
1160 u64::from(self.counts.injection),
1161 );
1162 }
1163 if self.decision.disposition == ContextDispositionV1::Deny {
1164 tally.add(SecurityKind::ContentWithheld, 1);
1165 } else if self.counts.incomplete > 0 {
1166 tally.add(SecurityKind::CoverageIncomplete, 1);
1167 }
1168 tally
1169 }
1170
1171 fn redactions(&self) -> u32 {
1172 if self.decision.disposition == ContextDispositionV1::AllowRedacted {
1173 self.counts.secrets.saturating_add(self.counts.pii)
1174 } else {
1175 0
1176 }
1177 }
1178}
1179
1180#[derive(Debug, Default)]
1182struct Overflow {
1183 objects: u32,
1184 permitted: u32,
1185 blocked: u32,
1186 redactions: u32,
1187 injection: u32,
1188 incomplete: u32,
1189 security: crate::core::security_events::SecurityCounts,
1190 injection_flags: Vec<crate::core::security_events::InjectionFlag>,
1191}
1192
1193impl Overflow {
1194 fn add(&mut self, record: &Recorded) {
1195 if record.decision.object == note_object() {
1198 return;
1199 }
1200 let disposition = record.decision.disposition;
1201 self.objects = self.objects.saturating_add(1);
1202 if disposition.delivers_content() {
1203 self.permitted = self.permitted.saturating_add(1);
1204 }
1205 if disposition == ContextDispositionV1::Deny {
1206 self.blocked = self.blocked.saturating_add(1);
1207 }
1208 self.redactions = self.redactions.saturating_add(record.redactions());
1209 self.injection = self.injection.saturating_add(record.counts.injection);
1210 if record.counts.incomplete > 0 {
1211 self.incomplete = self.incomplete.saturating_add(1);
1212 }
1213 self.security.merge(&record.security());
1214 self.injection_flags.extend(record.injection_flag());
1215 }
1216}
1217
1218pub struct CallIdentity<'a> {
1220 pub agent_id: Option<&'a str>,
1221 pub destination: Option<DestinationV1>,
1223}
1224
1225pub struct Delivered<'a> {
1227 pub text: &'a str,
1228 pub is_error: bool,
1229 pub tokens: u64,
1230}
1231
1232fn reference(prefix: &str, raw: &str) -> Option<ProtocolReference> {
1233 ProtocolReference::new(format!("{prefix}{raw}"))
1236 .ok()
1237 .or_else(|| {
1238 let digest = blake3::hash(raw.as_bytes()).to_hex();
1239 ProtocolReference::new(format!("{prefix}{}", &digest[..32])).ok()
1240 })
1241}
1242
1243fn sha256(bytes: &[u8]) -> Sha256Digest {
1244 use sha2::{Digest, Sha256};
1245 let hex = crate::core::agent_identity::hex_encode(&Sha256::digest(bytes));
1246 Sha256Digest::new(format!("sha256:{hex}")).expect("SHA-256 digest is canonical")
1247}
1248
1249fn saturating_u32(n: usize) -> u32 {
1250 u32::try_from(n).unwrap_or(u32::MAX)
1251}
1252
1253pub(super) fn note_object() -> Sha256Digest {
1259 sha256(b"proxy-block-kept-whole")
1260}
1261
1262pub fn is_note(decision: &ContextDecisionV1) -> bool {
1265 let note = note_object();
1266 decision.object == note
1268 || super::receipt_store::keyed::object_key()
1269 .is_ok_and(|key| decision.object == key.object(¬e))
1270}
1271
1272#[cfg(any(test, feature = "test-support"))]
1277pub fn finish(
1278 capture: &AdmissionCapture,
1279 identity: &CallIdentity<'_>,
1280 delivered: &Delivered<'_>,
1281) -> Option<ContextDecisionReceiptV1> {
1282 finish_for_presenter(capture, identity, None, delivered)
1283}
1284
1285pub(crate) fn finish_for_presenter(
1296 capture: &AdmissionCapture,
1297 identity: &CallIdentity<'_>,
1298 presenter_ref: Option<&str>,
1299 delivered: &Delivered<'_>,
1300) -> Option<ContextDecisionReceiptV1> {
1301 let records = capture
1302 .records
1303 .lock()
1304 .unwrap_or_else(PoisonError::into_inner)
1305 .clone();
1306 if records.is_empty() {
1307 return None;
1308 }
1309 let overflow = capture
1310 .overflow
1311 .lock()
1312 .unwrap_or_else(PoisonError::into_inner);
1313 let (mode, policy_digest) = capture
1314 .policy
1315 .lock()
1316 .unwrap_or_else(PoisonError::into_inner)
1317 .clone()?;
1318
1319 let decisions: Vec<ContextDecisionV1> = records
1320 .iter()
1321 .map(|record| record.decision.clone())
1322 .collect();
1323 let note = note_object();
1327 let notes = decisions
1328 .iter()
1329 .filter(|decision| decision.object == note)
1330 .count();
1331 if notes == decisions.len() && overflow.objects == 0 {
1332 return None;
1333 }
1334 let count = |disposition: ContextDispositionV1| {
1335 decisions
1336 .iter()
1337 .filter(|decision| decision.disposition == disposition)
1338 .count()
1339 };
1340 let blocked = count(ContextDispositionV1::Deny);
1344 let quarantined = count(ContextDispositionV1::Quarantine);
1345 let permitted = saturating_u32(
1346 decisions
1347 .iter()
1348 .filter(|decision| decision.object != note && decision.disposition.delivers_content())
1349 .count(),
1350 )
1351 .saturating_add(overflow.permitted);
1352 let redactions: u32 = records
1353 .iter()
1354 .map(Recorded::redactions)
1355 .fold(overflow.redactions, u32::saturating_add);
1356 let injection: u32 = records
1357 .iter()
1358 .map(|record| record.counts.injection)
1359 .fold(overflow.injection, u32::saturating_add);
1360 let incomplete = saturating_u32(
1361 records
1362 .iter()
1363 .filter(|record| record.counts.incomplete > 0)
1364 .count(),
1365 )
1366 .saturating_add(overflow.incomplete);
1367
1368 let object_key = super::receipt_store::keyed::object_key().ok()?;
1372 let decisions: Vec<ContextDecisionV1> = decisions
1373 .into_iter()
1374 .map(|mut decision| {
1375 decision.object = object_key.object(&decision.object);
1376 decision
1377 })
1378 .collect();
1379 let outcome = if permitted == 0 {
1380 DeliveryOutcomeV1::Withheld
1381 } else if delivered.is_error {
1382 DeliveryOutcomeV1::Failed
1383 } else {
1384 DeliveryOutcomeV1::Delivered
1385 };
1386 let final_context =
1387 (outcome == DeliveryOutcomeV1::Delivered).then(|| sha256(delivered.text.as_bytes()));
1388 let principal = presenter_ref
1389 .and_then(|presenter| ProtocolReference::new(presenter).ok())
1390 .and_then(|id| PrincipalV1::new(PrincipalKindV1::Session, id).ok())
1391 .or_else(|| {
1392 identity
1393 .agent_id
1394 .and_then(|agent| reference("agent:", agent))
1395 .and_then(|id| PrincipalV1::new(PrincipalKindV1::Agent, id).ok())
1396 })
1397 .unwrap_or_else(PrincipalV1::unknown);
1398 let receipt_seed = format!(
1399 "{}|{}|{}|{:?}",
1400 policy_digest.as_str(),
1401 decisions
1402 .iter()
1403 .map(|decision| decision.object.as_str())
1404 .collect::<Vec<_>>()
1405 .join(","),
1406 final_context.as_ref().map_or("", Sha256Digest::as_str),
1407 capture.started
1408 );
1409 let receipt = ContextDecisionReceiptV1 {
1410 schema_version: RECEIPT_V2_SCHEMA_VERSION,
1411 object_digest: Some(object_key.descriptor()),
1412 receipt_id: reference(
1413 "rcpt-",
1414 &blake3::hash(receipt_seed.as_bytes()).to_hex()[..32],
1415 )?,
1416 mode,
1417 principal,
1418 task: capture
1419 .task
1420 .lock()
1421 .unwrap_or_else(PoisonError::into_inner)
1422 .as_ref()
1423 .and_then(|(task, _)| TaskId::new(task.as_str()).ok()),
1424 destination: identity
1428 .destination
1429 .clone()
1430 .unwrap_or_else(|| DestinationV1 {
1431 provider: ProtocolReference::new("mcp-host").expect("valid provider ref"),
1432 model: None,
1433 locality: DestinationLocalityV1::Unknown,
1434 organization_managed: false,
1435 account_ref: None,
1436 region: None,
1437 }),
1438 policy: Some(PolicyRefV1 {
1439 id: PolicyId::new("context-gateway").expect("valid policy id"),
1440 version: None,
1441 digest: policy_digest,
1442 }),
1443 sources: SourceCountsV1 {
1444 inspected: saturating_u32(decisions.len() - notes).saturating_add(overflow.objects),
1445 permitted,
1446 selected: permitted,
1447 blocked: saturating_u32(blocked).saturating_add(overflow.blocked),
1448 },
1449 decisions,
1450 security: SecurityCountsV1 {
1451 redactions,
1452 blocked_objects: saturating_u32(blocked),
1453 quarantined_objects: saturating_u32(quarantined),
1454 injection_signals: injection,
1455 incomplete_coverage: incomplete,
1456 },
1457 tokens: TokenAccountV1 {
1458 original: capture
1461 .original_tokens
1462 .lock()
1463 .unwrap_or_else(PoisonError::into_inner)
1464 .filter(|tokens| *tokens > 0)
1465 .unwrap_or(delivered.tokens),
1466 delivered: delivered.tokens,
1467 },
1468 final_context,
1469 outcome,
1470 duration_us: u64::try_from(capture.started.elapsed().as_micros()).unwrap_or(u64::MAX),
1471 quality: None,
1472 };
1473 receipt.validate().ok()?;
1474 Some(receipt)
1475}