diff --git a/api/core/workflow_copilot/handlers_fix.py b/api/core/workflow_copilot/handlers_fix.py index 22c6884fb48..42563c30c04 100644 --- a/api/core/workflow_copilot/handlers_fix.py +++ b/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/ diff --git a/api/core/workflow_copilot/ports.py b/api/core/workflow_copilot/ports.py index 7e884c40c04..daec254be88 100644 --- a/api/core/workflow_copilot/ports.py +++ b/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: ... diff --git a/api/core/workflow_copilot/runner.py b/api/core/workflow_copilot/runner.py index cf70ec016d1..d6d45bf4414 100644 --- a/api/core/workflow_copilot/runner.py +++ b/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 diff --git a/api/services/workflow_copilot/dify_port.py b/api/services/workflow_copilot/dify_port.py index c37692d8210..e7e6e3268e3 100644 --- a/api/services/workflow_copilot/dify_port.py +++ b/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="") diff --git a/api/tasks/workflow_copilot_advance_task.py b/api/tasks/workflow_copilot_advance_task.py index 0d4ba0b255d..a04a340c446 100644 --- a/api/tasks/workflow_copilot_advance_task.py +++ b/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/ diff --git a/api/tests/unit_tests/core/workflow_copilot/fakes.py b/api/tests/unit_tests/core/workflow_copilot/fakes.py index 2df996b6154..49b65c012b4 100644 --- a/api/tests/unit_tests/core/workflow_copilot/fakes.py +++ b/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( diff --git a/api/tests/unit_tests/core/workflow_copilot/test_handlers_fix.py b/api/tests/unit_tests/core/workflow_copilot/test_handlers_fix.py index e6bed1f5981..5fd924ca9f5 100644 --- a/api/tests/unit_tests/core/workflow_copilot/test_handlers_fix.py +++ b/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 diff --git a/api/tests/unit_tests/services/workflow_copilot/test_advance_task.py b/api/tests/unit_tests/services/workflow_copilot/test_advance_task.py index 566d9110851..87429eaf8bd 100644 --- a/api/tests/unit_tests/services/workflow_copilot/test_advance_task.py +++ b/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 diff --git a/api/tests/unit_tests/services/workflow_copilot/test_dify_port.py b/api/tests/unit_tests/services/workflow_copilot/test_dify_port.py index b3b9f3f8ad9..268ea942987 100644 --- a/api/tests/unit_tests/services/workflow_copilot/test_dify_port.py +++ b/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 --------------------------------------------------------------