feat(examples-chat): sync itinerary — submit state + value hydration + SDK checkpoint push

Author: blove

examples/chat/angular/src/app/app.config.ts

@@ -5,6 +5,7 @@ import { provideThreadplaneTelemetry } from '@threadplane/telemetry/browser';
 import { LANGGRAPH_THREADS_CONFIG, LANGGRAPH_CLIENT_OPTIONS } from '@threadplane/langgraph';
 import { provideChat } from '@threadplane/chat';
 import { e2eClientOptions } from './shell/e2e-overrides';
+import { ItineraryStore } from './itinerary-store';
 import { routes } from './app.routes';
 import { environment } from '../environments/environment';
 
@@ -32,5 +33,9 @@ export const appConfig: ApplicationConfig = {
     provideChat({
       license: environment.license,
     }),
+    // App-wide singleton so DemoShell, the itinerary panel, and the map cockpit
+    // all read/write ONE working copy of the itinerary. Provided at root (not at
+    // the component) so routed children share the same instance.
+    ItineraryStore,
   ],
 };

examples/chat/angular/src/app/shell/demo-shell.component.spec.ts

@@ -2,8 +2,16 @@ import { signal } from '@angular/core';
 import { describe, it, expect, beforeEach } from 'vitest';
 import { TestBed } from '@angular/core/testing';
 import { provideRouter, Router, NavigationEnd } from '@angular/router';
-import { LangGraphThreadsAdapter } from '@threadplane/langgraph';
-import { DemoShell } from './demo-shell.component';
+import {
+  LangGraphThreadsAdapter,
+  LANGGRAPH_THREADS_CONFIG,
+  provideAgent,
+  FakeStreamTransport,
+  type AgentTransport,
+} from '@threadplane/langgraph';
+import { DemoShell, shouldSyncCheckpoint, extractItinerary } from './demo-shell.component';
+import { DEMO_AGENT_REF } from './agent-ref';
+import { ItineraryStore, type ItineraryStop } from '../itinerary-store';
 
 function createThreadsAdapterMock() {
   const threads = signal([]);
@@ -35,6 +43,7 @@ describe('DemoShell — mode signal', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: 'popup', component: DemoShell },
@@ -77,6 +86,7 @@ describe('DemoShell — toolbar layout', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: '', pathMatch: 'full', redirectTo: 'embed' },
@@ -93,6 +103,7 @@ describe('DemoShell — toolbar layout', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: '', pathMatch: 'full', redirectTo: 'embed' },
@@ -116,6 +127,7 @@ describe('DemoShell — toolbar dropdowns use chat-select', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: '', pathMatch: 'full', redirectTo: 'embed' },
@@ -143,6 +155,7 @@ describe('DemoShell — URL thread sync', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: 'embed/:threadId', component: DemoShell },
@@ -255,6 +268,7 @@ describe('DemoShell — URL knob hydration', () => {
     TestBed.configureTestingModule({
       providers: [
         threadsAdapterProvider,
+        ItineraryStore,
         provideRouter([
           { path: 'embed', component: DemoShell },
           { path: 'embed/:threadId', component: DemoShell },
@@ -372,3 +386,101 @@ describe('DemoShell — URL knob hydration', () => {
     expect(router.url).toContain('model=gpt-5-nano');
   });
 });
+
+// ── Itinerary ↔ checkpoint sync (Task 9) ────────────────────────────────────
+
+/** A FakeStreamTransport that records the payload of the most recent
+ *  submit so a spec can assert the shell's state injection. */
+class CapturingTransport extends FakeStreamTransport {
+  lastPayload: unknown = undefined;
+  override async *stream(
+    assistantId: string,
+    threadId: string | null,
+    payload: unknown,
+    signal: AbortSignal,
+    options?: Parameters<AgentTransport['stream']>[4],
+  ) {
+    this.lastPayload = payload;
+    yield* super.stream(assistantId, threadId, payload, signal, options);
+  }
+}
+
+describe('shouldSyncCheckpoint — push-gate predicate', () => {
+  it('pushes when settled, has a thread, and content changed', () => {
+    expect(shouldSyncCheckpoint(false, 'thread-1', '[1]', '[0]')).toBe(true);
+  });
+
+  it('never pushes while a run is loading (mid-run guard)', () => {
+    expect(shouldSyncCheckpoint(true, 'thread-1', '[1]', '[0]')).toBe(false);
+  });
+
+  it('never pushes without a thread id', () => {
+    expect(shouldSyncCheckpoint(false, null, '[1]', '[0]')).toBe(false);
+  });
+
+  it('skips when the content already matches lastSynced (echo-loop guard)', () => {
+    expect(shouldSyncCheckpoint(false, 'thread-1', '[1]', '[1]')).toBe(false);
+  });
+});
+
+describe('extractItinerary — value → stops', () => {
+  it('returns the itinerary array from a state value', () => {
+    const stops: ItineraryStop[] = [{ id: 'x', day: 1, place: 'Louvre' }];
+    expect(extractItinerary({ itinerary: stops })).toEqual(stops);
+  });
+
+  it('returns null when no itinerary present', () => {
+    expect(extractItinerary({ messages: [] })).toBeNull();
+    expect(extractItinerary(undefined)).toBeNull();
+    expect(extractItinerary({ itinerary: 'not-an-array' })).toBeNull();
+  });
+});
+
+describe('DemoShell — submit injects itinerary into state', () => {
+  beforeEach(() => {
+    localStorage.clear();
+    TestBed.configureTestingModule({
+      providers: [
+        threadsAdapterProvider,
+        ItineraryStore,
+        { provide: LANGGRAPH_THREADS_CONFIG, useValue: { apiUrl: 'http://localhost:2024' } },
+        provideRouter([
+          { path: 'embed', component: DemoShell },
+          { path: '', pathMatch: 'full', redirectTo: 'embed' },
+          { path: '**', redirectTo: 'embed' },
+        ]),
+      ],
+    });
+  });
+
+  it('forwards store.stops() as state.itinerary on submit', async () => {
+    const capturing = new CapturingTransport();
+    const store = new ItineraryStore();
+    // Override the component-scoped agent so its transport is capturable,
+    // and share the SAME store instance the shell injects.
+    TestBed.overrideComponent(DemoShell, {
+      set: {
+        providers: [
+          { provide: ItineraryStore, useValue: store },
+          ...provideAgent(DEMO_AGENT_REF, {
+            assistantId: 'fake',
+            transport: capturing,
+          }),
+        ],
+      },
+    });
+
+    const fx = TestBed.createComponent(DemoShell);
+    fx.detectChanges();
+    const shell = fx.componentInstance as unknown as {
+      agent: { submit: (i: unknown) => Promise<void> };
+    };
+
+    store.add(1, 'Louvre');
+    await shell.agent.submit({ messages: [{ role: 'user', content: 'hi' }] });
+
+    const payload = capturing.lastPayload as { itinerary?: ItineraryStop[] };
+    expect(payload.itinerary).toEqual(store.stops());
+    expect(payload.itinerary?.[0].place).toBe('Louvre');
+  });
+});

examples/chat/angular/src/app/shell/demo-shell.component.ts

@@ -13,8 +13,16 @@ import {
 import { Router, RouterOutlet, NavigationEnd } from '@angular/router';
 import { takeUntilDestroyed, toSignal } from '@angular/core/rxjs-interop';
 import { filter, map, startWith } from 'rxjs/operators';
-import { injectAgent, provideAgent, LangGraphThreadsAdapter, refreshOnRunEnd } from '@threadplane/langgraph';
+import {
+  injectAgent,
+  provideAgent,
+  LangGraphThreadsAdapter,
+  LANGGRAPH_THREADS_CONFIG,
+  createLangGraphClient,
+  refreshOnRunEnd,
+} from '@threadplane/langgraph';
 import { DEMO_AGENT_REF, type DemoState } from './agent-ref';
+import { ItineraryStore, type ItineraryStop } from '../itinerary-store';
 import { ThreadplaneTelemetryService } from '@threadplane/telemetry/browser';
 import {
   ChatInterruptPanelComponent,
@@ -65,6 +73,40 @@ function parseUrl(url: string): { mode: DemoMode; threadId: string | null } {
   return { mode, threadId };
 }
 
+// ── Itinerary ↔ checkpoint sync helpers (Task 9) ────────────────────────────
+// Extracted as pure functions so the sync decisions are unit-testable without
+// driving the agent's derived `value()`/`isLoading()` signals (which the fake
+// transport harness can't emit). The effects below stay thin wrappers.
+
+/**
+ * Decide whether to push the working itinerary to the durable checkpoint.
+ * Returns true only when a run has SETTLED (not loading), a thread exists,
+ * and the content actually changed since the last sync. The `json === lastJson`
+ * check is the echo-loop guard: hydration stamps `lastSyncedItinerary` with the
+ * incoming server JSON, so the re-fired push effect sees "no change" and skips —
+ * making hydrate→push→hydrate converge immediately.
+ */
+export function shouldSyncCheckpoint(
+  isLoading: boolean,
+  threadId: string | null,
+  json: string,
+  lastJson: string,
+): boolean {
+  if (isLoading) return false;
+  if (!threadId) return false;
+  return json !== lastJson;
+}
+
+/** Pull the itinerary array out of a graph-state `value()` snapshot, or null
+ *  when the state has no (well-formed) itinerary to hydrate from. */
+export function extractItinerary(value: unknown): ItineraryStop[] | null {
+  if (value && typeof value === 'object') {
+    const itin = (value as { itinerary?: unknown }).itinerary;
+    if (Array.isArray(itin)) return itin as ItineraryStop[];
+  }
+  return null;
+}
+
 @Component({
   selector: 'demo-shell',
   standalone: true,
@@ -115,6 +157,21 @@ export class DemoShell {
   protected readonly projectsSvc = inject(ProjectsService);
   private readonly telemetry = inject(ThreadplaneTelemetryService);
 
+  /** Shared working copy of the itinerary — an app-wide singleton (provided in
+   *  app.config.ts) so this shell, the panel, and the map read/write ONE store. */
+  protected readonly itinerary = inject(ItineraryStore);
+
+  /** Out-of-band SDK client used to push the working itinerary to the durable
+   *  checkpoint (`threads.updateState`) between runs. `LANGGRAPH_THREADS_CONFIG`
+   *  is optional so knob/routing unit tests that don't provide it still run. */
+  private readonly lgClient = createLangGraphClient(
+    inject(LANGGRAPH_THREADS_CONFIG, { optional: true })?.apiUrl ?? environment.langGraphApiUrl,
+  );
+
+  /** JSON of the last itinerary we synced (either pushed OR hydrated). Breaks
+   *  the hydrate→push→hydrate echo loop — see `shouldSyncCheckpoint`. */
+  private lastSyncedItinerary = '';
+
   constructor() {
     // Reflect the chosen theme onto <html data-theme="..."> so the
     // global stylesheet's scoped --a2ui-* overrides activate. Runs on
@@ -180,6 +237,49 @@ export class DemoShell {
     // needing a manual thread switch or reload.
     refreshOnRunEnd(this.agent, () => this.threadsSvc.refresh());
 
+    // ── Itinerary ↔ checkpoint sync (Task 9) ────────────────────────────────
+    // The checkpoint (per-thread graph state) is the durable record; the
+    // ItineraryStore is the live working copy. We sync client-authoritatively.
+
+    // (2) Hydrate the store from the checkpoint on reload / thread switch.
+    // Reads the agent's graph-state value(); when it carries an itinerary
+    // array, replaces the working copy. Stamps lastSyncedItinerary with the
+    // incoming JSON so the push effect below sees "no change" and skips —
+    // this is the other half of the echo-loop guard.
+    effect(() => {
+      const incoming = extractItinerary(this.agent.value());
+      if (incoming === null) return;
+      const json = JSON.stringify(incoming);
+      if (json === this.lastSyncedItinerary) return;
+      this.lastSyncedItinerary = json;
+      this.itinerary.hydrate(incoming);
+    });
+
+    // (3) Push the working copy to the checkpoint at run-settle and on user
+    // edits between runs. Run-gated (never mid-run) + debounced (~500ms).
+    // Depends on stops(), isLoading() (so it re-fires when a run SETTLES), and
+    // threadIdState(). Errors are swallowed — this is a best-effort sync.
+    effect((onCleanup) => {
+      const stops = this.itinerary.stops();
+      const isLoading = this.agent.isLoading();
+      const tid = threadIdState();
+      const json = JSON.stringify(stops);
+      if (!shouldSyncCheckpoint(isLoading, tid, json, this.lastSyncedItinerary)) return;
+      const timer = setTimeout(() => {
+        void (async () => {
+          try {
+            await this.lgClient.threads.updateState(tid as string, {
+              values: { itinerary: stops },
+            });
+            this.lastSyncedItinerary = json;
+          } catch {
+            // best-effort: a failed checkpoint push must not break the UI.
+          }
+        })();
+      }, 500);
+      onCleanup(() => clearTimeout(timer));
+    });
+
     if (typeof window !== 'undefined') {
       const onResize = () => this.viewportWidth.set(window.innerWidth);
       window.addEventListener('resize', onResize);
@@ -415,6 +515,7 @@ export class DemoShell {
             model: this.model(),
             reasoning_effort: this.effort(),
             gen_ui_mode: this.genUiMode(),
+            itinerary: this.itinerary.stops(),
           },
         },
         opts,