feat(workflow-copilot): opt-in per-intent canvas-event emission through apply_repair

api/core/workflow_copilot/handlers_fix.py

@@ -198,7 +198,7 @@ def handle_apply(env: Env, turn: Turn, s: Session, fc: FixContext) -> StepResult
     branches on ``fc.source``: the run-fix path always lands at
     ``fix.await_verify``; the checklist-fix path lands at
     ``checklist.await_recheck``. Port of ``handlers_fix.go:146``."""
-    result = env.dify.apply_repair(s.app_id, turn.actor, fc.staged_repair)
+    result = env.dify.apply_repair(s.app_id, turn.actor, fc.staged_repair, on_canvas=env.emit_canvas)
     fc.last_snapshot_hash = result.new_hash
     # Adapters that don't compute a real diff (e.g. FakeDifyPort in tests)
     # leave changes/scope empty -- fall back to the old changed_nodes/

api/core/workflow_copilot/ports.py

@@ -87,7 +87,13 @@ class DifyPort(Protocol):
 
     def node_outputs(self, app_id: str, actor: Actor, run_id: str) -> list[NodeOutput]: ...
 
-    def apply_repair(self, app_id: str, actor: Actor, intents: list[MutationIntent]) -> ApplyResult: ...
+    def apply_repair(
+        self,
+        app_id: str,
+        actor: Actor,
+        intents: list[MutationIntent],
+        on_canvas: Callable[[dict], None] | None = None,
+    ) -> ApplyResult: ...
 
     def run_draft(self, app_id: str, actor: Actor, inputs: Inputs, on_event: Callable[[NodeEvent], None]) -> Run: ...
 

api/core/workflow_copilot/runner.py

@@ -50,6 +50,12 @@ class Env:
     # usecase sets it per-Advance to publish to the session broadcaster; None
     # is treated as a no-op.
     emit: Callable[[NodeEvent], None] | None = None
+    # Emit forwards granular canvas mutations (add_*_node, apply_error_fix,
+    # ...) as each MutationIntent is applied. None is treated as a no-op --
+    # the caller (a handler) decides whether to wire it (opt-in, spec Sec 6):
+    # Build wants one event per intent, Edit suppresses these and fires one
+    # coarse apply_edit_plan itself instead.
+    emit_canvas: Callable[[dict], None] | None = None
 
 
 @dataclass

api/services/workflow_copilot/dify_port.py

@@ -41,6 +41,7 @@ from typing import Any
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.workflow_copilot.contract import CanvasEvent
 from core.workflow_copilot.models import (
     Actor,
     ApplyResult,
@@ -52,6 +53,7 @@ from core.workflow_copilot.models import (
     Run,
 )
 from extensions.ext_database import db
+from graphon.enums import BuiltinNodeTypes
 from libs.datetime_utils import naive_utc_now
 from models.model import App
 from models.workflow import Workflow
@@ -83,6 +85,40 @@ _APPLY_FNS: dict[str, Callable[..., tuple[Graph, list[str]]]] = {
     "insert_between": apply_insert_between,
 }
 
+_CREATE_NODE_CANVAS_EVENTS: dict[str, CanvasEvent] = {
+    BuiltinNodeTypes.START: CanvasEvent.ADD_START_NODE,
+    BuiltinNodeTypes.KNOWLEDGE_RETRIEVAL: CanvasEvent.ADD_KNOWLEDGE_NODE,
+    BuiltinNodeTypes.LLM: CanvasEvent.ADD_LLM_NODE,
+    BuiltinNodeTypes.END: CanvasEvent.ADD_OUTPUT_NODE,
+}
+
+
+def _canvas_event_for_intent(intent: MutationIntent) -> CanvasEvent:
+    """Map an applied ``MutationIntent`` to the canvas event that narrates it
+    (spec Sec 6). ``create_node`` dispatches by ``node_type`` to the matching
+    ``add_*_node`` event; ``set_node_config`` (Fix's only verb today) maps to
+    ``apply_error_fix``; every other verb (``delete_node``/``connect``/
+    ``insert_between``) and any unmapped ``node_type`` default to the
+    generic ``apply_edit_plan`` batch-mutation event.
+    """
+    if intent.op == "create_node":
+        return _CREATE_NODE_CANVAS_EVENTS.get(intent.args.get("node_type", ""), CanvasEvent.APPLY_EDIT_PLAN)
+    if intent.op == "set_node_config":
+        return CanvasEvent.APPLY_ERROR_FIX
+    return CanvasEvent.APPLY_EDIT_PLAN
+
+
+def _canvas_payload(intent: MutationIntent, changed: list[str]) -> dict[str, Any]:
+    """Build the ``{"event": ..., "node_id"?/"edge"?}`` dict passed to ``on_canvas``."""
+    payload: dict[str, Any] = {"event": str(_canvas_event_for_intent(intent))}
+    if intent.op == "connect":
+        payload["edge"] = {"source": intent.args.get("from_node"), "target": intent.args.get("to_node")}
+    elif intent.op == "delete_node":
+        payload["node_id"] = intent.args.get("node_id")
+    else:  # create_node | insert_between | set_node_config
+        payload["node_id"] = changed[0] if changed else None
+    return payload
+
 
 def _session_factory() -> sessionmaker[Session]:
     """One session per adapter call -- never shared/held across calls."""
@@ -135,7 +171,13 @@ class WorkflowServiceDifyPort:
             for node_exec in node_execs
         ]
 
-    def apply_repair(self, app_id: str, actor: Actor, intents: list[MutationIntent]) -> ApplyResult:
+    def apply_repair(
+        self,
+        app_id: str,
+        actor: Actor,
+        intents: list[MutationIntent],
+        on_canvas: Callable[[dict], None] | None = None,
+    ) -> ApplyResult:
         with _session_factory()() as session:
             account = resolve_account(session, actor)
             app = load_app(session, app_id, actor)
@@ -152,6 +194,8 @@ class WorkflowServiceDifyPort:
                 validate_intent_args(intent)
                 graph, changed = apply_fn(graph, **intent.args)
                 changed_nodes.extend(changed)
+                if on_canvas is not None:
+                    on_canvas(_canvas_payload(intent, changed))
 
             if not changed_nodes:
                 return ApplyResult(changed_nodes=[], new_hash=unique_hash, changes=[], scope="")

api/tasks/workflow_copilot_advance_task.py

@@ -65,8 +65,11 @@ def advance_session(session_id: str, action_dict: dict, actor_dict: dict, token:
                 {"kind": "node", "node_id": ne.node_id, "title": ne.title, "status": ne.status, "error": ne.error},
             )
 
+        def emit_canvas(event: dict) -> None:
+            progress_bus.publish(session_id, {"kind": "canvas", **event})
+
         actor = Actor(**actor_dict)
-        env = Env(dify=dify, agent=agent, repo=repo, now=naive_utc_now, emit=emit)
+        env = Env(dify=dify, agent=agent, repo=repo, now=naive_utc_now, emit=emit, emit_canvas=emit_canvas)
         runner = Runner(env, fix_registry())
         runner.advance(session_id, Turn(action=Action(**action_dict), actor=actor))
         # Project a SessionView (single source of truth for phase/run_status/

api/tests/unit_tests/core/workflow_copilot/fakes.py

@@ -235,10 +235,19 @@ class FakeDifyPort:
     def node_outputs(self, _app_id: str, _actor: Actor, _run_id: str) -> list[NodeOutput]:
         return [NodeOutput(node_id="output", status="failed", error="missing metrics")]
 
-    def apply_repair(self, _app_id: str, _actor: Actor, intents: list[MutationIntent]) -> ApplyResult:
+    def apply_repair(
+        self,
+        _app_id: str,
+        _actor: Actor,
+        intents: list[MutationIntent],
+        on_canvas: Callable[[dict], None] | None = None,
+    ) -> ApplyResult:
         self.applied = copy.deepcopy(intents)
         self.hash = "h1"
         changed = [intent.args["node_id"] for intent in intents if isinstance(intent.args.get("node_id"), str)]
+        if on_canvas is not None:
+            for node_id in changed:
+                on_canvas({"event": "apply_error_fix", "node_id": node_id})
         return ApplyResult(changed_nodes=changed, new_hash=self.hash)
 
     def run_draft(

api/tests/unit_tests/core/workflow_copilot/test_handlers_fix.py

@@ -160,6 +160,36 @@ def test_apply_writes_graph_locks_during_work_unlocks_at_await_verify():
     assert "output" in fc.change_set.changed_nodes
 
 
+def test_apply_forwards_env_emit_canvas_to_the_adapters_on_canvas_callback():
+    events: list[dict] = []
+    repo = InMemoryRepository()
+    env = Env(
+        dify=FakeDifyPort(),
+        agent=StubAgent(),
+        repo=repo,
+        now=lambda: datetime.min,
+        emit_canvas=events.append,
+    )
+    s = _seed_diagnose_session(repo)
+
+    runner = Runner(env, _run_fix_registry())
+    turn = Turn(action=Action(kind="request_fix", base_version=1), actor=_actor())
+    runner.advance(s.id, turn)
+
+    assert events == [{"event": "apply_error_fix", "node_id": "output"}]
+
+
+def test_apply_without_env_emit_canvas_does_not_error():
+    env, repo = _new_env()  # emit_canvas defaults to None
+    s = _seed_diagnose_session(repo)
+
+    runner = Runner(env, _run_fix_registry())
+    turn = Turn(action=Action(kind="request_fix", base_version=1), actor=_actor())
+    out = runner.advance(s.id, turn)
+
+    assert out.current_state == PcState.FIX_AWAIT_VERIFY
+
+
 def test_apply_emits_change_set_card_with_fallback_scope_and_changes_when_adapter_omits_them():
     # NOTE: this assertion already holds against the OLD hardcoded handle_apply
     # (scope="configuration", changes=changed_nodes) -- Task 8's fallback branch

api/tests/unit_tests/services/workflow_copilot/test_advance_task.py

@@ -116,6 +116,10 @@ def test_advance_session_drives_state_forward_emits_events_and_releases_lock(rep
     # table entries are run_validation (primary) + revert (destructive).
     assert [a["id"] for a in last_state_event["actions"]] == ["run_validation", "revert"]
 
+    canvas_events = [ev for _sid, ev in events if ev["kind"] == "canvas"]
+    assert canvas_events, "advance_session must publish canvas events once the adapter emits them"
+    assert canvas_events[0]["event"] == "apply_error_fix"
+
     assert released == [(s.id, "tok-1")]
 
     # 2) continue driving: run_verify -> fix.await_testdata -> provide_testdata

api/tests/unit_tests/services/workflow_copilot/test_dify_port.py

@@ -349,6 +349,88 @@ def test_apply_repair_computes_configuration_scope_for_set_node_config(mock_sess
     assert result.changes == ["node-1: code updated"]
 
 
+def test_apply_repair_invokes_on_canvas_once_per_applied_intent(mock_session: MagicMock):
+    account = SimpleNamespace(id="acc-1")
+    app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
+    _configure_session_get(mock_session, account=account, app=app)
+
+    workflow = SimpleNamespace(
+        graph_dict={"nodes": [{"id": "node-1", "data": {}}], "edges": []},
+        unique_hash="hash-1",
+        features_dict={},
+        conversation_variables=[],
+    )
+    updated_workflow = SimpleNamespace(unique_hash="hash-2")
+    intents = [MutationIntent(op="set_node_config", args={"node_id": "node-1", "path": "code", "value": "x"})]
+    events: list[dict] = []
+
+    with patch("services.workflow_copilot.dify_port.WorkflowService") as mock_ws_cls:
+        mock_ws_cls.return_value.get_draft_workflow.return_value = workflow
+        mock_ws_cls.return_value.sync_draft_workflow.return_value = updated_workflow
+
+        WorkflowServiceDifyPort().apply_repair("app-1", _actor(), intents, on_canvas=events.append)
+
+    assert events == [{"event": "apply_error_fix", "node_id": "node-1"}]
+
+
+@pytest.mark.parametrize(
+    ("node_type", "expected_event"),
+    [
+        ("start", "add_start_node"),
+        ("knowledge-retrieval", "add_knowledge_node"),
+        ("llm", "add_llm_node"),
+        ("end", "add_output_node"),
+        ("code", "apply_edit_plan"),  # unmapped node type falls back to the generic batch-mutate event
+    ],
+)
+def test_apply_repair_maps_create_node_by_node_type_to_the_right_add_node_event(
+    mock_session: MagicMock, node_type, expected_event
+):
+    account = SimpleNamespace(id="acc-1")
+    app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
+    _configure_session_get(mock_session, account=account, app=app)
+
+    workflow = SimpleNamespace(
+        graph_dict={"nodes": [], "edges": []},
+        unique_hash="hash-1",
+        features_dict={},
+        conversation_variables=[],
+    )
+    updated_workflow = SimpleNamespace(unique_hash="hash-2")
+    intents = [MutationIntent(op="create_node", args={"node_type": node_type, "config": {}})]
+    events: list[dict] = []
+
+    with patch("services.workflow_copilot.dify_port.WorkflowService") as mock_ws_cls:
+        mock_ws_cls.return_value.get_draft_workflow.return_value = workflow
+        mock_ws_cls.return_value.sync_draft_workflow.return_value = updated_workflow
+
+        WorkflowServiceDifyPort().apply_repair("app-1", _actor(), intents, on_canvas=events.append)
+
+    assert events[0]["event"] == expected_event
+
+
+def test_apply_repair_skips_on_canvas_when_not_provided(mock_session: MagicMock):
+    account = SimpleNamespace(id="acc-1")
+    app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
+    _configure_session_get(mock_session, account=account, app=app)
+
+    workflow = SimpleNamespace(
+        graph_dict={"nodes": [{"id": "node-1", "data": {}}], "edges": []},
+        unique_hash="hash-1",
+        features_dict={},
+        conversation_variables=[],
+    )
+    updated_workflow = SimpleNamespace(unique_hash="hash-2")
+    intents = [MutationIntent(op="set_node_config", args={"node_id": "node-1", "path": "code", "value": "x"})]
+
+    with patch("services.workflow_copilot.dify_port.WorkflowService") as mock_ws_cls:
+        mock_ws_cls.return_value.get_draft_workflow.return_value = workflow
+        mock_ws_cls.return_value.sync_draft_workflow.return_value = updated_workflow
+
+        # must not raise even though on_canvas is omitted (default None)
+        WorkflowServiceDifyPort().apply_repair("app-1", _actor(), intents)
+
+
 # ---- run_draft --------------------------------------------------------------