268 lines
18 KiB
Python
268 lines
18 KiB
Python
"""Independent stdlib-only saved-AST candidate review; no Dify/runtime imports.
|
|
Uses exact domain dataclass, processor, handlers and coordinator bodies from
|
|
snapshots. Queue/channel/clock/state/pool remain explicitly synthetic seams.
|
|
"""
|
|
from __future__ import annotations
|
|
import ast
|
|
import copy
|
|
from dataclasses import dataclass, field
|
|
import difflib
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import queue
|
|
from types import SimpleNamespace as NS
|
|
|
|
OUT = Path(__file__).resolve().parents[2] / "app/artifacts/ai_concurrency"
|
|
checks = 0
|
|
failures = []
|
|
observations = []
|
|
source = {}
|
|
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",
|
|
"v43r26c_deployed_dify_stop_source4.snapshot.json": "77e1247e4ccb052d50660839263a4b133f5935d24fb191b8b89b9f18af22eda0",
|
|
"v43r26c_deployed_dify_stop_source5.snapshot.json": "223bd4c0b81aedd559d10d0225932590c51cd7fb7a338309844ff3eab1ad4ed1",
|
|
"v43r26c_dispatcher_command_polling.patched.py": "77aa94c2a0bf660469149c7bae83c7bd59aa9ebabe183fd26138d9180895a2e6",
|
|
"v43r26c_dispatcher_command_polling.patch": "a78a1b24b0628a57da2967ddb1789d10e2f009c3a0afb2f83765cd7ff21f4a90",
|
|
"v43r26c_dispatcher_polling_candidate_regression.py": "3e7e9f68264878c6390b4594e3b10f3df599323a371721d4a328389d23c895bc",
|
|
"v43r26c_dispatcher_command_polling_candidate.manifest.safe.json": "0f4a02ed6237d7863b22b9b0901452daa1472420f7ab13af3568c6933ad7fb14",
|
|
}
|
|
|
|
|
|
def check(ok, label):
|
|
global checks
|
|
checks += 1
|
|
if not ok: failures.append(label)
|
|
|
|
|
|
for name, sha in pins.items():
|
|
raw = (OUT / name).read_bytes()
|
|
check(hashlib.sha256(raw).hexdigest() == sha, "exact fixed input " + name)
|
|
if name.endswith("snapshot.json"):
|
|
for f in json.loads(raw)["files"]:
|
|
text = f["text"].encode()
|
|
check(len(text) == f["bytes"] and hashlib.sha256(text).hexdigest() == f["sha256"], "embedded exact source " + f["path"])
|
|
check(f["path"] not in source, "unique saved source path")
|
|
source[f["path"]] = f["text"]
|
|
|
|
|
|
def class_node(text, name):
|
|
return next(n for n in ast.parse(text).body if isinstance(n, ast.ClassDef) and n.name == name)
|
|
|
|
|
|
def load_class(text, name, env):
|
|
node = copy.deepcopy(class_node(text, name))
|
|
node.decorator_list = [d for d in node.decorator_list if isinstance(d, ast.Name) and d.id == "dataclass"]
|
|
for n in node.body:
|
|
if isinstance(n, ast.FunctionDef):
|
|
n.decorator_list = [d for d in n.decorator_list if not (isinstance(d, ast.Name) and d.id == "override")]
|
|
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, __name__=__name__)
|
|
exec(compile(module, "<saved-source-class>", "exec"), scope)
|
|
return scope[name]
|
|
|
|
|
|
target = "core/workflow/graph_engine/orchestration/dispatcher.py"
|
|
original = source[target]
|
|
candidate = (OUT / "v43r26c_dispatcher_command_polling.patched.py").read_text(encoding="utf-8")
|
|
check(hashlib.sha256(original.encode()).hexdigest() == "733af3883a3edfc534be3fd53324ae917c17f07c67116136473d0cad5763c6ac", "original exact dispatcher")
|
|
expected = original.replace(" _COMMAND_TRIGGER_EVENTS = (", " _COMMAND_POLL_INTERVAL = 0.1\n\n _COMMAND_TRIGGER_EVENTS = (", 1)
|
|
expected = expected.replace(" self._start_time: float | None = None", " self._start_time: float | None = None\n self._last_command_poll: float | None = None", 1)
|
|
expected = expected.replace(" except queue.Empty:\n time.sleep(0.1)", " except queue.Empty:\n self._process_commands()\n time.sleep(0.1)", 1)
|
|
expected = expected.replace(" if event is None or isinstance(event, self._COMMAND_TRIGGER_EVENTS):\n self._execution_coordinator.process_commands()",
|
|
" now = time.monotonic()\n if (\n event is None\n or isinstance(event, self._COMMAND_TRIGGER_EVENTS)\n or self._last_command_poll is None\n or now - self._last_command_poll >= self._COMMAND_POLL_INTERVAL\n ):\n self._execution_coordinator.process_commands()\n self._last_command_poll = time.monotonic()", 1)
|
|
check(expected == candidate, "only four declared text changes")
|
|
diff = "".join(difflib.unified_diff(original.splitlines(True), candidate.splitlines(True), fromfile="a/"+target, tofile="b/"+target))
|
|
check(diff == (OUT / "v43r26c_dispatcher_command_polling.patch").read_text(encoding="utf-8"), "patch exactly regenerates from fixed inputs")
|
|
manifest = json.loads((OUT / "v43r26c_dispatcher_command_polling_candidate.manifest.safe.json").read_text(encoding="utf-8"))
|
|
check(manifest["applied"] is False and manifest["automatic_apply_available"] is False, "candidate remains unapplied")
|
|
for path, sha in manifest["input_and_artifact_sha256"].items(): check(hashlib.sha256((OUT / path).read_bytes()).hexdigest() == sha, "manifest binding " + path)
|
|
|
|
|
|
class Log:
|
|
def __init__(self): self.warnings = []; self.exceptions = []
|
|
def warning(self, *args): self.warnings.append(str(args[0]))
|
|
def exception(self, *args): self.exceptions.append(str(args[0]))
|
|
def debug(self, *args): pass
|
|
|
|
|
|
class Flag:
|
|
def __init__(self): self.value = False
|
|
def set(self): self.value = True
|
|
def clear(self): self.value = False
|
|
def is_set(self): return self.value
|
|
|
|
|
|
class Clock:
|
|
def __init__(self): self.ns = 0; self.wall = 1000000; self.sleeps = []
|
|
def monotonic(self): return self.ns / 1e9
|
|
def time(self): return self.wall
|
|
def advance(self, seconds): self.ns += round(seconds * 1e9); self.wall -= 1000
|
|
def sleep(self, seconds): self.sleeps.append(seconds); self.advance(seconds)
|
|
|
|
|
|
class Event:
|
|
def __init__(self, name="chunk", dt=.02): self.name = name; self.dt = dt
|
|
class Succeeded(Event): pass
|
|
class Failed(Event): pass
|
|
class NodeException(Event): pass
|
|
class AbortCommand:
|
|
def __init__(self, reason="owned"): self.reason = reason
|
|
class PauseCommand:
|
|
def __init__(self, reason="owned"): self.reason = reason
|
|
|
|
|
|
log = Log()
|
|
Graph = load_class(source["core/workflow/graph_engine/domain/graph_execution.py"], "GraphExecution",
|
|
dict(dataclass=dataclass, field=field, NodeExecution=NS))
|
|
Processor = load_class(source["core/workflow/graph_engine/command_processing/command_processor.py"], "CommandProcessor", dict(logger=log))
|
|
henv = dict(CommandHandler=object, AbortCommand=AbortCommand, PauseCommand=PauseCommand, logger=log, SchedulingPause=lambda **kw: NS(**kw))
|
|
Abort = load_class(source["core/workflow/graph_engine/command_processing/command_handlers.py"], "AbortCommandHandler", henv)
|
|
Pause = load_class(source["core/workflow/graph_engine/command_processing/command_handlers.py"], "PauseCommandHandler", henv)
|
|
Coordinator = load_class(source["core/workflow/graph_engine/orchestration/execution_coordinator.py"], "ExecutionCoordinator", {})
|
|
|
|
|
|
class Channel:
|
|
def __init__(self, clock, due=(), *, cost=0, error=False):
|
|
self.clock = clock; self.due = list(due); self.cost = cost; self.error = error; self.calls = []; self.taken = []
|
|
def fetch_commands(self):
|
|
self.calls.append(self.clock.monotonic()); self.clock.advance(self.cost)
|
|
if self.error: raise RuntimeError("synthetic channel read failure")
|
|
ready = [(t, c) for t, c in self.due if t <= self.clock.monotonic() + 1e-12]
|
|
self.due = [(t,c) for t,c in self.due if (t,c) not in ready]; self.taken += [(t,self.clock.monotonic(),type(c).__name__) for t,c in ready]
|
|
return [c for _,c in ready]
|
|
|
|
|
|
def scenario(text, mode="chunk", due=.025, action="abort", *, initial="running", cost=0, error=False, fail_at=None):
|
|
clock = Clock(); g = Graph("synthetic-workflow"); g.start()
|
|
if initial == "paused": g.pause(NS(message="owned"))
|
|
if initial == "error": g.fail(RuntimeError("preexisting"))
|
|
if initial in {"complete", "domain_only_complete"}: g.complete()
|
|
if initial == "aborted": g.abort("preexisting")
|
|
cmd = AbortCommand() if action == "abort" else PauseCommand()
|
|
channel = Channel(clock, [] if due is None else [(due,cmd)], cost=cost, error=error)
|
|
foreign = Channel(clock, [(0,AbortCommand("foreign"))])
|
|
processor = Processor(channel, g); processor.register_handler(AbortCommand, Abort()); processor.register_handler(PauseCommand, Pause())
|
|
state = NS(done=initial == "complete"); state.is_execution_complete = lambda: state.done
|
|
trace = []; pool = NS(check_and_scale=lambda: trace.append("scale"), stop=lambda: trace.append("worker-stop"))
|
|
coordinator = Coordinator(g, state, processor, pool)
|
|
items = [Event(str(i)) for i in range(30)] + [Succeeded("terminal", .001)] if mode == "chunk" else []
|
|
class Events:
|
|
def __init__(self): self.active = 0; self.done = 0; self.drained = 0; self.empty = 0
|
|
def get(self, block=True, timeout=None):
|
|
if not block:
|
|
if not items: raise queue.Empty()
|
|
self.drained += 1; return items.pop(0)
|
|
self.active += 1
|
|
if self.active > 60: raise AssertionError("finite loop guard")
|
|
if fail_at == "queue": raise RuntimeError("queue failure")
|
|
if items:
|
|
e = items.pop(0); clock.advance(e.dt); return e
|
|
check(timeout == .1, "original empty timeout")
|
|
clock.advance(timeout); self.empty += 1
|
|
if self.empty == 20: dispatcher._stop_event.set()
|
|
raise queue.Empty()
|
|
def task_done(self): self.done += 1
|
|
events = Events(); names=[]
|
|
def dispatch(e):
|
|
if fail_at == "dispatch": raise RuntimeError("dispatch failure")
|
|
names.append(e.name)
|
|
if isinstance(e, Succeeded): state.done = True
|
|
if fail_at == "scaling": pool.check_and_scale=lambda: (_ for _ in ()).throw(RuntimeError("scaling failure"))
|
|
completed=[]
|
|
env=dict(time=clock,queue=queue,threading=NS(Event=Flag),logger=log,NodeRunSucceededEvent=Succeeded,NodeRunFailedEvent=Failed,NodeRunExceptionEvent=NodeException)
|
|
Dispatcher=load_class(text,"Dispatcher",env)
|
|
dispatcher=Dispatcher(events,NS(dispatch=dispatch),coordinator,NS(mark_complete=lambda:completed.append(True)))
|
|
if initial == "external_stop": dispatcher._stop_event.set()
|
|
dispatcher._dispatcher_loop()
|
|
return NS(clock=clock,graph=g,channel=channel,foreign=foreign,events=events,names=names,complete=completed,dispatcher=dispatcher,trace=trace)
|
|
|
|
|
|
# Enumerate arrival phase/type with the real processor and abort/pause methods.
|
|
cases = 0
|
|
for mode in ["chunk", "empty"]:
|
|
for due in [0, .000001, .025, .099999, .1, .100001, .199999, .2, .31, .499]:
|
|
for action in ["abort", "pause"]:
|
|
a=scenario(original,mode,due,action); b=scenario(candidate,mode,due,action); cases += 1
|
|
check(bool(b.channel.taken), "candidate consumes owned command")
|
|
consumed=b.channel.taken[0][1]
|
|
check(consumed + 1e-9 >= due and consumed-due <= (.2 if mode=="empty" else .12)+1e-9, "bounded logical cooperative opportunity")
|
|
check(b.foreign.calls==[] and len(b.foreign.due)==1, "foreign channel not fetched")
|
|
check(a.names==b.names and a.events.done==b.events.done, "original queued drain/order/task_done retained")
|
|
check(b.complete==[True] and "worker-stop" not in b.trace, "finally exactly once, no new worker cancel hook")
|
|
check(b.graph.aborted if action=="abort" else b.graph.is_paused, "actual domain transition reached")
|
|
check(b.graph.completed if action=="abort" else not b.graph.completed, "actual coordinator respects completed versus paused")
|
|
if due == .025: observations.append(dict(case=mode+"_"+action,original_consumed=a.channel.taken[0][1],candidate_consumed=consumed))
|
|
|
|
for initial in ["paused","error","complete","aborted","external_stop"]:
|
|
a=scenario(original,due=None,initial=initial); b=scenario(candidate,due=None,initial=initial)
|
|
check(a.channel.calls==b.channel.calls and len(b.channel.calls)==2, "start/exit forced polling retains initial-state semantics")
|
|
check(a.graph.completed==b.graph.completed and a.names==b.names and b.complete==[True], "initial/error/pause/complete/external-stop drain retained")
|
|
|
|
# Domain.complete and StateManager completion are separate original contracts.
|
|
# Retain this distinct negative instead of confusing it with the complete case.
|
|
a=scenario(original,due=None,initial="domain_only_complete"); b=scenario(candidate,due=None,initial="domain_only_complete")
|
|
check(a.events.active==b.events.active==31 and a.names==b.names,"domain-only completed does not override original StateManager predicate")
|
|
check(a.complete==b.complete==[True] and a.graph.completed and b.graph.completed,"domain-only completed retains no duplicate complete transition")
|
|
|
|
for failure in ["queue","dispatch","scaling"]:
|
|
a=scenario(original,due=None,fail_at=failure); b=scenario(candidate,due=None,fail_at=failure)
|
|
check(a.graph.has_error and b.graph.has_error and a.graph.completed and b.graph.completed, "real mark_failed boundary preserved")
|
|
check(a.complete==b.complete==[True] and a.events.done==b.events.done, "exception finally/task_done behavior retained")
|
|
|
|
# A real processor suppresses fetch failures. The timestamp measures return of
|
|
# an attempted poll, not success or acknowledgement; neither version mark_failed.
|
|
before=len(log.warnings); b=scenario(candidate,"chunk",due=None,error=True)
|
|
check(len(log.warnings)>before and not b.graph.has_error, "actual command read failure warns without propagating")
|
|
check(b.dispatcher._last_command_poll==b.clock.monotonic() and b.graph.completed, "suppressed read failure still advances attempt timestamp")
|
|
observations.append(dict(case="real_processor_suppressed_read_error",warning=True,dispatcher_mark_failed=False,poll_timestamp_updated=True))
|
|
|
|
# Timestamp is completion-based; slow I/O is not a fresh 100ms cancellation SLA.
|
|
b=scenario(candidate,"chunk",due=None,cost=.04)
|
|
ordinary=b.channel.calls[1:-2]
|
|
check(all(y-x >= .14-1e-9 for x,y in zip(ordinary,ordinary[1:])), "supplementary cadence includes original command I/O cost")
|
|
check(all(x==.1 for x in scenario(candidate,"empty",due=.31).clock.sleeps), "empty retains original 100ms sleep")
|
|
|
|
# Direct selector: wall clock moves backwards, monotonic edge and forced calls.
|
|
clock=Clock(); calls=[]
|
|
env=dict(time=clock,queue=queue,threading=NS(Event=Flag),logger=log,NodeRunSucceededEvent=Succeeded,NodeRunFailedEvent=Failed,NodeRunExceptionEvent=NodeException)
|
|
D=load_class(candidate,"Dispatcher",env); d=D(None,None,NS(process_commands=lambda:calls.append(clock.monotonic())))
|
|
d._process_commands(); clock.ns=99999999; clock.wall=-1e12; d._process_commands(Event())
|
|
check(calls==[0.0], "99.999999ms and wall reversal do not poll")
|
|
clock.ns=100000000; d._process_commands(Event()); check(calls==[0.0,.1], "100ms monotonic boundary eligible")
|
|
for e in [Succeeded(),Failed(),NodeException(),None]: d._process_commands(e)
|
|
check(len(calls)==6, "all original forced selectors bypass throttle")
|
|
last=d._last_command_poll
|
|
d._execution_coordinator=NS(process_commands=lambda:(_ for _ in ()).throw(RuntimeError("outward coordinator failure")))
|
|
try: d._process_commands()
|
|
except RuntimeError: pass
|
|
else: check(False,"outward coordinator failure must propagate")
|
|
check(d._last_command_poll==last,"outward exception does not stamp successful return")
|
|
|
|
# LLM note assertions use only exact saved ASTs and preserve old report scopes.
|
|
def fn(path,name): return next(n for n in ast.walk(ast.parse(source[path])) if isinstance(n,ast.FunctionDef) and n.name==name)
|
|
def iterates(n, variable): return any(isinstance(x,ast.For) and isinstance(x.iter,ast.Name) and x.iter.id==variable for x in ast.walk(n))
|
|
def closes(n): return any(isinstance(x,ast.Call) and isinstance(x.func,ast.Attribute) and x.func.attr in {"close","aclose"} for x in ast.walk(n))
|
|
llm="core/workflow/nodes/llm/node.py"
|
|
check(iterates(fn(llm,"_run"),"generator") and not closes(fn(llm,"_run")),"LLM _run ordinary for, no explicit child close")
|
|
check(iterates(fn(llm,"handle_invoke_result"),"invoke_result") and not closes(fn(llm,"handle_invoke_result")),"LLM result ordinary for, no explicit child close")
|
|
invoke=fn(llm,"invoke_llm"); calls_llm=[n for n in ast.walk(invoke) if isinstance(n,ast.Call) and isinstance(n.func,ast.Attribute) and n.func.attr=="invoke_llm"]
|
|
check(len(calls_llm)==1 and {k.arg for k in calls_llm[0].keywords}=={"prompt_messages","model_parameters","stop","stream","user"},"explicit model call has stop sequence, no task cancellation argument")
|
|
check(iterates(fn("core/workflow/nodes/base/node.py","run"),"result") and not closes(fn("core/workflow/nodes/base/node.py","run")),"base Node ordinary result iteration")
|
|
check(iterates(fn("core/model_runtime/model_providers/__base/large_language_model.py","_invoke_result_generator"),"result") and not closes(fn("core/model_runtime/model_providers/__base/large_language_model.py","_invoke_result_generator")),"base model ordinary result iteration")
|
|
check(any(isinstance(n,ast.YieldFrom) and isinstance(n.value,ast.Name) and n.value.id=="response" for n in ast.walk(ast.parse(source["core/plugin/impl/model.py"]))),"plugin model delegates yield from response")
|
|
stream=fn("core/plugin/impl/base.py","_stream_request")
|
|
check(any(isinstance(n,ast.With) and any(isinstance(i.context_expr,ast.Call) and isinstance(i.context_expr.func,ast.Attribute) and i.context_expr.func.attr=="stream" for i in n.items) for n in ast.walk(stream)),"innermost plugin stream owns context manager")
|
|
preservation=json.loads((OUT/"v43r26c_deployed_dify_static43_preservation.safe.json").read_text(encoding="utf-8"))
|
|
check(preservation["source_after"]=={"checked":43,"changed":[]} and preservation["source_snapshot_stable"] is True,"saved later deployment rehash covers 43 unchanged paths")
|
|
for name,sha in pins.items():check(hashlib.sha256((OUT/name).read_bytes()).hexdigest()==sha,"review input unchanged after execution")
|
|
print(json.dumps(dict(scope="independent offline dispatcher-only candidate review",assertions=checks,failures=failures,
|
|
enumerated_command_scenarios=cases,embedded_sources_verified=len(source),observations=observations,pins=pins,
|
|
actual_domain_processor_handler_coordinator_methods=True,synthetic_clock_queue_channel_pool_state=True,
|
|
applied=False,actual_threads_started=0,network_auth_get_stop_model_calls=0,physical_gpu_release_proven=False,
|
|
historical_six_run_causes_proven=False),ensure_ascii=False))
|
|
raise SystemExit(1 if failures else 0)
|