use super::protocol::{
Binding, Digest, EvaluationResult, Id, Request, Response, Stage, Status, SuccessResponse,
Verdict, WirePath,
};
use crate::pack::evaluator::{Enforcement, Kind};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
pub const MAX_ATTEMPTS: usize = 2048;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Policy {
pub kind: Kind,
pub enforcement: Enforcement,
pub mission_policy_digest: Digest,
pub mechanical_prerequisites_passed: bool,
}
impl Policy {
pub fn digest(&self) -> Digest {
Digest::of(&serde_json::to_vec(self).expect("typed policy is serializable"))
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct RetainedArtifact {
pub path: WirePath,
pub raw_digest: Digest,
pub retained_digest: Digest,
pub retained_bytes: u64,
pub transformation: String,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Requested {
pub request: Request,
pub policy: Policy,
pub retained_inputs: Vec<RetainedArtifact>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub permission_request_id: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "status", rename_all = "kebab-case", deny_unknown_fields)]
pub enum Outcome {
Evaluated {
result: Box<EvaluationResult>,
#[serde(rename = "rawStdoutDigest")]
raw_stdout_digest: Digest,
},
Error {
message: String,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Finished {
pub attempt_id: Id,
pub outcome: Outcome,
pub exit_code: Option<i32>,
pub cleanup_confirmed: bool,
pub artifacts: Vec<RetainedArtifact>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Disposition {
Proceed,
Block,
RequireHuman,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Consent {
pub actor: crate::live_permission::Actor,
pub allow: bool,
pub reference: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Resolution {
pub id: Id,
pub attempt_id: Id,
pub binding: Binding,
pub disposition: Disposition,
pub rationale: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub consent: Option<Consent>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Action {
ApprovePlan,
AnswerPermission,
AcceptMilestone,
AcceptDeliverable,
AdvanceLocalBase,
}
impl Action {
fn stage(&self) -> Stage {
match self {
Self::ApprovePlan => Stage::PlanApproval,
Self::AnswerPermission => Stage::CommandPermission,
Self::AcceptMilestone => Stage::MilestoneValidation,
Self::AcceptDeliverable => Stage::FinalGate,
Self::AdvanceLocalBase => Stage::Merge,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct Consumed {
pub attempt_id: Id,
pub resolution_id: Id,
pub rechecked_binding: Binding,
pub action: Action,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct Record {
pub requested: Requested,
pub requested_at: DateTime<Utc>,
pub requested_seq: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished: Option<Finished>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub finished_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolution: Option<Resolution>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resolved_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub consumed: Option<Consumed>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub closed: Option<String>,
}
impl Record {
pub fn new(requested: Requested, at: DateTime<Utc>, seq: u64) -> Result<Self, String> {
requested.request.validate()?;
if requested.policy.digest() != requested.request.params.binding.policy_digest {
return Err("gate policy bytes differ from the request binding".into());
}
if requested.policy.kind == Kind::Judgment
&& !requested.policy.mechanical_prerequisites_passed
{
return Err("judgment requires successful mechanical prerequisites".into());
}
if at >= deadline(&requested.request)? {
return Err("gate request is already expired".into());
}
validate_artifacts(&requested.retained_inputs)?;
Ok(Self {
requested,
requested_at: at,
requested_seq: seq,
finished: None,
finished_at: None,
resolution: None,
resolved_at: None,
consumed: None,
closed: None,
})
}
pub fn finish(&mut self, finished: Finished, at: DateTime<Utc>) -> Result<(), String> {
let request = &self.requested.request;
if self.closed.is_some()
|| self.finished.is_some()
|| finished.attempt_id != request.params.attempt_id
|| at < self.requested_at
{
return Err("duplicate, foreign or out-of-order gate result".into());
}
validate_artifacts(&finished.artifacts)?;
match &finished.outcome {
Outcome::Evaluated { result, .. } => {
if at >= deadline(request)?
|| !finished.cleanup_confirmed
|| finished.exit_code != Some(0)
{
return Err("evaluation expired or process completion was not proved".into());
}
let wire = Response::Result(Box::new(SuccessResponse {
jsonrpc: "2.0".into(),
id: finished.attempt_id.clone(),
result: *result.clone(),
}));
Response::from_bytes(&serde_json::to_vec(&wire).map_err(|e| e.to_string())?)?
.correlate(request)?;
}
Outcome::Error { message } => bounded(message, 8192)?,
}
self.finished = Some(finished);
self.finished_at = Some(at);
Ok(())
}
pub fn disposition(&self, consent: Option<&Consent>) -> Result<Disposition, String> {
if self.closed.is_some() {
return Ok(Disposition::Block);
}
let finished = self
.finished
.as_ref()
.ok_or("gate has no terminal attempt result")?;
if let Some(consent) = consent {
bounded(&consent.reference, 1024)?;
match &consent.actor {
crate::live_permission::Actor::Policy => {
return Err("policy is not operator consent".into())
}
crate::live_permission::Actor::SlackUser(id) => bounded(id, 256)?,
_ => {}
}
if !consent.allow {
return Ok(Disposition::Block);
}
}
if !self.requested.policy.mechanical_prerequisites_passed || !finished.cleanup_confirmed {
return Ok(Disposition::Block);
}
match &finished.outcome {
Outcome::Evaluated { result, .. } if result.status == Status::Escalate => {
return Ok(Disposition::RequireHuman)
}
Outcome::Evaluated { result, .. }
if result.verdict == Some(Verdict::Fail)
&& self.requested.policy.enforcement == Enforcement::Blocking =>
{
return Ok(Disposition::Block)
}
Outcome::Error { .. } if self.requested.policy.enforcement == Enforcement::Blocking => {
return Ok(Disposition::Block)
}
_ => {}
}
if matches!(
self.requested.request.params.stage,
Stage::PlanApproval | Stage::CommandPermission | Stage::Merge
) && consent.is_none()
{
return Ok(Disposition::RequireHuman);
}
Ok(Disposition::Proceed)
}
pub fn resolve(&mut self, resolution: Resolution, at: DateTime<Utc>) -> Result<(), String> {
let request = &self.requested.request;
if self.closed.is_some()
|| self.finished_at.is_none_or(|finished| at < finished)
|| self.resolution.is_some()
|| resolution.attempt_id != request.params.attempt_id
|| resolution.binding != request.params.binding
{
return Err("duplicate, foreign or changed gate resolution".into());
}
bounded(&resolution.rationale, 8192)?;
if resolution.disposition != self.disposition(resolution.consent.as_ref())? {
return Err("gate disposition violates recorded policy or required consent".into());
}
self.resolution = Some(resolution);
self.resolved_at = Some(at);
Ok(())
}
pub fn consume(&mut self, consumed: Consumed, at: DateTime<Utc>) -> Result<(), String> {
let resolution = self.resolution.as_ref().ok_or("gate has no resolution")?;
let request = &self.requested.request;
if self.closed.is_some()
|| self.consumed.is_some()
|| resolution.disposition != Disposition::Proceed
|| at >= deadline(request)?
|| self.resolved_at.is_none_or(|resolved| at < resolved)
|| consumed.attempt_id != request.params.attempt_id
|| consumed.resolution_id != resolution.id
|| consumed.rechecked_binding != request.params.binding
|| consumed.action.stage() != request.params.stage
{
return Err("stale, duplicate or mismatched gate consumption".into());
}
self.consumed = Some(consumed);
Ok(())
}
pub fn close(&mut self, reason: String) -> Result<(), String> {
if self.closed.is_some() || self.consumed.is_some() {
return Err("gate attempt is already closed or consumed".into());
}
bounded(&reason, 8192)?;
self.closed = Some(reason);
Ok(())
}
}
fn deadline(request: &Request) -> Result<DateTime<Utc>, String> {
DateTime::parse_from_rfc3339(&request.params.deadline)
.map(|d| d.with_timezone(&Utc))
.map_err(|e| e.to_string())
}
fn bounded(value: &str, max: usize) -> Result<(), String> {
if value.trim().is_empty() || value.len() > max {
Err("invalid bounded gate text".into())
} else {
Ok(())
}
}
fn validate_artifacts(artifacts: &[RetainedArtifact]) -> Result<(), String> {
if artifacts.len() > 10_000 {
return Err("gate retention inventory exceeds limit".into());
}
super::protocol::validate_paths(artifacts.iter().map(|a| a.path.as_str()))?;
let mut names = std::collections::BTreeSet::new();
for artifact in artifacts {
if artifact.retained_bytes > 64 * 1024 * 1024
|| !artifact.path.as_str().starts_with("runs/gates/")
|| !names.insert(artifact.path.as_str().to_ascii_lowercase())
{
return Err("invalid retained gate artifact".into());
}
bounded(&artifact.transformation, 512)?;
}
Ok(())
}
pub(crate) fn fold(
state: &mut crate::types::MissionState,
event: &crate::events::Event,
) -> crate::error::Result<()> {
use crate::events::EventKind;
let invalid = |message: String| crate::error::EngineError::InvalidState(message);
if event.mission_id != state.mission.id {
return Err(invalid("foreign gate mission".into()));
}
match &event.kind {
EventKind::GateEvaluationClosed { attempt_id, reason } => {
state
.gate_evaluations
.get_mut(attempt_id.as_str())
.ok_or_else(|| invalid("closed gate has no request".into()))?
.close(reason.clone())
.map_err(invalid)?;
}
EventKind::GateEvaluationRequested { evaluation } => {
let request = &evaluation.request;
validate_stage_status(state, request.params.stage).map_err(invalid)?;
if state.gate_evaluations.len() >= MAX_ATTEMPTS
|| request.params.mission_id.as_str() != event.mission_id
|| state
.gate_evaluations
.contains_key(request.params.attempt_id.as_str())
{
return Err(invalid("duplicate, foreign or excess gate request".into()));
}
validate_permission_join(state, evaluation).map_err(invalid)?;
let record = Record::new(*evaluation.clone(), event.ts, event.seq).map_err(invalid)?;
state
.gate_evaluations
.insert(request.params.attempt_id.as_str().into(), record);
}
EventKind::GateEvaluationFinished { evaluation } => {
state
.gate_evaluations
.get_mut(evaluation.attempt_id.as_str())
.ok_or_else(|| invalid("gate result has no request".into()))?
.finish(*evaluation.clone(), event.ts)
.map_err(invalid)?;
}
EventKind::GateResolutionRecorded { resolution } => {
let record = state
.gate_evaluations
.get(resolution.attempt_id.as_str())
.ok_or_else(|| invalid("gate resolution has no request".into()))?;
if let (Some(id), Some(consent)) =
(&record.requested.permission_request_id, &resolution.consent)
{
let actual = state
.permissions
.get(id)
.and_then(|permission| permission.resolution.as_ref())
.ok_or_else(|| invalid("permission consent is absent".into()))?;
if actual.actor != consent.actor
|| actual.allow != consent.allow
|| consent.reference != *id
{
return Err(invalid(
"gate cannot replace or relabel permission consent".into(),
));
}
}
if state.gate_evaluations.values().any(|record| {
record
.resolution
.as_ref()
.is_some_and(|r| r.id == resolution.id)
}) {
return Err(invalid("gate resolution ID was reused".into()));
}
state
.gate_evaluations
.get_mut(resolution.attempt_id.as_str())
.ok_or_else(|| invalid("gate resolution has no request".into()))?
.resolve(resolution.clone(), event.ts)
.map_err(invalid)?;
}
EventKind::GateResolutionConsumed { consumption } => {
let record = state
.gate_evaluations
.get(consumption.attempt_id.as_str())
.ok_or_else(|| invalid("gate consumption has no request".into()))?;
validate_stage_status(state, record.requested.request.params.stage).map_err(invalid)?;
validate_permission_join(state, &record.requested).map_err(invalid)?;
if let Some(id) = &record.requested.permission_request_id {
let permission = &state.permissions[id];
let actual = permission
.resolution
.as_ref()
.ok_or_else(|| invalid("permission consent is absent".into()))?;
let consent = record
.resolution
.as_ref()
.and_then(|r| r.consent.as_ref())
.ok_or_else(|| invalid("permission consent join is absent".into()))?;
if !actual.allow
|| !consent.allow
|| actual.actor != consent.actor
|| consent.reference != *id
{
return Err(invalid(
"gate cannot replace or relabel permission consent".into(),
));
}
}
if state
.consumed_gate_resolutions
.contains(consumption.resolution_id.as_str())
{
return Err(invalid("gate resolution was already consumed".into()));
}
state
.gate_evaluations
.get_mut(consumption.attempt_id.as_str())
.ok_or_else(|| invalid("gate consumption has no request".into()))?
.consume(consumption.clone(), event.ts)
.map_err(invalid)?;
state
.consumed_gate_resolutions
.insert(consumption.resolution_id.as_str().into());
}
_ => return Err(invalid("not a gate lifecycle event".into())),
}
Ok(())
}
fn validate_permission_join(
state: &crate::types::MissionState,
evaluation: &Requested,
) -> Result<(), String> {
use super::protocol::Subject;
match (
&evaluation.request.params.subject,
&evaluation.permission_request_id,
) {
(
Subject::Invocation {
run_id,
peer_session_id,
tool_call_id,
peer_request_id,
action_digest,
options_digest,
cwd_id,
},
Some(id),
) => {
let record = state
.permissions
.get(id)
.ok_or("gate invocation has no live permission request")?;
let permission = &record.request;
let run = state
.runs
.get(&permission.binding.run_id)
.ok_or("permission run is absent")?;
let workspace = workspace_id(&permission.binding.workspace);
let peer = serde_json::to_value(peer_request_id).map_err(|e| e.to_string())?;
if record.closed.is_some()
|| record.delivery.is_some()
|| run.ended_at.is_some()
|| evaluation.request.params.binding.workspace_id != workspace
|| *cwd_id != workspace
|| evaluation.request.params.binding.plan_digest.as_str()
!= format!("sha256:{}", permission.binding.plan_digest)
|| evaluation.policy.mission_policy_digest.as_str()
!= format!("sha256:{}", permission.binding.policy_digest)
|| deadline(&evaluation.request)? > permission.proposal.deadline
|| permission.binding.run_id != run_id.as_str()
|| permission.proposal.peer_session_id != peer_session_id.as_str()
|| permission.proposal.tool_call_id != tool_call_id.as_str()
|| permission.proposal.peer_request_id != peer
|| action_digest.as_str() != format!("sha256:{}", permission.proposal.action_digest)
|| options_digest.as_str()
!= format!("sha256:{}", permission.proposal.options_digest)
{
return Err("gate invocation does not match its permission request".into());
}
}
(Subject::Invocation { .. }, None) => {
return Err("gate invocation requires a permission join".into())
}
(_, Some(_)) => return Err("only invocation evaluations may join a permission".into()),
_ => {}
}
Ok(())
}
pub fn workspace_id(path: &str) -> Id {
Id::try_from(format!(
"workspace-{}",
&Digest::of(path.as_bytes()).as_str()[7..]
))
.expect("a SHA-256 workspace label is a valid opaque ID")
}
fn validate_stage_status(state: &crate::types::MissionState, stage: Stage) -> Result<(), String> {
use crate::types::MissionStatus;
let ready = match stage {
Stage::PlanApproval => {
state.mission.status == MissionStatus::Planning
|| (state.pending_revision.is_some()
&& matches!(
state.mission.status,
MissionStatus::Running | MissionStatus::Blocked
))
}
Stage::CommandPermission | Stage::MilestoneValidation => matches!(
state.mission.status,
MissionStatus::Running | MissionStatus::Validating
),
Stage::FinalGate => state.mission.status == MissionStatus::Validating,
Stage::Merge => state.mission.status == MissionStatus::Complete,
};
if ready {
Ok(())
} else {
Err("gate stage does not match the mission lifecycle".into())
}
}