nodes/Artificial-Sweetener-Substitute-Backend
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(sugarcubes): persist native queue context

Browse Source
This commit is contained in:
Artificial Sweetener
2026-09-11 15:19:42 -04:00
parent fe2d3f115d
commit c5b20bd447
33 changed files with 760 additions and 1705 deletions
Download Patch File Download Diff File Expand all files Collapse all files
package-lock.json
Generated
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "substitute-backend",
"version": "1.9.1",
"version": "1.10.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "substitute-backend",
"version": "1.9.1",
"version": "1.10.0",
"license": "AGPL-3.0-or-later",
"devDependencies": {
"@semantic-release/changelog": "^6.0.3",
package.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "substitute-backend",
"private": true,
"version": "1.9.1",
"version": "1.10.0",
"license": "AGPL-3.0-or-later",
"type": "module",
"scripts": {
pyproject.toml
+1 -2
View File
@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "substitute-backend"
description = "Backend liaison services for SugarSubstitute and ComfyUI."
version = "1.9.1"
version = "1.10.0"
license = "AGPL-3.0-or-later"
license-files = ["LICENSE"]
requires-python = ">=3.12"
@@ -15,7 +15,6 @@ classifiers = [
dependencies = [
"aiohttp",
"requests",
"sugar-dsl==1.2.0",
]
[project.optional-dependencies]
substitute_backend/__init__.py
+1 -1
View File
@@ -17,7 +17,7 @@
from __future__ import annotations
__version__ = "1.9.1"
__version__ = "1.10.0"
EXTENSION_NAME = "Substitute BackEnd"
API_VERSION = 1
substitute_backend/features/cube_outputs/application/__init__.py
-20
View File
@@ -1,20 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Application services for cube-output publishing."""
from substitute_backend.features.cube_outputs.application.services import CubeOutputServices
__all__ = ["CubeOutputServices"]
substitute_backend/features/cube_outputs/application/services.py
-31
View File
@@ -1,31 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Service container for cube-output websocket publishing."""
from __future__ import annotations
from dataclasses import dataclass
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_registration import (
SugarCubesCubeOutputRegistration,
)
@dataclass(frozen=True)
class CubeOutputServices:
"""Own cube-output observer registration services."""
registration: SugarCubesCubeOutputRegistration
substitute_backend/features/prompt_queue/domain/run_context.py
+4
View File
@@ -48,6 +48,7 @@ class SubstituteRunContext:
workflow_id: str
generation_run_id: str
client_id: str
output_session_id: str | None = None
scene_run_id: str | None = None
scene_key: str | None = None
scene_title: str | None = None
@@ -66,6 +67,8 @@ class SubstituteRunContext:
"sourceKey": source.source_key,
"sourceLabel": source.source_label,
}
if self.output_session_id is not None:
payload["outputSessionId"] = self.output_session_id
if self.scene_run_id is not None:
payload["sceneRunId"] = self.scene_run_id
if self.scene_key is not None:
@@ -110,6 +113,7 @@ def parse_substitute_run_context(value: object) -> SubstituteRunContext | None:
workflow_id=workflow_id,
generation_run_id=generation_run_id,
client_id=client_id,
output_session_id=_optional_string(value.get("outputSessionId")),
scene_run_id=_optional_string(scene_mapping.get("runId")),
scene_key=_optional_string(scene_mapping.get("key")),
scene_title=_optional_string(scene_mapping.get("title")),
substitute_backend/features/prompt_queue/infrastructure/sugarcubes_queue_context.py
+121
View File
@@ -0,0 +1,121 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Capture Substitute routing context at SugarCubes' validated queue boundary."""
from __future__ import annotations
from collections.abc import Mapping
from copy import deepcopy
from typing import Protocol
from substitute_backend.features.prompt_queue.application.run_context_store import (
SubstituteRunContextStore,
)
from substitute_backend.features.prompt_queue.domain.run_context import (
parse_substitute_run_context,
)
class ValidatedQueueEventLike(Protocol):
"""Describe the stable SugarCubes queue event surface consumed by BackEnd."""
@property
def prompt_id(self) -> str:
"""Return the validated host prompt identity."""
@property
def prompt(self) -> object:
"""Return the final executable prompt."""
@property
def extra_data(self) -> object:
"""Return the host queue metadata envelope."""
@property
def report(self) -> object | None:
"""Return the optional SugarCubes execution report."""
class SugarCubesQueueContextObserver:
"""Persist validated Substitute routing metadata before native queue insertion."""
def __init__(self, *, run_context_store: SubstituteRunContextStore) -> None:
"""Store the authoritative process-lifetime run-context owner."""
self._run_context_store = run_context_store
def on_validated_queue(self, event: ValidatedQueueEventLike) -> None:
"""Capture valid Substitute context while leaving ordinary Comfy prompts alone."""
extra_data = event.extra_data
if not isinstance(extra_data, Mapping) or "substitute" not in extra_data:
return
context_payload = _context_with_report_sources(
extra_data["substitute"],
getattr(event, "report", None),
)
context = parse_substitute_run_context(context_payload)
if context is None:
raise ValueError("Substitute run context is malformed.")
self._run_context_store.store(
prompt_id=event.prompt_id,
context=context,
executable_prompt=event.prompt,
)
def _context_with_report_sources(value: object, report: object) -> object:
"""Fill native node routing from SugarCubes' authoritative execution report."""
if not isinstance(value, Mapping):
return value
enriched = deepcopy(dict(value))
raw_presentations = enriched.pop("cubePresentations", None)
presentations = raw_presentations if isinstance(raw_presentations, Mapping) else {}
if isinstance(value.get("sources"), Mapping):
return enriched
identities = getattr(report, "execution_node_identities", None)
if not isinstance(identities, tuple | list):
identities = getattr(report, "output_identities", None)
if not isinstance(identities, tuple | list):
return enriched
sources: dict[str, dict[str, str]] = {}
for identity in identities:
execution_id = getattr(identity, "execution_id", None)
instance_id = getattr(identity, "instance_id", None)
instance_alias = getattr(identity, "instance_alias", None)
if not isinstance(execution_id, str) or not isinstance(instance_id, str):
continue
presentation_label = presentations.get(instance_id)
source_label = (
presentation_label.strip()
if isinstance(presentation_label, str) and presentation_label.strip()
else (
instance_alias
if isinstance(instance_alias, str) and instance_alias.strip()
else instance_id
)
)
sources[execution_id] = {
"sourceKey": f"cube:{instance_id}",
"sourceLabel": source_label,
"cubeAlias": source_label,
}
enriched["sources"] = sources
return enriched
__all__ = ["SugarCubesQueueContextObserver", "ValidatedQueueEventLike"]
substitute_backend/features/sugar_compile/__init__.py
-16
View File
@@ -1,16 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Sugar compile feature package."""
substitute_backend/features/sugar_compile/api/__init__.py
-23
View File
@@ -1,23 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""HTTP API for backend-owned Sugar compilation."""
from substitute_backend.features.sugar_compile.api.routes import (
SugarCompileRouteHandlers,
build_sugar_compile_route_handlers,
)
__all__ = ["SugarCompileRouteHandlers", "build_sugar_compile_route_handlers"]
substitute_backend/features/sugar_compile/api/routes.py
-107
View File
@@ -1,107 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""HTTP route handlers for backend-owned Sugar compilation."""
from __future__ import annotations
import logging
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import cast
from aiohttp import web
from substitute_backend.api.errors import BackendHttpError, json_error
from substitute_backend.api.serialization import JsonObject
from substitute_backend.features.sugar_compile.application import SugarCompileServices
from substitute_backend.features.sugar_compile.domain import (
SugarCompileError,
SugarCompileRequest,
)
RouteHandler = Callable[[web.Request], Awaitable[web.StreamResponse]]
@dataclass(frozen=True)
class SugarCompileRouteHandlers:
"""Concrete Sugar compile route callables used by host registration."""
compile_sugar: RouteHandler
def build_sugar_compile_route_handlers(
services: SugarCompileServices,
logger: logging.Logger,
) -> SugarCompileRouteHandlers:
"""Build thin HTTP handlers over Sugar compile application services."""
async def compile_sugar(request: web.Request) -> web.Response:
"""Compile one Sugar script into prompt and workflow artifacts."""
try:
body = await _json_object_body(request)
result = services.compile.compile(SugarCompileRequest.from_payload(body))
return web.json_response(result.to_payload())
except BackendHttpError as exc:
return json_error(exc)
except SugarCompileError as exc:
return json_error(
BackendHttpError(
message=exc.message,
status=exc.status,
code=exc.code,
)
)
except Exception: # pragma: no cover - defensive host boundary.
logger.exception(
"Sugar compile route failed.",
extra={
"operation": "sugar-compile",
"route": "/substitute/v1/sugar/compile",
},
)
return json_error(
BackendHttpError(
message="Sugar workflow could not be compiled.",
status=500,
code="sugar-compile-failed",
)
)
return SugarCompileRouteHandlers(compile_sugar=compile_sugar)
async def _json_object_body(request: web.Request) -> JsonObject:
"""Parse and validate a JSON object request body."""
try:
body = await request.json()
except (TypeError, ValueError) as exc:
raise BackendHttpError(
message="Request body must be a JSON object.",
status=400,
code="sugar-compile-invalid-request",
) from exc
if not isinstance(body, dict):
raise BackendHttpError(
message="Request body must be a JSON object.",
status=400,
code="sugar-compile-invalid-request",
)
return cast(
JsonObject,
{str(key): value for key, value in body.items() if isinstance(key, str)},
)
substitute_backend/features/sugar_compile/application/services.py
-141
View File
@@ -1,141 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Use-case service for backend-owned Sugar compilation."""
from __future__ import annotations
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
from substitute_backend.api.errors import BackendHttpError
from substitute_backend.features.sugar_compile.domain import (
SugarCompileCapabilities,
SugarCompileError,
SugarCompileRequest,
SugarCompileResult,
SugarCompileUnavailableError,
)
class SugarWorkflowCompiler(Protocol):
"""Describe the infrastructure adapter that invokes Sugar-DSL."""
def is_available(self) -> bool:
"""Return whether Sugar-DSL can be imported in this backend runtime."""
def unavailable_reason(self) -> str:
"""Return a setup-oriented reason when Sugar-DSL is unavailable."""
def sugar_dsl_version(self) -> str:
"""Return the installed Sugar-DSL package version or an empty string."""
def compile(self, *, script_text: str, output_dir: Path) -> SugarCompileResult:
"""Compile Sugar text into Comfy artifacts."""
@dataclass(frozen=True)
class SugarCompileService:
"""Coordinate one Sugar compile request without owning Sugar-DSL details."""
compiler: SugarWorkflowCompiler
logger: logging.Logger
def capabilities(self) -> SugarCompileCapabilities:
"""Return the backend capability payload for Sugar compilation."""
if self.compiler.is_available():
return SugarCompileCapabilities(
available=True,
live_node_definitions=True,
sugar_dsl_version=self.compiler.sugar_dsl_version(),
)
return SugarCompileCapabilities(
available=False,
unavailable_reason=self.compiler.unavailable_reason(),
)
def compile(self, request: SugarCompileRequest) -> SugarCompileResult:
"""Compile one validated Sugar request through the injected compiler."""
self.logger.info(
"Sugar compile requested.",
extra={
"operation": "sugar-compile",
"output_path": str(request.output_dir),
"script_length": len(request.sugar_script_text),
},
)
try:
return self.compiler.compile(
script_text=request.sugar_script_text,
output_dir=request.output_dir,
)
except SugarCompileUnavailableError as exc:
self.logger.warning(
"Sugar compile unavailable.",
extra={
"operation": "sugar-compile",
"output_path": str(request.output_dir),
"script_length": len(request.sugar_script_text),
"error_category": exc.code,
},
)
raise _http_error(exc) from exc
except SugarCompileError as exc:
self.logger.warning(
"Sugar compile failed.",
extra={
"operation": "sugar-compile",
"output_path": str(request.output_dir),
"script_length": len(request.sugar_script_text),
"error_category": exc.code,
},
)
raise _http_error(exc) from exc
except Exception as exc:
self.logger.exception(
"Unexpected Sugar compile failure.",
extra={
"operation": "sugar-compile",
"output_path": str(request.output_dir),
"script_length": len(request.sugar_script_text),
"error_category": "unexpected",
},
)
raise BackendHttpError(
message=f"Sugar compile failed: {exc}",
status=500,
code="sugar-compile-failed",
) from exc
@dataclass(frozen=True)
class SugarCompileServices:
"""Group Sugar compile feature services for host route registration."""
compile: SugarCompileService
def _http_error(error: SugarCompileError) -> BackendHttpError:
"""Convert a domain compile error into the backend HTTP error contract."""
return BackendHttpError(
message=error.message,
status=error.status,
code=error.code,
)
substitute_backend/features/sugar_compile/domain/__init__.py
-34
View File
@@ -1,34 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Domain models for backend-owned Sugar compilation."""
from substitute_backend.features.sugar_compile.domain.models import (
SUGAR_COMPILE_ROUTE,
SugarCompileCapabilities,
SugarCompileError,
SugarCompileRequest,
SugarCompileResult,
SugarCompileUnavailableError,
)
__all__ = [
"SUGAR_COMPILE_ROUTE",
"SugarCompileCapabilities",
"SugarCompileError",
"SugarCompileRequest",
"SugarCompileResult",
"SugarCompileUnavailableError",
]
substitute_backend/features/sugar_compile/domain/models.py
-152
View File
@@ -1,152 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Domain contracts for the Sugar compile route."""
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path, PureWindowsPath
from substitute_backend.api.serialization import JsonObject
SUGAR_COMPILE_ROUTE = "/substitute/v1/sugar/compile"
SUGAR_COMPILE_SCHEMA_VERSION = 1
class SugarCompileError(RuntimeError):
"""Carry expected Sugar compilation failures across backend layers."""
def __init__(
self,
message: str,
*,
status: int = 400,
code: str = "sugar-compile-failed",
) -> None:
"""Create a typed compile failure with public HTTP error metadata."""
super().__init__(message)
self.message = message
self.status = status
self.code = code
class SugarCompileUnavailableError(SugarCompileError):
"""Report a missing Sugar-DSL runtime dependency."""
def __init__(self, message: str) -> None:
"""Create an unavailable compile failure."""
super().__init__(
message,
status=503,
code="sugar-compile-unavailable",
)
@dataclass(frozen=True)
class SugarCompileRequest:
"""Describe one Sugar script compile request from SugarSubstitute."""
sugar_script_text: str
output_dir: Path
schema_version: int = SUGAR_COMPILE_SCHEMA_VERSION
@classmethod
def from_payload(cls, payload: JsonObject) -> SugarCompileRequest:
"""Validate and construct a compile request from public JSON data."""
schema_version = payload.get("schemaVersion")
if schema_version != SUGAR_COMPILE_SCHEMA_VERSION:
raise SugarCompileError(
"Sugar compile request schemaVersion must be 1.",
status=400,
code="sugar-compile-invalid-request",
)
script_text = payload.get("sugarScriptText")
if not isinstance(script_text, str) or not script_text.strip():
raise SugarCompileError(
"Sugar compile request requires non-empty sugarScriptText.",
status=400,
code="sugar-compile-invalid-request",
)
output_dir = payload.get("outputDir")
if not isinstance(output_dir, str) or not output_dir.strip():
raise SugarCompileError(
"Sugar compile request requires non-empty outputDir.",
status=400,
code="sugar-compile-invalid-request",
)
if "\x00" in output_dir:
raise SugarCompileError(
"Sugar compile request outputDir contains an invalid character.",
status=400,
code="sugar-compile-invalid-request",
)
windows_path = PureWindowsPath(output_dir)
if not windows_path.is_absolute():
raise SugarCompileError(
"Sugar compile request outputDir must be an absolute Windows path.",
status=400,
code="sugar-compile-invalid-request",
)
return cls(
sugar_script_text=script_text,
output_dir=Path(output_dir),
)
@dataclass(frozen=True)
class SugarCompileResult:
"""Hold compiled Comfy prompt and UI workflow artifacts."""
prompt: JsonObject
workflow: JsonObject
def to_payload(self) -> JsonObject:
"""Return the public route response shape."""
return {
"prompt": self.prompt,
"workflow": self.workflow,
}
@dataclass(frozen=True)
class SugarCompileCapabilities:
"""Describe Sugar compilation support in the active backend runtime."""
available: bool
unavailable_reason: str = ""
live_node_definitions: bool = False
sugar_dsl_version: str = ""
schema_version: int = SUGAR_COMPILE_SCHEMA_VERSION
compile_route: str = SUGAR_COMPILE_ROUTE
def to_payload(self) -> JsonObject:
"""Return the public capability payload."""
payload: JsonObject = {
"schemaVersion": self.schema_version,
"available": self.available,
}
if self.available:
payload["compileRoute"] = self.compile_route
payload["liveNodeDefinitions"] = self.live_node_definitions
payload["sugarDslVersion"] = self.sugar_dsl_version
else:
payload["unavailableReason"] = self.unavailable_reason
return payload
substitute_backend/features/sugar_compile/infrastructure/__init__.py
-26
View File
@@ -1,26 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Infrastructure adapters for backend-owned Sugar compilation."""
from substitute_backend.features.sugar_compile.infrastructure.sugar_dsl_compiler import (
BackendCubeArtifactResolver,
SugarDslWorkflowCompiler,
)
__all__ = [
"BackendCubeArtifactResolver",
"SugarDslWorkflowCompiler",
]
substitute_backend/features/sugar_compile/infrastructure/sugar_dsl_compiler.py
-297
View File
@@ -1,297 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Sugar-DSL compiler adapter for the ComfyUI backend runtime."""
from __future__ import annotations
import copy
import importlib.util
import logging
from collections.abc import Mapping
from importlib import metadata
from pathlib import Path
from typing import Any, cast
from substitute_backend.api.errors import BackendHttpError
from substitute_backend.api.serialization import JsonObject
from substitute_backend.features.cube_library.application import CubeLibraryService
from substitute_backend.features.sugar_compile.domain import (
SugarCompileError,
SugarCompileResult,
SugarCompileUnavailableError,
)
_LATEST_VERSION_MEMO_KEY = "__latest__"
_UNAVAILABLE_REASON = "Sugar-DSL is not installed in the ComfyUI environment."
class SugarDslWorkflowCompiler:
"""Invoke Sugar-DSL while isolating optional Sugar imports."""
def __init__(
self,
*,
cube_library: CubeLibraryService,
logger: logging.Logger,
) -> None:
"""Store backend services needed by Sugar-DSL compilation."""
self._cube_library = cube_library
self._logger = logger
def is_available(self) -> bool:
"""Return whether Sugar-DSL's public builder can be imported."""
try:
return importlib.util.find_spec("sugar.api.builder") is not None
except (ImportError, ModuleNotFoundError, ValueError):
return False
def unavailable_reason(self) -> str:
"""Return the setup message for a missing Sugar-DSL install."""
return _UNAVAILABLE_REASON
def sugar_dsl_version(self) -> str:
"""Return the installed Sugar-DSL package version or an empty string."""
try:
return metadata.version("sugar-dsl")
except metadata.PackageNotFoundError:
return ""
def compile(self, *, script_text: str, output_dir: Path) -> SugarCompileResult:
"""Compile Sugar script text through Sugar-DSL."""
try:
from sugar.api.builder import (
build_comfy_artifacts_from_text,
)
from sugar.compiler.errors import (
SugarCompilerError as SugarDslCompilerError,
)
from sugar.runtime.live_definitions import (
ComfyRegistryLiveNodeDefinitionProvider,
)
except ImportError as exc:
raise SugarCompileUnavailableError(_UNAVAILABLE_REASON) from exc
try:
payload = cast(
Mapping[str, object],
build_comfy_artifacts_from_text(
script_text,
output_dir=output_dir,
cube_artifact_resolver=BackendCubeArtifactResolver(
cube_library=self._cube_library,
logger=self._logger,
),
live_node_definition_provider=ComfyRegistryLiveNodeDefinitionProvider(
logger=self._logger,
),
),
)
except SugarDslCompilerError as exc:
raise SugarCompileError(
exc.message,
status=400,
code=exc.code,
) from exc
except SugarCompileError:
raise
except Exception as exc:
raise SugarCompileError(
f"Sugar-DSL compile failed: {exc}",
status=500,
code="sugar-compile-failed",
) from exc
prompt = payload.get("prompt")
workflow = payload.get("workflow")
if not isinstance(prompt, dict) or not isinstance(workflow, dict):
raise SugarCompileError(
"Sugar-DSL did not return prompt/workflow artifacts.",
status=500,
code="sugar-compile-failed",
)
return SugarCompileResult(
prompt=cast(JsonObject, prompt),
workflow=cast(JsonObject, workflow),
)
class BackendCubeArtifactResolver:
"""Resolve Sugar cube artifacts through the backend Cube Library service."""
def __init__(
self,
*,
cube_library: CubeLibraryService,
logger: logging.Logger,
) -> None:
"""Create a resolver for one Sugar compile operation."""
self._cube_library = cube_library
self._logger = logger
self._memo: dict[tuple[str, str], Any] = {}
def resolve(
self,
*,
alias: str,
cube_id: str,
requested_version: str | None,
) -> Any:
"""Return one Sugar-DSL ``ResolvedCubeArtifact`` for a use statement."""
memo_key = (cube_id, requested_version or _LATEST_VERSION_MEMO_KEY)
resolved = self._memo.get(memo_key)
if resolved is None:
artifact = self._load_artifact(cube_id, requested_version)
resolved = self._resolved_artifact(
alias=alias,
cube_id=cube_id,
requested_version=requested_version,
artifact=artifact,
)
self._memo[memo_key] = resolved
return resolved
def _load_artifact(
self,
cube_id: str,
requested_version: str | None,
) -> JsonObject:
"""Load a raw cube artifact payload from the backend Cube Library."""
try:
if requested_version is None:
artifact = self._cube_library.load_cube(cube_id)
else:
artifact = self._cube_library.load_cube_version(
cube_id=cube_id,
version=requested_version,
)
except BackendHttpError as exc:
raise SugarCompileError(
f"Cube '{cube_id}' is not available in the active Cube Library.",
status=exc.status,
code="sugar-cube-artifact-unavailable",
) from exc
except Exception as exc:
self._logger.warning(
"Backend cube artifact load failed during Sugar compile.",
extra={
"operation": "sugar-cube-artifact-resolve",
"cube_id": cube_id,
"cube_version": requested_version or "",
"error": repr(exc),
},
)
raise SugarCompileError(
f"Cube '{cube_id}' could not be loaded from the active Cube Library.",
status=503,
code="sugar-cube-artifact-unavailable",
) from exc
return artifact
def _resolved_artifact(
self,
*,
alias: str,
cube_id: str,
requested_version: str | None,
artifact: Mapping[str, object],
) -> Any:
"""Validate and convert a Cube Library payload into Sugar-DSL's model."""
declared_cube_id = _required_text(artifact, "cubeId", cube_id=cube_id)
if declared_cube_id != cube_id:
raise SugarCompileError(
f"Cube '{cube_id}' returned artifact '{declared_cube_id}'.",
status=502,
code="sugar-cube-artifact-invalid",
)
resolved_version = _required_text(artifact, "version", cube_id=cube_id)
if requested_version is not None and resolved_version != requested_version:
raise SugarCompileError(
f"Cube '{cube_id}' version mismatch: expected "
f"'{requested_version}', got '{resolved_version}'.",
status=502,
code="sugar-cube-artifact-invalid",
)
cube_payload = artifact.get("cube")
if not isinstance(cube_payload, Mapping):
raise SugarCompileError(
f"Cube '{cube_id}' returned an invalid artifact payload.",
status=502,
code="sugar-cube-artifact-invalid",
)
try:
from sugar.catalog.artifacts import (
CubeArtifactIdentity,
ResolvedCubeArtifact,
)
from sugar.catalog.models import (
validate_cube_document,
)
except ImportError as exc:
raise SugarCompileUnavailableError(_UNAVAILABLE_REASON) from exc
try:
cube = validate_cube_document(dict(cube_payload))
except (RuntimeError, TypeError, ValueError) as exc:
raise SugarCompileError(
f"Cube '{cube_id}' returned an invalid artifact payload: {exc}",
status=502,
code="sugar-cube-artifact-invalid",
) from exc
self._logger.debug(
"Resolved Sugar cube artifact.",
extra={
"operation": "sugar-cube-artifact-resolve",
"cube_alias": alias,
"cube_id": cube_id,
"cube_version": resolved_version,
},
)
return ResolvedCubeArtifact(
cube=copy.deepcopy(cube),
identity=CubeArtifactIdentity(
cube_id=cube_id,
requested_version=requested_version,
resolved_version=resolved_version,
),
)
def _required_text(
payload: Mapping[str, object],
key: str,
*,
cube_id: str,
) -> str:
"""Read one required non-empty string from a Cube Library artifact."""
value = payload.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
raise SugarCompileError(
f"Cube '{cube_id}' returned an invalid artifact {key}.",
status=502,
code="sugar-cube-artifact-invalid",
)
substitute_backend/features/sugar_compile/application/__init__.py → substitute_backend/features/sugarcubes_integration/__init__.py
+7 -6
View File
@@ -13,12 +13,13 @@
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Application services for backend-owned Sugar compilation."""
"""Expose cohesive SugarCubes liaison construction and lifecycle services."""
from substitute_backend.features.sugar_compile.application.services import (
SugarCompileService,
SugarCompileServices,
SugarWorkflowCompiler,
from substitute_backend.features.sugarcubes_integration.composition import (
build_sugarcubes_integration,
)
from substitute_backend.features.sugarcubes_integration.services import (
SugarCubesIntegrationServices,
)
__all__ = ["SugarCompileService", "SugarCompileServices", "SugarWorkflowCompiler"]
__all__ = ["SugarCubesIntegrationServices", "build_sugarcubes_integration"]
substitute_backend/features/sugarcubes_integration/composition.py
+78
View File
@@ -0,0 +1,78 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Construct BackEnd's cohesive SugarCubes liaison service graph."""
from __future__ import annotations
from substitute_backend.features.cube_outputs.infrastructure.prompt_server_publisher import (
PromptServerCubeOutputPublisher,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_observer import (
SubstituteCubeOutputObserver,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_observer_hook import (
SugarCubesObserverHookResolver,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_registration import (
SugarCubesCubeOutputRegistration,
)
from substitute_backend.features.prompt_queue.application.run_context_store import (
SubstituteRunContextStore,
)
from substitute_backend.features.prompt_queue.infrastructure.sugarcubes_queue_context import (
SugarCubesQueueContextObserver,
)
from substitute_backend.features.sugarcubes_integration.queue_context_registration import (
SugarCubesQueueContextRegistration,
)
from substitute_backend.features.sugarcubes_integration.services import (
SugarCubesIntegrationServices,
)
from substitute_backend.infrastructure.logging import get_logger
def build_sugarcubes_integration(
*,
prompt_server: object,
run_context_store: SubstituteRunContextStore,
) -> SugarCubesIntegrationServices:
"""Build all BackEnd observers that participate in SugarCubes execution."""
output_publisher = PromptServerCubeOutputPublisher(
prompt_server=prompt_server,
logger=get_logger("cube_outputs.publisher"),
)
output_observer = SubstituteCubeOutputObserver(
publisher=output_publisher,
logger=get_logger("cube_outputs.observer"),
run_context_store=run_context_store,
)
return SugarCubesIntegrationServices(
cube_output_registration=SugarCubesCubeOutputRegistration(
hook_resolver=SugarCubesObserverHookResolver(
logger=get_logger("cube_outputs.sugarcubes")
),
observer=output_observer,
logger=get_logger("cube_outputs.registration"),
),
queue_context_registration=SugarCubesQueueContextRegistration(
observer=SugarCubesQueueContextObserver(run_context_store=run_context_store),
logger=get_logger("prompt_queue.sugarcubes_context"),
),
)
__all__ = ["build_sugarcubes_integration"]
substitute_backend/features/sugarcubes_integration/queue_context_registration.py
+74
View File
@@ -0,0 +1,74 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Own required Substitute queue-context registration with SugarCubes."""
from __future__ import annotations
import logging
from substitute_backend.features.prompt_queue.infrastructure.sugarcubes_queue_context import (
SugarCubesQueueContextObserver,
)
from substitute_backend.infrastructure.sugarcubes_host_api import (
SugarCubesHostApiResolutionStatus,
SugarCubesHostApiResolver,
)
class SugarCubesQueueContextRegistration:
"""Attach BackEnd context capture to SugarCubes exactly once per process."""
def __init__(
self,
*,
observer: SugarCubesQueueContextObserver,
logger: logging.Logger,
host_api_resolver: SugarCubesHostApiResolver | None = None,
) -> None:
"""Configure lazy public-API resolution and structured diagnostics."""
self._observer = observer
self._logger = logger
self._host_api_resolver = host_api_resolver or SugarCubesHostApiResolver()
self._registered = False
def register(self) -> None:
"""Register the required pre-queue observer when SugarCubes is available."""
if self._registered:
return
resolution = self._host_api_resolver.resolve()
if resolution.status is SugarCubesHostApiResolutionStatus.PENDING:
self._logger.debug(
"SugarCubes queue-context registration pending",
extra={"reason": resolution.message},
)
return
if resolution.status is SugarCubesHostApiResolutionStatus.UNAVAILABLE:
self._logger.warning(
"SugarCubes queue-context registration unavailable",
extra={"reason": resolution.message},
)
return
api = resolution.api
if api is None:
raise RuntimeError("SugarCubes host API resolved without an API object.")
api.register_validated_queue_observer(
self._observer.on_validated_queue,
required=True,
)
self._registered = True
self._logger.info("SugarCubes queue-context observer registered")
substitute_backend/features/sugarcubes_integration/services.py
+52
View File
@@ -0,0 +1,52 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Coordinate BackEnd's complete liaison lifecycle with SugarCubes."""
from __future__ import annotations
import asyncio
from dataclasses import dataclass
from typing import Protocol
class RegistrationLike(Protocol):
"""Describe one idempotent SugarCubes observer registration."""
def register(self) -> object:
"""Attempt observer registration without duplicating it."""
@dataclass(frozen=True)
class SugarCubesIntegrationServices:
"""Own BackEnd's output publication and queue-context liaison registrations."""
cube_output_registration: RegistrationLike
queue_context_registration: RegistrationLike
def register(self) -> None:
"""Attempt every SugarCubes liaison registration."""
self.cube_output_registration.register()
self.queue_context_registration.register()
def schedule_retry(self) -> None:
"""Retry after Comfy finishes the current asynchronous startup task."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
return
loop.call_soon(self.register)
substitute_backend/host/extension.py
+12 -87
View File
@@ -17,7 +17,6 @@
from __future__ import annotations
import asyncio
import os
from collections.abc import Mapping, MutableMapping
from dataclasses import dataclass
@@ -34,19 +33,6 @@ from substitute_backend.features.cube_library.infrastructure import (
PromptServerCubeLibraryPublisher,
SugarCubesLibraryAdapter,
)
from substitute_backend.features.cube_outputs.application import CubeOutputServices
from substitute_backend.features.cube_outputs.infrastructure.prompt_server_publisher import (
PromptServerCubeOutputPublisher,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_observer import (
SubstituteCubeOutputObserver,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_observer_hook import (
SugarCubesObserverHookResolver,
)
from substitute_backend.features.cube_outputs.infrastructure.sugarcubes_registration import (
SugarCubesCubeOutputRegistration,
)
from substitute_backend.features.downloads.application import DownloadServices
from substitute_backend.features.downloads.application.telemetry_service import (
DownloadTelemetryService,
@@ -203,11 +189,10 @@ from substitute_backend.features.prompt_queue.infrastructure.comfy_prompt_queue
PromptQueueLike,
PromptServerRuntimeLike,
)
from substitute_backend.features.sugar_compile.application import (
SugarCompileService,
SugarCompileServices,
from substitute_backend.features.sugarcubes_integration import (
SugarCubesIntegrationServices,
build_sugarcubes_integration,
)
from substitute_backend.features.sugar_compile.infrastructure import SugarDslWorkflowCompiler
from substitute_backend.host.routes import (
BackendRouteHandlers,
PromptServerClassLike,
@@ -238,10 +223,9 @@ class BackendServices:
model_loading: ModelLoadingServices
downloads: DownloadServices
preview_assets: PreviewAssetServices
cube_outputs: CubeOutputServices
sugarcubes_integration: SugarCubesIntegrationServices
prompt_queue: PromptQueueServices
preview_metadata_enrichment: PreviewMetadataEnrichmentInstaller
sugar_compile: SugarCompileServices
diagnostics: DiagnosticLogger
@@ -502,34 +486,6 @@ def build_download_services(prompt_server: object | None = None) -> DownloadServ
)
def build_cube_output_services(
extension_root: Path,
prompt_server: object | None = None,
run_context_store: SubstituteRunContextStore | None = None,
) -> CubeOutputServices:
"""Build services for SugarCubes cube-output websocket publishing."""
publisher = PromptServerCubeOutputPublisher(
prompt_server=prompt_server or object(),
logger=get_logger("cube_outputs.publisher"),
)
observer = SubstituteCubeOutputObserver(
publisher=publisher,
logger=get_logger("cube_outputs.observer"),
run_context_store=run_context_store,
)
hook_resolver = SugarCubesObserverHookResolver(
logger=get_logger("cube_outputs.sugarcubes"),
)
return CubeOutputServices(
registration=SugarCubesCubeOutputRegistration(
hook_resolver=hook_resolver,
observer=observer,
logger=get_logger("cube_outputs.registration"),
)
)
def build_preview_asset_services() -> PreviewAssetServices:
"""Build application services for backend-managed preview assets."""
@@ -578,21 +534,6 @@ def build_prompt_queue_services(
return PromptQueueServices(queue=PromptQueueService(adapter))
def build_sugar_compile_services(cube_library: CubeLibraryServices) -> SugarCompileServices:
"""Build services for backend-owned Sugar-DSL compilation."""
compiler = SugarDslWorkflowCompiler(
cube_library=cube_library.library,
logger=get_logger("sugar_compile.compiler"),
)
return SugarCompileServices(
compile=SugarCompileService(
compiler=compiler,
logger=get_logger("sugar_compile.service"),
)
)
def build_backend_services(
extension_root: Path,
model_roots: ModelRootsProvider | None = None,
@@ -641,12 +582,11 @@ def build_backend_services(
record_phase("downloads")
resolved_preview_assets = preview_assets or build_preview_asset_services()
record_phase("preview_assets")
cube_outputs = build_cube_output_services(
extension_root,
prompt_server=prompt_server,
sugarcubes_integration = build_sugarcubes_integration(
prompt_server=prompt_server or object(),
run_context_store=run_context_store,
)
record_phase("cube_outputs")
record_phase("sugarcubes_integration")
prompt_queue = build_prompt_queue_services(
extension_root,
prompt_server=prompt_server,
@@ -659,8 +599,6 @@ def build_backend_services(
logger=get_logger("preview_routing.metadata"),
)
record_phase("preview_metadata_enrichment")
sugar_compile = build_sugar_compile_services(cube_library)
record_phase("sugar_compile")
_log_startup_timing(
"backend_services",
total_duration_ms=round((perf_counter() - started_at) * 1000, 3),
@@ -675,10 +613,9 @@ def build_backend_services(
model_loading=model_loading,
downloads=downloads,
preview_assets=resolved_preview_assets,
cube_outputs=cube_outputs,
sugarcubes_integration=sugarcubes_integration,
prompt_queue=prompt_queue,
preview_metadata_enrichment=preview_metadata_enrichment,
sugar_compile=sugar_compile,
diagnostics=diagnostics,
)
@@ -718,10 +655,10 @@ def register_extension(
record_phase("install_download_patch")
services.preview_metadata_enrichment.install()
record_phase("install_preview_metadata_enrichment")
services.cube_outputs.registration.register()
record_phase("register_cube_outputs")
_schedule_cube_output_registration_retry(services.cube_outputs.registration)
record_phase("schedule_cube_output_retry")
services.sugarcubes_integration.register()
record_phase("register_sugarcubes_integration")
services.sugarcubes_integration.schedule_retry()
record_phase("schedule_sugarcubes_integration_retry")
services.cube_library_change_monitor.start()
record_phase("start_cube_library_change_monitor")
services.model_metadata.changes.start()
@@ -778,18 +715,6 @@ def _resolve_prompt_server_instance(
return prompt_server
def _schedule_cube_output_registration_retry(
registration: SugarCubesCubeOutputRegistration,
) -> None:
"""Retry cube-output registration after Comfy finishes the current startup task."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
return
loop.call_soon(registration.register)
def _catalog_revision_from_status(cube_library: CubeLibraryServices) -> str:
"""Read the current Cube Library catalog revision from status payloads."""
substitute_backend/host/routes.py
+14 -25
View File
@@ -35,7 +35,6 @@ from substitute_backend.features.cube_library.api.routes import (
from substitute_backend.features.cube_library.application.services import (
CubeLibraryServices,
)
from substitute_backend.features.cube_outputs.application import CubeOutputServices
from substitute_backend.features.downloads.application import DownloadServices
from substitute_backend.features.environment_management.api.routes import (
EnvironmentRouteHandlers,
@@ -72,12 +71,9 @@ from substitute_backend.features.prompt_queue.api.routes import (
build_prompt_queue_route_handlers,
)
from substitute_backend.features.prompt_queue.application.services import PromptQueueServices
from substitute_backend.features.sugar_compile.api.routes import (
SugarCompileRouteHandlers,
build_sugar_compile_route_handlers,
from substitute_backend.features.sugarcubes_integration import (
SugarCubesIntegrationServices,
)
from substitute_backend.features.sugar_compile.application import SugarCompileServices
from substitute_backend.features.sugar_compile.domain import SUGAR_COMPILE_ROUTE
from substitute_backend.infrastructure.diagnostics import DiagnosticLogger
from substitute_backend.infrastructure.logging import get_logger
@@ -149,17 +145,13 @@ class BackendServicesLike(Protocol):
"""Return preview asset preparation services."""
@property
def cube_outputs(self) -> CubeOutputServices:
"""Return cube-output publishing services."""
def sugarcubes_integration(self) -> SugarCubesIntegrationServices:
"""Return SugarCubes liaison services."""
@property
def prompt_queue(self) -> PromptQueueServices:
"""Return prompt queue facade services."""
@property
def sugar_compile(self) -> SugarCompileServices:
"""Return Sugar compile services."""
@property
def diagnostics(self) -> DiagnosticLogger:
"""Return opt-in diagnostic logging services."""
@@ -175,7 +167,6 @@ class BackendRouteHandlers:
local_assets: LocalAssetRouteHandlers
preview_assets: PreviewAssetRouteHandlers
prompt_queue: PromptQueueRouteHandlers
sugar_compile: SugarCompileRouteHandlers
def register_routes(
@@ -206,14 +197,9 @@ def register_routes(
services.prompt_queue,
logger=get_logger("prompt_queue.routes"),
)
sugar_compile_handlers = build_sugar_compile_route_handlers(
services.sugar_compile,
logger=get_logger("sugar_compile.routes"),
)
routes = _resolve_routes(prompt_server)
routes.get("/substitute/v1/capabilities")(_build_capabilities_handler(services))
routes.post("/substitute/v1/prompt/queue")(prompt_queue_handlers.queue_prompt)
routes.post(SUGAR_COMPILE_ROUTE)(sugar_compile_handlers.compile_sugar)
routes.post(LOCAL_ASSET_AUTHORIZE_ROUTE)(local_asset_handlers.authorize)
routes.get("/substitute/v1/models")(model_handlers.list_models)
routes.get("/substitute/v1/models/changes")(model_handlers.latest_model_changes)
@@ -292,7 +278,6 @@ def register_routes(
local_assets=local_asset_handlers,
preview_assets=preview_asset_handlers,
prompt_queue=prompt_queue_handlers,
sugar_compile=sugar_compile_handlers,
)
@@ -321,7 +306,7 @@ def _build_capabilities_handler(
try:
try:
services.cube_outputs.registration.register()
services.sugarcubes_integration.register()
except Exception as exc:
logger.warning(
"cube-output registration failed during capability probe",
@@ -348,12 +333,10 @@ def _build_capabilities_handler(
feature_list.append("prompt-queue-facade")
if "visual-routing" not in feature_list:
feature_list.append("visual-routing")
if "native-cube-queue-context" not in feature_list:
feature_list.append("native-cube-queue-context")
if "local-assets" not in feature_list:
feature_list.append("local-assets")
sugar_compile_capabilities = services.sugar_compile.compile.capabilities()
record_phase("sugar_compile_capabilities")
if sugar_compile_capabilities.available and "sugar-compile" not in feature_list:
feature_list.append("sugar-compile")
feature_payload: list[JsonValue] = list(feature_list)
payload["features"] = feature_payload
payload["cubeLibrary"] = services.cube_library.library.capabilities()
@@ -394,7 +377,13 @@ def _build_capabilities_handler(
"eventType": "substitute_cube_output",
"previewMetadataKey": "substitute",
}
payload["sugarCompile"] = sugar_compile_capabilities.to_payload()
payload["nativeCubeQueueContext"] = {
"schemaVersion": 1,
"queueObserverApiVersion": 1,
"requiredObserver": True,
"preQueuePersistence": True,
"sourceIdentityFromExecutionReport": True,
}
record_phase("static_payloads")
_log_capabilities_timing(
logger,
substitute_backend/infrastructure/sugarcubes_host_api.py
+25
View File
@@ -24,12 +24,14 @@ from typing import Protocol, cast
SUGARCUBES_HOST_API_MODULE = "sugarcubes.host_api"
SUPPORTED_SUGARCUBES_HOST_API_VERSION = 1
SUPPORTED_QUEUE_OBSERVER_API_VERSION = 1
class SugarCubesHostApi(Protocol):
"""Describe the cross-extension surface owned and published by SugarCubes."""
HOST_API_VERSION: int
QUEUE_OBSERVER_API_VERSION: int
def active_backend_services(self) -> object | None:
"""Return the service graph already created by SugarCubes."""
@@ -40,6 +42,17 @@ class SugarCubesHostApi(Protocol):
def unregister_cube_output_observer(self, observer: object) -> None:
"""Unregister one cube-output observer."""
def register_validated_queue_observer(
self,
observer: object,
*,
required: bool = False,
) -> None:
"""Register one observer of prompts validated for native queueing."""
def unregister_validated_queue_observer(self, observer: object) -> None:
"""Unregister one validated-queue observer."""
class SugarCubesHostApiResolutionStatus(StrEnum):
"""Describe whether the public SugarCubes API can be consumed now."""
@@ -81,6 +94,16 @@ class SugarCubesHostApiResolver:
f"{SUPPORTED_SUGARCUBES_HOST_API_VERSION}."
),
)
queue_observer_version = getattr(module, "QUEUE_OBSERVER_API_VERSION", None)
if queue_observer_version != SUPPORTED_QUEUE_OBSERVER_API_VERSION:
return SugarCubesHostApiResolution(
status=SugarCubesHostApiResolutionStatus.UNAVAILABLE,
message=(
"SugarCubes queue-observer API version "
f"{queue_observer_version!r} is not supported; expected "
f"{SUPPORTED_QUEUE_OBSERVER_API_VERSION}."
),
)
missing_operation = _missing_operation(module)
if missing_operation is not None:
return SugarCubesHostApiResolution(
@@ -101,6 +124,8 @@ def _missing_operation(module: object) -> str | None:
"active_backend_services",
"register_cube_output_observer",
"unregister_cube_output_observer",
"register_validated_queue_observer",
"unregister_validated_queue_observer",
):
if not callable(getattr(module, name, None)):
return name
tests/test_capabilities_and_routes.py
+15 -12
View File
@@ -30,7 +30,6 @@ import pytest
from aiohttp import web
from substitute_backend.api.serialization import JsonObject
from substitute_backend.features.cube_outputs.application import CubeOutputServices
from substitute_backend.features.model_metadata.application.capability_service import (
CapabilityService,
)
@@ -42,6 +41,9 @@ from substitute_backend.features.preview_assets.application import (
PreviewAssetServices,
TaesdAssetService,
)
from substitute_backend.features.sugarcubes_integration import (
SugarCubesIntegrationServices,
)
from substitute_backend.host.extension import build_backend_services
from substitute_backend.host.routes import PromptServerLike, register_routes
from substitute_backend.infrastructure.logging import get_logger
@@ -128,7 +130,6 @@ def test_register_routes_uses_expected_surface(tmp_path: Path) -> None:
assert prompt_server.routes.registered == [
("GET", "/substitute/v1/capabilities"),
("POST", "/substitute/v1/prompt/queue"),
("POST", "/substitute/v1/sugar/compile"),
("POST", "/substitute/v1/local-assets/authorize"),
("GET", "/substitute/v1/models"),
("GET", "/substitute/v1/models/changes"),
@@ -243,10 +244,11 @@ def test_capabilities_payload_advertises_preview_assets(
assert isinstance(response, web.Response)
assert response.text is not None
payload = json.loads(response.text)
assert "sugar-compile" not in payload["features"]
assert "sugarCompile" not in payload
assert "preview-assets" in payload["features"]
assert "cube-library" in payload["features"]
assert "prompt-queue-facade" in payload["features"]
assert "sugar-compile" in payload["features"]
assert "local-assets" in payload["features"]
assert payload["cubeLibrary"] == {
"schemaVersion": 1,
@@ -273,13 +275,6 @@ def test_capabilities_payload_advertises_preview_assets(
"optimizationReportSupported": True,
"debugDumpSupported": False,
}
assert payload["sugarCompile"] == {
"schemaVersion": 1,
"available": True,
"compileRoute": "/substitute/v1/sugar/compile",
"liveNodeDefinitions": True,
"sugarDslVersion": "0.8.1",
}
asyncio.run(run_capabilities())
@@ -306,8 +301,9 @@ def test_capabilities_payload_survives_cube_output_registration_failure(
)
services = replace(
services,
cube_outputs=CubeOutputServices(
registration=cast(Any, _FailingRegistration()),
sugarcubes_integration=SugarCubesIntegrationServices(
cube_output_registration=cast(Any, _FailingRegistration()),
queue_context_registration=cast(Any, _FailingRegistration()),
),
)
prompt_server = FakePromptServer()
@@ -324,6 +320,13 @@ def test_capabilities_payload_survives_cube_output_registration_failure(
payload = json.loads(response.text)
assert "visual-routing" in payload["features"]
assert payload["visualRouting"]["eventType"] == "substitute_cube_output"
assert payload["nativeCubeQueueContext"] == {
"schemaVersion": 1,
"queueObserverApiVersion": 1,
"requiredObserver": True,
"preQueuePersistence": True,
"sourceIdentityFromExecutionReport": True,
}
asyncio.run(run_capabilities())
tests/test_cube_outputs.py
+18
View File
@@ -88,6 +88,7 @@ class _Hook:
identity = "test-hook"
HOST_API_VERSION = 1
QUEUE_OBSERVER_API_VERSION = 1
def __init__(self) -> None:
"""Initialize empty observer lists."""
@@ -110,6 +111,21 @@ class _Hook:
return object()
def register_validated_queue_observer(
self,
observer: object,
*,
required: bool = False,
) -> None:
"""Accept the companion queue-observer API required by the shared contract."""
_ = (observer, required)
def unregister_validated_queue_observer(self, observer: object) -> None:
"""Accept queue-observer removal required by the shared contract."""
_ = observer
def test_cube_output_event_payload_matches_public_contract() -> None:
"""Cube-output domain events should serialize the versioned websocket payload."""
@@ -288,6 +304,7 @@ def test_observer_enriches_cube_output_with_substitute_context() -> None:
context=SubstituteRunContext(
workflow_id="wf-1",
generation_run_id="run-1",
output_session_id="generate-click-1",
client_id="client-run",
scene_key="scene-a",
sources={
@@ -329,6 +346,7 @@ def test_observer_enriches_cube_output_with_substitute_context() -> None:
"schemaVersion": 1,
"workflowId": "wf-1",
"generationRunId": "run-1",
"outputSessionId": "generate-click-1",
"clientId": "client-run",
"sourceKey": "wf-1:node-1",
"sourceLabel": "Demo",
tests/test_preview_routing.py
+2
View File
@@ -81,6 +81,7 @@ def test_preview_metadata_enrichment_adds_substitute_identity() -> None:
context=SubstituteRunContext(
workflow_id="wf-1",
generation_run_id="run-1",
output_session_id="generate-click-1",
client_id="client-1",
sources={
"node-1": SubstituteSourceRoute(
@@ -118,6 +119,7 @@ def test_preview_metadata_enrichment_adds_substitute_identity() -> None:
"schemaVersion": 1,
"workflowId": "wf-1",
"generationRunId": "run-1",
"outputSessionId": "generate-click-1",
"clientId": "client-1",
"sourceKey": "wf-1:node-1",
"sourceLabel": "Demo",
tests/test_project_dependencies.py
+3 -5
View File
@@ -24,14 +24,12 @@ from pathlib import Path
_PROJECT_ROOT = Path(__file__).resolve().parents[1]
def test_sugar_dsl_dependency_is_hard_pinned_to_published_release() -> None:
"""Keep Backend compilation on the explicitly verified Sugar-DSL release."""
def test_backend_does_not_depend_on_sugar_dsl() -> None:
"""Keep SugarScript compilation outside Substitute BackEnd."""
metadata = tomllib.loads((_PROJECT_ROOT / "pyproject.toml").read_text(encoding="utf-8"))
project = metadata.get("project")
assert isinstance(project, dict)
dependencies = project.get("dependencies")
assert isinstance(dependencies, list)
assert [value for value in dependencies if str(value).startswith("sugar-dsl")] == [
"sugar-dsl==1.2.0"
]
assert not [value for value in dependencies if str(value).startswith("sugar-dsl")]
tests/test_prompt_queue.py
+2
View File
@@ -494,6 +494,7 @@ def test_comfy_queue_adapter_stores_substitute_run_context_by_prompt_id() -> Non
"schemaVersion": 1,
"workflowId": "wf-1",
"generationRunId": "run-1",
"outputSessionId": "generate-click-1",
"clientId": "client-1",
"sources": {
"5": {
@@ -512,6 +513,7 @@ def test_comfy_queue_adapter_stores_substitute_run_context_by_prompt_id() -> Non
assert context is not None
assert context.workflow_id == "wf-1"
assert context.generation_run_id == "run-1"
assert context.output_session_id == "generate-click-1"
assert context.client_id == "client-1"
assert context.sources["5"].source_key == "wf-1:5"
tests/test_sugar_compile.py
-585
View File
@@ -1,585 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Tests for backend-owned Sugar compilation."""
from __future__ import annotations
import asyncio
import json
import logging
from collections.abc import Mapping
from importlib import metadata
from pathlib import Path
from typing import Any, cast
import pytest
from aiohttp import web
from substitute_backend.api.errors import BackendHttpError
from substitute_backend.api.serialization import JsonObject
from substitute_backend.features.cube_library.application import (
CubeLibraryService,
)
from substitute_backend.features.sugar_compile.api import (
SugarCompileRouteHandlers,
build_sugar_compile_route_handlers,
)
from substitute_backend.features.sugar_compile.application import (
SugarCompileService,
SugarCompileServices,
)
from substitute_backend.features.sugar_compile.domain import (
SugarCompileError,
SugarCompileRequest,
SugarCompileResult,
SugarCompileUnavailableError,
)
from substitute_backend.features.sugar_compile.infrastructure import (
BackendCubeArtifactResolver,
SugarDslWorkflowCompiler,
)
from substitute_backend.infrastructure.logging import get_logger
class FakeRequest:
"""Minimal aiohttp-like request with injectable JSON body."""
def __init__(self, body: object) -> None:
"""Store one request body."""
self._body = body
async def json(self) -> object:
"""Return the configured JSON body."""
return self._body
class RecordingCompiler:
"""Record Sugar compile service calls and return deterministic payloads."""
def __init__(self, *, available: bool = True, failure: Exception | None = None) -> None:
"""Configure availability and optional compile failure."""
self.calls: list[tuple[str, Path]] = []
self._available = available
self._failure = failure
def is_available(self) -> bool:
"""Return configured availability."""
return self._available
def unavailable_reason(self) -> str:
"""Return a deterministic unavailable message."""
return "Sugar-DSL is not installed in the ComfyUI environment."
def sugar_dsl_version(self) -> str:
"""Return a deterministic Sugar-DSL version."""
return "0.8.1"
def compile(self, *, script_text: str, output_dir: Path) -> SugarCompileResult:
"""Record the compile request and return a wrapped artifact payload."""
self.calls.append((script_text, output_dir))
if self._failure is not None:
raise self._failure
return SugarCompileResult(
prompt={"1": {"class_type": "KSampler"}},
workflow={"nodes": []},
)
class RecordingCubeLibraryGateway:
"""Return raw Cube Library artifacts while recording load calls."""
def __init__(
self,
*,
invalid_cube_id: str = "",
invalid_version: str = "",
missing_cube_id: str = "",
) -> None:
"""Configure load recording and failure controls."""
self.latest_loads: list[str] = []
self.version_loads: list[tuple[str, str]] = []
self._invalid_cube_id = invalid_cube_id
self._invalid_version = invalid_version
self._missing_cube_id = missing_cube_id
def status(self) -> JsonObject:
"""Return available status."""
return {"schemaVersion": 1, "available": True}
def capabilities(self) -> JsonObject:
"""Return available Cube Library capabilities."""
return {"schemaVersion": 1, "available": True}
def catalog(
self,
*,
include_disabled: bool,
diagnostic_context: object | None = None,
) -> JsonObject:
"""Return an empty catalog."""
_ = include_disabled, diagnostic_context
return {"schemaVersion": 1, "cubes": []}
def load_cube(
self,
cube_id: str,
*,
diagnostic_context: object | None = None,
) -> JsonObject:
"""Return a latest cube artifact."""
_ = diagnostic_context
self.latest_loads.append(cube_id)
return self._artifact(cube_id, "latest")
def list_cube_versions(self, cube_id: str) -> JsonObject:
"""Return no versions."""
_ = cube_id
return {"schemaVersion": 1, "versions": []}
def load_cube_version(
self,
*,
cube_id: str,
version: str,
diagnostic_context: object | None = None,
) -> JsonObject:
"""Return a versioned cube artifact."""
_ = diagnostic_context
self.version_loads.append((cube_id, version))
return self._artifact(cube_id, version)
def prewarm_cube_version(self, *, cube_id: str, version: str) -> JsonObject:
"""Return accepted prewarm status."""
_ = cube_id, version
return {"schemaVersion": 1, "accepted": True}
def icon_asset(self, cube_id: str) -> tuple[bytes, str]:
"""Return icon content."""
_ = cube_id
return b"", "image/png"
def list_packs(self) -> JsonObject:
"""Return no packs."""
return {"schemaVersion": 1, "packs": []}
def preflight_pack(self, *, owner: str, repo: str, branch: str) -> JsonObject:
"""Return a no-op preflight payload."""
_ = owner, repo, branch
return {"schemaVersion": 1}
def add_pack(
self,
*,
owner: str,
repo: str,
branch: str,
enabled: bool,
auto_update: bool,
sync_immediately: bool,
) -> JsonObject:
"""Return a no-op add payload."""
_ = owner, repo, branch, enabled, auto_update, sync_immediately
return {"schemaVersion": 1}
def update_pack(
self,
*,
owner: str,
repo: str,
branch: str | None,
enabled: bool | None,
auto_update: bool | None,
) -> JsonObject:
"""Return a no-op update payload."""
_ = owner, repo, branch, enabled, auto_update
return {"schemaVersion": 1}
def remove_pack(self, *, owner: str, repo: str) -> JsonObject:
"""Return a no-op remove payload."""
_ = owner, repo
return {"schemaVersion": 1}
def sync_pack(self, *, owner: str, repo: str) -> JsonObject:
"""Return a no-op sync payload."""
_ = owner, repo
return {"schemaVersion": 1}
def sync_all_packs(self) -> JsonObject:
"""Return a no-op sync-all payload."""
return {"schemaVersion": 1}
def readiness(self) -> JsonObject:
"""Return readiness."""
return {"schemaVersion": 1, "ready": True}
def dependency_readiness(self) -> JsonObject:
"""Return dependency readiness."""
return {"schemaVersion": 1, "ready": True}
def repair_dependencies(
self,
*,
baseline_only: bool,
approved_node_ids: tuple[str, ...],
sync_enabled_repos: bool,
) -> JsonObject:
"""Return repair result."""
_ = baseline_only, approved_node_ids, sync_enabled_repos
return {"schemaVersion": 1, "restartRequired": False}
def sync_and_check(self, payload: dict[str, object]) -> JsonObject:
"""Return a no-op sync/check result."""
_ = payload
return {"schemaVersion": 1, "restartRequired": False}
def _artifact(self, cube_id: str, version: str) -> JsonObject:
"""Build a raw Cube Library artifact."""
if cube_id == self._missing_cube_id:
raise BackendHttpError(
message="Missing cube.",
status=404,
code="cube-library-not-found",
)
declared_cube_id = self._invalid_cube_id or cube_id
declared_version = self._invalid_version or version
return {
"schemaVersion": 1,
"cubeId": declared_cube_id,
"version": declared_version,
"cube": {
"cube_id": cube_id,
"version": version,
"implementation": {
"nodes": {
"node": {
"class_type": "TestNode",
"inputs": {"value": 1},
}
},
"inputs": {},
"outputs": {},
"definitions": {},
"subgraphs": [],
"layout": {},
},
"surface": {
"default_flavor_id": "default",
"controls": [],
},
"flavors": {
"authored": [
{
"id": "default",
"name": "Default",
"values": {},
}
]
},
},
}
def _request(body: object) -> web.Request:
"""Return a typed fake aiohttp request."""
return cast(web.Request, FakeRequest(body))
def _payload(response: web.StreamResponse) -> dict[str, Any]:
"""Decode one JSON response."""
text = cast(web.Response, response).text or "{}"
return cast(dict[str, Any], json.loads(text))
def _handlers(compiler: RecordingCompiler) -> SugarCompileRouteHandlers:
"""Build Sugar compile route handlers with a fake compiler."""
return build_sugar_compile_route_handlers(
SugarCompileServices(
compile=SugarCompileService(
compiler=compiler,
logger=get_logger("tests.sugar_compile.service"),
)
),
logger=get_logger("tests.sugar_compile.routes"),
)
def test_compile_route_rejects_non_object_json() -> None:
"""Sugar compile route should reject array request bodies."""
async def run() -> None:
response = await _handlers(RecordingCompiler()).compile_sugar(_request([]))
assert response.status == 400
assert _payload(response)["error"]["code"] == "sugar-compile-invalid-request"
asyncio.run(run())
def test_compile_route_rejects_missing_script_text() -> None:
"""Sugar compile route should require script text before invoking compiler."""
async def run() -> None:
compiler = RecordingCompiler()
response = await _handlers(compiler).compile_sugar(
_request(
{
"schemaVersion": 1,
"outputDir": "E:\\outputs",
}
)
)
assert response.status == 400
assert _payload(response)["error"]["code"] == "sugar-compile-invalid-request"
assert compiler.calls == []
asyncio.run(run())
def test_compile_route_returns_wrapped_prompt_and_workflow() -> None:
"""Sugar compile route should return the compiler artifact payload."""
async def run() -> None:
compiler = RecordingCompiler()
response = await _handlers(compiler).compile_sugar(
_request(
{
"schemaVersion": 1,
"sugarScriptText": 'use "Owner/Repo/demo.cube" as demo',
"outputDir": "E:\\outputs",
}
)
)
assert response.status == 200
assert _payload(response) == {
"prompt": {"1": {"class_type": "KSampler"}},
"workflow": {"nodes": []},
}
assert compiler.calls == [('use "Owner/Repo/demo.cube" as demo', Path("E:\\outputs"))]
asyncio.run(run())
def test_compile_service_capabilities_include_sugar_dsl_version() -> None:
"""Sugar compile capabilities should expose the Backend-owned Sugar-DSL version."""
service = SugarCompileService(
compiler=RecordingCompiler(),
logger=logging.getLogger("tests.sugar_compile.service"),
)
assert service.capabilities().to_payload() == {
"schemaVersion": 1,
"available": True,
"compileRoute": "/substitute/v1/sugar/compile",
"liveNodeDefinitions": True,
"sugarDslVersion": "0.8.1",
}
def test_sugar_dsl_compiler_reads_installed_distribution_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Sugar-DSL infrastructure adapter should resolve installed package metadata."""
def installed_version(name: str) -> str:
"""Return deterministic Sugar-DSL metadata."""
_ = name
return "0.9.2"
monkeypatch.setattr(metadata, "version", installed_version)
compiler = SugarDslWorkflowCompiler(
cube_library=CubeLibraryService(gateway=RecordingCubeLibraryGateway()),
logger=get_logger("tests.sugar_compile.compiler"),
)
assert compiler.sugar_dsl_version() == "0.9.2"
def test_sugar_dsl_compiler_allows_missing_distribution_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Missing Sugar-DSL package metadata should not disable compile capability."""
def missing_version(name: str) -> str:
"""Raise the package metadata missing error."""
raise metadata.PackageNotFoundError(name)
monkeypatch.setattr(metadata, "version", missing_version)
compiler = SugarDslWorkflowCompiler(
cube_library=CubeLibraryService(gateway=RecordingCubeLibraryGateway()),
logger=get_logger("tests.sugar_compile.compiler"),
)
assert compiler.sugar_dsl_version() == ""
def test_compile_service_maps_unavailable_errors() -> None:
"""Sugar compile service should preserve unavailable setup failures."""
service = SugarCompileService(
compiler=RecordingCompiler(failure=SugarCompileUnavailableError("Sugar-DSL is missing.")),
logger=logging.getLogger("tests.sugar_compile.service"),
)
request = SugarCompileRequest(
sugar_script_text='use "Owner/Repo/demo.cube" as demo',
output_dir=Path("E:\\outputs"),
)
with pytest.raises(BackendHttpError) as error_info:
service.compile(request)
assert error_info.value.status == 503
assert error_info.value.code == "sugar-compile-unavailable"
def test_compile_service_maps_expected_compile_failures() -> None:
"""Sugar compile service should preserve structured compile failures."""
service = SugarCompileService(
compiler=RecordingCompiler(
failure=SugarCompileError(
"Cube invalid.",
status=502,
code="sugar-cube-artifact-invalid",
)
),
logger=logging.getLogger("tests.sugar_compile.service"),
)
request = SugarCompileRequest(
sugar_script_text='use "Owner/Repo/demo.cube" as demo',
output_dir=Path("E:\\outputs"),
)
with pytest.raises(BackendHttpError) as error_info:
service.compile(request)
assert error_info.value.status == 502
assert error_info.value.code == "sugar-cube-artifact-invalid"
def test_cube_artifact_resolver_loads_latest_and_pinned_artifacts_once() -> None:
"""Backend resolver should memoize equivalent cube/version requests."""
gateway = RecordingCubeLibraryGateway()
resolver = BackendCubeArtifactResolver(
cube_library=CubeLibraryService(gateway=gateway),
logger=get_logger("tests.sugar_compile.resolver"),
)
latest_a = resolver.resolve(
alias="latestA",
cube_id="Owner/Repo/latest.cube",
requested_version=None,
)
latest_b = resolver.resolve(
alias="latestB",
cube_id="Owner/Repo/latest.cube",
requested_version=None,
)
pinned_a = resolver.resolve(
alias="pinnedA",
cube_id="Owner/Repo/pinned.cube",
requested_version="1.0.0",
)
pinned_b = resolver.resolve(
alias="pinnedB",
cube_id="Owner/Repo/pinned.cube",
requested_version="1.0.0",
)
assert latest_a is latest_b
assert pinned_a is pinned_b
assert gateway.latest_loads == ["Owner/Repo/latest.cube"]
assert gateway.version_loads == [("Owner/Repo/pinned.cube", "1.0.0")]
def test_cube_artifact_resolver_rejects_identity_mismatch() -> None:
"""Backend resolver should reject Cube Library artifact id mismatches."""
resolver = BackendCubeArtifactResolver(
cube_library=CubeLibraryService(
gateway=RecordingCubeLibraryGateway(invalid_cube_id="Other/Repo/demo.cube")
),
logger=get_logger("tests.sugar_compile.resolver"),
)
with pytest.raises(SugarCompileError) as error_info:
resolver.resolve(
alias="demo",
cube_id="Owner/Repo/demo.cube",
requested_version=None,
)
assert error_info.value.code == "sugar-cube-artifact-invalid"
def test_cube_artifact_resolver_returns_sugar_resolved_artifact() -> None:
"""Backend resolver should return Sugar-DSL's resolved artifact model."""
resolver = BackendCubeArtifactResolver(
cube_library=CubeLibraryService(gateway=RecordingCubeLibraryGateway()),
logger=get_logger("tests.sugar_compile.resolver"),
)
resolved = resolver.resolve(
alias="demo",
cube_id="Owner/Repo/demo.cube",
requested_version="1.2.3",
)
identity = cast(Any, resolved).identity
cube = cast(Mapping[str, object], cast(Any, resolved).cube)
assert identity.cube_id == "Owner/Repo/demo.cube"
assert identity.requested_version == "1.2.3"
assert identity.resolved_version == "1.2.3"
assert cube["cube_id"] == "Owner/Repo/demo.cube"
tests/test_sugar_live_node_definitions.py
-132
View File
@@ -1,132 +0,0 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Tests for backend orchestration of Sugar-owned live definitions."""
from __future__ import annotations
import importlib
import logging
from pathlib import Path
from typing import Any, cast
import pytest
from substitute_backend.api.serialization import JsonObject
from substitute_backend.features.cube_library.application import CubeLibraryService
from substitute_backend.features.sugar_compile.domain import SugarCompileError
from substitute_backend.features.sugar_compile.infrastructure import (
SugarDslWorkflowCompiler,
)
def test_backend_compiler_passes_live_provider_to_sugar_dsl(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Backend compile calls should receive Sugar-DSL's runtime live provider."""
builder = importlib.import_module("sugar.api.builder")
calls: list[dict[str, Any]] = []
def fake_build_comfy_artifacts_from_text(
script_text: str,
*,
output_dir: Path,
cube_artifact_resolver: object,
live_node_definition_provider: object,
) -> JsonObject:
"""Record Sugar-DSL compile keyword arguments."""
calls.append(
{
"script_text": script_text,
"output_dir": output_dir,
"cube_artifact_resolver": cube_artifact_resolver,
"live_node_definition_provider": live_node_definition_provider,
}
)
return {"prompt": {}, "workflow": {}}
monkeypatch.setattr(
builder,
"build_comfy_artifacts_from_text",
fake_build_comfy_artifacts_from_text,
)
compiler = SugarDslWorkflowCompiler(
cube_library=_unused_cube_library(),
logger=logging.getLogger("tests.sugar.compiler"),
)
compiler.compile(script_text='use "demo" as Demo', output_dir=Path("E:\\outputs"))
[call] = calls
assert call["script_text"] == 'use "demo" as Demo'
assert call["output_dir"] == Path("E:\\outputs")
provider = call["live_node_definition_provider"]
assert type(provider).__name__ == "ComfyRegistryLiveNodeDefinitionProvider"
assert type(provider).__module__ == "sugar.runtime.live_definitions"
def test_backend_compiler_maps_live_definition_errors(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Sugar-DSL live-definition failures should keep their structured code."""
builder = importlib.import_module("sugar.api.builder")
errors = importlib.import_module("sugar.compiler.errors")
sugar_compiler_error = cast(Any, errors).SugarCompilerError
def fake_build_comfy_artifacts_from_text(
script_text: str,
*,
output_dir: Path,
cube_artifact_resolver: object,
live_node_definition_provider: object,
) -> JsonObject:
"""Raise the typed Sugar-DSL error from the compile boundary."""
_ = script_text, output_dir, cube_artifact_resolver, live_node_definition_provider
raise sugar_compiler_error(
"Required live input has no default.",
code="sugar-live-default-missing",
cube_alias="Demo",
cube_id="demo",
node_key="Demo.node",
node_class_type="SimpleNode",
input_name="new_widget",
)
monkeypatch.setattr(
builder,
"build_comfy_artifacts_from_text",
fake_build_comfy_artifacts_from_text,
)
compiler = SugarDslWorkflowCompiler(
cube_library=_unused_cube_library(),
logger=logging.getLogger("tests.sugar.compiler"),
)
with pytest.raises(SugarCompileError) as error_info:
compiler.compile(script_text='use "demo" as Demo', output_dir=Path("E:\\outputs"))
assert error_info.value.code == "sugar-live-default-missing"
assert error_info.value.status == 400
def _unused_cube_library() -> CubeLibraryService:
"""Return a cube library placeholder for compile calls that do not resolve cubes."""
return CubeLibraryService(gateway=cast(Any, object()))
tests/test_sugarcubes_adapter.py
+5
View File
@@ -244,4 +244,9 @@ def _publish_host_api(
module.__dict__["active_backend_services"] = lambda: services
module.__dict__["register_cube_output_observer"] = lambda _observer: None
module.__dict__["unregister_cube_output_observer"] = lambda _observer: None
module.__dict__["QUEUE_OBSERVER_API_VERSION"] = 1
module.__dict__["register_validated_queue_observer"] = lambda _observer, *, required=False: (
required
)
module.__dict__["unregister_validated_queue_observer"] = lambda _observer: None
monkeypatch.setitem(sys.modules, SUGARCUBES_HOST_API_MODULE, module)
tests/test_sugarcubes_host_api.py
+20
View File
@@ -43,6 +43,21 @@ def test_resolver_returns_the_canonical_versioned_host_api(
assert resolution.api is module
def test_resolver_requires_the_validated_queue_observer_contract(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Native queue context must fail closed when SugarCubes lacks its observer API."""
module = _host_api_module(version=1)
del module.__dict__["register_validated_queue_observer"]
monkeypatch.setitem(sys.modules, SUGARCUBES_HOST_API_MODULE, module)
resolution = SugarCubesHostApiResolver().resolve()
assert resolution.status is SugarCubesHostApiResolutionStatus.UNAVAILABLE
assert "register_validated_queue_observer" in resolution.message
def test_resolver_rejects_an_unsupported_host_api_version(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@@ -77,4 +92,9 @@ def _host_api_module(*, version: int) -> ModuleType:
module.__dict__["active_backend_services"] = lambda: object()
module.__dict__["register_cube_output_observer"] = lambda _observer: None
module.__dict__["unregister_cube_output_observer"] = lambda _observer: None
module.__dict__["QUEUE_OBSERVER_API_VERSION"] = 1
module.__dict__["register_validated_queue_observer"] = lambda _observer, *, required=False: (
required
)
module.__dict__["unregister_validated_queue_observer"] = lambda _observer: None
return module
tests/test_sugarcubes_queue_context.py
+303
View File
@@ -0,0 +1,303 @@
# Substitute BackEnd - backend liaison services for SugarSubstitute and ComfyUI
# Copyright (C) 2026 Artificial Sweetener and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Tests for Substitute context capture at SugarCubes' native queue boundary."""
from __future__ import annotations
import logging
from collections.abc import Callable
from dataclasses import dataclass
from typing import cast
import pytest
from substitute_backend.features.prompt_queue.application.run_context_store import (
SubstituteRunContextStore,
)
from substitute_backend.features.prompt_queue.infrastructure.sugarcubes_queue_context import (
SugarCubesQueueContextObserver,
)
from substitute_backend.features.sugarcubes_integration.queue_context_registration import (
SugarCubesQueueContextRegistration,
)
from substitute_backend.infrastructure.sugarcubes_host_api import (
SugarCubesHostApi,
SugarCubesHostApiResolution,
SugarCubesHostApiResolutionStatus,
SugarCubesHostApiResolver,
)
@dataclass(frozen=True)
class _ValidatedQueueEvent:
"""Provide the structural queue event consumed by the observer."""
prompt_id: str
prompt: object
extra_data: object
report: object | None = None
class _HostApi:
"""Capture the exact callable registered at the cross-extension boundary."""
def __init__(self) -> None:
"""Initialize empty registration state."""
self.observer: Callable[[object], None] | None = None
self.required = False
def register_validated_queue_observer(
self,
observer: object,
*,
required: bool = False,
) -> None:
"""Retain a callable observer and its fail-closed policy."""
assert callable(observer)
self.observer = observer
self.required = required
class _Resolver:
"""Resolve one configured host API for registration tests."""
def __init__(self, api: _HostApi) -> None:
"""Store the API returned by resolve."""
self._api = api
def resolve(self) -> SugarCubesHostApiResolution:
"""Return a successful structural resolution."""
return SugarCubesHostApiResolution(
status=SugarCubesHostApiResolutionStatus.RESOLVED,
message="resolved",
api=cast(SugarCubesHostApi, self._api),
)
def test_native_queue_observer_stores_substitute_context_before_queueing() -> None:
"""A validated native prompt should retain its Substitute visual routing context."""
store = SubstituteRunContextStore(time_source=lambda: 123.0)
observer = SugarCubesQueueContextObserver(run_context_store=store)
observer.on_validated_queue(
_ValidatedQueueEvent(
prompt_id="prompt-1",
prompt={"5": {"class_type": "TestNode", "inputs": {}}},
extra_data={"substitute": _substitute_context()},
)
)
context = store.resolve("prompt-1")
assert context is not None
assert context.workflow_id == "wf-1"
assert context.sources["5"].source_label == "CubeA"
def test_native_queue_observer_ignores_non_substitute_comfy_prompts() -> None:
"""Ordinary Comfy queue traffic should not create Substitute-owned context."""
store = SubstituteRunContextStore()
observer = SugarCubesQueueContextObserver(run_context_store=store)
observer.on_validated_queue(
_ValidatedQueueEvent(prompt_id="prompt-1", prompt={}, extra_data={})
)
assert store.resolve("prompt-1") is None
def test_native_queue_observer_derives_output_sources_from_execution_report() -> None:
"""SugarCubes output identities should become BackEnd routing sources atomically."""
@dataclass(frozen=True)
class _OutputIdentity:
"""Provide one SugarCubes report output identity."""
execution_id: str
instance_id: str
instance_alias: str | None
@dataclass(frozen=True)
class _Report:
"""Provide the output-identity portion of a SugarCubes execution report."""
output_identities: tuple[_OutputIdentity, ...]
store = SubstituteRunContextStore()
observer = SugarCubesQueueContextObserver(run_context_store=store)
context = _substitute_context()
context.pop("sources")
observer.on_validated_queue(
_ValidatedQueueEvent(
prompt_id="prompt-1",
prompt={"output-1": {"class_type": "SugarCubes.CubeOutput"}},
extra_data={"substitute": context},
report=_Report((_OutputIdentity("output-1", "instance-a", "CubeA"),)),
)
)
stored = store.resolve("prompt-1")
assert stored is not None
assert stored.sources["output-1"].source_key == "cube:instance-a"
assert stored.sources["output-1"].source_label == "CubeA"
def test_native_queue_observer_uses_substitute_cube_presentation_labels() -> None:
"""Use pre-queue presentation labels instead of qualified execution aliases."""
@dataclass(frozen=True)
class _OutputIdentity:
"""Provide one SugarCubes report output identity."""
execution_id: str
instance_id: str
instance_alias: str
@dataclass(frozen=True)
class _Report:
"""Provide output identities after SugarCubes lowers the graph."""
output_identities: tuple[_OutputIdentity, ...]
store = SubstituteRunContextStore()
observer = SugarCubesQueueContextObserver(run_context_store=store)
context = _substitute_context()
context.pop("sources")
context["cubePresentations"] = {"instance-a": "Prompt by Region"}
observer.on_validated_queue(
_ValidatedQueueEvent(
prompt_id="prompt-1",
prompt={"output-1": {"class_type": "SugarCubes.CubeOutput"}},
extra_data={"substitute": context},
report=_Report(
(
_OutputIdentity(
"output-1",
"instance-a",
"Anima/Prompt by Region",
),
)
),
)
)
stored = store.resolve("prompt-1")
assert stored is not None
assert stored.sources["output-1"].source_key == "cube:instance-a"
assert stored.sources["output-1"].source_label == "Prompt by Region"
def test_native_queue_observer_routes_every_cube_owned_execution_node() -> None:
"""Internal lowered nodes should carry Cube identity for previews and timing."""
@dataclass(frozen=True)
class _ExecutionIdentity:
"""Provide one SugarCubes lowered-node identity."""
execution_id: str
instance_id: str
instance_alias: str
@dataclass(frozen=True)
class _Report:
"""Provide the node-identity portion of a SugarCubes execution report."""
execution_node_identities: tuple[_ExecutionIdentity, ...]
store = SubstituteRunContextStore()
observer = SugarCubesQueueContextObserver(run_context_store=store)
context = _substitute_context()
context.pop("sources")
observer.on_validated_queue(
_ValidatedQueueEvent(
prompt_id="prompt-1",
prompt={"cube-a:sampler": {"class_type": "KSampler"}},
extra_data={"substitute": context},
report=_Report((_ExecutionIdentity("cube-a:sampler", "instance-a", "CubeA"),)),
)
)
stored = store.resolve("prompt-1")
assert stored is not None
assert stored.sources["cube-a:sampler"].source_key == "cube:instance-a"
assert stored.sources["cube-a:sampler"].source_label == "CubeA"
def test_native_queue_observer_rejects_malformed_substitute_context() -> None:
"""Malformed claimed Substitute context must stop queue insertion fail-closed."""
observer = SugarCubesQueueContextObserver(run_context_store=SubstituteRunContextStore())
with pytest.raises(ValueError, match="Substitute run context"):
observer.on_validated_queue(
_ValidatedQueueEvent(
prompt_id="prompt-1",
prompt={},
extra_data={"substitute": {"schemaVersion": 1}},
)
)
def test_registration_passes_the_observer_method_as_a_required_callable() -> None:
"""Registration must give SugarCubes a callable instead of its owning object."""
store = SubstituteRunContextStore()
observer = SugarCubesQueueContextObserver(run_context_store=store)
api = _HostApi()
registration = SugarCubesQueueContextRegistration(
observer=observer,
logger=logging.getLogger(__name__),
host_api_resolver=cast(SugarCubesHostApiResolver, _Resolver(api)),
)
registration.register()
assert api.required is True
assert api.observer is not None
api.observer(
_ValidatedQueueEvent(
prompt_id="prompt-registered",
prompt={"5": {"class_type": "TestNode", "inputs": {}}},
extra_data={"substitute": _substitute_context()},
)
)
assert store.resolve("prompt-registered") is not None
def _substitute_context() -> dict[str, object]:
"""Return a valid app-owned visual routing payload."""
return {
"schemaVersion": 1,
"workflowId": "wf-1",
"generationRunId": "run-1",
"clientId": "client-1",
"sources": {
"5": {
"sourceKey": "wf-1:5",
"sourceLabel": "CubeA",
"cubeAlias": "CubeA",
}
},
}