lean_ctx_protocol/hive/
compute.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Optional Task Compute (Release C, BAUPLAN v1.1 §9; C0 2.x
4//! `schemas/compute.schema.json`, `routes/compute_*`,
5//! `vectors/compute_cost.vectors.json`).
6//!
7//! An existing agent hands one clearly defined job (tests, static analysis,
8//! an index build) to an isolated, short-lived sandbox and gets a bounded
9//! result back. There is no Agent Compute: no model loop runs on the server.
10//!
11//! Engine routes (all with `Hive-Protocol: 2.x`; a server with Task Compute
12//! switched off answers every one `404 feature_disabled`, feature `compute`):
13//!
14//! - E47 `POST /v1/compute/jobs` ([`ComputeSubmitRequestV1`] →
15//!   [`ComputeJobResponseV1`], 201 new, 200 the job of the same `client_ref`)
16//! - E48 `GET /v1/compute/jobs/{job_id}` → [`ComputeJobResponseV1`]
17//! - E49 `POST /v1/compute/jobs/{job_id}/cancel` → [`ComputeJobResponseV1`]
18//! - E50 `GET /v1/compute/jobs/{job_id}/result` → [`ComputeResultResponseV1`]
19//!   (409 `invalid_state` while the job is queued or running)
20//! - E51 `PUT /v1/compute/bundles/{digest}` (`application/octet-stream`, at
21//!   most [`MAX_BUNDLE_BYTES`]) → [`ComputeBundleResponseV1`]
22//! - E52 `GET /v1/compute/jobs/{job_id}/blobs/{digest}`: the bytes of one
23//!   artifact or of the logs of that job
24//!
25//! Money: prepaid credit only, 1 credit = 1 vCPU-minute. The maximum cost
26//! ([`reserve_for`]) is reserved at submit and must fit the job's budget; the
27//! real cost ([`spent_for`]) is settled at the end and the rest refunded.
28//! Refusals: `402 credit_exhausted`, `409 cap_reached`, `404
29//! feature_disabled` ([`ComputeRefusalV1`]).
30//!
31//! Requests refuse unknown fields; responses accept them.
32
33use std::collections::BTreeMap;
34
35use serde::{Deserialize, Serialize};
36use serde_json::Value;
37use sha2::{Digest, Sha256};
38
39use super::require_schema;
40use crate::ValidationError;
41
42/// `schema` of an E47 submit request.
43pub const COMPUTE_JOB_SCHEMA: &str = "leanctx.hive.compute-job/v1";
44/// `feature` of the `feature_disabled` refusal.
45pub const COMPUTE_FEATURE: &str = "compute";
46/// Largest input bundle, artifact and upload body (64 MiB).
47pub const MAX_BUNDLE_BYTES: u64 = 64 * 1024 * 1024;
48/// Bundles per job.
49pub const MAX_INPUT_BUNDLES: usize = 16;
50/// `inputs.version` this engine writes and reads.
51pub const INPUTS_VERSION: u8 = 1;
52/// Command bounds.
53pub const MAX_ARGV: usize = 32;
54pub const MAX_ARG_BYTES: usize = 512;
55pub const MAX_WORKDIR_BYTES: usize = 128;
56/// Resource bounds of one job.
57pub const CPU_MIN: u32 = 1;
58pub const CPU_MAX: u32 = 8;
59pub const MEMORY_MB_MIN: u32 = 256;
60pub const MEMORY_MB_MAX: u32 = 16_384;
61pub const DISK_MB_MIN: u32 = 256;
62pub const DISK_MB_MAX: u32 = 20_480;
63pub const RUNTIME_S_MIN: u32 = 10;
64pub const RUNTIME_S_MAX: u32 = 3600;
65/// Hosts a job may reach (TCP 443); empty = no network.
66pub const MAX_NETWORK_HOSTS: usize = 16;
67/// Budget bounds of one job, in credits.
68pub const BUDGET_MIN: u64 = 1;
69pub const BUDGET_MAX: u64 = 1000;
70/// Result bounds.
71pub const MAX_ARTIFACTS: usize = 32;
72pub const MAX_LOG_BYTES: u64 = 262_144;
73pub const MAX_SUMMARY_TEXT_CHARS: usize = 2000;
74pub const MAX_SUMMARY_COUNTS: usize = 16;
75/// `rate.credits_per_cpu_minute` (fixed by the contract).
76pub const CREDITS_PER_CPU_MINUTE: u64 = 1;
77
78/// `tests`, `static_analysis` or `index_build`.
79#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
80#[serde(rename_all = "snake_case")]
81pub enum JobType {
82    Tests,
83    StaticAnalysis,
84    IndexBuild,
85}
86
87impl JobType {
88    pub const ALL: [JobType; 3] = [Self::Tests, Self::StaticAnalysis, Self::IndexBuild];
89
90    pub fn as_str(self) -> &'static str {
91        match self {
92            Self::Tests => "tests",
93            Self::StaticAnalysis => "static_analysis",
94            Self::IndexBuild => "index_build",
95        }
96    }
97
98    pub fn parse(value: &str) -> Option<Self> {
99        Self::ALL.into_iter().find(|t| t.as_str() == value)
100    }
101
102    /// The result formats the contract allows for this type.
103    pub fn formats(self) -> &'static [ResultFormat] {
104        match self {
105            Self::Tests => &[ResultFormat::JunitXml, ResultFormat::JsonSummary],
106            Self::StaticAnalysis => &[ResultFormat::Sarif, ResultFormat::JsonSummary],
107            Self::IndexBuild => &[ResultFormat::ArtifactBundle],
108        }
109    }
110
111    /// The format a job of this type gets when the caller names none.
112    pub fn default_format(self) -> ResultFormat {
113        self.formats()[0]
114    }
115}
116
117/// `queued -> running -> succeeded | failed | cancelled | expired`.
118#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
119#[serde(rename_all = "snake_case")]
120pub enum JobState {
121    Queued,
122    Running,
123    Succeeded,
124    Failed,
125    Cancelled,
126    Expired,
127}
128
129impl JobState {
130    pub const ALL: [JobState; 6] = [
131        Self::Queued,
132        Self::Running,
133        Self::Succeeded,
134        Self::Failed,
135        Self::Cancelled,
136        Self::Expired,
137    ];
138
139    pub fn as_str(self) -> &'static str {
140        match self {
141            Self::Queued => "queued",
142            Self::Running => "running",
143            Self::Succeeded => "succeeded",
144            Self::Failed => "failed",
145            Self::Cancelled => "cancelled",
146            Self::Expired => "expired",
147        }
148    }
149
150    pub fn is_terminal(self) -> bool {
151        !matches!(self, Self::Queued | Self::Running)
152    }
153
154    /// Whether a job in this state carries a terminal reason.
155    pub fn has_reason(self) -> bool {
156        matches!(self, Self::Failed | Self::Cancelled | Self::Expired)
157    }
158}
159
160/// Why a job is failed, cancelled or expired.
161#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
162#[serde(rename_all = "snake_case")]
163pub enum TerminalReason {
164    ExitNonzero,
165    ProviderError,
166    RunnerLost,
167    IsolationRefused,
168    Cancelled,
169    QueueTimeout,
170    RuntimeExceeded,
171}
172
173impl TerminalReason {
174    pub const ALL: [TerminalReason; 7] = [
175        Self::ExitNonzero,
176        Self::ProviderError,
177        Self::RunnerLost,
178        Self::IsolationRefused,
179        Self::Cancelled,
180        Self::QueueTimeout,
181        Self::RuntimeExceeded,
182    ];
183
184    pub fn as_str(self) -> &'static str {
185        match self {
186            Self::ExitNonzero => "exit_nonzero",
187            Self::ProviderError => "provider_error",
188            Self::RunnerLost => "runner_lost",
189            Self::IsolationRefused => "isolation_refused",
190            Self::Cancelled => "cancelled",
191            Self::QueueTimeout => "queue_timeout",
192            Self::RuntimeExceeded => "runtime_exceeded",
193        }
194    }
195
196    /// One line for people.
197    pub fn describe(self) -> &'static str {
198        match self {
199            Self::ExitNonzero => "the command exited non-zero",
200            Self::ProviderError => "the sandbox could not run it",
201            Self::RunnerLost => "the node running it stopped",
202            Self::IsolationRefused => "the job broke an isolation rule before anything ran",
203            Self::Cancelled => "cancelled",
204            Self::QueueTimeout => "not started before its queue deadline",
205            Self::RuntimeExceeded => "stopped at its maximum runtime",
206        }
207    }
208}
209
210/// What a job's result carries.
211#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
212#[serde(rename_all = "snake_case")]
213pub enum ResultFormat {
214    JunitXml,
215    Sarif,
216    JsonSummary,
217    ArtifactBundle,
218}
219
220impl ResultFormat {
221    pub const ALL: [ResultFormat; 4] = [
222        Self::JunitXml,
223        Self::Sarif,
224        Self::JsonSummary,
225        Self::ArtifactBundle,
226    ];
227
228    pub fn as_str(self) -> &'static str {
229        match self {
230            Self::JunitXml => "junit_xml",
231            Self::Sarif => "sarif",
232            Self::JsonSummary => "json_summary",
233            Self::ArtifactBundle => "artifact_bundle",
234        }
235    }
236
237    pub fn parse(value: &str) -> Option<Self> {
238        Self::ALL.into_iter().find(|f| f.as_str() == value)
239    }
240}
241
242/// `summary.outcome`.
243#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
244#[serde(rename_all = "snake_case")]
245pub enum Outcome {
246    Passed,
247    Failed,
248    Error,
249}
250
251impl Outcome {
252    pub const ALL: [Outcome; 3] = [Self::Passed, Self::Failed, Self::Error];
253
254    pub fn as_str(self) -> &'static str {
255        match self {
256            Self::Passed => "passed",
257            Self::Failed => "failed",
258            Self::Error => "error",
259        }
260    }
261}
262
263/// The hard cap a `cap_reached` refusal names.
264#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
265#[serde(rename_all = "snake_case")]
266pub enum CapName {
267    MaxRuntimeS,
268    MaxJobCredits,
269    JobBudget,
270    DailyCredits,
271    QueuedJobs,
272    BundleBytes,
273}
274
275impl CapName {
276    pub const ALL: [CapName; 6] = [
277        Self::MaxRuntimeS,
278        Self::MaxJobCredits,
279        Self::JobBudget,
280        Self::DailyCredits,
281        Self::QueuedJobs,
282        Self::BundleBytes,
283    ];
284
285    pub fn as_str(self) -> &'static str {
286        match self {
287            Self::MaxRuntimeS => "max_runtime_s",
288            Self::MaxJobCredits => "max_job_credits",
289            Self::JobBudget => "job_budget",
290            Self::DailyCredits => "daily_credits",
291            Self::QueuedJobs => "queued_jobs",
292            Self::BundleBytes => "bundle_bytes",
293        }
294    }
295
296    /// The cap as people read it.
297    pub fn describe(self) -> &'static str {
298        match self {
299            Self::MaxRuntimeS => "maximum runtime per job (seconds)",
300            Self::MaxJobCredits => "maximum credits per job",
301            Self::JobBudget => "this job's budget (credits)",
302            Self::DailyCredits => "daily compute credits",
303            Self::QueuedJobs => "queued jobs",
304            Self::BundleBytes => "stored input bundles (bytes)",
305        }
306    }
307}
308
309/// One uploaded input bundle, by digest.
310#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
311#[serde(deny_unknown_fields)]
312pub struct BundleRefV1 {
313    pub name: String,
314    pub digest: String,
315    pub size_bytes: u64,
316}
317
318impl BundleRefV1 {
319    pub fn validate(&self) -> Result<(), ValidationError> {
320        check(is_name(&self.name), "inputs.bundles[].name")?;
321        check(is_digest(&self.digest), "inputs.bundles[].digest")?;
322        check(
323            (1..=MAX_BUNDLE_BYTES).contains(&self.size_bytes),
324            "inputs.bundles[].size_bytes",
325        )
326    }
327}
328
329/// Versioned, content-addressed inputs.
330#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
331#[serde(deny_unknown_fields)]
332pub struct ComputeInputsV1 {
333    pub version: u8,
334    pub bundles: Vec<BundleRefV1>,
335}
336
337impl ComputeInputsV1 {
338    pub fn validate(&self) -> Result<(), ValidationError> {
339        check(self.version == INPUTS_VERSION, "inputs.version")?;
340        check(
341            (1..=MAX_INPUT_BUNDLES).contains(&self.bundles.len()),
342            "inputs.bundles",
343        )?;
344        self.bundles.iter().try_for_each(BundleRefV1::validate)
345    }
346}
347
348/// The command the sandbox runs.
349#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
350#[serde(deny_unknown_fields)]
351pub struct ComputeCommandV1 {
352    pub argv: Vec<String>,
353    /// Relative to `/work`, no `..`; empty = `/work`.
354    #[serde(default)]
355    pub workdir: String,
356}
357
358impl ComputeCommandV1 {
359    pub fn validate(&self) -> Result<(), ValidationError> {
360        check(
361            (1..=MAX_ARGV).contains(&self.argv.len()),
362            "command.argv (1-32 items)",
363        )?;
364        check(
365            self.argv
366                .iter()
367                .all(|a| (1..=MAX_ARG_BYTES).contains(&a.chars().count())),
368            "command.argv items (1-512 characters)",
369        )?;
370        check(is_workdir(&self.workdir), "command.workdir")
371    }
372}
373
374/// vCPUs, memory and scratch disk of one job.
375#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
376#[serde(deny_unknown_fields)]
377pub struct ComputeResourcesV1 {
378    pub cpu: u32,
379    pub memory_mb: u32,
380    pub disk_mb: u32,
381}
382
383impl ComputeResourcesV1 {
384    pub fn validate(&self) -> Result<(), ValidationError> {
385        check(
386            (CPU_MIN..=CPU_MAX).contains(&self.cpu),
387            "resources.cpu (1-8)",
388        )?;
389        check(
390            (MEMORY_MB_MIN..=MEMORY_MB_MAX).contains(&self.memory_mb),
391            "resources.memory_mb (256-16384)",
392        )?;
393        check(
394            (DISK_MB_MIN..=DISK_MB_MAX).contains(&self.disk_mb),
395            "resources.disk_mb (256-20480)",
396        )
397    }
398}
399
400/// The most a job may cost.
401#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
402#[serde(deny_unknown_fields)]
403pub struct ComputeBudgetV1 {
404    pub credits: u64,
405}
406
407/// E47 `POST /v1/compute/jobs`.
408#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
409#[serde(deny_unknown_fields)]
410pub struct ComputeSubmitRequestV1 {
411    pub schema: String,
412    pub vault_id: String,
413    #[serde(rename = "type")]
414    pub job_type: JobType,
415    pub inputs: ComputeInputsV1,
416    pub image: String,
417    pub command: ComputeCommandV1,
418    pub resources: ComputeResourcesV1,
419    pub max_runtime_s: u32,
420    pub network_allowlist: Vec<String>,
421    pub budget: ComputeBudgetV1,
422    pub result_format: ResultFormat,
423    #[serde(default, skip_serializing_if = "Option::is_none")]
424    pub client_ref: Option<String>,
425}
426
427impl ComputeSubmitRequestV1 {
428    /// Every rule of `submit_request` the engine can check before sending,
429    /// the job budget against the reserve included (`cap_reached` cap
430    /// `job_budget` on the server).
431    pub fn validate(&self) -> Result<(), ValidationError> {
432        require_schema(&self.schema, COMPUTE_JOB_SCHEMA)?;
433        check(is_lower_hex(&self.vault_id, 32), "vault_id")?;
434        self.inputs.validate()?;
435        check(is_image(&self.image), "image")?;
436        self.command.validate()?;
437        self.resources.validate()?;
438        check(
439            (RUNTIME_S_MIN..=RUNTIME_S_MAX).contains(&self.max_runtime_s),
440            "max_runtime_s (10-3600)",
441        )?;
442        validate_network(&self.network_allowlist)?;
443        check(
444            (BUDGET_MIN..=BUDGET_MAX).contains(&self.budget.credits),
445            "budget.credits (1-1000)",
446        )?;
447        check(
448            self.job_type.formats().contains(&self.result_format),
449            "result_format (not allowed for this job type)",
450        )?;
451        if let Some(client_ref) = &self.client_ref {
452            check(is_client_ref(client_ref), "client_ref")?;
453        }
454        budget_covers(self.resources.cpu, self.max_runtime_s, self.budget.credits).map(|_| ())
455    }
456}
457
458/// `rate` of a cost record.
459#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
460pub struct ComputeRateV1 {
461    pub credits_per_cpu_minute: u64,
462}
463
464/// The job's cost record.
465#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
466pub struct ComputeCostV1 {
467    pub credits_reserved: u64,
468    pub credits_spent: Option<u64>,
469    pub credits_refunded: Option<u64>,
470    pub wall_seconds: Option<u64>,
471    pub rate: ComputeRateV1,
472}
473
474impl ComputeCostV1 {
475    fn validate(&self, terminal: bool) -> Result<(), ValidationError> {
476        check(self.credits_reserved >= 1, "cost.credits_reserved")?;
477        check(
478            self.rate.credits_per_cpu_minute == CREDITS_PER_CPU_MINUTE,
479            "cost.rate.credits_per_cpu_minute",
480        )?;
481        match (terminal, self.credits_spent, self.credits_refunded) {
482            (true, Some(spent), Some(refunded)) => check(
483                spent.checked_add(refunded) == Some(self.credits_reserved),
484                "cost (spent + refunded must equal reserved)",
485            ),
486            (false, None, None) => Ok(()),
487            (true, _, _) => Err(ValidationError::new(
488                "cost: a finished job carries credits_spent and credits_refunded",
489            )),
490            (false, _, _) => Err(ValidationError::new(
491                "cost: an unfinished job has no credits_spent or credits_refunded yet",
492            )),
493        }
494    }
495}
496
497/// Who submitted a job.
498#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
499pub struct SubmittedByV1 {
500    pub instance_id: Option<String>,
501    pub user_id: Option<String>,
502}
503
504/// One job as E47, E48 and E49 return it.
505#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
506pub struct ComputeJobV1 {
507    pub job_id: String,
508    pub workspace: String,
509    pub vault_id: String,
510    #[serde(rename = "type")]
511    pub job_type: JobType,
512    pub state: JobState,
513    pub reason: Option<TerminalReason>,
514    pub inputs: ComputeInputsV1,
515    pub image: String,
516    pub command: ComputeCommandV1,
517    pub resources: ComputeResourcesV1,
518    pub max_runtime_s: u32,
519    pub network_allowlist: Vec<String>,
520    pub budget: ComputeBudgetV1,
521    pub result_format: ResultFormat,
522    pub client_ref: Option<String>,
523    pub submitted_by: SubmittedByV1,
524    pub cost: ComputeCostV1,
525    pub created_ms: u64,
526    pub started_ms: Option<u64>,
527    pub finished_ms: Option<u64>,
528    pub queue_deadline_ms: u64,
529    pub run_deadline_ms: Option<u64>,
530    pub cancel_requested: bool,
531    pub artifacts_expire_ms: Option<u64>,
532    pub result_available: bool,
533}
534
535impl ComputeJobV1 {
536    pub fn validate(&self) -> Result<(), ValidationError> {
537        check(is_job_id(&self.job_id), "job_id")?;
538        check(
539            prefixed_hex(&self.workspace, "org_", 32),
540            "workspace (org_ + 32 hex)",
541        )?;
542        check(is_lower_hex(&self.vault_id, 32), "vault_id")?;
543        check(
544            self.reason.is_some() == self.state.has_reason(),
545            "reason (set exactly when failed, cancelled or expired)",
546        )?;
547        self.inputs.validate()?;
548        check(is_image(&self.image), "image")?;
549        self.command.validate()?;
550        self.resources.validate()?;
551        check(
552            (RUNTIME_S_MIN..=RUNTIME_S_MAX).contains(&self.max_runtime_s),
553            "max_runtime_s",
554        )?;
555        validate_network(&self.network_allowlist)?;
556        check(
557            (BUDGET_MIN..=BUDGET_MAX).contains(&self.budget.credits),
558            "budget.credits",
559        )?;
560        if let Some(client_ref) = &self.client_ref {
561            check(is_client_ref(client_ref), "client_ref")?;
562        }
563        if let Some(instance) = &self.submitted_by.instance_id {
564            check(
565                prefixed_hex(instance, "inst_", 32),
566                "submitted_by.instance_id",
567            )?;
568        }
569        if let Some(user) = &self.submitted_by.user_id {
570            check(is_uuid(user), "submitted_by.user_id")?;
571        }
572        match self.state {
573            JobState::Queued => check(
574                self.started_ms.is_none() && self.finished_ms.is_none(),
575                "a queued job has neither started_ms nor finished_ms",
576            )?,
577            JobState::Running => check(
578                self.started_ms.is_some()
579                    && self.finished_ms.is_none()
580                    && self.run_deadline_ms.is_some(),
581                "a running job has started_ms and run_deadline_ms, no finished_ms",
582            )?,
583            _ => check(self.finished_ms.is_some(), "a finished job has finished_ms")?,
584        }
585        self.cost.validate(self.state.is_terminal())
586    }
587}
588
589/// E47, E48 and E49 answer.
590#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
591pub struct ComputeJobResponseV1 {
592    pub job: ComputeJobV1,
593}
594
595impl ComputeJobResponseV1 {
596    pub fn validate(&self) -> Result<(), ValidationError> {
597        self.job.validate()
598    }
599}
600
601/// One artifact below `/work/out`.
602#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
603pub struct ArtifactRefV1 {
604    pub name: String,
605    pub digest: String,
606    pub size_bytes: u64,
607}
608
609/// The job's logs (stdout and stderr, the last 256 KiB, screened).
610#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
611pub struct LogsRefV1 {
612    pub digest: String,
613    pub size_bytes: u64,
614    pub truncated: bool,
615    pub redactions: u64,
616}
617
618/// The job's summary.
619#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
620pub struct SummaryV1 {
621    pub outcome: Outcome,
622    pub counts: BTreeMap<String, u64>,
623    pub text: String,
624}
625
626/// A terminal job's bounded result.
627#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
628pub struct ComputeResultV1 {
629    pub exit_code: Option<i64>,
630    pub summary: SummaryV1,
631    pub artifacts: Vec<ArtifactRefV1>,
632    pub logs: Option<LogsRefV1>,
633}
634
635impl ComputeResultV1 {
636    pub fn validate(&self) -> Result<(), ValidationError> {
637        check(
638            self.summary.counts.len() <= MAX_SUMMARY_COUNTS
639                && self.summary.counts.keys().all(|k| is_count_key(k)),
640            "result.summary.counts",
641        )?;
642        check(
643            self.summary.text.chars().count() <= MAX_SUMMARY_TEXT_CHARS,
644            "result.summary.text",
645        )?;
646        check(
647            self.artifacts.len() <= MAX_ARTIFACTS,
648            "result.artifacts (at most 32)",
649        )?;
650        let mut total: u64 = 0;
651        for artifact in &self.artifacts {
652            check(is_artifact_name(&artifact.name), "result.artifacts[].name")?;
653            check(is_digest(&artifact.digest), "result.artifacts[].digest")?;
654            check(
655                artifact.size_bytes <= MAX_BUNDLE_BYTES,
656                "result.artifacts[].size_bytes",
657            )?;
658            total = total.saturating_add(artifact.size_bytes);
659        }
660        check(
661            total <= MAX_BUNDLE_BYTES,
662            "result.artifacts (64 MiB in total)",
663        )?;
664        if let Some(logs) = &self.logs {
665            check(is_digest(&logs.digest), "result.logs.digest")?;
666            check(logs.size_bytes <= MAX_LOG_BYTES, "result.logs.size_bytes")?;
667        }
668        Ok(())
669    }
670}
671
672/// E50 answer.
673#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
674pub struct ComputeResultResponseV1 {
675    pub job_id: String,
676    pub state: JobState,
677    pub reason: Option<TerminalReason>,
678    pub result: Option<ComputeResultV1>,
679    pub cost: ComputeCostV1,
680}
681
682impl ComputeResultResponseV1 {
683    pub fn validate(&self) -> Result<(), ValidationError> {
684        check(is_job_id(&self.job_id), "job_id")?;
685        check(self.state.is_terminal(), "state (a result is terminal)")?;
686        check(
687            self.reason.is_some() == self.state.has_reason(),
688            "reason (set exactly when failed, cancelled or expired)",
689        )?;
690        if let Some(result) = &self.result {
691            result.validate()?;
692        }
693        self.cost.validate(true)
694    }
695
696    /// The artifact or log of this result with `digest`, as `(name, size)`;
697    /// the logs are named `logs.txt`.
698    pub fn blob(&self, digest: &str) -> Option<(&str, u64)> {
699        let result = self.result.as_ref()?;
700        result
701            .artifacts
702            .iter()
703            .find(|a| a.digest == digest)
704            .map(|a| (a.name.as_str(), a.size_bytes))
705            .or_else(|| {
706                result
707                    .logs
708                    .as_ref()
709                    .filter(|l| l.digest == digest)
710                    .map(|l| ("logs.txt", l.size_bytes))
711            })
712    }
713}
714
715/// E51 answer.
716#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
717pub struct ComputeBundleResponseV1 {
718    pub digest: String,
719    pub size_bytes: u64,
720    pub expires_ms: u64,
721    /// `true` stored now (201), `false` already there, its expiry renewed (200).
722    pub created: bool,
723}
724
725impl ComputeBundleResponseV1 {
726    /// The answer for an upload of `digest` with `size_bytes`.
727    pub fn validate_for(&self, digest: &str, size_bytes: u64) -> Result<(), ValidationError> {
728        check(is_digest(&self.digest), "digest")?;
729        check(self.digest == digest, "digest (not the uploaded bundle)")?;
730        check(
731            self.size_bytes == size_bytes && (1..=MAX_BUNDLE_BYTES).contains(&size_bytes),
732            "size_bytes (not the uploaded bundle)",
733        )
734    }
735}
736
737/// A compute refusal the engine words for people.
738#[derive(Debug, Clone, PartialEq, Eq)]
739pub enum ComputeRefusalV1 {
740    /// 402: the prepaid balance does not cover the job's reserve.
741    CreditExhausted { balance: u64, needed: u64 },
742    /// 409: a hard cap refuses the job; nothing was queued.
743    CapReached {
744        cap: CapName,
745        limit: u64,
746        requested: u64,
747    },
748    /// 404: Task Compute is switched off on this Hive.
749    FeatureDisabled,
750}
751
752impl ComputeRefusalV1 {
753    /// The refusal in an error body, if it is one of the three.
754    pub fn parse(body: &Value) -> Option<Self> {
755        let error = body.get("error")?;
756        let number = |key: &str| error.get(key).and_then(Value::as_u64);
757        match error.get("code")?.as_str()? {
758            "credit_exhausted" => Some(Self::CreditExhausted {
759                balance: number("balance")?,
760                needed: number("needed")?,
761            }),
762            "cap_reached" => Some(Self::CapReached {
763                cap: serde_json::from_value(error.get("cap")?.clone()).ok()?,
764                limit: number("limit")?,
765                requested: number("requested")?,
766            }),
767            "feature_disabled"
768                if error.get("feature").and_then(Value::as_str) == Some(COMPUTE_FEATURE) =>
769            {
770                Some(Self::FeatureDisabled)
771            }
772            _ => None,
773        }
774    }
775}
776
777/// `ceil(a / b)` for `b > 0`.
778fn ceil_div(a: u64, b: u64) -> u64 {
779    a / b + u64::from(!a.is_multiple_of(b))
780}
781
782/// Credits reserved at submit: `ceil(cpu × max_runtime_s / 60)`.
783pub fn reserve_for(cpu: u32, max_runtime_s: u32) -> u64 {
784    ceil_div(u64::from(cpu) * u64::from(max_runtime_s), 60) * CREDITS_PER_CPU_MINUTE
785}
786
787/// Credits spent: 0 when the job never started, else
788/// `min(reserved, max(1, ceil(cpu × wall_seconds / 60)))`.
789pub fn spent_for(cpu: u32, max_runtime_s: u32, wall_seconds: Option<u64>) -> u64 {
790    let Some(wall) = wall_seconds else {
791        return 0;
792    };
793    let used = ceil_div(u64::from(cpu).saturating_mul(wall), 60).max(1) * CREDITS_PER_CPU_MINUTE;
794    used.min(reserve_for(cpu, max_runtime_s))
795}
796
797/// The reserve when `budget` covers it; else the reserve as the error
798/// (`cap_reached` cap `job_budget` on the server).
799pub fn budget_covers(cpu: u32, max_runtime_s: u32, budget: u64) -> Result<u64, ValidationError> {
800    let reserve = reserve_for(cpu, max_runtime_s);
801    if reserve > budget {
802        return Err(ValidationError::new(format!(
803            "budget.credits {budget} does not cover the reserve of {reserve} credits (cpu {cpu} x {max_runtime_s} s / 60)"
804        )));
805    }
806    Ok(reserve)
807}
808
809/// `"sha256:" + hex(sha256(bytes))`: the content address of a bundle,
810/// artifact or log.
811pub fn content_digest(bytes: &[u8]) -> String {
812    use std::fmt::Write as _;
813    let hash = Sha256::digest(bytes);
814    let mut out = String::with_capacity(71);
815    out.push_str("sha256:");
816    for byte in hash.iter() {
817        let _ = write!(out, "{byte:02x}");
818    }
819    out
820}
821
822fn check(ok: bool, what: &str) -> Result<(), ValidationError> {
823    if ok {
824        Ok(())
825    } else {
826        Err(ValidationError::new(format!("invalid {what}")))
827    }
828}
829
830fn is_lower_hex(value: &str, len: usize) -> bool {
831    value.len() == len
832        && value
833            .bytes()
834            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
835}
836
837fn prefixed_hex(value: &str, prefix: &str, len: usize) -> bool {
838    value
839        .strip_prefix(prefix)
840        .is_some_and(|rest| is_lower_hex(rest, len))
841}
842
843fn is_uuid(value: &str) -> bool {
844    let parts: Vec<&str> = value.split('-').collect();
845    parts.len() == 5
846        && parts.iter().map(|p| p.len()).eq([8, 4, 4, 4, 12])
847        && parts.iter().all(|p| is_lower_hex(p, p.len()))
848}
849
850/// `sha256:` + 64 lowercase hex.
851pub fn is_digest(value: &str) -> bool {
852    prefixed_hex(value, "sha256:", 64)
853}
854
855/// `cj_` + 32 lowercase hex.
856pub fn is_job_id(value: &str) -> bool {
857    prefixed_hex(value, "cj_", 32)
858}
859
860/// A bundle name: `^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`.
861pub fn is_name(value: &str) -> bool {
862    let bytes = value.as_bytes();
863    (1..=64).contains(&bytes.len())
864        && bytes[0].is_ascii_alphanumeric()
865        && bytes
866            .iter()
867            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'))
868}
869
870/// A runner image name: `^[a-z0-9][a-z0-9._-]{0,63}$`.
871pub fn is_image(value: &str) -> bool {
872    let bytes = value.as_bytes();
873    (1..=64).contains(&bytes.len())
874        && (bytes[0].is_ascii_lowercase() || bytes[0].is_ascii_digit())
875        && bytes.iter().all(|b| {
876            b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'.' | b'_' | b'-')
877        })
878}
879
880/// `^[A-Za-z0-9._:-]{1,64}$`.
881pub fn is_client_ref(value: &str) -> bool {
882    (1..=64).contains(&value.len())
883        && value
884            .bytes()
885            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b':' | b'-'))
886}
887
888fn relative_path(value: &str, min: usize) -> bool {
889    (min..=MAX_WORKDIR_BYTES).contains(&value.len())
890        && !value.starts_with('/')
891        && !value.contains("..")
892        && value
893            .bytes()
894            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'/' | b'-'))
895}
896
897/// `command.workdir`: relative to `/work`, no `..`, may be empty.
898pub fn is_workdir(value: &str) -> bool {
899    relative_path(value, 0)
900}
901
902/// An artifact name: a path below `/work/out`, no `..`.
903pub fn is_artifact_name(value: &str) -> bool {
904    relative_path(value, 1)
905}
906
907fn is_count_key(value: &str) -> bool {
908    let bytes = value.as_bytes();
909    (1..=32).contains(&bytes.len())
910        && bytes[0].is_ascii_lowercase()
911        && bytes
912            .iter()
913            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || *b == b'_')
914}
915
916/// A host of `network_allowlist`: lowercase DNS labels, a letters-only TLD.
917pub fn is_host(value: &str) -> bool {
918    if !(1..=253).contains(&value.len()) {
919        return false;
920    }
921    let labels: Vec<&str> = value.split('.').collect();
922    let Some((tld, rest)) = labels.split_last() else {
923        return false;
924    };
925    !rest.is_empty()
926        && (2..=63).contains(&tld.len())
927        && tld.bytes().all(|b| b.is_ascii_lowercase())
928        && rest.iter().all(|label| {
929            let b = label.as_bytes();
930            (1..=63).contains(&b.len())
931                && b[0] != b'-'
932                && b[b.len() - 1] != b'-'
933                && b.iter()
934                    .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == b'-')
935        })
936}
937
938fn validate_network(hosts: &[String]) -> Result<(), ValidationError> {
939    check(
940        hosts.len() <= MAX_NETWORK_HOSTS,
941        "network_allowlist (at most 16 hosts)",
942    )?;
943    check(
944        hosts.iter().all(|h| is_host(h)),
945        "network_allowlist (lowercase host names)",
946    )?;
947    let mut seen = std::collections::BTreeSet::new();
948    check(
949        hosts.iter().all(|h| seen.insert(h.as_str())),
950        "network_allowlist (each host once)",
951    )
952}
953
954#[cfg(test)]
955#[path = "compute_tests.rs"]
956mod tests;
lean_ctx_protocol/hive/compute.rs

lean_ctx_protocol/hive/
compute.rs
