Checkpoint、人工中断与重复执行：练习答案

参考实现

追加正文：

other = {"configurable":{"thread_id":"tenant-a:task-8"}}
g.invoke({"approved":False},other)
rejected = g.invoke(Command(resume=False),other)
assert rejected["approved"] is False and len(visits) == 4
assert g.get_state(config).values["approved"] is True
assert g.get_state(other).values["approved"] is False
print("independent approval states")

隔离检查点只证明状态隔离，访问控制仍由应用负责。

返回正文

两进程拒绝与错误线程答案

将正文持久图块保存为 persist.py。下列文件另存 persist_answer.py；它启动新进程，在错误线程先检查无检查点并拒绝恢复，再在原线程执行拒绝。thread_id 只定位状态，不替代业务鉴权。

import sys,subprocess
from pathlib import Path
from tempfile import TemporaryDirectory
from langgraph.graph import StateGraph,START,END
from langgraph.types import Command
from langgraph.checkpoint.sqlite import SqliteSaver
from persist import State,approve

def resume(db):
    b=StateGraph(State);b.add_node('approve',approve);b.add_edge(START,'approve');b.add_edge('approve',END)
    with SqliteSaver.from_conn_string(db) as saver:
        g=b.compile(checkpointer=saver)
        good={'configurable':{'thread_id':'review-1'}}
        wrong={'configurable':{'thread_id':'other'}}
        assert not g.get_state(wrong).values
        assert g.get_state(good).next==('approve',)
        # 服务入口没有待恢复 checkpoint 就拒绝，不把错误 thread 当成批准。
        denied=g.invoke(Command(resume=False),good,durability='sync')
        assert denied['approved'] is False and g.get_state(good).next==()
        assert not g.get_state(wrong).values
if __name__=='__main__':
    if len(sys.argv)==2:resume(sys.argv[1])
    else:
        import persist
        with TemporaryDirectory() as folder:
            db=str(Path(folder)/'state.sqlite')
            subprocess.run([sys.executable,persist.__file__,'start',db],check=True)
            subprocess.run([sys.executable,__file__,db],check=True)
        print('new process rejects wrong thread and persists refusal on original thread')