From 363bb41cf15fff4ae08a130199901b338ed2094b Mon Sep 17 00:00:00 2001 From: Rohit Ghumare Date: Thu, 23 Jul 2026 13:24:26 +0100 Subject: [PATCH 1/2] (MOT-4187) feat(iii-directory): version-aware skills reconcile Skills for a namespace were downloaded once and then frozen: the boot reconcile skips any namespace with a completion marker, so skills drift behind the worker they describe as installed versions move forward. - Completion marker records the resolved semver when the download was pinned to a concrete version (tolerant additive field; legacy markers parse as version-absent and converge on their first refresh pass). - reconcile_decision compares the marker version against the installed worker version and re-downloads on mismatch, behind a new opt-in auto_refresh config field (default false keeps existing behavior). - The worker-event subscription covers update as well as add, pinned to the version on the event when present, with a single worker::list lookup as fallback. - Re-downloads keep existing overwrite semantics: file-by-file, hand added sibling files survive. --- iii-directory/README.md | 25 +++- iii-directory/src/config.rs | 14 +++ iii-directory/src/functions/download.rs | 151 +++++++++++++++++++++--- iii-directory/src/main.rs | 115 +++++++++++++----- 4 files changed, 259 insertions(+), 46 deletions(-) diff --git a/iii-directory/README.md b/iii-directory/README.md index 8f11c41e3..226b1c0ce 100644 --- a/iii-directory/README.md +++ b/iii-directory/README.md @@ -94,17 +94,40 @@ or use the console Workers tab — all three propagate without a redeploy. # TOPOLOGY — changing any of these requires a worker restart. skills_folder: ~/.iii/skills # read/write root for skills + prompts local_skills_folder: ./.iii/skills # project-scoped overrides (whole-namespace local-wins) -auto_download: true # subscribe to worker-add + run the boot reconcile +auto_download: true # subscribe to worker add/update + run the boot reconcile # TUNABLE — hot-reload live on `configuration:updated`. registry_url: https://api.workers.iii.dev # workers registry base URL download_timeout_ms: 60000 # per git-clone / HTTP request timeout (ms) registry_cache_ttl_ms: 60000 # in-process TTL for registry::workers::* responses filter_unregistered: true # hide skills whose namespace isn't an installed worker +auto_refresh: false # boot reconcile also re-downloads namespaces whose marker + # records a different version than the installed worker ``` The `skills_folder` is created on first download if it doesn't exist. +### Keeping skills current + +Without `auto_refresh`, a namespace is downloaded once (on worker add or the +first boot reconcile) and then left alone: the completion marker makes every +later boot reconcile skip it, so skills can drift behind the worker they +describe as `iii worker update` moves the installed version forward. + +Two mechanisms close that gap: + +- The worker-event subscription now covers `update` as well as `add`, so + `iii worker update ` re-downloads that worker's skills immediately, + pinned to the newly installed version. +- With `auto_refresh: true`, the boot reconcile compares the version recorded + in each namespace's completion marker against the installed worker version + from `worker::list` and re-downloads on mismatch. Markers written by older + releases (or by tag downloads) carry no version and are converged to a + versioned marker on their first refresh pass. + +Re-downloads keep the existing overwrite semantics: files are replaced +file-by-file and hand-added sibling files in the namespace survive. + ### Zero-config default + seed With no seed and no stored value the worker uses built-in defaults diff --git a/iii-directory/src/config.rs b/iii-directory/src/config.rs index 42abedd4d..9e10a82a5 100644 --- a/iii-directory/src/config.rs +++ b/iii-directory/src/config.rs @@ -61,6 +61,10 @@ fn default_auto_download() -> bool { true } +fn default_auto_refresh() -> bool { + false +} + #[derive(Deserialize, Serialize, Debug, Clone, JsonSchema)] pub struct SkillsConfig { /// Folder that backs every read (`directory::skills::list`, @@ -113,6 +117,15 @@ pub struct SkillsConfig { /// folder. #[serde(default = "default_auto_download")] pub auto_download: bool, + + /// When `true`, the boot reconcile also re-downloads a namespace whose + /// completion marker records a different version than the one currently + /// installed, so skills follow `iii worker update` across restarts. + /// When `false` (default), a namespace with a completion marker is + /// never re-downloaded at boot (the pre-existing behavior). Requires + /// `auto_download: true` to have any effect. + #[serde(default = "default_auto_refresh")] + pub auto_refresh: bool, } impl Default for SkillsConfig { @@ -125,6 +138,7 @@ impl Default for SkillsConfig { registry_cache_ttl_ms: default_registry_cache_ttl_ms(), filter_unregistered: default_filter_unregistered(), auto_download: default_auto_download(), + auto_refresh: default_auto_refresh(), } } } diff --git a/iii-directory/src/functions/download.rs b/iii-directory/src/functions/download.rs index 57d3770bf..092dc52a4 100644 --- a/iii-directory/src/functions/download.rs +++ b/iii-directory/src/functions/download.rs @@ -401,12 +401,20 @@ async fn fan_out( /// Marker filename written inside a namespace after a complete download. const COMPLETION_MARKER: &str = ".iii-skill-complete"; -/// Marker payload shape: `{ worker, source, tag_or_version, schema }`. +/// Marker payload shape: `{ worker, source, tag_or_version, version?, schema }`. #[derive(Debug, serde::Serialize, serde::Deserialize)] struct CompletionMarker { worker: String, source: String, tag_or_version: String, + /// Resolved semver the namespace was downloaded at, when the download + /// was pinned to a concrete version. `None` for tag downloads (e.g. + /// `latest`) and for markers written before this field existed; both + /// deserialize tolerantly. Invariant: when `Some`, it equals + /// `tag_or_version` (kept separate because `tag_or_version` is frozen + /// schema and a tag can be semver-shaped). + #[serde(default, skip_serializing_if = "Option::is_none")] + version: Option, schema: u32, } @@ -423,6 +431,10 @@ fn write_completion_marker( VersionSpec::Version(v) => v.clone(), VersionSpec::Tag(t) => t.clone(), }, + version: match spec { + VersionSpec::Version(v) => Some(v.clone()), + VersionSpec::Tag(_) => None, + }, schema: 1, }; let json = serde_json::to_string_pretty(&marker).map_err(|e| format!("encode marker: {e}"))?; @@ -576,7 +588,12 @@ use std::path::Path; /// Skip guards (in order): /// 1. Name doesn't validate → skip. /// 2. Local override directory exists → skip. -/// 3. Completion marker already present in global root → skip. +/// 3. Completion marker already present in global root → skip, UNLESS +/// `auto_refresh` is enabled AND the installed `version` is a +/// concrete semver that differs from the version recorded in the +/// marker. A marker without a recorded version (tag download, or +/// written before the field existed) counts as differing, so one +/// refresh pass converges it to a versioned marker. /// /// When not skipped, `version` from the worker info determines the /// spec: `Some(v)` (non-empty) → `VersionSpec::Version(v)`, else @@ -586,6 +603,7 @@ pub fn reconcile_decision( version: Option<&str>, local_root: &Path, global_root: &Path, + auto_refresh: bool, ) -> Option { // Guard 1: invalid name. if crate::sources::registry::validate_worker_name(name).is_err() { @@ -595,14 +613,32 @@ pub fn reconcile_decision( if local_root.join(name).is_dir() { return None; } - // Guard 3: completion marker already present. - if has_completion_marker(global_root, name) { - return None; - } - // Determine version spec. - match version { - Some(v) if !v.is_empty() => Some(VersionSpec::Version(v.to_string())), - _ => Some(VersionSpec::Tag("latest".to_string())), + let installed = version.filter(|v| !v.is_empty()); + // Guard 3: completion marker. One read serves both the presence check + // and the recorded version; any read failure other than NotFound still + // counts as marker-present. + match std::fs::read_to_string(global_root.join(name).join(COMPLETION_MARKER)) { + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(match installed { + Some(v) => VersionSpec::Version(v.to_string()), + None => VersionSpec::Tag("latest".to_string()), + }), + marker => { + if !auto_refresh { + return None; + } + // Version-aware refresh: only a concrete installed version can + // signal drift; without one there is nothing to compare against. + // A marker whose version can't be read (tag download, older + // release, or parse failure) counts as differing, so one refresh + // pass converges it to a versioned marker. + let installed = installed?; + let recorded = marker + .ok() + .and_then(|raw| serde_json::from_str::(&raw).ok()) + .and_then(|m| m.version); + (recorded.as_deref() != Some(installed)) + .then(|| VersionSpec::Version(installed.to_string())) + } } } @@ -999,7 +1035,7 @@ mod tests { #[test] fn reconcile_skips_invalid_name() { let tmp = tempfile::tempdir().unwrap(); - let result = reconcile_decision("INVALID", None, tmp.path(), tmp.path()); + let result = reconcile_decision("INVALID", None, tmp.path(), tmp.path(), false); assert!(result.is_none(), "invalid name should be skipped"); } @@ -1011,7 +1047,7 @@ mod tests { // Create a local override directory. std::fs::create_dir_all(local_root.join("resend")).unwrap(); std::fs::create_dir_all(&global_root).unwrap(); - let result = reconcile_decision("resend", None, &local_root, &global_root); + let result = reconcile_decision("resend", None, &local_root, &global_root, false); assert!(result.is_none(), "local override should skip download"); } @@ -1025,14 +1061,14 @@ mod tests { // Write a completion marker. write_completion_marker(&global_root, "resend", &VersionSpec::Tag("latest".into())) .unwrap(); - let result = reconcile_decision("resend", None, &local_root, &global_root); + let result = reconcile_decision("resend", None, &local_root, &global_root, false); assert!(result.is_none(), "existing marker should skip download"); } #[test] fn reconcile_returns_version_spec_when_version_present() { let tmp = tempfile::tempdir().unwrap(); - let result = reconcile_decision("resend", Some("2.0.0"), tmp.path(), tmp.path()); + let result = reconcile_decision("resend", Some("2.0.0"), tmp.path(), tmp.path(), false); assert_eq!( result, Some(VersionSpec::Version("2.0.0".to_string())), @@ -1043,7 +1079,7 @@ mod tests { #[test] fn reconcile_returns_latest_tag_when_no_version() { let tmp = tempfile::tempdir().unwrap(); - let result = reconcile_decision("resend", None, tmp.path(), tmp.path()); + let result = reconcile_decision("resend", None, tmp.path(), tmp.path(), false); assert_eq!( result, Some(VersionSpec::Tag("latest".to_string())), @@ -1054,11 +1090,94 @@ mod tests { #[test] fn reconcile_returns_latest_tag_when_empty_version() { let tmp = tempfile::tempdir().unwrap(); - let result = reconcile_decision("resend", Some(""), tmp.path(), tmp.path()); + let result = reconcile_decision("resend", Some(""), tmp.path(), tmp.path(), false); assert_eq!( result, Some(VersionSpec::Tag("latest".to_string())), "empty version string should fall back to latest" ); } + + // ── reconcile_decision with auto_refresh ────────────────────────── + + /// Roots with a completion marker for "resend" already written under + /// the global root (`write_completion_marker` creates the parents). + fn roots_with_marker( + tmp: &tempfile::TempDir, + spec: &VersionSpec, + ) -> (std::path::PathBuf, std::path::PathBuf) { + let global_root = tmp.path().join("global"); + write_completion_marker(&global_root, "resend", spec).unwrap(); + (tmp.path().join("local"), global_root) + } + + #[test] + fn refresh_redownloads_on_version_drift() { + let tmp = tempfile::tempdir().unwrap(); + let (local, global) = roots_with_marker(&tmp, &VersionSpec::Version("1.0.0".into())); + assert_eq!( + reconcile_decision("resend", Some("1.2.0"), &local, &global, true), + Some(VersionSpec::Version("1.2.0".to_string())), + "marker at 1.0.0 with 1.2.0 installed should re-download" + ); + } + + #[test] + fn refresh_skips_when_versions_match() { + let tmp = tempfile::tempdir().unwrap(); + let (local, global) = roots_with_marker(&tmp, &VersionSpec::Version("1.2.0".into())); + assert!( + reconcile_decision("resend", Some("1.2.0"), &local, &global, true).is_none(), + "matching versions should skip" + ); + } + + #[test] + fn refresh_converges_tag_marker_to_version() { + let tmp = tempfile::tempdir().unwrap(); + let (local, global) = roots_with_marker(&tmp, &VersionSpec::Tag("latest".into())); + assert_eq!( + reconcile_decision("resend", Some("1.2.0"), &local, &global, true), + Some(VersionSpec::Version("1.2.0".to_string())), + "tag marker has no recorded version; a concrete installed version should re-download" + ); + } + + #[test] + fn refresh_skips_without_installed_version() { + let tmp = tempfile::tempdir().unwrap(); + let (local, global) = roots_with_marker(&tmp, &VersionSpec::Tag("latest".into())); + assert!( + reconcile_decision("resend", None, &local, &global, true).is_none(), + "no installed version means no drift signal; must not re-download" + ); + } + + #[test] + fn refresh_disabled_preserves_marker_skip() { + let tmp = tempfile::tempdir().unwrap(); + let (local, global) = roots_with_marker(&tmp, &VersionSpec::Version("1.0.0".into())); + assert!( + reconcile_decision("resend", Some("1.2.0"), &local, &global, false).is_none(), + "auto_refresh=false must keep the marker-skip behavior even on drift" + ); + } + + #[test] + fn refresh_converges_legacy_marker_without_version_field() { + let tmp = tempfile::tempdir().unwrap(); + let global = tmp.path().join("global"); + let dest = global.join("resend").join(COMPLETION_MARKER); + std::fs::create_dir_all(dest.parent().unwrap()).unwrap(); + std::fs::write( + &dest, + r#"{"worker":"resend","source":"registry","tag_or_version":"latest","schema":1}"#, + ) + .unwrap(); + assert_eq!( + reconcile_decision("resend", Some("1.2.0"), tmp.path(), &global, true), + Some(VersionSpec::Version("1.2.0".to_string())), + "a marker written before the version field existed converges on refresh" + ); + } } diff --git a/iii-directory/src/main.rs b/iii-directory/src/main.rs index cc9c8561d..303b40b5f 100644 --- a/iii-directory/src/main.rs +++ b/iii-directory/src/main.rs @@ -198,12 +198,17 @@ async fn main() -> Result<()> { Ok(()) } -/// `worker` trigger payload for `directory::__on_worker_added`. Only `worker` -/// is read; declared as a struct so the function publishes a typed schema. +/// `worker` trigger payload for `directory::__on_worker_added`. Declared as a +/// struct so the function publishes a typed schema. #[derive(Debug, Default, serde::Deserialize, schemars::JsonSchema)] struct WorkerAddedEvent { #[serde(default)] worker: Option, + /// Installed version, when the event carries it (the wire schema + /// reserves the field on terminal add/update stages). Preferred over a + /// `worker::list` lookup when present. + #[serde(default)] + version: Option, } #[derive(Debug, serde::Serialize, schemars::JsonSchema)] @@ -212,7 +217,8 @@ struct WorkerAddedAck { } /// Register the internal `directory::__on_worker_added` handler and -/// subscribe to the `worker` trigger type for `add` operations. +/// subscribe to the `worker` trigger type for `add` and `update` +/// operations. fn setup_auto_download( iii: &Arc, cfg: &SharedConfig, @@ -222,16 +228,18 @@ fn setup_auto_download( let cfg_inner = cfg.clone(); let cache_inner = cache.clone(); let in_flight_inner = in_flight.clone(); + let iii_inner = iii.clone(); - // Register the internal handler that fires on worker-add events. + // Register the internal handler that fires on worker add/update events. iii.register_function( "directory::__on_worker_added", RegisterFunction::new_async(move |event: WorkerAddedEvent| { let cfg = cfg_inner.load_full(); let cache = cache_inner.clone(); let in_flight = in_flight_inner.clone(); + let iii = iii_inner.clone(); async move { - handle_worker_added(&cfg, &cache, &in_flight, &event).await; + handle_worker_added(&iii, &cfg, &cache, &in_flight, &event).await; Ok::<_, Error>(WorkerAddedAck { ok: true }) } }) @@ -247,7 +255,7 @@ fn setup_auto_download( trigger_type: "worker".to_string(), function_id: "directory::__on_worker_added".to_string(), config: json!({ - "operations": ["add"], + "operations": ["add", "update"], "stages": ["done"] }), metadata: None, @@ -274,9 +282,14 @@ fn setup_auto_download( }); } -/// Handle a `worker` trigger add event. Downloads skills for the -/// newly added worker if not already in-flight. +/// Handle a `worker` trigger add/update event. Downloads skills for the +/// worker if not already in-flight. The spec is pinned to the installed +/// version when the event or `worker::list` reports one, so the +/// completion marker records the concrete semver the skills belong to; +/// otherwise it falls back to the `latest` tag (the pre-existing +/// behavior). async fn handle_worker_added( + iii: &IIIClient, cfg: &SkillsConfig, cache: &RegisteredWorkersCache, in_flight: &Arc, @@ -296,7 +309,15 @@ async fn handle_worker_added( return; }; - let spec = VersionSpec::Tag("latest".to_string()); + // Prefer the version on the event; fall back to one worker::list call. + let installed = match event.version.as_deref().filter(|v| !v.is_empty()) { + Some(v) => Some(v.to_string()), + None => installed_worker_version(iii, &worker).await, + }; + let spec = match installed { + Some(v) => VersionSpec::Version(v), + None => VersionSpec::Tag("latest".to_string()), + }; match download_worker_skills(cfg, &worker, &spec).await { Ok(true) => { tracing::info!(worker = %worker, "auto-download complete on worker add"); @@ -340,6 +361,42 @@ async fn reconcile_one( } } +/// One `worker::list` call, parsed to the worker row array. The single +/// place that encodes the request shape and response parse. +async fn worker_list_once(iii: &IIIClient) -> Result, Error> { + let val = iii + .trigger(TriggerRequest { + function_id: "worker::list".to_string(), + payload: json!({}), + action: None, + timeout_ms: Some(10_000), + }) + .await?; + Ok(val + .get("workers") + .and_then(|w| w.as_array()) + .cloned() + .unwrap_or_default()) +} + +/// Extract a worker row's non-empty `version` string. +fn worker_version(row: &serde_json::Value) -> Option<&str> { + row.get("version")?.as_str().filter(|v| !v.is_empty()) +} + +/// Look up one worker's installed version via a single `worker::list` +/// call. Returns `None` when the list is unavailable or the worker is +/// absent / reports no version; callers fall back to the `latest` tag. +async fn installed_worker_version(iii: &IIIClient, worker: &str) -> Option { + worker_list_once(iii) + .await + .ok()? + .iter() + .find(|w| w.get("name").and_then(|n| n.as_str()) == Some(worker)) + .and_then(worker_version) + .map(str::to_string) +} + /// Fetch the installed-worker list, retrying with backoff while the /// engine's worker-manager — which registers `worker::list` — is still /// coming up. On a cold engine start the worker-manager registers late, @@ -351,24 +408,10 @@ async fn reconcile_one( async fn fetch_worker_list_with_retry(iii: &IIIClient) -> Option> { const MAX_ATTEMPTS: u32 = 6; for attempt in 1..=MAX_ATTEMPTS { - let result = iii - .trigger(TriggerRequest { - function_id: "worker::list".to_string(), - payload: json!({}), - action: None, - timeout_ms: Some(10_000), - }) - .await; + let result = worker_list_once(iii).await; match result { - Ok(val) => { - return Some( - val.get("workers") - .and_then(|w| w.as_array()) - .cloned() - .unwrap_or_default(), - ); - } + Ok(workers) => return Some(workers), Err(e) if attempt == MAX_ATTEMPTS => { tracing::warn!( attempt, @@ -394,7 +437,9 @@ async fn fetch_worker_list_with_retry(iii: &IIIClient) -> Option, cfg: SharedConfig, @@ -419,7 +464,13 @@ fn spawn_boot_reconcile( // it directly (registry pull), independent of — and before — the // worker list, so it lands even when `worker::list` isn't ready // yet on a cold start. - if let Some(spec) = reconcile_decision(ENGINE_NAMESPACE, None, &local_root, &global_root) { + if let Some(spec) = reconcile_decision( + ENGINE_NAMESPACE, + None, + &local_root, + &global_root, + cfg.auto_refresh, + ) { if reconcile_one(&cfg, &in_flight, ENGINE_NAMESPACE, &spec).await { reconciled += 1; } @@ -449,8 +500,14 @@ fn spawn_boot_reconcile( None => continue, }; - let version = w.get("version").and_then(|v| v.as_str()); - let spec = match reconcile_decision(name, version, &local_root, &global_root) { + let version = worker_version(w); + let spec = match reconcile_decision( + name, + version, + &local_root, + &global_root, + cfg.auto_refresh, + ) { Some(s) => s, None => continue, }; From f28b92f369cd9160e4fa43ec5c557074071eb0d8 Mon Sep 17 00:00:00 2001 From: Rohit Ghumare Date: Fri, 24 Jul 2026 11:46:39 +0100 Subject: [PATCH 2/2] docs(iii-directory): clarify update events refresh skills regardless of auto_refresh --- iii-directory/README.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/iii-directory/README.md b/iii-directory/README.md index 226b1c0ce..76373d9ef 100644 --- a/iii-directory/README.md +++ b/iii-directory/README.md @@ -110,9 +110,11 @@ The `skills_folder` is created on first download if it doesn't exist. ### Keeping skills current Without `auto_refresh`, a namespace is downloaded once (on worker add or the -first boot reconcile) and then left alone: the completion marker makes every -later boot reconcile skip it, so skills can drift behind the worker they -describe as `iii worker update` moves the installed version forward. +first boot reconcile) and never re-fetched at boot: the completion marker makes +every later boot reconcile skip it. Worker update events still refresh that +worker's skills immediately (below); `auto_refresh` covers the drift those +events cannot see, such as updates applied while the directory was down or +markers written before versions were recorded. Two mechanisms close that gap: