Skip to content

(MOT-4073) refactor(workers): adopt engine recoverable triggers — remove hand-rolled pending-bind dances #523

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andersonleal merged 4 commits into from
Jul 16, 2026
Merged

(MOT-4073) refactor(workers): adopt engine recoverable triggers — remove hand-rolled pending-bind dances #523

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ebdcb25
(MOT-4073) refactor(web): one-shot harness hook bind via engine recov…
andersonleal Jul 16, 2026
6129083
(MOT-4073) refactor(workflow): one-shot harness hook binds via engine…
andersonleal Jul 16, 2026
6baf61c
(MOT-4073) refactor(scrapling): one-shot guidance hook bind via engin…
andersonleal Jul 16, 2026
ce983ff
(MOT-4073) refactor(memory): drop retry_bindings; keep a queue::defin…
andersonleal Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions memory/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
75 changes: 12 additions & 63 deletions memory/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<IIIClient>) {
/// 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<IIIClient>) {
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(),
Expand All @@ -178,8 +127,8 @@ fn retry_bindings(iii: Arc<IIIClient>) {
})
.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;
Expand Down Expand Up @@ -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).
Expand Down
8 changes: 8 additions & 0 deletions scrapling/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
134 changes: 9 additions & 125 deletions scrapling/src/guidance.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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]:
Expand All @@ -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,
Expand All @@ -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)")

@coderabbitai coderabbitai Bot Jul 16, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Log the binding request rather than immediate activation.

On a cold start, Line 122 executes while the intent may still be parked, so “guidance injection active” is inaccurate until the harness registers its trigger type. Use wording such as “binding requested.”

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@scrapling/src/guidance.py` at line 122, Update the log message in the
guidance binding setup to state that binding was requested rather than implying
guidance injection is already active. Preserve the existing log call and context
while using wording such as “binding requested.”

Loading
Loading