[Bug] <SDK accepts cross-journey transitions and later crashes journey projection with `KeyError`>

uv sync --group dev

import asyncio
import json
from typing import Any, Generic, Mapping, TypeVar

from parlant import sdk as p
from parlant.core.common import DefaultBaseModel
from parlant.core.journey_guideline_projection import JourneyGuidelineProjection
from parlant.core.journeys import JourneyStore
from parlant.core.nlp.embedding import NullEmbedder
from parlant.core.nlp.generation import SchematicGenerationResult, SchematicGenerator, StreamingTextGenerator
from parlant.core.nlp.moderation import ModerationService, NoModeration
from parlant.core.nlp.service import EmbedderHints, NLPService, SchematicGeneratorHints, StreamingTextGeneratorHints
from parlant.core.nlp.tokenization import ZeroEstimatingTokenizer

T = TypeVar("T", bound=DefaultBaseModel)


class ProbeComplete(Exception):
    pass


class DummyGenerator(SchematicGenerator[T], Generic[T]):
    def __init__(self, schema: type[T]) -> None:
        self._schema = schema
        self._tokenizer = ZeroEstimatingTokenizer()

    @property
    def schema(self) -> type[T]:
        return self._schema

    async def generate(self, prompt: str | Any, hints: Mapping[str, Any] = {}):
        raise RuntimeError("not used in this repro")

    @property
    def id(self) -> str:
        return f"dummy:{self._schema.__name__}"

    @property
    def max_tokens(self) -> int:
        return 8192

    @property
    def tokenizer(self):
        return self._tokenizer


class NoopNLPService(NLPService):
    @property
    def supports_streaming(self) -> bool:
        return False

    async def get_schematic_generator(
        self,
        t: type[T],
        hints: SchematicGeneratorHints = {},
    ) -> SchematicGenerator[T]:
        return DummyGenerator[t](t)  # type: ignore[index]

    async def get_streaming_text_generator(
        self,
        hints: StreamingTextGeneratorHints = {},
    ) -> StreamingTextGenerator:
        raise RuntimeError("not used in this repro")

    async def get_embedder(self, hints: EmbedderHints = {}):
        return NullEmbedder()

    async def get_moderation_service(self) -> ModerationService:
        return NoModeration()


async def main() -> None:
    server = p.Server(
        port=8800,
        tool_service_port=8818,
        log_level=p.LogLevel.TRACE,
        nlp_service=lambda c: NoopNLPService(),
    )

    try:
        async with server:
            agent = await server.create_agent(
                name="Probe Agent",
                description="Agent for reproducing cross-journey transition behavior",
            )

            journey1 = await agent.create_journey(
                title="Journey One",
                triggers=[],
                description="First journey",
            )
            journey2 = await agent.create_journey(
                title="Journey Two",
                triggers=[],
                description="Second journey",
            )

            journey2_transition = await journey2.initial_state.transition_to(
                chat_state="inside journey two"
            )
            foreign_state = journey2_transition.target

            cross_transition = await journey1.initial_state.transition_to(
                condition="jump across journeys",
                state=foreign_state,
            )

            journey_store = server.container[JourneyStore]
            projection = server.container[JourneyGuidelineProjection]

            stored_journey = await journey_store.read_journey(journey1.id)
            edges = await journey_store.list_edges(journey1.id)
            nodes = await journey_store.list_nodes(journey1.id)

            result = {
                "journey1_id": journey1.id,
                "journey2_id": journey2.id,
                "foreign_state_id": foreign_state.id,
                "cross_transition_target_id": cross_transition.target.id,
                "journey1_root_id": stored_journey.root_id,
                "journey1_node_ids": [n.id for n in nodes],
                "journey1_edges": [
                    {
                        "id": e.id,
                        "source": e.source,
                        "target": e.target,
                        "condition": e.condition,
                    }
                    for e in edges
                ],
            }

            try:
                guidelines = await projection.project_journey_to_guidelines(journey1.id)
                result["projection"] = {
                    "status": "ok",
                    "guideline_ids": [g.id for g in guidelines],
                }
            except Exception as exc:
                result["projection"] = {
                    "status": "error",
                    "error_type": type(exc).__name__,
                    "error": str(exc),
                }

            print(json.dumps(result, indent=2, default=str))
            raise ProbeComplete()
    except ProbeComplete:
        pass


asyncio.run(main())

uv run python cross_journey_probe.py

{
  "journey1_id": "8f2rGlcVin",
  "journey2_id": "htMqKzUPLW",
  "foreign_state_id": "hnJ5V3UIsq",
  "cross_transition_target_id": "hnJ5V3UIsq",
  "journey1_root_id": "r1HEKb2WOW",
  "journey1_node_ids": [
    "r1HEKb2WOW",
    "end"
  ],
  "journey1_edges": [
    {
      "id": "lvIDDCk8Xb",
      "source": "r1HEKb2WOW",
      "target": "hnJ5V3UIsq",
      "condition": "jump across journeys"
    }
  ],
  "projection": {
    "status": "error",
    "error_type": "KeyError",
    "error": "'hnJ5V3UIsq'"
  }
}