feat(dify-builder): live Edit test via run_draft, fail routes to repair gate

Rewrite handle_test_affected_paths from a canned always-success stub to a
live handler mirroring Build's handle_test_and_repair (A5): derive mock
inputs once via start_schema/generate_mock_inputs and persist a TestInput
(reused on retest), run run_draft (degrading to a synthesized failed Run on
exception), and mint a real immutable Run. Pass -> edit.review with a green
TestResultCard + summary. Fail -> real diagnose/propose_repair, stage
fc.diagnosis/staged_repair/risk, emit a red TestResultCard + ErrorCard (+
ChangeSetCard when intents are proposed) -> edit.await_repair (the B1 gate).
No auto-apply; apply_repair still runs only inside handle_await_repair's
approve_repair branch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

api/core/dify_builder/handlers_edit.py

@@ -18,6 +18,7 @@ from core.dify_builder.contract import (
     ChangeSetCard,
     CheckpointCard,
     DecisionItem,
+    ErrorCard,
     FormCard,
     PlanCard,
     PublishCard,
@@ -35,11 +36,13 @@ from core.dify_builder.handlers_fix import (
     build_change_set,
     build_form_fields,
     emit_canvas,
+    first_failed_node,
     merge_known_keys,
     mint_checkpoint,
     perform_revert,
+    start_schema,
 )
-from core.dify_builder.models import DifyBuilderContext, Session, Turn
+from core.dify_builder.models import DifyBuilderContext, Run, Session, TestInput, Turn
 from core.dify_builder.runner import Env, Handler, StepResult
 from core.dify_builder.state import PcState
 
@@ -272,53 +275,130 @@ def handle_apply_changes(env: Env, turn: Turn, s: Session, fc: DifyBuilderContex
 
 
 def handle_test_affected_paths(env: Env, turn: Turn, s: Session, fc: DifyBuilderContext) -> StepResult:
-    """(working, auto) Canned affected-path test: emit a success test_result +
-    the review summary + change_set, transition to edit.review. No live
-    run_draft (spec: canned)."""
-    emit_canvas(env, "mark_test_success")
-    test_result_items = append_card(
+    """(working, auto) Live affected-path test via run_draft. Success ->
+    edit.review; failure -> real diagnose + propose_repair, staged for the
+    edit.await_repair approval gate. No auto-apply (human-gated)."""
+    graph, _hash = env.dify.read_graph(s.app_id, turn.actor)
+    if fc.test_input_ref == "":
+        inputs = env.agent.generate_mock_inputs(start_schema(graph), {})
+        ti = TestInput(session_id=s.id, source="mock", inputs=inputs)
+        env.repo.save_test_input(ti)
+        fc.test_input_ref = ti.id
+    else:
+        inputs = env.repo.get_test_input(fc.test_input_ref).inputs
+
+    emit = env.emit if env.emit is not None else (lambda _e: None)
+    try:
+        raw = env.dify.run_draft(s.app_id, turn.actor, inputs, emit)
+        status, per_node, dify_run_id = raw.status, raw.per_node, raw.dify_run_id
+    except Exception:  # never crash the advance -- surface as a failed run
+        status, per_node, dify_run_id = "failed", [], ""
+
+    run = Run(
+        id=str(uuid.uuid4()),
+        kind="verify",
+        dify_run_id=dify_run_id,
+        status=status,
+        per_node=per_node,
+        inputs_ref=fc.test_input_ref,
+        immutable=True,
+    )
+
+    if status == "succeeded":
+        emit_canvas(env, "mark_test_success")
+        test_items = append_card(
+            fc,
+            TestResultCard(
+                title="Affected-path tests",
+                subtitle="All checks passed",
+                tone="success",
+                stats=[TestStat(value="1", label="runs"), TestStat(value="0", label="errors")],
+                run_ids=[run.id],
+            ),
+        )
+        emit_canvas(env, "mark_review_ready")
+        summary_items = append_card(
+            fc,
+            SummaryCard(
+                variant="review",
+                title="Review",
+                items=["Applied the change plan", "Affected paths tested", "Tests passing"],
+            ),
+        )
+        turn_items = append_card(
+            fc,
+            AssistantTurnItem(
+                turn_id=str(uuid.uuid4()),
+                stage_id="edit.review",
+                trace=Trace(status="completed", steps=[]),
+                reply_text="Tests passed; ready for review.",
+                cards=["test_result", "summary"],
+            ),
+        )
+        return StepResult(
+            next=PcState.EDIT_REVIEW,
+            context=fc,
+            items=[*test_items, *summary_items, *turn_items],
+            run=run,
+            run_id_sink=[run.id],
+        )
+
+    # failure: real diagnosis + proposed repair, staged for the approval gate
+    run.culprit_node_id = first_failed_node(per_node)
+    fc.verify_run_id = run.id
+    emit_canvas(env, "mark_test_error")
+    diagnosis = env.agent.diagnose(run, graph, per_node)
+    intents, risk = env.agent.propose_repair(diagnosis, graph)
+    fc.diagnosis = diagnosis
+    fc.staged_repair = list(intents)
+    fc.risk = risk
+    test_items = append_card(
         fc,
         TestResultCard(
             title="Affected-path tests",
-            subtitle="All checks passed",
-            tone="success",
-            stats=[TestStat(value="1", label="runs"), TestStat(value="0", label="errors")],
-            run_ids=[],
+            subtitle="Failed",
+            tone="error",
+            stats=[TestStat(value="1", label="runs"), TestStat(value="1", label="errors")],
+            run_ids=[run.id],
         ),
     )
-    changed = list(fc.change_set.changed_nodes) if fc.change_set else []
-    change_set_items = append_card(
+    error_items = append_card(
         fc,
-        ChangeSetCard(
-            count=len(changed),
-            changes=[f"edited {nid}" for nid in changed],
-            scope="configuration",
-            full_diff_open=False,
+        ErrorCard(
+            title="Test failed",
+            body=diagnosis.root_cause or "The run failed.",
+            tone="danger",
+            node_id=diagnosis.culprit_node_id,
         ),
     )
-    emit_canvas(env, "mark_review_ready")
-    summary_items = append_card(
-        fc,
-        SummaryCard(
-            variant="review",
-            title="Review",
-            items=["Applied the change plan", "Affected paths tested", "Tests passing"],
-        ),
+    proposed = [f"{i.op} {i.args.get('node_id', '')}".strip() for i in intents]
+    cs_items = (
+        append_card(
+            fc, ChangeSetCard(count=len(intents), changes=proposed, scope="configuration", full_diff_open=False)
+        )
+        if intents
+        else []
     )
     turn_items = append_card(
         fc,
         AssistantTurnItem(
             turn_id=str(uuid.uuid4()),
-            stage_id="edit.review",
+            stage_id="edit.await_repair",
             trace=Trace(status="completed", steps=[]),
-            reply_text="Tests passed; ready for review.",
-            cards=["test_result", "change_set", "summary"],
+            reply_text=(
+                "Test failed — here's a proposed fix to review."
+                if intents
+                else "Test failed — no safe automatic fix; edit or keep draft."
+            ),
+            cards=["test_result", "error"] + (["change_set"] if intents else []),
         ),
     )
     return StepResult(
-        next=PcState.EDIT_REVIEW,
+        next=PcState.EDIT_AWAIT_REPAIR,
         context=fc,
-        items=[*test_result_items, *change_set_items, *summary_items, *turn_items],
+        items=[*test_items, *error_items, *cs_items, *turn_items],
+        run=run,
+        run_id_sink=[run.id],
     )
 
 

api/tests/unit_tests/core/dify_builder/test_handlers_edit.py

@@ -15,18 +15,18 @@ from core.dify_builder.models import (
 from core.dify_builder.placeholder_agent import PlaceholderAgent
 from core.dify_builder.runner import Env, Runner
 from core.dify_builder.state import PcState
-from tests.unit_tests.core.dify_builder.fakes import FakeEditDifyPort, InMemoryRepository
+from tests.unit_tests.core.dify_builder.fakes import FakeEditDifyPort, InMemoryRepository, StubAgent
 
 
 def _actor() -> Actor:
     return Actor(account_id="acc-1", tenant_id="tenant-1")
 
 
-def _new_env(dify=None, emit_canvas=None) -> tuple[Env, InMemoryRepository]:
+def _new_env(dify=None, emit_canvas=None, agent=None) -> tuple[Env, InMemoryRepository]:
     repo = InMemoryRepository()
     env = Env(
         dify=dify or FakeEditDifyPort(),
-        agent=PlaceholderAgent(),
+        agent=agent or PlaceholderAgent(),
         repo=repo,
         now=lambda: datetime.min,
         emit_canvas=emit_canvas,
@@ -240,22 +240,90 @@ def test_apply_changes_revert_records_intent_only():
     assert {"event": "revert_checkpoint"} in events
 
 
-def test_test_affected_paths_emits_result_and_reaches_review():
+def test_edit_test_pass_goes_to_review_with_real_run():
+    from core.dify_builder.handlers_edit import handle_test_affected_paths
+
+    env, _ = _new_env()  # FakeEditDifyPort.verify_pass True by default
+    s = _session(entry_mode=EntryMode.EDIT, current_state=PcState.EDIT_TEST_AFFECTED_PATHS)
+    fc = DifyBuilderContext(edit_target_node_ids=["llm"])
+    result = handle_test_affected_paths(env, Turn(actor=_actor()), s, fc)
+
+    assert result.next == PcState.EDIT_REVIEW
+    assert result.run is not None
+    assert result.run.status == "succeeded"
+    assert result.context.test_input_ref  # inputs generated + persisted
+    test_result = next(i for i in result.items if i.kind == "test_result")
+    assert test_result.payload["tone"] == "success"
+    assistant = next(i for i in result.items if i.kind == "assistant_turn")
+    assert assistant.payload["cards"] == ["test_result", "summary"]
+
+
+def test_edit_test_fail_routes_to_await_repair():
     from core.dify_builder.handlers_edit import handle_test_affected_paths
 
     events: list[dict] = []
-    env, repo = _new_env(emit_canvas=events.append)
-    s = _seed_edit_session(repo, PcState.EDIT_TEST_AFFECTED_PATHS, edit_target_node_ids=["llm"])
-    res = handle_test_affected_paths(env, Turn(actor=_actor()), *repo.get_session(s.id))
+    env, _ = _new_env(agent=StubAgent(), emit_canvas=events.append)
+    env.dify.verify_pass = False
+    s = _session(entry_mode=EntryMode.EDIT, current_state=PcState.EDIT_TEST_AFFECTED_PATHS)
+    fc = DifyBuilderContext(edit_target_node_ids=["llm"])
+    result = handle_test_affected_paths(env, Turn(actor=_actor()), s, fc)
 
-    assert res.next == PcState.EDIT_REVIEW
-    kinds = [i.kind for i in res.items]
-    assert "test_result" in kinds
-    summary = next(i for i in res.items if i.kind == "summary")
-    assert summary.payload["variant"] == "review"
-    names = [e["event"] for e in events]
-    assert "mark_test_success" in names
-    assert "mark_review_ready" in names
+    assert result.next == PcState.EDIT_AWAIT_REPAIR
+    assert result.run is not None
+    assert result.run.status == "failed"
+    # StubAgent.propose_repair returns a repair -> staged
+    assert result.context.staged_repair
+    # card content: a red test_result, an error card carrying the real
+    # diagnosis (StubAgent.diagnose's culprit/root_cause), and a change_set
+    # since a repair was proposed -- the assistant_turn's cards list reflects
+    # exactly that trio.
+    test_result = next(i for i in result.items if i.kind == "test_result")
+    assert test_result.payload["tone"] == "error"
+    error_card = next(i for i in result.items if i.kind == "error")
+    assert error_card.payload["body"] == "Output node requires 'metrics'"
+    assert error_card.payload["node_id"] == "output"
+    assistant = next(i for i in result.items if i.kind == "assistant_turn")
+    assert assistant.payload["cards"] == ["test_result", "error", "change_set"]
+    assert {"event": "mark_test_error"} in events
+
+
+def test_edit_test_fail_with_no_proposed_repair_still_routes_to_gate():
+    """When propose_repair finds no safe fix (empty intents), the fail path
+    must still route to the gate, but WITHOUT a change_set card, and with the
+    "no safe automatic fix" reply_text variant -- the `if intents` branch the
+    handler takes to decide between the two card/reply-text shapes."""
+    from core.dify_builder.handlers_edit import handle_test_affected_paths
+    from core.dify_builder.models import Risk
+
+    env, _ = _new_env()
+    env.dify.verify_pass = False
+    env.agent.propose_repair = lambda _diagnosis, _graph: (
+        [],
+        Risk(level="high", reason="no fix", has_external_side_effect=False),
+    )
+    s = _session(entry_mode=EntryMode.EDIT, current_state=PcState.EDIT_TEST_AFFECTED_PATHS)
+    fc = DifyBuilderContext(edit_target_node_ids=["llm"])
+    result = handle_test_affected_paths(env, Turn(actor=_actor()), s, fc)
+
+    assert result.next == PcState.EDIT_AWAIT_REPAIR
+    assert result.context.staged_repair == []
+    kinds = [i.kind for i in result.items]
+    assert "change_set" not in kinds
+    assistant = next(i for i in result.items if i.kind == "assistant_turn")
+    assert assistant.payload["cards"] == ["test_result", "error"]
+    assert assistant.payload["reply_text"] == "Test failed — no safe automatic fix; edit or keep draft."
+
+
+def test_edit_test_reuses_persisted_inputs_on_retest():
+    from core.dify_builder.handlers_edit import handle_test_affected_paths
+
+    env, _ = _new_env()
+    s = _session(entry_mode=EntryMode.EDIT, current_state=PcState.EDIT_TEST_AFFECTED_PATHS)
+    fc = DifyBuilderContext(edit_target_node_ids=["llm"], test_input_ref="")
+    handle_test_affected_paths(env, Turn(actor=_actor()), s, fc)
+    ref = fc.test_input_ref
+    handle_test_affected_paths(env, Turn(actor=_actor()), s, fc)
+    assert fc.test_input_ref == ref  # reused, not regenerated
 
 
 def test_review_publish_reaches_terminal_edit_publish_with_publish_card():