feat(agent-manager): auto-generate conversation titles from the first message

docs/WIDGET_ARCHITECTURE.md

@@ -111,6 +111,13 @@ Streaming events include:
   arguments, and the identifiers required to resume it.
 - `resume_started` — the existing run has left suspension and is executing
   again under the same `run_id`.
+- `title` — the conversation's settled, persisted title, sent after the main
+  stream's terminal event on a conversation's first turn only. Titling runs
+  alongside the turn; the widget unlocks at `final` or `pending_approval` while
+  the response may remain open for this later event. The value may be either a
+  generated title or the trimmed opening message used after generation fails.
+  The event is absent when no model is configured for titling or when the title
+  could not be persisted.
 - `error` — stream failure.
 
 While a turn runs, the composer remains editable but cannot submit another

src/agent_engine/engine/langgraph/engine.py

@@ -8,6 +8,7 @@
 from typing import Any, cast
 
 from langchain_core.callbacks import BaseCallbackHandler
+from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
 from langchain_core.runnables import RunnableConfig
 from langchain_core.tools import BaseTool
 from langgraph.types import Command
@@ -28,7 +29,7 @@
 from agent_engine.approvals.session_approval_store import SessionApprovalStore
 from agent_engine.approvals.tool_execution_manager import ToolExecutionManager
 from agent_engine.core.execution import ExecutionPolicy
-from agent_engine.core.spec import AgentSpec, SystemSpec
+from agent_engine.core.spec import AgentSpec, BaseModelConfig, SystemSpec
 from agent_engine.engine.engine import Engine
 from agent_engine.engine.langgraph.approval_provider import InterruptApprovalProvider
 from agent_engine.engine.langgraph.checkpointing import (
@@ -38,7 +39,12 @@
 from agent_engine.engine.langgraph.execution.run_lifecycle import RunLifecycle
 from agent_engine.engine.langgraph.execution.stream_channel import StreamChannel
 from agent_engine.engine.langgraph.filters import AccessFilter, RouteFilter
-from agent_engine.engine.langgraph.graph.graph_builder import GraphBuilder, ModelFactory, RunGraph
+from agent_engine.engine.langgraph.graph.graph_builder import (
+    GraphBuilder,
+    ModelFactory,
+    RunGraph,
+    build_model,
+)
 from agent_engine.engine.langgraph.graph.traversal import (
     collect_mcp_specs,
     has_protected_nodes,
@@ -213,6 +219,7 @@ def __init__(
         self._callbacks: list[BaseCallbackHandler] = [*build_callbacks(), *(callbacks or [])]
 
         self._app: RunGraph | None = None
+        self._defaults_model: BaseModelConfig | None = None
         self._system_name = ""
         self._filters: list[RouteFilter] = []
         self._mcp_connector: MCPConnector | None = None
@@ -258,6 +265,7 @@ def __init__(
 
     async def build(self, spec: SystemSpec) -> None:
         self._system_name = spec.meta.name
+        self._defaults_model = spec.defaults.model if spec.defaults else None
         self._policy = spec.execution
         register_import_roots(self._base_dir, spec.plugins.import_roots)
         self._hook_manager = HookManager.from_config(
@@ -304,6 +312,50 @@ async def close(self) -> None:
             self._mcp_connector.clear()
         self._mcp_tools.clear()
 
+    @property
+    def can_complete_text(self) -> bool:
+        return self._defaults_model is not None
+
+    async def complete(
+        self,
+        prompt: str,
+        *,
+        model: BaseModelConfig | None = None,
+        system: str | None = None,
+        max_tokens: int | None = None,
+        trace_name: str | None = None,
+    ) -> str:
+        """One stateless model call, outside the compiled graph.
+
+        `model` defaults to `defaults.model` — the same model a node with no
+        `model:` of its own would get — but any caller with its own
+        `BaseModelConfig` (region, temperature, top_p included) can ask for a
+        different one; this never depends on what a specific caller needs.
+        Built fresh per call, not cached, so a per-call `max_tokens` reaches
+        the provider through its own constructor kwarg rather than relying on
+        every provider integration honoring an invoke-time override.
+        """
+        model_config = model or self._defaults_model
+        if model_config is None:
+            raise RuntimeError(f"{self._system_name or 'this system'} has no default model")
+        overrides: dict[str, object] | None = (
+            {"max_tokens": max_tokens} if max_tokens is not None else None
+        )
+        chat_model = build_model(self._model_factory, model_config, overrides)
+        messages: list[BaseMessage] = []
+        if system:
+            messages.append(SystemMessage(content=system))
+        messages.append(HumanMessage(content=prompt))
+        config = RunnableConfig(callbacks=self._callbacks)
+        if trace_name:
+            # `run_name` is LangChain's own RunnableConfig field: the label
+            # this call shows up as in trace tooling (Langfuse), the same
+            # mechanism `_thread_config` uses for a conversation turn — this
+            # is what tells the two apart there.
+            config["run_name"] = trace_name
+        response = await chat_model.ainvoke(messages, config=config)
+        return response.text
+
     def discovered_mcp_tools(self) -> dict[str, tuple[str, ...]]:
         """Return discovered MCP tool names grouped by server for diagnostics/UIs."""
         return {

src/agent_engine/engine/langgraph/graph/graph_builder.py

@@ -40,11 +40,37 @@
 RunGraph = CompiledStateGraph[GraphState, None, GraphState, GraphState]
 
 
+def build_model(
+    factory: ModelFactory,
+    model: BaseModelConfig,
+    overrides: dict[str, object] | None = None,
+) -> BaseChatModel:
+    """The one call shape every model-construction site in this codebase uses.
+
+    `overrides` wins over `model`'s own value for a key — a caller's per-call
+    need, e.g. a bounded output length for one completion rather than the
+    model's own configured size.
+    """
+    return factory(
+        model.provider,
+        model.name,
+        model.temperature,
+        **_model_factory_kwargs(factory, model, overrides),
+    )
+
+
 def _model_factory_kwargs(
     factory: ModelFactory,
     model: BaseModelConfig,
+    overrides: dict[str, object] | None = None,
 ) -> dict[str, object]:
+    """The optional kwargs a factory call for `model` should carry.
+
+    Filtered by the factory's signature, so a narrower test factory never
+    receives a kwarg it doesn't declare.
+    """
     optional = {key: getattr(model, key) for key in _MODEL_FACTORY_OPTIONAL_KWARGS}
+    optional.update(overrides or {})
     present: dict[str, object] = {
         key: value for key, value in optional.items() if value is not None
     }
@@ -202,12 +228,7 @@ def _build_tool_invoker(
         )
 
     def _build_model(self, model: BaseModelConfig) -> BaseChatModel:
-        return self._model_factory(
-            model.provider,
-            model.name,
-            model.temperature,
-            **_model_factory_kwargs(self._model_factory, model),
-        )
+        return build_model(self._model_factory, model)
 
     def _build_model_runnable(
         self,

src/agent_engine/engine/text_completion_engine.py

@@ -0,0 +1,44 @@
+"""Optional engine capability for a single stateless text completion.
+
+For side-work that wants a model's answer without the graph: no compiled
+nodes, no tools, no protected-node filtering, no prompt-template rendering.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from agent_engine.core.spec import BaseModelConfig
+
+
+@runtime_checkable
+class TextCompletionEngine(Protocol):
+    @property
+    def can_complete_text(self) -> bool:
+        """Whether the system has a model to complete with when `model` is omitted.
+
+        Ask before calling with no `model`: a system with no default model
+        answers `False` rather than `complete` raising, so a caller can
+        degrade without needing to catch an engine-configuration error
+        alongside real completion failures (a bad response, a provider
+        outage). Irrelevant to a call that supplies its own `model`.
+        """
+        ...
+
+    async def complete(
+        self,
+        prompt: str,
+        *,
+        model: BaseModelConfig | None = None,
+        system: str | None = None,
+        max_tokens: int | None = None,
+        trace_name: str | None = None,
+    ) -> str:
+        """Complete `prompt`, optionally preceded by a `system` instruction.
+
+        `model` picks the model for this call; omitted, the engine's own
+        default is used (see `can_complete_text`). `max_tokens` bounds this
+        call's output regardless of `model`'s own configured size.
+        `trace_name` labels the call in trace tooling.
+        """
+        ...

src/agent_engine/logging_config.py

@@ -28,8 +28,15 @@ def format(self, record: logging.LogRecord) -> str:
         return f"{base} {' '.join(pairs)}" if pairs else base
 
 
-def log(logger: logging.Logger, level: int, event: str, **fields: object) -> None:
-    logger.log(level, event, extra={"fields": fields})
+def log(
+    logger: logging.Logger,
+    level: int,
+    event: str,
+    *,
+    exc_info: bool = False,
+    **fields: object,
+) -> None:
+    logger.log(level, event, extra={"fields": fields}, exc_info=exc_info)
 
 
 def configure_logging(level: str | None = None) -> None:

src/agent_manager/api/app.py

@@ -15,6 +15,7 @@
 
 from agent_engine.core.validator import SystemSpecValidator
 from agent_engine.engine.langgraph.engine import LangGraphEngine
+from agent_engine.engine.text_completion_engine import TextCompletionEngine
 from agent_engine.logging_config import configure_logging
 from agent_engine.parsers.yaml.parser import YAMLParser
 from agent_manager.api.deps import CallerIdentity
@@ -23,6 +24,19 @@
 from agent_manager.application import ConversationService
 from agent_manager.composition import application_repositories, build_identity_resolver
 from agent_manager.config import Settings
+from agent_manager.domain import TitleGenerator
+from agent_manager.infrastructure.titles import ConversationTitler
+
+
+def _title_generator(engine: object) -> TitleGenerator | None:
+    """A titler backed by the engine's own default model, or none.
+
+    A system with no `defaults.model` simply doesn't title conversations —
+    checked once at startup rather than failing per-conversation.
+    """
+    if isinstance(engine, TextCompletionEngine) and engine.can_complete_text:
+        return ConversationTitler(engine)
+    return None
 
 
 def create_app(config_path: str, settings: Settings | None = None) -> FastAPI:
@@ -51,7 +65,7 @@ async def lifespan(app: FastAPI) -> Any:
             ) as engine,
         ):
             await engine.build(spec)
-            app.state.service = ConversationService(
+            service = ConversationService(
                 engine,
                 repositories.conversations,
                 window=settings.context_window,
@@ -61,8 +75,13 @@ async def lifespan(app: FastAPI) -> Any:
                 system_name=spec.meta.name,
                 config_path=str(Path(config_path).resolve()),
                 run_repository=repositories.runs,
+                title_generator=_title_generator(engine),
             )
-            yield
+            app.state.service = service
+            try:
+                yield
+            finally:
+                await service.close()
 
     app = FastAPI(lifespan=lifespan)
     app.state.caller_identity = CallerIdentity(

src/agent_manager/api/routes/conversations.py

@@ -156,6 +156,20 @@ async def event_source() -> AsyncIterator[str]:
                 payload = to_stream_event(event).model_dump(exclude_none=True)
                 yield f"event: {event.type}\ndata: {json.dumps(payload)}\n\n"
             exhausted = True
+            # Keep this response open for the independent title result. The
+            # client treats `final`/`pending_approval` as terminal for the main
+            # execution, so a slow title cannot hold the composer locked.
+            try:
+                title = await service.wait_for_generated_title(turn)
+            except Exception:
+                # The engine stream is already terminal and durable. A defect
+                # in secondary title delivery must not rewrite that outcome as
+                # a failed conversation turn.
+                logger.exception("conversation title delivery failed")
+                title = None
+            if title is not None:
+                titled = StreamEventOut(type="title", title=title).model_dump(exclude_none=True)
+                yield f"event: title\ndata: {json.dumps(titled)}\n\n"
         except Exception:
             await service.fail_turn(turn)
             # The run is already terminal; `finally` must not try to cancel it.

src/agent_manager/api/schemas.py

@@ -128,3 +128,4 @@ class StreamEventOut(BaseModel):
     agent_id: str | None = None
     description: str | None = None
     arguments: dict[str, Any] | None = None
+    title: str | None = None