Merge pull request #263 from OpenSquawk/feature/pm-python-backend-integration

Wire /pm to Python backend for stateful ATC training sessions

AGENTS.md

@@ -0,0 +1,44 @@
+# OpenSquawk - Project Guide
+
+## Architecture
+- **Nuxt 4** (Vue 3 SFC) frontend in `/app`
+- **H3 server** handlers in `/server`
+- **Shared types/utils** in `/shared`
+- MongoDB models in `/server/models`
+- **Python backend** (`OpenSquawk-LiveATC-api`) — owns PM session state and routing decisions; runs on `http://127.0.0.1:8000`
+
+## Key Files
+- `/shared/utils/communicationsEngine.ts` — Core state machine composable (used by `/pm` live ATC). Drives local cursor and TTS; Python backend owns the authoritative state.
+- `/app/composables/useRadioBackend.ts` — Typed wrapper around the Python backend REST API (`createSession`, `transmit`, `deleteSession`, `fetchFlows`)
+- `/server/utils/openai.ts` — Legacy LLM decision router (`routeDecision()`). No longer called by `/pm`; may still be used by other routes.
+- `/server/services/decisionFlowService.ts` — Builds runtime decision trees from MongoDB (used by Nuxt `/api/decision-flows/runtime`; `/pm` now fetches directly from the Python backend)
+- `/app/pages/pm.vue` — Live ATC page (speech-to-text, PTT, text input)
+- `/app/pages/classroom.vue` — Classroom learning mode (separate system, does NOT use communicationsEngine)
+
+## Live ATC Flow (/pm) — current
+
+1. `startMonitoring()` → `fetchRuntimeTree('icao_atc_decision_tree', radioBackendUrl)` loads the YAML flow from Python backend into the local engine (for cursor tracking / TTS)
+2. `startMonitoring()` → `radioBackend.createSession('icao_atc_decision_tree')` creates an authoritative server-side session; stores `backendSessionId`
+3. User inputs (PTT or text) → `handlePilotTransmission()`
+4. `radioBackend.transmit(backendSessionId, transcript)` → Python backend runs regex routing, readback evaluation, and side effects; returns `next_state_id`, `controller_say_template`, `auto_advanced_states`, `flags`
+5. `moveToSilent(stateId)` called for each auto-advanced state then the final state — advances local cursor without triggering further auto-transitions
+6. `scheduleControllerSpeech(controller_say_template)` speaks the ATC reply via TTS
+7. PTT path: STT still goes to Nuxt (`POST /api/atc/ptt`); transcription result then calls step 4
+
+## Decision Tree States
+States have `role: 'pilot' | 'atc' | 'system'`. The engine supports two template field naming conventions:
+- Old schema: `say_tpl`, `utterance_tpl` (MongoDB/legacy)
+- New schema: `say_template`, `expected_pilot_template` (Python backend YAML)
+
+Both are handled transparently by `stateSayTpl()` and `stateUtteranceTpl()` helpers in `communicationsEngine.ts`.
+
+Template variables use `{{variable}}` (new) or `{variable}` (old) — both are rendered by `renderTpl()`.
+
+Transitions: `next`, `ok_next`, `bad_next`, `timer_next`, `auto_transitions`.
+
+## Environment Variables
+- `NUXT_PUBLIC_RADIO_BACKEND_URL` — URL of the Python backend (default `http://127.0.0.1:8000`)
+
+## Commands
+- `bun run dev` — dev server (Nuxt)
+- Python backend: see `OpenSquawk-LiveATC-api/README.md`

app/composables/useRadioBackend.ts

@@ -0,0 +1,56 @@
+export interface RadioSessionResponse {
+    session_id: string
+    flow_slug: string
+    current_state: string
+    variables: Record<string, any>
+    flags: Record<string, boolean>
+}
+
+export interface RadioTransmitResponse {
+    session_id: string
+    next_state_id: string
+    controller_say_template: string | null
+    controller_say_rendered: string | null
+    expected_pilot_template: string | null
+    variables: Record<string, any>
+    flags: Record<string, boolean>
+    trace: Array<{ type: string; message: string }>
+    fallback_used: boolean
+    fallback_reason: string | null
+    auto_advanced_states?: string[]
+}
+
+export function useRadioBackend() {
+    const config = useRuntimeConfig()
+
+    function baseUrl(): string {
+        return (config.public.radioBackendUrl as string) || 'http://127.0.0.1:8000'
+    }
+
+    async function createSession(flowSlug: string): Promise<RadioSessionResponse> {
+        return await $fetch<RadioSessionResponse>(`${baseUrl()}/api/radio/session`, {
+            method: 'POST',
+            body: { flow_slug: flowSlug },
+        })
+    }
+
+    async function transmit(sessionId: string, pilotUtterance: string): Promise<RadioTransmitResponse> {
+        return await $fetch<RadioTransmitResponse>(
+            `${baseUrl()}/api/radio/session/${sessionId}/transmissions`,
+            {
+                method: 'POST',
+                body: { pilot_utterance: pilotUtterance },
+            },
+        )
+    }
+
+    async function deleteSession(sessionId: string): Promise<void> {
+        await $fetch(`${baseUrl()}/api/radio/session/${sessionId}`, { method: 'DELETE' })
+    }
+
+    async function fetchFlows(): Promise<any> {
+        return await $fetch(`${baseUrl()}/api/decision-flows/runtime`)
+    }
+
+    return { createSession, transmit, deleteSession, fetchFlows }
+}

app/pages/pm.vue

@@ -970,6 +970,8 @@ const engine = useCommunicationsEngine()
 const auth = useAuthStore()
 const api = useApi()
 const router = useRouter()
+const radioBackend = useRadioBackend()
+const config = useRuntimeConfig()
 
 const STORAGE_KEYS = {
   selectedPlan: 'pm_selected_plan',
@@ -1014,6 +1016,7 @@ const {
   buildLLMContext,
   applyLLMDecision,
   moveTo: forceMove,
+  moveToSilent,
   normalizeATCText,
   renderATCMessage,
   getStateDetails,
@@ -1021,6 +1024,8 @@ const {
   expectedPilotPhrases,
 } = engine
 
+const backendSessionId = ref<string | null>(null)
+
 const lastTransmission = ref('')
 const lastTransmissionFaulty = ref(false)
 const lastTransmissionFaultNote = ref('')
@@ -1409,9 +1414,9 @@ onMounted(async () => {
     }
 
     try {
-      await fetchRuntimeTree()
+      await fetchRuntimeTree('icao_atc_decision_tree', config.public.radioBackendUrl as string)
     } catch (err) {
-      console.error('Failed to load decision tree runtime', err)
+      console.error('Failed to load decision tree from Python backend', err)
       error.value = 'Decision engine konnte nicht initialisiert werden.'
       return
     }
@@ -1907,51 +1912,39 @@ const handlePilotTransmission = async (message: string, source: 'text' | 'ptt' =
     speakPilotReadback(transcript)
   }
 
-  const ctx = buildLLMContext(transcript)
+  if (!backendSessionId.value) {
+    console.error('No backend session — cannot transmit')
+    setLastTransmission(`${prefix}: ${transcript} (no session)`)
+    return
+  }
 
   try {
-    const result = await api.post('/api/llm/decide', ctx)
-    const decision =
-      result?.decision && typeof result.decision === 'object'
-        ? result.decision
-        : (result && typeof result === 'object' && 'next_state' in result)
-          ? result
-          : null
+    const response = await radioBackend.transmit(backendSessionId.value, transcript)
 
-    if (!decision) {
-      console.error('LLM decision response had unexpected shape:', result)
-      setLastTransmission(`${prefix}: ${transcript} (invalid decision response)`)
-      return
+    // Advance local cursor through every state the backend auto-walked, then
+    // the final state. moveToSilent updates current_unit, actions, handoffs,
+    // and the communication log without scheduling further auto-transitions.
+    for (const stateId of response.auto_advanced_states ?? []) {
+      moveToSilent(stateId)
+    }
+    moveToSilent(response.next_state_id)
+
+    // Sync boolean routing flags (in_air, emergency_active, etc.) from backend.
+    for (const [k, v] of Object.entries(response.flags ?? {})) {
+      if (typeof v === 'boolean') (flags as any).value[k] = v
     }
 
-    const normalizedTrace = normalizeDecisionTraceResult(result)
-
-    applyLLMDecision(decision, normalizedTrace ?? null)
-
-    // If decision explicitly has controller_say_tpl, speak it
-    if (decision.controller_say_tpl && !decision.radio_check) {
-      scheduleControllerSpeech(decision.controller_say_tpl)
+    // TTS: use the template so local variables (callsign, squawk, etc.) render correctly.
+    if (response.controller_say_template) {
+      scheduleControllerSpeech(response.controller_say_template)
     }
 
-    // Auto-advance through ATC/system states and speak any say_tpl messages.
-    // This is the key fix: after a decision moves us to a new state, we need
-    // to walk through all non-pilot states (ATC replies, system transitions)
-    // until we reach the next pilot state, speaking each ATC message via TTS.
-    if (!decision.radio_check) {
-      await nextTick()
-      const atcMessages = collectAtcStatesUntilPilotTurn()
-      for (const msg of atcMessages) {
-        // Don't double-speak if the decision already had controller_say_tpl
-        // for this exact template
-        if (decision.controller_say_tpl && msg.say_tpl === decision.controller_say_tpl) {
-          continue
-        }
-        scheduleControllerSpeech(msg.say_tpl)
-      }
+    if (response.fallback_used) {
+      console.warn('[Backend] Fallback used:', response.fallback_reason)
     }
   } catch (e) {
-    console.error('LLM decision failed', e)
-    setLastTransmission(`${prefix}: ${transcript} (LLM failed)`)
+    console.error('Backend transmission failed', e)
+    setLastTransmission(`${prefix}: ${transcript} (backend failed)`)
   }
 }
 
@@ -1982,9 +1975,10 @@ const loadFlightPlans = async () => {
 }
 
 const startMonitoring = async (flightPlan: any) => {
+  // 1. Ensure the local tree is loaded from the Python backend (same source as session)
   try {
     if (!engineReady.value) {
-      await fetchRuntimeTree()
+      await fetchRuntimeTree('icao_atc_decision_tree', config.public.radioBackendUrl as string)
     }
   } catch (err) {
     console.error('Failed to prepare decision engine', err)
@@ -1992,22 +1986,41 @@ const startMonitoring = async (flightPlan: any) => {
     return
   }
 
+  // 2. Create a backend session — this is the authoritative state from here on
+  try {
+    const session = await radioBackend.createSession('icao_atc_decision_tree')
+    backendSessionId.value = session.session_id
+    // Sync cursor to wherever the backend initialised (usually start_state)
+    moveToSilent(session.current_state)
+    // Sync boolean routing flags from session
+    for (const [k, v] of Object.entries(session.flags ?? {})) {
+      if (typeof v === 'boolean') (flags as any).value[k] = v
+    }
+  } catch (err) {
+    console.error('Failed to create backend session', err)
+    error.value = 'Verbindung zum Training-Backend fehlgeschlagen.'
+    return
+  }
+
   error.value = ''
   selectedPlan.value = flightPlan
+
+  // 3. initializeFlight sets the real flight plan variables (callsign, squawk, dep/dest, etc.)
+  //    This runs AFTER session creation so these values win over backend defaults.
   initializeFlight(flightPlan)
   currentScreen.value = 'monitor'
   persistSelectedPlan(flightPlan)
 
-  // Set appropriate frequency based on departure airport
   if (flightPlan.dep === 'EDDF') {
-    frequencies.value.active = '121.900' // Frankfurt Delivery
-    frequencies.value.standby = '121.700' // Frankfurt Ground
+    frequencies.value.active = '121.900'
+    frequencies.value.standby = '121.700'
   }
 
   await fetchAirportFrequencies(flightPlan.dep || flightPlan.departure)
 
-  // If the start state is an ATC state, auto-advance and speak its message
-  // so the pilot sees the first prompt immediately after connecting.
+  // 4. Walk the initial ATC/system states locally (deterministic, no LLM).
+  //    Safe because we loaded the tree from the same Python backend, so the
+  //    walk is identical to what the backend will do on the first transmission.
   try {
     await nextTick()
     const startMessages = collectAtcStatesUntilPilotTurn()

nuxt.config.ts

@@ -57,6 +57,7 @@ export default defineNuxtConfig({
         },
         public: {
             apiDocumentationUrl: '/api-docs',
+            radioBackendUrl: process.env.NUXT_PUBLIC_RADIO_BACKEND_URL || 'http://127.0.0.1:8000',
         },
     },
     vuetify: {

shared/types/decision.ts

@@ -220,7 +220,9 @@ export interface RuntimeDecisionState {
   name?: string
   summary?: string
   say_tpl?: string
+  say_template?: string
   utterance_tpl?: string
+  expected_pilot_template?: string
   else_say_tpl?: string
   next?: Array<{ to: string; label?: string; when?: string; guard?: string }>
   ok_next?: Array<{ to: string; label?: string; when?: string; guard?: string }>

shared/utils/communicationsEngine.ts

@@ -159,7 +159,10 @@ function createDefaultFlightContext(): FlightContext {
 }
 
 function renderTpl(tpl: string, ctx: Record<string, any>): string {
-    return tpl.replace(/\{([\w.]+)\}/g, (_m, key) => {
+    // Handles both {variable} (old schema) and {{variable}} (new YAML/Jinja2 schema).
+    // Double-brace is matched first so {{x}} isn't partially matched as {x} with trailing brace.
+    return tpl.replace(/\{\{([\w.]+)\}\}|\{([\w.]+)\}/g, (_m, key1, key2) => {
+        const key = key1 ?? key2
         const parts = key.split('.')
         let cur: any = ctx
         for (const p of parts) cur = cur?.[p]
@@ -167,6 +170,14 @@ function renderTpl(tpl: string, ctx: Record<string, any>): string {
     })
 }
 
+function stateSayTpl(s: RuntimeDecisionState): string | undefined {
+    return s.say_tpl ?? s.say_template
+}
+
+function stateUtteranceTpl(s: RuntimeDecisionState): string | undefined {
+    return s.utterance_tpl ?? s.expected_pilot_template
+}
+
 export default function useCommunicationsEngine() {
     const runtimeSystem = ref<RuntimeDecisionSystem | null>(null)
     const flowOrder = ref<string[]>([])
@@ -562,13 +573,20 @@ export default function useCommunicationsEngine() {
             })
     })
 
-    async function fetchRuntimeTree(slug = 'icao_atc_decision_tree') {
+    async function fetchRuntimeTree(slug = 'icao_atc_decision_tree', baseUrl?: string) {
         ready.value = false
-        const fetcher: any = (globalThis as any).$fetch
-        if (typeof fetcher !== 'function') {
-            throw new Error('Universal fetch is not available in this context')
+        let data: RuntimeDecisionSystem
+        if (baseUrl) {
+            const res = await fetch(`${baseUrl}/api/decision-flows/runtime`)
+            if (!res.ok) throw new Error(`Failed to load flows from ${baseUrl}: ${res.status}`)
+            data = await res.json() as RuntimeDecisionSystem
+        } else {
+            const fetcher: any = (globalThis as any).$fetch
+            if (typeof fetcher !== 'function') {
+                throw new Error('Universal fetch is not available in this context')
+            }
+            data = (await fetcher('/api/decision-flows/runtime')) as RuntimeDecisionSystem
         }
-        const data = (await fetcher('/api/decision-flows/runtime')) as RuntimeDecisionSystem
         const activeSlug = slug && data.flows[slug] ? slug : data.main
         resetEngineFromSystem(data, { activeSlug })
     }
@@ -959,12 +977,13 @@ export default function useCommunicationsEngine() {
             if (endStates.includes(s.id)) break
 
             // Collect ATC/system messages for TTS
-            if (s.say_tpl) {
+            const sayTpl = stateSayTpl(s)
+            if (sayTpl) {
                 messages.push({
                     stateId: s.id,
-                    say_tpl: s.say_tpl,
-                    rendered: renderTpl(s.say_tpl, exposeCtx()),
-                    normalized: normalizeATCText(s.say_tpl, exposeCtxFlat()),
+                    say_tpl: sayTpl,
+                    rendered: renderTpl(sayTpl, exposeCtx()),
+                    normalized: normalizeATCText(sayTpl, exposeCtxFlat()),
                 })
             }
 
@@ -991,10 +1010,16 @@ export default function useCommunicationsEngine() {
             }
 
             if (!nextId) {
-                // Collect all eligible transitions
+                // Collect all eligible transitions — also include new-schema auto_transitions
+                // (used by ATC states in the new YAML that have no ok_next/next entries)
+                const autoTransitionTargets = (s.auto_transitions ?? [])
+                    .map(t => ({ to: (t as any).to as string | undefined }))
+                    .filter(t => !!t.to)
+
                 const allTransitions = [
                     ...(s.next ?? []),
                     ...(s.ok_next ?? []),
+                    ...autoTransitionTargets,
                 ] as Array<{ to?: string; when?: string; guard?: string }>
 
                 const eligible = allTransitions.filter(t => {
@@ -1028,12 +1053,13 @@ export default function useCommunicationsEngine() {
         const s = currentState.value
         if (!s) return []
 
-        // If current state IS a pilot state with utterance_tpl, show it
-        if (s.role === 'pilot' && s.utterance_tpl) {
+        // If current state IS a pilot state with utterance_tpl / expected_pilot_template, show it
+        const currentUtteranceTpl = stateUtteranceTpl(s)
+        if (s.role === 'pilot' && currentUtteranceTpl) {
             return [{
                 stateId: s.id,
-                text: renderTpl(s.utterance_tpl, exposeCtx()),
-                normalized: normalizeATCText(s.utterance_tpl, exposeCtxFlat()),
+                text: renderTpl(currentUtteranceTpl, exposeCtx()),
+                normalized: normalizeATCText(currentUtteranceTpl, exposeCtxFlat()),
             }]
         }
 
@@ -1042,11 +1068,12 @@ export default function useCommunicationsEngine() {
         for (const id of nextCandidates.value) {
             const state = states.value[id]
             if (!state) continue
-            if (state.role === 'pilot' && state.utterance_tpl) {
+            const utteranceTpl = stateUtteranceTpl(state)
+            if (state.role === 'pilot' && utteranceTpl) {
                 results.push({
                     stateId: id,
-                    text: renderTpl(state.utterance_tpl, exposeCtx()),
-                    normalized: normalizeATCText(state.utterance_tpl, exposeCtxFlat()),
+                    text: renderTpl(utteranceTpl, exposeCtx()),
+                    normalized: normalizeATCText(utteranceTpl, exposeCtxFlat()),
                 })
             }
         }
@@ -1111,7 +1138,7 @@ export default function useCommunicationsEngine() {
 
         const state = currentState.value
         if (!state) return
-        if (state.role === 'atc' && state.say_tpl) return
+        if (state.role === 'atc' && stateSayTpl(state)) return
 
         const transitions = [
             ...(state.next ?? []),
@@ -1141,7 +1168,7 @@ export default function useCommunicationsEngine() {
         const targetState = states.value[targetId]
         if (!targetState) return
 
-        const silentSystemHop = targetState.role === 'system' && !targetState.say_tpl
+        const silentSystemHop = targetState.role === 'system' && !stateSayTpl(targetState)
         const delay = silentSystemHop
             ? 50
             : Math.floor(Math.random() * 1000) + 1000
@@ -1230,8 +1257,9 @@ export default function useCommunicationsEngine() {
         }
 
         // Auto-Say
-        if (s.say_tpl) {
-            speak(s.role, s.say_tpl, s.id!)
+        const sayTplMoveTo = stateSayTpl(s)
+        if (sayTplMoveTo) {
+            speak(s.role, sayTplMoveTo, s.id!)
         }
 
         // Update flight context phase
@@ -1239,6 +1267,49 @@ export default function useCommunicationsEngine() {
         queueMicrotask(() => evaluateAutoTransitions())
     }
 
+    // Like moveTo but does NOT schedule auto-transitions — used when the backend
+    // is driving state and we only need to sync the local cursor + side-effects
+    // (actions, handoffs, communication log) without the engine trying to advance further.
+    function moveToSilent(stateId: string) {
+        ensureTree()
+        if (!states.value[stateId]) {
+            console.warn(`[Engine] Unknown state for silent move: ${stateId}`)
+            return
+        }
+
+        if (stateId.startsWith('INT_')) {
+            flags.value.stack.push(currentStateId.value)
+        }
+
+        setActiveStateId(stateId)
+        resetAutoHistory(stateId)
+        const s = currentState.value
+        if (!s) return
+
+        for (const act of s.actions ?? []) {
+            if (typeof act === 'string') continue
+            if (act.if && !safeEvalBoolean(act.if)) continue
+            if (act.set) {
+                setByPath({ variables: variables.value, flags: flags.value, telemetry: telemetry.value }, act.set, act.to)
+            }
+        }
+
+        if (s.handoff?.to) {
+            flags.value.current_unit = unitFromHandoff(s.handoff.to)
+            if (s.handoff.freq) {
+                variables.value.handoff_freq = renderTpl(s.handoff.freq, exposeCtx())
+            }
+        }
+
+        const sayTplSilent = stateSayTpl(s)
+        if (sayTplSilent) {
+            speak(s.role, sayTplSilent, s.id!)
+        }
+
+        updateFlightPhase(s.phase)
+        // deliberately no evaluateAutoTransitions — backend owns the next move
+    }
+
     function updateFlightPhase(phase: Phase) {
         const phaseMap: Record<Phase, string> = {
             'Clearance': 'clearance',
@@ -1419,6 +1490,7 @@ export default function useCommunicationsEngine() {
 
         // Flow Control
         moveTo,
+        moveToSilent,
 
         // Utilities
         normalizeATCText,