用户把周六改成周日，Agent 哪些工作必须重做？

"""Offline read-only workflow; no LLM, network, purchase or real clock."""
from copy import deepcopy
from dataclasses import dataclass


class SourceExpired(Exception):
    pass


class NotReady(Exception):
    pass


@dataclass(frozen=True)
class Value:
    revision: int
    data: object
    expires_at: int | None = None
    dependencies: tuple = ()


@dataclass(frozen=True)
class Ticket:
    node: str
    attempt: int
    dependencies: tuple
    values: dict
    expires_at: int | None


class Planner:
    # Every input read by compute() is declared here. Logic is fixed in this run.
    dependencies = {
        "events": ("day", "catalog"),
        "route": ("events",),
        "select": ("events", "budget"),
        "draft": ("route", "select", "style"),
    }

    def __init__(self):
        self.now = 0
        self.inputs = {
            "day": Value(1, "sat"), "budget": Value(1, 50),
            "style": Value(1, "brief"),
            "catalog": Value(1, [
                {"id": "park-sat", "day": "sat", "cost": 20},
                {"id": "museum-sat", "day": "sat", "cost": 45},
                {"id": "museum-sun", "day": "sun", "cost": 30},
            ], expires_at=10),
        }
        self.artifacts, self.pending, self.revisions = {}, {}, {}
        self.next_attempt = 0
        self.executed = []

    def view(self, name):
        value = self.inputs.get(name) if name in self.inputs else self.artifacts.get(name)
        if value is None:
            raise NotReady(name)
        if value.expires_at is not None and self.now >= value.expires_at:
            raise SourceExpired(name)
        if name in self.dependencies:
            current = tuple((dep, self.view(dep).revision) for dep in self.dependencies[name])
            if current != value.dependencies:
                raise NotReady(name)
        return value

    def affected(self, changed):
        found = set(changed)
        while True:
            previous = len(found)
            for node, deps in self.dependencies.items():
                if any(dep in found for dep in deps):
                    found.add(node)
            if len(found) == previous:
                return [node for node in self.dependencies if node in found]

    def invalidate(self, changed):
        nodes = self.affected(changed)
        for node in nodes:
            self.artifacts.pop(node, None)
        # Pending work can still finish. Accept/reject is checked at finish().
        return nodes

    def update(self, name, data, expires_at=None):
        if name not in self.inputs:
            raise KeyError(name)
        old = self.inputs[name]
        if old.data == data and old.expires_at == expires_at:
            return []
        self.inputs[name] = Value(old.revision + 1, deepcopy(data), expires_at)
        return self.invalidate([name])

    def begin(self, node):
        values = {dep: self.view(dep) for dep in self.dependencies[node]}
        signature = tuple((dep, item.revision) for dep, item in values.items())
        deadlines = [item.expires_at for item in values.values() if item.expires_at is not None]
        self.next_attempt += 1
        ticket = Ticket(node, self.next_attempt, signature,
                        {dep: deepcopy(item.data) for dep, item in values.items()},
                        min(deadlines) if deadlines else None)
        self.pending[node] = ticket.attempt
        self.executed.append(node)
        return ticket

    def finish(self, ticket, data):
        if self.pending.get(ticket.node) != ticket.attempt:
            return "superseded_attempt"
        try:
            signature = tuple((dep, self.view(dep).revision)
                              for dep in self.dependencies[ticket.node])
        except SourceExpired:
            return "expired_dependencies"
        except NotReady:
            return "stale_dependencies"
        if signature != ticket.dependencies:
            return "stale_dependencies"
        if ticket.expires_at is not None and self.now >= ticket.expires_at:
            return "expired_dependencies"
        revision = self.revisions.get(ticket.node, 0) + 1
        self.revisions[ticket.node] = revision
        self.invalidate([ticket.node])
        self.artifacts[ticket.node] = Value(revision, deepcopy(data),
                                           ticket.expires_at, ticket.dependencies)
        del self.pending[ticket.node]
        return "committed"

    @staticmethod
    def compute(ticket):
        value = ticket.values
        if ticket.node == "events":
            return [item for item in value["catalog"] if item["day"] == value["day"]]
        if ticket.node == "route":
            return {item["id"]: "walk" for item in value["events"]}
        if ticket.node == "select":
            return [item for item in value["events"] if item["cost"] <= value["budget"]]
        if ticket.node == "draft":
            return {"style": value["style"], "items": [
                {"id": item["id"], "cost": item["cost"], "route": value["route"][item["id"]]}
                for item in value["select"]]}
        raise KeyError(ticket.node)

    def ensure(self, name):
        try:
            return self.view(name)
        except (NotReady, SourceExpired):
            if name in self.inputs:
                raise  # A tool/source refresh is required; never invent freshness.
        for dependency in self.dependencies[name]:
            self.ensure(dependency)
        ticket = self.begin(name)
        result = self.finish(ticket, self.compute(ticket))
        if result != "committed":
            raise NotReady(result)
        return self.view(name)


def main():
    planner = Planner()
    first = deepcopy(planner.ensure("draft").data)
    assert len(first["items"]) == 2
    print("initial computed:", ",".join(planner.executed))
    planner.executed.clear()
    changed = planner.update("budget", 35)
    assert changed == ["select", "draft"]
    planner.ensure("draft")
    assert planner.executed == ["select", "draft"]
    print("budget recomputed:", ",".join(planner.executed))
    planner.executed.clear()
    planner.update("style", "detailed")
    planner.ensure("draft")
    assert planner.executed == ["draft"]
    print("style recomputed:", ",".join(planner.executed))
    assert planner.update("style", "detailed") == []

    old_ticket = planner.begin("events")
    planner.update("day", "sun")
    rejected = planner.finish(old_ticket, planner.compute(old_ticket))
    assert rejected == "stale_dependencies"
    print("late old-day result:", rejected)
    current = planner.ensure("draft").data
    assert [row["id"] for row in current["items"]] == ["museum-sun"]
    assert [row["id"] for row in first["items"]] == ["park-sat", "museum-sat"]

    still_valid = planner.begin("events")
    planner.update("budget", 40)  # This input is not an events dependency.
    accepted = planner.finish(still_valid, planner.compute(still_valid))
    assert accepted == "committed"
    print("late result after unrelated change:", accepted)
    older, newer = planner.begin("events"), planner.begin("events")
    assert planner.finish(older, planner.compute(older)) == "superseded_attempt"
    assert planner.finish(newer, planner.compute(newer)) == "committed"
    assert planner.finish(newer, planner.compute(newer)) == "superseded_attempt"

    expires_while_running = planner.begin("events")
    planner.now = 10
    assert planner.finish(expires_while_running, planner.compute(expires_while_running)) == "expired_dependencies"
    try:
        planner.ensure("draft")
    except SourceExpired:
        print("at t=10: source refresh required")
    else:
        raise AssertionError("expired source must not be silently refreshed")

    planner.executed.clear()
    # A scripted successful source recheck; same facts, a new accepted expiry.
    planner.update("catalog", planner.inputs["catalog"].data, expires_at=20)
    planner.ensure("draft")
    assert planner.executed == ["events", "route", "select", "draft"]
    print("after source refresh:", ",".join(planner.executed))
    print("PASS: dependency snapshots, selective rebuild, late results, source expiry")


if __name__ == "__main__":
    main()