feat: add opt-in conversation checkpoint and resume

.github/workflows/conversation_state.yml

@@ -0,0 +1,28 @@
+name: Conversation State
+
+on:
+  pull_request:
+    paths:
+      - "packages/core_systems/ten_conversation_state/**"
+      - ".github/workflows/conversation_state.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+        python: ["3.10", "3.12"]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python }}
+      - name: Install package
+        run: python -m pip install ./packages/core_systems/ten_conversation_state
+      - name: Test checkpoint lifecycle and process recovery
+        run: python -m unittest discover -s packages/core_systems/ten_conversation_state/tests -v

packages/core_systems/ten_conversation_state/LICENSE

@@ -0,0 +1,13 @@
+Copyright © 2025 Agora
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

packages/core_systems/ten_conversation_state/README.md

@@ -0,0 +1,117 @@
+# Conversation checkpoint and resume
+
+An opt-in Python proof of concept for
+[issue #2340](https://github.com/TEN-framework/ten-framework/issues/2340).
+Applications explicitly capture JSON state and restore it into a new session.
+The package has no runtime or vendor dependencies and does not change existing
+extension lifecycle behavior. Its API is experimental pending maintainer review.
+
+## Install and run
+
+From the repository root, using Python 3.10 or newer:
+
+```sh
+python -m pip install ./packages/core_systems/ten_conversation_state
+python -m unittest discover -s packages/core_systems/ten_conversation_state/tests -v
+```
+
+Alternatively, set `PYTHONPATH=packages/core_systems/ten_conversation_state`
+to run the tests and example directly from source.
+
+The example uses an application-owned SQLite backend. Run these as **separate
+processes**, copying the checkpoint ID printed by the first command:
+
+```sh
+python packages/core_systems/ten_conversation_state/examples/resume_workflow.py checkpoint /tmp/conversation.db
+python packages/core_systems/ten_conversation_state/examples/resume_workflow.py resume /tmp/conversation.db --checkpoint-id <printed-id>
+```
+
+On Windows, use a writable path such as `conversation.db`. The first process
+simulates gathering information and completing a tool lookup, saves a checkpoint,
+then exits. The second process creates a fresh workflow and resumes at
+`confirm_itinerary`, retaining the completed lookup reference without replaying it.
+This demonstrates logical state recovery after a process interruption; it does not
+open or reconnect a WebRTC, WebSocket, or SIP transport.
+
+## Participants and TEN integration
+
+Register an explicit mapping of stable component names to objects implementing:
+
+- `state_version: str`: application-defined schema version.
+- `async snapshot_state()`: return allowlisted JSON state.
+- `async validate_state(state)`: validate without mutating live state.
+- `async restore_state(state)`: apply state without replaying tool side effects.
+
+An application-owned `AsyncExtension` can implement these hooks and construct a
+`CheckpointManager(store, {"workflow": self})` during `on_init`. Call
+`manager.checkpoint(conversation_id, ttl_seconds=3600)` at an explicit workflow
+boundary. In a fresh extension, await
+`manager.restore(conversation_id, checkpoint_id, policy=resume_policy)` after
+configuration and before accepting new input or sending a greeting. Handle the
+returned decision before processing audio or user requests. Existing third-party
+extensions do not participate automatically; use adapters for their state APIs.
+
+The application owns orchestration. Pause input and wait for in-flight tool calls
+to complete before snapshotting or restoring. A manager serializes its own calls
+on one event loop, but cannot pause extensions or provide a distributed snapshot.
+Do not pass raw extension instances across runtime threads: an adapter must route
+snapshot/validate/restore commands onto the owning extension's loop using TEN
+messages. This initial package does not define graph-wide commands or automatic
+disconnect/reconnect hooks.
+
+## Resume policy
+
+Every restore requires an async policy receiving a detached checkpoint copy:
+
+```python
+async def resume_policy(checkpoint):
+    if not await application.authorized_for(checkpoint.conversation_id):
+        return ResumeDecision.REVALIDATE
+    return ResumeDecision.RESTORE
+```
+
+Import `ResumeDecision` from `ten_conversation_state`. `application` above is the
+application's own authorization service, bound to the current caller.
+
+`RESTORE` permits validation and application of state. `REVALIDATE` and `EXPIRED`
+leave all participants untouched. After revalidation succeeds, the application
+can explicitly retry restore with a policy that checks the fresh authorization.
+Policies may reject an entire checkpoint; selective field migrations or redaction
+must happen explicitly in application-owned state before checkpointing.
+
+Expiration is enforced before policy execution and checked again after policy and
+validation complete. An expired checkpoint cannot be approved by a policy.
+Conversation ID and checkpoint ID must match the request. IDs are lookup keys,
+not authorization credentials. Authenticate and authorize the current caller
+independently; never restore authentication or privileges from saved state.
+
+## Format, storage, and failure behavior
+
+The version 1 envelope contains `conversation_id`, a UUID `checkpoint_id`, UTC Unix
+`created_at` and optional `expires_at`, `schema_version`, and `components`. Each
+component has a `version` and `state`. Only JSON objects with string keys, arrays,
+strings, booleans, null, integers, and finite floats are accepted. Tuples, bytes,
+cycles, sockets, provider clients, and arbitrary objects are rejected. JSON
+validation cannot detect secrets: participants must explicitly exclude them.
+
+Implement `CheckpointStore.save`, `load`, and `delete` to use a database, Redis,
+object storage, or application persistence. Store a complete validated envelope
+atomically; `Checkpoint.to_json()` and `Checkpoint.from_json()` provide the wire
+format. `load` returns `None` for a missing key. Backend errors propagate to the
+caller. The memory store isolates objects but does not survive a process restart.
+Storage access control, encryption, retention, size limits, expired-checkpoint
+cleanup, and concurrent-writer ownership belong to the application/backend.
+
+Unknown envelope versions, component versions, or participant sets are rejected
+before restore. Migrations are application-owned. All participants validate
+before any restore hook runs. A failing restore hook raises `RestoreError` with
+the failed `component` and the names already `restored`; there is no rollback of
+arbitrary extension state. Discard that new session and construct fresh
+participants before retrying. Cancellation propagates and may also leave a partial
+restore; discard the session in that case. Do not expose a session until restore
+has completed successfully.
+
+Checkpointing does not guarantee exactly-once tool execution. Preserve operation
+IDs and use application-level idempotency for external side effects. Automatic
+scheduling, cross-agent migration, distributed failover, provider conversation
+recreation, and automatic transport reconnection are outside this initial scope.

packages/core_systems/ten_conversation_state/examples/resume_workflow.py

@@ -0,0 +1,141 @@
+#
+# This file is part of TEN Framework, an open source project.
+# Licensed under the Apache License, Version 2.0.
+# See the LICENSE file for more information.
+#
+"""Two-process recovery demonstration with an application-owned SQLite store."""
+
+import argparse
+import asyncio
+import sqlite3
+from pathlib import Path
+
+from ten_conversation_state import (
+    Checkpoint,
+    CheckpointManager,
+    ResumeDecision,
+)
+
+
+class SQLiteStore:
+    """Example backend; use an application-controlled path, not a client path."""
+
+    def __init__(self, path: Path):
+        self.path = path
+        with sqlite3.connect(path) as connection:
+            connection.execute(
+                "CREATE TABLE IF NOT EXISTS checkpoints ("
+                "conversation TEXT, checkpoint TEXT, payload TEXT, "
+                "PRIMARY KEY (conversation, checkpoint))"
+            )
+
+    async def save(self, checkpoint):
+        payload = checkpoint.to_json()
+
+        def write():
+            with sqlite3.connect(self.path) as connection:
+                connection.execute(
+                    "INSERT INTO checkpoints VALUES (?, ?, ?)",
+                    (
+                        checkpoint.conversation_id,
+                        checkpoint.checkpoint_id,
+                        payload,
+                    ),
+                )
+
+        await asyncio.to_thread(write)
+
+    async def load(self, conversation_id, checkpoint_id):
+        def read():
+            with sqlite3.connect(self.path) as connection:
+                return connection.execute(
+                    "SELECT payload FROM checkpoints "
+                    "WHERE conversation = ? AND checkpoint = ?",
+                    (conversation_id, checkpoint_id),
+                ).fetchone()
+
+        row = await asyncio.to_thread(read)
+        return Checkpoint.from_json(row[0]) if row else None
+
+    async def delete(self, conversation_id, checkpoint_id):
+        def remove():
+            with sqlite3.connect(self.path) as connection:
+                connection.execute(
+                    "DELETE FROM checkpoints "
+                    "WHERE conversation = ? AND checkpoint = ?",
+                    (conversation_id, checkpoint_id),
+                )
+
+        await asyncio.to_thread(remove)
+
+
+class Workflow:
+    """The same hooks can live on an application-owned TEN extension."""
+
+    state_version = "1"
+
+    def __init__(self):
+        self.state = {
+            "topic": "travel",
+            "step": "ask_destination",
+            "completed_operations": [],
+        }
+
+    async def snapshot_state(self):
+        # Explicit allowlist: never include provider clients or authentication.
+        return {
+            key: self.state[key]
+            for key in ("topic", "step", "completed_operations")
+        }
+
+    async def validate_state(self, state):
+        if (
+            not isinstance(state, dict)
+            or set(state) != {"topic", "step", "completed_operations"}
+            or state["topic"] != "travel"
+            or state["step"] != "confirm_itinerary"
+            or state["completed_operations"] != ["lookup:demo-trip"]
+        ):
+            raise ValueError("Unsupported workflow state")
+
+    async def restore_state(self, state):
+        self.state = state
+
+
+async def demo_policy(_checkpoint):
+    # This local demo contains only non-sensitive sample data.
+    # A real app must verify the current caller's access to the conversation
+    # and revalidate identity/privileges independently of checkpoint contents.
+    return ResumeDecision.RESTORE
+
+
+async def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=("checkpoint", "resume"))
+    parser.add_argument("database", type=Path)
+    parser.add_argument("--checkpoint-id")
+    args = parser.parse_args()
+    if args.action == "resume" and not args.checkpoint_id:
+        parser.error("resume requires --checkpoint-id")
+    workflow = Workflow()
+    store = await asyncio.to_thread(SQLiteStore, args.database)
+    manager = CheckpointManager(store, {"workflow": workflow})
+    if args.action == "checkpoint":
+        # Simulate information gathering and a completed tool operation.
+        workflow.state["step"] = "confirm_itinerary"
+        workflow.state["completed_operations"].append("lookup:demo-trip")
+        checkpoint = await manager.checkpoint("demo", ttl_seconds=3600)
+        print(checkpoint.checkpoint_id)
+        # Exit this process, losing every in-memory object and connection.
+    else:
+        decision = await manager.restore(
+            "demo", args.checkpoint_id, policy=demo_policy
+        )
+        if decision is not ResumeDecision.RESTORE:
+            raise RuntimeError(f"Cannot resume: {decision.value}")
+        print(f"Continue at: {workflow.state['step']}")
+        print("Completed lookup retained; no tool operation replayed.")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

packages/core_systems/ten_conversation_state/manifest.json

@@ -0,0 +1,20 @@
+{
+  "type": "system",
+  "name": "ten_conversation_state",
+  "version": "0.1.0",
+  "dependencies": [],
+  "description": {
+    "locales": {
+      "en-US": {
+        "content": "Opt-in conversation checkpoint and resume for Python applications."
+      }
+    }
+  },
+  "readme": {
+    "locales": {
+      "en-US": {
+        "import_uri": "README.md"
+      }
+    }
+  }
+}

packages/core_systems/ten_conversation_state/pyproject.toml

@@ -0,0 +1,12 @@
+[build-system]
+requires = ["setuptools>=77"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "ten-conversation-state"
+version = "0.1.0"
+description = "Opt-in conversation checkpoint and resume for TEN applications"
+requires-python = ">=3.10"
+
+[tool.setuptools]
+packages = ["ten_conversation_state"]