LangGraph 1.2.15 ย้อน Checkpoint อย่างไรไม่ให้ State คนละ Branch ปนกัน

"""Synthetic checkpoint/fork lesson. No model, network or external writes.

Editorial test status: NOT RUN. Assertions are acceptance checks for the reader
to execute; their presence is not evidence that this program has been tested.
"""
from __future__ import annotations

from copy import deepcopy
import json
from typing import Any, TypedDict
from uuid import uuid4

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph


FIXTURES = {
    "original": {"path": "A", "text": "เอกสารขั้นตอนเดิม"},
    "alternate": {"path": "B", "text": "เอกสารขั้นตอนทางเลือก"},
}


class LessonState(TypedDict, total=False):
    scenario: str
    fixture: dict[str, str]
    report: str


def read_fixture(state: LessonState) -> dict[str, Any]:
    """Mock tool node: select local synthetic data; no API call."""
    scenario = state.get("scenario")
    if scenario not in FIXTURES:
        raise ValueError("Unknown synthetic scenario.")
    return {"fixture": deepcopy(FIXTURES[scenario])}


def format_report(state: LessonState) -> dict[str, str]:
    """Pure formatter. Reject incomplete mock tool data."""
    fixture = state.get("fixture")
    if not isinstance(fixture, dict) or set(fixture) != {"path", "text"}:
        raise ValueError("Fixture must contain only path and text.")
    if any(not isinstance(value, str) or not value.strip()
           for value in fixture.values()):
        raise ValueError("Fixture values must be nonempty strings.")
    return {"report": f"เส้นทาง {fixture['path']} | {fixture['text']}"}


def build_graph() -> Any:
    builder = StateGraph(LessonState)
    builder.add_node("read_fixture", read_fixture)
    builder.add_node("format_report", format_report)
    builder.add_edge(START, "read_fixture")
    builder.add_edge("read_fixture", "format_report")
    builder.add_edge("format_report", END)
    return builder.compile(checkpointer=InMemorySaver())


def select_before_report(
    graph: Any,
    thread_config: dict[str, Any],
    checkpoint_id: str | None = None,
) -> Any:
    """Select by execution boundary, never by a history-list index.

    This lesson accepts exactly one matching checkpoint. With multiple turns or
    branches, the caller must specify an ID from the displayed history.
    """
    candidates = [
        snapshot
        for snapshot in graph.get_state_history(thread_config)
        if snapshot.next == ("format_report",)
        and (
            checkpoint_id is None
            or snapshot.config["configurable"]["checkpoint_id"] == checkpoint_id
        )
    ]
    if len(candidates) != 1:
        raise ValueError(
            "Expected one checkpoint before format_report; "
            "inspect history and supply its exact checkpoint_id."
        )
    return candidates[0]


def checkpoint_lineage(
    graph: Any, snapshot: Any, ancestor_config: dict[str, Any], limit: int = 12
) -> list[str]:
    """Read a bounded parent chain; stop if the expected ancestor is absent."""
    ancestor_id = ancestor_config["configurable"]["checkpoint_id"]
    ids: list[str] = []
    current = snapshot
    for _ in range(limit):
        current_id = current.config["configurable"]["checkpoint_id"]
        if current_id in ids:
            raise RuntimeError("Unexpected cycle in checkpoint lineage.")
        ids.append(current_id)
        if current_id == ancestor_id:
            return ids
        if current.parent_config is None:
            break
        current = graph.get_state(current.parent_config)
    raise RuntimeError("Fork start not found in the bounded parent chain.")


def run_demo() -> dict[str, Any]:
    graph = build_graph()
    thread_config = {"configurable": {"thread_id": str(uuid4())}}
    graph.invoke({"scenario": "original"}, thread_config)

    original_final = graph.get_state(thread_config)
    if original_final.next:
        raise RuntimeError("Original branch has not finished.")
    original_config = deepcopy(original_final.config)
    original_values = deepcopy(original_final.values)

    selected = select_before_report(graph, thread_config)
    fork_start_config = graph.update_state(
        selected.config,
        {
            "scenario": "alternate",
            "fixture": deepcopy(FIXTURES["alternate"]),
        },
        as_node="read_fixture",
    )
    if graph.get_state(fork_start_config).next != ("format_report",):
        raise RuntimeError("Fork is not at the expected boundary.")

    fork_output = graph.invoke(None, fork_start_config)

    # ONE WORKER demo: no other invocation can advance this thread between
    # invoke and this capture. A concurrent production thread needs its own
    # run ownership/correlation; its latest head is not necessarily our fork.
    fork_final = graph.get_state(thread_config)
    if fork_final.next:
        raise RuntimeError("Captured fork is not terminal.")
    fork_final_config = deepcopy(fork_final.config)
    lineage = checkpoint_lineage(graph, fork_final, fork_start_config)
    fork_readback = deepcopy(graph.get_state(fork_final_config).values)
    assert fork_readback == fork_output

    original_readback = deepcopy(graph.get_state(original_config).values)

    assert original_readback == original_values
    assert original_readback["report"] == (
        "เส้นทาง A | เอกสารขั้นตอนเดิม"
    )
    assert fork_output["report"] == (
        "เส้นทาง B | เอกสารขั้นตอนทางเลือก"
    )
    assert original_readback["fixture"]["path"] == "A"

    return {
        "mode": "synthetic_no_model_no_network",
        "thread_id": thread_config["configurable"]["thread_id"],
        "selected_checkpoint_id": selected.config["configurable"]["checkpoint_id"],
        "original_final_checkpoint_id": (
            original_config["configurable"]["checkpoint_id"]
        ),
        "fork_start_checkpoint_id": (
            fork_start_config["configurable"]["checkpoint_id"]
        ),
        "fork_final_checkpoint_id": (
            fork_final_config["configurable"]["checkpoint_id"]
        ),
        "fork_final_parent_checkpoint_id": (
            fork_final.parent_config["configurable"]["checkpoint_id"]
            if fork_final.parent_config else None
        ),
        "fork_lineage_checkpoint_ids": lineage,
        "original_report": original_readback["report"],
        "fork_report": fork_readback["report"],
        "original_unchanged": original_readback == original_values,
    }


if __name__ == "__main__":
    print(json.dumps(run_demo(), ensure_ascii=False, indent=2))

"""Synthetic acceptance tests. Editorial status: NOT RUN."""
from copy import deepcopy
import unittest

from checkpoint_fork import (
    FIXTURES,
    build_graph,
    run_demo,
    select_before_report,
)


class CheckpointForkTests(unittest.TestCase):
    def completed_graph(self):
        graph = build_graph()
        config = {"configurable": {"thread_id": self.id()}}
        graph.invoke({"scenario": "original"}, config)
        return graph, config

    def test_fork_returns_alternative_and_original_is_unchanged(self):
        result = run_demo()
        self.assertTrue(result["original_unchanged"])
        self.assertEqual(result["original_report"], "เส้นทาง A | เอกสารขั้นตอนเดิม")
        self.assertEqual(result["fork_report"], "เส้นทาง B | เอกสารขั้นตอนทางเลือก")
        self.assertNotEqual(
            result["original_final_checkpoint_id"],
            result["fork_start_checkpoint_id"],
        )
        self.assertNotEqual(
            result["fork_start_checkpoint_id"],
            result["fork_final_checkpoint_id"],
        )
        self.assertEqual(
            result["fork_lineage_checkpoint_ids"][0],
            result["fork_final_checkpoint_id"],
        )
        self.assertIn(
            result["fork_start_checkpoint_id"],
            result["fork_lineage_checkpoint_ids"],
        )

    def test_selection_requires_boundary_not_history_position(self):
        graph, config = self.completed_graph()
        selected = select_before_report(graph, config)
        self.assertEqual(selected.next, ("format_report",))
        selected_id = selected.config["configurable"]["checkpoint_id"]
        addressed = select_before_report(graph, config, selected_id)
        self.assertEqual(addressed.config, selected.config)

    def test_terminal_checkpoint_is_not_a_before_report_checkpoint(self):
        graph, config = self.completed_graph()
        terminal = graph.get_state(config)
        self.assertEqual(terminal.next, ())
        terminal_id = terminal.config["configurable"]["checkpoint_id"]
        with self.assertRaises(ValueError):
            select_before_report(graph, config, terminal_id)

    def test_unknown_checkpoint_id_stops_instead_of_guessing(self):
        graph, config = self.completed_graph()
        with self.assertRaises(ValueError):
            select_before_report(graph, config, "missing-checkpoint")

    def test_ambiguous_history_requires_explicit_checkpoint_id(self):
        graph, config = self.completed_graph()
        first = select_before_report(graph, config)
        first_id = first.config["configurable"]["checkpoint_id"]
        graph.invoke({"scenario": "alternate"}, config)
        with self.assertRaises(ValueError):
            select_before_report(graph, config)
        self.assertEqual(
            select_before_report(graph, config, first_id).config,
            first.config,
        )

    def test_failed_fork_leaves_addressed_original_values_unchanged(self):
        graph, config = self.completed_graph()
        original = graph.get_state(config)
        original_config = deepcopy(original.config)
        original_values = deepcopy(original.values)
        selected = select_before_report(graph, config)
        bad_fork = graph.update_state(
            selected.config,
            {"fixture": {"path": "B"}},  # Missing text: synthetic failure.
            as_node="read_fixture",
        )
        with self.assertRaises(ValueError):
            graph.invoke(None, bad_fork)
        self.assertEqual(graph.get_state(original_config).values, original_values)
        self.assertEqual(FIXTURES["original"]["path"], "A")


if __name__ == "__main__":
    unittest.main()

# Checkpoint fork lesson — synthetic example

Editorial status: **NOT RUN**. The author has not executed, imported, compiled,
installed or tested these files. Expected output and tests describe acceptance
criteria for a reader's own authorized local run.

## Scope

Python 3.12+; LangGraph pinned to1.2.15 (package supports Python>=3.10).
No LLM, API key, environment variables, network requests, files/database writes,
publishing tools or paid calls exist in the example. The mock tool reads two
local fixtures. InMemorySaver stores checkpoints in RAM and loses them at
process exit. This is not a production restart-recovery or live-integration demo.

## Files

- checkpoint_fork.py: independently authored smallStateGraph and fork manifest.
- test_checkpoint_fork.py: six synthetic acceptance/failure tests.
- ../requirements.txt: langgraph==1.2.15; transitive requirements come from its package metadata.
- No pytest dependency: tests use Python standard-library unittest.

## Installation (reader instructions; NOT performed by the writer)

Run these commands in this run directory in a local environment you control.
Package installation accesses the package registry and changes your chosen
virtual environment; it is not part of the automatic editorial run.

Windows PowerShell:

```powershell
py -3.12 -m venv .venv
.venv\Scripts\python.exe -m pip install -r requirements.txt
.venv\Scripts\python.exe code\checkpoint_fork.py
.venv\Scripts\python.exe -m unittest discover -s code -p "test_checkpoint_fork.py" -v
```

Linux/macOS withPython3.12:

```sh
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt
.venv/bin/python code/checkpoint_fork.py
.venv/bin/python -m unittest discover -s code -p 'test_checkpoint_fork.py' -v
```

No environment variable configuration is required. There is no model/provider
billing from this example; local resources and package download still exist.

## Input and expected output

Input inside run_demo: scenario="original". The controlled fork replaces the
recorded mockfixture with scenario="alternate" afterread_fixture and before
format_report. You may inspect or alter the fixtures locally.

Expected (NOT observed by the writer), omitting runtime-generated IDs:

```json
{
  "mode": "synthetic_no_model_no_network",
  "original_report": "เส้นทาง A | เอกสารขั้นตอนเดิม",
  "fork_report": "เส้นทาง B | เอกสารขั้นตอนทางเลือก",
  "original_unchanged": true
}
```

The executable emits thread, selected, original-final, fork-start, fork-final
and parent/lineage IDs too. Values are generated at a reader's runtime and must
not be copied as claimed logs. fork-start addresses the new checkpoint before
the formatter; fork-final addresses the captured terminal checkpoint.

ONE WORKER assumption: after invoke finishes, this demonstration alone can
advance its thread, so it captures the latest snapshot and stores its full config.
It verifies terminal next, expected parent ancestry through a bounded read chain,
and rereads the fork by that captured config. In a concurrent production thread,
latest cannot be assumed to belong to our fork: use explicit run ownership and
correlation before accepting it.

Acceptance: address the saved original-final config after the fork, compare
deepcopied values, show the alternative output, and record IDs. Selecting a
terminal checkpoint or an unknown/ambiguous checkpoint must stop rather than
guessing an index.

## Design and limits

- StateGraph -> read_fixture mocknode -> format_report pure node -> END.
- select_before_report selects snapshot.next==("format_report",).
- Multiple matches require a checkpoint_id chosen from inspected history.
- update_state usesas_node="read_fixture" because the patch replaces that
  node's already-recorded synthetic output; it does not call the mock tool again.
- invoke(None,fork_start_config) runs the downstream formatter.
- Original verification uses original_final.config, never the latest thread head.
- Fork final is captured only under ONE WORKER, then reread by fork_final.config;
  terminal next, fork-start lineage, and fork-output equality are checked.
- Standardstate channels are used; this is not a repro ofDeltaChannel's
  exit-durability regression and does not prove that bug fixed locally.
- Replay/fork can execute downstream node operations again in a real graph.
  Checkpoints do not undo external sends or publications. Do not add such a
  tool without a separate authorization/idempotency/reconciliation design.
- It intentionally tests one sequential graph without concurrent workers,
  subgraphs, durable storage, migrations, process crashes or model decisions.
- Python assert can be disabled withoptimization(-O); do not use -O for this
  acceptance demo, and use explicit validation exceptions in production logic.

## Sources

Official APIs and package contract:

- https://docs.langchain.com/oss/python/langgraph/use-time-travel
- https://docs.langchain.com/oss/python/langgraph/checkpointers
- https://github.com/langchain-ai/langgraph/blob/1.2.15/libs/langgraph/pyproject.toml
- https://github.com/langchain-ai/langgraph/blob/1.2.15/libs/langgraph/langgraph/graph/__init__.py

Original independent demonstration code; uses documented public API patterns.
Not a copied upstream test. No production-ready claim.

langgraph==1.2.15