diff --git a/Cargo.lock b/Cargo.lock index 4ebbd2f8..63eb2401 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -48,6 +48,15 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "android_system_properties" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" +dependencies = [ + "libc", +] + [[package]] name = "anstream" version = "1.0.0" @@ -223,6 +232,17 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "num-traits", + "windows-link", +] + [[package]] name = "clap" version = "4.6.6" @@ -287,6 +307,16 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "combine" +version = "4.6.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfc320937d09e6de266b31b9afb480f197d7a861be86be7cb2ea7e5d1bfffc5e" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "compression-codecs" version = "0.4.38" @@ -310,6 +340,22 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" +[[package]] +name = "core-foundation" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpp_demangle" version = "0.5.1" @@ -492,6 +538,36 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crypto-common" version = "0.1.7" @@ -826,6 +902,76 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hickory-net" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2295ed2f9c31e471e1428a8f88a3f0e1f4b27c15049592138d1eebe9c35b183" +dependencies = [ + "async-trait", + "cfg-if", + "data-encoding", + "futures-channel", + "futures-io", + "futures-util", + "hickory-proto", + "idna", + "ipnet", + "jni", + "rand 0.10.2", + "thiserror", + "tinyvec", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "hickory-proto" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bab31817bfb44672a252e97fe81cd0c18d1b2cf892108922f6818820df8c643" +dependencies = [ + "data-encoding", + "idna", + "ipnet", + "jni", + "once_cell", + "prefix-trie", + "rand 0.10.2", + "ring", + "thiserror", + "tinyvec", + "tracing", + "url", +] + +[[package]] +name = "hickory-resolver" +version = "0.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0d58d28879ceecde6607729660c2667a081ccdc082e082675042793960f178c" +dependencies = [ + "cfg-if", + "futures-util", + "hickory-net", + "hickory-proto", + "ipconfig", + "ipnet", + "jni", + "moka", + "ndk-context", + "once_cell", + "parking_lot", + "rand 0.10.2", + "resolv-conf", + "smallvec", + "system-configuration", + "thiserror", + "tokio", + "tracing", +] + [[package]] name = "http" version = "1.5.0" @@ -933,6 +1079,30 @@ dependencies = [ "tracing", ] +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "icu_collections" version = "2.3.0" @@ -1075,11 +1245,27 @@ dependencies = [ "libc", ] +[[package]] +name = "ipconfig" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" +dependencies = [ + "socket2", + "widestring", + "windows-registry", + "windows-result", + "windows-sys 0.61.2", +] + [[package]] name = "ipnet" version = "2.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" +dependencies = [ + "serde", +] [[package]] name = "is_terminal_polyfill" @@ -1102,6 +1288,55 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys", + "log", + "simd_cesu8", + "thiserror", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.119", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn 2.0.119", +] + [[package]] name = "js-sys" version = "0.3.104" @@ -1275,6 +1510,29 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "moka" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9" +dependencies = [ + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + +[[package]] +name = "ndk-context" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" + [[package]] name = "notify" version = "8.2.0" @@ -1337,6 +1595,10 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "critical-section", + "portable-atomic", +] [[package]] name = "once_cell_polyfill" @@ -1488,6 +1750,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "portable-atomic" +version = "1.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" + [[package]] name = "postcard" version = "1.1.3" @@ -1518,6 +1786,17 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prefix-trie" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cf6e3177f0684016a5c209b00882e15f8bdd3f3bb48f0491df10cd102d0c6e7" +dependencies = [ + "either", + "ipnet", + "num-traits", +] + [[package]] name = "prettyplease" version = "0.2.37" @@ -1636,7 +1915,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1798,6 +2077,12 @@ dependencies = [ "webpki-roots", ] +[[package]] +name = "resolv-conf" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" + [[package]] name = "ring" version = "0.17.14" @@ -1824,6 +2109,15 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" +[[package]] +name = "rustc_version" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" +dependencies = [ + "semver", +] + [[package]] name = "rustix" version = "1.1.4" @@ -2016,6 +2310,22 @@ version = "0.3.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", +] + +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + [[package]] name = "slab" version = "0.4.12" @@ -2083,6 +2393,16 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "st2-github-pr-component" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "sha2", + "wit-bindgen", +] + [[package]] name = "st2-pty-stats-component" version = "0.1.0" @@ -2106,9 +2426,12 @@ dependencies = [ name = "st2-resource-providers" version = "0.1.0" dependencies = [ + "chrono", + "hickory-resolver", "libc", "parking_lot", "reqwest", + "serde", "serde_json", "st2-resource-protocol", "st2-resource-wasip2", @@ -2131,6 +2454,16 @@ dependencies = [ "wat", ] +[[package]] +name = "st2-vista-component" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "sha2", + "wit-bindgen", +] + [[package]] name = "st2-wire" version = "0.1.0" @@ -2199,6 +2532,33 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "system-configuration" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" +dependencies = [ + "bitflags", + "core-foundation", + "system-configuration-sys", +] + +[[package]] +name = "system-configuration-sys" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "target-lexicon" version = "0.13.5" @@ -2610,6 +2970,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.26.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.1" @@ -3079,6 +3450,12 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + [[package]] name = "winapi-util" version = "0.1.11" @@ -3088,12 +3465,76 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-registry" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" +dependencies = [ + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.52.0" diff --git a/Cargo.toml b/Cargo.toml index da39f63b..a66a0b51 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,9 @@ [workspace] members = [ "components/github-issue", + "components/github-pr", "components/pty-stats", + "components/vista", "crates/agent-spec", "crates/demo-resolver-wasm", "crates/st2-resource-protocol", diff --git a/components/github-pr/Cargo.toml b/components/github-pr/Cargo.toml new file mode 100644 index 00000000..cb6ed23c --- /dev/null +++ b/components/github-pr/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "st2-github-pr-component" +version = "0.1.0" +edition = "2024" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" +wit-bindgen = "0.57.1" diff --git a/components/github-pr/src/lib.rs b/components/github-pr/src/lib.rs new file mode 100644 index 00000000..710ce8e0 --- /dev/null +++ b/components/github-pr/src/lib.rs @@ -0,0 +1,511 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use sha2::{Digest as _, Sha256}; + +wit_bindgen::generate!({ + path: "../../wit/github-pr", + world: "github-pr-provider", + with: { + "compoundingtech:st2-github-pr/github-pr@0.1.0": generate, + }, +}); + +use compoundingtech::st2_github_pr::github_pr; +use exports::st2::resource_provider::provider_api; + +const SNAPSHOT_SCHEMA: &str = "dev.schickling.github-pr.snapshot.v1"; +const TOPICS: [&str; 4] = [ + "ci.failure", + "mergeability.conflict", + "review.requested", + "terminal", +]; +const SELECTOR_SCHEMA: &str = r#"{ + "type": "object", + "properties": { + "owner": { "type": "string" }, + "repo": { "type": "string" }, + "number": { "type": "integer" }, + "topics": { + "type": "array", + "items": { "type": "string" }, + "uniqueItems": true + } + }, + "required": ["owner", "repo", "number"], + "additionalProperties": false +}"#; + +struct Component; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct Selector { + owner: String, + repo: String, + number: u64, + #[serde(default)] + topics: Vec, +} + +#[derive(Deserialize)] +struct PullRequestResponse { + number: u64, + #[serde(default = "null")] + url: Value, + #[serde(default = "null")] + html_url: Value, + #[serde(default = "null")] + state: Value, + #[serde(default)] + draft: Option, + #[serde(default)] + merged: Option, + #[serde(default = "null")] + merged_at: Value, + #[serde(default = "null")] + closed_at: Value, + #[serde(default)] + mergeable: Option, + #[serde(default = "null")] + mergeable_state: Value, + head: PullRequestHeadResponse, + base: PullRequestBaseResponse, + #[serde(default)] + requested_reviewers: Option>, + #[serde(default)] + requested_teams: Option>, +} + +#[derive(Deserialize)] +struct PullRequestHeadResponse { + sha: String, + #[serde(default = "null")] + r#ref: Value, +} + +#[derive(Deserialize)] +struct PullRequestBaseResponse { + #[serde(default = "null")] + r#ref: Value, +} + +#[derive(Deserialize)] +struct RequestedReviewerResponse { + login: String, +} + +#[derive(Deserialize)] +struct RequestedTeamResponse { + slug: String, +} + +#[derive(Deserialize)] +struct CheckRunsResponse { + #[serde(default)] + check_runs: Option>, +} + +#[derive(Deserialize)] +struct CheckRunResponse { + name: String, + #[serde(default = "null")] + status: Value, + #[serde(default = "null")] + conclusion: Value, + #[serde(default = "null")] + details_url: Value, +} + +#[derive(Deserialize)] +struct CombinedStatusResponse { + #[serde(default)] + state: Option, + #[serde(default)] + statuses: Option>, +} + +#[derive(Deserialize)] +struct StatusResponse { + context: String, + state: String, + #[serde(default = "null")] + target_url: Value, + #[serde(default = "null")] + description: Value, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Snapshot<'a> { + schema: &'static str, + uri: &'a str, + observed_at: &'a str, + repository: RepositorySnapshot<'a>, + number: u64, + pull_request: PullRequestSnapshot, + ci: CiSnapshot, + facets: Facets, +} + +#[derive(Serialize)] +struct RepositorySnapshot<'a> { + owner: &'a str, + name: &'a str, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct PullRequestSnapshot { + api_url: Value, + html_url: Value, + state: Value, + draft: bool, + merged: bool, + merged_at: Value, + closed_at: Value, + mergeable: Option, + mergeable_state: Value, + head: HeadSnapshot, + base: BaseSnapshot, + requested_reviewers: Vec, + requested_teams: Vec, +} + +#[derive(Serialize)] +struct HeadSnapshot { + sha: String, + r#ref: Value, +} + +#[derive(Serialize)] +struct BaseSnapshot { + r#ref: Value, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct CiSnapshot { + state: String, + check_runs: Vec, + statuses: Vec, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct CheckRunSnapshot { + name: String, + status: Value, + conclusion: Value, + details_url: Value, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct StatusSnapshot { + context: String, + state: String, + target_url: Value, + description: Value, +} + +#[derive(Serialize, Clone, Copy, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +struct Facets { + review_requested: bool, + ci_failure: bool, + merge_conflict: bool, + terminal: bool, +} + +impl provider_api::Guest for Component { + fn describe() -> Result { + Ok(provider_api::ProviderDescriptor { + capabilities: vec![provider_api::SchedulingCapability::Demand], + selector_schema_json: SELECTOR_SCHEMA.into(), + default_selector_json: "{}".into(), + topics: TOPICS.iter().map(|topic| (*topic).to_owned()).collect(), + snapshot_media_type: "application/json".into(), + snapshot_schema_id: SNAPSHOT_SCHEMA.into(), + }) + } + + fn observe(request: provider_api::ObserveRequest) -> provider_api::ObservationResult { + observe(request).unwrap_or_else(|diagnostic| { + provider_api::ObservationResult::Failed(Some(diagnostic)) + }) + } +} + +fn observe( + request: provider_api::ObserveRequest, +) -> Result { + let selector: Selector = serde_json::from_str(&request.selector_json) + .map_err(|_| "invalid GitHub pull request selector".to_owned())?; + if selector.owner.is_empty() || selector.repo.is_empty() || selector.number == 0 { + return Err("GitHub pull request selector fields must be non-empty".into()); + } + let expected_uri = format!( + "github-pr://{}/{}/{}", + selector.owner, selector.repo, selector.number + ); + if request.uri != expected_uri { + return Err("GitHub pull request URI did not match the selector".into()); + } + let response = github_pr::get(&github_pr::PullRequestRequest { + owner: selector.owner.clone(), + repo: selector.repo.clone(), + number: selector.number, + }) + .map_err(map_source_error)?; + let observation = match response { + github_pr::PullRequestResponse::NotModified => { + return Ok(provider_api::ObservationResult::Unchanged); + } + github_pr::PullRequestResponse::Ok(observation) => observation, + }; + + let current = build_snapshot( + &request.uri, + &selector, + &observation.current, + )?; + let previous = observation + .previous + .as_ref() + .map(|source| build_snapshot(&request.uri, &selector, source)) + .transpose()?; + if previous + .as_ref() + .is_some_and(|previous| same_semantics(previous, ¤t)) + { + let prior_digest = request + .prior_digest + .as_deref() + .ok_or_else(|| "GitHub prior source lacked a snapshot digest".to_owned())?; + github_pr::bind_snapshot(prior_digest).map_err(map_source_error)?; + return Ok(provider_api::ObservationResult::Unchanged); + } + let bytes = serde_json::to_vec(¤t) + .map_err(|_| "GitHub pull request snapshot normalization failed".to_owned())?; + let digest = Sha256::digest(&bytes); + github_pr::bind_snapshot(digest.as_slice()).map_err(map_source_error)?; + if request.prior_digest.as_deref() == Some(digest.as_slice()) { + return Ok(provider_api::ObservationResult::Unchanged); + } + let publication_topics = topics(previous.as_ref(), ¤t); + let facts = facet_facts(previous.as_ref(), ¤t); + let _ = (request.demand_watermark, selector.topics); + Ok(provider_api::ObservationResult::Published( + provider_api::Publication { + schema_id: SNAPSHOT_SCHEMA.into(), + media_type: "application/json".into(), + bytes, + topics: publication_topics, + facts: Some(facts), + }, + )) +} + +fn build_snapshot( + uri: &str, + selector: &Selector, + source: &github_pr::SourceSnapshot, +) -> Result { + let pull: PullRequestResponse = serde_json::from_slice(&source.pull_request.body) + .map_err(|_| "GitHub pull request response was invalid".to_owned())?; + if pull.number != selector.number || !valid_head_sha(&pull.head.sha) { + return Err("GitHub response did not match the requested pull request".into()); + } + let checks: CheckRunsResponse = serde_json::from_slice(&source.check_runs.body) + .map_err(|_| "GitHub check runs response was invalid".to_owned())?; + let status: CombinedStatusResponse = serde_json::from_slice(&source.combined_status.body) + .map_err(|_| "GitHub combined status response was invalid".to_owned())?; + + let mut requested_reviewers: Vec<_> = pull + .requested_reviewers + .unwrap_or_default() + .into_iter() + .map(|reviewer| reviewer.login) + .collect(); + requested_reviewers.sort(); + let mut requested_teams: Vec<_> = pull + .requested_teams + .unwrap_or_default() + .into_iter() + .map(|team| team.slug) + .collect(); + requested_teams.sort(); + let mut check_runs: Vec<_> = checks + .check_runs + .unwrap_or_default() + .into_iter() + .map(|check| CheckRunSnapshot { + name: check.name, + status: check.status, + conclusion: check.conclusion, + details_url: check.details_url, + }) + .collect(); + check_runs.sort_by(|left, right| left.name.cmp(&right.name)); + let mut statuses: Vec<_> = status + .statuses + .unwrap_or_default() + .into_iter() + .map(|status| StatusSnapshot { + context: status.context, + state: status.state, + target_url: status.target_url, + description: status.description, + }) + .collect(); + statuses.sort_by(|left, right| left.context.cmp(&right.context)); + + let check_failure = check_runs.iter().any(|check| { + check.status.as_str() == Some("completed") + && matches!( + check.conclusion.as_str(), + Some( + "failure" + | "timed_out" + | "cancelled" + | "action_required" + | "startup_failure" + | "stale" + ) + ) + }); + let combined_state = status.state.unwrap_or_else(|| "pending".to_owned()); + let facets = Facets { + review_requested: !requested_reviewers.is_empty() || !requested_teams.is_empty(), + ci_failure: check_failure || matches!(combined_state.as_str(), "failure" | "error"), + merge_conflict: pull.mergeable == Some(false) + || pull.mergeable_state.as_str() == Some("dirty"), + terminal: pull.merged == Some(true) || pull.state.as_str() == Some("closed"), + }; + + serde_json::to_value(Snapshot { + schema: SNAPSHOT_SCHEMA, + uri, + observed_at: &source.observed_at, + repository: RepositorySnapshot { + owner: &selector.owner, + name: &selector.repo, + }, + number: selector.number, + pull_request: PullRequestSnapshot { + api_url: pull.url, + html_url: pull.html_url, + state: pull.state, + draft: pull.draft.unwrap_or(false), + merged: pull.merged.unwrap_or(false), + merged_at: pull.merged_at, + closed_at: pull.closed_at, + mergeable: pull.mergeable, + mergeable_state: pull.mergeable_state, + head: HeadSnapshot { + sha: pull.head.sha, + r#ref: pull.head.r#ref, + }, + base: BaseSnapshot { + r#ref: pull.base.r#ref, + }, + requested_reviewers, + requested_teams, + }, + ci: CiSnapshot { + state: combined_state, + check_runs, + statuses, + }, + facets, + }) + .map_err(|_| "GitHub pull request snapshot normalization failed".to_owned()) +} + +fn same_semantics(before: &Value, after: &Value) -> bool { + let (Some(before), Some(after)) = (before.as_object(), after.as_object()) else { + return false; + }; + let before_len = before + .keys() + .filter(|key| key.as_str() != "observedAt") + .count(); + let after_len = after + .keys() + .filter(|key| key.as_str() != "observedAt") + .count(); + before_len == after_len + && before + .iter() + .filter(|(key, _)| key.as_str() != "observedAt") + .all(|(key, value)| after.get(key) == Some(value)) +} + +fn topics(previous: Option<&Value>, current: &Value) -> Vec { + let Some(previous) = previous else { + return TOPICS.iter().map(|topic| (*topic).to_owned()).collect(); + }; + [ + ("ciFailure", "ci.failure"), + ("mergeConflict", "mergeability.conflict"), + ("reviewRequested", "review.requested"), + ("terminal", "terminal"), + ] + .into_iter() + .filter_map(|(facet, topic)| { + (facet_value(previous, facet) != facet_value(current, facet)).then(|| topic.to_owned()) + }) + .collect() +} + +fn facet_facts(previous: Option<&Value>, current: &Value) -> Vec { + [ + ("facets.ciFailure", "ciFailure"), + ("facets.mergeConflict", "mergeConflict"), + ("facets.reviewRequested", "reviewRequested"), + ("facets.terminal", "terminal"), + ] + .into_iter() + .filter_map(|(key, facet)| { + let before = previous.and_then(|value| facet_value(value, facet)); + let after = facet_value(current, facet); + (before != after).then(|| provider_api::Fact { + key: key.into(), + before: before.map_or(provider_api::FactValue::Omitted, |value| { + provider_api::FactValue::Value(value.to_string()) + }), + after: after.map_or(provider_api::FactValue::Null, |value| { + provider_api::FactValue::Value(value.to_string()) + }), + }) + }) + .collect() +} + +fn facet_value(snapshot: &Value, facet: &str) -> Option { + snapshot + .get("facets") + .and_then(|facets| facets.get(facet)) + .and_then(Value::as_bool) +} + +fn valid_head_sha(value: &str) -> bool { + value.len() == 40 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) +} + +fn null() -> Value { + Value::Null +} + +fn map_source_error(error: github_pr::PullRequestError) -> String { + match error { + github_pr::PullRequestError::Denied => "GitHub pull request scope denied", + github_pr::PullRequestError::Unavailable => "GitHub is unavailable", + github_pr::PullRequestError::ResourceExhausted => "GitHub response exceeded limits", + github_pr::PullRequestError::DeadlineExceeded => "GitHub request deadline exceeded", + } + .into() +} + +export!(Component); diff --git a/components/vista/Cargo.toml b/components/vista/Cargo.toml new file mode 100644 index 00000000..5bc793b2 --- /dev/null +++ b/components/vista/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "st2-vista-component" +version = "0.1.0" +edition = "2024" +publish = false + +[lib] +crate-type = ["cdylib"] + +[dependencies] +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" +wit-bindgen = "0.57.1" diff --git a/components/vista/src/lib.rs b/components/vista/src/lib.rs new file mode 100644 index 00000000..5abcc220 --- /dev/null +++ b/components/vista/src/lib.rs @@ -0,0 +1,298 @@ +use serde::{Deserialize, Serialize}; +use sha2::{Digest as _, Sha256}; + +wit_bindgen::generate!({ + path: "../../wit/vista", + world: "vista-provider", + with: { + "compoundingtech:st2-vista/vista@0.1.0": generate, + }, +}); + +use compoundingtech::st2_vista::vista; +use exports::st2::resource_provider::provider_api; + +const SNAPSHOT_SCHEMA: &str = "dev.schickling.vista.snapshot.v1"; +const MAX_SNAPSHOT_BYTES: usize = 1024 * 1024; +const MAX_VERSION: u64 = 9_007_199_254_740_991; +const TOPICS: [&str; 4] = ["ready", "updated", "failed", "expired"]; +const SELECTOR_SCHEMA: &str = r#"{ + "type": "object", + "properties": { + "slug": { "type": "string", "minLength": 1, "maxLength": 128, "pattern": "^[a-z0-9]+(?:-[a-z0-9]+)*$" }, + "version": { "type": "integer", "minimum": 1, "maximum": 9007199254740991 }, + "topics": { + "type": "array", + "items": { "type": "string", "enum": ["ready", "updated", "failed", "expired"] }, + "uniqueItems": true + } + }, + "required": ["slug", "version"], + "additionalProperties": false +}"#; + +struct Component; + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct Selector { + slug: String, + version: u64, + #[serde(default)] + topics: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] +#[serde(rename_all = "lowercase")] +enum ArtifactState { + Ready, + Failed, + Expired, +} + +impl ArtifactState { + const fn as_str(self) -> &'static str { + match self { + Self::Ready => "ready", + Self::Failed => "failed", + Self::Expired => "expired", + } + } +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct StatusCounts { + locked: u64, + open: u64, + awaiting: u64, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ArtifactManifest { + schema_version: u8, + uri: String, + slug: String, + version: u64, + author: String, + timestamp: String, + change_summary: String, + parent: Option, + retired: bool, + state: ArtifactState, + canonical_url: String, + #[serde(skip_serializing_if = "Option::is_none")] + title: Option, + #[serde(skip_serializing_if = "Option::is_none")] + template: Option, + #[serde(skip_serializing_if = "Option::is_none")] + status: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Snapshot<'a> { + schema: &'static str, + #[serde(flatten)] + artifact: &'a ArtifactManifest, +} + +impl provider_api::Guest for Component { + fn describe() -> Result { + Ok(provider_api::ProviderDescriptor { + capabilities: vec![provider_api::SchedulingCapability::Demand], + selector_schema_json: SELECTOR_SCHEMA.into(), + default_selector_json: "{}".into(), + topics: TOPICS.into_iter().map(str::to_owned).collect(), + snapshot_media_type: "application/json".into(), + snapshot_schema_id: SNAPSHOT_SCHEMA.into(), + }) + } + + fn observe(request: provider_api::ObserveRequest) -> provider_api::ObservationResult { + observe(request).unwrap_or_else(|diagnostic| { + provider_api::ObservationResult::Failed(Some(diagnostic)) + }) + } +} + +fn observe( + request: provider_api::ObserveRequest, +) -> Result { + let selector: Selector = serde_json::from_str(&request.selector_json) + .map_err(|_| "invalid Vista selector".to_owned())?; + let Some((uri_slug, uri_version)) = parse_uri(&request.uri) else { + return Err("invalid Vista URI".into()); + }; + if !valid_slug(&selector.slug) + || !(1..=MAX_VERSION).contains(&selector.version) + || selector.slug != uri_slug + || selector.version != uri_version + { + return Err("Vista selector does not match URI identity".into()); + } + + let outcome = vista::get(&vista::ArtifactRequest { + slug: selector.slug.clone(), + version: selector.version, + }) + .map_err(map_source_error)?; + if outcome.stdout_truncated || outcome.stderr_truncated { + return Err("Vista output exceeded limits".into()); + } + match outcome.exit { + vista::ExitStatus::Code(0) => {} + vista::ExitStatus::Code(_) | vista::ExitStatus::Signal(_) => { + let detail = bounded_detail(&outcome.stderr); + let diagnostic = if detail.is_empty() { + "vista artifact get exited unsuccessfully".into() + } else { + format!("vista artifact get failed: {detail}") + }; + return Ok(provider_api::ObservationResult::Failed(Some(diagnostic))); + } + } + + let artifact: ArtifactManifest = serde_json::from_slice(&outcome.stdout) + .map_err(|error| format!("vista returned invalid manifest: {error}"))?; + if artifact.schema_version != 1 + || artifact.uri != request.uri + || artifact.slug != selector.slug + || artifact.version != selector.version + { + return Err("vista returned a different artifact identity".into()); + } + + let state = artifact.state.as_str(); + let bytes = serde_json::to_vec(&Snapshot { + schema: SNAPSHOT_SCHEMA, + artifact: &artifact, + }) + .map_err(|_| "Vista snapshot normalization failed".to_owned())?; + if bytes.len() > MAX_SNAPSHOT_BYTES { + return Err("Vista snapshot exceeded limits".into()); + } + let digest = Sha256::digest(&bytes); + if request.prior_digest.as_deref() == Some(digest.as_slice()) { + return Ok(provider_api::ObservationResult::Unchanged); + } + let topics = if request.prior_digest.is_some() { + vec!["updated".into(), state.into()] + } else { + vec![state.into()] + }; + let _ = (request.demand_watermark, selector.topics); + Ok(provider_api::ObservationResult::Published( + provider_api::Publication { + schema_id: SNAPSHOT_SCHEMA.into(), + media_type: "application/json".into(), + bytes, + topics, + facts: Some(vec![provider_api::Fact { + key: "state".into(), + before: provider_api::FactValue::Omitted, + after: provider_api::FactValue::Value(state.into()), + }]), + }, + )) +} + +fn parse_uri(uri: &str) -> Option<(&str, u64)> { + let subject = uri.strip_prefix("vista://")?; + let (slug, version) = subject.split_once('/')?; + let digits = version.strip_prefix('v')?; + if version.contains('/') + || !valid_slug(slug) + || digits.is_empty() + || digits.len() > 16 + || digits.starts_with('0') + || !digits.bytes().all(|byte| byte.is_ascii_digit()) + { + return None; + } + let version = digits.parse::().ok()?; + (1..=MAX_VERSION).contains(&version).then_some((slug, version)) +} + +fn valid_slug(slug: &str) -> bool { + !slug.is_empty() + && slug.len() <= 128 + && slug.bytes().enumerate().all(|(index, byte)| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || (byte == b'-' && index > 0) + }) + && !slug.ends_with('-') + && !slug.contains("--") +} + +fn bounded_detail(stderr: &[u8]) -> String { + const MAX_DETAIL: usize = 4096; + String::from_utf8_lossy(&stderr[..stderr.len().min(MAX_DETAIL)]) + .trim() + .to_owned() +} + +fn map_source_error(error: vista::VistaError) -> String { + match error { + vista::VistaError::Denied => "Vista artifact scope denied", + vista::VistaError::Unavailable => "Vista is unavailable", + vista::VistaError::ResourceExhausted => "Vista output exceeded limits", + vista::VistaError::DeadlineExceeded => "Vista deadline exceeded", + vista::VistaError::Cancelled => "Vista observation was cancelled", + } + .into() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn vista_version_matches_the_javascript_safe_integer_contract() { + assert_eq!( + parse_uri("vista://release-notes/v9007199254740991"), + Some(("release-notes", 9_007_199_254_740_991)) + ); + for invalid in [ + "vista://release-notes/v0", + "vista://release-notes/v01", + "vista://release-notes/v9007199254740992", + "vista://release-notes/v10000000000000000000", + "vista://-release/v1", + "vista://release-/v1", + "vista://release--notes/v1", + "vista://Release/v1", + "vista://release/v1/extra", + "vista://release/1", + ] { + assert_eq!(parse_uri(invalid), None, "{invalid}"); + } + + let schema: serde_json::Value = serde_json::from_str(SELECTOR_SCHEMA).unwrap(); + assert_eq!( + schema["properties"]["version"]["maximum"], + serde_json::json!(9_007_199_254_740_991_u64) + ); + } + + #[test] + fn artifact_manifest_rejects_unknown_fields() { + let unknown = br#"{ + "schemaVersion": 1, + "uri": "vista://release/v1", + "slug": "release", + "version": 1, + "author": "agent", + "timestamp": "2026-09-02T10:00:00Z", + "changeSummary": "created", + "parent": null, + "retired": false, + "state": "ready", + "canonicalUrl": "https://vista.example/release/v1", + "unexpected": true + }"#; + assert!(serde_json::from_slice::(unknown).is_err()); + } +} + +export!(Component); diff --git a/crates/st2-resource-providers/Cargo.toml b/crates/st2-resource-providers/Cargo.toml index 4653a471..38a36e08 100644 --- a/crates/st2-resource-providers/Cargo.toml +++ b/crates/st2-resource-providers/Cargo.toml @@ -2,13 +2,16 @@ name = "st2-resource-providers" version = "0.1.0" edition = "2024" -description = "Typed host capabilities for the st2 GitHub Issue and PTY stats resource components." +description = "Typed host capabilities for the st2 GitHub and PTY stats resource components." license = "MIT" publish = false [dependencies] +chrono = { version = "0.4", default-features = false, features = ["clock", "std"] } +hickory-resolver = { version = "0.26", default-features = false, features = ["system-config", "tokio"] } libc = "0.2" parking_lot = "0.12" +serde = { version = "1", features = ["derive"] } reqwest = { version = "0.12", default-features = false, features = ["gzip", "rustls-tls", "stream"] } serde_json = "1" st2-resource-wasip2 = { path = "../st2-resource-wasip2", features = ["runtime"] } diff --git a/crates/st2-resource-providers/src/github_pr.rs b/crates/st2-resource-providers/src/github_pr.rs new file mode 100644 index 00000000..e0b4d811 --- /dev/null +++ b/crates/st2-resource-providers/src/github_pr.rs @@ -0,0 +1,813 @@ +use std::collections::BTreeMap; +use std::future::Future; +use std::net::{IpAddr, SocketAddr}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use chrono::{SecondsFormat, Utc}; +use serde::Deserialize; +use st2_resource_wasip2::{ + CapabilityContext, CapabilityModule, CapabilityPhase, InterruptionReason, InvocationControl, + InvocationStore, +}; +use wasmtime::component::{HasSelf, Linker}; + +mod bindings { + wasmtime::component::bindgen!({ + path: "../../wit/github-pr", + world: "github-pr-provider", + }); +} + +use bindings::compoundingtech::st2_github_pr::github_pr::{ + Host, PullRequestError, PullRequestRequest, PullRequestResponse, SourceObject, + SourceObservation, SourceSnapshot, +}; + +const IMPORT_NAME: &str = "compoundingtech:st2-github-pr/github-pr@0.1.0"; +const API_HOST: &str = "api.github.com"; +const API_PORT: u16 = 443; +const MAX_HEADERS_BYTES: usize = 16 * 1024; +const MAX_SOURCE_BYTES: usize = 1024 * 1024; +const MAX_ETAG_BYTES: usize = 1024; +const SNAPSHOT_DIGEST_BYTES: usize = 32; +const MAX_CACHED_SNAPSHOTS: usize = 16; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct GitHubPrConfig { + pub owner: String, + pub repo: String, + pub number: u64, + pub connect_timeout: Duration, + pub total_timeout: Duration, +} + +impl GitHubPrConfig { + pub fn validate(&self) -> Result<(), &'static str> { + if !valid_component(&self.owner, 39) + || !valid_component(&self.repo, 100) + || self.number == 0 + { + return Err("GitHub pull request scope is invalid"); + } + if self.connect_timeout.is_zero() + || self.total_timeout.is_zero() + || self.connect_timeout > self.total_timeout + || self.total_timeout > Duration::from_secs(60) + { + return Err("GitHub pull request deadlines are invalid"); + } + Ok(()) + } +} + +#[derive(Clone)] +pub struct GitHubPrModule { + config: GitHubPrConfig, + cache: Arc>, +} + +impl GitHubPrModule { + pub fn new(config: GitHubPrConfig) -> Result { + config.validate()?; + Ok(Self { + config, + cache: Arc::new(Mutex::new(SnapshotCache::default())), + }) + } +} + +#[derive(Debug, Clone)] +struct PullRequestKey { + owner: String, + repo: String, + number: u64, +} + +#[derive(Debug, Clone)] +struct CachedObject { + etag: Option, + body: Vec, +} + +#[derive(Debug, Clone)] +struct CachedSource { + pull_request: CachedObject, + check_runs: CachedObject, + combined_status: CachedObject, + observed_at: String, +} + +#[derive(Default)] +struct SnapshotCache { + sources: BTreeMap<[u8; SNAPSHOT_DIGEST_BYTES], CachedSource>, +} + +impl SnapshotCache { + fn get(&self, digest: &[u8; SNAPSHOT_DIGEST_BYTES]) -> Option<&CachedSource> { + self.sources.get(digest) + } + + fn insert(&mut self, digest: [u8; SNAPSHOT_DIGEST_BYTES], source: CachedSource) { + if !self.sources.contains_key(&digest) && self.sources.len() >= MAX_CACHED_SNAPSHOTS { + if let Some(evicted) = self.sources.keys().next().cloned() { + self.sources.remove(&evicted); + } + } + self.sources.insert(digest, source); + } +} + +pub struct GitHubPrInvocation { + config: GitHubPrConfig, + cache: Arc>, + prior_digest: Option<[u8; SNAPSHOT_DIGEST_BYTES]>, + current_source: Option, + control: InvocationControl, +} + +impl CapabilityModule for GitHubPrModule { + type Invocation = GitHubPrInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[IMPORT_NAME] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + bindings::GithubPrProvider::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, context: CapabilityContext<'_>) -> Self::Invocation { + let prior_digest = match context.phase() { + CapabilityPhase::Describe => None, + CapabilityPhase::Observe(request) => { + request.prior_digest.as_ref().map(|digest| *digest.as_bytes()) + } + }; + GitHubPrInvocation { + config: self.config.clone(), + cache: Arc::clone(&self.cache), + prior_digest, + current_source: None, + control: context.control().clone(), + } + } +} + +impl Host for InvocationStore { + fn get( + &mut self, + request: PullRequestRequest, + ) -> Result { + self.capability_mut().get(request) + } + + fn bind_snapshot(&mut self, digest: Vec) -> Result<(), PullRequestError> { + self.capability_mut().bind_snapshot(digest) + } +} + +impl GitHubPrInvocation { + fn get( + &mut self, + request: PullRequestRequest, + ) -> Result { + run_on_runtime(self.get_async(request)) + } + + async fn get_async( + &mut self, + request: PullRequestRequest, + ) -> Result { + if !request_matches_scope(&self.config, &request) { + return Err(PullRequestError::Denied); + } + if let Some(reason) = self.control.interruption_reason() { + return Err(interruption_error(reason)); + } + let key = PullRequestKey { + owner: request.owner, + repo: request.repo, + number: request.number, + }; + let prior = match self.prior_digest.as_ref() { + Some(digest) => self + .cache + .lock() + .map_err(|_| PullRequestError::Unavailable)? + .get(digest) + .cloned(), + None => None, + }; + let deadline = Instant::now() + .checked_add(self.config.total_timeout) + .ok_or(PullRequestError::DeadlineExceeded)?; + let address = resolve_public_api_address(&self.control, deadline).await?; + let client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(self.config.connect_timeout) + .gzip(true) + .resolve(API_HOST, address) + .build() + .map_err(|_| PullRequestError::Unavailable)?; + + let pull_endpoint = format!( + "https://{API_HOST}/repos/{}/{}/pulls/{}", + key.owner, key.repo, key.number + ); + let pull_request = self + .fetch_object( + &client, + pull_endpoint, + prior.as_ref().map(|source| &source.pull_request), + deadline, + MAX_SOURCE_BYTES, + ) + .await?; + let pull: PullRequestHead = serde_json::from_slice(&pull_request.object.body) + .map_err(|_| PullRequestError::Unavailable)?; + if pull.number != key.number || !valid_head_sha(&pull.head.sha) { + return Err(PullRequestError::Denied); + } + + let mut remaining = MAX_SOURCE_BYTES + .checked_sub(pull_request.object.body.len()) + .ok_or(PullRequestError::ResourceExhausted)?; + let checks_endpoint = format!( + "https://{API_HOST}/repos/{}/{}/commits/{}/check-runs?per_page=100", + key.owner, key.repo, pull.head.sha + ); + let check_runs = require_complete( + self.fetch_object( + &client, + checks_endpoint, + prior.as_ref().map(|source| &source.check_runs), + deadline, + remaining, + ) + .await?, + )?; + remaining = remaining + .checked_sub(check_runs.object.body.len()) + .ok_or(PullRequestError::ResourceExhausted)?; + let status_endpoint = format!( + "https://{API_HOST}/repos/{}/{}/commits/{}/status?per_page=100", + key.owner, key.repo, pull.head.sha + ); + let combined_status = require_complete( + self.fetch_object( + &client, + status_endpoint, + prior.as_ref().map(|source| &source.combined_status), + deadline, + remaining, + ) + .await?, + )?; + + if !pull_request.modified && !check_runs.modified && !combined_status.modified { + return Ok(PullRequestResponse::NotModified); + } + let observed_at = prior + .as_ref() + .filter(|prior| { + prior.pull_request.body == pull_request.object.body + && prior.check_runs.body == check_runs.object.body + && prior.combined_status.body == combined_status.object.body + }) + .map_or_else( + || Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + |prior| prior.observed_at.clone(), + ); + let source = CachedSource { + pull_request: pull_request.object, + check_runs: check_runs.object, + combined_status: combined_status.object, + observed_at, + }; + let current = source_to_wit(&source); + self.current_source = Some(source); + Ok(PullRequestResponse::Ok(SourceObservation { + current, + previous: prior.as_ref().map(source_to_wit), + })) + } + + fn bind_snapshot(&mut self, digest: Vec) -> Result<(), PullRequestError> { + let digest: [u8; SNAPSHOT_DIGEST_BYTES] = digest + .try_into() + .map_err(|_| PullRequestError::Denied)?; + let source = self + .current_source + .take() + .ok_or(PullRequestError::Unavailable)?; + self.cache + .lock() + .map_err(|_| PullRequestError::Unavailable)? + .insert(digest, source); + Ok(()) + } + + async fn fetch_object( + &self, + client: &reqwest::Client, + endpoint: String, + cached: Option<&CachedObject>, + deadline: Instant, + max_body_bytes: usize, + ) -> Result { + if max_body_bytes == 0 { + return Err(PullRequestError::ResourceExhausted); + } + if let Some(reason) = self.control.interruption_reason() { + return Err(interruption_error(reason)); + } + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or(PullRequestError::DeadlineExceeded)?; + let builder = request_builder(client, endpoint, remaining, cached); + let mut response = tokio::select! { + biased; + reason = wait_for_interruption(&self.control) => { + return Err(interruption_error(reason)); + } + response = builder.send() => response.map_err(map_transport_error)?, + }; + let status = response.status(); + if status.is_redirection() && status.as_u16() != 304 { + return Err(PullRequestError::Denied); + } + let header_bytes = response.headers().iter().try_fold( + 0_usize, + |total, (name, value)| { + total + .checked_add(name.as_str().len()) + .and_then(|total| total.checked_add(value.as_bytes().len())) + .ok_or(PullRequestError::ResourceExhausted) + }, + )?; + if header_bytes > MAX_HEADERS_BYTES { + return Err(PullRequestError::ResourceExhausted); + } + let response_etag = response + .headers() + .get(reqwest::header::ETAG) + .and_then(|value| value.to_str().ok()) + .filter(|value| valid_etag(value)) + .map(str::to_owned); + let has_next_page = response_has_next_page(response.headers()); + match status.as_u16() { + 304 => replay_not_modified(cached, response_etag, has_next_page), + 200 => { + let mut body = Vec::new(); + loop { + let chunk = tokio::select! { + biased; + reason = wait_for_interruption(&self.control) => { + return Err(interruption_error(reason)); + } + chunk = response.chunk() => chunk.map_err(map_transport_error)?, + }; + let Some(chunk) = chunk else { + break; + }; + if body.len().saturating_add(chunk.len()) > max_body_bytes { + return Err(PullRequestError::ResourceExhausted); + } + body.extend_from_slice(&chunk); + } + Ok(FetchedObject { + object: CachedObject { + etag: response_etag, + body, + }, + has_next_page, + modified: true, + }) + } + 401 | 403 | 404 => Err(PullRequestError::Denied), + _ => Err(PullRequestError::Unavailable), + } + } +} + +#[derive(Deserialize)] +struct PullRequestHead { + number: u64, + head: PullRequestHeadSha, +} + +#[derive(Deserialize)] +struct PullRequestHeadSha { + sha: String, +} + +struct FetchedObject { + object: CachedObject, + modified: bool, + has_next_page: bool, +} + +fn require_complete(object: FetchedObject) -> Result { + if object.has_next_page { + Err(PullRequestError::ResourceExhausted) + } else { + Ok(object) + } +} + +fn source_to_wit(source: &CachedSource) -> SourceSnapshot { + SourceSnapshot { + pull_request: object_to_wit(&source.pull_request), + + check_runs: object_to_wit(&source.check_runs), + combined_status: object_to_wit(&source.combined_status), + observed_at: source.observed_at.clone(), + } +} + +fn run_on_runtime( + future: impl Future>, +) -> Result { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|_| PullRequestError::Unavailable)?; + runtime.block_on(future) +} + +fn request_builder( + client: &reqwest::Client, + endpoint: String, + timeout: Duration, + cached: Option<&CachedObject>, +) -> reqwest::RequestBuilder { + let mut builder = client + .get(endpoint) + .timeout(timeout) + .header("accept", "application/vnd.github+json") + .header("x-github-api-version", "2022-11-28") + .header("user-agent", "st2-github-pr-resource-profile/1"); + if let Some(etag) = cached.and_then(|object| object.etag.as_deref()) { + builder = builder.header(reqwest::header::IF_NONE_MATCH, etag); + } + builder +} + +fn response_has_next_page(headers: &reqwest::header::HeaderMap) -> bool { + headers + .get_all(reqwest::header::LINK) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .flat_map(|link| link.split(';').skip(1)) + .any(|parameter| parameter.trim().eq_ignore_ascii_case(r#"rel="next""#)) +} + +fn replay_not_modified( + cached: Option<&CachedObject>, + response_etag: Option, + has_next_page: bool, +) -> Result { + let cached = cached.ok_or(PullRequestError::Unavailable)?; + let effective_etag = response_etag.or_else(|| cached.etag.clone()); + if cached.etag != effective_etag { + return Err(PullRequestError::Unavailable); + } + Ok(FetchedObject { + object: cached.clone(), + modified: false, + has_next_page, + }) +} + +fn object_to_wit(object: &CachedObject) -> SourceObject { + SourceObject { + etag: object.etag.clone(), + body: object.body.clone(), + } +} + +async fn wait_for_interruption(control: &InvocationControl) -> InterruptionReason { + loop { + if let Some(reason) = control.interruption_reason() { + return reason; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +fn interruption_error(reason: InterruptionReason) -> PullRequestError { + match reason { + InterruptionReason::Cancelled => PullRequestError::Unavailable, + InterruptionReason::TimedOut => PullRequestError::DeadlineExceeded, + } +} + +fn request_matches_scope(config: &GitHubPrConfig, request: &PullRequestRequest) -> bool { + request.owner == config.owner + && request.repo == config.repo + && request.number == config.number +} + +fn valid_component(value: &str, maximum: usize) -> bool { + !value.is_empty() + && value.len() <= maximum + && !matches!(value, "." | "..") + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.')) +} + +fn valid_head_sha(value: &str) -> bool { + value.len() == 40 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) +} + +fn valid_etag(value: &str) -> bool { + !value.is_empty() + && value.len() <= MAX_ETAG_BYTES + && !value + .bytes() + .any(|byte| byte == b'\r' || byte == b'\n' || byte == 0) +} + +fn map_transport_error(error: reqwest::Error) -> PullRequestError { + if error.is_timeout() { + PullRequestError::DeadlineExceeded + } else { + PullRequestError::Unavailable + } +} + +async fn resolve_public_api_address( + control: &InvocationControl, + deadline: Instant, +) -> Result { + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or(PullRequestError::DeadlineExceeded)?; + let resolver = hickory_resolver::Resolver::builder_tokio() + .and_then(hickory_resolver::ResolverBuilder::build) + .map_err(|_| PullRequestError::Unavailable)?; + let lookup = tokio::select! { + biased; + reason = wait_for_interruption(control) => return Err(interruption_error(reason)), + result = await_dns_lookup(remaining, resolver.lookup_ip(API_HOST)) => result?, + }; + let mut addresses = lookup.iter(); + let first = addresses.next().ok_or(PullRequestError::Unavailable)?; + if !is_public(first) || addresses.any(|address| !is_public(address)) { + return Err(PullRequestError::Denied); + } + Ok(SocketAddr::new(first, API_PORT)) +} + +async fn await_dns_lookup( + remaining: Duration, + lookup: impl Future>, +) -> Result { + tokio::time::timeout(remaining, lookup) + .await + .map_err(|_| PullRequestError::DeadlineExceeded)? + .map_err(|_| PullRequestError::Unavailable) +} + +fn is_public(address: IpAddr) -> bool { + match address { + IpAddr::V4(address) => { + let octets = address.octets(); + !(address.is_private() + || address.is_loopback() + || address.is_link_local() + || address.is_broadcast() + || address.is_documentation() + || address.is_unspecified() + || address.is_multicast() + || octets[0] == 0 + || (octets[0] == 100 && (octets[1] & 0xc0) == 64) + || (octets[0] == 192 && octets[1] == 0 && octets[2] == 0) + || (octets[0] == 198 && matches!(octets[1], 18 | 19)) + || octets[0] >= 240) + } + IpAddr::V6(address) => { + let segments = address.segments(); + let first = segments[0]; + let mapped_private = address + .to_ipv4_mapped() + .is_some_and(|mapped| !is_public(IpAddr::V4(mapped))); + !(address.is_loopback() + || address.is_unspecified() + || address.is_multicast() + || mapped_private + || (first & 0xfe00) == 0xfc00 + || (first & 0xffc0) == 0xfe80 + || (segments[0] == 0x2001 && segments[1] == 0x0db8)) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn live_config() -> GitHubPrConfig { + GitHubPrConfig { + owner: "example".into(), + repo: "demo".into(), + number: 389, + connect_timeout: Duration::from_secs(3), + total_timeout: Duration::from_secs(10), + } + } + + #[test] + fn exact_scope_denies_before_transport() { + let config = live_config(); + for request in [ + PullRequestRequest { + owner: "other".into(), + repo: "demo".into(), + number: 389, + }, + PullRequestRequest { + owner: "example".into(), + repo: "other".into(), + number: 389, + }, + PullRequestRequest { + owner: "example".into(), + repo: "demo".into(), + number: 390, + }, + ] { + assert!(!request_matches_scope(&config, &request)); + } + } + + #[test] + fn private_special_and_documentation_addresses_are_never_admitted() { + for address in [ + "0.0.0.0", + "10.0.0.1", + "127.0.0.1", + "169.254.1.1", + "192.0.2.1", + "224.0.0.1", + "::", + "::1", + "fc00::1", + "fe80::1", + "2001:db8::1", + "ff02::1", + ] { + assert!(!is_public(address.parse().unwrap()), "{address}"); + } + assert!(is_public("8.8.8.8".parse().unwrap())); + assert!(is_public("2606:4700:4700::1111".parse().unwrap())); + } + + #[test] + fn deadlines_are_bounded_and_ordered() { + let mut config = live_config(); + config.connect_timeout = Duration::from_secs(11); + assert!(config.validate().is_err()); + config.connect_timeout = Duration::from_secs(1); + config.total_timeout = Duration::from_secs(61); + assert!(config.validate().is_err()); + } + + #[test] + fn interruption_reasons_preserve_cancel_and_deadline_semantics() { + assert!(matches!( + interruption_error(InterruptionReason::Cancelled), + PullRequestError::Unavailable + )); + assert!(matches!( + interruption_error(InterruptionReason::TimedOut), + PullRequestError::DeadlineExceeded + )); + } + + #[test] + fn conditional_cache_is_selected_by_the_exact_prior_digest() { + let object_v1 = CachedObject { + etag: Some("\"v1\"".into()), + body: b"v1".to_vec(), + }; + let object_v2 = CachedObject { + etag: Some("\"v2\"".into()), + body: b"v2".to_vec(), + }; + let source = |object: CachedObject| CachedSource { + pull_request: object.clone(), + check_runs: object.clone(), + combined_status: object, + observed_at: "2026-09-02T10:00:00Z".into(), + }; + let digest_v1 = [1; SNAPSHOT_DIGEST_BYTES]; + let digest_v2 = [2; SNAPSHOT_DIGEST_BYTES]; + let mut cache = SnapshotCache::default(); + cache.insert(digest_v1, source(object_v1)); + cache.insert(digest_v2, source(object_v2)); + let prior = cache.get(&digest_v1).unwrap(); + assert_eq!(prior.pull_request.body, b"v1"); + + let client = reqwest::Client::new(); + let unbound = request_builder( + &client, + "https://api.github.com/example".into(), + Duration::from_secs(1), + None, + ) + .build() + .unwrap(); + assert!(unbound.headers().get(reqwest::header::IF_NONE_MATCH).is_none()); + let bound = request_builder( + &client, + "https://api.github.com/example".into(), + Duration::from_secs(1), + Some(&prior.pull_request), + ) + .build() + .unwrap(); + assert_eq!( + bound.headers()[reqwest::header::IF_NONE_MATCH], + "\"v1\"" + ); + } + + #[test] + fn paginated_ci_responses_are_rejected_instead_of_published_incomplete() { + let mut headers = reqwest::header::HeaderMap::new(); + headers.insert( + reqwest::header::LINK, + r#"; rel="next", ; rel="last""# + .parse() + .unwrap(), + ); + assert!(response_has_next_page(&headers)); + let incomplete = FetchedObject { + object: CachedObject { + etag: None, + body: Vec::new(), + }, + modified: true, + has_next_page: true, + }; + assert!(matches!( + require_complete(incomplete), + Err(PullRequestError::ResourceExhausted) + )); + + headers.insert( + reqwest::header::LINK, + r#"; rel="prev""# + .parse() + .unwrap(), + ); + assert!(!response_has_next_page(&headers)); + } + + #[test] + fn async_dns_deadline_drops_the_pending_lookup() { + struct PendingLookup(Arc); + + impl Future for PendingLookup { + type Output = Result<(), ()>; + + fn poll( + self: std::pin::Pin<&mut Self>, + _context: &mut std::task::Context<'_>, + ) -> std::task::Poll { + std::task::Poll::Pending + } + } + + impl Drop for PendingLookup { + fn drop(&mut self) { + self.0.store(true, std::sync::atomic::Ordering::SeqCst); + } + } + + let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let result = run_on_runtime(await_dns_lookup( + Duration::ZERO, + PendingLookup(Arc::clone(&dropped)), + )); + assert!(matches!(result, Err(PullRequestError::DeadlineExceeded))); + assert!(dropped.load(std::sync::atomic::Ordering::SeqCst)); + } + + #[test] + fn etags_and_head_shas_reject_header_and_path_injection() { + assert!(valid_etag("\"safe\"")); + assert!(!valid_etag("\"safe\"\r\nx-injected: true")); + assert!(valid_head_sha("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")); + assert!(!valid_head_sha("../heads/main")); + } +} diff --git a/crates/st2-resource-providers/src/lib.rs b/crates/st2-resource-providers/src/lib.rs index 949c1241..a0527f7e 100644 --- a/crates/st2-resource-providers/src/lib.rs +++ b/crates/st2-resource-providers/src/lib.rs @@ -1,7 +1,11 @@ //! Closed host capabilities for production resource-observer components. mod github_issue; +mod github_pr; mod pty_stats; +mod vista; pub use github_issue::{GitHubIssueConfig, GitHubIssueModule}; +pub use github_pr::{GitHubPrConfig, GitHubPrModule}; pub use pty_stats::{PtyStatsConfig, PtyStatsModule, PtyStatsScope}; +pub use vista::{VistaConfig, VistaModule}; diff --git a/crates/st2-resource-providers/src/vista.rs b/crates/st2-resource-providers/src/vista.rs new file mode 100644 index 00000000..6ae0d6e3 --- /dev/null +++ b/crates/st2-resource-providers/src/vista.rs @@ -0,0 +1,801 @@ +use std::io::Write as _; +use std::os::unix::fs::PermissionsExt as _; +use std::os::unix::process::{CommandExt as _, ExitStatusExt as _}; +use std::path::{Path, PathBuf}; +use std::process::{Command, Stdio}; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::{Arc, mpsc}; +use std::thread; +use std::time::{Duration, Instant}; + +use parking_lot::Mutex; +use st2_resource_wasip2::{ + CapabilityContext, CapabilityModule, InterruptionReason, + InvocationControl as ExecutorInvocationControl, InvocationStore, +}; +use wasmtime::component::{HasSelf, Linker}; + +mod bindings { + wasmtime::component::bindgen!({ + path: "../../wit/vista", + world: "vista-provider", + }); +} + +use bindings::compoundingtech::st2_vista::vista::{ + ArtifactRequest, ExitStatus, Host, Outcome, VistaError, +}; + +const IMPORT_NAME: &str = "compoundingtech:st2-vista/vista@0.1.0"; +const MAX_STDOUT_BYTES: usize = 1024 * 1024; +const MAX_STDERR_BYTES: usize = 64 * 1024; +const MAX_VERSION: u64 = 9_007_199_254_740_991; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct VistaConfig { + pub executable: PathBuf, + pub cwd: PathBuf, + pub slug: String, + pub version: u64, + pub deadline: Duration, +} + +impl VistaConfig { + pub fn resolve( + executable: impl AsRef, + cwd: impl Into, + slug: String, + version: u64, + deadline: Duration, + ) -> Result { + if deadline.is_zero() || deadline > Duration::from_secs(60) { + return Err("Vista deadline is invalid"); + } + let executable = + resolve_executable(executable.as_ref()).ok_or("Vista executable is unavailable")?; + let cwd = cwd.into(); + if !cwd.is_absolute() { + return Err("Vista cwd must be absolute"); + } + if !valid_slug(&slug) || !(1..=MAX_VERSION).contains(&version) { + return Err("Vista artifact scope is invalid"); + } + Ok(Self { + executable, + cwd, + slug, + version, + deadline, + }) + } +} + +#[derive(Clone)] +pub struct VistaModule { + config: VistaConfig, +} + +impl VistaModule { + pub fn new(config: VistaConfig) -> Self { + Self { config } + } +} + +pub struct VistaInvocation { + config: VistaConfig, + control: Arc, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +enum Termination { + None = 0, + Cancelled = 1, + TimedOut = 2, +} + +struct ProcessControl { + termination: AtomicU8, + child: Mutex, + #[cfg(not(test))] + invocation: ExecutorInvocationControl, + #[cfg(test)] + invocation: Option, +} + +enum ChildOwnership { + Pending, + Live(i32), + Reaping, + Reaped, +} + +impl ProcessControl { + fn new(invocation: ExecutorInvocationControl) -> Self { + Self { + termination: AtomicU8::new(Termination::None as u8), + child: Mutex::new(ChildOwnership::Pending), + #[cfg(not(test))] + invocation, + #[cfg(test)] + invocation: Some(invocation), + } + } + + #[cfg(test)] + fn detached() -> Self { + Self { + termination: AtomicU8::new(Termination::None as u8), + child: Mutex::new(ChildOwnership::Pending), + invocation: None, + } + } + + fn termination(&self) -> Termination { + match self.termination.load(Ordering::Acquire) { + 1 => Termination::Cancelled, + 2 => Termination::TimedOut, + _ => match self.executor_interruption() { + Some(InterruptionReason::Cancelled) => Termination::Cancelled, + Some(InterruptionReason::TimedOut) => Termination::TimedOut, + None => Termination::None, + }, + } + } + + fn executor_interruption(&self) -> Option { + #[cfg(not(test))] + { + self.invocation.interruption_reason() + } + #[cfg(test)] + { + self.invocation + .as_ref() + .and_then(ExecutorInvocationControl::interruption_reason) + } + } + + fn install(&self, process_group: i32) { + let mut ownership = self.child.lock(); + debug_assert!(matches!(*ownership, ChildOwnership::Pending)); + *ownership = ChildOwnership::Live(process_group); + if self.termination() != Termination::None { + let _ = kill_process_group(process_group); + } + } + + fn terminate(&self, reason: Termination) -> bool { + let changed = self + .termination + .compare_exchange( + Termination::None as u8, + reason as u8, + Ordering::AcqRel, + Ordering::Acquire, + ) + .is_ok(); + if changed + && let ChildOwnership::Live(process_group) = *self.child.lock() + { + let _ = kill_process_group(process_group); + } + changed + } + + fn synchronize_interruption(&self) -> Termination { + let reason = self.termination(); + if reason != Termination::None { + self.terminate(reason); + } + reason + } + + fn kill_live_group(&self) { + if let ChildOwnership::Live(process_group) = *self.child.lock() { + let _ = kill_process_group(process_group); + } + } + + fn reap_exited( + &self, + child: &mut std::process::Child, + ) -> std::io::Result { + let mut ownership = self.child.lock(); + debug_assert!(matches!(*ownership, ChildOwnership::Live(_))); + *ownership = ChildOwnership::Reaping; + let status = child.wait(); + *ownership = ChildOwnership::Reaped; + status + } + + fn kill_and_reap(&self, child: &mut std::process::Child) { + let mut ownership = self.child.lock(); + if let ChildOwnership::Live(process_group) = *ownership { + let _ = kill_process_group(process_group); + } + *ownership = ChildOwnership::Reaping; + let _ = child.wait(); + *ownership = ChildOwnership::Reaped; + } +} + +impl CapabilityModule for VistaModule { + type Invocation = VistaInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[IMPORT_NAME] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + bindings::VistaProvider::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, context: CapabilityContext<'_>) -> Self::Invocation { + VistaInvocation { + config: self.config.clone(), + control: Arc::new(ProcessControl::new(context.control().clone())), + } + } +} + +impl Host for InvocationStore { + fn get(&mut self, request: ArtifactRequest) -> Result { + self.capability_mut().get(request) + } +} + +impl VistaInvocation { + fn get(&mut self, request: ArtifactRequest) -> Result { + if !request_is_valid(&request) + || request.slug != self.config.slug + || request.version != self.config.version + { + return Err(VistaError::Denied); + } + match self.control.termination() { + Termination::Cancelled => return Err(VistaError::Cancelled), + Termination::TimedOut => return Err(VistaError::DeadlineExceeded), + Termination::None => {} + } + + let mut command = Command::new(&self.config.executable); + command + .arg("artifact") + .arg("get") + .arg(&request.slug) + .arg(format!("v{}", request.version)) + .arg("--output") + .arg("json") + .current_dir(&self.config.cwd) + .env_clear() + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + // SAFETY: this runs in the freshly-forked child before exec and calls only async-signal-safe + // setpgid. The dedicated process group is the cancellation/reaping boundary. + unsafe { + command.pre_exec(|| { + if libc::setpgid(0, 0) == -1 { + Err(std::io::Error::last_os_error()) + } else { + Ok(()) + } + }); + } + let mut child = command.spawn().map_err(|_| VistaError::Unavailable)?; + self.control.install(child.id() as i32); + let Some(stdout) = child.stdout.take() else { + self.control.kill_and_reap(&mut child); + return Err(VistaError::Unavailable); + }; + let Some(stderr) = child.stderr.take() else { + self.control.kill_and_reap(&mut child); + return Err(VistaError::Unavailable); + }; + let stdout_reader = match thread::Builder::new() + .name("st2-vista-stdout".into()) + .spawn(move || drain_bounded(stdout, MAX_STDOUT_BYTES)) + { + Ok(reader) => reader, + Err(_) => { + self.control.kill_and_reap(&mut child); + return Err(VistaError::Unavailable); + } + }; + let stderr_reader = match thread::Builder::new() + .name("st2-vista-stderr".into()) + .spawn(move || drain_bounded(stderr, MAX_STDERR_BYTES)) + { + Ok(reader) => reader, + Err(_) => { + self.control.kill_and_reap(&mut child); + let _ = stdout_reader.join(); + return Err(VistaError::Unavailable); + } + }; + let (completed_tx, completed_rx) = mpsc::sync_channel(1); + let deadline = Instant::now() + self.config.deadline; + let deadline_control = Arc::clone(&self.control); + let timer = match thread::Builder::new() + .name("st2-vista-deadline".into()) + .spawn(move || loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + deadline_control.terminate(Termination::TimedOut); + return; + } + match completed_rx.recv_timeout(remaining.min(Duration::from_millis(10))) { + Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return, + Err(mpsc::RecvTimeoutError::Timeout) => {} + } + if deadline_control.synchronize_interruption() != Termination::None { + return; + } + }) + { + Ok(timer) => timer, + Err(_) => { + self.control.kill_and_reap(&mut child); + let _ = stdout_reader.join(); + let _ = stderr_reader.join(); + return Err(VistaError::Unavailable); + } + }; + if wait_without_reaping(child.id()).is_err() { + self.control.kill_and_reap(&mut child); + let _ = completed_tx.send(()); + let _ = timer.join(); + let _ = stdout_reader.join(); + let _ = stderr_reader.join(); + return Err(VistaError::Unavailable); + } + let stdout = stdout_reader.join(); + let stderr = stderr_reader.join(); + self.control.kill_live_group(); + let _ = completed_tx.send(()); + let _ = timer.join(); + let status = self.control.reap_exited(&mut child); + let status = status.map_err(|_| VistaError::Unavailable)?; + let (stdout, stdout_truncated) = stdout.map_err(|_| VistaError::Unavailable)??; + let (stderr, stderr_truncated) = stderr.map_err(|_| VistaError::Unavailable)??; + match self.control.termination() { + Termination::Cancelled => return Err(VistaError::Cancelled), + Termination::TimedOut => return Err(VistaError::DeadlineExceeded), + Termination::None => {} + } + if stdout_truncated || stderr_truncated { + return Err(VistaError::ResourceExhausted); + } + let exit = status.code().map_or_else( + || ExitStatus::Signal(status.signal().unwrap_or(0)), + ExitStatus::Code, + ); + Ok(Outcome { + stdout, + stderr, + stdout_truncated, + stderr_truncated, + exit, + }) + } +} + +fn request_is_valid(request: &ArtifactRequest) -> bool { + valid_slug(&request.slug) && (1..=MAX_VERSION).contains(&request.version) +} + +fn valid_slug(slug: &str) -> bool { + !slug.is_empty() + && slug.len() <= 128 + && slug.bytes().enumerate().all(|(index, byte)| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || (byte == b'-' && index > 0) + }) + && !slug.ends_with('-') + && !slug.contains("--") +} + +fn drain_bounded( + mut input: impl std::io::Read, + limit: usize, +) -> Result<(Vec, bool), VistaError> { + let mut retained = Vec::with_capacity(limit.min(64 * 1024)); + let mut truncated = false; + let mut buffer = [0_u8; 16 * 1024]; + loop { + let read = input + .read(&mut buffer) + .map_err(|_| VistaError::Unavailable)?; + if read == 0 { + break; + } + let remaining = limit.saturating_sub(retained.len()); + retained + .write_all(&buffer[..read.min(remaining)]) + .map_err(|_| VistaError::Unavailable)?; + truncated |= read > remaining; + } + Ok((retained, truncated)) +} + +fn kill_process_group(process_group: i32) -> bool { + // SAFETY: negative pid addresses the process group created by pre_exec; SIGKILL makes the + // deadline a hard bound even when the provider subprocess ignores graceful signals. + unsafe { libc::kill(-process_group, libc::SIGKILL) == 0 } +} + +fn wait_without_reaping(pid: u32) -> std::io::Result<()> { + loop { + // SAFETY: `info` is initialized for the kernel, and WNOWAIT keeps the child waitable so its + // process-group identity cannot be recycled before ownership is fenced. + let result = unsafe { + let mut info = std::mem::zeroed::(); + libc::waitid( + libc::P_PID, + pid, + &mut info, + libc::WEXITED | libc::WNOWAIT, + ) + }; + if result == 0 { + return Ok(()); + } + let error = std::io::Error::last_os_error(); + if error.kind() != std::io::ErrorKind::Interrupted { + return Err(error); + } + } +} + +fn resolve_executable(executable: &Path) -> Option { + if executable.is_absolute() { + return executable_is_runnable(executable).then(|| executable.to_path_buf()); + } + let validation_cwd = std::env::current_dir().ok()?; + let search_path = std::env::var_os("PATH"); + resolve_executable_at(executable, &validation_cwd, search_path.as_deref()) +} + +fn resolve_executable_at( + executable: &Path, + validation_cwd: &Path, + search_path: Option<&std::ffi::OsStr>, +) -> Option { + debug_assert!(validation_cwd.is_absolute()); + if executable.components().count() > 1 { + let candidate = validation_cwd.join(executable); + return executable_is_runnable(&candidate).then_some(candidate); + } + let search_path = search_path?; + std::env::split_paths(search_path) + .map(|directory| { + let directory = if directory.is_absolute() { + directory + } else { + validation_cwd.join(directory) + }; + directory.join(executable) + }) + .find(|candidate| executable_is_runnable(candidate)) +} + +fn executable_is_runnable(path: &Path) -> bool { + std::fs::metadata(path).is_ok_and(|metadata| { + metadata.is_file() && metadata.permissions().mode() & 0o111 != 0 + }) +} + +#[cfg(test)] +mod tests { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt as _; + + use super::*; + + fn write_executable(path: &Path, contents: &str) { + std::fs::write(path, contents).unwrap(); + let mut permissions = std::fs::metadata(path).unwrap().permissions(); + permissions.set_mode(0o700); + std::fs::set_permissions(path, permissions).unwrap(); + } + + fn config( + executable: &Path, + cwd: &Path, + deadline: Duration, + slug: &str, + version: u64, + ) -> VistaConfig { + VistaConfig::resolve( + executable, + cwd.to_path_buf(), + slug.to_owned(), + version, + deadline, + ) + .unwrap() + } + + fn request(slug: &str, version: u64) -> ArtifactRequest { + ArtifactRequest { + slug: slug.into(), + version, + } + } + + #[test] + fn fixed_command_uses_exact_read_only_argv() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + write_executable( + &executable, + "#!/bin/sh\nprintf '%s\\n' \"$#|$1|$2|$3|$4|$5|$6\"\n", + ); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_secs(1), + "release-notes", + 7, + ), + control: Arc::new(ProcessControl::detached()), + }; + let outcome = invocation.get(request("release-notes", 7)).unwrap(); + assert_eq!( + outcome.stdout, + b"6|artifact|get|release-notes|v7|--output|json\n" + ); + } + + #[test] + fn exact_artifact_scope_is_denied_before_spawn() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + let marker = temporary.path().join("spawned"); + write_executable(&executable, "#!/bin/sh\ntouch \"$PWD/spawned\"\n"); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_secs(1), + "release-notes", + 7, + ), + control: Arc::new(ProcessControl::detached()), + }; + for denied in [ + request("other-valid-slug", 7), + request("release-notes", 8), + request("", 7), + request("-leading", 7), + request("trailing-", 7), + request("two--dashes", 7), + request("Upper", 7), + request("release-notes", 0), + request("release-notes", MAX_VERSION + 1), + ] { + assert!(matches!(invocation.get(denied), Err(VistaError::Denied))); + } + assert!(!marker.exists()); + } + + #[test] + fn artifact_version_is_bounded_by_the_javascript_safe_integer_contract() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + write_executable(&executable, "#!/bin/sh\nexit 0\n"); + assert!(request_is_valid(&request("release", MAX_VERSION))); + assert!(!request_is_valid(&request("release", MAX_VERSION + 1))); + assert!( + VistaConfig::resolve( + &executable, + temporary.path().to_path_buf(), + "release".into(), + MAX_VERSION + 1, + Duration::from_secs(1), + ) + .is_err() + ); + } + + #[test] + fn cancelled_before_spawn_is_effect_free() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + let marker = temporary.path().join("spawned"); + write_executable(&executable, "#!/bin/sh\ntouch \"$PWD/spawned\"\n"); + let control = Arc::new(ProcessControl::detached()); + control.terminate(Termination::Cancelled); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_secs(1), + "valid", + 1, + ), + control, + }; + assert!(matches!( + invocation.get(request("valid", 1)), + Err(VistaError::Cancelled) + )); + assert!(!marker.exists()); + } + + #[test] + fn deadline_owns_the_process_group_until_inherited_output_pipes_close() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + let holder = temporary.path().join("pipe-holder"); + std::os::unix::fs::symlink(std::env::current_exe().unwrap(), &holder).unwrap(); + let fifo = temporary.path().join("block"); + let started = temporary.path().join("started"); + for path in [&fifo, &started] { + let path = CString::new(path.as_os_str().as_bytes()).unwrap(); + // SAFETY: the pathname is a live NUL-terminated byte string owned for the call. + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o600) }, 0); + } + write_executable( + &executable, + "#!/bin/sh\n(trap '' HUP; exec \"$PWD/pipe-holder\" --ignored --exact vista::tests::inherited_pipe_holder --nocapture) &\nread marker < \"$PWD/started\"\nprintf 'direct child exited\\n'\nexit 0\n", + ); + let control = Arc::new(ProcessControl::detached()); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_millis(100), + "valid", + 1, + ), + control: Arc::clone(&control), + }; + assert!(matches!( + invocation.get(request("valid", 1)), + Err(VistaError::DeadlineExceeded) + )); + assert!(matches!(*control.child.lock(), ChildOwnership::Reaped)); + } + + #[test] + fn cancellation_owns_the_process_group_until_inherited_output_pipes_close() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + let holder = temporary.path().join("pipe-holder"); + std::os::unix::fs::symlink(std::env::current_exe().unwrap(), &holder).unwrap(); + for name in ["block", "started", "cancel-ready"] { + let path = temporary.path().join(name); + let path = CString::new(path.as_os_str().as_bytes()).unwrap(); + // SAFETY: the pathname is a live NUL-terminated byte string owned for the call. + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o600) }, 0); + } + write_executable( + &executable, + "#!/bin/sh\n(trap '' HUP; exec \"$PWD/pipe-holder\" --ignored --exact vista::tests::inherited_pipe_holder --nocapture) &\nread marker < \"$PWD/started\"\nprintf 'ready\\n' > \"$PWD/cancel-ready\"\nprintf 'direct child exited\\n'\nexit 0\n", + ); + let control = Arc::new(ProcessControl::detached()); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_secs(10), + "valid", + 1, + ), + control: Arc::clone(&control), + }; + let worker = thread::spawn(move || invocation.get(request("valid", 1))); + let mut ready = String::new(); + let mut ready_pipe = std::fs::File::open(temporary.path().join("cancel-ready")).unwrap(); + std::io::Read::read_to_string(&mut ready_pipe, &mut ready).unwrap(); + assert_eq!(ready, "ready\n"); + assert!(control.terminate(Termination::Cancelled)); + assert!(matches!(worker.join().unwrap(), Err(VistaError::Cancelled))); + assert!(matches!(*control.child.lock(), ChildOwnership::Reaped)); + } + + #[test] + fn successful_completion_kills_descendants_that_closed_output_pipes() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + let holder = temporary.path().join("pipe-holder"); + std::os::unix::fs::symlink(std::env::current_exe().unwrap(), &holder).unwrap(); + for name in ["block", "started"] { + let path = temporary.path().join(name); + let path = CString::new(path.as_os_str().as_bytes()).unwrap(); + // SAFETY: the pathname is a live NUL-terminated byte string owned for the call. + assert_eq!(unsafe { libc::mkfifo(path.as_ptr(), 0o600) }, 0); + } + std::fs::File::create(temporary.path().join("descendant-lock")).unwrap(); + write_executable( + &executable, + "#!/bin/sh\n(trap '' HUP; exec \"$PWD/pipe-holder\" --ignored --exact vista::tests::closed_pipe_descendant_holder --nocapture) &\nread marker < \"$PWD/started\"\nexit 0\n", + ); + let control = Arc::new(ProcessControl::detached()); + let mut invocation = VistaInvocation { + config: config( + &executable, + temporary.path(), + Duration::from_secs(1), + "valid", + 1, + ), + control: Arc::clone(&control), + }; + let outcome = invocation.get(request("valid", 1)).unwrap(); + assert!(matches!(outcome.exit, ExitStatus::Code(0))); + let lock_path = temporary.path().join("descendant-lock"); + let (acquired_tx, acquired_rx) = mpsc::sync_channel(0); + let waiter = thread::spawn(move || { + let lock = std::fs::OpenOptions::new() + .write(true) + .open(lock_path) + .unwrap(); + // SAFETY: `lock` owns a live file descriptor for the duration of the call. + let result = + unsafe { libc::flock(std::os::fd::AsRawFd::as_raw_fd(&lock), libc::LOCK_EX) }; + let _ = acquired_tx.send(result); + }); + let acquired = acquired_rx.recv_timeout(Duration::from_secs(1)); + if let Err(error) = &acquired { + drop(acquired_rx); + let pid: i32 = + std::fs::read_to_string(temporary.path().join("descendant-pid")) + .unwrap() + .parse() + .unwrap(); + // SAFETY: the fixture wrote its live pid after acquiring the lock. + unsafe { libc::kill(pid, libc::SIGKILL) }; + waiter.join().unwrap(); + panic!("the descendant retained its lock after provider completion: {error}"); + } + assert_eq!(acquired.unwrap(), 0); + waiter.join().unwrap(); + assert!(matches!(*control.child.lock(), ChildOwnership::Reaped)); + } + + #[test] + #[ignore = "subprocess fixture for inherited output pipe ownership"] + fn inherited_pipe_holder() { + let mut started = std::fs::OpenOptions::new() + .write(true) + .open("started") + .unwrap(); + writeln!(started, "ready").unwrap(); + drop(started); + let _blocked = std::fs::File::open("block").unwrap(); + } + + #[test] + #[ignore = "subprocess fixture for descendant process-group ownership"] + fn closed_pipe_descendant_holder() { + let lock = std::fs::OpenOptions::new() + .write(true) + .open("descendant-lock") + .unwrap(); + // SAFETY: `lock` owns a live file descriptor for the duration of the call. + assert_eq!( + unsafe { libc::flock(std::os::fd::AsRawFd::as_raw_fd(&lock), libc::LOCK_EX) }, + 0 + ); + std::fs::write("descendant-pid", std::process::id().to_string()).unwrap(); + // SAFETY: the subprocess fixture intentionally releases inherited output pipes while + // retaining the process-group-owned lock. + unsafe { + libc::close(libc::STDOUT_FILENO); + libc::close(libc::STDERR_FILENO); + } + let mut started = std::fs::OpenOptions::new() + .write(true) + .open("started") + .unwrap(); + writeln!(started, "ready").unwrap(); + drop(started); + let _blocked = std::fs::File::open("block").unwrap(); + } +} diff --git a/crates/st2-resource-providers/tests/github_pr_component.rs b/crates/st2-resource-providers/tests/github_pr_component.rs new file mode 100644 index 00000000..889c08c1 --- /dev/null +++ b/crates/st2-resource-providers/tests/github_pr_component.rs @@ -0,0 +1,373 @@ +use std::fs; +use std::path::PathBuf; +use std::sync::{Arc, Barrier}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use serde_json::{Value, json}; +use st2_resource_protocol::{ObservationResult, SnapshotDigest}; +use st2_resource_wasip2::{ + CapabilityContext, CapabilityModule, Executor, InterruptionReason, InvocationControl, + InvocationStore, ObservationRequest, ObserveError, RuntimeConfig, +}; +use wasmtime::component::{HasSelf, Linker}; + +mod bindings { + wasmtime::component::bindgen!({ + path: "../../wit/github-pr", + world: "github-pr-provider", + }); +} + +use bindings::compoundingtech::st2_github_pr::github_pr::{ + Host, PullRequestError, PullRequestRequest, PullRequestResponse, SourceObject, + SourceObservation, SourceSnapshot, +}; + +const IMPORT_NAME: &str = "compoundingtech:st2-github-pr/github-pr@0.1.0"; +const HEAD_SHA: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + +#[derive(Clone, Default)] +struct FixtureModule { + calls: Arc, + bindings: Arc, +} + +struct FixtureInvocation { + calls: Arc, + bindings: Arc, +} + +impl CapabilityModule for FixtureModule { + type Invocation = FixtureInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[IMPORT_NAME] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + bindings::GithubPrProvider::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, _context: CapabilityContext<'_>) -> Self::Invocation { + FixtureInvocation { + calls: Arc::clone(&self.calls), + bindings: Arc::clone(&self.bindings), + } + } +} + +impl Host for InvocationStore { + fn get( + &mut self, + request: PullRequestRequest, + ) -> Result { + assert_eq!(request.owner, "example"); + assert_eq!(request.repo, "demo"); + assert_eq!(request.number, 389); + let call = self.capability().calls.fetch_add(1, Ordering::SeqCst); + Ok(match call { + 0 => PullRequestResponse::Ok(SourceObservation { + current: source(false, "2026-08-30T12:34:56Z"), + previous: None, + }), + 1 => PullRequestResponse::Ok(SourceObservation { + current: source(true, "2026-08-30T12:35:56Z"), + previous: Some(source(false, "2026-08-30T12:34:56Z")), + }), + _ => PullRequestResponse::Ok(SourceObservation { + current: source(true, "2026-08-30T12:36:56Z"), + previous: Some(source(true, "2026-08-30T12:35:56Z")), + }), + }) + } + + fn bind_snapshot(&mut self, digest: Vec) -> Result<(), PullRequestError> { + assert_eq!(digest.len(), 32); + self.capability().bindings.fetch_add(1, Ordering::SeqCst); + Ok(()) + } +} + +#[derive(Clone)] +struct BlockingModule { + entered: Arc, +} + +struct BlockingInvocation { + entered: Arc, + control: InvocationControl, +} + +impl CapabilityModule for BlockingModule { + type Invocation = BlockingInvocation; + + fn import_names(&self) -> &'static [&'static str] { + &[IMPORT_NAME] + } + + fn add_to_linker( + &self, + linker: &mut Linker>, + ) -> Result<(), wasmtime::Error> { + bindings::GithubPrProvider::add_to_linker::<_, HasSelf<_>>(linker, |state| state) + } + + fn begin(&self, context: CapabilityContext<'_>) -> Self::Invocation { + BlockingInvocation { + entered: Arc::clone(&self.entered), + control: context.control().clone(), + } + } +} + +impl Host for InvocationStore { + fn get( + &mut self, + _request: PullRequestRequest, + ) -> Result { + let capability = self.capability(); + capability.entered.wait(); + match capability.control.wait_for_interruption() { + InterruptionReason::Cancelled => Err(PullRequestError::Unavailable), + InterruptionReason::TimedOut => Err(PullRequestError::DeadlineExceeded), + } + } + + fn bind_snapshot(&mut self, _digest: Vec) -> Result<(), PullRequestError> { + Ok(()) + } +} + +#[test] +fn component_preserves_snapshot_facets_delta_topics_and_semantic_replay() { + let module = FixtureModule::default(); + let bindings = Arc::clone(&module.bindings); + let executor = Executor::new(RuntimeConfig::default(), None, module).unwrap(); + let bytes = fs::read(component()).unwrap(); + let loaded = executor.load(&bytes).unwrap(); + let descriptor = executor.describe(&loaded, None).unwrap(); + assert_eq!( + descriptor.topics, + [ + "ci.failure", + "mergeability.conflict", + "review.requested", + "terminal" + ] + ); + assert_eq!( + descriptor.snapshot_schema_id, + "dev.schickling.github-pr.snapshot.v1" + ); + assert_eq!(descriptor.snapshot_media_type, "application/json"); + + let first = executor + .observe(&loaded, &request(1, None), None) + .unwrap(); + let first = match first { + ObservationResult::Published { publication } => publication, + other => panic!("first observation must publish, got {other:?}"), + }; + assert_eq!( + first.topics, + [ + "ci.failure", + "mergeability.conflict", + "review.requested", + "terminal" + ] + ); + let snapshot: Value = serde_json::from_slice(first.bytes.as_slice()).unwrap(); + assert_eq!( + snapshot, + json!({ + "ci": { + "checkRuns": [ + {"conclusion": "failure", "detailsUrl": "https://example.invalid/a", "name": "a-build", "status": "completed"}, + {"conclusion": "success", "detailsUrl": "https://example.invalid/z", "name": "z-test", "status": "completed"} + ], + "state": "failure", + "statuses": [ + {"context": "a/build", "description": "failed", "state": "failure", "targetUrl": "https://example.invalid/a"}, + {"context": "z/lint", "description": null, "state": "success", "targetUrl": "https://example.invalid/z"} + ] + }, + "facets": {"ciFailure": true, "mergeConflict": false, "reviewRequested": true, "terminal": false}, + "number": 389, + "observedAt": "2026-08-30T12:34:56Z", + "pullRequest": { + "apiUrl": "https://api.github.com/repos/example/demo/pulls/389", + "base": {"ref": "main"}, + "closedAt": null, + "draft": false, + "head": {"ref": "resources", "sha": HEAD_SHA}, + "htmlUrl": "https://github.com/example/demo/pull/389", + "mergeable": true, + "mergeableState": "clean", + "merged": false, + "mergedAt": null, + "requestedReviewers": ["a-reviewer", "z-reviewer"], + "requestedTeams": ["team-a", "team-b"], + "state": "open" + }, + "repository": {"name": "demo", "owner": "example"}, + "schema": "dev.schickling.github-pr.snapshot.v1", + "uri": "github-pr://example/demo/389" + }) + ); + + let second = executor + .observe( + &loaded, + &request(2, Some(SnapshotDigest::of(first.bytes.as_slice()))), + None, + ) + .unwrap(); + let second = match second { + ObservationResult::Published { publication } => publication, + other => panic!("terminal transition must publish, got {other:?}"), + }; + assert_eq!(second.topics, ["terminal"]); + assert_eq!( + second + .facts + .as_ref() + .unwrap() + .iter() + .map(|fact| fact.key()) + .collect::>(), + ["facets.terminal"] + ); + + let third = executor + .observe( + &loaded, + &request(3, Some(SnapshotDigest::of(second.bytes.as_slice()))), + None, + ) + .unwrap(); + assert_eq!(third, ObservationResult::Unchanged); + assert_eq!(bindings.load(Ordering::SeqCst), 3); +} + +#[test] +fn component_import_cancellation_is_deterministic() { + assert!(matches!( + interrupt_blocked_import(BlockedInterruption::Cancel), + ObserveError::Cancelled + )); +} + +#[test] +fn component_import_deadline_is_deterministic() { + assert!(matches!( + interrupt_blocked_import(BlockedInterruption::TimeOut), + ObserveError::TimedOut + )); +} + +enum BlockedInterruption { + Cancel, + TimeOut, +} + +fn interrupt_blocked_import(interruption: BlockedInterruption) -> ObserveError { + let entered = Arc::new(Barrier::new(2)); + let executor = Executor::new( + RuntimeConfig::default(), + None, + BlockingModule { + entered: Arc::clone(&entered), + }, + ) + .unwrap(); + let bytes = fs::read(component()).unwrap(); + let loaded = executor.load(&bytes).unwrap(); + let handle = executor.interruption_handle(); + let trigger = handle.clone(); + let observing = + std::thread::spawn(move || executor.observe(&loaded, &request(4, None), Some(&handle))); + entered.wait(); + match interruption { + BlockedInterruption::Cancel => assert!(trigger.cancel()), + BlockedInterruption::TimeOut => assert!(trigger.time_out()), + } + observing.join().unwrap().unwrap_err() +} + +fn request(invocation_id: u64, prior_digest: Option) -> ObservationRequest { + ObservationRequest { + invocation_id, + uri: "github-pr://example/demo/389".into(), + selector: json!({ + "owner": "example", + "repo": "demo", + "number": 389, + "topics": ["ci.failure", "terminal"] + }), + prior_digest, + demand_watermark: Some(invocation_id), + } +} + +fn source(terminal: bool, observed_at: &str) -> SourceSnapshot { + let state = if terminal { "closed" } else { "open" }; + SourceSnapshot { + pull_request: object( + "\"pull-v1\"", + json!({ + "number": 389, + "url": "https://api.github.com/repos/example/demo/pulls/389", + "html_url": "https://github.com/example/demo/pull/389", + "state": state, + "draft": false, + "merged": terminal, + "merged_at": if terminal { Some("2026-08-30T12:35:00Z") } else { None }, + "closed_at": if terminal { Some("2026-08-30T12:35:00Z") } else { None }, + "mergeable": true, + "mergeable_state": "clean", + "head": {"sha": HEAD_SHA, "ref": "resources"}, + "base": {"ref": "main"}, + "requested_reviewers": [{"login": "z-reviewer"}, {"login": "a-reviewer"}], + "requested_teams": [{"slug": "team-b"}, {"slug": "team-a"}] + }), + ), + check_runs: object( + "\"checks-v1\"", + json!({ + "check_runs": [ + {"name": "z-test", "status": "completed", "conclusion": "success", "details_url": "https://example.invalid/z"}, + {"name": "a-build", "status": "completed", "conclusion": "failure", "details_url": "https://example.invalid/a"} + ] + }), + ), + combined_status: object( + "\"status-v1\"", + json!({ + "state": "failure", + "statuses": [ + {"context": "z/lint", "state": "success", "target_url": "https://example.invalid/z", "description": null}, + {"context": "a/build", "state": "failure", "target_url": "https://example.invalid/a", "description": "failed"} + ] + }), + ), + observed_at: observed_at.into(), + } +} + +fn object(etag: &str, value: Value) -> SourceObject { + SourceObject { + etag: Some(etag.into()), + body: serde_json::to_vec(&value).unwrap(), + } +} + +fn component() -> PathBuf { + PathBuf::from( + std::env::var_os("ST2_GITHUB_PR_COMPONENT") + .expect("ST2_GITHUB_PR_COMPONENT is not set"), + ) +} diff --git a/flake.nix b/flake.nix index 52556394..f4963f1d 100644 --- a/flake.nix +++ b/flake.nix @@ -144,7 +144,11 @@ "--exclude" "st2-github-issue-component" "--exclude" + "st2-github-pr-component" + "--exclude" "st2-pty-stats-component" + "--exclude" + "st2-vista-component" "--lib" "--bins" "--test" @@ -192,7 +196,11 @@ "--exclude" "st2-github-issue-component" "--exclude" + "st2-github-pr-component" + "--exclude" "st2-pty-stats-component" + "--exclude" + "st2-vista-component" "--test" "resync" "--test" @@ -253,11 +261,21 @@ wasmName = "st2_github_issue_component"; }; + st2GitHubPrComponent = buildProviderComponent { + package = "st2-github-pr-component"; + wasmName = "st2_github_pr_component"; + }; + st2PtyStatsComponent = buildProviderComponent { package = "st2-pty-stats-component"; wasmName = "st2_pty_stats_component"; }; + st2VistaComponent = buildProviderComponent { + package = "st2-vista-component"; + wasmName = "st2_vista_component"; + }; + st2ProviderRuntime = st2.overrideAttrs (old: { pname = "st2-provider-runtime"; cargoBuildFeatures = (old.cargoBuildFeatures or [ ]) ++ [ "wasip2-provider-runtime" ]; @@ -271,11 +289,15 @@ st2ProviderRuntimeCheck = st2ProviderRuntime.overrideAttrs (_: { pname = "st2-provider-runtime-check"; ST2_GITHUB_ISSUE_COMPONENT = "${st2GitHubIssueComponent}/share/st2/providers/st2_github_issue_component.component.wasm"; + ST2_GITHUB_PR_COMPONENT = "${st2GitHubPrComponent}/share/st2/providers/st2_github_pr_component.component.wasm"; ST2_PTY_STATS_COMPONENT = "${st2PtyStatsComponent}/share/st2/providers/st2_pty_stats_component.component.wasm"; + ST2_VISTA_COMPONENT = "${st2VistaComponent}/share/st2/providers/st2_vista_component.component.wasm"; cargoTestFlags = [ "-p" "st2-resource-providers" "--lib" + "--test" + "github_pr_component" "-p" "st2" "--features" @@ -388,7 +410,9 @@ packages.st2-wasm-resolver = st2WasmResolver; packages.st2-provider-runtime = st2ProviderRuntime; packages.st2-github-issue-component = st2GitHubIssueComponent; + packages.st2-github-pr-component = st2GitHubPrComponent; packages.st2-pty-stats-component = st2PtyStatsComponent; + packages.st2-vista-component = st2VistaComponent; packages.default = st2; # `nix flake check` is the whole CI: it builds the package — which runs @@ -410,7 +434,9 @@ checks.wasip2-resource-executor = st2Wasip2ExecutorCheck; checks.wasip2-resource-providers = st2ProviderRuntimeCheck; checks.github-issue-component = st2GitHubIssueComponent; + checks.github-pr-component = st2GitHubPrComponent; checks.pty-stats-component = st2PtyStatsComponent; + checks.vista-component = st2VistaComponent; # Exercise the shipped binary, not a cargo-side surrogate: its version entrypoint runs and # the same artifact strictly admits a catalog carrying a real wasm profile module. checks.wasm-resolver-artifact = pkgs.runCommand "st2-wasm-resolver-artifact-${version}" { } '' diff --git a/src/catalog.rs b/src/catalog.rs index 5c70d2ff..f71d41b7 100644 --- a/src/catalog.rs +++ b/src/catalog.rs @@ -65,12 +65,26 @@ pub enum DeclaredProviderCapability { connect_timeout_ms: u64, total_timeout_ms: u64, }, + GitHubPr { + owner: String, + repo: String, + number: u64, + connect_timeout_ms: u64, + total_timeout_ms: u64, + }, PtyStats { executable: String, cwd: String, scope: DeclaredPtyStatsScope, deadline_ms: u64, }, + Vista { + executable: String, + cwd: String, + slug: String, + version: u64, + deadline_ms: u64, + }, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -260,6 +274,13 @@ fn parse_profile(node: &kdl::KdlNode) -> anyhow::Result { ); capability = Some(parse_github_issue_capability(scheme, runtime_child)?); } + "github-pr" => { + anyhow::ensure!( + capability.is_none() && runtime_child.children().is_none(), + "profile '{scheme}': runtime declares more than one capability" + ); + capability = Some(parse_github_pr_capability(scheme, runtime_child)?); + } "pty-stats" => { anyhow::ensure!( capability.is_none() && runtime_child.children().is_none(), @@ -267,9 +288,16 @@ fn parse_profile(node: &kdl::KdlNode) -> anyhow::Result { ); capability = Some(parse_pty_stats_capability(scheme, runtime_child)?); } + "vista" => { + anyhow::ensure!( + capability.is_none() && runtime_child.children().is_none(), + "profile '{scheme}': runtime declares more than one capability" + ); + capability = Some(parse_vista_capability(scheme, runtime_child)?); + } other => anyhow::bail!( "profile '{scheme}': runtime field '{other}' is unknown \ - (expected component, demand, github-issue, or pty-stats)" + (expected component, demand, github-issue, github-pr, pty-stats, or vista)" ), } } @@ -385,6 +413,39 @@ fn parse_github_issue_capability( }) } +fn parse_github_pr_capability( + scheme: &str, + node: &kdl::KdlNode, +) -> anyhow::Result { + anyhow::ensure!( + node.entries().len() == 5 && node.entries().iter().all(|entry| entry.name().is_some()), + "profile '{scheme}': github-pr requires owner, repo, number, \ + connect-timeout-ms, and total-timeout-ms properties" + ); + let owner = required_string_property(scheme, node, "owner")?; + let repo = required_string_property(scheme, node, "repo")?; + let number = required_u64_property(scheme, node, "number")?; + let connect_timeout_ms = required_u64_property(scheme, node, "connect-timeout-ms")?; + let total_timeout_ms = required_u64_property(scheme, node, "total-timeout-ms")?; + anyhow::ensure!( + number > 0, + "profile '{scheme}': GitHub pull request number must be positive" + ); + anyhow::ensure!( + connect_timeout_ms > 0 + && connect_timeout_ms <= total_timeout_ms + && total_timeout_ms <= 60_000, + "profile '{scheme}': GitHub deadlines must be positive, ordered, and at most 60000ms" + ); + Ok(DeclaredProviderCapability::GitHubPr { + owner, + repo, + number, + connect_timeout_ms, + total_timeout_ms, + }) +} + fn parse_pty_stats_capability( scheme: &str, node: &kdl::KdlNode, @@ -418,6 +479,46 @@ fn parse_pty_stats_capability( }) } +fn parse_vista_capability( + scheme: &str, + node: &kdl::KdlNode, +) -> anyhow::Result { + anyhow::ensure!( + node.entries().len() == 5 && node.entries().iter().all(|entry| entry.name().is_some()), + "profile '{scheme}': vista requires executable, cwd, slug, version, and deadline-ms properties" + ); + let executable = required_string_property(scheme, node, "executable")?; + let cwd = required_string_property(scheme, node, "cwd")?; + let slug = required_string_property(scheme, node, "slug")?; + let version = required_u64_property(scheme, node, "version")?; + let deadline_ms = required_u64_property(scheme, node, "deadline-ms")?; + anyhow::ensure!( + valid_vista_slug(&slug) && (1..=9_007_199_254_740_991).contains(&version), + "profile '{scheme}': Vista artifact scope is invalid" + ); + anyhow::ensure!( + deadline_ms > 0 && deadline_ms <= 60_000, + "profile '{scheme}': Vista deadline must be between 1ms and 60000ms" + ); + Ok(DeclaredProviderCapability::Vista { + executable, + cwd, + slug, + version, + deadline_ms, + }) +} + +fn valid_vista_slug(slug: &str) -> bool { + !slug.is_empty() + && slug.len() <= 128 + && slug.bytes().enumerate().all(|(index, byte)| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || (byte == b'-' && index > 0) + }) + && !slug.ends_with('-') + && !slug.contains("--") +} + fn required_string_property( scheme: &str, node: &kdl::KdlNode, @@ -920,6 +1021,85 @@ mod tests { } } + #[test] + fn github_pr_runtime_capability_is_exact_and_bounded() { + let config = parse( + r#" + profile "github-pr" { + wasm "github-pr-resolver.wasm" + runtime { + component "components/github-pr.component.wasm" + demand #true + github-pr owner="example" repo="demo" number=389 connect-timeout-ms=3000 total-timeout-ms=10000 + } + } + "#, + ) + .unwrap(); + assert_eq!( + config.profiles[0].runtime, + Some(DeclaredProfileRuntime { + component: "components/github-pr.component.wasm".into(), + capability: DeclaredProviderCapability::GitHubPr { + owner: "example".into(), + repo: "demo".into(), + number: 389, + connect_timeout_ms: 3000, + total_timeout_ms: 10000, + }, + demand: true, + }) + ); + for malformed in [ + r#"profile "github-pr" { wasm "x"; runtime { component "x"; github-pr owner="o" repo="r" number=0 connect-timeout-ms=1 total-timeout-ms=2 } }"#, + r#"profile "github-pr" { wasm "x"; runtime { component "x"; github-pr owner="o" repo="r" number=1 connect-timeout-ms=3 total-timeout-ms=2 } }"#, + r#"profile "github-pr" { wasm "x"; runtime { component "x"; github-pr owner="o" repo="r" number=1 connect-timeout-ms=1 total-timeout-ms=60001 } }"#, + ] { + assert!(parse(malformed).is_err(), "expected error for: {malformed}"); + } + } + + #[test] + fn vista_runtime_capability_is_exact_and_bounded() { + let config = parse( + r#" + profile "vista" { + wasm "vista-resolver.wasm" + runtime { + component "components/vista.component.wasm" + demand #true + vista executable="/nix/store/example/bin/vista" cwd="/var/empty" slug="release-notes" version=7 deadline-ms=10000 + } + } + "#, + ) + .unwrap(); + assert_eq!( + config.profiles[0].runtime, + Some(DeclaredProfileRuntime { + component: "components/vista.component.wasm".into(), + capability: DeclaredProviderCapability::Vista { + executable: "/nix/store/example/bin/vista".into(), + cwd: "/var/empty".into(), + slug: "release-notes".into(), + version: 7, + deadline_ms: 10000, + }, + demand: true, + }) + ); + for malformed in [ + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="release" version=1 deadline-ms=0 } }"#, + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="release" version=1 deadline-ms=60001 } }"#, + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="-release" version=1 deadline-ms=1 } }"#, + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="release" version=0 deadline-ms=1 } }"#, + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="release" version=9007199254740992 deadline-ms=1 } }"#, + r#"profile "vista" { wasm "x"; runtime { component "x"; vista executable="vista" cwd="/" slug="release" version=1 deadline-ms=1 extra="no" } }"#, + ] { + assert!(parse(malformed).is_err(), "expected error for: {malformed}"); + } + } + #[test] fn malformed_profile_blocks_fail_validation_loudly() { let loud = [ diff --git a/src/resource_profile_supervisor.rs b/src/resource_profile_supervisor.rs index dd9ede94..abc77753 100644 --- a/src/resource_profile_supervisor.rs +++ b/src/resource_profile_supervisor.rs @@ -25,7 +25,8 @@ use sha2::{Digest as _, Sha256}; use st2_resource_protocol::ProposalFence; #[cfg(feature = "wasip2-provider-runtime")] use st2_resource_providers::{ - GitHubIssueConfig, GitHubIssueModule, PtyStatsConfig, PtyStatsModule, PtyStatsScope, + GitHubIssueConfig, GitHubIssueModule, GitHubPrConfig, GitHubPrModule, PtyStatsConfig, + PtyStatsModule, PtyStatsScope, VistaConfig, VistaModule, }; #[cfg(feature = "wasip2-provider-runtime")] use st2_resource_wasip2::{ @@ -1170,10 +1171,18 @@ enum ProviderRuntime { executor: Wasip2Executor, component: LoadedComponent, }, + GitHubPr { + executor: Wasip2Executor, + component: LoadedComponent, + }, PtyStats { executor: Wasip2Executor, component: LoadedComponent, }, + Vista { + executor: Wasip2Executor, + component: LoadedComponent, + }, } #[cfg(feature = "wasip2-provider-runtime")] fn validate_provider_descriptor( @@ -1218,7 +1227,9 @@ impl ProviderRuntime { fn cancellation(&self) -> ObservationCancellation { let interruption = match self { Self::GitHubIssue { executor, .. } => executor.interruption_handle(), + Self::GitHubPr { executor, .. } => executor.interruption_handle(), Self::PtyStats { executor, .. } => executor.interruption_handle(), + Self::Vista { executor, .. } => executor.interruption_handle(), }; ObservationCancellation { interruption } } @@ -1234,11 +1245,21 @@ impl ProviderRuntime { component, .. } => executor.observe(component, request, Some(&cancellation.interruption)), + Self::GitHubPr { + executor, + component, + .. + } => executor.observe(component, request, Some(&cancellation.interruption)), Self::PtyStats { executor, component, .. } => executor.observe(component, request, Some(&cancellation.interruption)), + Self::Vista { + executor, + component, + .. + } => executor.observe(component, request, Some(&cancellation.interruption)), } } } @@ -1372,6 +1393,34 @@ impl RuntimeProcess { component, } } + crate::catalog::DeclaredProviderCapability::GitHubPr { + owner, + repo, + number, + connect_timeout_ms, + total_timeout_ms, + } => { + let module = GitHubPrModule::new(GitHubPrConfig { + owner: owner.clone(), + repo: repo.clone(), + number: *number, + connect_timeout: Duration::from_millis(*connect_timeout_ms), + total_timeout: Duration::from_millis(*total_timeout_ms), + }) + .map_err(anyhow::Error::msg)?; + let executor = Wasip2Executor::new( + Wasip2RuntimeConfig::default(), + None, + module, + )?; + let component = executor.load(&sample.component.bytes)?; + let descriptor = executor.describe(&component, None)?; + validate_provider_descriptor(&descriptor, &sample.descriptor)?; + ProviderRuntime::GitHubPr { + executor, + component, + } + } crate::catalog::DeclaredProviderCapability::PtyStats { executable, cwd, @@ -1408,6 +1457,38 @@ impl RuntimeProcess { component, } } + crate::catalog::DeclaredProviderCapability::Vista { + executable, + cwd, + slug, + version, + deadline_ms, + } => { + let executable = + crate::expand::expand_catalog(executable, catalog_root); + let cwd = crate::expand::expand_catalog(cwd, catalog_root); + let config = VistaConfig::resolve( + executable, + PathBuf::from(cwd), + slug.clone(), + *version, + Duration::from_millis(*deadline_ms), + ) + .map_err(anyhow::Error::msg)?; + let module = VistaModule::new(config); + let executor = Wasip2Executor::new( + Wasip2RuntimeConfig::default(), + None, + module, + )?; + let component = executor.load(&sample.component.bytes)?; + let descriptor = executor.describe(&component, None)?; + validate_provider_descriptor(&descriptor, &sample.descriptor)?; + ProviderRuntime::Vista { + executor, + component, + } + } }; let sequence = ID_SEQUENCE.fetch_add(1, Ordering::Relaxed); let incarnation = diff --git a/tests/resource_profile_supervisor_e2e.rs b/tests/resource_profile_supervisor_e2e.rs index f54ab9eb..a5eef930 100755 --- a/tests/resource_profile_supervisor_e2e.rs +++ b/tests/resource_profile_supervisor_e2e.rs @@ -97,6 +97,52 @@ fn supervisor_compatibility_contract_uses_the_production_pty_component() { } +#[test] +fn supervisor_spawns_vista_capability_and_preserves_stable_snapshot() { + let _guard = STATE_ENV.lock(); + let temporary = tempfile::tempdir().unwrap(); + unsafe { std::env::set_var("XDG_STATE_HOME", temporary.path().join("state")) }; + let executable = temporary.path().join("vista"); + write_executable( + &executable, + r#"#!/bin/sh +if [ "$#" -ne 6 ] || [ "$1" != artifact ] || [ "$2" != get ] || [ "$3" != release-notes ] || [ "$4" != v7 ] || [ "$5" != --output ] || [ "$6" != json ]; then + exit 64 +fi +printf '%s\n' '{"schemaVersion":1,"uri":"vista://release-notes/v7","slug":"release-notes","version":7,"author":"agent","timestamp":"2026-09-02T10:00:00Z","changeSummary":"created","parent":null,"retired":false,"state":"ready","canonicalUrl":"https://vista.example/release-notes/v7"}' +"#, + ); + let selector = + r#"{"slug":"release-notes","version":7,"topics":["ready","updated","failed","expired"]}"#; + let vista = ProviderFixture::new_with_uri( + temporary.path().join("catalog"), + "vista", + "vista://release-notes/v7", + component("ST2_VISTA_COMPONENT"), + selector, + &format!( + "vista executable={:?} cwd={:?} slug=\"release-notes\" version=7 deadline-ms=10000", + executable, + temporary.path() + ), + "dev.schickling.vista.snapshot.v1", + &["ready", "updated", "failed", "expired"], + ); + + let first = vista.observe(None); + assert_eq!(first.status, ObserveReceiptStatus::SettledChanged, "{first:?}"); + let first_bytes = fs::read(vista.snapshot()).unwrap(); + let snapshot: serde_json::Value = serde_json::from_slice(&first_bytes).unwrap(); + assert_eq!( + snapshot.get("schema").and_then(serde_json::Value::as_str), + Some("dev.schickling.vista.snapshot.v1") + ); + assert!(snapshot.get("observedAt").is_none()); + let replay = vista.observe(first.digest); + assert_eq!(replay.status, ObserveReceiptStatus::SettledUnchanged); + assert_eq!(fs::read(vista.snapshot()).unwrap(), first_bytes); +} + #[test] fn production_component_preserves_resync_filter_catch_up_and_scope_isolation() { let _guard = STATE_ENV.lock(); @@ -519,7 +565,7 @@ struct ProviderFixture { root: PathBuf, agent: PathBuf, host: String, - scheme: String, + uri: String, selector: Mutex, supervisor: ResourceProfileSupervisor, } @@ -534,6 +580,29 @@ impl ProviderFixture { capability: &str, schema_id: &str, topic: &str, + ) -> Self { + Self::new_with_uri( + root, + scheme, + &format!("{scheme}://subject"), + component, + selector, + capability, + schema_id, + &[topic], + ) + } + + #[allow(clippy::too_many_arguments)] + fn new_with_uri( + root: PathBuf, + scheme: &str, + uri: &str, + component: PathBuf, + selector: &str, + capability: &str, + schema_id: &str, + topics: &[&str], ) -> Self { let host = "e2e".to_owned(); let agent = root.join("agents/e2e/worker"); @@ -546,7 +615,7 @@ impl ProviderFixture { fs::copy(component, installed_component).unwrap(); fs::write( root.join("resolver.wasm"), - observable_resolver_wasm(schema_id, topic, selector), + observable_resolver_wasm(schema_id, topics, selector), ) .unwrap(); fs::write( @@ -556,14 +625,14 @@ impl ProviderFixture { ), ) .unwrap(); - write_agent(&agent, &host, scheme, selector); + write_agent(&agent, &host, uri, selector); st2::event::publish_owner_binding_for_test(&root, &host).unwrap(); let supervisor = ResourceProfileSupervisor::new(root.clone(), host.clone()).unwrap(); let fixture = Self { root, agent, host, - scheme: scheme.to_owned(), + uri: uri.to_owned(), selector: Mutex::new(selector.to_owned()), supervisor, }; @@ -587,7 +656,7 @@ impl ProviderFixture { fn rewrite_selector(&self, selector: &str) { *self.selector.lock() = selector.to_owned(); - write_agent(&self.agent, &self.host, &self.scheme, selector); + write_agent(&self.agent, &self.host, &self.uri, selector); } fn observe(&self, prior: Option) -> ObserveReceipt { @@ -660,24 +729,24 @@ impl ProviderFixture { } } -fn write_agent(agent: &Path, host: &str, scheme: &str, selector: &str) { +fn write_agent(agent: &Path, host: &str, uri: &str, selector: &str) { fs::write( agent.join("agent.kdl"), format!( - "agent \"worker\" {{\n host {host:?}\n command \"true\"\n resource \"observed\" uri=\"{scheme}://subject\" reason=\"Observed state.\" selector=#\"{selector}\"#\n}}\n" + "agent \"worker\" {{\n host {host:?}\n command \"true\"\n resource \"observed\" uri={uri:?} reason=\"Observed state.\" selector=#\"{selector}\"#\n}}\n" ), ) .unwrap(); } -fn observable_resolver_wasm(schema_id: &str, topic: &str, selector: &str) -> Vec { +fn observable_resolver_wasm(schema_id: &str, topics: &[&str], selector: &str) -> Vec { let selector_value: serde_json::Value = serde_json::from_str(selector).unwrap(); let descriptor = serde_json::to_vec(&serde_json::json!({ "abiVersion": 3, "capabilities": ["resolve", "read", "observe"], "selectorSchema": { "type": "object", "additionalProperties": true }, "defaultSelector": selector_value, - "topics": [{"name": topic}, {"name": "ignored"}], + "topics": topics.iter().map(|name| serde_json::json!({"name": name})).chain([serde_json::json!({"name": "ignored"})]).collect::>(), "runtime": {"topology": "shared"}, "snapshot": {"mediaType": "application/json", "schemaId": schema_id} })) diff --git a/tests/resource_provider_e2e.rs b/tests/resource_provider_e2e.rs index 354aa2e3..7441550a 100755 --- a/tests/resource_provider_e2e.rs +++ b/tests/resource_provider_e2e.rs @@ -8,7 +8,8 @@ use std::time::Duration; use serde_json::json; use st2_resource_protocol::{ObservationResult, SnapshotDigest}; use st2_resource_providers::{ - GitHubIssueConfig, GitHubIssueModule, PtyStatsConfig, PtyStatsModule, PtyStatsScope, + GitHubIssueConfig, GitHubIssueModule, GitHubPrConfig, GitHubPrModule, PtyStatsConfig, + PtyStatsModule, PtyStatsScope, VistaConfig, VistaModule, }; use st2_resource_wasip2::{Executor, ObservationRequest, RuntimeConfig}; @@ -106,6 +107,211 @@ fn pty_component_observes_replays_and_enforces_capability_scope() { )); } +#[test] +fn github_pr_component_describes_and_denies_out_of_scope_before_transport() { + let module = GitHubPrModule::new(GitHubPrConfig { + owner: "example".into(), + repo: "demo".into(), + number: 389, + connect_timeout: Duration::from_secs(3), + total_timeout: Duration::from_secs(10), + }) + .unwrap(); + let executor = Executor::new(RuntimeConfig::default(), None, module).unwrap(); + let component_bytes = fs::read(component("ST2_GITHUB_PR_COMPONENT")).unwrap(); + let loaded = executor.load(&component_bytes).unwrap(); + let descriptor = executor.describe(&loaded, None).unwrap(); + assert_eq!( + descriptor.topics, + [ + "ci.failure", + "mergeability.conflict", + "review.requested", + "terminal" + ] + ); + assert_eq!( + descriptor.snapshot_schema_id, + "dev.schickling.github-pr.snapshot.v1" + ); + assert_eq!(descriptor.snapshot_media_type, "application/json"); + + let denied = executor + .observe( + &loaded, + &ObservationRequest { + invocation_id: 1, + uri: "github-pr://other/demo/389".into(), + selector: json!({ + "owner": "other", + "repo": "demo", + "number": 389, + "topics": ["ci.failure"] + }), + prior_digest: None, + demand_watermark: Some(1), + }, + None, + ) + .unwrap(); + assert!(matches!( + &denied, + ObservationResult::Failed { + diagnostic: Some(diagnostic) + } if diagnostic.contains("GitHub pull request scope denied") + )); +} + +#[test] +fn vista_component_observes_replays_and_enforces_identity_before_spawn() { + let temporary = tempfile::tempdir().unwrap(); + let executable = temporary.path().join("vista"); + write_executable( + &executable, + r#"#!/bin/sh +set -eu +if [ "$#" -ne 6 ] || [ "$1" != artifact ] || [ "$2" != get ] || [ "$3" != release-notes ] || [ "$4" != v7 ] || [ "$5" != --output ] || [ "$6" != json ]; then + printf 'unexpected argv: %s\n' "$*" >&2 + exit 64 +fi +printf '%s\n' "$*" >> "$PWD/invocations" +IFS= read -r mode < "$PWD/mode" +case "$mode" in + ready) + printf '%s\n' '{"schemaVersion":1,"uri":"vista://release-notes/v7","slug":"release-notes","version":7,"author":"agent","timestamp":"2026-09-02T10:00:00Z","changeSummary":"created","parent":null,"retired":false,"state":"ready","canonicalUrl":"https://vista.example/release-notes/v7","title":"Release notes","status":{"locked":1,"open":2,"awaiting":3}}' + ;; + changed) + printf '%s\n' '{"schemaVersion":1,"uri":"vista://release-notes/v7","slug":"release-notes","version":7,"author":"agent","timestamp":"2026-09-02T10:00:00Z","changeSummary":"revised","parent":null,"retired":false,"state":"ready","canonicalUrl":"https://vista.example/release-notes/v7","title":"Revised release notes","status":{"locked":1,"open":1,"awaiting":2}}' + ;; + mismatch) + printf '%s\n' '{"schemaVersion":1,"uri":"vista://different/v7","slug":"different","version":7,"author":"agent","timestamp":"2026-09-02T10:00:00Z","changeSummary":"wrong","parent":null,"retired":false,"state":"ready","canonicalUrl":"https://vista.example/different/v7"}' + ;; + unknown-field) + printf '%s\n' '{"schemaVersion":1,"uri":"vista://release-notes/v7","slug":"release-notes","version":7,"author":"agent","timestamp":"2026-09-02T10:00:00Z","changeSummary":"wrong","parent":null,"retired":false,"state":"ready","canonicalUrl":"https://vista.example/release-notes/v7","extra":true}' + ;; + nonzero) + printf 'artifact unavailable for release-notes\n' >&2 + exit 7 + ;; +esac +"#, + ); + fs::write(temporary.path().join("mode"), "ready\n").unwrap(); + + let module = VistaModule::new( + VistaConfig::resolve( + &executable, + temporary.path().to_path_buf(), + "release-notes".into(), + 7, + Duration::from_secs(5), + ) + .unwrap(), + ); + let executor = Executor::new(RuntimeConfig::default(), None, module).unwrap(); + let component_bytes = fs::read(component("ST2_VISTA_COMPONENT")).unwrap(); + let loaded = executor.load(&component_bytes).unwrap(); + let descriptor = executor.describe(&loaded, None).unwrap(); + assert_eq!(descriptor.capabilities.len(), 1); + assert_eq!( + descriptor.topics, + ["ready", "updated", "failed", "expired"] + ); + assert_eq!( + descriptor.snapshot_schema_id, + "dev.schickling.vista.snapshot.v1" + ); + assert_eq!(descriptor.snapshot_media_type, "application/json"); + + let request = |invocation_id, prior_digest| ObservationRequest { + invocation_id, + uri: "vista://release-notes/v7".into(), + selector: json!({ + "slug": "release-notes", + "version": 7, + "topics": ["ready", "updated", "failed", "expired"] + }), + prior_digest, + demand_watermark: Some(invocation_id), + }; + let first = executor.observe(&loaded, &request(1, None), None).unwrap(); + let publication = match first { + ObservationResult::Published { publication } => publication, + other => panic!("first Vista observation must publish, got {other:?}"), + }; + assert_eq!(publication.topics, ["ready"]); + assert_eq!( + publication + .facts + .as_ref() + .unwrap() + .iter() + .map(|fact| fact.key()) + .collect::>(), + ["state"] + ); + let carrier: serde_json::Value = + serde_json::from_slice(publication.bytes.as_slice()).unwrap(); + assert_eq!( + carrier.get("schema").and_then(serde_json::Value::as_str), + Some("dev.schickling.vista.snapshot.v1") + ); + assert!(carrier.get("observedAt").is_none()); + let prior = SnapshotDigest::of(publication.bytes.as_slice()); + + assert_eq!( + executor + .observe(&loaded, &request(2, Some(prior)), None) + .unwrap(), + ObservationResult::Unchanged + ); + + fs::write(temporary.path().join("mode"), "changed\n").unwrap(); + let changed = executor.observe(&loaded, &request(3, Some(prior)), None).unwrap(); + assert!(matches!( + changed, + ObservationResult::Published { publication } + if publication.topics == ["updated", "ready"] + )); + + let invocations_before = fs::read_to_string(temporary.path().join("invocations")).unwrap(); + let malformed = executor + .observe( + &loaded, + &ObservationRequest { + invocation_id: 4, + uri: "vista://bad--slug/v7".into(), + selector: json!({ "slug": "bad--slug", "version": 7 }), + prior_digest: None, + demand_watermark: Some(4), + }, + None, + ) + .unwrap(); + assert!(matches!(malformed, ObservationResult::Failed { .. })); + assert_eq!( + fs::read_to_string(temporary.path().join("invocations")).unwrap(), + invocations_before + ); + + for (mode, expected) in [ + ("mismatch", "different artifact identity"), + ("unknown-field", "invalid manifest"), + ("nonzero", "artifact unavailable for release-notes"), + ] { + fs::write(temporary.path().join("mode"), format!("{mode}\n")).unwrap(); + let result = executor + .observe(&loaded, &request(5, None), None) + .unwrap(); + assert!(matches!( + &result, + ObservationResult::Failed { + diagnostic: Some(diagnostic) + } if diagnostic.contains(expected) + )); + } +} + #[test] #[ignore = "explicit read-only public GitHub smoke; requires network and ST2_GITHUB_ISSUE_COMPONENT"] fn github_component_public_read_only_smoke() { diff --git a/wit/github-pr/deps/st2-github-pr/github-pr.wit b/wit/github-pr/deps/st2-github-pr/github-pr.wit new file mode 100644 index 00000000..b9f2fa26 --- /dev/null +++ b/wit/github-pr/deps/st2-github-pr/github-pr.wit @@ -0,0 +1,41 @@ +package compoundingtech:st2-github-pr@0.1.0; + +interface github-pr { + record pull-request-request { + owner: string, + repo: string, + number: u64, + } + + record source-object { + etag: option, + body: list, + } + + record source-snapshot { + pull-request: source-object, + check-runs: source-object, + combined-status: source-object, + observed-at: string, + } + + record source-observation { + current: source-snapshot, + previous: option, + } + + variant pull-request-response { + ok(source-observation), + not-modified, + } + + variant pull-request-error { + denied, + unavailable, + resource-exhausted, + deadline-exceeded, + } + + get: func(request: pull-request-request) -> result; + bind-snapshot: func(digest: list) -> result<_, pull-request-error>; +} diff --git a/wit/github-pr/provider-api.wit b/wit/github-pr/provider-api.wit new file mode 100644 index 00000000..ecf8fbe0 --- /dev/null +++ b/wit/github-pr/provider-api.wit @@ -0,0 +1,57 @@ +package st2:resource-provider@0.1.0; + +interface provider-api { + enum scheduling-capability { + demand, + } + + record provider-descriptor { + capabilities: list, + selector-schema-json: string, + default-selector-json: string, + topics: list, + snapshot-media-type: string, + snapshot-schema-id: string, + } + + variant descriptor-error { + invalid-descriptor(string), + unavailable(string), + } + + record observe-request { + uri: string, + selector-json: string, + prior-digest: option>, + demand-watermark: option, + } + + variant fact-value { + omitted, + null, + value(string), + } + + record fact { + key: string, + before: fact-value, + after: fact-value, + } + + record publication { + schema-id: string, + media-type: string, + bytes: list, + topics: list, + facts: option>, + } + + variant observation-result { + unchanged, + failed(option), + published(publication), + } + + describe: func() -> result; + observe: func(request: observe-request) -> observation-result; +} diff --git a/wit/github-pr/provider.wit b/wit/github-pr/provider.wit new file mode 100644 index 00000000..ba309fcb --- /dev/null +++ b/wit/github-pr/provider.wit @@ -0,0 +1,6 @@ +package st2:resource-provider@0.1.0; + +world github-pr-provider { + import compoundingtech:st2-github-pr/github-pr@0.1.0; + export provider-api; +} diff --git a/wit/vista/deps/st2-vista/vista.wit b/wit/vista/deps/st2-vista/vista.wit new file mode 100644 index 00000000..df334dc7 --- /dev/null +++ b/wit/vista/deps/st2-vista/vista.wit @@ -0,0 +1,31 @@ +package compoundingtech:st2-vista@0.1.0; + +interface vista { + record artifact-request { + slug: string, + version: u64, + } + + variant exit-status { + code(s32), + signal(s32), + } + + record outcome { + stdout: list, + stderr: list, + stdout-truncated: bool, + stderr-truncated: bool, + exit: exit-status, + } + + variant vista-error { + denied, + unavailable, + resource-exhausted, + deadline-exceeded, + cancelled, + } + + get: func(request: artifact-request) -> result; +} diff --git a/wit/vista/provider-api.wit b/wit/vista/provider-api.wit new file mode 100644 index 00000000..ecf8fbe0 --- /dev/null +++ b/wit/vista/provider-api.wit @@ -0,0 +1,57 @@ +package st2:resource-provider@0.1.0; + +interface provider-api { + enum scheduling-capability { + demand, + } + + record provider-descriptor { + capabilities: list, + selector-schema-json: string, + default-selector-json: string, + topics: list, + snapshot-media-type: string, + snapshot-schema-id: string, + } + + variant descriptor-error { + invalid-descriptor(string), + unavailable(string), + } + + record observe-request { + uri: string, + selector-json: string, + prior-digest: option>, + demand-watermark: option, + } + + variant fact-value { + omitted, + null, + value(string), + } + + record fact { + key: string, + before: fact-value, + after: fact-value, + } + + record publication { + schema-id: string, + media-type: string, + bytes: list, + topics: list, + facts: option>, + } + + variant observation-result { + unchanged, + failed(option), + published(publication), + } + + describe: func() -> result; + observe: func(request: observe-request) -> observation-result; +} diff --git a/wit/vista/provider.wit b/wit/vista/provider.wit new file mode 100644 index 00000000..5f58811e --- /dev/null +++ b/wit/vista/provider.wit @@ -0,0 +1,6 @@ +package st2:resource-provider@0.1.0; + +world vista-provider { + import compoundingtech:st2-vista/vista@0.1.0; + export provider-api; +}