diff --git a/crates/skilld-command/src/lib.rs b/crates/skilld-command/src/lib.rs index 6729f0aa..82e6fd4b 100644 --- a/crates/skilld-command/src/lib.rs +++ b/crates/skilld-command/src/lib.rs @@ -23,10 +23,10 @@ pub use local_store::{ pub use output::{CommandPlatform, OutputContext}; pub use remote::{ Cancellation, HeaderValue, HttpAdapter, HttpHeader, HttpMethod, HttpRequest, HttpResponse, - NativeRemoteConfig, NeverCancelled, NoTokenProvider, PreparedRemoteSkill, + NativeRemoteConfig, NeverCancelled, NoRemoteProgress, NoTokenProvider, PreparedRemoteSkill, RemoteComparisonAccess, RemoteComparisonOutcome, RemoteComparisonRelation, RemoteLatestCommit, - RemoteProvider, RemoteSourceState, RemoteUpdateComparison, RemoteUpdateResult, SecretValue, - SkilldRemote, Sleeper, ThreadSleeper, TokenProvider, + RemoteProgress, RemoteProgressStage, RemoteProvider, RemoteSourceState, RemoteUpdateComparison, + RemoteUpdateResult, SecretValue, SkilldRemote, Sleeper, ThreadSleeper, TokenProvider, }; pub use run::{ FileContent, FileKind, PulledFile, RunOutcome, SkillOrigin, SupportingFile, TransientSkill, diff --git a/crates/skilld-command/src/remote.rs b/crates/skilld-command/src/remote.rs index 3d4518c7..73764498 100644 --- a/crates/skilld-command/src/remote.rs +++ b/crates/skilld-command/src/remote.rs @@ -649,11 +649,50 @@ pub enum NativeRemoteConfig { Unconfigured, } +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq)] +#[serde(rename_all = "kebab-case")] +pub enum RemoteProgressStage { + RequestingResolution, + Requested, + Resolving, + Fetching, + Checking, + Packaging, + Encrypting, + Signing, + Publishing, + RetryWait, + VerifyingAttestation, + RequestingDownload, + DownloadingArtifact, + VerifyingArtifact, +} + +pub trait RemoteProgress: Send + Sync { + fn stage(&self, stage: RemoteProgressStage); +} + +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum RemotePendingStage { + Known(RemoteProgressStage), + #[allow(dead_code)] + Unknown(String), +} + +#[derive(Default)] +pub struct NoRemoteProgress; + +impl RemoteProgress for NoRemoteProgress { + fn stage(&self, _stage: RemoteProgressStage) {} +} + pub struct SkilldRemote { adapter: Arc, tokens: Arc, cancellation: Arc, sleeper: Arc, + progress: Arc, endpoint: Url, root_pin: NativeRemoteConfig, } @@ -669,6 +708,7 @@ impl SkilldRemote { tokens, cancellation: Arc::new(NeverCancelled), sleeper: Arc::new(ThreadSleeper), + progress: Arc::new(NoRemoteProgress), endpoint: Url::parse("https://skilld.dev").expect("the fixed endpoint is valid"), root_pin, } @@ -684,6 +724,11 @@ impl SkilldRemote { self } + pub fn with_progress(mut self, progress: Arc) -> Self { + self.progress = progress; + self + } + pub fn with_endpoint(mut self, endpoint: &str) -> Result { let endpoint = Url::parse(endpoint) .map_err(|_| RemoteError::new("INVALID_ENDPOINT", "the API endpoint is invalid"))?; @@ -843,6 +888,8 @@ impl SkilldRemote { fn resolve(&self, source: &SourceRequest) -> Result { let mut deadline = ResolutionDeadline::new(); + self.progress + .stage(RemoteProgressStage::RequestingResolution); let body = serde_json::to_vec(&json!({ "source": source })).map_err(|_| { RemoteError::new("INVALID_SOURCE", "the source request cannot be encoded") })?; @@ -891,9 +938,12 @@ impl SkilldRemote { } Resolution::Pending { resolution_id, + stage, poll_after_ms, - .. } => { + if let RemotePendingStage::Known(stage) = stage { + self.progress.stage(stage); + } if !(250..=60_000).contains(&poll_after_ms) { return Err(RemoteError::new( "INVALID_RESPONSE", @@ -1468,10 +1518,16 @@ impl RemoteProvider for SkilldRemote { return self.direct(selector); } let descriptor = self.resolve(selector.source())?; + self.progress + .stage(RemoteProgressStage::VerifyingAttestation); let root = self.verified_root()?; verify_attestation(&descriptor.attestation, &root)?; + self.progress.stage(RemoteProgressStage::RequestingDownload); let grant = self.grant(&descriptor.artifact_id)?; + self.progress + .stage(RemoteProgressStage::DownloadingArtifact); let archive = self.download_grant(&descriptor, grant)?; + self.progress.stage(RemoteProgressStage::VerifyingArtifact); let verified = verify_artifact(descriptor.attestation, &root, &archive)?; if matches!( &selector.source().selector, @@ -2261,8 +2317,7 @@ enum Resolution { Pending { #[serde(rename = "resolutionId")] resolution_id: String, - #[serde(rename = "stage")] - _stage: String, + stage: RemotePendingStage, #[serde(rename = "pollAfterMs")] poll_after_ms: u64, }, diff --git a/crates/skilld-command/tests/remote.rs b/crates/skilld-command/tests/remote.rs index 9245cd88..ca8064f4 100644 --- a/crates/skilld-command/tests/remote.rs +++ b/crates/skilld-command/tests/remote.rs @@ -11,8 +11,9 @@ use sha2::{Digest, Sha256}; use skilld_command::{ Cancellation, HeaderValue, Host, HttpAdapter, HttpRequest, HttpResponse, LocalHost, NativeRemoteConfig, NoTokenProvider, PreparedRemoteSkill, RemoteComparisonAccess, - RemoteComparisonOutcome, RemoteComparisonRelation, RemoteProvider, RemoteSourceState, - RemoteUpdateComparison, SecretValue, SkilldRemote, Sleeper, TokenProvider, run, + RemoteComparisonOutcome, RemoteComparisonRelation, RemoteProgress, RemoteProgressStage, + RemoteProvider, RemoteSourceState, RemoteUpdateComparison, SecretValue, SkilldRemote, Sleeper, + TokenProvider, run, }; use skilld_core::{ AgentTargetId, ArtifactAttestation, ArtifactFile, AttestationSignature, CheckOutcome, @@ -141,6 +142,15 @@ impl Sleeper for NoSleep { } } +#[derive(Default)] +struct RecordingProgress(Mutex>); + +impl RemoteProgress for RecordingProgress { + fn stage(&self, stage: RemoteProgressStage) { + self.0.lock().unwrap().push(stage); + } +} + #[derive(Default)] struct RecordingSleeper { elapsed: Mutex, @@ -1020,6 +1030,111 @@ fn a_resolution_cannot_change_its_identity_while_polling() { assert_eq!(error.code, "INVALID_RESPONSE"); } +#[test] +fn a_hosted_resolution_reports_each_service_stage() { + let resolution_id = "018f47a4-2d38-7c5f-8d3e-1c5a6b7d8e9f"; + let stages = [ + "requested", + "resolving", + "fetching", + "checking", + "packaging", + "encrypting", + "signing", + "publishing", + "retry-wait", + ]; + let mut responses = stages + .iter() + .map(|stage| { + response( + 200, + serde_json::to_vec(&json!({ + "state": "pending", + "resolutionId": resolution_id, + "stage": stage, + "pollAfterMs": 250 + })) + .unwrap(), + ) + }) + .collect::>(); + responses.push(response( + 200, + serde_json::to_vec(&json!({ + "state": "blocked", + "resolutionId": resolution_id, + "checkResults": [] + })) + .unwrap(), + )); + let progress = Arc::new(RecordingProgress::default()); + let remote = search_remote(Arc::new(FakeHttp::with(responses))).with_progress(progress.clone()); + + let error = remote.prepare(&skilld_selector(), false).unwrap_err(); + + assert_eq!(error.code, "CHECK_BLOCKED"); + assert_eq!( + *progress.0.lock().unwrap(), + [ + RemoteProgressStage::RequestingResolution, + RemoteProgressStage::Requested, + RemoteProgressStage::Resolving, + RemoteProgressStage::Fetching, + RemoteProgressStage::Checking, + RemoteProgressStage::Packaging, + RemoteProgressStage::Encrypting, + RemoteProgressStage::Signing, + RemoteProgressStage::Publishing, + RemoteProgressStage::RetryWait, + ] + ); +} + +#[test] +fn an_unknown_pending_stage_does_not_abort_the_resolution() { + let (pin, mut responses) = verified_remote_responses(); + let pending = response( + 200, + serde_json::to_vec(&json!({ + "state": "pending", + "resolutionId": "018f47a4-2d38-7c5f-8d3e-1c5a6b7d8e9f", + "stage": "queueing", + "pollAfterMs": 250 + })) + .unwrap(), + ); + responses.insert(0, pending); + let progress = Arc::new(RecordingProgress::default()); + let remote = SkilldRemote::new( + Arc::new(FakeHttp::with(responses)), + Arc::new(NoTokenProvider), + NativeRemoteConfig::Pinned(pin), + ) + .with_endpoint("http://127.0.0.1:8787") + .unwrap() + .with_sleeper(Arc::new(NoSleep)) + .with_progress(progress.clone()); + let selector = RemoteSelector::parse("skilld:skilld-dev/skills/example").unwrap(); + + let prepared = remote.prepare(&selector, false).unwrap(); + + assert!(matches!( + prepared.source_status, + SourceStatus::Verified { .. } + )); + assert_eq!( + *progress.0.lock().unwrap(), + [ + RemoteProgressStage::RequestingResolution, + RemoteProgressStage::VerifyingAttestation, + RemoteProgressStage::RequestingDownload, + RemoteProgressStage::DownloadingArtifact, + RemoteProgressStage::VerifyingArtifact, + ] + ); +} + #[test] fn a_pending_resolution_times_out_after_at_most_sixty_seconds() { let resolution_id = "018f47a4-2d38-7c5f-8d3e-1c5a6b7d8e9f"; diff --git a/crates/skilld-native/src/main.rs b/crates/skilld-native/src/main.rs index 050ce9c4..aa49745a 100644 --- a/crates/skilld-native/src/main.rs +++ b/crates/skilld-native/src/main.rs @@ -60,17 +60,39 @@ fn main() -> ExitCode { }; let global_root = global_root(); let detection = detection_environment(); + let output = OutputContext::auto( + std::io::stdout().is_terminal(), + active_agent_detected(), + environment_enabled("CI"), + environment_present("NO_COLOR"), + env::var("TERM").is_ok_and(|term| term.eq_ignore_ascii_case("dumb")), + terminal_width(), + CommandPlatform::current(), + ); + let label = if interactive { + None + } else { + status::status_label(args.iter().map(|arg| arg.to_string_lossy())) + }; + let status = match label { + Some(label) => StatusLine::for_terminal(label, output), + None => StatusLine::disabled(), + }; + let remote_progress = status.remote_progress(); let account = Arc::new(NativeAccount::new()); let host = LocalHost::new(project_root, global_root) .with_target_roots(target_roots()) .with_detection_environment(detection.clone()) .with_bundled_provider(Arc::new(EmbeddedSkilld::new())) .with_account_provider(account.clone()) - .with_remote_provider(Arc::new(SkilldRemote::new( - Arc::new(NativeHttpAdapter::new()), - account, - native_remote_config(), - ))); + .with_remote_provider(Arc::new( + SkilldRemote::new( + Arc::new(NativeHttpAdapter::new()), + account, + native_remote_config(), + ) + .with_progress(remote_progress), + )); let host = if args.iter().skip(1).any(|arg| arg == "outdated") && !args.iter().any(|arg| arg == "--json" || arg == "--plain") { @@ -115,20 +137,6 @@ fn main() -> ExitCode { let mut stdout = std::io::stdout().lock(); let mut stderr = std::io::stderr(); - let output = OutputContext::auto( - stdout.is_terminal(), - active_agent_detected(), - environment_enabled("CI"), - environment_present("NO_COLOR"), - env::var("TERM").is_ok_and(|term| term.eq_ignore_ascii_case("dumb")), - terminal_width(), - CommandPlatform::current(), - ); - let label = status::status_label(args.iter().map(|arg| arg.to_string_lossy())); - let status = match label { - Some(label) => StatusLine::for_terminal(label, output), - None => StatusLine::disabled(), - }; let mut gated = status::GatedStderr::new(&mut stderr, status); let result = run_with_output(args, host.as_ref(), output, &mut stdout, &mut gated); gated.finish_status(); diff --git a/crates/skilld-native/src/status.rs b/crates/skilld-native/src/status.rs index fe5cfe09..9309a8ba 100644 --- a/crates/skilld-native/src/status.rs +++ b/crates/skilld-native/src/status.rs @@ -1,4 +1,4 @@ -use skilld_command::OutputContext; +use skilld_command::{OutputContext, RemoteProgress, RemoteProgressStage}; use skilld_ui::spinner; use skilld_ui::theme::{RESET, Role, paint}; @@ -38,6 +38,40 @@ pub struct StatusLine { thread: Option>, } +struct StatusProgress { + shared: Option>>, +} + +impl RemoteProgress for StatusProgress { + fn stage(&self, stage: RemoteProgressStage) { + let Some(shared) = &self.shared else { + return; + }; + if let Ok(mut state) = shared.lock() { + state.label = remote_status_label(stage).to_owned(); + } + } +} + +fn remote_status_label(stage: RemoteProgressStage) -> &'static str { + match stage { + RemoteProgressStage::RequestingResolution => "Starting Artifact resolution", + RemoteProgressStage::Requested => "Waiting for Artifact resolution", + RemoteProgressStage::Resolving => "Resolving Skill source", + RemoteProgressStage::Fetching => "Fetching Repository", + RemoteProgressStage::Checking => "Checking Skill", + RemoteProgressStage::Packaging => "Packaging Artifact", + RemoteProgressStage::Encrypting => "Encrypting Artifact", + RemoteProgressStage::Signing => "Signing attestation", + RemoteProgressStage::Publishing => "Publishing Artifact", + RemoteProgressStage::RetryWait => "Waiting to retry", + RemoteProgressStage::VerifyingAttestation => "Verifying attestation", + RemoteProgressStage::RequestingDownload => "Requesting Artifact download", + RemoteProgressStage::DownloadingArtifact => "Downloading Artifact", + RemoteProgressStage::VerifyingArtifact => "Verifying Artifact", + } +} + impl StatusLine { pub fn disabled() -> Self { Self { @@ -111,6 +145,12 @@ impl StatusLine { ) } + pub fn remote_progress(&self) -> Arc { + Arc::new(StatusProgress { + shared: self.shared.clone(), + }) + } + pub fn stop(&mut self) { let Some(shared) = &self.shared else { return; @@ -259,8 +299,11 @@ impl skilld_command::OutdatedProgress for OutdatedProgressLine { #[cfg(test)] mod tests { - use super::{GatedStderr, OutputContext, StatusLine, frame_line, status_label}; + use super::{ + GatedStderr, OutputContext, StatusLine, frame_line, remote_status_label, status_label, + }; use skilld_command::CommandPlatform; + use skilld_command::RemoteProgressStage; use std::io::Write; use std::sync::Mutex; use std::thread; @@ -290,6 +333,26 @@ mod tests { assert!(colored.contains("Searching")); } + #[test] + fn remote_stages_name_the_current_work() { + assert_eq!( + remote_status_label(RemoteProgressStage::Resolving), + "Resolving Skill source" + ); + assert_eq!( + remote_status_label(RemoteProgressStage::Checking), + "Checking Skill" + ); + assert_eq!( + remote_status_label(RemoteProgressStage::Packaging), + "Packaging Artifact" + ); + assert_eq!( + remote_status_label(RemoteProgressStage::DownloadingArtifact), + "Downloading Artifact" + ); + } + #[test] fn nothing_paints_before_the_start_delay() { let buffer = std::sync::Arc::new(Mutex::new(Vec::new()));