fix(builder): stabilize retained knowledge goal waiting (ENG-1449)

This commit is contained in:
Charles Yao committed 2026-11-10 00:57:01 -08:00
1 parent 016ca0c35e
commit af9401a98d
2 files changed
+404 -19

No files matched your search

@@ -22,7 +22,7 @@ from core.dify_builder.contract import (
ResourceSelectCard,
TestResultCard,
)
from core.dify_builder.errors import DraftWouldNotStartError, HashMismatchError
from core.dify_builder.errors import DraftWouldNotStartError, HashMismatchError, NotFoundError
from core.dify_builder.input_schema import schema_hash, validate_runtime_inputs
from core.dify_builder.models import (
ConversationItem,
@@ -582,7 +582,32 @@ def handle_initial_plan(env: Env, turn: Turn, s: Session, fc: DifyBuilderContext
def _knowledge_scope(fc: DifyBuilderContext) -> str:
return canonical_digest({"goal": fc.goal_text, "generation": fc.task_control.get("generation", 0)})
"""Bind goal intent to its logical task, independently of execution consent.
Resume changes the execution generation, not the user's goal decision.
Legacy empty task IDs remain session-local through the owning repository;
a new logical task or a changed goal invalidates this scope.
"""
return canonical_digest({"goal": fc.goal_text, "logical_task_id": fc.logical_task_id})
def _knowledge_scope_matches(fc: DifyBuilderContext, decision: dict, *, effective: bool = False) -> bool:
"""Validate stored goal scope without upgrading legacy execution authority.
New decisions survive Resume only within the same logical task and goal.
Unmarked legacy decisions require their original current-generation scope;
unknown scope kinds refuse. This performs no I/O or context mutation.
"""
digest = (
decision.get("effective_scope_digest", decision.get("scope_digest"))
if effective
else decision.get("scope_digest")
)
if decision.get("scope_kind") == "logical_task":
return digest == _knowledge_scope(fc)
if "scope_kind" not in decision:
return digest == canonical_digest({"goal": fc.goal_text, "generation": fc.task_control.get("generation", 0)})
return False
def _knowledge_waived(fc: DifyBuilderContext) -> bool:
@@ -590,7 +615,7 @@ def _knowledge_waived(fc: DifyBuilderContext) -> bool:
return (
isinstance(decision, dict)
and decision.get("choice") == "waive"
and decision.get("effective_scope_digest", decision.get("scope_digest")) == _knowledge_scope(fc)
and _knowledge_scope_matches(fc, decision, effective=True)
and bool(decision.get("card_key"))
)
@@ -648,6 +673,7 @@ def _ask_knowledge_decision(env: Env, turn: Turn, s: Session, fc: DifyBuilderCon
fc.resource_selection["knowledge_decision_pending"] = {
"card_key": key,
"scope_digest": _knowledge_scope(fc),
"scope_kind": "logical_task",
"original_goal": fc.goal_text,
"actor_id": turn.actor.account_id,
"tenant_id": turn.actor.tenant_id,
@@ -683,17 +709,19 @@ def _ask_knowledge_decision(env: Env, turn: Turn, s: Session, fc: DifyBuilderCon
def _handle_knowledge_decision(env: Env, turn: Turn, s: Session, fc: DifyBuilderContext) -> StepResult:
"""Apply only the current actor's explicit choice; retain keeps its form open.
"""Apply only a current, actor-bound explicit goal choice.
Repeated retain choices append no new message and perform no cognition or
workflow operations. Keeping the pending nonce lets the user later choose
a goal change through the same current form, with ordinary version checks.
First retain acknowledges a selected waiting form; repeated retains append
no card/message and perform no cognition or workflow operations. The nonce
still permits explicit later adjustment with ordinary version checks.
Valid legacy submissions gain task scope. A resumed legacy gate can only
reoffer after its saved task/goal is verified; the old choice is ignored.
Foreign or changed-task/goal choices change nothing.
"""
pending = fc.resource_selection.get("knowledge_decision_pending")
payload = turn.action.payload if turn.action and isinstance(turn.action.payload, dict) else {}
if (
not isinstance(pending, dict)
or pending.get("scope_digest") != _knowledge_scope(fc)
or pending.get("actor_id") != turn.actor.account_id
or pending.get("tenant_id") != turn.actor.tenant_id
):
@@ -703,11 +731,15 @@ def _handle_knowledge_decision(env: Env, turn: Turn, s: Session, fc: DifyBuilder
value = payload.get(key) if isinstance(key, str) else None
if not isinstance(options, list) or len(options) != 2 or value not in options:
return StepResult(next=PcState.BUILD_GOAL_ANALYSIS, context=fc)
if not _knowledge_scope_matches(fc, pending):
return _restore_legacy_knowledge_decision(env, turn, s, fc, pending)
choice = "retain" if value == options[0] else "waive"
previous = fc.resource_selection.get("knowledge_scope_decision")
if choice == "retain" and isinstance(previous, dict) and previous.get("card_key") == key:
if previous.get("choice") == "retain":
if previous.get("choice") == "retain" and previous.get("scope_kind") == "logical_task":
return StepResult(next=PcState.BUILD_GOAL_ANALYSIS, context=fc)
pending = {**pending, "scope_digest": _knowledge_scope(fc), "scope_kind": "logical_task"}
fc.resource_selection["knowledge_decision_pending"] = pending
decision = {**pending, "choice": choice, "user_choice": value}
fc.resource_selection["knowledge_scope_decision"] = decision
if choice == "retain":
@@ -741,13 +773,59 @@ def _handle_knowledge_decision(env: Env, turn: Turn, s: Session, fc: DifyBuilder
return StepResult(next=PcState.BUILD_INITIAL_PLAN, context=fc)
def _restore_legacy_knowledge_decision(
env: Env, turn: Turn, s: Session, fc: DifyBuilderContext, pending: dict
) -> StepResult:
"""Reoffer a resumed legacy gate only from matching saved task/goal facts.
The caller has validated the current actor and submitted old control.
Missing history, unknown scope kinds, or changed tasks/goals refuse without
mutation. Success issues one fresh nonce and task scope; it never accepts
the old choice, waives a goal, refreshes resources, or runs a workflow.
"""
unchanged = StepResult(next=PcState.BUILD_GOAL_ANALYSIS, context=fc)
source_version = pending.get("source_version")
if (
"scope_kind" in pending
or pending.get("original_goal") != fc.goal_text
or type(source_version) is not int
or not 1 <= source_version < s.version
):
return unchanged
try:
original_session, original = env.repo.get_session_at_version(s.id, source_version)
except NotFoundError:
return unchanged
if (
original.goal_text != fc.goal_text
or original.logical_task_id != fc.logical_task_id
or original_session.owner_account_id != s.owner_account_id
or original_session.tenant_id != s.tenant_id
or pending.get("scope_digest")
!= canonical_digest({"goal": original.goal_text, "generation": original.task_control.get("generation", 0)})
):
return unchanged
result = _ask_knowledge_decision(env, turn, s, fc)
message = (
"已恢复当前任务的知识库目标确认。先前表单来自恢复任务之前,请使用新表单重新选择;刚才的旧表单提交未更改目标,也未生成或执行工作流。"
if fc.reply_language.lower().startswith("zh")
else (
"The knowledge-goal confirmation for this task has been restored. The earlier form predates Resume; "
"choose using the refreshed form. The old submission did not change the goal, generate or run a workflow."
)
)
result.items.extend(append_assistant(env, s, fc, message, cards=["form"]))
return result
def _knowledge_wait(
env: Env, s: Session, fc: DifyBuilderContext, *, unavailable: bool = False, execution: Any = None
) -> StepResult:
"""Explain the capability block while preserving the current decision gate.
"""Acknowledge retained intent without asking the unanswered question again.
A pending retain/waive form stays actionable; an invalid resource selection
stays at resource selection. Neither path authorizes planning or execution.
Valid first retention projects its accepted value in the existing form
contract, leaving explicit adjustment and Stop available. Invalid resource
selections stay at resource selection. Neither authorizes execution.
"""
chinese = fc.reply_language.lower().startswith("zh")
if unavailable:
@@ -770,7 +848,31 @@ def _knowledge_wait(
"support; this selection did not generate or execute a workflow."
)
)
items = append_assistant(env, s, fc, message, execution=execution)
items = []
pending = fc.resource_selection.get("knowledge_decision_pending")
if not unavailable and isinstance(pending, dict):
items.extend(
append_card(
fc,
FormCard(
variant="build_requirements",
title="已保留知识库目标,等待后续能力"
if chinese
else "Knowledge goal retained; waiting for support",
description=(
"当前保留选择已记录,无需重复提交。可通过下方选项明确调整目标,也可停止任务。"
if chinese
else (
"Your choice to retain the goal is recorded; no resubmission is needed. "
"Use the options below to explicitly adjust the goal, or stop the task."
)
),
fields=build_form_fields(fc.form_fields),
values={pending["card_key"]: pending["options"][0]},
),
)
)
items.extend(append_assistant(env, s, fc, message, execution=execution, cards=["form"] if items else None))
waiting_state = (
PcState.BUILD_GOAL_ANALYSIS
if fc.resource_selection.get("knowledge_decision_pending")
@@ -2,7 +2,7 @@
from dataclasses import replace
from datetime import datetime
from typing import NoReturn, override
from typing import TYPE_CHECKING, NoReturn, override
import pytest
@@ -26,6 +26,13 @@ from tests.unit_tests.services.dify_builder.test_knowledge_source_cue import (
MEETING_GOAL,
)
if TYPE_CHECKING:
from services.dify_builder.repository import SqlDifyBuilderRepository
from services.dify_builder.service import DifyBuilderService
from tests.unit_tests.services.dify_builder.test_service import Enqueued
pytest_plugins = ["tests.unit_tests.services.dify_builder.test_service"]
DATASET = "25f76fa8-db3b-499d-b7fc-2bfbddad778a"
KNOWLEDGE = "knowledge:" + DATASET
MODEL = "provider/model"
@@ -612,6 +619,9 @@ def test_refusing_goal_change_keeps_current_actionable_form_and_does_not_repeat_
assert view.active_interaction.action_id == "submit_requirements"
assert view.active_interaction.valid_at_version == result.version
assert view.active_interaction.card.payload["fields"][0]["key"] == key
# Retention is an accepted choice, not the original unanswered question.
assert view.active_interaction.card.payload["values"] == {key: context.form_fields[0]["options"][0]}
assert len([item for item in repo.list_conversation(sid) if item.kind == "form"]) == 2
assert _internal_action_allowed(result.current_state, context, "message")
assert _internal_action_allowed(result.current_state, context, "stop")
conversation = repo.list_conversation(sid)
@@ -635,6 +645,167 @@ def test_refusing_goal_change_keeps_current_actionable_form_and_does_not_repeat_
assert agent.requests == requests
@pytest.mark.parametrize("logical_task_id", ["", "stable-knowledge-task"])
@pytest.mark.parametrize("legacy", [False, True])
def test_retained_goal_stays_adjustable_after_sql_stop_resume(
monkeypatch: pytest.MonkeyPatch,
service: "DifyBuilderService",
repo: "SqlDifyBuilderRepository",
enqueued: list["Enqueued"],
logical_task_id: str,
legacy: bool,
) -> None:
from sqlalchemy import func, select
from core.dify_builder.models import Graph
from models.dify_builder import DifyBuilderRun
from runtime_contracts.builder.execution import canonical_digest
from services.dify_builder.agent.llm_agent import LlmBuilderAgent
from services.dify_builder.dify_port import WorkflowServiceDifyPort
from services.dify_builder.task_control_service import claim_continuation, refresh_continuation
from tests.unit_tests.services.dify_builder.test_service import ACCOUNT_ID, APP_ID, TENANT_ID, _actor
reads: list[tuple[str, str | None]] = []
resolutions: list[str] = []
model_calls: list[dict[str, object]] = []
def catalog(
tenant_id: str, *, account_id: str | None = None, selected_agent_ids: frozenset[str] = frozenset()
) -> resources.TenantResources:
assert selected_agent_ids == frozenset()
reads.append((tenant_id, account_id))
return inventory()
def resolve() -> object:
resolutions.append("model")
return object()
def invoke(_model: object, **kwargs: object) -> dict[str, object]:
model_calls.append(kwargs)
return {"resource_ids": []}
monkeypatch.setattr(resources, "list_tenant_resources", catalog)
monkeypatch.setattr(build.llm, "invoke_json", invoke)
monkeypatch.setattr(build, "assess_capability_gap", lambda *_a, **_k: "")
agent = LlmBuilderAgent(TENANT_ID, account_id=ACCOUNT_ID, model_resolver_fn=resolve)
session = Session(
app_id=APP_ID,
owner_account_id=ACCOUNT_ID,
tenant_id=TENANT_ID,
entry_mode=EntryMode.BUILD,
current_state=PcState.BUILD_RESOURCE_RECOMMENDATION,
)
goal = "必须从知识库检索依据后回答"
repo.create_session(
session,
DifyBuilderContext(
goal_text=goal,
reply_language="zh-Hans",
logical_task_id=logical_task_id,
task_status="active",
resource_selection={"offered_knowledge_ids": [KNOWLEDGE]},
),
[],
)
dify = FakeDifyPort()
def refuse_effect(*_args: object, **_kwargs: object) -> NoReturn:
raise AssertionError("A knowledge waiting/adjustment gate cannot execute a workflow effect")
for operation in ("run_draft", "apply_repair", "publish", "restore_graph"):
monkeypatch.setattr(dify, operation, refuse_effect)
runner = Runner(Env(agent=agent, repo=repo, dify=dify, now=lambda: datetime.min), build_registry())
def submit(kind: str, payload: dict[str, object]) -> Session:
current, _ = repo.get_session(session.id)
action = Action(kind=kind, payload=payload, base_version=current.version)
service._prepare_action(session.id, _actor(), action)
return runner.advance(session.id, Turn(actor=_actor(), action=action))
submit("confirm_resources", {"resource_ids": [KNOWLEDGE]})
field = repo.get_session(session.id)[1].form_fields[0]
key, options = field["key"], field["options"]
retained = submit("submit_requirements", {key: options[0]})
_, context = repo.get_session(session.id)
if legacy:
# Persist the pre-fix context shape, then use the real Resume owner.
for name in ("knowledge_decision_pending", "knowledge_scope_decision"):
context.resource_selection[name].pop("scope_kind")
context.resource_selection[name]["scope_digest"] = canonical_digest({"goal": goal, "generation": 0})
repo.compare_and_advance(session.id, retained.version, retained.current_state, context, [])
retained, context = repo.get_session(session.id)
decision = dict(context.resource_selection["knowledge_scope_decision"])
waiting = service._build_session_view(retained, context).active_interaction
assert waiting is not None
conversation = repo.list_conversation(session.id)
# A GET/refresh reads the same accepted card without rediscovery/cognition.
assert service.get_session_view(session.id, _actor()).active_interaction == waiting
assert repo.list_conversation(session.id) == conversation
service._read_task_dsl_fn = lambda *_: (Graph(graph=Graph(nodes=[], edges=[])), "waiting-revision")
stopped = service.stop_task(session.id, _actor(), base_version=retained.version, client_action_id="knowledge-stop")
service.resume_task(session.id, _actor(), base_version=stopped.version, client_action_id="knowledge-resume")
continuation = enqueued[-1][1]
claim_continuation(repo, session.id, _actor(), continuation)
continuation_port = WorkflowServiceDifyPort()
monkeypatch.setattr(continuation_port, "read_task_dsl", service._read_task_dsl_fn)
refresh_continuation(repo, continuation_port, session.id, _actor(), continuation)
runner.advance(session.id, Turn(actor=_actor(), action=continuation))
resumed, context = repo.get_session(session.id)
assert context.task_control["generation"] == 1
assert context.logical_task_id == logical_task_id
assert context.goal_text == goal
assert context.resource_selection["knowledge_preferences"] == [KNOWLEDGE]
assert context.resource_selection["knowledge_scope_decision"] == decision
resumed_view = service._build_session_view(resumed, context)
assert resumed_view.active_interaction is not None
assert resumed_view.active_interaction.card == waiting.card
assert resumed_view.active_interaction.valid_at_version == resumed.version
if legacy:
restoration = submit("submit_requirements", {key: options[1]})
_, context = repo.get_session(session.id)
assert restoration.current_state == PcState.BUILD_GOAL_ANALYSIS
assert context.goal_text == goal
assert context.resource_selection["knowledge_scope_decision"] == decision
fresh = context.form_fields[0]
assert fresh["key"] != key
key, options = fresh["key"], fresh["options"]
submit("submit_requirements", {key: options[0]})
conversation = repo.list_conversation(session.id)
repeated = submit("submit_requirements", {key: options[0]})
assert [item for item in repo.list_conversation(session.id) if item.kind in {"form", "assistant_turn"}] == [
item for item in conversation if item.kind in {"form", "assistant_turn"}
]
assert repeated.current_state == PcState.BUILD_GOAL_ANALYSIS
assert (reads, resolutions, model_calls) == ([(TENANT_ID, ACCOUNT_ID)], [], [])
assert context.task_status == "active"
assert not context.checkpoint_id
assert not context.verify_run_id
assert not context.built_node_ids
assert dify.applied == []
assert dify.run_draft_inputs == {}
assert not dify.published
with repo._session_factory() as sql:
assert sql.scalar(select(func.count()).select_from(DifyBuilderRun)) == 0
plan_calls: list[dict[str, object]] = []
def revised_plan(requirements: dict[str, object], **_kwargs: object) -> WorkflowPlan:
plan_calls.append(requirements)
return WorkflowPlan("Format text", "Text", "Formatted text", ["Format the supplied text"])
monkeypatch.setattr(agent, "propose_plan_v1", revised_plan)
adjusted = submit("submit_requirements", {key: options[1]})
_, context = repo.get_session(session.id)
assert adjusted.current_state == PcState.BUILD_RESOURCE_RECOMMENDATION
assert context.resource_selection["knowledge_scope_decision"]["choice"] == "waive"
assert context.resource_selection["knowledge_scope_decision"]["original_goal"] == goal
assert context.goal_text.startswith(goal)
assert context.goal_text != goal
assert len(plan_calls) == 1
assert len(reads) == 2
assert len(resolutions) == len(model_calls) == 1
@pytest.mark.parametrize("selected", [False, True])
@pytest.mark.parametrize("retain_first", [False, True])
def test_explicit_goal_change_is_traceable_and_allows_no_knowledge_plan(selected: bool, retain_first: bool) -> None:
@@ -742,22 +913,134 @@ def test_replaying_resolved_scope_card_does_not_reauthorize_or_replan() -> None:
assert len(agent.requests) == calls
def test_decision_is_rejected_after_goal_or_actor_changes() -> None:
@pytest.mark.parametrize("already_retained", [False, True])
@pytest.mark.parametrize("resumed", [False, True])
def test_legacy_goal_form_requires_its_original_generation_before_task_scope_upgrade(
already_retained: bool, resumed: bool
) -> None:
from runtime_contracts.builder.execution import canonical_digest
runner, repo, sid, agent = decision_dialog()
session, context = repo.get_session(sid)
pending = context.resource_selection["knowledge_decision_pending"]
pending.pop("scope_kind", None)
pending["scope_digest"] = canonical_digest({"goal": context.goal_text, "generation": 0})
if already_retained:
context.resource_selection["knowledge_scope_decision"] = {**pending, "choice": "retain"}
if resumed:
context.task_control["generation"] = 1
previous = dict(context.resource_selection)
repo.compare_and_advance(sid, session.version, session.current_state, context, [])
conversation = repo.list_conversation(sid)
result = answer_decision(runner, repo, sid, option=1 if resumed else 0)
_, stored = repo.get_session(sid)
assert result.current_state == PcState.BUILD_GOAL_ANALYSIS
assert stored.goal_text == context.goal_text
assert agent.requests == []
if resumed:
# Restore a new gate; the submitted old choice grants no authority.
fresh = stored.resource_selection["knowledge_decision_pending"]
assert fresh["card_key"] != pending["card_key"]
assert fresh["scope_kind"] == "logical_task"
assert stored.resource_selection.get("knowledge_scope_decision") == previous.get("knowledge_scope_decision")
restored = repo.list_conversation(sid)
assert (
len([item for item in restored if item.kind == "form"])
== len([item for item in conversation if item.kind == "form"]) + 1
)
# An old option cannot submit the new current card or cause another restore.
runner.advance(
sid,
Turn(
actor=Actor(account_id="actor", tenant_id="tenant"),
action=Action(
kind="submit_requirements",
payload={pending["card_key"]: pending["options"][1]},
base_version=result.version,
),
),
)
assert repo.list_conversation(sid) == restored
assert repo.get_session(sid)[1].goal_text == context.goal_text
answer_decision(runner, repo, sid, option=0)
waiting = repo.list_conversation(sid)
answer_decision(runner, repo, sid, option=0)
assert repo.list_conversation(sid) == waiting
assert agent.requests == []
adjusted = answer_decision(runner, repo, sid, option=1)
assert adjusted.current_state == PcState.BUILD_RESOURCE_RECOMMENDATION
assert repo.get_session(sid)[1].resource_selection["knowledge_scope_decision"]["choice"] == "waive"
assert len(agent.requests) == 1
else:
assert stored.resource_selection["knowledge_scope_decision"]["choice"] == "retain"
assert stored.resource_selection["knowledge_decision_pending"]["scope_kind"] == "logical_task"
card = repo.list_conversation(sid)[-2]
assert card.kind == "form"
assert card.payload["values"] == {pending["card_key"]: pending["options"][0]}
assert not stored.checkpoint_id
assert not stored.verify_run_id
@pytest.mark.parametrize("changed", ["goal", "task", "actor", "tenant", "missing_history"])
def test_stale_legacy_form_cannot_restore_for_a_different_goal_task_or_actor(changed: str) -> None:
from runtime_contracts.builder.execution import canonical_digest
runner, repo, sid, agent = decision_dialog()
session, context = repo.get_session(sid)
pending = context.resource_selection["knowledge_decision_pending"]
pending.pop("scope_kind", None)
pending["scope_digest"] = canonical_digest({"goal": context.goal_text, "generation": 0})
context.task_control["generation"] = 1
actor = Actor(account_id="actor", tenant_id="tenant")
if changed == "goal":
context.goal_text = "A different current goal"
if changed == "task":
context.logical_task_id = "another-task"
if changed == "actor":
actor = replace(actor, account_id="another-actor")
if changed == "tenant":
actor = replace(actor, tenant_id="another-tenant")
if changed == "missing_history":
pending["source_version"] = 10000
version = repo.compare_and_advance(sid, session.version, session.current_state, context, [])
conversation = repo.list_conversation(sid)
runner.advance(
sid,
Turn(
actor=actor,
action=Action(
kind="submit_requirements",
payload={pending["card_key"]: pending["options"][1]},
base_version=version,
),
),
)
assert repo.list_conversation(sid) == conversation
assert "knowledge_scope_decision" not in repo.get_session(sid)[1].resource_selection
assert agent.requests == []
@pytest.mark.parametrize("option", [0, 1])
def test_decision_is_rejected_after_goal_task_or_actor_changes(option: int) -> None:
from services.dify_builder.application.handlers_build import handle_goal_analysis
runner, repo, sid, agent = decision_dialog()
del runner
session, ctx = repo.get_session(sid)
field = ctx.form_fields[0]
for goal, actor in [("Different goal", "actor"), (ctx.goal_text, "another-actor")]:
changed = replace(ctx, goal_text=goal)
for changed, actor, tenant in [
(replace(ctx, goal_text="Different goal"), "actor", "tenant"),
(replace(ctx, logical_task_id="another-task"), "actor", "tenant"),
(replace(ctx), "another-actor", "tenant"),
(replace(ctx), "actor", "another-tenant"),
]:
result = handle_goal_analysis(
Env(agent=agent, repo=repo, dify=FakeDifyPort(), now=lambda: datetime.min),
Turn(
actor=Actor(account_id=actor, tenant_id="tenant"),
actor=Actor(account_id=actor, tenant_id=tenant),
action=Action(
kind="submit_requirements",
payload={field["key"]: field["options"][1]},
payload={field["key"]: field["options"][option]},
base_version=session.version,
),
),