fix(workflows): derive honest media metadata for single imports (HEL-809)

Co-authored-by: multica-agent <github@multica.ai>

calliope-backend/src/calliope/comfyui/metadata.py

@@ -124,6 +124,19 @@ def note_description(ui_workflow: dict[str, Any], limit: int = 160) -> str | Non
     return None
 
 
+def media_io_from_schema(
+    inputs: list[dict[str, Any]], outputs: list[dict[str, Any]]
+) -> dict[str, list[str]]:
+    """Media declared by the final input schema and real graph outputs."""
+    media_kinds = ("image", "image_list", "video", "video_list", "audio", "audio_list")
+    return {
+        "in": sorted(
+            {e["kind"].replace("_list", "") for e in inputs if e.get("kind") in media_kinds}
+        ),
+        "out": sorted({e["kind"] for e in outputs if e.get("kind") in ("image", "video", "audio")}),
+    }
+
+
 def suggested_metadata(
     converted: dict[str, Any],
     source_path: str | None = None,
@@ -131,21 +144,10 @@ def suggested_metadata(
 ) -> dict[str, Any]:
     """The import suggestion bundle for one converted workflow."""
     judgment = classify_usage(converted)
-    media_kinds = ("image", "image_list", "video", "video_list", "audio", "audio_list")
-    media_in = sorted(
-        {
-            entry["kind"].replace("_list", "")
-            for entry in parse_dynamic_inputs(converted)
-            if entry.get("kind") in media_kinds
-        }
-    )
-    media_out = sorted(
-        {
-            str(entry.get("kind"))
-            for entry in parse_dynamic_outputs(converted)
-            if entry.get("kind") in ("image", "video", "audio")
-        }
+    media_io = media_io_from_schema(
+        parse_dynamic_inputs(converted), parse_dynamic_outputs(converted)
     )
+    media_in, media_out = media_io["in"], media_io["out"]
     category, category_label = category_from_source_path(source_path)
     return {
         "kind": "video" if "video" in media_out and "image" not in media_out else "image",

calliope-backend/src/calliope/routers/workflows.py

@@ -7,7 +7,7 @@ from fastapi import APIRouter, HTTPException
 
 from calliope.comfyui.endpoints import client_for, endpoint_snapshot
 from calliope.comfyui.lint import blocking_problem, workflow_warnings
-from calliope.comfyui.metadata import suggested_metadata
+from calliope.comfyui.metadata import media_io_from_schema, suggested_metadata
 from calliope.comfyui.output_map import (
     auto_output_map,
     load_output_map,
@@ -76,7 +76,9 @@ def _serialize_workflow(row: Any) -> dict[str, Any]:
         mapping_problem(data["workflow_json"], raw_map, data.get("output_map_source")) is not None
     )
     # HEL-681: structured library metadata (raw JSON columns → values).
-    data["media_io"] = _load_json_col(data.get("media_io"), {"in": [], "out": []})
+    data["media_io"] = _resolved_media_io(
+        data.get("media_io"), data["input_schema"], data["output_schema"]
+    )
     data["tags"] = _load_json_col(data.get("tags"), [])
     data["favorite"] = bool(data.get("favorite"))
     return data
@@ -90,6 +92,20 @@ def _load_json_col(raw: Any, default: Any) -> Any:
     return value if value is not None else default
 
 
+def _resolved_media_io(
+    raw: Any, inputs: list[dict[str, Any]], outputs: list[dict[str, Any]]
+) -> dict[str, list[str]]:
+    """Fill missing declarations only; an explicit empty list is meaningful."""
+    stored = _load_json_col(raw, {})
+    derived = media_io_from_schema(inputs, outputs)
+    if not isinstance(stored, dict):
+        return derived
+    return {
+        key: stored[key] if isinstance(stored.get(key), list) else derived[key]
+        for key in ("in", "out")
+    }
+
+
 async def _connection_check(
     endpoint_id: str | None, workflow_json: dict[str, Any], refresh: bool = False
 ) -> dict[str, Any]:
@@ -197,13 +213,15 @@ async def reanalyze_workflow(workflow_id: int) -> dict[str, Any]:
             output_map, output_map_source = None, None
         else:
             output_map, output_map_source = json.dumps(auto_output_map(workflow)), "auto"
+        media_io = _resolved_media_io(data["media_io"], schema, outputs)
+        media_io["in"] = media_io_from_schema(schema, outputs)["in"]
         is_enabled = data["is_enabled"]
         if output_map_source is None or usage is None:
             is_enabled = 0
         conn.execute(
             "UPDATE workflows SET input_schema = ?, output_schema = ?,"
             " usage = ?, usage_source = ?, output_map = ?, output_map_source = ?,"
-            " is_enabled = ? WHERE id = ?",
+            " is_enabled = ?, media_io = ? WHERE id = ?",
             (
                 json.dumps(schema),
                 json.dumps(outputs),
@@ -212,6 +230,7 @@ async def reanalyze_workflow(workflow_id: int) -> dict[str, Any]:
                 output_map,
                 output_map_source,
                 is_enabled,
+                json.dumps(media_io),
                 workflow_id,
             ),
         )
@@ -247,7 +266,8 @@ async def list_workflows(view: str | None = None) -> list[dict[str, Any]]:
     try:
         if view == "meta":
             # HEL-681: metadata-only list for 150+ scale browsing — no
-            # workflow_json, no per-row re-parse. `needs_mapping` here is the
+            # workflow_json in the response. Only legacy/missing media
+            # declarations need graph parsing. `needs_mapping` here is the
             # cheap pending check (usage/output_map not yet settled); rows
             # that never generated keep showing as pending exactly like the
             # full view reports them.
@@ -260,11 +280,27 @@ async def list_workflows(view: str | None = None) -> list[dict[str, Any]]:
                 data = row_to_dict(row)
                 data["is_enabled"] = bool(data.get("is_enabled"))
                 data["favorite"] = bool(data.get("favorite"))
-                data["media_io"] = _load_json_col(data.get("media_io"), {"in": [], "out": []})
+                media = _load_json_col(data.get("media_io"), {})
+                if isinstance(media, dict) and all(
+                    isinstance(media.get(key), list) for key in ("in", "out")
+                ):
+                    data["media_io"] = {key: media[key] for key in ("in", "out")}
+                else:
+                    legacy = conn.execute(
+                        "SELECT workflow_json, input_schema FROM workflows WHERE id = ?",
+                        (data["id"],),
+                    ).fetchone()
+                    graph = json.loads(legacy["workflow_json"])
+                    schema = effective_input_schema(
+                        _load_json_col(legacy["input_schema"], []), graph
+                    )
+                    data["media_io"] = _resolved_media_io(
+                        data.get("media_io"), schema, parse_dynamic_outputs(graph)
+                    )
                 data["tags"] = _load_json_col(data.get("tags"), [])
-                data["needs_mapping"] = data.get("usage") is None or data.get(
-                    "output_map_source"
-                ) is None
+                data["needs_mapping"] = (
+                    data.get("usage") is None or data.get("output_map_source") is None
+                )
                 out.append(data)
             return out
         rows = conn.execute("SELECT * FROM workflows ORDER BY id DESC").fetchall()
@@ -369,6 +405,9 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
         output_map = json.dumps(auto_map) if auto_map else None
         output_map_source = "auto" if auto_map else None
     mapping_ok = not needs_user_output_map(payload.workflow_json) or bool(payload.output_map)
+    schema = _stored_schema(payload.input_schema, payload.workflow_json)
+    outputs = parse_dynamic_outputs(payload.workflow_json)
+    media_io = media_io_from_schema(json.loads(schema), outputs)
     conn = get_db(settings.db_path)
     try:
         # Byte-identical re-imports are the main way the library fills up with
@@ -395,15 +434,15 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
             INSERT INTO workflows (name, kind, workflow_json, input_schema,
                                    output_schema, description, prompt_profile,
                                    endpoint_id, is_enabled, usage, usage_source,
-                                   output_map, output_map_source)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                                   output_map, output_map_source, media_io)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 payload.name,
                 kind,
                 json.dumps(payload.workflow_json),
-                _stored_schema(payload.input_schema, payload.workflow_json),
-                json.dumps(parse_dynamic_outputs(payload.workflow_json)),
+                schema,
+                json.dumps(outputs),
                 payload.description,
                 profile,
                 payload.endpoint_id,
@@ -412,6 +451,7 @@ async def create_workflow(payload: WorkflowCreate) -> dict[str, Any]:
                 usage_source,
                 output_map,
                 output_map_source,
+                json.dumps(media_io),
             ),
         )
         conn.commit()
@@ -474,7 +514,20 @@ async def update_workflow(workflow_id: int, payload: WorkflowUpdate) -> dict[str
         if "input_schema" in data and data["input_schema"] is not None:
             if not all(isinstance(e, dict) for e in data["input_schema"]):
                 raise HTTPException(status_code=422, detail="input_schema 必须是参数对象数组")
+            media = _resolved_media_io(
+                existing["media_io"], [], parse_dynamic_outputs(workflow_json)
+            )
+            schema = effective_input_schema(data["input_schema"], workflow_json)
+            media["in"] = media_io_from_schema(schema, [])["in"]
+            data["media_io"] = json.dumps(media)
             data["input_schema"] = json.dumps(data["input_schema"])
+        elif existing["media_io"] is None:
+            schema = effective_input_schema(
+                _load_json_col(existing["input_schema"], []), workflow_json
+            )
+            data["media_io"] = json.dumps(
+                media_io_from_schema(schema, parse_dynamic_outputs(workflow_json))
+            )
         # endpoint_id: explicit null clears the binding; other Nones are skips
         # (an output_map reset keeps its explicit NULLs).
         explicit_null = {"endpoint_id", "output_map", "output_map_source"} & {
@@ -792,6 +845,8 @@ def _insert_converted(
             "existing_id": existing["id"],
         }
     is_enabled = 1 if (usage and mapping_ok) else 0
+    schema = _stored_schema(item.input_schema, converted)
+    outputs = parse_dynamic_outputs(converted)
     cur = conn.execute(
         """
         INSERT INTO workflows (name, kind, workflow_json, input_schema, output_schema,
@@ -805,10 +860,8 @@ def _insert_converted(
             report["name"],
             kind,
             workflow_text,
-            json.dumps(
-                item.input_schema if item.input_schema else parse_dynamic_inputs(converted)
-            ),
-            json.dumps(parse_dynamic_outputs(converted)),
+            schema,
+            json.dumps(outputs),
             item.description if item.description is not None else suggested["description"],
             profile,
             is_enabled,
@@ -818,7 +871,7 @@ def _insert_converted(
             output_map_source,
             suggested["category"],
             suggested["model_family"],
-            json.dumps({"in": suggested["media_in"], "out": suggested["media_out"]}),
+            json.dumps(media_io_from_schema(json.loads(schema), outputs)),
             json.dumps(item.tags or []),
             item.source_path,
         ),

calliope-backend/tests/test_workflow_import_batch.py

@@ -339,7 +339,7 @@ def test_meta_view_has_metadata_but_no_full_json(client):
 
 def test_legacy_list_shape_and_old_records_unchanged(client):
     """Old imports (API format via POST /api/workflows) keep working; the
-    legacy list still carries workflow_json and the new columns default."""
+    legacy list still carries workflow_json and truthful media declarations."""
     created = client.post(
         "/api/workflows",
         json={"name": "旧导入", "workflow_json": _api_workflow()},
@@ -352,7 +352,7 @@ def test_legacy_list_shape_and_old_records_unchanged(client):
     assert "usage" in row and "output_map" in row
     assert row["favorite"] is False
     assert row["tags"] == []
-    assert row["media_io"] == {"in": [], "out": []}
+    assert row["media_io"] == {"in": ["image"], "out": ["video"]}
     meta = client.get("/api/workflows", params={"view": "meta"}).json()
     assert meta[0]["needs_mapping"] is False or meta[0]["needs_mapping"] is True
     assert meta[0]["source"] is None

calliope-backend/tests/test_workflow_media_io.py

@@ -0,0 +1,205 @@
+"""HEL-809: import declarations, persistence and both library views agree."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from calliope.config import settings
+from calliope.db import get_db
+from tests.test_workflow_import_batch import _ui_graph
+
+
+def graph(*classes):
+    nodes = {str(i): {"class_type": cls, "inputs": {}} for i, cls in enumerate(classes, 1)}
+    nodes["90"] = {
+        "class_type": "PrimitiveStringMultiline",
+        "inputs": {"value": "HEL809 isolated prompt"},
+        "_meta": {"title": "Prompt (Input:prompt)"},
+    }
+    for i in range(1, len(classes) + 1):
+        role = "mask" if classes[i - 1] == "LoadImageMask" else "image"
+        nodes[str(i)]["_meta"] = {"title": f"Reference (Input:{role})"}
+    return nodes
+
+
+def create(client, workflow, schema=None):
+    analyzed = client.post("/api/workflows/analyze", json={"workflow_json": workflow})
+    assert analyzed.status_code == 200
+    response = client.post(
+        "/api/workflows",
+        json={
+            "name": "HEL809 input contract",
+            "workflow_json": workflow,
+            "input_schema": analyzed.json()["inputs"] if schema is None else schema,
+        },
+    )
+    assert response.status_code == 200, response.text
+    return response.json()
+
+
+def assert_views(client, wid, expected):
+    detail = client.get(f"/api/workflows/{wid}").json()
+    assert detail["media_io"] == expected
+    for query in ("", "?view=meta"):
+        rows = client.get("/api/workflows" + query).json()
+        row = next(r for r in rows if r["id"] == wid)
+        assert row["media_io"] == expected
+        # This is the library's exact media predicate, fed by the meta API.
+        for media in ("image", "video", "audio"):
+            hits = [
+                r["id"]
+                for r in rows
+                if media in r["media_io"]["in"] or media in r["media_io"]["out"]
+            ]
+            assert (wid in hits) == (media in expected["in"] or media in expected["out"])
+        if query:
+            assert not {"workflow_json", "input_schema", "output_schema"} & row.keys()
+
+
+def stored_media(wid):
+    with get_db(settings.db_path) as conn:
+        return json.loads(
+            conn.execute("SELECT media_io FROM workflows WHERE id = ?", (wid,)).fetchone()[
+                "media_io"
+            ]
+        )
+
+
+@pytest.mark.parametrize(
+    "classes", [("LoadImage",), ("LoadImageMask",), ("LoadImage", "LoadImageMask"), ()]
+)
+def test_analyze_create_persists_media_and_agrees_with_library(client, classes):
+    row = create(client, graph(*classes))
+    expected = {"in": ["image"] if classes else [], "out": []}
+    if "LoadImageMask" in classes:
+        assert any(e["role"] == "mask" and e["kind"] == "image" for e in row["input_schema"])
+    assert stored_media(row["id"]) == expected
+    assert row["model_family"] is None  # never infer a model or output from kind
+    assert_views(client, row["id"], expected)
+
+
+def test_create_uses_confirmed_schema_not_original_graph(client):
+    row = create(client, graph(), [{"nodeId": "90", "kind": "audio_list", "manual": True}])
+    assert stored_media(row["id"]) == {"in": ["audio"], "out": []}
+    assert_views(client, row["id"], {"in": ["audio"], "out": []})
+
+
+def test_update_changes_inputs_preserving_explicit_outputs_and_other_metadata(client):
+    row = create(client, graph("LoadImage"))
+    with get_db(settings.db_path) as conn:
+        conn.execute(
+            "UPDATE workflows SET media_io = ?, model_family = ?, tags = ? WHERE id = ?",
+            (json.dumps({"in": ["image"], "out": []}), "existing", '["tag"]', row["id"]),
+        )
+    for kind, expected in (("audio_list", ["audio"]), ("text", []), ("video_list", ["video"])):
+        response = client.patch(
+            f"/api/workflows/{row['id']}",
+            json={
+                "input_schema": [{"nodeId": "90", "kind": kind, "manual": True}],
+            },
+        )
+        assert response.status_code == 200, response.text
+        assert response.json()["model_family"] == "existing"
+        assert response.json()["tags"] == ["tag"]
+        assert stored_media(row["id"]) == {"in": expected, "out": []}
+        assert_views(client, row["id"], {"in": expected, "out": []})
+
+
+@pytest.mark.parametrize(
+    "raw, expected",
+    [
+        (None, {"in": ["image"], "out": ["image"]}),
+        ('{"in": [], "out": []}', {"in": [], "out": []}),
+        ('{"out": []}', {"in": ["image"], "out": []}),
+        ('{"in": []}', {"in": [], "out": ["image"]}),
+    ],
+)
+def test_legacy_null_and_missing_keys_are_distinct_from_empty_lists(client, raw, expected):
+    workflow = graph("LoadImageMask")
+    workflow["99"] = {"class_type": "SaveImage", "inputs": {"images": ["1", 0]}}
+    row = create(client, workflow)
+    with get_db(settings.db_path) as conn:
+        conn.execute("UPDATE workflows SET media_io = ? WHERE id = ?", (raw, row["id"]))
+    assert_views(client, row["id"], expected)
+    with get_db(settings.db_path) as conn:
+        assert (
+            conn.execute("SELECT media_io FROM workflows WHERE id = ?", (row["id"],)).fetchone()[
+                "media_io"
+            ]
+            == raw
+        )  # read only fallback
+    if raw is None:
+        response = client.patch(f"/api/workflows/{row['id']}", json={"favorite": True})
+        assert response.status_code == 200
+        assert stored_media(row["id"]) == expected
+
+
+def test_legacy_prompt_only_has_no_invented_media(client):
+    row = create(client, graph())
+    with get_db(settings.db_path) as conn:
+        conn.execute("UPDATE workflows SET media_io = NULL WHERE id = ?", (row["id"],))
+    assert_views(client, row["id"], {"in": [], "out": []})
+
+
+def test_batch_import_metadata_survives_unrelated_update(client):
+    response = client.post(
+        "/api/workflows/import-batch",
+        json={
+            "files": [
+                {
+                    "source_path": "3-图像生成/Qwen-Image-2.1/demo.json",
+                    "workflow_json": _ui_graph(),
+                    "tags": ["existing batch"],
+                }
+            ]
+        },
+    )
+    assert response.status_code == 200, response.text
+    wid = response.json()["imported_ids"][0]
+    before = client.get(f"/api/workflows/{wid}").json()
+    response = client.patch(f"/api/workflows/{wid}", json={"favorite": True})
+    after = response.json()
+    for key in ("media_io", "source", "source_path", "category", "model_family", "tags"):
+        assert before[key] == after[key]
+    assert_views(client, wid, {"in": [], "out": ["image"]})
+
+
+def test_reanalyze_refreshes_inputs_and_preserves_empty_output_declaration(client):
+    row = create(client, graph("LoadImageMask"))
+    client.patch(
+        f"/api/workflows/{row['id']}",
+        json={
+            "input_schema": [{"nodeId": "90", "kind": "text"}],
+        },
+    )
+    # Legacy auto schema: read/reanalyze must rederive it before media classification.
+    with get_db(settings.db_path) as conn:
+        conn.execute(
+            "UPDATE workflows SET input_schema = ? WHERE id = ?",
+            ('[{"nodeId": "90", "kind": "text", "auto": true}]', row["id"]),
+        )
+    response = client.post(f"/api/workflows/{row['id']}/reanalyze")
+    assert response.status_code == 200
+    expected = {"in": ["image"], "out": []}
+    assert stored_media(row["id"]) == expected
+    assert_views(client, row["id"], expected)
+
+
+def test_batch_import_uses_final_confirmed_input_schema(client):
+    response = client.post(
+        "/api/workflows/import-batch",
+        json={
+            "files": [
+                {
+                    "workflow_json": _ui_graph(),
+                    "input_schema": [{"nodeId": "2", "kind": "image_list", "manual": True}],
+                }
+            ]
+        },
+    )
+    assert response.status_code == 200, response.text
+    wid = response.json()["imported_ids"][0]
+    assert stored_media(wid) == {"in": ["image"], "out": ["image"]}
+    assert_views(client, wid, {"in": ["image"], "out": ["image"]})

docs/changelogs/2026-10-23-hel809-workflow-media.md

@@ -0,0 +1,10 @@
+HEL-809 fixes single-workflow imports being shown as prompt-only and excluded from media filters. It continues draft PR #12 from `cb410fbdeaa87528aaaab1f0dbffeb0240a9a32e`, retaining all existing fixes; target is `leefer/hel-731-studio-project@2952af42b6232483640f1f27803980367cd62dd0`.
+
+- `comfyui/metadata.py`: extracts the existing input-kind normalization and real output-kind rules into `media_io_from_schema`, shared by import suggestions and persistence. Mask inputs already have kind=image; lists normalize to their media kind. Model family inference is unchanged.
+- `routers/workflows.py`: single create and schema updates persist media declarations from the final schema. Reanalyze also refreshes inputs. Batch import uses the same helper with the confirmed schema while retaining category, model family, tags and source. Existing output declarations, including empty arrays, survive schema edits; missing outputs derive only from actual parsed graph declarations, never workflow kind/usage.
+- Full and metadata-only serialization derive absent media declarations for legacy NULL rows without database writes. Present arrays, including valid empty arrays, are preserved independently per key. The lightweight list fetches/parses graph data only for records missing declarations; its response still excludes heavy schemas/graphs. The existing frontend row and filter both consume this result; no frontend product file or visual rule changes.
+- `tests/test_workflow_media_io.py`: 14 targeted regressions cover real analyze→create image/mask/prompt paths, confirmed schema overrides, update classification, NULL and partial legacy metadata versus empty arrays, no invented outputs/families, batch metadata retention, reanalyze and agreement among detail/full/meta library views. The previous batch test expecting missing declarations now expects the graph's actual image→video declaration.
+
+Validation: Python 3.12 relevant backend suites: 93 passed, one existing Starlette deprecation warning. Ruff on changed product/new regression files and git diff whitespace check pass. Unchanged frontend production build passes. `e2e/hel809.cjs` and `e2e/run-hel809.py` use the same frontend build against frozen pre-fix and current backend, capture actual drawer analyze/create requests and responses plus meta/detail reads, and assert old prompt-only/filter miss versus corrected image/filter hit. Pure prompt row retains its label and is excluded from the image filter. Screenshots/responses/logs and exact final head/base are delivered on HEL-809.
+
+Reproduce: use an existing backend Python environment with repository dependencies and build `calliope-web`; set `PLAYWRIGHT_MODULE_PATH` and browser library path for the machine, then run `python e2e/run-hel809.py`. The runner uses loopback-only ports 18249/18250, checks availability first, pauses generation, points ComfyUI/LLM to a closed loopback port, starts no real model service, terminates its exact backend children and removes its temporary databases in finally. Evidence remains in the caller's workdir. No new dependency, migration, production data/config, deployment or merge. Rollback this task's commit only; existing NULL records remain compatible with the preceding version.

e2e/hel809.cjs

@@ -0,0 +1,69 @@
+// Targeted real drawer -> HTTP -> database -> library regression; no generation.
+const { chromium } = require(process.env.PLAYWRIGHT_MODULE_PATH || 'playwright');
+const fs = require('node:fs'), path = require('node:path'), http = require('node:http');
+const assert = require('node:assert/strict');
+const phase = process.env.HEL809_PHASE, out = process.env.HEL809_EVIDENCE;
+const graph = {
+ '1': {class_type:'PrimitiveStringMultiline',inputs:{value:'isolated test prompt'},_meta:{title:'Prompt (Input:prompt)'}},
+ '2': {class_type:'LoadImage',inputs:{image:'isolated.png'},_meta:{title:'Image (Input:image)'}},
+ '3': {class_type:'LoadImageMask',inputs:{image:'isolated.png',channel:'red'},_meta:{title:'Mask (Input:mask)'}},
+ '4': {class_type:'SaveImage',inputs:{images:['2',0]},_meta:{title:'Result (Output:image)'}}
+};
+const build = path.resolve('calliope-web/build');
+const mime = {'.js':'text/javascript','.css':'text/css','.html':'text/html;charset=utf-8','.json':'application/json','.svg':'image/svg+xml','.woff2':'font/woff2'};
+const sockets = new Set();
+const server = http.createServer((req,res)=>{
+ if(req.url.startsWith('/api/')) {
+  const proxy=http.request({hostname:'127.0.0.1',port:18249,path:req.url,method:req.method,headers:req.headers},r=>{res.writeHead(r.statusCode,r.headers);r.pipe(res)});
+  proxy.on('error',()=>{res.writeHead(502);res.end()});req.pipe(proxy);return;
+ }
+ let file=path.join(build,new URL(req.url,'http://localhost').pathname);
+ if(!fs.existsSync(file)||fs.statSync(file).isDirectory())file=path.join(build,'index.html');
+ res.setHeader('Content-Type',mime[path.extname(file)]||'application/octet-stream');fs.createReadStream(file).pipe(res);
+});
+server.on('connection',s=>{sockets.add(s);s.on('close',()=>sockets.delete(s))});
+(async()=>{
+ await new Promise((resolve,reject)=>{server.once('error',reject);server.listen(18250,'127.0.0.1',resolve)});
+ let browser; const errors=[],responses=[],checks=[];
+ try {
+  browser=await chromium.launch({args:['--no-sandbox']});
+  const context=await browser.newContext({viewport:{width:1440,height:900}});
+  await context.addInitScript(()=>{localStorage.setItem('aigc-lang','zh');localStorage.setItem('aigc-theme','night')});
+  const page=await context.newPage();page.on('pageerror',e=>errors.push(e.message));
+  page.on('response',r=>{if(r.url().includes('/api/workflows') && ['POST','PATCH'].includes(r.request().method()))responses.push((async()=>({method:r.request().method(),url:new URL(r.url()).pathname,request:r.request().postDataJSON(),status:r.status(),response:await r.json()}))())});
+  await page.goto('http://127.0.0.1:18250/workflows');
+  await page.getByRole('button',{name:'导入工作流',exact:true}).click();
+  await page.getByPlaceholder('Qwen Image 2.1 文生图').fill('HEL809 图像与遮罩');
+  await page.locator('textarea.json').fill(JSON.stringify(graph,null,2));
+  await page.getByRole('button',{name:'分析',exact:true}).click();
+  await page.getByRole('button',{name:'确认保存',exact:true}).click();
+  const row=page.locator('tbody tr').filter({hasText:'HEL809 图像与遮罩'});
+  await row.waitFor();
+  await page.screenshot({path:path.join(out,phase+'-library.png'),fullPage:true});
+  const rowText=await row.innerText();
+  assert(rowText.includes(phase==='before'?'仅提示词':'图像'),rowText);checks.push('real drawer row label');
+  await page.getByRole('combobox',{name:'素材类型',exact:true}).selectOption('image');
+  // The output is image too: media filter uses both input and output declarations.
+  if(phase==='before')await row.waitFor({state:'hidden'});else await row.waitFor();
+  await page.screenshot({path:path.join(out,phase+'-image-filter.png'),fullPage:true});
+  checks.push('image filter '+(phase==='before'?'miss reproduced':'hit'));
+  const api=async(p,b,method)=>{const r=await fetch('http://127.0.0.1:18249/api'+p,{method:method||(b?'POST':'GET'),headers:{'Content-Type':'application/json'},body:b?JSON.stringify(b):undefined});assert(r.ok,await r.clone().text());return r.json()};
+  const meta=await api('/workflows?view=meta'),detail=await api('/workflows/'+meta[0].id);
+  assert(detail.input_schema.some(x=>x.role==='mask'&&x.kind==='image'));checks.push('mask is image');
+  if(phase==='after'){
+   assert.deepEqual(meta[0].media_io,{in:['image'],out:['image']});
+   const promptGraph={'1':graph['1']};
+   const analyzed=await api('/workflows/analyze',{workflow_json:promptGraph});
+   const prompt=await api('/workflows',{name:'HEL809 纯提示词',workflow_json:promptGraph,input_schema:analyzed.inputs,usage:'text2image'});
+   assert.deepEqual(prompt.media_io,{in:[],out:[]});
+   await page.reload();await page.getByRole('combobox',{name:'素材类型',exact:true}).selectOption('all');
+   const promptRow=page.locator('tbody tr').filter({hasText:'HEL809 纯提示词'});await promptRow.waitFor();assert((await promptRow.innerText()).includes('仅提示词'));
+   await page.getByRole('combobox',{name:'素材类型',exact:true}).selectOption('image');await promptRow.waitFor({state:'hidden'});await row.waitFor();
+   checks.push('pure prompt label and exclusion from image filter');
+   await page.screenshot({path:path.join(out,phase+'-prompt-excluded.png'),fullPage:true});
+  }
+  const network=await Promise.all(responses);
+  fs.writeFileSync(path.join(out,phase+'-responses.json'),JSON.stringify({network,meta,detail,rowText,checks,errors},null,2));
+  assert.deepEqual(errors,[]);console.log(phase,checks);
+ } finally {if(browser)await browser.close();for(const s of sockets)s.destroy();await new Promise(r=>server.close(r));}
+})().catch(e=>{console.error(e);process.exitCode=1});

e2e/run-hel809.py

@@ -0,0 +1,69 @@
+"""Foreground isolated before/after API + browser proof; always remove own state."""
+import io
+import json
+import os
+from pathlib import Path
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import tarfile
+import tempfile
+import time
+import urllib.request
+
+root = Path(__file__).resolve().parents[1]
+os.chdir(root)
+out = root.parent / 'evidence/hel809'
+out.mkdir(parents=True, exist_ok=True)
+for port in (18249, 18250):
+    with socket.socket() as s:
+        assert s.connect_ex(('127.0.0.1', port)) != 0, f'Port in use: {port}'
+bootstrap = '''from pathlib import Path
+import os
+from calliope import config
+root = Path(os.environ['HEL809_DATA'])
+config.CONFIG_FILE = root / 'settings.json'
+config.settings = config.Settings(data_dir=root/'data', assets_dir=root/'data/assets',
+    comfyui_base_url='http://127.0.0.1:1', llm_base_url='http://127.0.0.1:1/v1', dry_run=False)
+config.settings.ensure_storage_dirs()
+from calliope.queue.manager import queue_manager
+queue_manager.paused = True
+from calliope.main import create_app
+import uvicorn
+uvicorn.run(create_app(),host='127.0.0.1',port=18249,log_level='warning')
+'''
+cleanup = []
+with tempfile.TemporaryDirectory(prefix='hel809-', dir=out) as tmp:
+    work = Path(tmp)
+    old = work/'before'; old.mkdir()
+    archive = subprocess.check_output(['git','archive','cb410fbdeaa87528aaaab1f0dbffeb0240a9a32e','calliope-backend/src'])
+    with tarfile.open(fileobj=io.BytesIO(archive)) as bundle:
+        bundle.extractall(old, filter='data')
+    server = work/'server.py'; server.write_text(bootstrap)
+    for phase, src in [('before',old/'calliope-backend/src'),('after',root/'calliope-backend/src')]:
+        data = work/phase/'runtime'; data.mkdir(parents=True)
+        env = {**os.environ, 'PYTHONPATH':str(src), 'HEL809_DATA':str(data),
+               'HEL809_PHASE':phase,'HEL809_EVIDENCE':str(out)}
+        with (out/(phase+'-backend.log')).open('w') as log:
+            child = subprocess.Popen([sys.executable,str(server)],env=env,stdout=log,stderr=subprocess.STDOUT)
+            try:
+                for attempt in range(100):
+                    try:
+                        urllib.request.urlopen('http://127.0.0.1:18249/api/health',timeout=1);break
+                    except Exception:
+                        if child.poll() is not None: raise RuntimeError('Backend failed: '+phase)
+                        if attempt==99:raise
+                        time.sleep(.1)
+                subprocess.run(['node','e2e/hel809.cjs'],env=env,check=True,timeout=100)
+            finally:
+                daemon = json.loads(subprocess.check_output(['multica','daemon','status','--output','json']))
+                assert child.pid != daemon['pid']
+                if child.poll() is None:child.send_signal(signal.SIGTERM)
+                try:child.wait(timeout=10)
+                except subprocess.TimeoutExpired:child.kill();child.wait()
+                cleanup.append({'phase':phase,'pid':child.pid,'exit':child.returncode})
+    # TemporaryDirectory removes only this run's database and before-source copy.
+(out/'cleanup.json').write_text(json.dumps({'children':cleanup,'temporary_state_removed':not work.exists()},indent=2))
+print('CLEANUP',cleanup, 'state removed',not work.exists())