zyt/server/tests/DifyDeployedStopPersistenceR26cDiagnostic.py

"""Offline diagnostics over exact saved deployed source ASTs; no Dify imports.

Extracted functions/subtrees execute unchanged against memory-only Redis/queue,
repository, events and worker stubs. No network, real DB, thread or stop call.
"""
import ast
import copy
import hashlib
import json
from pathlib import Path
import queue
from types import SimpleNamespace as NS
from enum import Enum, IntEnum
from contextlib import contextmanager

OUT = Path(__file__).resolve().parents[2] / "app/artifacts/ai_concurrency"
PINS = {
    "v43r26c_deployed_dify_stop_source.snapshot.json": "700f05516c29e325319a2fe9bfb7c008bbfc9810ec3211fb7614bac08e13cd38",
    "v43r26c_deployed_dify_stop_source2.snapshot.json": "2e053d67c98ec321f5da19745a605a86ede83c919a49dd7b8476612397825dee",
    "v43r26c_deployed_dify_stop_source3.snapshot.json": "7357ea267484307c781dfaa6d05665cc3e5cc9bf19306237f2b671c20e51c408",
}
checks = 0
failures = []
observations = []
files = {}
trees = {}
extracted = []


def check(ok, label):
    global checks
    checks += 1
    if not ok:
        failures.append(label)


for name, pin in PINS.items():
    raw = (OUT / name).read_bytes()
    check(hashlib.sha256(raw).hexdigest() == pin, "exact source snapshot pin")
    for f in json.loads(raw)["files"]:
        data = f["text"].encode("utf-8")
        check(len(data) == f["bytes"], "snapshot embedded byte count")
        check(hashlib.sha256(data).hexdigest() == f["sha256"], "snapshot embedded hash")
        check(f["path"] not in files, "unique source path")
        files[f["path"]] = f
        trees[f["path"]] = ast.parse(f["text"], filename=f["path"])


def method(path, cls, name):
    c = next(n for n in trees[path].body if isinstance(n, ast.ClassDef) and n.name == cls)
    n = next(n for n in c.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == name)
    extracted.append(dict(path=path, cls=cls, method=name, line=n.lineno, end_line=n.end_lineno))
    return copy.deepcopy(n)


def compile_node(node, env):
    node.decorator_list = []
    module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), node], type_ignores=[])
    ast.fix_missing_locations(module)
    scope = dict(env)
    exec(compile(module, "<saved-source-ast>", "exec"), scope)
    return scope[node.name]


class Logger:
    def __init__(self): self.errors = 0
    def exception(self, *args, **kwargs): self.errors += 1
    def warning(self, *args, **kwargs): self.errors += 1
    def debug(self, *args, **kwargs): pass


class Event:
    def __init__(self, **kw): self.__dict__.update(kw)
    def model_dump(self): return {}


class QueueStopEvent(Event):
    class StopBy(Enum):
        USER_MANUAL = "user"
        INPUT_MODERATION = "input"
        ANNOTATION_REPLY = "annotation"
    def get_stop_reason(self): return "synthetic stop"


class PublishFrom(IntEnum):
    APPLICATION_MANAGER = 1
    TASK_PIPELINE = 2


class GenerateTaskStoppedError(Exception): pass
class QueueErrorEvent(Event): pass
class QueueMessageEndEvent(Event): pass
class QueueAdvancedChatMessageEndEvent(Event): pass
class QueuePingEvent(Event): pass


queue_env = dict(QueueStopEvent=QueueStopEvent, QueueErrorEvent=QueueErrorEvent,
    QueueMessageEndEvent=QueueMessageEndEvent, QueueAdvancedChatMessageEndEvent=QueueAdvancedChatMessageEndEvent,
    QueuePingEvent=QueuePingEvent, MessageQueueMessage=NS, GenerateTaskStoppedError=GenerateTaskStoppedError,
    PublishFrom=PublishFrom, queue=queue)
bp = "core/app/apps/base_app_queue_manager.py"
mp = "core/app/apps/message_based_app_queue_manager.py"
publish = compile_node(method(bp, "AppQueueManager", "publish"), queue_env)
stop_listen = compile_node(method(bp, "AppQueueManager", "stop_listen"), queue_env)
private_publish = compile_node(method(mp, "MessageBasedAppQueueManager", "_publish"), queue_env)


def make_queue(stopped=False):
    q = NS(_q=queue.Queue(), _task_id="synthetic-task", _message_id="synthetic-message",
        _conversation_id="synthetic-conversation", _app_mode="advanced-chat", stopped=stopped, clear_count=0)
    q._is_stopped = lambda: q.stopped
    q._clear_task_belong_cache = lambda: setattr(q, "clear_count", q.clear_count + 1)
    q._check_for_sqlalchemy_models = lambda _: None
    q.stop_listen = lambda: stop_listen(q)
    q._publish = lambda e, p: private_publish(q, e, p)
    q.publish = lambda e, p: publish(q, e, p)
    return q


# A real listener emits QueueStopEvent using only its legacy flag or timeout.
class FakeQueue:
    def __init__(self): self.items = []; self.calls = 0
    def put(self, value): self.items.append(value)
    def get(self, **kwargs):
        self.calls += 1
        if self.calls > 8: raise AssertionError("finite queue bound")
        if not self.items: raise queue.Empty()
        return self.items.pop(0)


for cause in ["legacy_flag", "listen_timeout"]:
    q = make_queue(cause == "legacy_flag"); q._q = FakeQueue()
    clock = iter([0.0, 2.0, 2.0, 2.0, 2.0])
    listen = compile_node(method(bp, "AppQueueManager", "listen"),
        dict(queue_env, time=NS(time=lambda: next(clock)), dify_config=NS(APP_MAX_EXECUTION_TIME=1 if cause == "listen_timeout" else 100)))
    gen = listen(q); message = next(gen)
    check(isinstance(message.event, QueueStopEvent), "actual listen yields independent stop event")
    check(message.event.stopped_by == QueueStopEvent.StopBy.USER_MANUAL, "timeout also labels stop as USER_MANUAL")
    check(q.clear_count > 0, "queue stop clears belonging cache and enqueues sentinel")
    gen.close()
    observations.append(dict(case=cause, stop_event=True, engine_abort_command_sent_by_listener=False))

# The producer publishes first, then raises a local cancellation exception.
q = make_queue(True)
try:
    q.publish(Event(), PublishFrom.APPLICATION_MANAGER)
    check(False, "stopped producer should throw")
except GenerateTaskStoppedError:
    check(q._q.qsize() == 1, "producer event was queued before stop exception")


class Status(Enum):
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    STOPPED = "stopped"


graph_names = ["GraphRunStartedEvent", "GraphRunSucceededEvent", "GraphRunPartialSucceededEvent", "GraphRunFailedEvent",
    "GraphRunAbortedEvent", "GraphRunPausedEvent", "NodeRunStartedEvent", "NodeRunRetryEvent", "NodeRunSucceededEvent",
    "NodeRunFailedEvent", "NodeRunExceptionEvent", "NodeRunPauseRequestedEvent"]
events = {n: type(n, (Event,), {}) for n in graph_names}
class WorkflowExecution:
    @staticmethod
    def new(**kw): return NS(status=Status.RUNNING, outputs={}, **kw)


class Repository:
    def __init__(self, fail_terminal=False): self.persisted = None; self.saves = []; self.fail_terminal = fail_terminal
    def save(self, obj):
        if self.fail_terminal and obj.status != Status.RUNNING: raise RuntimeError("synthetic DB failure")
        self.persisted = copy.deepcopy(obj)
        self.saves.append(obj.status.value)


pp = "core/workflow/graph_engine/layers/persistence.py"
penv = dict(events, WorkflowExecution=WorkflowExecution, WorkflowExecutionStatus=Status, naive_utc_now=lambda: 1)
pmethods = {n: compile_node(method(pp, "WorkflowPersistenceLayer", n), penv) for n in
    ["on_graph_start", "on_event", "on_graph_end", "_handle_graph_run_started", "_handle_graph_run_succeeded", "_handle_graph_run_aborted", "_get_workflow_execution"]}


def persistence(fail_terminal=False):
    p = NS(_workflow_execution=None, _node_execution_cache={}, _node_snapshots={}, _node_sequence=0,
        _workflow_info=NS(workflow_id="synthetic-workflow", workflow_type="chat", version="1", graph_data={}),
        _workflow_execution_repository=Repository(fail_terminal))
    for n, f in pmethods.items(): setattr(p, n, lambda *a, _f=f, **kw: _f(p, *a, **kw))
    p._get_execution_id = lambda: "synthetic-run"
    p._prepare_workflow_inputs = lambda: {}
    p._populate_completion_statistics = lambda *a, **kw: None
    p._fail_running_node_executions = lambda *a, **kw: None
    p._enqueue_trace_task = lambda *a: None
    return p


ep = "core/workflow/graph_engine/event_management/event_manager.py"
log = Logger()
notify = compile_node(method(ep, "EventManager", "_notify_layers"), {"_logger": log})
gp = "core/workflow/graph_engine/graph_engine.py"
engine_run = compile_node(method(gp, "GraphEngine", "run"), events)
engine_stop = compile_node(method(gp, "GraphEngine", "_stop_execution"), {"logging": NS(getLogger=lambda _: log)})


def make_engine(p, *, aborted=False):
    execution = NS(started=False, paused=False, pause_reasons=[], is_paused=False, aborted=aborted,
        error=None, has_error=False, exceptions_count=0)
    execution.start = lambda: setattr(execution, "started", True)
    stopped = []
    e = NS(_graph_execution=execution, _graph_runtime_state=NS(outputs={}), _layers=[p],
        _dispatcher=NS(stop=lambda: stopped.append("dispatcher")), _worker_pool=NS(stop=lambda: stopped.append("workers")))
    em = NS(_layers=[p]); em.notify_layers = lambda event: notify(em, event); em.emit_events = lambda: iter(())
    e._event_manager = em; e._initialize_layers = p.on_graph_start; e._start_execution = lambda **kw: None
    e._stop_execution = lambda: engine_stop(e); e.run = lambda: engine_run(e)
    return e, stopped


# Close a suspended source generator before it reaches its graph terminal block.
p = persistence(); e, stopped = make_engine(p)
g = e.run(); check(type(next(g)).__name__ == "GraphRunStartedEvent", "engine start delivered")
check(p._workflow_execution_repository.persisted.status == Status.RUNNING, "start was persisted running")
g.close()
check(stopped == ["dispatcher", "workers"], "actual engine finally calls local stop subsystems")
check(p._workflow_execution_repository.saves == ["running"], "GeneratorExit path has no terminal repository save")
p.on_graph_end(None)
check(p._workflow_execution_repository.persisted.status == Status.RUNNING, "persistence on_graph_end remains no-op")
observations.append(dict(case="generator_close_before_terminal", persisted_status="running", terminal_save_count=0, local_stop_hooks=2))

# The exact AdvancedChat runner for-loop does not inject a consumer publish
# exception into the suspended graph generator. Explicit disposal then follows
# the GeneratorExit path above; no scheduler/CPython GC timing is assumed.
p = persistence(); e, _ = make_engine(p); q = make_queue(False)
e._event_manager.emit_events = lambda: iter([Event()])
runner_method = method("core/app/apps/advanced_chat/app_runner.py", "AdvancedChatAppRunner", "run")
loop = next(n for n in runner_method.body if isinstance(n, ast.For) and isinstance(n.iter, ast.Name) and n.iter.id == "generator")
wrapper = ast.FunctionDef(name="saved_runner_loop", args=ast.arguments(posonlyargs=[], args=[ast.arg(arg="self"), ast.arg(arg="generator")],
    vararg=None, kwonlyargs=[], kw_defaults=[], kwarg=None, defaults=[]), body=[copy.deepcopy(loop)], decorator_list=[])
runner_loop = compile_node(wrapper, {"workflow_entry": NS(graph_engine=e)})
def handle_event(entry, event):
    if not isinstance(event, events["GraphRunStartedEvent"]): q.stopped = True
    q.publish(Event(), PublishFrom.APPLICATION_MANAGER)
runner = NS(_handle_event=handle_event)
g = e.run()
try:
    runner_loop(runner, g)
    check(False, "actual runner for-loop must propagate producer publish stop")
except GenerateTaskStoppedError:
    check(p._workflow_execution_repository.saves == ["running"], "consumer stop exception occurs before graph terminal save")
finally:
    g.close()
check(p._workflow_execution_repository.persisted.status == Status.RUNNING, "disposing interrupted runner's graph leaves durable status running")
observations.append(dict(case="runner_producer_publish_interruption", consumer_stop_exception=True, graph_generator_explicitly_closed=True,
    graph_terminal_save_count=0, persisted_status="running", implicit_generator_disposal_timing_tested=False))

# Counter-control: an actual graph abort or success does persist before yield.
for terminal in ["aborted", "succeeded"]:
    p = persistence(); e, _ = make_engine(p, aborted=terminal == "aborted")
    result = list(e.run())
    wanted = Status.STOPPED if terminal == "aborted" else Status.SUCCEEDED
    check(p._workflow_execution_repository.persisted.status == wanted, "normal terminal path persists exact enum")
    check(p._workflow_execution_repository.saves == ["running", wanted.value], "normal terminal save follows start")
    observations.append(dict(case="normal_" + terminal, persisted_status=wanted.value, graph_terminal_emitted=type(result[-1]).__name__))

# Counter-control: swallowed persistence failure can leave DB running even when
# the graph itself emits its real aborted event.
p = persistence(True); e, _ = make_engine(p, aborted=True); before_errors = log.errors
result = list(e.run())
check(type(result[-1]).__name__ == "GraphRunAbortedEvent", "terminal event still emitted after layer save failure")
check(p._workflow_execution_repository.persisted.status == Status.RUNNING and log.errors > before_errors,
    "layer persistence exception is logged and suppressed, durable row remains running")
observations.append(dict(case="terminal_persistence_exception", graph_aborted_event=True, persisted_status="running", logged=True))

# Execute the exact pipeline stop handler with converter/database seams stubbed.
# This proves explicit STOPPED conversion + Message save, not unseen SSE encoding.
ap = "core/app/apps/advanced_chat/generate_task_pipeline.py"
stop_handler = compile_node(method(ap, "AdvancedChatAppGenerateTaskPipeline", "_handle_stop_event"),
    dict(QueueStopEvent=QueueStopEvent, WorkflowExecutionStatus=Status))
p = persistence(); p.on_graph_start(); p.on_event(events["GraphRunStartedEvent"]()); saves = []; converter_calls = []
@contextmanager
def session(): yield NS()
pipeline = NS(_workflow_run_id="synthetic-run", _workflow_id="synthetic-workflow", _application_generate_entity=NS(task_id="synthetic-task"))
pipeline._resolve_graph_runtime_state = lambda *a: NS()
pipeline._database_session = session
pipeline._save_message = lambda **kw: saves.append("message")
pipeline._message_end_to_stream_response = lambda: dict(event="message_end")
def convert(**kw): converter_calls.append(kw["status"]); return dict(event="workflow_finished", status=kw["status"].value)
pipeline._workflow_response_converter = NS(workflow_finish_to_stream_response=convert)
response = list(stop_handler(pipeline, QueueStopEvent(stopped_by=QueueStopEvent.StopBy.USER_MANUAL)))
check(response[0]["status"] == "stopped" and response[-1]["event"] == "message_end", "pipeline requests STOPPED conversion and message end")
check(saves == ["message"] and p._workflow_execution_repository.persisted.status == Status.RUNNING,
    "pipeline handler itself saves Message without graph-terminal persistence")
observations.append(dict(case="pipeline_stop_handler", requested_stream_status="stopped", message_saved=True,
    persisted_status_in_stub="running", converter_implementation_not_in_snapshots=True))

# Actual dispatcher subtree shows no command poll on empty queue or nonterminal
# events, whereas start/end and node terminal events are polling boundaries.
dp = "core/workflow/graph_engine/orchestration/dispatcher.py"
node = method(dp, "Dispatcher", "_dispatcher_loop")
empty_handlers = [h for n in ast.walk(node) if isinstance(n, ast.Try) for h in n.handlers
    if isinstance(h.type, ast.Attribute) and h.type.attr == "Empty"]
check(all(not any(isinstance(c, ast.Attribute) and c.attr == "_process_commands" for c in ast.walk(h)) for h in empty_handlers),
    "no command poll in dispatcher empty-queue branches")
process = method(dp, "Dispatcher", "_process_commands")
process_fn = compile_node(process, {})
term_type = type("NodeTerminal", (), {}); polls = []
d = NS(_COMMAND_TRIGGER_EVENTS=(term_type,), _execution_coordinator=NS(process_commands=lambda: polls.append(1)))
process_fn(d, Event()); check(polls == [], "nonterminal node event does not poll command channel")
process_fn(d, term_type()); process_fn(d); check(len(polls) == 2, "terminal and boundary events do poll command channel")
observations.append(dict(case="dispatcher_command_polling", empty_queue_poll=False, nonterminal_event_poll=False, terminal_boundary_poll=True))

# A failed graph-command send is suppressed; the route's plain method body can
# still return success. Authentication decorators are not executed or tested.
manager_path = "core/workflow/graph_engine/manager.py"
manager_log = Logger(); command_calls = []
class FailingChannel:
    def __init__(self, client, key): command_calls.append("constructed")
    def send_command(self, command): command_calls.append("send_failed"); raise RuntimeError("synthetic Redis failure")
manager = NS()
manager_env = dict(GraphEngineManager=manager, RedisChannel=FailingChannel, redis_client=NS(),
    logger=manager_log, AbortCommand=Event)
for name in ["send_stop_command", "_send_command"]:
    setattr(manager, name, compile_node(method(manager_path, "GraphEngineManager", name), manager_env))
class AppMode(Enum):
    CHAT = "chat"
    AGENT_CHAT = "agent-chat"
    ADVANCED_CHAT = "advanced-chat"
    WORKFLOW = "workflow"
    @classmethod
    def value_of(cls, value): return cls(value)
legacy = []; service = NS()
service.stop_task = compile_node(method("services/app_task_service.py", "AppTaskService", "stop_task"),
    dict(AppQueueManager=NS(set_stop_flag=lambda *a: legacy.append("flag_requested")),
        GraphEngineManager=manager, AppMode=AppMode))
route = compile_node(method("controllers/service_api/app/completion.py", "ChatStopApi", "post"),
    dict(AppMode=AppMode, AppTaskService=service, InvokeFrom=NS(SERVICE_API="api")))
reply = route(NS(), NS(mode="advanced-chat"), NS(id="synthetic-user"), "synthetic-task")
check(reply == ({"result": "success"}, 200), "plain route body returns success despite suppressed graph send failure")
check(legacy == ["flag_requested"] and command_calls == ["constructed", "send_failed"] and manager_log.errors == 1,
    "legacy request and failed graph send are separate observations")
observations.append(dict(case="stop_api_command_send_failure", response_success=True, graph_command_send_failed=True,
    command_ack_or_persisted_terminal_checked=False, auth_decorators_not_executed=True))

# Worker.stop sets an event. The exact inner node iterator does not check it;
# whether a real provider cooperates remains outside these snapshots/stubs.
wp = "core/workflow/graph_engine/worker.py"
worker_stop = compile_node(method(wp, "Worker", "stop"), {})
execute_node = compile_node(method(wp, "Worker", "_execute_node"), {})
worker = NS(_flask_app=None, _context_vars=None, _stop_event=NS(is_set=False), _event_queue=queue.Queue())
worker._stop_event.set = lambda: setattr(worker._stop_event, "is_set", True)
worker._invoke_node_run_start_hooks = lambda node: None
worker._invoke_node_run_end_hooks = lambda node, error: None
def node_events():
    yield Event()
    worker_stop(worker)
    yield Event()
node = NS(ensure_execution_id=lambda: None, run=node_events)
execute_node(worker, node)
check(worker._stop_event.is_set and worker._event_queue.qsize() == 2, "inner node iterator is not interrupted by Worker.stop event alone")
observations.append(dict(case="worker_stop_during_node_iterator", stop_flag_set=True, subsequent_synthetic_node_event=True,
    provider_cancel_observed=False))

for name, pin in PINS.items(): check(hashlib.sha256((OUT / name).read_bytes()).hexdigest() == pin, "snapshot unchanged after diagnostics")
print(json.dumps(dict(scope="offline saved deployed Dify AST diagnostic", assertions=checks, failures=failures,
    snapshots=PINS, embedded_source_files=len(files), extracted_source_methods=extracted, observations=observations,
    real_network_calls=0, real_database_calls=0, actual_stop_calls=0, real_threads_started=0, model_calls=0,
    actual_six_run_causes_proven=False, physical_gpu_release_proven=False), ensure_ascii=False, separators=(",", ":")))
raise SystemExit(1 if failures else 0)