Time Travel

"""
LangGraph Time Travel Graph
 
Demonstrates checkpoint-based time travel. Each message exchange is saved as
a checkpoint snapshot. The client reads the thread's checkpoint history and can
start a new run from any past checkpoint by submitting with that checkpoint.
 
The LangGraph API server provides checkpointing automatically.
"""
 
from pathlib import Path
from typing import NotRequired
from langgraph.graph import StateGraph, MessagesState, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage
 
PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
 
 
class TimeTravelState(MessagesState):
    """Canonical final identities let the React preview verify adopted history."""
 
    completed_turn_id: NotRequired[str]
    completed_answer_id: NotRequired[str]
    completed_message_ids: NotRequired[list[str]]
 
 
def build_time_travel_graph():
    """
    Constructs a StateGraph that the LangGraph API server checkpoints.
 
    The LangGraph API checkpointer saves a snapshot after each node execution,
    producing a history of ThreadState objects that the client can replay or
    branch from using checkpoint IDs.
    """
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
 
    async def generate(state: TimeTravelState) -> dict:
        """Generate a response, checkpointed for time travel."""
        system_prompt = (PROMPTS_DIR / "time-travel.md").read_text()
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        response = await llm.ainvoke(messages)
        canonical = [*state["messages"], response]
        ids = [getattr(message, "id", None) for message in canonical]
        completion = {}
        if (len(canonical) >= 2
                and getattr(canonical[-2], "type", None) == "human"
                and getattr(response, "type", None) == "ai"
                and all(isinstance(identity, str) and identity for identity in ids)
                and len(ids) == len(set(ids))):
            completion = {
                "completed_turn_id": ids[-2],
                "completed_answer_id": ids[-1],
                "completed_message_ids": ids,
            }
        return {"messages": [response], **completion}
 
    graph = StateGraph(TimeTravelState)
    graph.add_node("generate", generate)
    graph.set_entry_point("generate")
    graph.add_edge("generate", END)
    return graph.compile()
 
 
# The graph instance — referenced by langgraph.json
graph = build_time_travel_graph()

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/langgraph';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'langgraph') {
        throw new Error('incompatible runtime');
      }
      return {
        apiUrl: connection.apiUrl,
        assistantId: connection.assistantId,
        clientOptions: connection.clientOptions,
      };
    }),
  ],
};

protected readonly agent = injectAgent();
 
/** Selection survives new checkpoints being prepended to the history. */
protected readonly selectedCheckpointId = signal<string | null>(null);
 
/** Checkpoint history derived from the agent. */
protected readonly checkpoints = computed(
  (): ThreadState<any>[] => this.agent.langGraphHistory(),
);

<!-- Checkpoint timeline sidebar -->
<div sidebar class="panel">
  <div class="head">
    <h2 class="cap">Timeline</h2>
    <p class="sub">{{ checkpoints().length }} checkpoint(s)</p>
  </div>
 
  <div class="list">
    @if (checkpoints().length === 0) {
      <p class="empty">No checkpoints yet. Send a message to begin.</p>
    }
 
    @for (state of checkpoints(); track state.checkpoint?.checkpoint_id ?? $index; let i = $index) {
      <div
        class="row"
        [class.row--active]="state.checkpoint?.checkpoint_id === selectedCheckpointId()"
      >
        <!-- Numbered badge -->
        <span
          class="badge"
          [class.badge--active]="state.checkpoint?.checkpoint_id === selectedCheckpointId()"
        >
          {{ i + 1 }}
        </span>
 
        <!-- Checkpoint info -->
        <div class="info">
          <p class="label">{{ checkpointLabel(state, i) }}</p>
          @if (state.checkpoint?.checkpoint_id) {
            <p class="id">{{ state.checkpoint.checkpoint_id }}</p>
          }
        </div>
 
        <!-- Action buttons -->
        <div class="actions">
          <button
            class="btn"
            title="Select this checkpoint as the active branch"
            (click)="select(state)"
          >
            Select
          </button>
          <button
            class="btn"
            title="Fork from this checkpoint"
            (click)="fork(state)"
          >
            Fork
          </button>
        </div>
      </div>
    }
  </div>
</div>

/**
 * Mark a checkpoint as the active branch.
 *
 * `setBranch()` records the identifier in the agent's `branch()` signal.
 * It starts no run: it is a pointer the interface can read back.
 */
protected select(state: ThreadState<any>): void {
  if (state.checkpoint?.checkpoint_id) {
    this.selectedCheckpointId.set(state.checkpoint.checkpoint_id);
    this.agent.setBranch(state.checkpoint.checkpoint_id);
  }
}
 
/**
 * Start a new run from the given checkpoint.
 *
 * Passing `checkpointId` to `submit()` resumes the thread at that point
 * rather than at its tip, so the new run hangs off the chosen checkpoint
 * and the original path stays intact.
 */
protected fork(state: ThreadState<any>): void {
  const checkpointId = state.checkpoint?.checkpoint_id;
  if (!checkpointId) return;
  this.selectedCheckpointId.set(checkpointId);
  void this.agent.submit(
    { message: 'Try a different approach from here.' },
    { checkpointId },
  );
}