mirror of
https://github.com/rookiestar28/ComfyUI-OpenClaw.git
synced 2026-08-14 00:48:07 +00:00
fix(parameter-lab): correlate host queue receipts
This commit is contained in:
rookiestar28
@@ -0,0 +1,90 @@
|
||||
"""Authoritative, transient prompt-ID carrier for Parameter Lab submissions."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
PARAMETER_LAB_RECEIPT_KEY = "__openclaw_parameter_lab_receipt__"
|
||||
PARAMETER_LAB_RECEIPT_VERSION = 1
|
||||
|
||||
_CANONICAL_UUID_RE = re.compile(
|
||||
r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-" r"[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
|
||||
)
|
||||
|
||||
|
||||
def _valid_marker(marker: Any) -> str | None:
|
||||
if not isinstance(marker, dict):
|
||||
return None
|
||||
if set(marker) != {"version", "prompt_id"}:
|
||||
return None
|
||||
if marker.get("version") != PARAMETER_LAB_RECEIPT_VERSION:
|
||||
return None
|
||||
prompt_id = marker.get("prompt_id")
|
||||
if not isinstance(prompt_id, str) or not _CANONICAL_UUID_RE.fullmatch(prompt_id):
|
||||
return None
|
||||
return prompt_id
|
||||
|
||||
|
||||
def consume_parameter_lab_queue_receipt(json_data: Any) -> Any:
|
||||
"""Strip one transient marker and promote its UUID to native ``prompt_id``.
|
||||
|
||||
The function is copy-on-write so other prompt handlers never observe an in-place
|
||||
mutation of their input. Invalid markers are still stripped, but never gain
|
||||
identifier authority.
|
||||
"""
|
||||
|
||||
if not isinstance(json_data, dict):
|
||||
return json_data
|
||||
extra_data = json_data.get("extra_data")
|
||||
if not isinstance(extra_data, dict):
|
||||
return json_data
|
||||
extra_pnginfo = extra_data.get("extra_pnginfo")
|
||||
if not isinstance(extra_pnginfo, dict):
|
||||
return json_data
|
||||
workflow = extra_pnginfo.get("workflow")
|
||||
if not isinstance(workflow, dict):
|
||||
return json_data
|
||||
workflow_extra = workflow.get("extra")
|
||||
if (
|
||||
not isinstance(workflow_extra, dict)
|
||||
or PARAMETER_LAB_RECEIPT_KEY not in workflow_extra
|
||||
):
|
||||
return json_data
|
||||
|
||||
marker = workflow_extra.get(PARAMETER_LAB_RECEIPT_KEY)
|
||||
prompt_id = _valid_marker(marker)
|
||||
|
||||
# CRITICAL: the carrier is transient; never retain it in queue/history/image metadata.
|
||||
next_workflow_extra = dict(workflow_extra)
|
||||
next_workflow_extra.pop(PARAMETER_LAB_RECEIPT_KEY, None)
|
||||
next_workflow = dict(workflow)
|
||||
next_workflow["extra"] = next_workflow_extra
|
||||
next_pnginfo = dict(extra_pnginfo)
|
||||
next_pnginfo["workflow"] = next_workflow
|
||||
next_extra_data = dict(extra_data)
|
||||
next_extra_data["extra_pnginfo"] = next_pnginfo
|
||||
result = dict(json_data)
|
||||
result["extra_data"] = next_extra_data
|
||||
|
||||
if prompt_id is not None:
|
||||
# CRITICAL: the frontend assigns this exact UUID only after promptQueued.
|
||||
# Preserving a different earlier handler value would cross-assign lifecycle
|
||||
# events to the wrong Parameter Lab run.
|
||||
result["prompt_id"] = prompt_id
|
||||
return result
|
||||
|
||||
|
||||
def register_parameter_lab_queue_receipt_handler(server: Any) -> bool:
|
||||
"""Register the official ComfyUI on-prompt handler exactly once."""
|
||||
|
||||
handlers = getattr(server, "on_prompt_handlers", ())
|
||||
if isinstance(handlers, (list, tuple)) and (
|
||||
consume_parameter_lab_queue_receipt in handlers
|
||||
):
|
||||
return False
|
||||
add_handler = getattr(server, "add_on_prompt_handler", None)
|
||||
if not callable(add_handler):
|
||||
raise RuntimeError("ComfyUI host does not expose add_on_prompt_handler")
|
||||
add_handler(consume_parameter_lab_queue_receipt)
|
||||
return True
|
||||
@@ -208,6 +208,9 @@ def _initialize_registries_and_security_gate() -> None:
|
||||
def _do_full_registration(server) -> None:
|
||||
"""Register all OpenClaw routes including bridge/scheduler bindings."""
|
||||
from .access_control import require_admin_token
|
||||
from .parameter_lab_queue_receipt import (
|
||||
register_parameter_lab_queue_receipt_handler,
|
||||
)
|
||||
from .plugins.async_bridge import run_async_in_sync_context
|
||||
from .queue_submit import submit_prompt
|
||||
from .route_bootstrap_contract import load_route_bootstrap_contract
|
||||
@@ -223,6 +226,8 @@ def _do_full_registration(server) -> None:
|
||||
register_trigger_routes = contract["register_trigger_routes"]
|
||||
|
||||
register_routes(server)
|
||||
# CRITICAL: receipt promotion is required for exact Parameter Lab run ownership.
|
||||
register_parameter_lab_queue_receipt_handler(server)
|
||||
register_preset_routes(server.app)
|
||||
register_schedule_routes(server.app, require_admin_token_fn=require_admin_token)
|
||||
|
||||
@@ -449,7 +454,7 @@ def _run_registration_retry_loop(
|
||||
_mark_startup_fatal("route_registration", exc)
|
||||
_store_registration_failure(exc, generation=owner_generation)
|
||||
logger.error(
|
||||
"Route registration failed " "(attempt=%s, error_type=%s)",
|
||||
"Route registration failed (attempt=%s, error_type=%s)",
|
||||
attempt,
|
||||
type(exc).__name__,
|
||||
)
|
||||
@@ -574,7 +579,7 @@ def register_routes_once() -> None:
|
||||
)
|
||||
_store_registration_failure(exc, generation=generation)
|
||||
logger.error(
|
||||
"Route registration retry owner failed to start " "(error_type=%s)",
|
||||
"Route registration retry owner failed to start (error_type=%s)",
|
||||
type(exc).__name__,
|
||||
)
|
||||
raise
|
||||
|
||||
@@ -233,6 +233,7 @@
|
||||
"services/packs/pack_types.py",
|
||||
"services/parameter_lab.py",
|
||||
"services/parameter_lab_policy.py",
|
||||
"services/parameter_lab_queue_receipt.py",
|
||||
"services/paths.py",
|
||||
"services/permission_posture.py",
|
||||
"services/planner.py",
|
||||
|
||||
@@ -379,4 +379,217 @@ test.describe('Parameter Lab - Dynamic Dimensions', () => {
|
||||
await page.selectOption('.dim-candidate-select', { value: 'xl.ckpt' });
|
||||
await expect(page.locator('.openclaw-chip >> text=xl.ckpt')).toBeVisible();
|
||||
});
|
||||
|
||||
test('uses an authoritative receipt when the host queue API returns only boolean', async ({ page }) => {
|
||||
await page.evaluate(async () => {
|
||||
const mod = await import('/web/openclaw_api.js');
|
||||
const { api } = await import('/scripts/api.js');
|
||||
const receiptMod = await import('/web/openclaw_parameter_lab_receipt.js');
|
||||
window.__labRunUpdates = [];
|
||||
window.__labQueueCalls = 0;
|
||||
window.__labSubmittedPromptIds = [];
|
||||
window.app.rootGraph = window.app.graph;
|
||||
window.app.processingQueue = false;
|
||||
window.app.queueItems = [];
|
||||
window.app.nextQueueRequestId = 1;
|
||||
window.app.graph.serialize = function () {
|
||||
const data = { nodes: [], extra: {} };
|
||||
this.onSerialize?.(data);
|
||||
return data;
|
||||
};
|
||||
window.app.queuePrompt = async function (number, batchCount = 1) {
|
||||
window.__labQueueCalls += 1;
|
||||
const requestId = this.nextQueueRequestId++;
|
||||
this.queueItems.push({ requestId, number, batchCount });
|
||||
api.dispatchCustomEvent('promptQueueing', { requestId, batchCount });
|
||||
if (this.processingQueue) return false;
|
||||
|
||||
this.processingQueue = true;
|
||||
await Promise.resolve();
|
||||
try {
|
||||
while (this.queueItems.length) {
|
||||
const request = this.queueItems.pop();
|
||||
let queuedCount = 0;
|
||||
for (let index = 0; index < request.batchCount; index += 1) {
|
||||
for (const node of this.graph._nodes) {
|
||||
for (const widget of node.widgets || []) {
|
||||
widget.beforeQueued?.({ isPartialExecution: false });
|
||||
}
|
||||
}
|
||||
const workflow = this.graph.serialize();
|
||||
const marker =
|
||||
workflow.extra?.[receiptMod.PARAMETER_LAB_RECEIPT_KEY];
|
||||
if (!marker?.prompt_id) throw new Error('missing receipt marker');
|
||||
delete workflow.extra[receiptMod.PARAMETER_LAB_RECEIPT_KEY];
|
||||
window.__labSubmittedPromptIds.push(marker.prompt_id);
|
||||
for (const node of this.graph._nodes) {
|
||||
for (const widget of node.widgets || []) {
|
||||
widget.afterQueued?.({ isPartialExecution: false });
|
||||
}
|
||||
}
|
||||
queuedCount += 1;
|
||||
}
|
||||
api.dispatchCustomEvent('promptQueued', {
|
||||
requestId: request.requestId,
|
||||
batchCount: queuedCount,
|
||||
number: request.number
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.processingQueue = false;
|
||||
}
|
||||
return true;
|
||||
};
|
||||
|
||||
const originalFetch = mod.openclawApi.fetch.bind(mod.openclawApi);
|
||||
mod.openclawApi.fetch = async (url, options = {}) => {
|
||||
const normalizedPath = String(url || '').replace(/^\/moltbot/, '/openclaw');
|
||||
if (normalizedPath.endsWith('/lab/sweep')) {
|
||||
return {
|
||||
ok: true,
|
||||
status: 200,
|
||||
data: {
|
||||
plan: {
|
||||
experiment_id: 'exp_receipt',
|
||||
dimensions: [
|
||||
{
|
||||
node_id: 10,
|
||||
widget_name: 'seed',
|
||||
values: [42],
|
||||
strategy: 'grid'
|
||||
}
|
||||
],
|
||||
runs: [{ '10.seed': 42 }]
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
if (normalizedPath.includes('/lab/experiments/exp_receipt/runs/0')) {
|
||||
window.__labRunUpdates.push(JSON.parse(options?.body || '{}'));
|
||||
return { ok: true, status: 200, data: {} };
|
||||
}
|
||||
return originalFetch(url, options);
|
||||
};
|
||||
});
|
||||
|
||||
await page.click('#lab-add-dim');
|
||||
await page.selectOption('.dim-node-select', { value: '10' });
|
||||
await page.selectOption('.dim-widget-select', { value: 'seed' });
|
||||
await page.fill('.dim-manual-input', '42');
|
||||
await page.press('.dim-manual-input', 'Enter');
|
||||
await page.click('#lab-generate');
|
||||
await expect(page.locator('#lab-run-all')).toBeVisible();
|
||||
await page.click('#lab-run-all');
|
||||
|
||||
await expect.poll(() => page.evaluate(() => window.__labQueueCalls)).toBe(1);
|
||||
await expect(page.locator('.openclaw-lab-run-item .run-status')).toContainText('Queued');
|
||||
const updates = await page.evaluate(() => window.__labRunUpdates);
|
||||
expect(updates).toEqual([
|
||||
expect.objectContaining({
|
||||
status: 'queued',
|
||||
output: { prompt_id: expect.any(String) }
|
||||
})
|
||||
]);
|
||||
const submitted = await page.evaluate(() => window.__labSubmittedPromptIds);
|
||||
expect(submitted).toHaveLength(1);
|
||||
expect(updates[0].output.prompt_id).toBe(submitted[0]);
|
||||
|
||||
await page.evaluate(async (promptId) => {
|
||||
const { api } = await import('/scripts/api.js');
|
||||
api.dispatchCustomEvent('execution_start', { prompt_id: promptId });
|
||||
}, submitted[0]);
|
||||
await expect(page.locator('.openclaw-lab-run-item .run-status')).toHaveText('Running');
|
||||
await expect.poll(() => page.evaluate(() => window.__labRunUpdates)).toEqual([
|
||||
expect.objectContaining({ status: 'queued' }),
|
||||
{ status: 'running' }
|
||||
]);
|
||||
|
||||
await page.evaluate(async (promptId) => {
|
||||
const { api } = await import('/scripts/api.js');
|
||||
api.dispatchCustomEvent('execution_success', {
|
||||
prompt_id: '00000000-0000-4000-8000-000000000000'
|
||||
});
|
||||
api.dispatchCustomEvent('execution_success', { prompt_id: promptId });
|
||||
api.dispatchCustomEvent('execution_error', { prompt_id: promptId });
|
||||
}, submitted[0]);
|
||||
await expect(page.locator('.openclaw-lab-run-item .run-status')).toHaveText('Completed');
|
||||
await expect.poll(() => page.evaluate(() => window.__labRunUpdates)).toEqual([
|
||||
expect.objectContaining({ status: 'queued' }),
|
||||
{ status: 'running' },
|
||||
{ status: 'completed' }
|
||||
]);
|
||||
|
||||
await expect(page.locator('.openclaw-banner')).toContainText(
|
||||
'All experiment runs finished.'
|
||||
);
|
||||
|
||||
await page.evaluate(async () => {
|
||||
const { api } = await import('/scripts/api.js');
|
||||
const watchedEvents = new Set([
|
||||
'promptQueueing',
|
||||
'promptQueued',
|
||||
'execution_start',
|
||||
'execution_success',
|
||||
'execution_error',
|
||||
'execution_interrupted'
|
||||
]);
|
||||
const widget = window.app.graph.getNodeById(10).widgets[0];
|
||||
const originalAddEventListener = api.addEventListener.bind(api);
|
||||
const originalRemoveEventListener = api.removeEventListener.bind(api);
|
||||
window.__labDisposeProbe = {
|
||||
listenerBalance: 0,
|
||||
originalBeforeQueued: widget.beforeQueued,
|
||||
originalAfterQueued: widget.afterQueued
|
||||
};
|
||||
api.addEventListener = (type, callback, options) => {
|
||||
if (watchedEvents.has(type)) {
|
||||
window.__labDisposeProbe.listenerBalance += 1;
|
||||
}
|
||||
return originalAddEventListener(type, callback, options);
|
||||
};
|
||||
api.removeEventListener = (type, callback, options) => {
|
||||
if (watchedEvents.has(type)) {
|
||||
window.__labDisposeProbe.listenerBalance -= 1;
|
||||
}
|
||||
return originalRemoveEventListener(type, callback, options);
|
||||
};
|
||||
window.app.processingQueue = false;
|
||||
window.app.queuePrompt = function (_number, batchCount = 1) {
|
||||
const requestId = this.nextQueueRequestId++;
|
||||
api.dispatchCustomEvent('promptQueueing', { requestId, batchCount });
|
||||
this.processingQueue = true;
|
||||
return new Promise(() => {});
|
||||
};
|
||||
});
|
||||
|
||||
await page.click('#lab-run-all');
|
||||
await expect.poll(() => page.evaluate(() => {
|
||||
const widget = window.app.graph.getNodeById(10).widgets[0];
|
||||
const probe = window.__labDisposeProbe;
|
||||
return {
|
||||
callbacksWrapped:
|
||||
widget.beforeQueued !== probe.originalBeforeQueued &&
|
||||
widget.afterQueued !== probe.originalAfterQueued,
|
||||
listenerBalance: probe.listenerBalance
|
||||
};
|
||||
})).toEqual({ callbacksWrapped: true, listenerBalance: 6 });
|
||||
|
||||
await clickTab(page, 'Settings');
|
||||
await expect.poll(() => page.evaluate(() => {
|
||||
const widget = window.app.graph.getNodeById(10).widgets[0];
|
||||
const probe = window.__labDisposeProbe;
|
||||
return {
|
||||
callbacksRestored:
|
||||
widget.beforeQueued === probe.originalBeforeQueued &&
|
||||
widget.afterQueued === probe.originalAfterQueued,
|
||||
listenerBalance: probe.listenerBalance,
|
||||
parameterPaneChildren:
|
||||
document.querySelector('#openclaw-tab-parameter-lab')?.childElementCount
|
||||
};
|
||||
})).toEqual({
|
||||
callbacksRestored: true,
|
||||
listenerBalance: 0,
|
||||
parameterPaneChildren: 0
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -213,15 +213,20 @@ export async function mockComfyUiCore(page, options = {}) {
|
||||
status: 200,
|
||||
contentType: 'application/javascript',
|
||||
body: `
|
||||
export const api = {
|
||||
fetchApi: async (route, options) => {
|
||||
// Prefix with /api if not already present (shim logic simulation)
|
||||
const url = "/api" + route;
|
||||
return fetch(url, options);
|
||||
},
|
||||
apiURL: (route) => "/api" + route,
|
||||
fileURL: (route) => route // Simplified for test
|
||||
};
|
||||
class OpenClawMockComfyApi extends EventTarget {
|
||||
async fetchApi(route, options) {
|
||||
// Prefix with /api if not already present (shim logic simulation)
|
||||
const url = "/api" + route;
|
||||
return fetch(url, options);
|
||||
}
|
||||
apiURL(route) { return "/api" + route; }
|
||||
fileURL(route) { return route; }
|
||||
dispatchCustomEvent(type, detail) {
|
||||
this.dispatchEvent(new CustomEvent(type, { detail }));
|
||||
}
|
||||
}
|
||||
window.__openclawMockComfyApi ||= new OpenClawMockComfyApi();
|
||||
export const api = window.__openclawMockComfyApi;
|
||||
`,
|
||||
});
|
||||
});
|
||||
|
||||
@@ -289,7 +289,7 @@ class RepositoryArchitecturePolicyTests(unittest.TestCase):
|
||||
analysis = dependency_policy.analyze_repository(self.repo_root, policy)
|
||||
|
||||
self.assertEqual(analysis.findings, ())
|
||||
self.assertEqual(len(analysis.owned_paths), 299)
|
||||
self.assertEqual(len(analysis.owned_paths), 300)
|
||||
self.assertEqual(len(policy["accepted_cycles"]), 2)
|
||||
self.assertEqual(len(policy["dynamic_imports"]), 8)
|
||||
self.assertEqual(len(policy["compatibility_exceptions"]), 9)
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
import copy
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
|
||||
from services.parameter_lab_queue_receipt import (
|
||||
PARAMETER_LAB_RECEIPT_KEY,
|
||||
PARAMETER_LAB_RECEIPT_VERSION,
|
||||
consume_parameter_lab_queue_receipt,
|
||||
register_parameter_lab_queue_receipt_handler,
|
||||
)
|
||||
|
||||
|
||||
def _payload(marker):
|
||||
return {
|
||||
"prompt": {"1": {"class_type": "Test", "inputs": {}}},
|
||||
"extra_data": {
|
||||
"extra_pnginfo": {
|
||||
"workflow": {
|
||||
"nodes": [],
|
||||
"extra": {
|
||||
"preserved": {"safe": True},
|
||||
PARAMETER_LAB_RECEIPT_KEY: marker,
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class TestParameterLabQueueReceipt(unittest.TestCase):
|
||||
def test_valid_marker_promotes_native_uuid_and_is_removed_from_metadata(self):
|
||||
prompt_id = "11111111-1111-4111-8111-111111111111"
|
||||
source = _payload(
|
||||
{"version": PARAMETER_LAB_RECEIPT_VERSION, "prompt_id": prompt_id}
|
||||
)
|
||||
result = consume_parameter_lab_queue_receipt(source)
|
||||
|
||||
self.assertEqual(result["prompt_id"], prompt_id)
|
||||
extra = result["extra_data"]["extra_pnginfo"]["workflow"]["extra"]
|
||||
self.assertNotIn(PARAMETER_LAB_RECEIPT_KEY, extra)
|
||||
self.assertEqual(extra["preserved"], {"safe": True})
|
||||
self.assertNotIn(PARAMETER_LAB_RECEIPT_KEY, repr(result))
|
||||
|
||||
def test_invalid_or_conflicting_markers_are_stripped_without_identifier_authority(
|
||||
self,
|
||||
):
|
||||
invalid_markers = (
|
||||
None,
|
||||
"bad",
|
||||
{},
|
||||
{"version": 999, "prompt_id": "11111111-1111-4111-8111-111111111111"},
|
||||
{"version": PARAMETER_LAB_RECEIPT_VERSION, "prompt_id": "not-a-uuid"},
|
||||
{
|
||||
"version": PARAMETER_LAB_RECEIPT_VERSION,
|
||||
"prompt_id": "11111111-1111-4111-8111-111111111111",
|
||||
"extra": True,
|
||||
},
|
||||
)
|
||||
for marker in invalid_markers:
|
||||
with self.subTest(marker=marker):
|
||||
result = consume_parameter_lab_queue_receipt(_payload(marker))
|
||||
self.assertNotIn("prompt_id", result)
|
||||
self.assertNotIn(
|
||||
PARAMETER_LAB_RECEIPT_KEY,
|
||||
result["extra_data"]["extra_pnginfo"]["workflow"]["extra"],
|
||||
)
|
||||
|
||||
source = _payload(
|
||||
{
|
||||
"version": PARAMETER_LAB_RECEIPT_VERSION,
|
||||
"prompt_id": "11111111-1111-4111-8111-111111111111",
|
||||
}
|
||||
)
|
||||
source["prompt_id"] = "22222222-2222-4222-8222-222222222222"
|
||||
result = consume_parameter_lab_queue_receipt(source)
|
||||
# CRITICAL: the transient marker is the exact ID the frontend will own after
|
||||
# promptQueued. Preserving a different earlier handler value would cross-assign.
|
||||
self.assertEqual(result["prompt_id"], "11111111-1111-4111-8111-111111111111")
|
||||
self.assertNotIn(
|
||||
PARAMETER_LAB_RECEIPT_KEY,
|
||||
result["extra_data"]["extra_pnginfo"]["workflow"]["extra"],
|
||||
)
|
||||
|
||||
def test_copy_on_write_preserves_input_and_unrelated_shapes(self):
|
||||
source = _payload(
|
||||
{
|
||||
"version": PARAMETER_LAB_RECEIPT_VERSION,
|
||||
"prompt_id": "33333333-3333-4333-8333-333333333333",
|
||||
}
|
||||
)
|
||||
original = copy.deepcopy(source)
|
||||
result = consume_parameter_lab_queue_receipt(source)
|
||||
|
||||
self.assertEqual(source, original)
|
||||
self.assertIsNot(result, source)
|
||||
self.assertEqual(result["extra_data"]["extra_pnginfo"]["workflow"]["nodes"], [])
|
||||
untouched = {"prompt": {}}
|
||||
self.assertIs(consume_parameter_lab_queue_receipt(untouched), untouched)
|
||||
|
||||
def test_registration_is_idempotent_and_uses_official_host_handler(self):
|
||||
handlers = []
|
||||
server = SimpleNamespace(
|
||||
on_prompt_handlers=handlers,
|
||||
add_on_prompt_handler=handlers.append,
|
||||
)
|
||||
|
||||
self.assertTrue(register_parameter_lab_queue_receipt_handler(server))
|
||||
self.assertFalse(register_parameter_lab_queue_receipt_handler(server))
|
||||
self.assertEqual(len(handlers), 1)
|
||||
promoted = handlers[0](
|
||||
_payload(
|
||||
{
|
||||
"version": PARAMETER_LAB_RECEIPT_VERSION,
|
||||
"prompt_id": "44444444-4444-4444-8444-444444444444",
|
||||
}
|
||||
)
|
||||
)
|
||||
self.assertEqual(promoted["prompt_id"], "44444444-4444-4444-8444-444444444444")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -151,13 +151,19 @@ class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
|
||||
release.wait(timeout=1.0)
|
||||
|
||||
app = SimpleNamespace(router=_DummyRouter())
|
||||
server = SimpleNamespace(routes=_DummyRoutes(), app=app)
|
||||
prompt_handlers = []
|
||||
server = SimpleNamespace(
|
||||
routes=_DummyRoutes(),
|
||||
app=app,
|
||||
on_prompt_handlers=prompt_handlers,
|
||||
add_on_prompt_handler=prompt_handlers.append,
|
||||
)
|
||||
|
||||
contract = {
|
||||
"register_routes": lambda server: setattr(server, "core_routes", True),
|
||||
"register_preset_routes": lambda app: setattr(app, "presets", True),
|
||||
"register_schedule_routes": lambda app, require_admin_token_fn=None: setattr(
|
||||
app, "schedules", True
|
||||
"register_schedule_routes": lambda app, require_admin_token_fn=None: (
|
||||
setattr(app, "schedules", True)
|
||||
),
|
||||
"BridgeHandlers": _DummyBridgeHandlers,
|
||||
"register_trigger_routes": lambda app, **kwargs: setattr(
|
||||
@@ -194,6 +200,7 @@ class TestRouteBootstrapWarmupBoundary(unittest.TestCase):
|
||||
self.assertTrue(server.core_routes)
|
||||
self.assertTrue(app.triggers)
|
||||
self.assertTrue(app.approvals)
|
||||
self.assertEqual(len(prompt_handlers), 1)
|
||||
self.assertTrue(diagnostics["ready"])
|
||||
warmup = next(
|
||||
item for item in diagnostics["warmups"] if item["name"] == "slow_provider"
|
||||
|
||||
@@ -0,0 +1,617 @@
|
||||
/**
|
||||
* Exact Parameter Lab queue receipts over ComfyUI's boolean app.queuePrompt API.
|
||||
*
|
||||
* The coordinator observes the host request boundary, mirrors its reviewed LIFO
|
||||
* selection, and adds a transient UUID only to the matching serialized workflow.
|
||||
*/
|
||||
|
||||
export const PARAMETER_LAB_RECEIPT_KEY = "__openclaw_parameter_lab_receipt__";
|
||||
export const PARAMETER_LAB_RECEIPT_VERSION = 1;
|
||||
|
||||
const MAX_ACTIVE_ATTEMPTS = 64;
|
||||
const MAX_TRACKED_REQUESTS = 128;
|
||||
const MAX_BUFFERED_LIFECYCLE_EVENTS = 8;
|
||||
const DEFAULT_TIMEOUT_MS = 30_000;
|
||||
const UUID_RE =
|
||||
/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
|
||||
const TERMINAL_EVENTS = new Set([
|
||||
"execution_success",
|
||||
"execution_error",
|
||||
"execution_interrupted",
|
||||
]);
|
||||
const LIFECYCLE_EVENTS = [
|
||||
"execution_start",
|
||||
"execution_success",
|
||||
"execution_error",
|
||||
"execution_interrupted",
|
||||
];
|
||||
|
||||
export class ParameterLabReceiptError extends Error {
|
||||
constructor(code) {
|
||||
super(code);
|
||||
this.name = "ParameterLabReceiptError";
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
function receiptError(code) {
|
||||
return new ParameterLabReceiptError(code);
|
||||
}
|
||||
|
||||
function isPositiveInteger(value) {
|
||||
return Number.isSafeInteger(value) && value > 0;
|
||||
}
|
||||
|
||||
function normalizePromptId(value) {
|
||||
return typeof value === "string" && UUID_RE.test(value) ? value : "";
|
||||
}
|
||||
|
||||
function safeCall(callback, thisArg, args) {
|
||||
if (typeof callback !== "function") return undefined;
|
||||
return callback.apply(thisArg, args);
|
||||
}
|
||||
|
||||
class ParameterLabReceiptCoordinator {
|
||||
constructor({
|
||||
app,
|
||||
api,
|
||||
uuidFactory = () => globalThis.crypto.randomUUID(),
|
||||
timeoutMs = DEFAULT_TIMEOUT_MS,
|
||||
}) {
|
||||
if (!app || typeof app.queuePrompt !== "function") {
|
||||
throw receiptError("unsupported_host_queue");
|
||||
}
|
||||
if (
|
||||
!api ||
|
||||
typeof api.addEventListener !== "function" ||
|
||||
typeof api.removeEventListener !== "function"
|
||||
) {
|
||||
throw receiptError("unsupported_host_events");
|
||||
}
|
||||
if (typeof uuidFactory !== "function") {
|
||||
throw receiptError("invalid_uuid_factory");
|
||||
}
|
||||
if (!isPositiveInteger(timeoutMs)) {
|
||||
throw receiptError("invalid_receipt_timeout");
|
||||
}
|
||||
|
||||
this.app = app;
|
||||
this.api = api;
|
||||
this.uuidFactory = uuidFactory;
|
||||
this.timeoutMs = timeoutMs;
|
||||
this.disposed = false;
|
||||
this.installed = false;
|
||||
this.hostWindowOwned = false;
|
||||
this.requestTrackingInvalid = false;
|
||||
this.captureAttempt = null;
|
||||
this.nextAttemptId = 1;
|
||||
this.attempts = new Map();
|
||||
this.attemptsByPromptId = new Map();
|
||||
this.requests = new Map();
|
||||
this.pendingRequests = [];
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
this.hookWidget = null;
|
||||
this.originalBeforeQueued = undefined;
|
||||
this.originalAfterQueued = undefined;
|
||||
this.wrappedBeforeQueued = null;
|
||||
this.wrappedAfterQueued = null;
|
||||
|
||||
this.onPromptQueueing = (event) => this._handlePromptQueueing(event);
|
||||
this.onPromptQueued = (event) => this._handlePromptQueued(event);
|
||||
this.lifecycleHandlers = new Map(
|
||||
LIFECYCLE_EVENTS.map((type) => [
|
||||
type,
|
||||
(event) => this._handleLifecycle(type, event),
|
||||
]),
|
||||
);
|
||||
}
|
||||
|
||||
debugSnapshot() {
|
||||
let lifecycleSubscriptions = 0;
|
||||
for (const attempt of this.attempts.values()) {
|
||||
if (typeof attempt.lifecycleCallback === "function") {
|
||||
lifecycleSubscriptions += 1;
|
||||
}
|
||||
}
|
||||
return {
|
||||
activeAttempts: this.attempts.size,
|
||||
pendingRequests: this.requests.size,
|
||||
lifecycleSubscriptions,
|
||||
installed: this.installed,
|
||||
};
|
||||
}
|
||||
|
||||
queue({ experimentId, runId, widget, signal } = {}) {
|
||||
if (this.disposed) {
|
||||
return Promise.reject(receiptError("coordinator_disposed"));
|
||||
}
|
||||
if (signal?.aborted) {
|
||||
return Promise.reject(receiptError("attempt_cancelled"));
|
||||
}
|
||||
if (!widget || typeof widget !== "object") {
|
||||
return Promise.reject(receiptError("receipt_widget_required"));
|
||||
}
|
||||
if (this.attempts.size >= MAX_ACTIVE_ATTEMPTS) {
|
||||
return Promise.reject(receiptError("too_many_receipt_attempts"));
|
||||
}
|
||||
if (this.requestTrackingInvalid) {
|
||||
return Promise.reject(receiptError("request_tracking_unavailable"));
|
||||
}
|
||||
|
||||
const promptId = normalizePromptId(this.uuidFactory());
|
||||
if (!promptId) {
|
||||
return Promise.reject(receiptError("invalid_receipt_id"));
|
||||
}
|
||||
if (this.attemptsByPromptId.has(promptId)) {
|
||||
return Promise.reject(receiptError("duplicate_receipt_id"));
|
||||
}
|
||||
|
||||
const ownsNewWindow = !this.hostWindowOwned;
|
||||
if (ownsNewWindow && this.app.processingQueue !== false) {
|
||||
return Promise.reject(receiptError("host_queue_unobserved_busy"));
|
||||
}
|
||||
if (this.captureAttempt) {
|
||||
return Promise.reject(receiptError("request_capture_busy"));
|
||||
}
|
||||
|
||||
try {
|
||||
this._install(widget);
|
||||
} catch (error) {
|
||||
return Promise.reject(error);
|
||||
}
|
||||
|
||||
const attempt = this._createAttempt({
|
||||
experimentId,
|
||||
runId,
|
||||
promptId,
|
||||
signal,
|
||||
});
|
||||
if (ownsNewWindow) this.hostWindowOwned = true;
|
||||
this.captureAttempt = attempt;
|
||||
|
||||
let hostResult;
|
||||
try {
|
||||
hostResult = this.app.queuePrompt(0, 1);
|
||||
} catch (_error) {
|
||||
this.captureAttempt = null;
|
||||
this._failAttempt(attempt, "host_queue_failed");
|
||||
if (ownsNewWindow) this._finishHostWindow("host_queue_failed");
|
||||
return attempt.promise;
|
||||
}
|
||||
this.captureAttempt = null;
|
||||
|
||||
if (attempt.requestId === null) {
|
||||
this._failAttempt(attempt, "missing_request_boundary");
|
||||
}
|
||||
|
||||
const hostPromise = Promise.resolve(hostResult);
|
||||
if (ownsNewWindow) {
|
||||
hostPromise.then(
|
||||
() => this._finishHostWindow(),
|
||||
() => this._finishHostWindow("host_queue_failed"),
|
||||
);
|
||||
} else {
|
||||
hostPromise.catch(() => this._failAttempt(attempt, "host_queue_failed"));
|
||||
}
|
||||
return attempt.promise;
|
||||
}
|
||||
|
||||
dispose() {
|
||||
if (this.disposed) return;
|
||||
this.disposed = true;
|
||||
for (const attempt of [...this.attempts.values()]) {
|
||||
this._failAttempt(attempt, "coordinator_disposed");
|
||||
}
|
||||
this.requests.clear();
|
||||
this.pendingRequests.length = 0;
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
this.captureAttempt = null;
|
||||
this.hostWindowOwned = false;
|
||||
this._uninstall();
|
||||
}
|
||||
|
||||
_createAttempt({ experimentId, runId, promptId, signal }) {
|
||||
const id = this.nextAttemptId++;
|
||||
const attempt = {
|
||||
id,
|
||||
experimentId: String(experimentId ?? ""),
|
||||
runId: String(runId ?? ""),
|
||||
promptId,
|
||||
requestId: null,
|
||||
state: "pending",
|
||||
serializedCount: 0,
|
||||
lifecycleCallback: null,
|
||||
lifecycleBuffer: [],
|
||||
signal,
|
||||
abortHandler: null,
|
||||
timer: null,
|
||||
receipt: null,
|
||||
settled: false,
|
||||
resolve: null,
|
||||
reject: null,
|
||||
promise: null,
|
||||
};
|
||||
attempt.promise = new Promise((resolve, reject) => {
|
||||
attempt.resolve = resolve;
|
||||
attempt.reject = reject;
|
||||
});
|
||||
attempt.timer = setTimeout(
|
||||
() => this._failAttempt(attempt, "receipt_timeout"),
|
||||
this.timeoutMs,
|
||||
);
|
||||
if (signal) {
|
||||
attempt.abortHandler = () =>
|
||||
this._failAttempt(attempt, "attempt_cancelled");
|
||||
signal.addEventListener("abort", attempt.abortHandler, { once: true });
|
||||
}
|
||||
this.attempts.set(id, attempt);
|
||||
this.attemptsByPromptId.set(promptId, attempt);
|
||||
return attempt;
|
||||
}
|
||||
|
||||
_install(widget) {
|
||||
if (this.installed) {
|
||||
if (widget !== this.hookWidget) {
|
||||
throw receiptError("receipt_widget_conflict");
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
this.hookWidget = widget;
|
||||
this.originalBeforeQueued = widget.beforeQueued;
|
||||
this.originalAfterQueued = widget.afterQueued;
|
||||
const coordinator = this;
|
||||
this.wrappedBeforeQueued = function (...args) {
|
||||
const result = safeCall(coordinator.originalBeforeQueued, this, args);
|
||||
coordinator._handleBeforeQueued();
|
||||
return result;
|
||||
};
|
||||
this.wrappedAfterQueued = function (...args) {
|
||||
const result = safeCall(coordinator.originalAfterQueued, this, args);
|
||||
coordinator._handleAfterQueued();
|
||||
return result;
|
||||
};
|
||||
widget.beforeQueued = this.wrappedBeforeQueued;
|
||||
widget.afterQueued = this.wrappedAfterQueued;
|
||||
|
||||
this.api.addEventListener("promptQueueing", this.onPromptQueueing);
|
||||
this.api.addEventListener("promptQueued", this.onPromptQueued);
|
||||
for (const [type, handler] of this.lifecycleHandlers) {
|
||||
this.api.addEventListener(type, handler);
|
||||
}
|
||||
this.installed = true;
|
||||
}
|
||||
|
||||
_uninstall() {
|
||||
if (!this.installed) return;
|
||||
if (this.hookWidget?.beforeQueued === this.wrappedBeforeQueued) {
|
||||
this.hookWidget.beforeQueued = this.originalBeforeQueued;
|
||||
}
|
||||
if (this.hookWidget?.afterQueued === this.wrappedAfterQueued) {
|
||||
this.hookWidget.afterQueued = this.originalAfterQueued;
|
||||
}
|
||||
this.api.removeEventListener("promptQueueing", this.onPromptQueueing);
|
||||
this.api.removeEventListener("promptQueued", this.onPromptQueued);
|
||||
for (const [type, handler] of this.lifecycleHandlers) {
|
||||
this.api.removeEventListener(type, handler);
|
||||
}
|
||||
this.hookWidget = null;
|
||||
this.wrappedBeforeQueued = null;
|
||||
this.wrappedAfterQueued = null;
|
||||
this.originalBeforeQueued = undefined;
|
||||
this.originalAfterQueued = undefined;
|
||||
this.installed = false;
|
||||
}
|
||||
|
||||
_handlePromptQueueing(event) {
|
||||
if (this.requestTrackingInvalid) return;
|
||||
const detail = event?.detail;
|
||||
const requestId = detail?.requestId;
|
||||
const batchCount = detail?.batchCount;
|
||||
if (
|
||||
!Number.isSafeInteger(requestId) ||
|
||||
requestId < 0 ||
|
||||
!isPositiveInteger(batchCount) ||
|
||||
this.requests.has(requestId)
|
||||
) {
|
||||
if (this.captureAttempt) {
|
||||
this._failAttempt(this.captureAttempt, "invalid_request_boundary");
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (this.requests.size >= MAX_TRACKED_REQUESTS) {
|
||||
this._invalidateRequestTracking();
|
||||
return;
|
||||
}
|
||||
|
||||
const attempt = this.captureAttempt;
|
||||
if (attempt) {
|
||||
if (attempt.requestId !== null) {
|
||||
this._failAttempt(attempt, "duplicate_request_boundary");
|
||||
return;
|
||||
}
|
||||
attempt.requestId = requestId;
|
||||
}
|
||||
const request = {
|
||||
requestId,
|
||||
batchCount,
|
||||
remaining: batchCount,
|
||||
successful: 0,
|
||||
attempt,
|
||||
};
|
||||
this.requests.set(requestId, request);
|
||||
this.pendingRequests.push(request);
|
||||
}
|
||||
|
||||
_handleBeforeQueued() {
|
||||
if (this.batchInFlight) {
|
||||
this._failRequest(this.batchInFlight, "request_batch_incomplete");
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
}
|
||||
if (!this.currentRequest) {
|
||||
this.currentRequest = this.pendingRequests.pop() ?? null;
|
||||
}
|
||||
const request = this.currentRequest;
|
||||
if (!request) return;
|
||||
this.batchInFlight = request;
|
||||
if (request.attempt?.state === "pending") {
|
||||
this._armSerialization(request.attempt);
|
||||
}
|
||||
}
|
||||
|
||||
_handleAfterQueued() {
|
||||
const request = this.batchInFlight;
|
||||
if (!request) return;
|
||||
this.batchInFlight = null;
|
||||
request.successful += 1;
|
||||
request.remaining -= 1;
|
||||
if (
|
||||
request.attempt?.state === "pending" &&
|
||||
request.attempt.serializedCount !== request.successful
|
||||
) {
|
||||
this._failAttempt(request.attempt, "receipt_not_serialized");
|
||||
}
|
||||
}
|
||||
|
||||
_armSerialization(attempt) {
|
||||
const coordinator = this;
|
||||
const graph = this.app.rootGraph ?? this.app.graph;
|
||||
if (!graph || typeof graph.serialize !== "function") {
|
||||
this._failAttempt(attempt, "unsupported_graph_serialization");
|
||||
return;
|
||||
}
|
||||
const previous = graph.onSerialize;
|
||||
if (previous !== undefined && typeof previous !== "function") {
|
||||
this._failAttempt(attempt, "unsupported_graph_callback");
|
||||
return;
|
||||
}
|
||||
|
||||
let armed = true;
|
||||
const wrapper = function (data) {
|
||||
if (!armed) return;
|
||||
armed = false;
|
||||
if (graph.onSerialize === wrapper) graph.onSerialize = previous;
|
||||
safeCall(previous, this, [data]);
|
||||
if (
|
||||
!data ||
|
||||
typeof data !== "object" ||
|
||||
!data.extra ||
|
||||
typeof data.extra !== "object" ||
|
||||
Object.prototype.hasOwnProperty.call(
|
||||
data.extra,
|
||||
PARAMETER_LAB_RECEIPT_KEY,
|
||||
)
|
||||
) {
|
||||
coordinator._failAttempt(attempt, "receipt_marker_collision");
|
||||
throw receiptError("receipt_marker_collision");
|
||||
}
|
||||
data.extra[PARAMETER_LAB_RECEIPT_KEY] = {
|
||||
version: PARAMETER_LAB_RECEIPT_VERSION,
|
||||
prompt_id: attempt.promptId,
|
||||
};
|
||||
attempt.serializedCount += 1;
|
||||
};
|
||||
graph.onSerialize = wrapper;
|
||||
queueMicrotask(() => {
|
||||
if (!armed) return;
|
||||
armed = false;
|
||||
if (graph.onSerialize === wrapper) graph.onSerialize = previous;
|
||||
});
|
||||
}
|
||||
|
||||
_handlePromptQueued(event) {
|
||||
const detail = event?.detail;
|
||||
const requestId = detail?.requestId;
|
||||
const request = this.requests.get(requestId);
|
||||
if (!request) return;
|
||||
if (
|
||||
request !== this.currentRequest ||
|
||||
request.remaining !== 0 ||
|
||||
detail?.batchCount !== request.successful
|
||||
) {
|
||||
this._failRequest(request, "request_boundary_mismatch");
|
||||
return;
|
||||
}
|
||||
|
||||
this.requests.delete(requestId);
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
const attempt = request.attempt;
|
||||
if (!attempt || attempt.state !== "pending") return;
|
||||
if (
|
||||
request.batchCount !== 1 ||
|
||||
request.successful !== 1 ||
|
||||
attempt.serializedCount !== 1
|
||||
) {
|
||||
this._failAttempt(attempt, "receipt_count_mismatch");
|
||||
return;
|
||||
}
|
||||
this._acceptAttempt(attempt);
|
||||
}
|
||||
|
||||
_acceptAttempt(attempt) {
|
||||
attempt.state = "accepted";
|
||||
clearTimeout(attempt.timer);
|
||||
if (attempt.signal && attempt.abortHandler) {
|
||||
attempt.signal.removeEventListener("abort", attempt.abortHandler);
|
||||
}
|
||||
const coordinator = this;
|
||||
const receipt = Object.freeze({
|
||||
promptId: attempt.promptId,
|
||||
requestId: attempt.requestId,
|
||||
subscribeLifecycle(callback) {
|
||||
if (typeof callback !== "function") {
|
||||
throw receiptError("invalid_lifecycle_callback");
|
||||
}
|
||||
if (!coordinator.attempts.has(attempt.id)) return () => {};
|
||||
attempt.lifecycleCallback = callback;
|
||||
coordinator._flushLifecycle(attempt);
|
||||
return () => {
|
||||
if (attempt.lifecycleCallback === callback) {
|
||||
attempt.lifecycleCallback = null;
|
||||
}
|
||||
};
|
||||
},
|
||||
release() {
|
||||
coordinator._releaseAttempt(attempt);
|
||||
},
|
||||
});
|
||||
attempt.receipt = receipt;
|
||||
if (!this.hostWindowOwned) {
|
||||
attempt.settled = true;
|
||||
attempt.resolve(receipt);
|
||||
}
|
||||
}
|
||||
|
||||
_handleLifecycle(type, event) {
|
||||
const promptId = event?.detail?.prompt_id;
|
||||
if (typeof promptId !== "string") return;
|
||||
const attempt = this.attemptsByPromptId.get(promptId);
|
||||
if (!attempt || attempt.state === "failed") return;
|
||||
if (attempt.lifecycleBuffer.length >= MAX_BUFFERED_LIFECYCLE_EVENTS) {
|
||||
this._failAttempt(attempt, "lifecycle_buffer_exceeded");
|
||||
return;
|
||||
}
|
||||
// CRITICAL: host lifecycle payloads can contain node/error content.
|
||||
// The receipt boundary needs only the opaque prompt ID and event type.
|
||||
attempt.lifecycleBuffer.push(Object.freeze({ type, promptId }));
|
||||
if (attempt.state === "accepted" && attempt.lifecycleCallback) {
|
||||
this._flushLifecycle(attempt);
|
||||
}
|
||||
}
|
||||
|
||||
_flushLifecycle(attempt) {
|
||||
while (
|
||||
this.attempts.has(attempt.id) &&
|
||||
attempt.lifecycleCallback &&
|
||||
attempt.lifecycleBuffer.length
|
||||
) {
|
||||
const event = attempt.lifecycleBuffer.shift();
|
||||
try {
|
||||
attempt.lifecycleCallback(event);
|
||||
} catch (_error) {
|
||||
// CRITICAL: an internal UI consumer failure must not retain prompt ownership
|
||||
// or expose its potentially private exception detail through host dispatch.
|
||||
this._releaseAttempt(attempt);
|
||||
return;
|
||||
}
|
||||
if (TERMINAL_EVENTS.has(event.type)) {
|
||||
this._releaseAttempt(attempt);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
_failRequest(request, code) {
|
||||
this.requests.delete(request.requestId);
|
||||
const pendingIndex = this.pendingRequests.indexOf(request);
|
||||
if (pendingIndex >= 0) this.pendingRequests.splice(pendingIndex, 1);
|
||||
if (request.attempt) this._failAttempt(request.attempt, code);
|
||||
}
|
||||
|
||||
_invalidateRequestTracking() {
|
||||
// CRITICAL: dropping one LIFO boundary while retaining others could cross-assign
|
||||
// a later receipt, so overflow invalidates the whole request-correlation window.
|
||||
this.requestTrackingInvalid = true;
|
||||
const pendingAttempts = [...this.attempts.values()].filter(
|
||||
(attempt) => attempt.state === "pending",
|
||||
);
|
||||
this.requests.clear();
|
||||
this.pendingRequests.length = 0;
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
for (const attempt of pendingAttempts) {
|
||||
this._failAttempt(attempt, "request_boundary_overflow");
|
||||
}
|
||||
this._maybeUninstall();
|
||||
}
|
||||
|
||||
_failAttempt(attempt, code) {
|
||||
if (!attempt || !this.attempts.has(attempt.id)) return;
|
||||
attempt.state = "failed";
|
||||
clearTimeout(attempt.timer);
|
||||
if (attempt.signal && attempt.abortHandler) {
|
||||
attempt.signal.removeEventListener("abort", attempt.abortHandler);
|
||||
}
|
||||
this.attempts.delete(attempt.id);
|
||||
this.attemptsByPromptId.delete(attempt.promptId);
|
||||
attempt.reject(receiptError(code));
|
||||
this._maybeUninstall();
|
||||
}
|
||||
|
||||
_releaseAttempt(attempt) {
|
||||
if (!attempt || !this.attempts.has(attempt.id)) return;
|
||||
clearTimeout(attempt.timer);
|
||||
if (attempt.signal && attempt.abortHandler) {
|
||||
attempt.signal.removeEventListener("abort", attempt.abortHandler);
|
||||
}
|
||||
this.attempts.delete(attempt.id);
|
||||
this.attemptsByPromptId.delete(attempt.promptId);
|
||||
attempt.lifecycleCallback = null;
|
||||
attempt.lifecycleBuffer.length = 0;
|
||||
this._maybeUninstall();
|
||||
}
|
||||
|
||||
_finishHostWindow(failureCode = "") {
|
||||
this.hostWindowOwned = false;
|
||||
if (failureCode) {
|
||||
for (const request of [...this.requests.values()]) {
|
||||
if (request.attempt) this._failAttempt(request.attempt, failureCode);
|
||||
}
|
||||
} else {
|
||||
for (const request of [...this.requests.values()]) {
|
||||
if (request.attempt) {
|
||||
this._failAttempt(request.attempt, "missing_queued_boundary");
|
||||
}
|
||||
}
|
||||
}
|
||||
this.requests.clear();
|
||||
this.pendingRequests.length = 0;
|
||||
this.currentRequest = null;
|
||||
this.batchInFlight = null;
|
||||
for (const attempt of this.attempts.values()) {
|
||||
if (attempt.state === "accepted" && attempt.receipt && !attempt.settled) {
|
||||
attempt.settled = true;
|
||||
attempt.resolve(attempt.receipt);
|
||||
}
|
||||
}
|
||||
this._maybeUninstall();
|
||||
}
|
||||
|
||||
_maybeUninstall() {
|
||||
if (
|
||||
!this.hostWindowOwned &&
|
||||
this.attempts.size === 0 &&
|
||||
this.requests.size === 0
|
||||
) {
|
||||
this._uninstall();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function createParameterLabReceiptCoordinator(options) {
|
||||
return new ParameterLabReceiptCoordinator(options);
|
||||
}
|
||||
+154
-56
@@ -1,6 +1,7 @@
|
||||
// CRITICAL: this tab module is loaded under /extensions/<pack>/web/tabs/*.js.
|
||||
// Must resolve ComfyUI core app from /scripts/app.js via ../../../ prefix.
|
||||
import { app } from "../../../scripts/app.js";
|
||||
import { api } from "../../../scripts/api.js";
|
||||
import { openclawApi } from "../openclaw_api.js";
|
||||
import {
|
||||
findComparableWidget,
|
||||
@@ -17,6 +18,7 @@ import {
|
||||
validateParameterLabScalar,
|
||||
validateParameterLabWorkflow,
|
||||
} from "../openclaw_parameter_lab_policy.js";
|
||||
import { createParameterLabReceiptCoordinator } from "../openclaw_parameter_lab_receipt.js";
|
||||
import { openclawUI } from "../openclaw_ui.js";
|
||||
|
||||
/**
|
||||
@@ -37,8 +39,14 @@ export const ParameterLabTab = {
|
||||
experimentId: null,
|
||||
isRunning: false,
|
||||
results: [],
|
||||
_receiptCoordinator: null,
|
||||
_runController: null,
|
||||
_activePromptIds: null,
|
||||
_queuedRunCount: 0,
|
||||
_queueingComplete: false,
|
||||
|
||||
render(container) {
|
||||
this._disposeRunRuntime();
|
||||
container.innerHTML = "";
|
||||
container.className = "openclaw-tab-content openclaw-tab-content moltbot-tab-content openclaw-lab-container openclaw-lab-container moltbot-lab-container";
|
||||
|
||||
@@ -146,6 +154,27 @@ export const ParameterLabTab = {
|
||||
}
|
||||
},
|
||||
|
||||
dispose() {
|
||||
const hadRuntime = Boolean(
|
||||
this._receiptCoordinator || this._runController || this.es
|
||||
);
|
||||
this._disposeRunRuntime();
|
||||
return hadRuntime;
|
||||
},
|
||||
|
||||
_disposeRunRuntime() {
|
||||
this._runController?.abort();
|
||||
this._runController = null;
|
||||
this._receiptCoordinator?.dispose();
|
||||
this._receiptCoordinator = null;
|
||||
this._activePromptIds = null;
|
||||
this._queuedRunCount = 0;
|
||||
this._queueingComplete = false;
|
||||
this.es?.close?.();
|
||||
this.es = null;
|
||||
this.isRunning = false;
|
||||
},
|
||||
|
||||
async showHistory() {
|
||||
this.resultsContainer.innerHTML = "<div class='openclaw-loading openclaw-loading moltbot-loading'>Loading history...</div>";
|
||||
try {
|
||||
@@ -712,48 +741,33 @@ export const ParameterLabTab = {
|
||||
|
||||
async runExperiment() {
|
||||
if (this.isRunning) return;
|
||||
this._disposeRunRuntime();
|
||||
this.isRunning = true;
|
||||
openclawUI.showBanner("info", "Starting experiment...");
|
||||
|
||||
const items = this.resultsContainer.querySelectorAll(".openclaw-lab-run-item");
|
||||
|
||||
// Subscribe to events for status updates
|
||||
const es = openclawApi.subscribeEvents((data) => {
|
||||
if (!this.isRunning) return; // Note: we might want to keep listening even after queuing finishes
|
||||
const pid = data.prompt_id;
|
||||
if (!pid) return;
|
||||
|
||||
// Find run with this prompt_id
|
||||
const runIdx = this.plan.runs.findIndex(r => r.prompt_id === pid);
|
||||
if (runIdx !== -1) {
|
||||
const item = items[runIdx];
|
||||
const statusSpan = item.querySelector(".run-status");
|
||||
|
||||
if (data.event_type === "execution_success" || data.event_type === "completed") {
|
||||
statusSpan.className = "run-status success";
|
||||
statusSpan.textContent = "Completed";
|
||||
// Update backend
|
||||
openclawApi.fetch(openclawApi._path(`/lab/experiments/${this.experimentId}/runs/${runIdx}`), {
|
||||
method: "POST", body: JSON.stringify({ status: "completed" })
|
||||
});
|
||||
} else if (data.event_type === "execution_error" || data.event_type === "failed") {
|
||||
statusSpan.className = "run-status error";
|
||||
statusSpan.textContent = "Failed";
|
||||
openclawApi.fetch(openclawApi._path(`/lab/experiments/${this.experimentId}/runs/${runIdx}`), {
|
||||
method: "POST", body: JSON.stringify({ status: "failed" })
|
||||
});
|
||||
} else if (data.event_type === "executing") {
|
||||
statusSpan.className = "run-status running";
|
||||
statusSpan.textContent = "Executing Node " + data.node;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
this.es = es;
|
||||
this._runController = new AbortController();
|
||||
this._activePromptIds = new Set();
|
||||
this._queuedRunCount = 0;
|
||||
this._queueingComplete = false;
|
||||
try {
|
||||
this._receiptCoordinator = createParameterLabReceiptCoordinator({
|
||||
app,
|
||||
api,
|
||||
});
|
||||
} catch (_error) {
|
||||
this.isRunning = false;
|
||||
openclawUI.showBanner(
|
||||
"error",
|
||||
"Parameter Lab queue receipt is unavailable."
|
||||
);
|
||||
return;
|
||||
}
|
||||
const signal = this._runController.signal;
|
||||
|
||||
try {
|
||||
for (let i = 0; i < this.plan.runs.length; i++) {
|
||||
// If user stops? (TODO: Add stop button)
|
||||
if (signal.aborted) break;
|
||||
|
||||
const run = this.plan.runs[i];
|
||||
const item = items[i];
|
||||
@@ -764,35 +778,116 @@ export const ParameterLabTab = {
|
||||
|
||||
try {
|
||||
// 1. Apply overrides
|
||||
this.applyOverrides(run);
|
||||
|
||||
// 2. Queue Prompt & Capture ID
|
||||
const res = await app.queuePrompt(0, 1);
|
||||
|
||||
if (res && res.prompt_id) {
|
||||
run.prompt_id = res.prompt_id;
|
||||
statusSpan.textContent = "Queued (" + res.prompt_id.slice(0, 4) + ")";
|
||||
|
||||
// Register with backend
|
||||
openclawApi.fetch(openclawApi._path(`/lab/experiments/${this.experimentId}/runs/${i}`), {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ status: "queued", output: { prompt_id: res.prompt_id } })
|
||||
});
|
||||
} else {
|
||||
throw new Error("No prompt_id returned");
|
||||
const receiptWidget = this.applyOverrides(run);
|
||||
if (!receiptWidget) {
|
||||
throw new Error("receipt_widget_required");
|
||||
}
|
||||
|
||||
} catch (e) {
|
||||
// 2. Queue through the host-owned path and bind its exact receipt.
|
||||
const receipt = await this._receiptCoordinator.queue({
|
||||
experimentId: this.experimentId,
|
||||
runId: String(i),
|
||||
widget: receiptWidget,
|
||||
signal,
|
||||
});
|
||||
if (signal.aborted) {
|
||||
receipt.release();
|
||||
break;
|
||||
}
|
||||
|
||||
run.prompt_id = receipt.promptId;
|
||||
this._activePromptIds.add(receipt.promptId);
|
||||
this._queuedRunCount += 1;
|
||||
statusSpan.textContent =
|
||||
"Queued (" + receipt.promptId.slice(0, 4) + ")";
|
||||
await this._updateRun(i, {
|
||||
status: "queued",
|
||||
output: { prompt_id: receipt.promptId },
|
||||
});
|
||||
|
||||
receipt.subscribeLifecycle((event) => {
|
||||
if (signal.aborted) return;
|
||||
this._handleRunLifecycle({
|
||||
event,
|
||||
runIndex: i,
|
||||
statusSpan,
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
if (signal.aborted) break;
|
||||
statusSpan.className = "run-status error";
|
||||
statusSpan.textContent = "Queue Failed";
|
||||
console.error(e);
|
||||
await this._updateRun(i, { status: "failed" });
|
||||
console.error("[OpenClaw] Parameter Lab queue failed", {
|
||||
code: error?.code || "queue_failed",
|
||||
});
|
||||
}
|
||||
|
||||
await new Promise(r => setTimeout(r, 1000));
|
||||
await new Promise(resolve => setTimeout(resolve, 1000));
|
||||
}
|
||||
} finally {
|
||||
// Keep monitoring
|
||||
openclawUI.showBanner("success", "All runs queued. Monitoring progress...");
|
||||
this._queueingComplete = true;
|
||||
if (!signal.aborted) {
|
||||
if (this._queuedRunCount > 0 && this._activePromptIds.size > 0) {
|
||||
openclawUI.showBanner(
|
||||
"success",
|
||||
"All runs queued. Monitoring progress..."
|
||||
);
|
||||
} else if (this._queuedRunCount > 0) {
|
||||
this.isRunning = false;
|
||||
openclawUI.showBanner(
|
||||
"success",
|
||||
"All experiment runs finished."
|
||||
);
|
||||
} else {
|
||||
this.isRunning = false;
|
||||
openclawUI.showBanner(
|
||||
"error",
|
||||
"No experiment runs were queued."
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
async _updateRun(runIndex, payload) {
|
||||
try {
|
||||
await openclawApi.fetch(
|
||||
openclawApi._path(
|
||||
`/lab/experiments/${this.experimentId}/runs/${runIndex}`
|
||||
),
|
||||
{
|
||||
method: "POST",
|
||||
body: JSON.stringify(payload),
|
||||
}
|
||||
);
|
||||
} catch (_error) {
|
||||
console.warn("[OpenClaw] Parameter Lab run update failed");
|
||||
}
|
||||
},
|
||||
|
||||
_handleRunLifecycle({ event, runIndex, statusSpan }) {
|
||||
if (!this._activePromptIds?.has(event.promptId)) return;
|
||||
if (event.type === "execution_start") {
|
||||
statusSpan.className = "run-status running";
|
||||
statusSpan.textContent = "Running";
|
||||
void this._updateRun(runIndex, { status: "running" });
|
||||
return;
|
||||
}
|
||||
|
||||
const succeeded = event.type === "execution_success";
|
||||
statusSpan.className = succeeded ? "run-status success" : "run-status error";
|
||||
statusSpan.textContent = succeeded ? "Completed" : "Failed";
|
||||
void this._updateRun(runIndex, {
|
||||
status: succeeded ? "completed" : "failed",
|
||||
});
|
||||
this._activePromptIds.delete(event.promptId);
|
||||
if (this._queueingComplete && this._activePromptIds.size === 0) {
|
||||
this.isRunning = false;
|
||||
openclawUI.showBanner(
|
||||
"success",
|
||||
"All experiment runs finished."
|
||||
);
|
||||
}
|
||||
},
|
||||
|
||||
@@ -804,6 +899,7 @@ export const ParameterLabTab = {
|
||||
},
|
||||
|
||||
applyOverrides(run) {
|
||||
let receiptWidget = null;
|
||||
Object.entries(run).forEach(([key, value]) => {
|
||||
if (key === "prompt_id" || key === "status") return;
|
||||
const separatorIndex = key.indexOf(".");
|
||||
@@ -823,7 +919,9 @@ export const ParameterLabTab = {
|
||||
: null);
|
||||
if (widget) {
|
||||
widget.value = value;
|
||||
receiptWidget ||= widget;
|
||||
}
|
||||
});
|
||||
return receiptWidget;
|
||||
}
|
||||
};
|
||||
|
||||
@@ -0,0 +1,499 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import {
|
||||
PARAMETER_LAB_RECEIPT_KEY,
|
||||
createParameterLabReceiptCoordinator,
|
||||
} from "../../openclaw_parameter_lab_receipt.js";
|
||||
|
||||
class FakeApi extends EventTarget {
|
||||
emit(type, detail) {
|
||||
this.dispatchEvent(new CustomEvent(type, { detail }));
|
||||
}
|
||||
}
|
||||
|
||||
function createHostFixture() {
|
||||
const api = new FakeApi();
|
||||
const submissions = [];
|
||||
const beforeQueuedSpy = vi.fn();
|
||||
const afterQueuedSpy = vi.fn();
|
||||
const widget = {
|
||||
beforeQueued: beforeQueuedSpy,
|
||||
afterQueued: afterQueuedSpy,
|
||||
};
|
||||
const graph = {
|
||||
extra: { preserved: true },
|
||||
onSerialize: vi.fn(),
|
||||
serialize() {
|
||||
const data = { nodes: [], extra: { ...this.extra } };
|
||||
this.onSerialize?.(data);
|
||||
return data;
|
||||
},
|
||||
};
|
||||
let nextRequestId = 1;
|
||||
let nextForeignId = 1;
|
||||
const app = {
|
||||
graph,
|
||||
rootGraph: graph,
|
||||
processingQueue: false,
|
||||
queueItems: [],
|
||||
async queuePrompt(number, batchCount = 1) {
|
||||
const requestId = nextRequestId++;
|
||||
this.queueItems.push({ requestId, number, batchCount });
|
||||
api.emit("promptQueueing", { requestId, batchCount });
|
||||
if (this.processingQueue) return false;
|
||||
|
||||
this.processingQueue = true;
|
||||
await Promise.resolve();
|
||||
try {
|
||||
while (this.queueItems.length) {
|
||||
const request = this.queueItems.pop();
|
||||
let queuedCount = 0;
|
||||
for (let index = 0; index < request.batchCount; index += 1) {
|
||||
widget.beforeQueued?.({ isPartialExecution: false });
|
||||
const workflow = graph.serialize();
|
||||
const marker = workflow.extra?.[PARAMETER_LAB_RECEIPT_KEY];
|
||||
const promptId = marker?.prompt_id || `foreign-${nextForeignId++}`;
|
||||
submissions.push({
|
||||
requestId: request.requestId,
|
||||
promptId,
|
||||
marker: marker || null,
|
||||
});
|
||||
widget.afterQueued?.({ isPartialExecution: false });
|
||||
queuedCount += 1;
|
||||
}
|
||||
api.emit("promptQueued", {
|
||||
requestId: request.requestId,
|
||||
batchCount: queuedCount,
|
||||
number: request.number,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.processingQueue = false;
|
||||
}
|
||||
return true;
|
||||
},
|
||||
};
|
||||
return {
|
||||
api,
|
||||
app,
|
||||
graph,
|
||||
submissions,
|
||||
widget,
|
||||
beforeQueuedSpy,
|
||||
afterQueuedSpy,
|
||||
};
|
||||
}
|
||||
|
||||
describe("Parameter Lab authoritative queue receipt", () => {
|
||||
it("correlates two overlapping runs across LIFO host work without tagging unrelated submission", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const ids = [
|
||||
"11111111-1111-4111-8111-111111111111",
|
||||
"22222222-2222-4222-8222-222222222222",
|
||||
];
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => ids.shift(),
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
|
||||
const first = coordinator.queue({
|
||||
experimentId: "exp-one",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const second = coordinator.queue({
|
||||
experimentId: "exp-two",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const unrelated = fixture.app.queuePrompt(0, 1);
|
||||
|
||||
const [firstReceipt, secondReceipt] = await Promise.all([first, second]);
|
||||
await unrelated;
|
||||
|
||||
expect(firstReceipt.promptId).toBe("11111111-1111-4111-8111-111111111111");
|
||||
expect(secondReceipt.promptId).toBe("22222222-2222-4222-8222-222222222222");
|
||||
expect(fixture.submissions).toEqual([
|
||||
expect.objectContaining({ marker: null }),
|
||||
expect.objectContaining({
|
||||
promptId: "22222222-2222-4222-8222-222222222222",
|
||||
marker: {
|
||||
version: 1,
|
||||
prompt_id: "22222222-2222-4222-8222-222222222222",
|
||||
},
|
||||
}),
|
||||
expect.objectContaining({
|
||||
promptId: "11111111-1111-4111-8111-111111111111",
|
||||
marker: {
|
||||
version: 1,
|
||||
prompt_id: "11111111-1111-4111-8111-111111111111",
|
||||
},
|
||||
}),
|
||||
]);
|
||||
expect(fixture.graph.extra).toEqual({ preserved: true });
|
||||
expect(fixture.graph.onSerialize).toHaveBeenCalledTimes(3);
|
||||
expect(fixture.beforeQueuedSpy).toHaveBeenCalledTimes(3);
|
||||
expect(fixture.afterQueuedSpy).toHaveBeenCalledTimes(3);
|
||||
expect(coordinator.debugSnapshot()).toEqual({
|
||||
activeAttempts: 2,
|
||||
pendingRequests: 0,
|
||||
lifecycleSubscriptions: 0,
|
||||
installed: true,
|
||||
});
|
||||
|
||||
firstReceipt.release();
|
||||
secondReceipt.release();
|
||||
coordinator.dispose();
|
||||
expect(coordinator.debugSnapshot()).toEqual({
|
||||
activeAttempts: 0,
|
||||
pendingRequests: 0,
|
||||
lifecycleSubscriptions: 0,
|
||||
installed: false,
|
||||
});
|
||||
});
|
||||
|
||||
it("routes only exact native lifecycle IDs and disposes terminal and late events", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const promptId = "33333333-3333-4333-8333-333333333333";
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => promptId,
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
const receipt = await coordinator.queue({
|
||||
experimentId: "exp-life",
|
||||
runId: "7",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const observed = [];
|
||||
receipt.subscribeLifecycle((event) => observed.push(event.type));
|
||||
|
||||
fixture.api.emit("execution_start", { prompt_id: "unrelated" });
|
||||
fixture.api.emit("execution_start", {
|
||||
prompt_id: promptId,
|
||||
private_node_value: "must-not-cross-receipt-boundary",
|
||||
});
|
||||
fixture.api.emit("execution_success", { prompt_id: promptId });
|
||||
fixture.api.emit("execution_error", { prompt_id: promptId });
|
||||
|
||||
expect(observed).toEqual(["execution_start", "execution_success"]);
|
||||
expect(coordinator.debugSnapshot().lifecycleSubscriptions).toBe(0);
|
||||
expect(coordinator.debugSnapshot().activeAttempts).toBe(0);
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("fails closed and releases ownership when the lifecycle consumer throws", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const promptId = "35353535-3535-4535-8535-353535353535";
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => promptId,
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
const receipt = await coordinator.queue({
|
||||
experimentId: "exp-consumer-failure",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
receipt.subscribeLifecycle(() => {
|
||||
throw new Error("private lifecycle consumer detail");
|
||||
});
|
||||
|
||||
expect(() =>
|
||||
fixture.api.emit("execution_start", { prompt_id: promptId }),
|
||||
).not.toThrow();
|
||||
expect(coordinator.debugSnapshot()).toEqual({
|
||||
activeAttempts: 0,
|
||||
pendingRequests: 0,
|
||||
lifecycleSubscriptions: 0,
|
||||
installed: false,
|
||||
});
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("redacts raw lifecycle detail at the coordinator boundary", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const promptId = "34343434-3434-4434-8434-343434343434";
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => promptId,
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
const receipt = await coordinator.queue({
|
||||
experimentId: "exp-private",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const observed = [];
|
||||
receipt.subscribeLifecycle((event) => observed.push(event));
|
||||
|
||||
fixture.api.emit("execution_error", {
|
||||
prompt_id: promptId,
|
||||
exception_message: "secret=private-host-detail",
|
||||
node_id: "private-node",
|
||||
});
|
||||
|
||||
expect(observed).toEqual([
|
||||
{
|
||||
type: "execution_error",
|
||||
promptId,
|
||||
},
|
||||
]);
|
||||
expect(JSON.stringify(observed)).not.toContain("private-host-detail");
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("settles the host owner before a sequential run opens the next queue window", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const ids = [
|
||||
"36363636-3636-4636-8636-363636363636",
|
||||
"37373737-3737-4737-8737-373737373737",
|
||||
];
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => ids.shift(),
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
|
||||
const first = await coordinator.queue({
|
||||
experimentId: "exp-sequential",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
expect(fixture.app.processingQueue).toBe(false);
|
||||
|
||||
const second = await coordinator.queue({
|
||||
experimentId: "exp-sequential",
|
||||
runId: "1",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
|
||||
expect(first.promptId).toBe("36363636-3636-4636-8636-363636363636");
|
||||
expect(second.promptId).toBe("37373737-3737-4737-8737-373737373737");
|
||||
first.release();
|
||||
second.release();
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("fails closed before enqueueing when the pre-existing host owner was not observed", async () => {
|
||||
const fixture = createHostFixture();
|
||||
fixture.app.processingQueue = true;
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => "44444444-4444-4444-8444-444444444444",
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
|
||||
await expect(
|
||||
coordinator.queue({
|
||||
experimentId: "exp-busy",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "host_queue_unobserved_busy" });
|
||||
expect(fixture.app.queueItems).toEqual([]);
|
||||
expect(fixture.submissions).toEqual([]);
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("restores callbacks and rejects malformed UUID, cancellation, and dispose without leaks", async () => {
|
||||
const malformedFixture = createHostFixture();
|
||||
const malformed = createParameterLabReceiptCoordinator({
|
||||
app: malformedFixture.app,
|
||||
api: malformedFixture.api,
|
||||
uuidFactory: () => "not-a-uuid",
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
await expect(
|
||||
malformed.queue({
|
||||
experimentId: "exp-bad",
|
||||
runId: "0",
|
||||
widget: malformedFixture.widget,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "invalid_receipt_id" });
|
||||
malformed.dispose();
|
||||
|
||||
const fixture = createHostFixture();
|
||||
const originalBefore = fixture.widget.beforeQueued;
|
||||
const originalAfter = fixture.widget.afterQueued;
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => "55555555-5555-4555-8555-555555555555",
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
await expect(
|
||||
coordinator.queue({
|
||||
experimentId: "exp-cancel",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
signal: controller.signal,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "attempt_cancelled" });
|
||||
coordinator.dispose();
|
||||
|
||||
expect(fixture.widget.beforeQueued).toBe(originalBefore);
|
||||
expect(fixture.widget.afterQueued).toBe(originalAfter);
|
||||
expect(coordinator.debugSnapshot().installed).toBe(false);
|
||||
});
|
||||
|
||||
it("buffers an exact fast terminal event until subscription and ignores it after release", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const promptId = "66666666-6666-4666-8666-666666666666";
|
||||
fixture.afterQueuedSpy.mockImplementationOnce(() => {
|
||||
fixture.api.emit("execution_start", { prompt_id: promptId });
|
||||
fixture.api.emit("execution_success", { prompt_id: promptId });
|
||||
});
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => promptId,
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
|
||||
const receipt = await coordinator.queue({
|
||||
experimentId: "exp-fast",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const observed = [];
|
||||
receipt.subscribeLifecycle((event) => observed.push(event.type));
|
||||
fixture.api.emit("execution_error", { prompt_id: promptId });
|
||||
|
||||
expect(observed).toEqual(["execution_start", "execution_success"]);
|
||||
expect(coordinator.debugSnapshot().activeAttempts).toBe(0);
|
||||
coordinator.dispose();
|
||||
});
|
||||
|
||||
it("rejects marker collision and original callback failure with exact restoration", async () => {
|
||||
const collision = createHostFixture();
|
||||
collision.graph.extra[PARAMETER_LAB_RECEIPT_KEY] = {
|
||||
version: 1,
|
||||
prompt_id: "77777777-7777-4777-8777-777777777777",
|
||||
};
|
||||
const collisionCoordinator = createParameterLabReceiptCoordinator({
|
||||
app: collision.app,
|
||||
api: collision.api,
|
||||
uuidFactory: () => "88888888-8888-4888-8888-888888888888",
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
await expect(
|
||||
collisionCoordinator.queue({
|
||||
experimentId: "exp-collision",
|
||||
runId: "0",
|
||||
widget: collision.widget,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "receipt_marker_collision" });
|
||||
collisionCoordinator.dispose();
|
||||
|
||||
const callbackFailure = createHostFixture();
|
||||
const originalBefore = callbackFailure.widget.beforeQueued;
|
||||
const originalAfter = callbackFailure.widget.afterQueued;
|
||||
callbackFailure.beforeQueuedSpy.mockImplementationOnce(() => {
|
||||
throw new Error("private callback detail");
|
||||
});
|
||||
const failedCoordinator = createParameterLabReceiptCoordinator({
|
||||
app: callbackFailure.app,
|
||||
api: callbackFailure.api,
|
||||
uuidFactory: () => "99999999-9999-4999-8999-999999999999",
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
await expect(
|
||||
failedCoordinator.queue({
|
||||
experimentId: "exp-callback",
|
||||
runId: "0",
|
||||
widget: callbackFailure.widget,
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "host_queue_failed" });
|
||||
expect(callbackFailure.widget.beforeQueued).toBe(originalBefore);
|
||||
expect(callbackFailure.widget.afterQueued).toBe(originalAfter);
|
||||
expect(failedCoordinator.debugSnapshot().installed).toBe(false);
|
||||
failedCoordinator.dispose();
|
||||
});
|
||||
|
||||
it("bounds a missing host boundary with timeout cleanup", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const fixture = createHostFixture();
|
||||
fixture.app.queuePrompt = function () {
|
||||
const requestId = 1;
|
||||
fixture.api.emit("promptQueueing", {
|
||||
requestId,
|
||||
batchCount: 1,
|
||||
});
|
||||
this.processingQueue = true;
|
||||
return new Promise(() => {});
|
||||
};
|
||||
const originalBefore = fixture.widget.beforeQueued;
|
||||
const originalAfter = fixture.widget.afterQueued;
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa",
|
||||
timeoutMs: 25,
|
||||
});
|
||||
const pending = coordinator.queue({
|
||||
experimentId: "exp-timeout",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const rejection = expect(pending).rejects.toMatchObject({
|
||||
code: "receipt_timeout",
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(25);
|
||||
|
||||
await rejection;
|
||||
coordinator.dispose();
|
||||
expect(fixture.widget.beforeQueued).toBe(originalBefore);
|
||||
expect(fixture.widget.afterQueued).toBe(originalAfter);
|
||||
expect(coordinator.debugSnapshot().installed).toBe(false);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("bounds unrelated request-boundary floods without losing exact lifecycle cleanup", async () => {
|
||||
const fixture = createHostFixture();
|
||||
const promptId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
|
||||
const coordinator = createParameterLabReceiptCoordinator({
|
||||
app: fixture.app,
|
||||
api: fixture.api,
|
||||
uuidFactory: () => promptId,
|
||||
timeoutMs: 1_000,
|
||||
});
|
||||
const receipt = await coordinator.queue({
|
||||
experimentId: "exp-boundary-flood",
|
||||
runId: "0",
|
||||
widget: fixture.widget,
|
||||
});
|
||||
const observed = [];
|
||||
receipt.subscribeLifecycle((event) => observed.push(event.type));
|
||||
|
||||
for (let index = 0; index < 1_000; index += 1) {
|
||||
fixture.api.emit("promptQueueing", {
|
||||
requestId: 10_000 + index,
|
||||
batchCount: 1,
|
||||
});
|
||||
}
|
||||
|
||||
expect(coordinator.debugSnapshot().pendingRequests).toBe(0);
|
||||
fixture.api.emit("execution_success", { prompt_id: promptId });
|
||||
expect(observed).toEqual(["execution_success"]);
|
||||
expect(coordinator.debugSnapshot()).toEqual({
|
||||
activeAttempts: 0,
|
||||
pendingRequests: 0,
|
||||
lifecycleSubscriptions: 0,
|
||||
installed: false,
|
||||
});
|
||||
coordinator.dispose();
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user