(MOT-3748) fix(harness): preserve registrations across engine reloads

.github/workflows/harness-quickstart.yml

@@ -1,6 +1,18 @@
 name: Harness quickstart
 
 on:
+  workflow_call:
+    inputs:
+      channel:
+        description: Installer channel to validate
+        required: false
+        type: string
+        default: latest
+    secrets:
+      SLACK_BOT_TOKEN:
+        required: false
+      ZAI_API_KEY:
+        required: true
   schedule:
     - cron: "17 5 * * *"
   workflow_dispatch:

.github/workflows/release.yml

@@ -91,6 +91,7 @@ jobs:
       experimental: ${{ steps.meta.outputs.experimental }}
       web_bundle: ${{ steps.web.outputs.needed }}
       interface_smoke: ${{ steps.smoke.outputs.interface_smoke }}
+      harness_smoke: ${{ steps.harness_smoke.outputs.enabled }}
     steps:
       - uses: actions/checkout@v5
         with:
@@ -150,6 +151,29 @@ jobs:
             echo "::notice::$WORKER opts out of interface collection; registry publish will be skipped"
           fi
 
+      # The published Harness quickstart also covers every worker declared as
+      # a mandatory Harness dependency. Keep this derived from the manifest so
+      # the post-deploy smoke follows dependency changes automatically.
+      - name: Detect Harness quickstart smoke target
+        id: harness_smoke
+        env:
+          WORKER: ${{ steps.meta.outputs.worker }}
+        run: |
+          set -euo pipefail
+          enabled=$(WORKER="$WORKER" python3 - <<'PY'
+          import os
+          from pathlib import Path
+          import yaml
+
+          worker = os.environ["WORKER"]
+          manifest = yaml.safe_load(Path("harness/iii.worker.yaml").read_text())
+          dependencies = manifest.get("dependencies", {}) or {}
+          print("true" if worker == "harness" or worker in dependencies else "false")
+          PY
+          )
+          echo "enabled=$enabled" >> "$GITHUB_OUTPUT"
+          echo "::notice::Harness quickstart smoke target=$enabled worker=$WORKER"
+
   # ──────────────────────────────────────────────────────────────
   # Create the GitHub Release shell (skipped on dry runs).
   # ──────────────────────────────────────────────────────────────
@@ -241,6 +265,26 @@ jobs:
       experimental: ${{ needs.setup.outputs.experimental }}
     secrets: inherit
 
+  # Run the published Harness quickstart immediately after the registry
+  # deployment of Harness or one of its mandatory dependencies. The validator
+  # supports the stable release channels only.
+  quickstart:
+    name: Harness quickstart
+    needs: [setup, publish]
+    if: >-
+      ${{
+        !failure() &&
+        !cancelled() &&
+        needs.setup.outputs.harness_smoke == 'true' &&
+        needs.setup.outputs.dry_run != 'true' &&
+        needs.publish.result == 'success' &&
+        (needs.setup.outputs.registry_tag == 'latest' || needs.setup.outputs.registry_tag == 'next')
+      }}
+    uses: ./.github/workflows/harness-quickstart.yml
+    with:
+      channel: ${{ needs.setup.outputs.registry_tag }}
+    secrets: inherit
+
   # ──────────────────────────────────────────────────────────────
   # Announce the release in Slack (#worker-releases). Terminal job:
   # nothing depends on it, but a bad token or channel fails red

harness/Cargo.toml

@@ -4,7 +4,7 @@ resolver = "2"
 
 [package]
 name = "harness"
-version = "1.6.4"
+version = "1.6.7"
 edition = "2021"
 publish = false
 
@@ -17,12 +17,14 @@ name = "harness"
 path = "src/lib.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+# 0.21.8 adds the reconnect reattach handshake. Older SDKs can lose this
+# worker's registrations when the engine reloads another worker.
+iii-sdk = "=0.21.8"
 # Re-exports the same `opentelemetry` the SDK uses, so we can carry the active
 # OTel context across `tokio::spawn` boundaries (shared `Context` global).
 # 0.21.5 adds the live span-start push (`LiveSpanStartProcessor`), so the
 # console renders `harness::turn step` while it runs instead of on close.
-iii-helpers = "=0.21.6"
+iii-helpers = "=0.21.8"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

harness/iii.worker.yaml

@@ -9,7 +9,7 @@ description: Thin durable turn loop that wires session-manager, context-manager,
 
 dependencies:
   state: "^0.21.3"
-  queue: "^0.2.0"
+  queue: "^0.21.2"
   cron: "^0.21.0"
   configuration: "^0.21.6"
   iii-observability: "^0.21.6"

harness/src/queue.rs

@@ -23,7 +23,8 @@ const DEFINE_RETRY_BACKOFF_MS: u64 = 250;
 /// Exhaustion is fatal to harness startup: accepting sends without this queue
 /// would acknowledge work that cannot run durably.
 pub async fn ensure_turn_queue(iii: &IIIClient) -> Result<(), String> {
-    let payload = turn_queue_definition();
+    let mut payload = turn_queue_definition();
+    let mut restart_redelivery_requested = true;
     let mut last_error = String::new();
 
     for attempt in 1..=DEFINE_ATTEMPTS {
@@ -42,6 +43,16 @@ pub async fn ensure_turn_queue(iii: &IIIClient) -> Result<(), String> {
             }
             Err(error) => {
                 last_error = error.to_string();
+                if restart_redelivery_requested && is_legacy_queue_schema_error(&last_error) {
+                    tracing::warn!(
+                        queue = TURN_QUEUE,
+                        error = %last_error,
+                        "queue worker does not support restart redelivery; retrying with the legacy queue schema"
+                    );
+                    payload = legacy_turn_queue_definition();
+                    restart_redelivery_requested = false;
+                    continue;
+                }
                 if attempt < DEFINE_ATTEMPTS {
                     tracing::warn!(
                         queue = TURN_QUEUE,
@@ -60,6 +71,10 @@ pub async fn ensure_turn_queue(iii: &IIIClient) -> Result<(), String> {
     ))
 }
 
+fn is_legacy_queue_schema_error(error: &str) -> bool {
+    error.contains("unknown field `redeliver_on_engine_restart`")
+}
+
 fn turn_queue_definition() -> Value {
     json!({
         "queue": TURN_QUEUE,
@@ -75,6 +90,20 @@ fn turn_queue_definition() -> Value {
     })
 }
 
+fn legacy_turn_queue_definition() -> Value {
+    json!({
+        "queue": TURN_QUEUE,
+        "config": {
+            "type": "fifo",
+            "message_group_field": "session_id",
+            "concurrency": 10,
+            "max_retries": 3,
+            "backoff_ms": 1_000,
+            "poll_interval_ms": 100
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -100,4 +129,35 @@ mod tests {
             .get("timeout_ms")
             .is_none());
     }
+
+    #[test]
+    fn legacy_turn_queue_definition_omits_restart_redelivery() {
+        assert_eq!(
+            legacy_turn_queue_definition(),
+            json!({
+                "queue": "harness-turn",
+                "config": {
+                    "type": "fifo",
+                    "message_group_field": "session_id",
+                    "concurrency": 10,
+                    "max_retries": 3,
+                    "backoff_ms": 1_000,
+                    "poll_interval_ms": 100
+                }
+            })
+        );
+        assert!(legacy_turn_queue_definition()["config"]
+            .get("redeliver_on_engine_restart")
+            .is_none());
+    }
+
+    #[test]
+    fn only_the_unsupported_restart_redelivery_field_triggers_fallback() {
+        assert!(is_legacy_queue_schema_error(
+            "serialization error: unknown field `redeliver_on_engine_restart`"
+        ));
+        assert!(!is_legacy_queue_schema_error(
+            "serialization error: unknown field `message_group_field`"
+        ));
+    }
 }

harness/tests/e2e/Cargo.toml

@@ -6,7 +6,7 @@ publish = false
 
 [dependencies]
 harness = { path = "../.." }
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.21.8"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

harness/tests/integration/Cargo.toml

@@ -13,7 +13,7 @@ name = "harness_integration"
 path = "src/lib.rs"
 
 [dependencies]
-iii-sdk = "=0.21.6"
+iii-sdk = "=0.21.8"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time", "signal", "net", "io-util"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

harness/tests/manifest.rs

@@ -54,7 +54,7 @@ fn worker_manifest_uses_the_standalone_queue_worker() {
 
     assert_eq!(
         dependencies.get(serde_yaml::Value::String("queue".into())),
-        Some(&serde_yaml::Value::String("^0.2.0".into()))
+        Some(&serde_yaml::Value::String("^0.21.2".into()))
     );
     assert!(!dependencies.contains_key(serde_yaml::Value::String("iii-queue".into())));
 }