From ebdcb25453273d04e0b64853c38fed7295bf260b Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 16 Jul 2026 18:14:50 -0300 Subject: [PATCH 1/4] (MOT-4073) refactor(web): one-shot harness hook bind via engine recoverable triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The engine now parks a register_trigger for an unregistered trigger type as a pending intent and activates it when the type registers, re-parking and re-activating across harness restarts (recoverable triggers, iii #1962). Delete the web::on-registry-changed watcher, its two engine::*-available event bindings, the engine::triggers::list warm-start probe, the AtomicBool bind claim, and the explicit registry-changed wire schemas; bind the pre-generate guidance hook once at startup. Requires engine >= 0.21.8 — noted in the README (older engines silently drop the bind). --- web/README.md | 8 ++ web/src/configuration.rs | 213 ++------------------------------------- web/src/main.rs | 9 +- 3 files changed, 22 insertions(+), 208 deletions(-) diff --git a/web/README.md b/web/README.md index 5d431187f..917b43028 100644 --- a/web/README.md +++ b/web/README.md @@ -11,6 +11,14 @@ The worker is a faithful Rust port of the original TypeScript `web::fetch` implementation — same request fields and success/error/image envelopes, so existing callers and the harness consumer are unaffected. +While connected it also injects a usage section into the agent system prompt +via the harness `pre-generate` hook (`web::inject-guidance`), so the guidance +is presence-gated: no web worker, no prompt text. The binding is one-shot at +startup and relies on the engine's recoverable triggers (iii #1962, engine ≥ +0.21.8): bound before the harness is up, it parks as a pending intent and +activates when the harness registers the trigger type. On older engines the +bind is silently dropped. + ## Table of contents 1. [Install](#install) diff --git a/web/src/configuration.rs b/web/src/configuration.rs index 9ae44c846..338d7d811 100644 --- a/web/src/configuration.rs +++ b/web/src/configuration.rs @@ -3,8 +3,6 @@ //! `configuration` trigger so `configuration:updated` re-fetches and applies //! the change. All WebConfig fields hot-reload (no topology partition). -use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; use std::time::Duration; use iii_sdk::errors::Error; @@ -21,17 +19,6 @@ const CONFIG_FN_ID: &str = "web::on-config-change"; const CONFIG_TIMEOUT_MS: u64 = 5_000; const CONFIG_RETRIES: u32 = 3; -const ON_REGISTRY_CHANGED_FN_ID: &str = "web::on-registry-changed"; -const WORKERS_AVAILABLE_TRIGGER_TYPE: &str = "engine::workers-available"; -const FUNCTIONS_AVAILABLE_TRIGGER_TYPE: &str = "engine::functions-available"; - -/// The harness-PROVIDED trigger type the web worker binds to (NOT an engine built-in). -/// The web worker can boot before the harness has registered it; since `register_trigger` -/// is fire-and-forget (the engine async-rejects a bind to an unknown trigger type and -/// silently drops it), binding too early loses the guidance hook, so we bind only once -/// it is present (driven by `setup_harness_hooks`). -const HARNESS_TRIGGER_TYPES: [&str; 1] = ["harness::hook::pre-generate"]; - #[derive(Clone)] pub struct SharedState { pub config: SharedConfig, @@ -57,154 +44,20 @@ fn bind(iii: &IIIClient, trigger_type: &str, function_id: &str, config: Value) - } } -/// True iff every id in `needed` appears as a trigger-type `id` in an -/// `engine::triggers::list` response. Pure, so it's unit-testable. -fn all_trigger_types_present(resp: &Value, needed: &[&str]) -> bool { - let Some(types) = resp.get("triggers").and_then(|v| v.as_array()) else { - return false; - }; - needed.iter().all(|n| { - types - .iter() - .any(|t| t.get("id").and_then(|v| v.as_str()) == Some(*n)) - }) -} - -/// One probe: is the harness pre-generate hook trigger type registered right now? -async fn harness_trigger_types_ready(iii: &IIIClient) -> bool { - match iii - .trigger(TriggerRequest { - function_id: "engine::triggers::list".into(), - payload: json!({ "include_internal": true }), - action: None, - timeout_ms: Some(CONFIG_TIMEOUT_MS), - }) - .await - { - Ok(resp) => all_trigger_types_present(&resp, &HARNESS_TRIGGER_TYPES), - Err(_) => false, // engine not reachable / harness not up yet - } -} - -/// Bind the `web::inject-guidance` pre_generate hook EXACTLY ONCE, and only after the -/// harness has registered its trigger type. `register_trigger` is fire-and-forget — the -/// engine silently drops a bind to an unknown trigger type — so binding before the -/// harness is up would lose the hook. The shared `bound` flag keeps this idempotent -/// across the boot-time attempt and every registry-change firing. `on_error: fail_open` -/// is MANDATORY: pre_generate defaults fail-CLOSED, which would abort generation if this -/// hook ever errored/timed out; a missing guidance line must never block a turn. -async fn try_bind_harness_hooks(iii: &IIIClient, bound: &AtomicBool) { - if bound.load(Ordering::Acquire) { - return; - } - if !harness_trigger_types_ready(iii).await { - return; // harness not up yet; a later registry-change event retries - } - // Win the bind race exactly once even if several worker events fire together. - if bound - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() - { - return; - } - if bind( +/// Register the `web::inject-guidance` pre_generate hook. One shot: if the harness is +/// not up yet, the engine parks the binding as a pending intent and activates it when +/// the trigger type registers (recoverable triggers, iii #1962) — and re-parks/ +/// re-activates it across harness restarts. Nothing to watch or retry. +/// `on_error: fail_open` is MANDATORY: pre_generate defaults fail-CLOSED, which would +/// abort generation if this hook ever errored/timed out; a missing guidance line must +/// never block a turn. +pub fn setup_harness_hooks(iii: &IIIClient) { + let _ = bind( iii, "harness::hook::pre-generate", inject_guidance::GUIDANCE_HOOK_ID, json!({ "on_error": "fail_open" }), - ) - .is_some() - { - tracing::info!("web pre-generate hook bound: inject-guidance (guidance injection active)"); - } else { - // Release the claim so a later engine::*-available event retries instead of - // leaving the hook unbound until restart. - bound.store(false, Ordering::Release); - tracing::warn!( - "web inject-guidance bind failed; releasing the claim to retry on the next \ - registry-change event" - ); - } -} - -/// Bind the harness guidance hook without racing the harness's startup — and without -/// polling. The web worker can boot before the harness registers its hook trigger type, -/// so we subscribe to two engine registry-change triggers and (re)try the bind on each -/// firing; `try_bind_harness_hooks` no-ops until the harness's type appears, then binds -/// once. We need BOTH: `engine::workers-available` is immediate but can precede a -/// worker's own type registration, while `engine::functions-available` fires after -/// function registration and so reliably post-dates the harness's types. The initial -/// attempt covers the warm-start case (harness already connected, so no new event fires). -/// -// ponytail: this presence-gated bind is hand-rolled per worker (workflow and rbac-proxy -// already do the same); extract a shared crate only if a 4th consumer appears. -pub async fn setup_harness_hooks(iii: &Arc) { - let bound = Arc::new(AtomicBool::new(false)); - - let iii_for_fn = iii.clone(); - let bound_for_fn = bound.clone(); - iii.register_function( - ON_REGISTRY_CHANGED_FN_ID, - RegisterFunction::new_async(move |_event: Value| { - let iii = iii_for_fn.clone(); - let bound = bound_for_fn.clone(); - async move { - try_bind_harness_hooks(&iii, &bound).await; - Ok::(json!({ "ok": true })) - } - }) - .description( - "Internal: on engine::workers-available / engine::functions-available, bind the \ - web inject-guidance pre-generate hook once the harness has registered its \ - trigger type. Not called directly.", - ) - .request_format(registry_changed_request_schema()) - .response_format(registry_changed_response_schema()) - .metadata(json!({ "internal": true })), ); - - // Arm the event-driven retries BEFORE the initial probe, so a harness that comes up - // mid-probe is still caught. Both binds are race-free (mandatory in-process types). - for trigger_type in [ - WORKERS_AVAILABLE_TRIGGER_TYPE, - FUNCTIONS_AVAILABLE_TRIGGER_TYPE, - ] { - let _ = bind(iii, trigger_type, ON_REGISTRY_CHANGED_FN_ID, json!({})); - } - - // Warm start: the harness may already be connected, in which case no further - // registry-change event fires for it — attempt the bind now. - try_bind_harness_hooks(iii, &bound).await; -} - -/// Explicit wire schemas for `web::on-registry-changed`: the handler keeps a lossless -/// `Value` signature (registry-change event shapes belong to the engine, and the -/// payload is ignored anyway), so schemars would emit the AnyValue schema — which the -/// registry publish gate rejects (this blocked the web/v1.2.0 release). -fn registry_changed_request_schema() -> Value { - json!({ - "$schema": "http://json-schema.org/draft-07/schema#", - "title": "OnRegistryChangedEvent", - "description": "Registry-change event fired by engine::workers-available / \ - engine::functions-available. The payload is ignored — any JSON \ - is accepted; the handler just retries the guidance-hook bind.", - "type": ["null", "boolean", "number", "string", "array", "object"] - }) -} - -fn registry_changed_response_schema() -> Value { - json!({ - "$schema": "http://json-schema.org/draft-07/schema#", - "title": "OnRegistryChangedResponse", - "type": "object", - "properties": { - "ok": { - "type": "boolean", - "description": "Always true; the bind attempt is fire-and-forget." - } - }, - "required": ["ok"] - }) } pub async fn register_config(iii: &IIIClient, seed: Option<&WebConfig>) -> Result<(), String> { @@ -334,51 +187,3 @@ async fn trigger_with_retry( "{function_id} failed after {CONFIG_RETRIES} attempts: {last_err}" )) } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn all_trigger_types_present_requires_every_needed_id() { - // The harness pre-generate hook type present → ready. - let ready = json!({ - "triggers": [ - { "id": "harness::hook::pre-generate" }, - { "id": "cron" }, - ] - }); - assert!(all_trigger_types_present(&ready, &HARNESS_TRIGGER_TYPES)); - - // Harness up but pre-generate not yet registered → NOT ready (keep waiting). - let partial = json!({ "triggers": [ { "id": "harness::turn-completed" } ] }); - assert!(!all_trigger_types_present(&partial, &HARNESS_TRIGGER_TYPES)); - - // No triggers field at all (engine unreachable / empty) → NOT ready. - assert!(!all_trigger_types_present( - &json!({}), - &HARNESS_TRIGGER_TYPES - )); - } - - /// Mirrors the registry publish gate (`collect_worker_interface.py`): both - /// explicit schemas must carry a schema-defining keyword, not the AnyValue - /// schema (which blocked the web/v1.2.0 release). - #[test] - fn registry_changed_schemas_pass_the_publish_typed_gate() { - for (field, schema) in [ - ("request_schema", registry_changed_request_schema()), - ("response_schema", registry_changed_response_schema()), - ] { - let obj = schema - .as_object() - .unwrap_or_else(|| panic!("{field} must be a JSON object")); - assert!( - ["type", "properties", "$ref"] - .iter() - .any(|k| obj.contains_key(*k)), - "web::on-registry-changed {field} is untyped: {schema}" - ); - } - } -} diff --git a/web/src/main.rs b/web/src/main.rs index 70c1666af..0fccc3041 100644 --- a/web/src/main.rs +++ b/web/src/main.rs @@ -104,10 +104,11 @@ async fn main() -> Result<()> { let shared = cfg.into_shared(); functions::register_all(&iii, &shared); - // Bind the harness pre-generate hook (web::inject-guidance) once the harness's hook - // trigger type appears — reacts to engine::*-available, no polling. Presence-gated: - // the guidance is injected only while this worker is connected. - configuration::setup_harness_hooks(&iii).await; + // Bind the harness pre-generate hook (web::inject-guidance). One shot: the engine + // parks the binding until the harness registers the trigger type (recoverable + // triggers, iii #1962). Presence-gated: the guidance is injected only while this + // worker is connected. + configuration::setup_harness_hooks(&iii); configuration::register_config_trigger( &iii, From 6129083a091045e321020ec15264c7ddf1cc79c8 Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 16 Jul 2026 18:15:05 -0300 Subject: [PATCH 2/4] (MOT-4073) refactor(workflow): one-shot harness hook binds via engine recoverable triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Keep the three harness hook binds (turn-completed → wake, pre-trigger → stamp-reply, pre-generate → inject-guidance) but drop the watching/claim machinery around them: the workflow::on-registry-changed watcher, its two engine::*-available event bindings, the engine::triggers::list readiness probe, and the AtomicBool bind claim. The engine parks binds to unregistered trigger types and activates them when the type registers, re-parking across harness restarts (recoverable triggers, iii #1962). Requires engine >= 0.21.8 — noted in the README (older engines silently drop the binds). --- workflow/README.md | 7 ++ workflow/src/configuration.rs | 188 ++-------------------------------- workflow/src/main.rs | 14 ++- 3 files changed, 24 insertions(+), 185 deletions(-) diff --git a/workflow/README.md b/workflow/README.md index 79659103a..669654598 100644 --- a/workflow/README.md +++ b/workflow/README.md @@ -5,6 +5,13 @@ A deterministic DAG orchestrator running over the iii harness. Accepts state in iii-state; fans out nodes as child harness sessions with deterministic ids so duplicate deliveries are idempotent. +The three harness hook bindings (turn-completed → wake, pre-trigger → +stamp-reply, pre-generate → inject-guidance) are one-shot at startup and rely +on the engine's recoverable triggers (iii #1962, engine ≥ 0.21.8): bound +before the harness is up, they park as pending intents and activate when the +harness registers the trigger types. On older engines the binds are silently +dropped. + --- ## Getting the result back (don't poll) diff --git a/workflow/src/configuration.rs b/workflow/src/configuration.rs index a55f2eb5d..f34ee28ae 100644 --- a/workflow/src/configuration.rs +++ b/workflow/src/configuration.rs @@ -8,7 +8,6 @@ //! Every other field is a per-call tuning knob read from the live snapshot via //! [`Deps::cfg`](crate::functions::Deps::cfg); a change swaps the snapshot. -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; @@ -109,154 +108,19 @@ fn bind(iii: &IIIClient, trigger_type: &str, function_id: &str, config: Value) - } } -/// Harness-PROVIDED trigger types the workflow worker binds to (NOT engine -/// built-ins like `cron`). On a cold start the workflow worker can boot before the -/// harness has registered these; since `register_trigger` is fire-and-forget (the -/// engine async-rejects a bind to an unknown trigger type and silently drops it), -/// binding too early loses the wake / reply-stamp / guidance hooks, so we bind only -/// once these are present (driven by `setup_harness_hooks`). -const HARNESS_TRIGGER_TYPES: [&str; 3] = [ - "harness::turn-completed", - "harness::hook::pre-trigger", - "harness::hook::pre-generate", -]; - -/// True iff every id in `needed` appears as a trigger-type `id` in an -/// `engine::triggers::list` response. Pure, so it's unit-testable. -fn all_trigger_types_present(resp: &Value, needed: &[&str]) -> bool { - let Some(types) = resp.get("triggers").and_then(|v| v.as_array()) else { - return false; - }; - needed.iter().all(|n| { - types - .iter() - .any(|t| t.get("id").and_then(|v| v.as_str()) == Some(*n)) - }) -} - -/// One probe: are all harness trigger types registered on the engine right now? -async fn harness_trigger_types_ready(iii: &IIIClient) -> bool { - match iii - .trigger(TriggerRequest { - function_id: "engine::triggers::list".into(), - payload: json!({ "include_internal": true }), - action: None, - timeout_ms: Some(CONFIG_TIMEOUT_MS), - }) - .await - { - Ok(resp) => all_trigger_types_present(&resp, &HARNESS_TRIGGER_TYPES), - Err(_) => false, // engine not reachable / harness not up yet - } -} - -const ON_REGISTRY_CHANGED_FN_ID: &str = "workflow::on-registry-changed"; -/// Two mandatory in-process engine triggers we subscribe to so we bind the harness hooks -/// as soon as the harness is up. Binding to them never races (both registered at engine -/// boot). They cover DIFFERENT timings, and the harness needs the second one: -/// - `engine::workers-available` fires IMMEDIATELY on any worker connect / metadata / -/// disconnect. But the SDK auto-sends a worker's metadata register right after connect, -/// which can PRECEDE that worker registering its own trigger types: the harness -/// registers `harness::hook::*` only after two awaited config RPCs (harness main.rs), -/// so its workers-available firings land BEFORE its types exist — a probe then misses. -/// - `engine::functions-available` fires via the engine's internal function-set poll, -/// AFTER a worker registers its functions. The harness registers its trigger types -/// before its functions, so this firing reliably post-dates the harness's types. This -/// is the event that closes the cold-start hole workers-available alone left open. -const WORKERS_AVAILABLE_TRIGGER_TYPE: &str = "engine::workers-available"; -const FUNCTIONS_AVAILABLE_TRIGGER_TYPE: &str = "engine::functions-available"; - /// Bind the three harness hooks (turn-completed → wake, pre-trigger → stamp-reply, -/// pre-generate → inject-guidance) EXACTLY ONCE, and only after the harness has -/// registered their trigger types (one `engine::triggers::list` probe). Because -/// `register_trigger` is fire-and-forget — the engine silently drops a bind to an -/// unknown trigger type — binding before the harness is up would lose the hooks. The -/// shared `bound` flag keeps this idempotent across the boot-time attempt and every -/// registry-change firing (which may run concurrently). After the first success the -/// engine re-binds these automatically if the harness later restarts -/// (`register_trigger_type` re-scans stored triggers), so one pass is enough. -async fn try_bind_harness_hooks(iii: &IIIClient, bound: &AtomicBool) { - if bound.load(Ordering::Acquire) { - return; - } - if !harness_trigger_types_ready(iii).await { - return; // harness not up yet; a later registry-change event retries - } - // Win the bind race exactly once even if several worker events fire together. - if bound - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() - { - return; - } - let turn_completed = bind_turn_completed(iii).is_some(); - let pre_trigger = bind_pre_trigger_hook(iii).is_some(); - let pre_generate = bind_pre_generate_hook(iii).is_some(); - if turn_completed && pre_trigger && pre_generate { - tracing::info!( - "workflow harness hooks registered: turn-completed → wake, pre-trigger → \ - stamp-reply, pre-generate → inject-guidance (guidance injection active)" - ); - } else { - // A bind transiently failed. Release the claim so a later engine::*-available - // event retries, instead of leaving the missing hook(s) unbound until restart. - // Re-binding an already-bound hook on retry is tolerated: the bound hooks - // (turn-completed → wake, pre-trigger → stamp) are idempotent downstream. - bound.store(false, Ordering::Release); - tracing::warn!( - turn_completed, - pre_trigger, - pre_generate, - "not all workflow harness hooks bound; releasing the bind claim to retry on the \ - next registry-change event" - ); - } -} - -/// Bind the harness hooks without racing the harness's startup — and without polling. -/// -/// The workflow worker can boot before the harness has registered its hook trigger types. -/// Rather than poll, we subscribe to two engine registry-change triggers and (re)try the -/// binds on each firing; `try_bind_harness_hooks` no-ops until the harness's types appear, -/// then binds once. We need BOTH because they fire at different moments (see the trigger -/// consts): `engine::workers-available` is immediate but can precede a worker's own type -/// registration, while `engine::functions-available` fires after function registration and -/// so reliably post-dates the harness's types. The initial attempt covers the warm-start -/// case (harness already connected, so no new event fires for it). -pub async fn setup_harness_hooks(iii: &Arc) { - let bound = Arc::new(AtomicBool::new(false)); - - let iii_for_fn = iii.clone(); - let bound_for_fn = bound.clone(); - iii.register_function( - ON_REGISTRY_CHANGED_FN_ID, - RegisterFunction::new_async(move |_event: Value| { - let iii = iii_for_fn.clone(); - let bound = bound_for_fn.clone(); - async move { - try_bind_harness_hooks(&iii, &bound).await; - Ok::(json!({ "ok": true })) - } - }) - .description( - "Internal: on engine::workers-available / engine::functions-available, bind the \ - workflow harness hooks (turn-completed, pre-trigger, pre-generate) once the \ - harness has registered their trigger types. Not called directly.", - ), +/// pre-generate → inject-guidance). One shot: if the harness is not up yet, the +/// engine parks each binding as a pending intent and activates it when the trigger +/// type registers (recoverable triggers, iii #1962) — and re-parks/re-activates +/// them across harness restarts. Nothing to watch or retry. +pub fn setup_harness_hooks(iii: &IIIClient) { + let _ = bind_turn_completed(iii); + let _ = bind_pre_trigger_hook(iii); + let _ = bind_pre_generate_hook(iii); + tracing::info!( + "workflow harness hooks registered: turn-completed → wake, pre-trigger → \ + stamp-reply, pre-generate → inject-guidance (guidance injection active)" ); - - // Arm the event-driven retries BEFORE the initial probe, so a harness that comes up - // mid-probe is still caught. Both binds are race-free (mandatory in-process types). - for trigger_type in [ - WORKERS_AVAILABLE_TRIGGER_TYPE, - FUNCTIONS_AVAILABLE_TRIGGER_TYPE, - ] { - let _ = bind(iii, trigger_type, ON_REGISTRY_CHANGED_FN_ID, json!({})); - } - - // Warm start: the harness may already be connected, in which case no further - // registry-change event fires for it — attempt the bind now. - try_bind_harness_hooks(iii, &bound).await; } /// (Re)bind the cron node-timeout sweep from the current config. @@ -473,34 +337,4 @@ mod tests { .await; assert_eq!(cell.read().await.default_pending_timeout_ms, 9999); } - - #[test] - fn all_trigger_types_present_requires_every_needed_id() { - // All three harness trigger types present → ready. - let ready = serde_json::json!({ - "triggers": [ - { "id": "harness::turn-completed" }, - { "id": "harness::hook::pre-trigger" }, - { "id": "harness::hook::pre-generate" }, - { "id": "cron" }, - ] - }); - assert!(all_trigger_types_present(&ready, &HARNESS_TRIGGER_TYPES)); - - // One missing (the exact cold-start bug: harness up but pre-generate not - // yet registered) → NOT ready, so we keep waiting instead of binding early. - let partial = serde_json::json!({ - "triggers": [ - { "id": "harness::turn-completed" }, - { "id": "harness::hook::pre-trigger" }, - ] - }); - assert!(!all_trigger_types_present(&partial, &HARNESS_TRIGGER_TYPES)); - - // No triggers field at all (engine unreachable / empty) → NOT ready. - assert!(!all_trigger_types_present( - &serde_json::json!({}), - &HARNESS_TRIGGER_TYPES - )); - } } diff --git a/workflow/src/main.rs b/workflow/src/main.rs index 0b72e4f7e..6bf2eef7a 100644 --- a/workflow/src/main.rs +++ b/workflow/src/main.rs @@ -8,8 +8,8 @@ //! 5. Build ConfigCell + Deps. //! 6. Register all functions. //! 7. Bind the cron sweep trigger (retain handle for hot-reload). -//! 8. Set up the harness hooks: react to `engine::workers-available` and bind the -//! turn-completed / pre-trigger / pre-generate hooks once the harness is up (no poll). +//! 8. Set up the harness hooks: one-shot binds for turn-completed / pre-trigger / +//! pre-generate (the engine parks them until the harness registers the types). //! 9. Resume-scan: re-enqueue a tick for every non-terminal run (crash recovery). //! 10. LAST: bind the configuration-change trigger so its handler closes over //! the fully-built snapshot cell + the cron handle. @@ -123,12 +123,10 @@ async fn main() -> Result<()> { }); // Bind the three HARNESS-provided hooks (turn-completed → wake, hook::pre-trigger - // → reply stamping, hook::pre-generate → guidance injection). register_trigger - // is fire-and-forget — the engine silently drops a bind to a trigger type it does - // not know yet — so on a cold start where this worker boots before the harness has - // registered them the binds would be lost. setup_harness_hooks reacts to - // engine::workers-available (no polling) and binds once the harness's types appear. - configuration::setup_harness_hooks(&iii).await; + // → reply stamping, hook::pre-generate → guidance injection). One shot: the engine + // parks each binding until the harness registers the trigger type (recoverable + // triggers, iii #1962) and re-activates them across harness restarts. + configuration::setup_harness_hooks(&iii); // Crash recovery: a parked AwaitingNodes run has no enqueued tick. // Re-drive each non-terminal run so it can make progress. From 6baf61c74f46ba18c06e4f3f5949a5c00e42d469 Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 16 Jul 2026 18:15:16 -0300 Subject: [PATCH 3/4] (MOT-4073) refactor(scrapling): one-shot guidance hook bind via engine recoverable triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python port of the same cleanup: delete scrapling::on-registry-changed and its two engine::*-available bindings, the engine::triggers::list warm-start probe, the BindOnce claim, try_bind_sync/try_bind_async, and the registry-changed wire schemas; register_trigger the pre-generate hook once at startup. The engine parks the bind until the harness registers the type and re-parks across harness restarts (recoverable triggers, iii #1962). Requires engine >= 0.21.8 — noted in the README. --- scrapling/README.md | 8 ++ scrapling/src/guidance.py | 134 +++---------------------------- scrapling/tests/test_guidance.py | 95 ++++------------------ 3 files changed, 34 insertions(+), 203 deletions(-) diff --git a/scrapling/README.md b/scrapling/README.md index bda7e0c00..e757ae25f 100644 --- a/scrapling/README.md +++ b/scrapling/README.md @@ -6,6 +6,14 @@ on the iii bus: fast HTTP with TLS impersonation, a Camoufox anti-bot browser, a full Playwright/Chromium browser, screenshots, and CSS/XPath/regex/adaptive extraction. +While connected it also injects a usage section into the agent system prompt +via the harness `pre-generate` hook (`scrapling::inject-guidance`), so the +guidance is presence-gated: no scrapling worker, no prompt text. The binding is +one-shot at startup and relies on the engine's recoverable triggers (iii #1962, +engine ≥ 0.21.8): bound before the harness is up, it parks as a pending intent +and activates when the harness registers the trigger type. On older engines the +bind is silently dropped. + ## Install ```bash diff --git a/scrapling/src/guidance.py b/scrapling/src/guidance.py index 8613ebc8a..eb6486a3d 100644 --- a/scrapling/src/guidance.py +++ b/scrapling/src/guidance.py @@ -3,28 +3,22 @@ connected. The binding dies with the worker, so the guidance is presence-gated for free: a deployment without scrapling never pays for it. -The hook may only be bound after the harness has registered its -`harness::hook::pre-generate` trigger type — `register_trigger` is -fire-and-forget and the engine silently drops binds to unknown types. So -`setup()` also registers `scrapling::on-registry-changed`, bound to the engine -registry-change triggers, and (re)tries the bind on every firing, plus once at -boot for the warm-start case (harness already up → no event will fire). +The bind is one shot: if the harness is not up yet, the engine parks the +binding as a pending intent and activates it when the trigger type registers +(recoverable triggers, iii #1962) — and re-parks/re-activates it across +harness restarts. Nothing to watch or retry. Mirrors web/src/functions/inject_guidance.rs + configuration.rs. """ from __future__ import annotations import logging -import threading from typing import Any log = logging.getLogger("scrapling.guidance") HOOK_ID = "scrapling::inject-guidance" -ON_REGISTRY_CHANGED_ID = "scrapling::on-registry-changed" HOOK_TRIGGER_TYPE = "harness::hook::pre-generate" -REGISTRY_TRIGGER_TYPES = ("engine::workers-available", "engine::functions-available") -PROBE_TIMEOUT_MS = 5_000 # The single canonical copy of the scrapling usage guidance (same role as # WEB_GUIDANCE in the web worker). Pure USAGE guidance: the hook only fires @@ -86,15 +80,6 @@ }, "required": ["mutations"], } -REGISTRY_CHANGED_REQUEST: dict[str, Any] = { - "description": "Registry-change event; payload is ignored — the handler just retries the hook bind.", - "type": ["null", "boolean", "number", "string", "array", "object"], -} -REGISTRY_CHANGED_RESPONSE: dict[str, Any] = { - "type": "object", - "properties": {"ok": {"type": "boolean"}}, - "required": ["ok"], -} def mutations_for(base: str) -> dict[str, Any]: @@ -117,88 +102,8 @@ async def inject_guidance(payload: dict[str, Any] | None) -> dict[str, Any]: return mutations_for(base if isinstance(base, str) else "") -def hook_type_present(resp: Any) -> bool: - """True iff an `engine::triggers::list` response carries the harness hook type.""" - if not isinstance(resp, dict): - return False - triggers = resp.get("triggers") - if not isinstance(triggers, list): - return False - return any(isinstance(t, dict) and t.get("id") == HOOK_TRIGGER_TYPE for t in triggers) - - -class BindOnce: - """Claim/release around the one hook bind. Thread-safe because the boot-time - attempt runs on the main thread while event retries run on the SDK loop.""" - - def __init__(self) -> None: - self._lock = threading.Lock() - self._bound = False - - def is_bound(self) -> bool: - with self._lock: - return self._bound - - def claim(self) -> bool: - with self._lock: - if self._bound: - return False - self._bound = True - return True - - def release(self) -> None: - with self._lock: - self._bound = False - - -def _probe_request() -> dict[str, Any]: - return { - "function_id": "engine::triggers::list", - "payload": {"include_internal": True}, - "timeout_ms": PROBE_TIMEOUT_MS, - } - - -def _bind_hook(iii: Any, state: BindOnce) -> None: - if not state.claim(): - return - try: - # on_error fail_open is MANDATORY: pre_generate defaults fail-CLOSED, and a - # missing guidance line must never abort a turn. - iii.register_trigger({"type": HOOK_TRIGGER_TYPE, "function_id": HOOK_ID, "config": {"on_error": "fail_open"}}) - log.info("scrapling pre-generate hook bound (guidance injection active)") - except Exception: - state.release() - log.warning("inject-guidance bind failed; retrying on the next registry-change event", exc_info=True) - - -def try_bind_sync(iii: Any, state: BindOnce) -> None: - """Boot-time warm start; a probe failure just defers to the event retries.""" - if state.is_bound(): - return - try: - resp = iii.trigger(_probe_request()) - except Exception: - return - if hook_type_present(resp): - _bind_hook(iii, state) - - -async def try_bind_async(iii: Any, state: BindOnce) -> None: - if state.is_bound(): - return - try: - resp = await iii.trigger_async(_probe_request()) - except Exception: - return - if hook_type_present(resp): - _bind_hook(iii, state) - - def setup(iii: Any) -> None: - """Register the hook + its presence-gated bind plumbing. Call once at boot.""" - state = BindOnce() - + """Register the hook function and bind it one-shot. Call once at boot.""" iii.register_function( HOOK_ID, inject_guidance, @@ -211,28 +116,7 @@ def setup(iii: Any) -> None: metadata={"internal": True}, ) - async def on_registry_changed(_payload: Any) -> dict[str, Any]: - await try_bind_async(iii, state) - return {"ok": True} - - iii.register_function( - ON_REGISTRY_CHANGED_ID, - on_registry_changed, - description=( - "Internal: on engine registry changes, bind the scrapling inject-guidance pre-generate " - "hook once the harness has registered its trigger type. Not called directly." - ), - request_format=REGISTRY_CHANGED_REQUEST, - response_format=REGISTRY_CHANGED_RESPONSE, - metadata={"internal": True}, - ) - - # Arm the event-driven retries BEFORE the warm-start probe, so a harness that - # comes up mid-probe is still caught. - for trigger_type in REGISTRY_TRIGGER_TYPES: - try: - iii.register_trigger({"type": trigger_type, "function_id": ON_REGISTRY_CHANGED_ID, "config": {}}) - except Exception: - log.warning("registry-change bind failed for %s", trigger_type, exc_info=True) - - try_bind_sync(iii, state) + # on_error fail_open is MANDATORY: pre_generate defaults fail-CLOSED, and a + # missing guidance line must never abort a turn. + iii.register_trigger({"type": HOOK_TRIGGER_TYPE, "function_id": HOOK_ID, "config": {"on_error": "fail_open"}}) + log.info("scrapling pre-generate hook bound (guidance injection active)") diff --git a/scrapling/tests/test_guidance.py b/scrapling/tests/test_guidance.py index 9f60b31c0..39152b691 100644 --- a/scrapling/tests/test_guidance.py +++ b/scrapling/tests/test_guidance.py @@ -1,4 +1,4 @@ -"""Guidance injection: pure mutation logic + the presence-gated hook bind.""" +"""Guidance injection: pure mutation logic + the one-shot hook bind.""" from __future__ import annotations @@ -7,43 +7,22 @@ from src import guidance -READY = {"triggers": [{"id": guidance.HOOK_TRIGGER_TYPE}, {"id": "http"}]} -NOT_READY = {"triggers": [{"id": "http"}]} - class FakeIII: """Duck-types the slice of IIIClient that guidance.setup touches.""" - def __init__(self, triggers_resp: Any = None, fail_hook_binds: int = 0): + def __init__(self): self.functions: dict[str, dict[str, Any]] = {} self.handlers: dict[str, Any] = {} self.trigger_binds: list[dict[str, Any]] = [] - self.triggers_resp = triggers_resp - self.fail_hook_binds = fail_hook_binds def register_function(self, function_id, handler, **kwargs): self.functions[function_id] = kwargs self.handlers[function_id] = handler def register_trigger(self, spec): - if spec["type"] == guidance.HOOK_TRIGGER_TYPE and self.fail_hook_binds > 0: - self.fail_hook_binds -= 1 - raise RuntimeError("engine hiccup") self.trigger_binds.append(spec) - def trigger(self, request): - assert request["function_id"] == "engine::triggers::list" - assert request["payload"] == {"include_internal": True} - if self.triggers_resp is None: - raise RuntimeError("engine down") - return self.triggers_resp - - async def trigger_async(self, request): - return self.trigger(request) - - def hook_binds(self): - return [b for b in self.trigger_binds if b["type"] == guidance.HOOK_TRIGGER_TYPE] - # ---- pure mutation logic ---------------------------------------------------- @@ -85,62 +64,22 @@ def test_guidance_names_the_full_surface(): assert needle in guidance.GUIDANCE, f"missing: {needle}" -def test_hook_type_present(): - assert guidance.hook_type_present(READY) - assert not guidance.hook_type_present(NOT_READY) - assert not guidance.hook_type_present({}) - assert not guidance.hook_type_present({"triggers": "nope"}) - assert not guidance.hook_type_present(None) - +# ---- setup: one-shot bind ---------------------------------------------------- -# ---- setup + bind lifecycle -------------------------------------------------- - -def test_setup_warm_start_binds_hook_when_harness_ready(): - iii = FakeIII(triggers_resp=READY) +def test_setup_registers_hook_and_binds_one_shot(): + iii = FakeIII() guidance.setup(iii) - assert set(iii.functions) == {guidance.HOOK_ID, guidance.ON_REGISTRY_CHANGED_ID} - for kwargs in iii.functions.values(): - assert kwargs["request_format"].get("type"), "untyped request schema" - assert kwargs["response_format"].get("type"), "untyped response schema" - - registry_binds = [b for b in iii.trigger_binds if b["type"] in guidance.REGISTRY_TRIGGER_TYPES] - assert {b["type"] for b in registry_binds} == set(guidance.REGISTRY_TRIGGER_TYPES) - assert all(b["function_id"] == guidance.ON_REGISTRY_CHANGED_ID for b in registry_binds) - - hooks = iii.hook_binds() - assert len(hooks) == 1 - assert hooks[0]["function_id"] == guidance.HOOK_ID - assert hooks[0]["config"] == {"on_error": "fail_open"} - - -def test_setup_defers_bind_until_registry_event_says_ready(): - iii = FakeIII(triggers_resp=NOT_READY) - guidance.setup(iii) - assert iii.hook_binds() == [] - - on_changed = iii.handlers[guidance.ON_REGISTRY_CHANGED_ID] - iii.triggers_resp = READY - assert asyncio.run(on_changed({})) == {"ok": True} - assert len(iii.hook_binds()) == 1 - - # Further events are no-ops: the bind is claimed exactly once. - asyncio.run(on_changed({})) - assert len(iii.hook_binds()) == 1 - - -def test_setup_survives_engine_down_at_boot(): - iii = FakeIII(triggers_resp=None) # probe raises - guidance.setup(iii) - assert iii.hook_binds() == [] - - -def test_failed_hook_bind_releases_the_claim_for_a_retry(): - iii = FakeIII(triggers_resp=READY, fail_hook_binds=1) - guidance.setup(iii) # warm-start bind attempt raises → claim released - assert iii.hook_binds() == [] - - on_changed = iii.handlers[guidance.ON_REGISTRY_CHANGED_ID] - asyncio.run(on_changed({})) - assert len(iii.hook_binds()) == 1 + assert set(iii.functions) == {guidance.HOOK_ID} + kwargs = iii.functions[guidance.HOOK_ID] + assert kwargs["request_format"].get("type"), "untyped request schema" + assert kwargs["response_format"].get("type"), "untyped response schema" + + assert iii.trigger_binds == [ + { + "type": guidance.HOOK_TRIGGER_TYPE, + "function_id": guidance.HOOK_ID, + "config": {"on_error": "fail_open"}, + } + ] From ce983ffb8796d60ed959236cb92b78480547f548 Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Thu, 16 Jul 2026 18:15:29 -0300 Subject: [PATCH 4/4] (MOT-4073) refactor(memory): drop retry_bindings; keep a queue::define poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The delta-attribution retry loop (snapshot engine::triggers::info counts, re-request, confirm on increase) existed only because binds to not-yet-registered sibling trigger types were silently dropped. The engine now parks them as pending intents and activates them when the owner registers, re-parking across restarts (recoverable triggers, iii #1962), so the one-shot bind pass suffices for all four seams. queue::define is the one guarantee parking does not cover — it is an RPC the queue worker must serve before enqueue works — so keep a minimal poll for just that. Requires engine >= 0.21.8 — noted in the README. --- memory/README.md | 5 ++++ memory/src/main.rs | 75 ++++++++-------------------------------------- 2 files changed, 17 insertions(+), 63 deletions(-) diff --git a/memory/README.md b/memory/README.md index b24c37560..d709681a2 100644 --- a/memory/README.md +++ b/memory/README.md @@ -51,6 +51,11 @@ All fields hot-reload through the `configuration` worker (rendered as a form in Bank selection order: turn metadata `memory_bank` → session metadata `memory_bank` (`session::set-meta`) → configured `default_bank`. A session-lookup failure injects nothing rather than falling back across banks. +All seam bindings are one-shot at startup and rely on the engine's recoverable +triggers (iii #1962, engine ≥ 0.21.8): bound before the owning sibling (harness, +session-manager, queue) is up, they park as pending intents and activate when +the trigger type registers. On older engines the binds are silently dropped. + ## Functions | Function | Purpose | diff --git a/memory/src/main.rs b/memory/src/main.rs index 90d56e127..7110450c5 100644 --- a/memory/src/main.rs +++ b/memory/src/main.rs @@ -111,64 +111,13 @@ fn required_bindings() -> Vec<(&'static str, &'static str, serde_json::Value)> { ] } -/// Boot-order safety net: in an orderly startup wave the siblings that -/// own these trigger types (harness, session-manager, queue) may register -/// AFTER this worker, so the boot-time binding requests die with -/// `trigger_type_not_found` — silently, since registration acks are -/// async. Poll `engine::triggers::info` and re-request each binding until -/// every one is live; without this, injection and extraction never run -/// after an unlucky boot (mirrors approval-gate's hook retry). -fn retry_bindings(iii: Arc) { +/// Ensure the extraction queue exists. Trigger parking covers the binds +/// (recoverable triggers, iii #1962), but `queue::define` is an RPC the +/// queue worker must be up to serve — enqueue rejects an undefined queue — +/// so poll until the (idempotent) definition lands. +fn ensure_extraction_queue(iii: Arc) { tokio::spawn(async move { - // engine::triggers::info exposes only an AGGREGATE instance_count, - // and other workers bind the same trigger types (web also binds - // pre-generate; several workers use durable:subscriber). A bare - // `count > 0` therefore confirms SOMEONE's binding, not ours — - // observed live: memory's hook binding died at boot while web's - // kept the count positive, and injection silently never ran. - // Until the engine exposes binding membership, attribute by - // DELTA: snapshot the count, re-request our binding, and treat - // only an increase over the snapshot as our registration landing. - let mut baselines: std::collections::HashMap<&'static str, u64> = - std::collections::HashMap::new(); - let mut confirmed: std::collections::HashSet<&'static str> = - std::collections::HashSet::new(); loop { - let mut all_ready = true; - for (trigger_type, function_id, config) in required_bindings() { - if confirmed.contains(trigger_type) { - continue; - } - let count = iii - .trigger(iii_sdk::protocol::TriggerRequest { - function_id: "engine::triggers::info".into(), - payload: json!({ "id": trigger_type }), - action: None, - timeout_ms: Some(5_000), - }) - .await - .ok() - .and_then(|v| v.get("instance_count").and_then(serde_json::Value::as_u64)); - let Some(count) = count else { - // Trigger type not registered yet (owner still booting). - all_ready = false; - baselines.remove(trigger_type); - continue; - }; - match baselines.get(trigger_type) { - Some(&baseline) if count > baseline => { - // The count rose after our request: ours landed. - confirmed.insert(trigger_type); - } - _ => { - all_ready = false; - baselines.insert(trigger_type, count); - bind_best_effort(&iii, trigger_type, function_id, config); - } - } - } - // The extraction queue definition is idempotent; keep asking - // until the queue worker is up. let defined = iii .trigger(iii_sdk::protocol::TriggerRequest { function_id: "queue::define".into(), @@ -178,8 +127,8 @@ fn retry_bindings(iii: Arc) { }) .await .is_ok(); - if all_ready && defined { - tracing::info!("memory bindings confirmed (hooks, turn events, session GC, queue)"); + if defined { + tracing::info!("memory extraction queue defined"); break; } tokio::time::sleep(std::time::Duration::from_secs(15)).await; @@ -264,14 +213,14 @@ async fn main() -> Result<()> { // Bind the harness/session/queue seams (injection fail-OPEN — a // memory failure must never block a turn — plus turn-completed - // extraction, session GC, and the durable extraction consumer), then - // keep retrying until every binding is confirmed live: in an orderly - // boot wave the owning siblings may register their trigger types - // after this worker. + // extraction, session GC, and the durable extraction consumer). + // One shot: if an owning sibling boots after this worker, the engine + // parks the binding and activates it when the trigger type registers + // (recoverable triggers, iii #1962). for (trigger_type, function_id, config) in required_bindings() { bind_best_effort(&iii, trigger_type, function_id, config); } - retry_bindings(iii.clone()); + ensure_extraction_queue(iii.clone()); // Catch-up vector backfill for memories saved while embeddings were // unavailable (delayed so the router and providers finish booting).